From 2c0f9e94dba84a0e33069fe00984cfab7405fb8f Mon Sep 17 00:00:00 2001 From: Antonin Bas Date: Wed, 2 Sep 2026 10:17:21 -0700 Subject: [PATCH 1/2] Fix flaky TestTCPCollectingProcess_ConcurrentClient The test stopped the collector from inside one of the client goroutines, as soon as that goroutine observed at least 2 sessions. Nothing guaranteed that the other client had connected by then, so the collector's listener could be closed before the other client dialed, and the test failed with "Cannot establish connection". The session count is not a reliable signal for this, as it also includes the connection that waitForCollectorReady opens to probe the collector. Instead, stop the collector when the test returns, once all the clients are known to be connected. The clients still connect concurrently, which is the point of the test. The connections are also kept open and referenced until the end of the test: they used to be discarded right after being established, which allowed the garbage collector to close them at any point. Signed-off-by: Antonin Bas --- pkg/collector/process_test.go | 70 +++++++++++++++++++---------------- 1 file changed, 39 insertions(+), 31 deletions(-) diff --git a/pkg/collector/process_test.go b/pkg/collector/process_test.go index 6b8b3346..5ec53f3d 100644 --- a/pkg/collector/process_test.go +++ b/pkg/collector/process_test.go @@ -340,39 +340,47 @@ func TestUDPCollectingProcess_ReceiveDataRecord(t *testing.T) { func TestTCPCollectingProcess_ConcurrentClient(t *testing.T) { input := getCollectorInput(tcpTransport, false, false) cp, _ := InitCollectingProcess(input) + go cp.Start() + // Stop the collector when the test returns, and never before: stopping it from one + // of the client goroutines below used to race with the other clients, which could + // then fail to connect. + defer cp.Stop() + // wait until collector is ready + waitForCollectorReady(t, cp) + collectorAddr := cp.GetAddress() + + const numClients = 2 + // The connections are kept open until the end of the test, so that all the clients + // are connected to the collector simultaneously. They also need to be referenced + // until then, as the garbage collector closes unreachable connections. + conns := make([]net.Conn, numClients) var wg sync.WaitGroup - wg.Add(2) - go func() { - defer wg.Done() - // wait until collector is ready - waitForCollectorReady(t, cp) - collectorAddr := cp.GetAddress() - _, err := net.Dial(collectorAddr.Network(), collectorAddr.String()) - if err != nil { - t.Errorf("Cannot establish connection to %s", collectorAddr.String()) - } - }() - go func() { - defer wg.Done() - // wait until collector is ready - waitForCollectorReady(t, cp) - collectorAddr := cp.GetAddress() - _, err := net.Dial(collectorAddr.Network(), collectorAddr.String()) - if err != nil { - t.Errorf("Cannot establish connection to %s", collectorAddr.String()) - } - // Poll until both connections are registered by the collector's accept loop, - // rather than relying on a fixed sleep that can be too short on slow CI runners. - assert.Eventually(t, func() bool { - return cp.GetNumConnToCollector() >= 2 - }, 5*time.Second, 10*time.Millisecond, "There should be at least two tcp clients.") - cp.Stop() - }() - cp.Start() - // Ensure both goroutines (and their calls into t) have finished before the - // test returns, otherwise a slow goroutine can call t.Errorf after the - // test has already completed, causing a panic. + for i := range numClients { + wg.Add(1) + go func() { + defer wg.Done() + conn, err := net.Dial(collectorAddr.Network(), collectorAddr.String()) + // Use assert and not require, which must not be called from a goroutine + // other than the one running the test. + if !assert.NoErrorf(t, err, "Cannot establish connection to %s", collectorAddr.String()) { + return + } + conns[i] = conn + }() + } + // All the clients connect concurrently, but we wait for all of them to be connected + // before asserting anything about the collector's sessions. wg.Wait() + for _, conn := range conns { + if conn != nil { + defer conn.Close() + } + } + + // Sessions are registered asynchronously by the accept loop, hence Eventually. + assert.Eventually(t, func() bool { + return cp.GetNumConnToCollector() >= numClients + }, 5*time.Second, 10*time.Millisecond, "There should be at least two tcp clients.") } func TestUDPCollectingProcess_ConcurrentClient(t *testing.T) { From 67b82884c39a4006d245c4279fd47ff84d15e741 Mon Sep 17 00:00:00 2001 From: Antonin Bas Date: Wed, 2 Sep 2026 10:17:45 -0700 Subject: [PATCH 2/2] Stop dialing the collector to check for readiness in tests waitForCollectorReady used to dial the collector to determine whether it was ready. For TCP, such a connection is registered as a session by the accept loop, and is only unregistered asynchronously once it has been closed, so it can be observed by tests which assert on the collector's sessions. For UDP, it did not prove anything at all, as dialing a UDP socket is a purely local operation. Both the TCP and the UDP code paths bind the socket before publishing the collector's address, so waiting for that address is enough, and has no side effect on the collector. Because it no longer needs to wait for a connection to be accepted, the helper can also poll more frequently and return as soon as the collector is ready. With this source of interference gone, the concurrent client test can assert that the collector has a session for each client, and no other session, instead of asserting on the number of sessions only. Signed-off-by: Antonin Bas --- pkg/collector/process_test.go | 46 +++++++++++++++++++++++------------ 1 file changed, 31 insertions(+), 15 deletions(-) diff --git a/pkg/collector/process_test.go b/pkg/collector/process_test.go index 5ec53f3d..26b17f8d 100644 --- a/pkg/collector/process_test.go +++ b/pkg/collector/process_test.go @@ -354,6 +354,7 @@ func TestTCPCollectingProcess_ConcurrentClient(t *testing.T) { // are connected to the collector simultaneously. They also need to be referenced // until then, as the garbage collector closes unreachable connections. conns := make([]net.Conn, numClients) + sessionIDs := make([]string, numClients) var wg sync.WaitGroup for i := range numClients { wg.Add(1) @@ -366,10 +367,9 @@ func TestTCPCollectingProcess_ConcurrentClient(t *testing.T) { return } conns[i] = conn + sessionIDs[i] = localConnSessionID(conn) }() } - // All the clients connect concurrently, but we wait for all of them to be connected - // before asserting anything about the collector's sessions. wg.Wait() for _, conn := range conns { if conn != nil { @@ -377,10 +377,22 @@ func TestTCPCollectingProcess_ConcurrentClient(t *testing.T) { } } - // Sessions are registered asynchronously by the accept loop, hence Eventually. + // Check for these specific sessions, and not just the session count, so that the + // assertion cannot be satisfied by an unrelated session. Sessions are registered + // asynchronously by the accept loop, hence Eventually. assert.Eventually(t, func() bool { - return cp.GetNumConnToCollector() >= numClients - }, 5*time.Second, 10*time.Millisecond, "There should be at least two tcp clients.") + cp.mutex.RLock() + defer cp.mutex.RUnlock() + if len(cp.sessions) != numClients { + return false + } + for _, id := range sessionIDs { + if _, ok := cp.sessions[id]; !ok { + return false + } + } + return true + }, 5*time.Second, 10*time.Millisecond, "The collector should have a session for each tcp client.") } func TestUDPCollectingProcess_ConcurrentClient(t *testing.T) { @@ -1032,18 +1044,22 @@ func getCollectorInput(network string, isEncrypted bool, isIPv6 bool) CollectorI } } +// waitForCollectorReady waits until the collector's socket is bound and its listening +// address is available. Both the TCP and the UDP code paths bind the socket before +// calling updateAddress, so a non-nil address means that the kernel is already queuing +// incoming connections / datagrams for the collector, even if its accept or read loop +// has not been scheduled yet. +// +// It deliberately does not dial the collector to probe for readiness: such a connection +// is registered as a session by the TCP accept loop, and is only unregistered +// asynchronously after it is closed, which contaminates session assertions in tests. func waitForCollectorReady(t *testing.T, cp *CollectingProcess) { - checkConn := func(ctx context.Context) (bool, error) { - if conn, err := net.Dial(cp.GetAddress().Network(), cp.GetAddress().String()); err != nil { - return false, err - } else { - defer conn.Close() - return true, nil - } - } - if err := wait.PollUntilContextTimeout(context.Background(), 100*time.Millisecond, 500*time.Millisecond, false, checkConn); err != nil { - t.Errorf("Cannot establish connection to %s", cp.GetAddress().String()) + t.Helper() + checkAddr := func(ctx context.Context) (bool, error) { + return cp.GetAddress() != nil, nil } + err := wait.PollUntilContextTimeout(context.Background(), 10*time.Millisecond, 5*time.Second, true, checkAddr) + require.NoError(t, err, "Collecting process is not ready") } func disableLogToStderr() {