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

@@ -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
 }

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 {

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 {