Split PacketCount to Egress and Ingress

Also, 1 packet may be counted in N topologies, so you can't rely on the
sum of all packet counts across topologies having any relation to the
sampling data.

app/api_topology_test.go

@@ -96,8 +96,8 @@ func TestAPITopologyApplications(t *testing.T) {
 			t.Fatalf("JSON parse error: %s", err)
 		}
 		if want, have := (report.EdgeMetadata{
-			PacketCount:     newu64(10),
-			EgressByteCount: newu64(100),
+			EgressPacketCount: newu64(10),
+			EgressByteCount:   newu64(100),
 		}), edge.Metadata; !reflect.DeepEqual(want, have) {
 			t.Error(test.Diff(want, have))
 		}

probe/sniff/sniffer.go

@@ -119,8 +119,11 @@ func interpolateCounts(r report.Report) {
 	factor := 1.0 / rate
 	for _, topology := range r.Topologies() {
 		for _, emd := range topology.EdgeMetadatas {
-			if emd.PacketCount != nil {
-				*emd.PacketCount = uint64(float64(*emd.PacketCount) * factor)
+			if emd.EgressPacketCount != nil {
+				*emd.EgressPacketCount = uint64(float64(*emd.EgressPacketCount) * factor)
+			}
+			if emd.IngressPacketCount != nil {
+				*emd.IngressPacketCount = uint64(float64(*emd.IngressPacketCount) * factor)
 			}
 			if emd.EgressByteCount != nil {
 				*emd.EgressByteCount = uint64(float64(*emd.EgressByteCount) * factor)
@@ -261,17 +264,21 @@ func (s *Sniffer) Merge(p Packet, rpt report.Report) {
 		}
 
 		emd := rpt.Address.EdgeMetadatas[edgeID]
-		if emd.PacketCount == nil {
-			emd.PacketCount = new(uint64)
-		}
-		*emd.PacketCount++
 
 		if egress {
+			if emd.EgressPacketCount == nil {
+				emd.EgressPacketCount = new(uint64)
+			}
+			*emd.EgressPacketCount++
 			if emd.EgressByteCount == nil {
 				emd.EgressByteCount = new(uint64)
 			}
 			*emd.EgressByteCount += uint64(p.Network)
 		} else {
+			if emd.IngressPacketCount == nil {
+				emd.IngressPacketCount = new(uint64)
+			}
+			*emd.IngressPacketCount++
 			if emd.IngressByteCount == nil {
 				emd.IngressByteCount = new(uint64)
 			}
@@ -296,17 +303,21 @@ func (s *Sniffer) Merge(p Packet, rpt report.Report) {
 		}
 
 		emd := rpt.Endpoint.EdgeMetadatas[edgeID]
-		if emd.PacketCount == nil {
-			emd.PacketCount = new(uint64)
-		}
-		*emd.PacketCount++
 
 		if egress {
+			if emd.EgressPacketCount == nil {
+				emd.EgressPacketCount = new(uint64)
+			}
+			*emd.EgressPacketCount++
 			if emd.EgressByteCount == nil {
 				emd.EgressByteCount = new(uint64)
 			}
 			*emd.EgressByteCount += uint64(p.Transport)
 		} else {
+			if emd.IngressPacketCount == nil {
+				emd.IngressPacketCount = new(uint64)
+			}
+			*emd.IngressPacketCount++
 			if emd.IngressByteCount == nil {
 				emd.IngressByteCount = new(uint64)
 			}

probe/sniff/sniffer_internal_test.go

@@ -22,9 +22,10 @@ func TestInterpolateCounts(t *testing.T) {
 	r.Sampling.Count = samplingCount
 	r.Sampling.Total = samplingTotal
 	r.Endpoint.EdgeMetadatas[edgeID] = report.EdgeMetadata{
-		PacketCount:      newu64(packetCount),
-		IngressByteCount: newu64(byteCount),
-		EgressByteCount:  newu64(byteCount),
+		EgressPacketCount:  newu64(packetCount),
+		IngressPacketCount: newu64(packetCount),
+		EgressByteCount:    newu64(byteCount),
+		IngressByteCount:   newu64(byteCount),
 	}
 
 	interpolateCounts(r)
@@ -35,7 +36,10 @@ func TestInterpolateCounts(t *testing.T) {
 		apply  = func(v uint64) uint64 { return uint64(factor * float64(v)) }
 		emd    = r.Endpoint.EdgeMetadatas[edgeID]
 	)
-	if want, have := apply(packetCount), (*emd.PacketCount); want != have {
+	if want, have := apply(packetCount), (*emd.EgressPacketCount); want != have {
+		t.Errorf("want %d packets, have %d", want, have)
+	}
+	if want, have := apply(packetCount), (*emd.IngressPacketCount); want != have {
 		t.Errorf("want %d packets, have %d", want, have)
 	}
 	if want, have := apply(byteCount), (*emd.EgressByteCount); want != have {

probe/sniff/sniffer_test.go

@@ -72,8 +72,8 @@ func TestMerge(t *testing.T) {
 		},
 		EdgeMetadatas: report.EdgeMetadatas{
 			report.MakeEdgeID(srcEndpointNodeID, dstEndpointNodeID): report.EdgeMetadata{
-				PacketCount:     newu64(1),
-				EgressByteCount: newu64(256),
+				EgressPacketCount: newu64(1),
+				EgressByteCount:   newu64(256),
 			},
 		},
 		NodeMetadatas: report.NodeMetadatas{
@@ -95,8 +95,8 @@ func TestMerge(t *testing.T) {
 		},
 		EdgeMetadatas: report.EdgeMetadatas{
 			report.MakeEdgeID(srcAddressNodeID, dstAddressNodeID): report.EdgeMetadata{
-				PacketCount:     newu64(1),
-				EgressByteCount: newu64(512),
+				EgressPacketCount: newu64(1),
+				EgressByteCount:   newu64(512),
 			},
 		},
 		NodeMetadatas: report.NodeMetadatas{

render/detailed_node.go

@@ -64,8 +64,11 @@ func MakeDetailedNode(r report.Report, n RenderableNode) DetailedNode {
 		if n.EdgeMetadata.MaxConnCountTCP != nil {
 			rows = append(rows, Row{"TCP connections", strconv.FormatUint(*n.EdgeMetadata.MaxConnCountTCP, 10), ""})
 		}
-		if n.EdgeMetadata.PacketCount != nil {
-			rows = append(rows, Row{"Packets", strconv.FormatUint(*n.EdgeMetadata.PacketCount, 10), ""})
+		if n.EdgeMetadata.EgressPacketCount != nil {
+			rows = append(rows, Row{"Egress packets", strconv.FormatUint(*n.EdgeMetadata.EgressPacketCount, 10), ""})
+		}
+		if n.EdgeMetadata.IngressPacketCount != nil {
+			rows = append(rows, Row{"Ingress packets", strconv.FormatUint(*n.EdgeMetadata.IngressPacketCount, 10), ""})
 		}
 		if n.EdgeMetadata.EgressByteCount != nil {
 			rows = append(rows, Row{"Egress bytes", strconv.FormatUint(*n.EdgeMetadata.EgressByteCount, 10), ""}) // TODO rate

render/detailed_node_test.go

@@ -66,7 +66,7 @@ func TestMakeDetailedNode(t *testing.T) {
 				Numeric: true,
 				Rank:    100,
 				Rows: []render.Row{
-					{"Packets", "150", ""},
+					{"Egress packets", "150", ""},
 					{"Egress bytes", "1500", ""},
 				},
 			},

render/expected/expected.go

@@ -55,8 +55,8 @@ var (
 			),
 			NodeMetadata: report.MakeNodeMetadata(),
 			EdgeMetadata: report.EdgeMetadata{
-				PacketCount:     newu64(10),
-				EgressByteCount: newu64(100),
+				EgressPacketCount: newu64(10),
+				EgressByteCount:   newu64(100),
 			},
 		},
 		ClientProcess2ID: {
@@ -73,8 +73,8 @@ var (
 			),
 			NodeMetadata: report.MakeNodeMetadata(),
 			EdgeMetadata: report.EdgeMetadata{
-				PacketCount:     newu64(20),
-				EgressByteCount: newu64(200),
+				EgressPacketCount: newu64(20),
+				EgressByteCount:   newu64(200),
 			},
 		},
 		ServerProcessID: {
@@ -97,8 +97,8 @@ var (
 			),
 			NodeMetadata: report.MakeNodeMetadata(),
 			EdgeMetadata: report.EdgeMetadata{
-				PacketCount:     newu64(150),
-				EgressByteCount: newu64(1500),
+				EgressPacketCount: newu64(150),
+				EgressByteCount:   newu64(1500),
 			},
 		},
 		nonContainerProcessID: {
@@ -137,8 +137,8 @@ var (
 			),
 			NodeMetadata: report.MakeNodeMetadata(),
 			EdgeMetadata: report.EdgeMetadata{
-				PacketCount:     newu64(30),
-				EgressByteCount: newu64(300),
+				EgressPacketCount: newu64(30),
+				EgressByteCount:   newu64(300),
 			},
 		},
 		"apache": {
@@ -160,8 +160,8 @@ var (
 			),
 			NodeMetadata: report.MakeNodeMetadata(),
 			EdgeMetadata: report.EdgeMetadata{
-				PacketCount:     newu64(150),
-				EgressByteCount: newu64(1500),
+				EgressPacketCount: newu64(150),
+				EgressByteCount:   newu64(1500),
 			},
 		},
 		"bash": {
@@ -200,8 +200,8 @@ var (
 			),
 			NodeMetadata: report.MakeNodeMetadata(),
 			EdgeMetadata: report.EdgeMetadata{
-				PacketCount:     newu64(30),
-				EgressByteCount: newu64(300),
+				EgressPacketCount: newu64(30),
+				EgressByteCount:   newu64(300),
 			},
 		},
 		test.ServerContainerID: {
@@ -219,8 +219,8 @@ var (
 			),
 			NodeMetadata: report.MakeNodeMetadata(),
 			EdgeMetadata: report.EdgeMetadata{
-				PacketCount:     newu64(150),
-				EgressByteCount: newu64(1500),
+				EgressPacketCount: newu64(150),
+				EgressByteCount:   newu64(1500),
 			},
 		},
 		uncontainedServerID: {
@@ -258,8 +258,8 @@ var (
 			),
 			NodeMetadata: report.MakeNodeMetadata(),
 			EdgeMetadata: report.EdgeMetadata{
-				PacketCount:     newu64(30),
-				EgressByteCount: newu64(300),
+				EgressPacketCount: newu64(30),
+				EgressByteCount:   newu64(300),
 			},
 		},
 		test.ServerContainerImageName: {
@@ -277,8 +277,8 @@ var (
 				test.ServerHostNodeID),
 			NodeMetadata: report.MakeNodeMetadata(),
 			EdgeMetadata: report.EdgeMetadata{
-				PacketCount:     newu64(150),
-				EgressByteCount: newu64(1500),
+				EgressPacketCount: newu64(150),
+				EgressByteCount:   newu64(1500),
 			},
 		},
 		uncontainedServerID: {

render/render_test.go

@@ -36,11 +36,11 @@ func TestReduceRender(t *testing.T) {
 
 func TestReduceEdge(t *testing.T) {
 	renderer := render.Reduce([]render.Renderer{
-		mockRenderer{edgeMetadata: report.EdgeMetadata{PacketCount: newu64(1)}},
-		mockRenderer{edgeMetadata: report.EdgeMetadata{PacketCount: newu64(2)}},
+		mockRenderer{edgeMetadata: report.EdgeMetadata{EgressPacketCount: newu64(1)}},
+		mockRenderer{edgeMetadata: report.EdgeMetadata{EgressPacketCount: newu64(2)}},
 	})
 
-	want := report.EdgeMetadata{PacketCount: newu64(3)}
+	want := report.EdgeMetadata{EgressPacketCount: newu64(3)}
 	have := renderer.EdgeMetadata(report.MakeReport(), "", "")
 
 	if !reflect.DeepEqual(want, have) {
@@ -118,8 +118,8 @@ func TestMapEdge(t *testing.T) {
 				">bar": report.MakeIDList("foo"),
 			},
 			EdgeMetadatas: report.EdgeMetadatas{
-				"foo|bar": report.EdgeMetadata{PacketCount: newu64(1), EgressByteCount: newu64(2)},
-				"bar|foo": report.EdgeMetadata{PacketCount: newu64(3), EgressByteCount: newu64(4)},
+				"foo|bar": report.EdgeMetadata{EgressPacketCount: newu64(1), EgressByteCount: newu64(2)},
+				"bar|foo": report.EdgeMetadata{EgressPacketCount: newu64(3), EgressByteCount: newu64(4)},
 			},
 		}
 	}
@@ -140,8 +140,8 @@ func TestMapEdge(t *testing.T) {
 	}
 
 	if want, have := (report.EdgeMetadata{
-		PacketCount:     newu64(1),
-		EgressByteCount: newu64(2),
+		EgressPacketCount: newu64(1),
+		EgressByteCount:   newu64(2),
 	}), mapper.EdgeMetadata(report.MakeReport(), "_foo", "_bar"); !reflect.DeepEqual(want, have) {
 		t.Errorf("want %+v, have %+v", want, have)
 	}

report/merge.go

@@ -3,7 +3,8 @@ package report
 // Merge functions for all topology datatypes. The general semantics are that
 // the receiver is modified, and what's merged in isn't.
 
-// Merge merges another Report into the receiver.
+// Merge merges another Report into the receiver. Pass addWindows true if the
+// reports represent distinct (non-overlapping) periods of time.
 func (r *Report) Merge(other Report) {
 	r.Endpoint.Merge(other.Endpoint)
 	r.Address.Merge(other.Address)
@@ -63,7 +64,8 @@ func (e *EdgeMetadatas) Merge(other EdgeMetadatas) {
 // Merge merges another EdgeMetadata into the receiver. The two edge metadatas
 // should represent the same edge on different times.
 func (m *EdgeMetadata) Merge(other EdgeMetadata) {
-	m.PacketCount = merge(m.PacketCount, other.PacketCount, sum)
+	m.EgressPacketCount = merge(m.EgressPacketCount, other.EgressPacketCount, sum)
+	m.IngressPacketCount = merge(m.IngressPacketCount, other.IngressPacketCount, sum)
 	m.EgressByteCount = merge(m.EgressByteCount, other.EgressByteCount, sum)
 	m.IngressByteCount = merge(m.IngressByteCount, other.IngressByteCount, sum)
 	m.MaxConnCountTCP = merge(m.MaxConnCountTCP, other.MaxConnCountTCP, max)
@@ -72,7 +74,8 @@ func (m *EdgeMetadata) Merge(other EdgeMetadata) {
 // Flatten sums two EdgeMetadatas. Their windows should be the same duration;
 // they should represent different edges at the same time.
 func (m *EdgeMetadata) Flatten(other EdgeMetadata) {
-	m.PacketCount = merge(m.PacketCount, other.PacketCount, sum)
+	m.EgressPacketCount = merge(m.EgressPacketCount, other.EgressPacketCount, sum)
+	m.IngressPacketCount = merge(m.IngressPacketCount, other.IngressPacketCount, sum)
 	m.EgressByteCount = merge(m.EgressByteCount, other.EgressByteCount, sum)
 	m.IngressByteCount = merge(m.IngressByteCount, other.IngressByteCount, sum)
 	// Note that summing of two maximums doesn't always give us the true

report/merge_test.go

@@ -112,82 +112,82 @@ func TestMergeEdgeMetadatas(t *testing.T) {
 			a: report.EdgeMetadatas{},
 			b: report.EdgeMetadatas{
 				"hostA|:192.168.1.1:12345|:192.168.1.2:80": report.EdgeMetadata{
-					PacketCount:     newu64(1),
-					MaxConnCountTCP: newu64(2),
+					EgressPacketCount: newu64(1),
+					MaxConnCountTCP:   newu64(2),
 				},
 			},
 			want: report.EdgeMetadatas{
 				"hostA|:192.168.1.1:12345|:192.168.1.2:80": report.EdgeMetadata{
-					PacketCount:     newu64(1),
-					MaxConnCountTCP: newu64(2),
+					EgressPacketCount: newu64(1),
+					MaxConnCountTCP:   newu64(2),
 				},
 			},
 		},
 		"Empty b": {
 			a: report.EdgeMetadatas{
 				"hostA|:192.168.1.1:12345|:192.168.1.2:80": report.EdgeMetadata{
-					PacketCount:     newu64(12),
-					EgressByteCount: newu64(999),
+					EgressPacketCount: newu64(12),
+					EgressByteCount:   newu64(999),
 				},
 			},
 			b: report.EdgeMetadatas{},
 			want: report.EdgeMetadatas{
 				"hostA|:192.168.1.1:12345|:192.168.1.2:80": report.EdgeMetadata{
-					PacketCount:     newu64(12),
-					EgressByteCount: newu64(999),
+					EgressPacketCount: newu64(12),
+					EgressByteCount:   newu64(999),
 				},
 			},
 		},
 		"Host merge": {
 			a: report.EdgeMetadatas{
 				"hostA|:192.168.1.1:12345|:192.168.1.2:80": report.EdgeMetadata{
-					PacketCount:     newu64(12),
-					EgressByteCount: newu64(500),
-					MaxConnCountTCP: newu64(4),
+					EgressPacketCount: newu64(12),
+					EgressByteCount:   newu64(500),
+					MaxConnCountTCP:   newu64(4),
 				},
 			},
 			b: report.EdgeMetadatas{
 				"hostQ|:192.168.1.1:12345|:192.168.1.2:80": report.EdgeMetadata{
-					PacketCount:     newu64(1),
-					EgressByteCount: newu64(2),
-					MaxConnCountTCP: newu64(6),
+					EgressPacketCount: newu64(1),
+					EgressByteCount:   newu64(2),
+					MaxConnCountTCP:   newu64(6),
 				},
 			},
 			want: report.EdgeMetadatas{
 				"hostA|:192.168.1.1:12345|:192.168.1.2:80": report.EdgeMetadata{
-					PacketCount:     newu64(12),
-					EgressByteCount: newu64(500),
-					MaxConnCountTCP: newu64(4),
+					EgressPacketCount: newu64(12),
+					EgressByteCount:   newu64(500),
+					MaxConnCountTCP:   newu64(4),
 				},
 				"hostQ|:192.168.1.1:12345|:192.168.1.2:80": report.EdgeMetadata{
-					PacketCount:     newu64(1),
-					EgressByteCount: newu64(2),
-					MaxConnCountTCP: newu64(6),
+					EgressPacketCount: newu64(1),
+					EgressByteCount:   newu64(2),
+					MaxConnCountTCP:   newu64(6),
 				},
 			},
 		},
 		"Edge merge": {
 			a: report.EdgeMetadatas{
 				"hostA|:192.168.1.1:12345|:192.168.1.2:80": report.EdgeMetadata{
-					PacketCount:     newu64(12),
-					EgressByteCount: newu64(1000),
-					MaxConnCountTCP: newu64(7),
+					EgressPacketCount: newu64(12),
+					EgressByteCount:   newu64(1000),
+					MaxConnCountTCP:   newu64(7),
 				},
 			},
 			b: report.EdgeMetadatas{
 				"hostA|:192.168.1.1:12345|:192.168.1.2:80": report.EdgeMetadata{
-					PacketCount:      newu64(1),
-					IngressByteCount: newu64(123),
-					EgressByteCount:  newu64(2),
-					MaxConnCountTCP:  newu64(9),
+					EgressPacketCount: newu64(1),
+					IngressByteCount:  newu64(123),
+					EgressByteCount:   newu64(2),
+					MaxConnCountTCP:   newu64(9),
 				},
 			},
 			want: report.EdgeMetadatas{
 				"hostA|:192.168.1.1:12345|:192.168.1.2:80": report.EdgeMetadata{
-					PacketCount:      newu64(13),
-					IngressByteCount: newu64(123),
-					EgressByteCount:  newu64(1002),
-					MaxConnCountTCP:  newu64(9),
+					EgressPacketCount: newu64(13),
+					IngressByteCount:  newu64(123),
+					EgressByteCount:   newu64(1002),
+					MaxConnCountTCP:   newu64(9),
 				},
 			},
 		},

report/report.go

@@ -1,6 +1,9 @@
 package report
 
-import "fmt"
+import (
+	"fmt"
+	"strings"
+)
 
 // Report is the core data type. It's produced by probes, and consumed and
 // stored by apps. It's composed of multiple topologies, each representing
@@ -72,22 +75,17 @@ func (r Report) Topologies() []Topology {
 
 // Validate checks the report for various inconsistencies.
 func (r Report) Validate() error {
-	var packets uint64
+	var errs []string
 	for _, topology := range r.Topologies() {
 		if err := topology.Validate(); err != nil {
-			return err
-		}
-		for _, emd := range topology.EdgeMetadatas {
-			if emd.PacketCount != nil {
-				packets += *emd.PacketCount
-			}
+			errs = append(errs, err.Error())
 		}
 	}
 	if r.Sampling.Count > r.Sampling.Total {
-		return fmt.Errorf("sampling count (%d) bigger than total (%d)", r.Sampling.Count, r.Sampling.Total)
+		errs = append(errs, fmt.Sprintf("sampling count (%d) bigger than total (%d)", r.Sampling.Count, r.Sampling.Total))
 	}
-	if packets > 0 && (r.Sampling.Count == 0 || r.Sampling.Total == 0) {
-		return fmt.Errorf("packets exist in EdgeMetadata, but no sampling count or total in the base report")
+	if len(errs) > 0 {
+		return fmt.Errorf("%d error(s): %s", len(errs), strings.Join(errs, "; "))
 	}
 	return nil
 }

report/topology.go

@@ -31,10 +31,11 @@ type NodeMetadatas map[string]NodeMetadata
 // EdgeMetadata describes a superset of the metadata that probes can possibly
 // collect about a directed edge between two nodes in any topology.
 type EdgeMetadata struct {
-	PacketCount      *uint64 `json:"packet_count,omitempty"`
-	EgressByteCount  *uint64 `json:"egress_byte_count,omitempty"`
-	IngressByteCount *uint64 `json:"ingress_byte_count,omitempty"`
-	MaxConnCountTCP  *uint64 `json:"max_conn_count_tcp,omitempty"`
+	EgressPacketCount  *uint64 `json:"egress_packet_count,omitempty"`
+	IngressPacketCount *uint64 `json:"ingress_packet_count,omitempty"`
+	EgressByteCount    *uint64 `json:"egress_byte_count,omitempty"`  // Transport layer
+	IngressByteCount   *uint64 `json:"ingress_byte_count,omitempty"` // Transport layer
+	MaxConnCountTCP    *uint64 `json:"max_conn_count_tcp,omitempty"`
 }
 
 // NodeMetadata describes a superset of the metadata that probes can collect

test/report_fixture.go

@@ -108,33 +108,33 @@ var (
 			},
 			EdgeMetadatas: report.EdgeMetadatas{
 				report.MakeEdgeID(Client54001NodeID, Server80NodeID): report.EdgeMetadata{
-					PacketCount:     newu64(10),
-					EgressByteCount: newu64(100),
+					EgressPacketCount: newu64(10),
+					EgressByteCount:   newu64(100),
 				},
 				report.MakeEdgeID(Client54002NodeID, Server80NodeID): report.EdgeMetadata{
-					PacketCount:     newu64(20),
-					EgressByteCount: newu64(200),
+					EgressPacketCount: newu64(20),
+					EgressByteCount:   newu64(200),
 				},
 
 				report.MakeEdgeID(Server80NodeID, Client54001NodeID): report.EdgeMetadata{
-					PacketCount:     newu64(10),
-					EgressByteCount: newu64(100),
+					EgressPacketCount: newu64(10),
+					EgressByteCount:   newu64(100),
 				},
 				report.MakeEdgeID(Server80NodeID, Client54002NodeID): report.EdgeMetadata{
-					PacketCount:     newu64(20),
-					EgressByteCount: newu64(200),
+					EgressPacketCount: newu64(20),
+					EgressByteCount:   newu64(200),
 				},
 				report.MakeEdgeID(Server80NodeID, UnknownClient1NodeID): report.EdgeMetadata{
-					PacketCount:     newu64(30),
-					EgressByteCount: newu64(300),
+					EgressPacketCount: newu64(30),
+					EgressByteCount:   newu64(300),
 				},
 				report.MakeEdgeID(Server80NodeID, UnknownClient2NodeID): report.EdgeMetadata{
-					PacketCount:     newu64(40),
-					EgressByteCount: newu64(400),
+					EgressPacketCount: newu64(40),
+					EgressByteCount:   newu64(400),
 				},
 				report.MakeEdgeID(Server80NodeID, UnknownClient3NodeID): report.EdgeMetadata{
-					PacketCount:     newu64(50),
-					EgressByteCount: newu64(500),
+					EgressPacketCount: newu64(50),
+					EgressByteCount:   newu64(500),
 				},
 			},
 		},