From a6da810261fbec0544e062138f75b1e091602daf Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Sun, 22 Sep 2019 10:37:29 +0000 Subject: [PATCH 01/10] refactor(probe): move host/pid encoding into addConnection() function --- probe/endpoint/connection_tracker.go | 29 +++++++++++----------------- 1 file changed, 11 insertions(+), 18 deletions(-) diff --git a/probe/endpoint/connection_tracker.go b/probe/endpoint/connection_tracker.go index bc62476f1..df42a3608 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, "", 0, false, tuple, 0) }) if t.conf.WalkProc && t.conf.Scanner != nil { @@ -135,14 +135,7 @@ 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, - } - } - t.addConnection(rpt, incoming, tuple, namespaceID, fromNodeInfo, toNodeInfo) + t.addConnection(rpt, hostNodeID, conn.Proc.PID, incoming, tuple, namespaceID) } return nil } @@ -180,19 +173,19 @@ func feedEBPFInitialState(conf ReporterConfig, ebpfTracker *EbpfTracker) { 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, - } - } - t.addConnection(rpt, e.incoming, e.tuple, e.networkNamespace, fromNodeInfo, toNodeInfo) + t.addConnection(rpt, hostNodeID, uint(e.pid), e.incoming, e.tuple, e.networkNamespace) }) return nil } -func (t *connectionTracker) addConnection(rpt *report.Report, incoming bool, ft fourTuple, namespaceID uint32, extraFromNode, extraToNode map[string]string) { +func (t *connectionTracker) addConnection(rpt *report.Report, hostNodeID string, pid uint, incoming bool, ft fourTuple, namespaceID uint32) { + var extraToNode, extraFromNode map[string]string + if pid > 0 { + extraFromNode = map[string]string{ + process.PID: strconv.FormatUint(uint64(pid), 10), + report.HostNodeID: hostNodeID, + } + } if incoming { ft = reverse(ft) extraFromNode, extraToNode = extraToNode, extraFromNode From 2767e2b319a13f23b3146208a14aa25e8d3c352a Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Tue, 8 Oct 2019 07:49:17 +0000 Subject: [PATCH 02/10] improvement: only record connections that we have a PID for --- probe/endpoint/ebpf.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/probe/endpoint/ebpf.go b/probe/endpoint/ebpf.go index 254ee39eb..b35674cd1 100644 --- a/probe/endpoint/ebpf.go +++ b/probe/endpoint/ebpf.go @@ -323,6 +323,9 @@ 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 { From d57a4df3b26f09fd099651448b5a7160485438e2 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Thu, 10 Oct 2019 08:05:11 +0000 Subject: [PATCH 03/10] enhancement(probe): debug message for initial connection --- probe/endpoint/ebpf.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/probe/endpoint/ebpf.go b/probe/endpoint/ebpf.go index b35674cd1..744b2cb4c 100644 --- a/probe/endpoint/ebpf.go +++ b/probe/endpoint/ebpf.go @@ -329,6 +329,8 @@ func (t *EbpfTracker) feedInitialConnections(conns procspy.ConnIter, seenTuples tuple, namespaceID, incoming := connectionTuple(conn, seenTuples) if _, ok := t.closedDuringInit[tuple]; !ok { if _, ok := t.openConnections[tuple]; !ok { + log.Debugf("initialConnection([%v], in=%v, pid=%v, netNS=%v)", + tuple, incoming, conn.Proc.PID, namespaceID) t.openConnections[tuple] = ebpfConnection{ incoming: incoming, tuple: tuple, From 9758c8173625db81aa4acf9c99386850156ac496 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Thu, 10 Oct 2019 16:45:24 +0000 Subject: [PATCH 04/10] comment: add explanatory comment on handleFdInstall() --- probe/endpoint/ebpf.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/probe/endpoint/ebpf.go b/probe/endpoint/ebpf.go index 744b2cb4c..93230dc60 100644 --- a/probe/endpoint/ebpf.go +++ b/probe/endpoint/ebpf.go @@ -247,6 +247,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)) From fc46ea17eea1b5af4c58cab6f8358b5d01e405b4 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Thu, 10 Oct 2019 16:46:22 +0000 Subject: [PATCH 05/10] refactor(probe/ebpf): track connections by four-tuple+namespace The previous code tracked only by four-tuple, which meant that two connections with same address/port combinations in different namespace would clash and one would get dropped. Also previously the tuple was duplicated between the map key and value, so we remove it from the value. We only add the namespace in the case that the local address is loopback, which matches how the rest of Scope treats addresses. --- probe/endpoint/connection_tracker.go | 4 +- probe/endpoint/ebpf.go | 96 ++++++++++++++++------------ probe/endpoint/ebpf_test.go | 72 +++++++++------------ 3 files changed, 86 insertions(+), 86 deletions(-) diff --git a/probe/endpoint/connection_tracker.go b/probe/endpoint/connection_tracker.go index df42a3608..4622cb408 100644 --- a/probe/endpoint/connection_tracker.go +++ b/probe/endpoint/connection_tracker.go @@ -172,8 +172,8 @@ func feedEBPFInitialState(conf ReporterConfig, ebpfTracker *EbpfTracker) { } func (t *connectionTracker) performEbpfTrack(rpt *report.Report, hostNodeID string) error { - t.ebpfTracker.walkConnections(func(e ebpfConnection) { - t.addConnection(rpt, hostNodeID, uint(e.pid), e.incoming, e.tuple, e.networkNamespace) + t.ebpfTracker.walkConnections(func(key ebpfKey, e ebpfDetail) { + t.addConnection(rpt, hostNodeID, uint(e.pid), e.incoming, key.fourTuple, key.networkNamespace) }) return nil } diff --git a/probe/endpoint/ebpf.go b/probe/endpoint/ebpf.go index 93230dc60..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 @@ -263,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), } } @@ -278,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) } @@ -310,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] } @@ -330,15 +345,14 @@ func (t *EbpfTracker) feedInitialConnections(conns procspy.ConnIter, seenTuples 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 { + 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[tuple] = ebpfConnection{ - incoming: incoming, - tuple: tuple, - pid: int(conn.Proc.PID), - networkNamespace: namespaceID, + t.openConnections[key] = ebpfDetail{ + incoming: incoming, + pid: uint32(conn.Proc.PID), } } } @@ -389,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 { From de3c34ddc6a764467db33f39635a76fc5cc05edc Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Sun, 22 Sep 2019 18:38:10 +0000 Subject: [PATCH 06/10] performance(probe): thin out many connections between the same point The app will only show one line, regardless of how many connections we have, so reduce the number to save bandwidth and rendering time. We filter by choosing a modulus, e.g. send every connection that is a multiple of 3, or 9, and so on. We avoid multiples of 2 because port numbers are often a multiple of 2 or 4 for bit-encoding reasons. --- probe/endpoint/connection_tracker.go | 156 +++++++++++++++++++++++++-- report/map_keys.go | 2 + 2 files changed, 147 insertions(+), 11 deletions(-) diff --git a/probe/endpoint/connection_tracker.go b/probe/endpoint/connection_tracker.go index 4622cb408..51cbad1d2 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, "", 0, false, tuple, 0) + t.addConnection(rpt, "", tuple, 0, 0, 0, 1) }) if t.conf.WalkProc && t.conf.Scanner != nil { @@ -135,7 +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) - t.addConnection(rpt, hostNodeID, conn.Proc.PID, incoming, tuple, namespaceID) + 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) + } } return nil } @@ -171,24 +175,154 @@ 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 { + /* 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) { - t.addConnection(rpt, hostNodeID, uint(e.pid), e.incoming, key.fourTuple, key.networkNamespace) + 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 + } + 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) + i, sent, skipped := 0, 0, 0 + // Now do the actual sends + for fromPort, pids := range portToPids { + if filter(fromPort) { + tuple := fourTuple{ + fromAddr: triple.fromAddr, + fromPort: fromPort, + toAddr: triple.toAddr, + toPort: triple.toPort, + } + sent++ + if sent == count { + skipped += (len(portToPids) - i - 1) + } + t.addConnection(rpt, hostNodeID, tuple, uint(pids.fromPid), uint(pids.toPid), triple.networkNamespace, skipped+1) + skipped = 0 + } else { + skipped++ + } + i++ + } + } + return nil } -func (t *connectionTracker) addConnection(rpt *report.Report, hostNodeID string, pid uint, incoming bool, ft fourTuple, namespaceID uint32) { - var extraToNode, extraFromNode map[string]string - if pid > 0 { +// 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(pid), 10), + process.PID: strconv.FormatUint(uint64(fromPid), 10), report.HostNodeID: hostNodeID, } } - if incoming { - ft = reverse(ft) - extraFromNode, extraToNode = extraToNode, extraFromNode + 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[:]) @@ -204,7 +338,7 @@ func (t *connectionTracker) addConnection(rpt *report.Report, hostNodeID string, 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/report/map_keys.go b/report/map_keys.go index 388f39dc2..aa7436086 100644 --- a/report/map_keys.go +++ b/report/map_keys.go @@ -6,6 +6,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 From 4c9cf138d6f4b7c92b8f49117798fad6be6e4bd2 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Tue, 24 Sep 2019 18:21:01 +0000 Subject: [PATCH 07/10] ui: include skipped connection count in rendering --- render/detailed/connections.go | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) 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) { From b7b245ed48c18d881f7d250277ea55ad874870bf Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Tue, 7 Jan 2020 09:04:35 +0000 Subject: [PATCH 08/10] tests: connection subset testing Utility functions to create fake sets of connections for testing, and then exercising the subset filtering code to check that quantities come out as expected. --- probe/endpoint/connection_tracker_test.go | 112 ++++++++++++++++++++++ 1 file changed, 112 insertions(+) create mode 100644 probe/endpoint/connection_tracker_test.go 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) + } + }) + } +} From de939cc39b1bb66ff5be18ff4765dd3e93c30710 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Thu, 24 Oct 2019 18:55:01 +0000 Subject: [PATCH 09/10] tests: add an integration test to check connection thinning --- integration/321_many_connections_2_test.sh | 28 ++++++++++++++++++++++ integration/config.sh | 6 +++-- 2 files changed, 32 insertions(+), 2 deletions(-) create mode 100755 integration/321_many_connections_2_test.sh 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 From 7dc7215a264fc8fa65e8ae9811d0502cc442fc9e Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Thu, 23 Jan 2020 15:04:51 +0000 Subject: [PATCH 10/10] Refactor: improve readability based on review feedback --- probe/endpoint/connection_tracker.go | 48 +++++++++++++++------------- 1 file changed, 25 insertions(+), 23 deletions(-) diff --git a/probe/endpoint/connection_tracker.go b/probe/endpoint/connection_tracker.go index 51cbad1d2..bf0948932 100644 --- a/probe/endpoint/connection_tracker.go +++ b/probe/endpoint/connection_tracker.go @@ -184,12 +184,12 @@ type mapPortToPids map[uint16]pidPair func (t *connectionTracker) performEbpfTrack(rpt *report.Report, hostNodeID string) error { /* 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. + * 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 @@ -233,26 +233,28 @@ func (t *connectionTracker) performEbpfTrack(rpt *report.Report, hostNodeID stri for triple, portToPids := range connectionsByTriple { filter, count := makeFilter(portToPids) - i, sent, skipped := 0, 0, 0 - // Now do the actual sends + 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 { - if filter(fromPort) { - tuple := fourTuple{ - fromAddr: triple.fromAddr, - fromPort: fromPort, - toAddr: triple.toAddr, - toPort: triple.toPort, - } - sent++ - if sent == count { - skipped += (len(portToPids) - i - 1) - } - t.addConnection(rpt, hostNodeID, tuple, uint(pids.fromPid), uint(pids.toPid), triple.networkNamespace, skipped+1) - skipped = 0 - } else { + seen++ + if !filter(fromPort) { skipped++ + continue } - i++ + 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 } }