diff --git a/integration/330_process_edge_test.sh b/integration/330_process_edge_test.sh index e3f3ef803..7fadbe94a 100755 --- a/integration/330_process_edge_test.sh +++ b/integration/330_process_edge_test.sh @@ -3,10 +3,10 @@ # shellcheck disable=SC1091 . ./config.sh -start_suite "Test long connections (procspy) between processes" +start_suite "Test long connections between processes" weave_on "$HOST1" launch -scope_on "$HOST1" launch --probe.ebpf.connections=false --probe.conntrack=false +scope_on "$HOST1" launch server_on "$HOST1" weave_proxy_on "$HOST1" run -dti --name client alpine /bin/sh -c "while true; do \ diff --git a/integration/331_process_edge_without_ebpf_test.sh b/integration/331_process_edge_without_ebpf_test.sh new file mode 100755 index 000000000..712820d4b --- /dev/null +++ b/integration/331_process_edge_without_ebpf_test.sh @@ -0,0 +1,23 @@ +#! /bin/bash + +# shellcheck disable=SC1091 +. ./config.sh + +start_suite "Test long connections (procspy) between processes" + +weave_on "$HOST1" launch +scope_on "$HOST1" launch --probe.ebpf.connections=false + +server_on "$HOST1" +weave_proxy_on "$HOST1" run -dti --name client alpine /bin/sh -c "while true; do \ + nc nginx.weave.local 80 || true; \ + sleep 1; \ +done" + +wait_for processes "$HOST1" 60 "nginx: worker process" nc + +has processes "$HOST1" "nginx: worker process" +has processes "$HOST1" nc +has_connection processes "$HOST1" nc "nginx: worker process" + +scope_end_suite diff --git a/integration/340_process_edge_across_host_2_test.sh b/integration/340_process_edge_across_host_2_test.sh index 161c24929..50d8e1006 100755 --- a/integration/340_process_edge_across_host_2_test.sh +++ b/integration/340_process_edge_across_host_2_test.sh @@ -3,13 +3,13 @@ # shellcheck disable=SC1091 . ./config.sh -start_suite "Test long connections (procspy) between processes on different hosts" +start_suite "Test long connections between processes on different hosts" weave_on "$HOST1" launch "$HOST1" "$HOST2" weave_on "$HOST2" launch "$HOST1" "$HOST2" -scope_on "$HOST1" launch --probe.ebpf.connections=false --probe.conntrack=false -scope_on "$HOST2" launch --probe.ebpf.connections=false --probe.conntrack=false +scope_on "$HOST1" launch +scope_on "$HOST2" launch server_on "$HOST1" weave_proxy_on "$HOST2" run -dti --name client alpine /bin/sh -c "while true; do \ diff --git a/integration/341_process_edge_across_host_without_ebpf_2_test.sh b/integration/341_process_edge_across_host_without_ebpf_2_test.sh new file mode 100755 index 000000000..5dd41c507 --- /dev/null +++ b/integration/341_process_edge_across_host_without_ebpf_2_test.sh @@ -0,0 +1,31 @@ +#! /bin/bash + +# shellcheck disable=SC1091 +. ./config.sh + +start_suite "Test long connections (procspy) between processes on different hosts" + +weave_on "$HOST1" launch "$HOST1" "$HOST2" +weave_on "$HOST2" launch "$HOST1" "$HOST2" + +scope_on "$HOST1" launch --probe.ebpf.connections=false +scope_on "$HOST2" launch --probe.ebpf.connections=false + +server_on "$HOST1" +weave_proxy_on "$HOST2" run -dti --name client alpine /bin/sh -c "while true; do \ + nc nginx.weave.local 80 || true; \ + sleep 1; \ +done" + +sleep 30 # need to allow the scopes to poll dns, resolve the other app ids, and send them reports + +check() { + has processes "$1" "nginx: worker process" + has processes "$1" nc + has_connection processes "$1" nc "nginx: worker process" +} + +check "$HOST1" +check "$HOST2" + +scope_end_suite diff --git a/probe/endpoint/ebpf.go b/probe/endpoint/ebpf.go index 0e4c24674..290e0b3f2 100644 --- a/probe/endpoint/ebpf.go +++ b/probe/endpoint/ebpf.go @@ -28,13 +28,14 @@ type ebpfConnection struct { // Closed connections are kept in the `closedConnections` slice for one iteration of `walkConnections`. type EbpfTracker struct { sync.Mutex - tracer *tracer.Tracer - readyToHandleConnections bool - dead bool - lastTimestampV4 uint64 + tracer *tracer.Tracer + ready bool + dead bool + lastTimestampV4 uint64 openConnections map[fourTuple]ebpfConnection closedConnections []ebpfConnection + closedDuringInit map[fourTuple]struct{} } var releaseRegex = regexp.MustCompile(`^(\d+)\.(\d+).*$`) @@ -77,7 +78,8 @@ func newEbpfTracker() (*EbpfTracker, error) { } tracker := &EbpfTracker{ - openConnections: map[fourTuple]ebpfConnection{}, + openConnections: map[fourTuple]ebpfConnection{}, + closedDuringInit: map[fourTuple]struct{}{}, } tracer, err := tracer.NewTracer(tracker) @@ -204,10 +206,6 @@ func (t *EbpfTracker) handleConnection(ev tracer.EventType, tuple fourTuple, pid t.Lock() defer t.Unlock() - if !t.isReadyToHandleConnections() { - return - } - log.Debugf("handleConnection(%v, [%v:%v --> %v:%v], pid=%v, netNS=%v)", ev, tuple.fromAddr, tuple.fromPort, tuple.toAddr, tuple.toPort, pid, networkNamespace) @@ -227,6 +225,9 @@ func (t *EbpfTracker) handleConnection(ev tracer.EventType, tuple fourTuple, pid networkNamespace: networkNamespace, } case tracer.EventClose: + if !t.ready { + t.closedDuringInit[tuple] = struct{}{} + } if deadConn, ok := t.openConnections[tuple]; ok { delete(t.openConnections, tuple) t.closedConnections = append(t.closedConnections, deadConn) @@ -257,14 +258,19 @@ func (t *EbpfTracker) feedInitialConnections(conns procspy.ConnIter, seenTuples t.Lock() for conn := conns.Next(); conn != nil; conn = conns.Next() { tuple, namespaceID, incoming := connectionTuple(conn, seenTuples) - t.openConnections[tuple] = ebpfConnection{ - incoming: incoming, - tuple: tuple, - pid: int(conn.Proc.PID), - networkNamespace: namespaceID, + 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, + } + } } } - t.readyToHandleConnections = true + t.closedDuringInit = nil + t.ready = true t.Unlock() for _, p := range processesWaitingInAccept { @@ -273,10 +279,6 @@ func (t *EbpfTracker) feedInitialConnections(conns procspy.ConnIter, seenTuples } } -func (t *EbpfTracker) isReadyToHandleConnections() bool { - return t.readyToHandleConnections -} - func (t *EbpfTracker) isDead() bool { t.Lock() defer t.Unlock() diff --git a/probe/endpoint/ebpf_test.go b/probe/endpoint/ebpf_test.go index a3a85f216..14f9c8064 100644 --- a/probe/endpoint/ebpf_test.go +++ b/probe/endpoint/ebpf_test.go @@ -9,6 +9,15 @@ import ( "github.com/weaveworks/tcptracer-bpf/pkg/tracer" ) +func newMockEbpfTracker() *EbpfTracker { + return &EbpfTracker{ + ready: true, + dead: false, + + openConnections: map[fourTuple]ebpfConnection{}, + } +} + func TestHandleConnection(t *testing.T) { var ( ServerPid uint32 = 42 @@ -92,13 +101,7 @@ func TestHandleConnection(t *testing.T) { } ) - mockEbpfTracker := &EbpfTracker{ - readyToHandleConnections: true, - dead: false, - - openConnections: map[fourTuple]ebpfConnection{}, - closedConnections: []ebpfConnection{}, - } + mockEbpfTracker := newMockEbpfTracker() tuple := fourTuple{IPv4ConnectEvent.SAddr.String(), IPv4ConnectEvent.DAddr.String(), uint16(IPv4ConnectEvent.SPort), uint16(IPv4ConnectEvent.DPort)} mockEbpfTracker.handleConnection(IPv4ConnectEvent.Type, tuple, int(IPv4ConnectEvent.Pid), strconv.FormatUint(uint64(IPv4ConnectEvent.NetNS), 10)) @@ -114,13 +117,7 @@ func TestHandleConnection(t *testing.T) { mockEbpfTracker.openConnections[tuple]) } - mockEbpfTracker = &EbpfTracker{ - readyToHandleConnections: true, - dead: false, - - openConnections: map[fourTuple]ebpfConnection{}, - closedConnections: []ebpfConnection{}, - } + mockEbpfTracker = newMockEbpfTracker() tuple = fourTuple{IPv4AcceptEvent.SAddr.String(), IPv4AcceptEvent.DAddr.String(), uint16(IPv4AcceptEvent.SPort), uint16(IPv4AcceptEvent.DPort)} mockEbpfTracker.handleConnection(IPv4AcceptEvent.Type, tuple, int(IPv4AcceptEvent.Pid), strconv.FormatUint(uint64(IPv4AcceptEvent.NetNS), 10)) @@ -155,26 +152,20 @@ func TestWalkConnections(t *testing.T) { toPort: 0, } ) - mockEbpfTracker := &EbpfTracker{ - readyToHandleConnections: true, - dead: false, - openConnections: map[fourTuple]ebpfConnection{ - activeTuple: { - tuple: activeTuple, - networkNamespace: "12345", - incoming: true, - pid: 0, - }, - }, - closedConnections: []ebpfConnection{ - { - tuple: inactiveTuple, - networkNamespace: "12345", - incoming: false, - pid: 0, - }, - }, + mockEbpfTracker := newMockEbpfTracker() + mockEbpfTracker.openConnections[activeTuple] = ebpfConnection{ + tuple: activeTuple, + networkNamespace: "12345", + incoming: true, + pid: 0, } + mockEbpfTracker.closedConnections = append(mockEbpfTracker.closedConnections, + ebpfConnection{ + tuple: inactiveTuple, + networkNamespace: "12345", + incoming: false, + pid: 0, + }) mockEbpfTracker.walkConnections(func(e ebpfConnection) { cnt++ }) @@ -204,11 +195,7 @@ func TestInvalidTimeStampDead(t *testing.T) { NetNS: NetNS, } ) - mockEbpfTracker := &EbpfTracker{ - readyToHandleConnections: true, - dead: false, - openConnections: map[fourTuple]ebpfConnection{}, - } + mockEbpfTracker := newMockEbpfTracker() event.Timestamp = 0 mockEbpfTracker.TCPEventV4(event) event2 := event