diff --git a/output/hec/ack_test.go b/output/hec/ack_test.go index f29754c..2db6ce3 100644 --- a/output/hec/ack_test.go +++ b/output/hec/ack_test.go @@ -183,18 +183,18 @@ func TestHEC_ACKConfirmFlow(t *testing.T) { require.NoError(t, err) // Wait for the event to be sent and tracked - time.Sleep(500 * time.Millisecond) - - // Verify ackId is being tracked + require.Eventually(t, func() bool { + return h.workers[0].tracker.size() == 1 + }, 5*time.Second, 10*time.Millisecond) assert.Equal(t, 1, h.workers[0].tracker.size(), "expected 1 pending ackId") // Confirm the ackId in the mock server state.setACKStatus(0, true) - // Wait for the poller to pick it up - time.Sleep(500 * time.Millisecond) - - // Should be confirmed and removed + // Wait for the poller to pick it up and remove the confirmed ackId + require.Eventually(t, func() bool { + return h.workers[0].tracker.size() == 0 + }, 5*time.Second, 10*time.Millisecond) assert.Equal(t, 0, h.workers[0].tracker.size(), "expected 0 pending ackIds after confirmation") require.NoError(t, h.Stop(ctx)) @@ -230,9 +230,12 @@ func TestHEC_ACKExpireAndResend(t *testing.T) { }) require.NoError(t, err) - // Wait for event to be sent, ackId 0 to expire, and resend to produce a new ackId. - // With 300ms timeout and 100ms poll, after ~500ms the first resend should have happened. - time.Sleep(600 * time.Millisecond) + // Wait for event to be sent, ackId 0 to expire, and a resend to produce a + // new ackId. The mock server auto-increments ackIds: the initial send is + // ackId 0 (nextAckID -> 1); once a resend happens nextAckID reaches 2. + require.Eventually(t, func() bool { + return state.nextAckID.Load() >= 2 + }, 5*time.Second, 10*time.Millisecond) // Confirm ALL ackIds the server has seen so far (one of them will be the current pending) // The server auto-increments ackIds, so let's confirm a range @@ -241,8 +244,9 @@ func TestHEC_ACKExpireAndResend(t *testing.T) { } // Wait for the poller to pick up the confirmation - time.Sleep(300 * time.Millisecond) - + require.Eventually(t, func() bool { + return h.workers[0].tracker.size() == 0 + }, 5*time.Second, 10*time.Millisecond) assert.Equal(t, 0, h.workers[0].tracker.size(), "expected 0 pending after resend+confirm") require.NoError(t, h.Stop(ctx)) @@ -278,11 +282,18 @@ func TestHEC_ACKMaxRetriesDrop(t *testing.T) { }) require.NoError(t, err) - // Wait for initial send, expiry, resend, second expiry, and drop - // With 50ms poll + 100ms timeout, this should cycle through quickly - time.Sleep(2 * time.Second) - - // After maxRetries exhausted, tracker should be empty (batch dropped) + // First wait for the event to be sent and tracked, so the subsequent + // wait-for-drop cannot pass vacuously against an empty tracker. + require.Eventually(t, func() bool { + return h.workers[0].tracker.size() == 1 + }, 5*time.Second, 10*time.Millisecond) + + // After initial send, expiry, resend, second expiry, and drop + // (50ms poll + 100ms timeout), maxRetries is exhausted and the batch + // is dropped, leaving the tracker empty. + require.Eventually(t, func() bool { + return h.workers[0].tracker.size() == 0 + }, 5*time.Second, 10*time.Millisecond) assert.Equal(t, 0, h.workers[0].tracker.size(), "expected 0 pending after max retries drop") require.NoError(t, h.Stop(ctx)) @@ -317,7 +328,7 @@ func TestHEC_ACKDisabledNoTracker(t *testing.T) { }) require.NoError(t, err) - time.Sleep(300 * time.Millisecond) + // Stop flushes any pending event; no state to wait on beforehand. require.NoError(t, h.Stop(ctx)) } diff --git a/output/hec/hec_test.go b/output/hec/hec_test.go index fac6d6b..2e998ac 100644 --- a/output/hec/hec_test.go +++ b/output/hec/hec_test.go @@ -187,12 +187,14 @@ func TestHEC_WriteAndBatchByTimeout(t *testing.T) { } func TestHEC_AuthHeader(t *testing.T) { - var gotAuth string - var gotChannel string + // atomic.Value so the test goroutine can poll these while the server + // goroutine writes them without a data race. + var gotAuth atomic.Value + var gotChannel atomic.Value server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - gotAuth = r.Header.Get("Authorization") - gotChannel = r.Header.Get("X-Splunk-Request-Channel") + gotAuth.Store(r.Header.Get("Authorization")) + gotChannel.Store(r.Header.Get("X-Splunk-Request-Channel")) resp := hecResponse{Text: "Success", Code: 0} w.WriteHeader(200) @@ -204,8 +206,8 @@ func TestHEC_AuthHeader(t *testing.T) { logger := zaptest.NewLogger(t) t.Run("with ACK enabled sends channel header", func(t *testing.T) { - gotAuth = "" - gotChannel = "" + gotAuth.Store("") + gotChannel.Store("") h, err := New(logger, WithHost(host), @@ -225,17 +227,19 @@ func TestHEC_AuthHeader(t *testing.T) { }) require.NoError(t, err) - time.Sleep(500 * time.Millisecond) + require.Eventually(t, func() bool { + return gotChannel.Load().(string) != "" + }, 5*time.Second, 10*time.Millisecond) - assert.Equal(t, "Splunk my-secret-token", gotAuth) - assert.NotEmpty(t, gotChannel, "expected X-Splunk-Request-Channel header when ACK enabled") + assert.Equal(t, "Splunk my-secret-token", gotAuth.Load().(string)) + assert.NotEmpty(t, gotChannel.Load().(string), "expected X-Splunk-Request-Channel header when ACK enabled") require.NoError(t, h.Stop(ctx)) }) t.Run("with ACK disabled omits channel header", func(t *testing.T) { - gotAuth = "" - gotChannel = "" + gotAuth.Store("") + gotChannel.Store("") h, err := New(logger, WithHost(host), @@ -255,20 +259,27 @@ func TestHEC_AuthHeader(t *testing.T) { }) require.NoError(t, err) - time.Sleep(500 * time.Millisecond) + // Wait for the request to arrive (auth header is always set); the + // channel header must remain empty when ACK is disabled. + require.Eventually(t, func() bool { + return gotAuth.Load().(string) != "" + }, 5*time.Second, 10*time.Millisecond) - assert.Equal(t, "Splunk my-secret-token", gotAuth) - assert.Empty(t, gotChannel, "expected no X-Splunk-Request-Channel header when ACK disabled") + assert.Equal(t, "Splunk my-secret-token", gotAuth.Load().(string)) + assert.Empty(t, gotChannel.Load().(string), "expected no X-Splunk-Request-Channel header when ACK disabled") require.NoError(t, h.Stop(ctx)) }) } func TestHEC_EventFormatRaw(t *testing.T) { - var receivedBody []byte + // atomic.Value so the test goroutine can poll the captured body while the + // server goroutine writes it without a data race. + var receivedBody atomic.Value server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - receivedBody, _ = io.ReadAll(r.Body) + body, _ := io.ReadAll(r.Body) + receivedBody.Store(body) resp := hecResponse{Text: "Success", Code: 0} w.WriteHeader(200) json.NewEncoder(w).Encode(resp) @@ -297,20 +308,26 @@ func TestHEC_EventFormatRaw(t *testing.T) { }) require.NoError(t, err) - time.Sleep(500 * time.Millisecond) + require.Eventually(t, func() bool { + b, _ := receivedBody.Load().([]byte) + return len(b) > 0 + }, 5*time.Second, 10*time.Millisecond) var event hecEvent - require.NoError(t, json.Unmarshal(receivedBody, &event)) + require.NoError(t, json.Unmarshal(receivedBody.Load().([]byte), &event)) assert.Equal(t, `{"key":"value"}`, event.Event) require.NoError(t, h.Stop(ctx)) } func TestHEC_EventFormatParsed(t *testing.T) { - var receivedBody []byte + // atomic.Value so the test goroutine can poll the captured body while the + // server goroutine writes it without a data race. + var receivedBody atomic.Value server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - receivedBody, _ = io.ReadAll(r.Body) + body, _ := io.ReadAll(r.Body) + receivedBody.Store(body) resp := hecResponse{Text: "Success", Code: 0} w.WriteHeader(200) json.NewEncoder(w).Encode(resp) @@ -342,10 +359,13 @@ func TestHEC_EventFormatParsed(t *testing.T) { }) require.NoError(t, err) - time.Sleep(500 * time.Millisecond) + require.Eventually(t, func() bool { + b, _ := receivedBody.Load().([]byte) + return len(b) > 0 + }, 5*time.Second, 10*time.Millisecond) var event hecEvent - require.NoError(t, json.Unmarshal(receivedBody, &event)) + require.NoError(t, json.Unmarshal(receivedBody.Load().([]byte), &event)) // In parsed mode, event should be a map eventMap, ok := event.Event.(map[string]any) require.True(t, ok, "expected event to be a map, got %T", event.Event) @@ -356,10 +376,13 @@ func TestHEC_EventFormatParsed(t *testing.T) { } func TestHEC_EventFormatParsedFallback(t *testing.T) { - var receivedBody []byte + // atomic.Value so the test goroutine can poll the captured body while the + // server goroutine writes it without a data race. + var receivedBody atomic.Value server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - receivedBody, _ = io.ReadAll(r.Body) + body, _ := io.ReadAll(r.Body) + receivedBody.Store(body) resp := hecResponse{Text: "Success", Code: 0} w.WriteHeader(200) json.NewEncoder(w).Encode(resp) @@ -389,10 +412,13 @@ func TestHEC_EventFormatParsedFallback(t *testing.T) { }) require.NoError(t, err) - time.Sleep(500 * time.Millisecond) + require.Eventually(t, func() bool { + b, _ := receivedBody.Load().([]byte) + return len(b) > 0 + }, 5*time.Second, 10*time.Millisecond) var event hecEvent - require.NoError(t, json.Unmarshal(receivedBody, &event)) + require.NoError(t, json.Unmarshal(receivedBody.Load().([]byte), &event)) assert.Equal(t, "raw fallback", event.Event) require.NoError(t, h.Stop(ctx)) @@ -431,9 +457,10 @@ func TestHEC_HECErrorResponse(t *testing.T) { }) require.NoError(t, err) - time.Sleep(500 * time.Millisecond) - // Should have attempted the request (error is logged, not returned) + require.Eventually(t, func() bool { + return requestCount.Load() >= int32(1) + }, 5*time.Second, 10*time.Millisecond) assert.GreaterOrEqual(t, requestCount.Load(), int32(1)) require.NoError(t, h.Stop(ctx)) @@ -480,9 +507,10 @@ func TestHEC_GracefulShutdown(t *testing.T) { // Stop should flush remaining events require.NoError(t, h.Stop(ctx)) - // Give a moment for the final flush - time.Sleep(200 * time.Millisecond) - + // Wait for the final flush to reach the server + require.Eventually(t, func() bool { + return batchCount.Load() >= int32(1) + }, 5*time.Second, 10*time.Millisecond) assert.GreaterOrEqual(t, batchCount.Load(), int32(1), "expected at least one batch to be sent during shutdown") } @@ -523,8 +551,9 @@ func TestHEC_MultipleWorkers(t *testing.T) { require.NoError(t, err) } - time.Sleep(1 * time.Second) - + require.Eventually(t, func() bool { + return requestCount.Load() >= int32(10) + }, 5*time.Second, 10*time.Millisecond) assert.GreaterOrEqual(t, requestCount.Load(), int32(10), "expected at least 10 requests with batch size 1") require.NoError(t, h.Stop(ctx))