Skip to content

chore(tests): poll received events in hec output tests instead of sleeping (PIPE-986) #266

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Dylan-M merged 1 commit into from
Jul 31, 2026
Merged

chore(tests): poll received events in hec output tests instead of sleeping (PIPE-986) #266

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 29 additions & 18 deletions output/hec/ack_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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))
Expand Down Expand Up @@ -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
Expand All @@ -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))
Expand Down Expand Up @@ -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))
Expand Down Expand Up @@ -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))
}

Expand Down
95 changes: 62 additions & 33 deletions output/hec/hec_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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),
Expand All @@ -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),
Expand All @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand All @@ -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)
Expand Down Expand Up @@ -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))
Expand Down Expand Up @@ -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))
Expand Down Expand Up @@ -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")
}

Expand Down Expand Up @@ -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))
Expand Down
Loading