Merge pull request #3709 from weaveworks/report-endpoint-subset

Report a subset of connections from/to the same endpoint

integration/321_many_connections_2_test.sh

@@ -0,0 +1,28 @@
+#! /bin/bash
+
+# shellcheck disable=SC1091
+. ./config.sh
+
+start_suite "Test short lived connections between containers on different hosts"
+
+weave_on "$HOST1" launch "$HOST1" "$HOST2"
+weave_on "$HOST2" launch "$HOST1" "$HOST2"
+
+scope_on "$HOST1" launch
+scope_on "$HOST2" launch
+
+server_on "$HOST1"
+client_on "$HOST2"
+
+sleep 30 # need to allow the scopes to poll dns, resolve the other app ids, and send them reports
+
+check() {
+    has_container "$1" nginx
+    has_container "$1" client
+    has_connection containers "$1" client nginx
+}
+
+check "$HOST1"
+check "$HOST2"
+
+scope_end_suite

integration/config.sh

@@ -109,15 +109,17 @@ has_connection_by_id() {
     local from_id="$3"
     local to_id="$4"
     local timeout="${5:-60}"
+    local max_edges="$6:10"
 
     for i in $(seq "$timeout"); do
         local nodes
         local edge
-        edge=$(echo "$nodes" | (jq -r ".nodes[\"$from_id\"].adjacency | contains([\"$to_id\"])" || true) 2>/dev/null)
         nodes=$(curl -s "http://$host:4040/api/topology/${view}?system=show" || true)
+        edge=$(echo "$nodes" | (jq -r ".nodes[\"$from_id\"].adjacency | contains([\"$to_id\"])" || true) 2>/dev/null)
         if [ "$edge" = "true" ]; then
             echo "Found edge $from -> $to after $i secs"
-            assert "curl -s http://$host:4040/api/topology/${view}?system=show |  jq -r '.nodes[\"$from_id\"].adjacency | contains([\"$to_id\"])'" true
+            count=$(echo "$nodes" | jq -r ".nodes[\"$from_id\"].adjacency | length" 2>/dev/null)
+            assert "[ $count -le $max_edges ]"
             return
         fi
         sleep 1

probe/endpoint/connection_tracker.go

@@ -100,7 +100,7 @@ func (t *connectionTracker) ReportConnections(rpt *report.Report) {
 	t.flowWalker.walkFlows(func(f conntrack.Conn, alive bool) {
 		tuple := flowToTuple(f)
 		seenTuples[tuple.key()] = tuple
-		t.addConnection(rpt, false, tuple, 0, nil, nil)
+		t.addConnection(rpt, "", tuple, 0, 0, 0, 1)
 	})
 
 	if t.conf.WalkProc && t.conf.Scanner != nil {
@@ -135,14 +135,11 @@ func (t *connectionTracker) performWalkProc(rpt *report.Report, hostNodeID strin
 	}
 	for conn := conns.Next(); conn != nil; conn = conns.Next() {
 		tuple, namespaceID, incoming := connectionTuple(conn, seenTuples)
-		var toNodeInfo, fromNodeInfo map[string]string
-		if conn.Proc.PID > 0 {
-			fromNodeInfo = map[string]string{
-				process.PID:       strconv.FormatUint(uint64(conn.Proc.PID), 10),
-				report.HostNodeID: hostNodeID,
-			}
+		if incoming {
+			t.addConnection(rpt, hostNodeID, reverse(tuple), 0, conn.Proc.PID, namespaceID, 1)
+		} else {
+			t.addConnection(rpt, hostNodeID, tuple, conn.Proc.PID, 0, namespaceID, 1)
 		}
-		t.addConnection(rpt, incoming, tuple, namespaceID, fromNodeInfo, toNodeInfo)
 	}
 	return nil
 }
@@ -178,24 +175,156 @@ func feedEBPFInitialState(conf ReporterConfig, ebpfTracker *EbpfTracker) {
 	ebpfTracker.feedInitialConnections(conns, seenTuples, processesWaitingInAccept, report.MakeHostNodeID(conf.HostID))
 }
 
+type pidPair struct {
+	fromPid uint32 // zero if unknown
+	toPid   uint32
+}
+type mapPortToPids map[uint16]pidPair
+
 func (t *connectionTracker) performEbpfTrack(rpt *report.Report, hostNodeID string) error {
-	t.ebpfTracker.walkConnections(func(e ebpfConnection) {
-		var toNodeInfo, fromNodeInfo map[string]string
-		if e.pid > 0 {
-			fromNodeInfo = map[string]string{
-				process.PID:       strconv.Itoa(e.pid),
-				report.HostNodeID: hostNodeID,
+	/* Collect the connections by from/to address pairs (scoped by namespace) plus destination port
+	   There are three main cases:
+		 * connections from address+port off-box to a local process
+		   - in this case we know the pid of the local process
+		 * connections from local processes to an off-box address+port
+		   - we will know the pids of the local processes but not the remote
+		 * connections from local processes to a local process
+		   - these connections will each be reported twice by ebpf, as incoming and as outgoing.
+	*/
+	type triple struct {
+		fromAddr, toAddr [net.IPv4len]byte
+		networkNamespace uint32
+		toPort           uint16
+	}
+	connectionsByTriple := make(map[triple]mapPortToPids, 1000)
+	t.ebpfTracker.walkConnections(func(key ebpfKey, e ebpfDetail) {
+		var t triple
+		var fromPort uint16
+		if e.incoming {
+			t = triple{
+				fromAddr:         key.toAddr,
+				toAddr:           key.fromAddr,
+				toPort:           key.fromPort,
+				networkNamespace: key.networkNamespace,
 			}
+			fromPort = key.toPort
+		} else {
+			t = triple{
+				fromAddr:         key.fromAddr,
+				toAddr:           key.toAddr,
+				toPort:           key.toPort,
+				networkNamespace: key.networkNamespace,
+			}
+			fromPort = key.fromPort
 		}
-		t.addConnection(rpt, e.incoming, e.tuple, e.networkNamespace, fromNodeInfo, toNodeInfo)
+		portToPids := connectionsByTriple[t]
+		if portToPids == nil {
+			portToPids = make(mapPortToPids)
+		}
+		pids := portToPids[fromPort]
+		if e.incoming {
+			pids.toPid = e.pid
+		} else {
+			pids.fromPid = e.pid
+		}
+		portToPids[fromPort] = pids
+		connectionsByTriple[t] = portToPids
 	})
+
+	for triple, portToPids := range connectionsByTriple {
+		filter, count := makeFilter(portToPids)
+		seen, sent, skipped := 0, 0, 0
+		// Now go over everything we collected, reporting connections if they pass the filter.
+		// With each connection is a count of how many it stands for.
+		for fromPort, pids := range portToPids {
+			seen++
+			if !filter(fromPort) {
+				skipped++
+				continue
+			}
+			tuple := fourTuple{
+				fromAddr: triple.fromAddr,
+				fromPort: fromPort,
+				toAddr:   triple.toAddr,
+				toPort:   triple.toPort,
+			}
+			sent++
+			if sent == count {
+				// Last one in a group: add in the connections that come after this one.
+				skipped += (len(portToPids) - seen)
+			}
+			t.addConnection(rpt, hostNodeID, tuple, uint(pids.fromPid), uint(pids.toPid), triple.networkNamespace, skipped+1)
+			skipped = 0
+		}
+	}
+
 	return nil
 }
 
-func (t *connectionTracker) addConnection(rpt *report.Report, incoming bool, ft fourTuple, namespaceID uint32, extraFromNode, extraToNode map[string]string) {
-	if incoming {
-		ft = reverse(ft)
-		extraFromNode, extraToNode = extraToNode, extraFromNode
+// Pick a subset of the connections to send, such that if two probes
+// on different machines go through the same process there is a good
+// chance of overlap.
+// return value is a function to filter from ports, and a count of how many will match
+func makeFilter(ports mapPortToPids) (filter func(uint16) bool, count int) {
+	var modulus uint16 = 1
+	count = len(ports)
+	// Check they all come from/to the same pid (or zero): if differing we need another strategy to thin them down
+	var firstToPid, firstFromPid uint32
+	for _, pids := range ports {
+		firstFromPid = pids.fromPid
+		firstToPid = pids.toPid
+		break
+	}
+	for _, pids := range ports {
+		if pids.fromPid != firstFromPid || pids.toPid != firstToPid {
+			return func(uint16) bool { return true }, count
+		}
+	}
+	const (
+		power      = 3 // Don't use powers of two to reduce aliasing with ephemeral port number selection.
+		lowerBound = 3
+		upperBound = 5
+	)
+	// Find modulus such that we choose at least the lower bound, and
+	// ideally no more than the upper bound
+	for count > upperBound {
+		modulus *= power
+		prevCount := count
+		// Count how many are sent for this modulus
+		count = 0
+		for fromPort := range ports {
+			if (fromPort % modulus) == 0 {
+				count++
+			}
+		}
+		if count < lowerBound { // too few: step back and stop there
+			modulus /= power
+			count = prevCount
+			break
+		}
+	}
+	return func(port uint16) bool { return (port % modulus) == 0 }, count
+}
+
+// tuple is canonicalised - always opened from-to
+func (t *connectionTracker) addConnection(rpt *report.Report, hostNodeID string, ft fourTuple, fromPid, toPid uint, namespaceID uint32, connectionCount int) {
+	extraToNode := map[string]string{}
+	extraFromNode := map[string]string{}
+	if fromPid > 0 {
+		extraFromNode = map[string]string{
+			process.PID:       strconv.FormatUint(uint64(fromPid), 10),
+			report.HostNodeID: hostNodeID,
+		}
+	}
+	if toPid > 0 {
+		extraToNode = map[string]string{
+			process.PID:       strconv.FormatUint(uint64(toPid), 10),
+			report.HostNodeID: hostNodeID,
+		}
+	}
+	if connectionCount > 1 {
+		// Tell the app we have elided several connections to a common IP and port onto this one
+		extraFromNode[report.ConnectionCount] = strconv.Itoa(connectionCount)
 	}
 	var (
 		fromAddr = net.IP(ft.fromAddr[:])
@@ -211,7 +340,7 @@ func (t *connectionTracker) addConnection(rpt *report.Report, incoming bool, ft
 
 func (t *connectionTracker) makeEndpointNode(namespaceID uint32, addr net.IP, port uint16, extra map[string]string) report.Node {
 	node := report.MakeNodeWith(report.MakeEndpointNodeIDB(t.conf.HostID, namespaceID, addr, port), nil)
-	if extra != nil {
+	if len(extra) > 0 {
 		node = node.WithLatests(extra)
 	}
 	return node

probe/endpoint/connection_tracker_test.go

@@ -0,0 +1,112 @@
+package endpoint
+
+import (
+	"testing"
+)
+
+// mapPortToPids collects info about connections between specific
+// address pairs and destination port.
+
+// a set of connections from a single pid to a destination off-box (pid is zero)
+func fakeConnectionsOffBox(count int, fromPid uint32, startPort uint16) mapPortToPids {
+	ret := make(mapPortToPids, count)
+	for i := 0; i < count; i++ {
+		ret[uint16(i)+startPort] = pidPair{fromPid: fromPid, toPid: 0}
+	}
+	return ret
+}
+
+// a set of connections to a single pid from a destination off-box (pid is zero)
+func fakeConnectionsFromOffBox(count int, toPid uint32, startPort uint16) mapPortToPids {
+	ret := make(mapPortToPids, count)
+	for i := 0; i < count; i++ {
+		ret[uint16(i)+startPort] = pidPair{fromPid: 0, toPid: toPid}
+	}
+	return ret
+}
+
+// a set of connections from a range of pids and a range of ports between two pids
+func fakeConnections(count int, fromPid, toPid uint32, startPort uint16) mapPortToPids {
+	ret := make(mapPortToPids, count)
+	for i := 0; i < count; i++ {
+		ret[uint16(i)+startPort] = pidPair{fromPid: fromPid, toPid: toPid}
+	}
+	return ret
+}
+
+// union N existing sets
+func concatMapPortToPids(maps ...mapPortToPids) mapPortToPids {
+	ret := make(mapPortToPids)
+	for _, m := range maps {
+		for port, pair := range m {
+			ret[port] = pair
+		}
+	}
+	return ret
+}
+
+func TestConnectionThinning(t *testing.T) {
+	for _, d := range []struct {
+		name string
+		m    mapPortToPids
+		expC int
+	}{
+		{
+			name: "0 ports",
+			m:    mapPortToPids{},
+			expC: 0,
+		},
+		{
+			name: "5 connections off-box",
+			m:    fakeConnectionsOffBox(5, 1000, 30000),
+			expC: 5, // 5 is too few to thin down
+		},
+		{
+			name: "50 connections off-box",
+			m:    fakeConnectionsOffBox(50, 1000, 30000),
+			expC: 5, // 50 connections should be thinned down
+		},
+		{
+			name: "50 connections from off-box",
+			m:    fakeConnectionsFromOffBox(50, 1000, 30000),
+			expC: 5, // 50 connections should be thinned down
+		},
+		{
+			name: "5 connections from pid 1000 to 2000",
+			m:    fakeConnections(5, 1000, 2000, 30000),
+			expC: 5, // 5 is too few to thin down
+		},
+		{
+			name: "50 connections from pid 1000 to 2000",
+			m:    fakeConnections(50, 1000, 2000, 30000),
+			expC: 5, // 50 connections should be thinned down
+		},
+		{
+			name: "connections both ways",
+			m: concatMapPortToPids(
+				fakeConnections(50, 1000, 2000, 30000),
+				fakeConnections(50, 2000, 1000, 40000)),
+			expC: 100, // no thinning because in both directions
+		},
+		{
+			name: "connections to two different pids",
+			m: concatMapPortToPids(
+				fakeConnections(50, 1000, 2000, 30000),
+				fakeConnections(50, 1000, 3000, 40000)),
+			expC: 100, // no thinning because two different pids
+		},
+		{
+			name: "connections from off-box and on-box",
+			m:    concatMapPortToPids(fakeConnectionsFromOffBox(50, 1000, 30000), fakeConnections(50, 2000, 1000, 40000)),
+			expC: 100, // no thinning because pid and no-pid
+		},
+	} {
+		t.Run(d.name, func(t *testing.T) {
+			filter, count := makeFilter(d.m)
+			_ = filter
+			if d.expC != count {
+				t.Errorf("expected count %d, got %d", d.expC, count)
+			}
+		})
+	}
+}

probe/endpoint/ebpf.go

@@ -6,6 +6,7 @@ import (
 	"bytes"
 	"fmt"
 	"io/ioutil"
+	"net"
 	"os"
 	"regexp"
 	"strconv"
@@ -22,12 +23,31 @@ import (
 	"github.com/weaveworks/tcptracer-bpf/pkg/tracer"
 )
 
-// An ebpfConnection represents a TCP connection
-type ebpfConnection struct {
-	tuple            fourTuple
+// Open connections are held by four-tuple, and loopback addresses (e.g. 127.0.0.1)
+// are scoped by network namespace.  We only need one namespace: either
+// just one from or to address is loopback, or both are in the same namespace.
+type ebpfKey struct {
+	fourTuple
 	networkNamespace uint32
-	incoming         bool
-	pid              int
+}
+
+func makeKey(tuple fourTuple, namespace uint32) ebpfKey {
+	ret := ebpfKey{fourTuple: tuple}
+	if net.IP(tuple.fromAddr[:]).IsLoopback() || net.IP(tuple.toAddr[:]).IsLoopback() {
+		ret.networkNamespace = namespace
+	}
+	return ret
+}
+
+// For each connection we also record which direction it was opened in, and the pid of the 'from' end.
+type ebpfDetail struct {
+	incoming bool
+	pid      uint32 // zero if unknown
+}
+
+type ebpfClosedConnection struct {
+	key ebpfKey
+	ebpfDetail
 }
 
 // EbpfTracker contains the sets of open and closed TCP connections.
@@ -51,9 +71,9 @@ type EbpfTracker struct {
 	//   $ echo stop | sudo tee /proc/$(pidof scope-probe)/root/var/run/scope/debug-bpf
 	debugBPF bool
 
-	openConnections   map[fourTuple]ebpfConnection
-	closedConnections []ebpfConnection
-	closedDuringInit  map[fourTuple]struct{}
+	openConnections   map[ebpfKey]ebpfDetail
+	closedConnections []ebpfClosedConnection
+	closedDuringInit  map[ebpfKey]struct{}
 }
 
 // releaseRegex should match all possible variations of a common Linux
@@ -247,6 +267,9 @@ func tupleFromPidFd(pid int, fd int) (tuple fourTuple, netns uint32, ok bool) {
 	return fourTuple{}, 0, false
 }
 
+// this callback exists to close a hole whereby we don't get a kprobe
+// for tcp_accept if accept was called before the probe started.
+// It's fairly safe to assume all such connections are incoming, but not 100%
 func (t *EbpfTracker) handleFdInstall(ev tracer.EventType, pid int, fd int) {
 	if !process.IsProcInAccept("/proc", strconv.Itoa(pid)) {
 		t.tracer.RemoveFdInstallWatcher(uint32(pid))
@@ -260,11 +283,9 @@ func (t *EbpfTracker) handleFdInstall(ev tracer.EventType, pid int, fd int) {
 	t.Lock()
 	defer t.Unlock()
 
-	t.openConnections[tuple] = ebpfConnection{
-		incoming:         true,
-		tuple:            tuple,
-		pid:              pid,
-		networkNamespace: netns,
+	t.openConnections[makeKey(tuple, netns)] = ebpfDetail{
+		incoming: true,
+		pid:      uint32(pid),
 	}
 }
 
@@ -275,28 +296,25 @@ func (t *EbpfTracker) handleConnection(ev tracer.EventType, tuple fourTuple, pid
 	log.Debugf("handleConnection(%v, [%v:%v --> %v:%v], pid=%v, netNS=%v)",
 		ev, tuple.fromAddr, tuple.fromPort, tuple.toAddr, tuple.toPort, pid, networkNamespace)
 
+	key := makeKey(tuple, networkNamespace)
 	switch ev {
 	case tracer.EventConnect:
-		t.openConnections[tuple] = ebpfConnection{
-			incoming:         false,
-			tuple:            tuple,
-			pid:              pid,
-			networkNamespace: networkNamespace,
+		t.openConnections[key] = ebpfDetail{
+			incoming: false,
+			pid:      uint32(pid),
 		}
 	case tracer.EventAccept:
-		t.openConnections[tuple] = ebpfConnection{
-			incoming:         true,
-			tuple:            tuple,
-			pid:              pid,
-			networkNamespace: networkNamespace,
+		t.openConnections[key] = ebpfDetail{
+			incoming: true,
+			pid:      uint32(pid),
 		}
 	case tracer.EventClose:
 		if !t.ready {
-			t.closedDuringInit[tuple] = struct{}{}
+			t.closedDuringInit[key] = struct{}{}
 		}
-		if deadConn, ok := t.openConnections[tuple]; ok {
-			delete(t.openConnections, tuple)
-			t.closedConnections = append(t.closedConnections, deadConn)
+		if deadConn, ok := t.openConnections[key]; ok {
+			delete(t.openConnections, key)
+			t.closedConnections = append(t.closedConnections, ebpfClosedConnection{key: key, ebpfDetail: deadConn})
 		} else {
 			log.Debugf("EbpfTracker: unmatched close event: %s pid=%d netns=%v", tuple, pid, networkNamespace)
 		}
@@ -307,15 +325,15 @@ func (t *EbpfTracker) handleConnection(ev tracer.EventType, tuple fourTuple, pid
 
 // walkConnections calls f with all open connections and connections that have come and gone
 // since the last call to walkConnections
-func (t *EbpfTracker) walkConnections(f func(ebpfConnection)) {
+func (t *EbpfTracker) walkConnections(f func(ebpfKey, ebpfDetail)) {
 	t.Lock()
 	defer t.Unlock()
 
-	for _, connection := range t.openConnections {
-		f(connection)
+	for tuple, detail := range t.openConnections {
+		f(tuple, detail)
 	}
 	for _, connection := range t.closedConnections {
-		f(connection)
+		f(connection.key, connection.ebpfDetail)
 	}
 	t.closedConnections = t.closedConnections[:0]
 }
@@ -323,14 +341,18 @@ func (t *EbpfTracker) walkConnections(f func(ebpfConnection)) {
 func (t *EbpfTracker) feedInitialConnections(conns procspy.ConnIter, seenTuples map[string]fourTuple, processesWaitingInAccept []int, hostNodeID string) {
 	t.Lock()
 	for conn := conns.Next(); conn != nil; conn = conns.Next() {
+		if conn.Proc.PID == 0 {
+			continue // no point in tracking a connection which we can't associate to a process
+		}
 		tuple, namespaceID, incoming := connectionTuple(conn, seenTuples)
-		if _, ok := t.closedDuringInit[tuple]; !ok {
-			if _, ok := t.openConnections[tuple]; !ok {
-				t.openConnections[tuple] = ebpfConnection{
-					incoming:         incoming,
-					tuple:            tuple,
-					pid:              int(conn.Proc.PID),
-					networkNamespace: namespaceID,
+		key := makeKey(tuple, namespaceID)
+		if _, ok := t.closedDuringInit[key]; !ok {
+			if _, ok := t.openConnections[key]; !ok {
+				log.Debugf("initialConnection([%v], in=%v, pid=%v, netNS=%v)",
+					tuple, incoming, conn.Proc.PID, namespaceID)
+				t.openConnections[key] = ebpfDetail{
+					incoming: incoming,
+					pid:      uint32(conn.Proc.PID),
 				}
 			}
 		}
@@ -381,9 +403,9 @@ func (t *EbpfTracker) restart() error {
 	t.dead = false
 	t.ready = false
 
-	t.openConnections = map[fourTuple]ebpfConnection{}
-	t.closedDuringInit = map[fourTuple]struct{}{}
-	t.closedConnections = []ebpfConnection{}
+	t.openConnections = map[ebpfKey]ebpfDetail{}
+	t.closedDuringInit = map[ebpfKey]struct{}{}
+	t.closedConnections = []ebpfClosedConnection{}
 
 	tracer, err := tracer.NewTracer(t)
 	if err != nil {

probe/endpoint/ebpf_test.go

@@ -18,7 +18,7 @@ func newMockEbpfTracker() *EbpfTracker {
 		ready: true,
 		dead:  false,
 
-		openConnections: map[fourTuple]ebpfConnection{},
+		openConnections: map[ebpfKey]ebpfDetail{},
 	}
 }
 
@@ -46,16 +46,9 @@ func TestHandleConnection(t *testing.T) {
 			NetNS: NetNS,
 		}
 
-		IPv4ConnectEbpfConnection = ebpfConnection{
-			tuple: fourTuple{
-				fromAddr: ClientAddr,
-				toAddr:   ServerAddr,
-				fromPort: ClientPort,
-				toPort:   ServerPort,
-			},
-			networkNamespace: NetNS,
-			incoming:         false,
-			pid:              int(ClientPid),
+		IPv4ConnectEbpfConnection = ebpfDetail{
+			incoming: false,
+			pid:      ClientPid,
 		}
 
 		IPv4ConnectCloseEvent = tracer.TcpV4{
@@ -82,16 +75,9 @@ func TestHandleConnection(t *testing.T) {
 			NetNS: NetNS,
 		}
 
-		IPv4AcceptEbpfConnection = ebpfConnection{
-			tuple: fourTuple{
-				fromAddr: ServerAddr,
-				toAddr:   ClientAddr,
-				fromPort: ServerPort,
-				toPort:   ClientPort,
-			},
-			networkNamespace: NetNS,
-			incoming:         true,
-			pid:              int(ServerPid),
+		IPv4AcceptEbpfConnection = ebpfDetail{
+			incoming: true,
+			pid:      ServerPid,
 		}
 
 		IPv4AcceptCloseEvent = tracer.TcpV4{
@@ -109,27 +95,30 @@ func TestHandleConnection(t *testing.T) {
 
 	mockEbpfTracker := newMockEbpfTracker()
 
-	tuple := fourTuple{ClientAddr, ServerAddr, uint16(IPv4ConnectEvent.SPort), uint16(IPv4ConnectEvent.DPort)}
+	tuple := fourTuple{ClientAddr, ServerAddr, IPv4ConnectEvent.SPort, IPv4ConnectEvent.DPort}
 	mockEbpfTracker.handleConnection(IPv4ConnectEvent.Type, tuple, int(IPv4ConnectEvent.Pid), NetNS)
-	if !reflect.DeepEqual(mockEbpfTracker.openConnections[tuple], IPv4ConnectEbpfConnection) {
+	key := makeKey(tuple, NetNS)
+	if !reflect.DeepEqual(mockEbpfTracker.openConnections[key], IPv4ConnectEbpfConnection) {
 		t.Errorf("Connection mismatch connect event\nTarget connection:%v\nParsed connection:%v",
-			IPv4ConnectEbpfConnection, mockEbpfTracker.openConnections[tuple])
+			IPv4ConnectEbpfConnection, mockEbpfTracker.openConnections[key])
 	}
 
 	tuple = fourTuple{ClientAddr, ServerAddr, uint16(IPv4ConnectCloseEvent.SPort), uint16(IPv4ConnectCloseEvent.DPort)}
 	mockEbpfTracker.handleConnection(IPv4ConnectCloseEvent.Type, tuple, int(IPv4ConnectCloseEvent.Pid), NetNS)
+	key = makeKey(tuple, NetNS)
 	if len(mockEbpfTracker.openConnections) != 0 {
 		t.Errorf("Connection mismatch close event\nConnection to close:%v",
-			mockEbpfTracker.openConnections[tuple])
+			mockEbpfTracker.openConnections[key])
 	}
 
 	mockEbpfTracker = newMockEbpfTracker()
 
 	tuple = fourTuple{ServerAddr, ClientAddr, uint16(IPv4AcceptEvent.SPort), uint16(IPv4AcceptEvent.DPort)}
 	mockEbpfTracker.handleConnection(IPv4AcceptEvent.Type, tuple, int(IPv4AcceptEvent.Pid), NetNS)
-	if !reflect.DeepEqual(mockEbpfTracker.openConnections[tuple], IPv4AcceptEbpfConnection) {
+	key = makeKey(tuple, NetNS)
+	if !reflect.DeepEqual(mockEbpfTracker.openConnections[key], IPv4AcceptEbpfConnection) {
 		t.Errorf("Connection mismatch connect event\nTarget connection:%v\nParsed connection:%v",
-			IPv4AcceptEbpfConnection, mockEbpfTracker.openConnections[tuple])
+			IPv4AcceptEbpfConnection, mockEbpfTracker.openConnections[key])
 	}
 
 	tuple = fourTuple{ServerAddr, ClientAddr, uint16(IPv4AcceptCloseEvent.SPort), uint16(IPv4AcceptCloseEvent.DPort)}
@@ -143,25 +132,22 @@ func TestHandleConnection(t *testing.T) {
 
 func TestWalkConnections(t *testing.T) {
 	var (
-		cnt           int
-		activeTuple   = fourTuple{}
-		inactiveTuple = fourTuple{}
+		cnt         int
+		activeKey   = ebpfKey{networkNamespace: 12345}
+		inactiveKey = ebpfKey{networkNamespace: 12345}
 	)
 	mockEbpfTracker := newMockEbpfTracker()
-	mockEbpfTracker.openConnections[activeTuple] = ebpfConnection{
-		tuple:            activeTuple,
-		networkNamespace: 12345,
-		incoming:         true,
-		pid:              0,
+	mockEbpfTracker.openConnections[activeKey] = ebpfDetail{
+		pid: 0,
 	}
 	mockEbpfTracker.closedConnections = append(mockEbpfTracker.closedConnections,
-		ebpfConnection{
-			tuple:            inactiveTuple,
-			networkNamespace: 12345,
-			incoming:         false,
-			pid:              0,
+		ebpfClosedConnection{
+			key: inactiveKey,
+			ebpfDetail: ebpfDetail{
+				pid: 0,
+			},
 		})
-	mockEbpfTracker.walkConnections(func(e ebpfConnection) {
+	mockEbpfTracker.walkConnections(func(k ebpfKey, e ebpfDetail) {
 		cnt++
 	})
 	if cnt != 2 {
@@ -197,7 +183,7 @@ func TestInvalidTimeStampDead(t *testing.T) {
 	event2.SPort = 1
 	event2.Timestamp = 2
 	mockEbpfTracker.TCPEventV4(event2)
-	mockEbpfTracker.walkConnections(func(e ebpfConnection) {
+	mockEbpfTracker.walkConnections(func(ebpfKey, ebpfDetail) {
 		cnt++
 	})
 	if cnt != 2 {
@@ -209,7 +195,7 @@ func TestInvalidTimeStampDead(t *testing.T) {
 	cnt = 0
 	event.Timestamp = 1
 	mockEbpfTracker.TCPEventV4(event)
-	mockEbpfTracker.walkConnections(func(e ebpfConnection) {
+	mockEbpfTracker.walkConnections(func(ebpfKey, ebpfDetail) {
 		cnt++
 	})
 	if cnt != 2 {

render/detailed/connections.go

@@ -101,8 +101,15 @@ func (c *connectionCounters) add(dns report.DNSRecords, outgoing bool, localNode
 		return
 	}
 
+	count := 1
+	if countStr, _, ok := srcEndpoint.Latest.LookupEntry(report.ConnectionCount); ok {
+		if i, err := strconv.Atoi(countStr); err == nil {
+			count = i
+		}
+	}
+
 	c.counted[connectionID] = struct{}{}
-	c.counts[conn]++
+	c.counts[conn] += count
 }
 
 func internetAddr(dns report.DNSRecords, node report.Node, ep report.Node) (string, bool) {

report/map_keys.go

@@ -8,6 +8,8 @@ const (
 	ReverseDNSNames = "reverse_dns_names"
 	SnoopedDNSNames = "snooped_dns_names"
 	CopyOf          = "copy_of"
+	ConnectionCount = "conn_count"
+
 	// probe/process
 	PID     = "pid"
 	Name    = "name" // also used by probe/docker