diff --git a/integration/321_many_connections_2_test.sh b/integration/321_many_connections_2_test.sh new file mode 100755 index 000000000..4e0a057a9 --- /dev/null +++ b/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 diff --git a/integration/config.sh b/integration/config.sh index 01b2eb5bc..d51ee9999 100644 --- a/integration/config.sh +++ b/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 diff --git a/probe/endpoint/connection_tracker.go b/probe/endpoint/connection_tracker.go index bc62476f1..bf0948932 100644 --- a/probe/endpoint/connection_tracker.go +++ b/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 diff --git a/probe/endpoint/connection_tracker_test.go b/probe/endpoint/connection_tracker_test.go new file mode 100644 index 000000000..1b1c82f31 --- /dev/null +++ b/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) + } + }) + } +} diff --git a/probe/endpoint/ebpf.go b/probe/endpoint/ebpf.go index 254ee39eb..1120b343f 100644 --- a/probe/endpoint/ebpf.go +++ b/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 { diff --git a/probe/endpoint/ebpf_test.go b/probe/endpoint/ebpf_test.go index d2b5d711a..f5daa688e 100644 --- a/probe/endpoint/ebpf_test.go +++ b/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 { diff --git a/render/detailed/connections.go b/render/detailed/connections.go index e7ac88fb5..dbf35b607 100644 --- a/render/detailed/connections.go +++ b/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) { diff --git a/report/map_keys.go b/report/map_keys.go index e8bfac3f8..e68c8dc94 100644 --- a/report/map_keys.go +++ b/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