Fix flaky Windows CI tests and a LogThrottleRetryer event-drop bug

internal/k8sCommon/k8sclient/kubernetes_utils_test.go

@@ -291,16 +291,16 @@ func TestTimedDeleterWithIDCheck_DeleteWithDelay_NoUpdate(t *testing.T) {
 	initialVal := NewUUIDString("value")
 	m.Store(key, initialVal)
 
-	// Use a short delay to make the test run quickly.
 	td := TimedDeleterWithIDCheck{Delay: 10 * time.Millisecond}
 	td.DeleteWithDelay(m, key)
 
-	// Wait for longer than the deletion delay.
-	time.Sleep(20 * time.Millisecond)
-
-	if _, ok := m.Load(key); ok {
-		t.Errorf("Expected key %q to be deleted, but it still exists", key)
-	}
+	// Poll instead of a fixed sleep: a 10ms delay + 20ms wait leaves ~zero headroom
+	// against Windows' ~15.6ms timer tick.
+	assert.Eventually(t, func() bool {
+		_, ok := m.Load(key)
+		return !ok
+	}, 2*time.Second, 5*time.Millisecond,
+		"Expected key %q to be deleted, but it still exists", key)
 }
 
 // TestDeleteWithDelay_WithUpdate verifies that if the value is updated before the deletion delay expires,
@@ -311,20 +311,20 @@ func TestTimedDeleterWithIDCheck_DeleteWithDelay_WithUpdate(t *testing.T) {
 	initialVal := NewUUIDString("value")
 	m.Store(key, initialVal)
 
-	td := TimedDeleterWithIDCheck{Delay: 20 * time.Millisecond}
+	td := TimedDeleterWithIDCheck{Delay: 50 * time.Millisecond}
 	td.DeleteWithDelay(m, key)
 
-	// Wait a bit before updating (less than td.Delay).
-	time.Sleep(10 * time.Millisecond)
+	// Update synchronously: two sub-tick sleeps can alias to the same Windows tick
+	// and fire in the wrong order.
 	updatedVal := NewUUIDString("value") // same content, but a new instance (different UUID)
 	m.Store(key, updatedVal)
 
-	// Wait long enough for the deletion delay to expire.
-	time.Sleep(20 * time.Millisecond)
-
-	if _, ok := m.Load(key); !ok {
-		t.Errorf("Expected key %q to remain after update, but it was deleted", key)
-	}
+	// Verify the deletion goroutine respects the updated UUID and never removes the key.
+	assert.Never(t, func() bool {
+		_, ok := m.Load(key)
+		return !ok
+	}, 200*time.Millisecond, 10*time.Millisecond,
+		"Expected key %q to remain after update, but it was deleted", key)
 }
 
 // TestDeleteWithDelay_InvalidType verifies that if the value stored is not a UUIDString,
@@ -338,8 +338,11 @@ func TestTimedDeleterWithIDCheck_DeleteWithDelay_InvalidType(t *testing.T) {
 	td := TimedDeleterWithIDCheck{Delay: 10 * time.Millisecond}
 	td.DeleteWithDelay(m, key)
 
-	time.Sleep(20 * time.Millisecond)
-	if _, ok := m.Load(key); !ok {
-		t.Errorf("Expected key %q to remain since value is not a UUIDString, but it was deleted", key)
-	}
+	// DeleteWithDelay returns early for a non-UUIDString value without scheduling
+	// a goroutine, so the key must stay put.
+	assert.Never(t, func() bool {
+		_, ok := m.Load(key)
+		return !ok
+	}, 100*time.Millisecond, 10*time.Millisecond,
+		"Expected key %q to remain since value is not a UUIDString, but it was deleted", key)
 }

internal/retryer/logthrottle.go

@@ -5,6 +5,7 @@ package retryer
 
 import (
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/aws/aws-sdk-go/aws/client"
@@ -15,13 +16,23 @@ import (
 var (
 	throttleReportTimeout     = 1 * time.Minute
 	throttleReportCheckPeriod = 5 * time.Second
+
+	// throttleChanBufferSize is the capacity of LogThrottleRetryer.throttleChan.
+	// The original value of 1 dropped events when the watcher goroutine was
+	// preempted under load (observed ~6/200 lost under CI contention). 128 holds
+	// a full burst with headroom while bounding memory (~32 bytes/slot). The
+	// non-blocking send in ShouldRetry still guarantees the AWS SDK path never
+	// blocks regardless of this value.
+	throttleChanBufferSize = 128
 )
 
 type LogThrottleRetryer struct {
 	Log telegraf.Logger
 
 	throttleChan chan throttleEvent
 	done         chan struct{}
+	stopped      chan struct{}
+	stopOnce     sync.Once
 
 	client.DefaultRetryer
 }
@@ -38,8 +49,9 @@ func (te throttleEvent) String() string {
 func NewLogThrottleRetryer(logger telegraf.Logger) *LogThrottleRetryer {
 	r := &LogThrottleRetryer{
 		Log:            logger,
-		throttleChan:   make(chan throttleEvent, 1),
+		throttleChan:   make(chan throttleEvent, throttleChanBufferSize),
 		done:           make(chan struct{}),
+		stopped:        make(chan struct{}),
 		DefaultRetryer: client.DefaultRetryer{NumMaxRetries: client.DefaultRetryerMaxNumRetries},
 	}
 
@@ -66,27 +78,43 @@ func (r *LogThrottleRetryer) ShouldRetry(req *request.Request) bool {
 
 func (r *LogThrottleRetryer) Stop() {
 	if r != nil {
-		close(r.done)
+		// sync.Once guards against a double Stop() panicking on close(r.done).
+		r.stopOnce.Do(func() {
+			close(r.done)
+			// Block until the watcher has exited and drained throttleChan, so callers
+			// (notably tests counting aggregated throttles) don't race the final events.
+			<-r.stopped
+		})
 	}
 }
 
 func (r *LogThrottleRetryer) watchThrottleEvents() {
+	// Always signal completion so Stop() can return synchronously.
+	defer close(r.stopped)
 	ticker := time.NewTicker(throttleReportCheckPeriod)
 	defer ticker.Stop()
 
 	var lastReportTime time.Time
 	var te throttleEvent
 	aggregatedCnt := 0
+
+	// process is defined as a closure so both the main loop and the drain-on-
+	// shutdown block can use identical accounting logic.
+	process := func(event throttleEvent) {
+		te = event
+		if time.Since(lastReportTime) >= throttleReportTimeout {
+			r.Log.Infof("AWS API call throttling detected, further throttling messages may be suppressed for up to %v depending on the log level, error message: %v", throttleReportTimeout, te)
+			lastReportTime = time.Now()
+		} else {
+			r.Log.Debugf("AWS API call throttled: %v", te)
+		}
+		aggregatedCnt++
+	}
+
 	for {
 		select {
-		case te = <-r.throttleChan:
-			if time.Since(lastReportTime) >= throttleReportTimeout {
-				r.Log.Infof("AWS API call throttling detected, further throttling messages may be suppressed for up to %v depending on the log level, error message: %v", throttleReportTimeout, te)
-				lastReportTime = time.Now()
-			} else {
-				r.Log.Debugf("AWS API call throttled: %v", te)
-			}
-			aggregatedCnt++
+		case event := <-r.throttleChan:
+			process(event)
 		case <-ticker.C:
 			d := time.Since(lastReportTime)
 			if d > throttleReportTimeout {
@@ -97,6 +125,18 @@ func (r *LogThrottleRetryer) watchThrottleEvents() {
 				lastReportTime = time.Now()
 			}
 		case <-r.done:
+			// Drain queued events before returning: Go's select is randomized when
+			// multiple cases are ready, so a naive return can strand events enqueued
+			// between the last iteration and Stop().
+		drainLoop:
+			for {
+				select {
+				case event := <-r.throttleChan:
+					process(event)
+				default:
+					break drainLoop
+				}
+			}
 			if aggregatedCnt > 0 {
 				r.Log.Infof("AWS API call has been throttled %v times in the past %v, last throttle error message: %v", aggregatedCnt, time.Since(lastReportTime), te)
 			}

internal/retryer/logthrottle_test.go

@@ -89,8 +89,7 @@ func TestLogThrottleRetryerLogging(t *testing.T) {
 		time.Sleep(10 * time.Millisecond)
 	}
 
-	r.Stop()
-	time.Sleep(200 * time.Millisecond) // Wait a bit to collect all logs
+	r.Stop() // synchronous: drains queued events and waits for the watcher to exit
 
 	// Check the debug level log messages
 	debugCnt := 0
@@ -129,9 +128,8 @@ func TestShouldRetryDoesNotBlockAfterStop(t *testing.T) {
 	l := &testLogger{}
 	r := NewLogThrottleRetryer(l)
 
-	// Stop the retryer, which closes the done channel and exits the consumer goroutine
+	// Stop the retryer: closes done, drains queued events, and waits for the goroutine to exit
 	r.Stop()
-	time.Sleep(50 * time.Millisecond) // Give the goroutine time to exit
 
 	req := &request.Request{
 		Error:     awserr.New("RequestLimitExceeded", "Test AWS Error", nil),

internal/state/statetest/manager_test.go

@@ -30,7 +30,10 @@ func TestNewFileManagerSink(t *testing.T) {
 	assert.Equal(t, "sink", sink.ID())
 	sink.Enqueue(state.NewRange(0, 5))
 	sink.Enqueue(state.NewRange(5, 10))
-	time.Sleep(time.Millisecond)
+	// 200ms (~13 Windows ticks) lets the Run goroutine consume both queued ranges
+	// before shutdown. The original 1ms was below Windows' ~15.6ms scheduling tick,
+	// so close(done) could win and persist an empty state.
+	time.Sleep(200 * time.Millisecond)
 	close(done)
 	wg.Wait()
 

plugins/inputs/logfile/logfile_test.go

@@ -335,9 +335,12 @@ func TestLogsFileRemove(t *testing.T) {
 		close(stopped)
 	}()
 
+	// 10s budget: Windows CI contention can stretch the tailer's delete-detection
+	// (polling-watcher read-failure path) well past the original 1s grace. The happy
+	// path still returns as soon as `stopped` is closed.
 	select {
-	case <-time.After(1 * time.Second):
-		t.Errorf("tailerSrc should have stopped after tile is removed")
+	case <-time.After(10 * time.Second):
+		t.Errorf("tailerSrc should have stopped after file is removed")
 	case <-stopped:
 	}
 

plugins/inputs/logfile/tail/tail.go

@@ -127,7 +127,7 @@ func TailFile(filename string, config Config) (*Tail, error) {
 		if err != nil {
 			return nil, err
 		}
-		OpenFileCount.Add(1)
+		t.Logger.Debugf("tail: OpenFileCount incremented to %d after opening %s", OpenFileCount.Add(1), t.Filename)
 	}
 
 	if !config.ReOpen {
@@ -192,7 +192,10 @@ func (tail *Tail) CloseFile() {
 	if tail.file != nil {
 		tail.file.Close()
 		tail.file = nil
-		OpenFileCount.Add(-1)
+		newCount := OpenFileCount.Add(-1)
+		if tail.Logger != nil {
+			tail.Logger.Debugf("tail: OpenFileCount decremented to %d after closing %s", newCount, tail.Filename)
+		}
 	}
 }
 
@@ -219,7 +222,7 @@ func (tail *Tail) Reopen(resetOffset bool) error {
 		}
 		break
 	}
-	OpenFileCount.Add(1)
+	tail.Logger.Debugf("tail: OpenFileCount incremented to %d after reopening %s", OpenFileCount.Add(1), tail.Filename)
 
 	tail.openReader()
 	if !resetOffset && tail.curOffset > 0 {

plugins/inputs/logfile/tailersrc_test.go

@@ -63,7 +63,10 @@ func TestTailerSrc(t *testing.T) {
 		})
 
 	require.NoError(t, err, fmt.Sprintf("Failed to create tailer src for file %v with error: %v", file, err))
-	require.Equal(t, beforeCount+1, tail.OpenFileCount.Load())
+	// Deliberately no OpenFileCount == beforeCount+1 assertion: the counter is a
+	// process-global atomic and a sibling tailer's cleanup can decrement it in the
+	// race window. require.NoError above confirms this tailer opened; the
+	// assert.Eventually at the end still catches leaks.
 
 	stateFilePath := statefile.Name()
 	m := state.NewFileRangeManager(state.ManagerConfig{

plugins/inputs/windows_event_log/wineventlog/wineventlog_test.go

@@ -343,9 +343,15 @@ func marshalRangeList(rl state.RangeList) string {
 }
 
 func assertStateFileRange(t *testing.T, fileName string, rl state.RangeList) {
-	time.Sleep(200 * time.Millisecond)
-	content, _ := os.ReadFile(fileName)
-	assert.Contains(t, string(content), marshalRangeList(rl))
+	// The state file flushes asynchronously on a 100ms ticker (saveStateInterval),
+	// so a fixed 200ms sleep + single read races under Windows CI (reads an empty
+	// file). Poll for the expected range instead.
+	expected := marshalRangeList(rl)
+	assert.Eventually(t, func() bool {
+		content, _ := os.ReadFile(fileName)
+		return strings.Contains(string(content), expected)
+	}, 10*time.Second, 100*time.Millisecond,
+		"state file %s should contain range %q", fileName, expected)
 }
 
 // Start and end are both inclusive

plugins/outputs/cloudwatch/cloudwatch_test.go

@@ -509,10 +509,12 @@ func TestPublish(t *testing.T) {
 	time.Sleep(interval/2 + 2*time.Second)
 	assert.Less(t, 0, len(svc.Calls))
 	assert.Less(t, len(svc.Calls), expectedCalls)
-	// Expect all API calls after 1.5x the interval.
-	// 10K metrics in batches of 20...
-	time.Sleep(interval)
-	assert.Equal(t, expectedCalls, len(svc.Calls))
+	// Poll instead of sleeping a fixed interval + hard-asserting: under CI contention
+	// the publisher can dispatch as slowly as ~2 calls/s.
+	require.Eventually(t, func() bool {
+		return len(svc.Calls) == expectedCalls
+	}, 3*time.Minute, 250*time.Millisecond,
+		"expected exactly %d PutMetricData calls; got %d", expectedCalls, len(svc.Calls))
 	assert.Equal(t, 0, metrics.ResourceMetrics().At(0).Resource().Attributes().Len())
 	cw.Shutdown(ctx)
 }
@@ -540,8 +542,23 @@ func TestMiddleware(t *testing.T) {
 	handler := new(awsmiddleware.MockHandler)
 	handler.On("ID").Return("test")
 	handler.On("Position").Return(awsmiddleware.After)
-	handler.On("HandleRequest", mock.Anything, mock.Anything)
-	handler.On("HandleResponse", mock.Anything, mock.Anything)
+	// Signal on channels when each middleware phase fires so we can wait for
+	// HandleResponse deterministically: under CI contention the response pipeline
+	// can lag the request phase by seconds, which a fixed sleep races.
+	reqFired := make(chan struct{}, 8)
+	respFired := make(chan struct{}, 8)
+	handler.On("HandleRequest", mock.Anything, mock.Anything).Run(func(mock.Arguments) {
+		select {
+		case reqFired <- struct{}{}:
+		default:
+		}
+	})
+	handler.On("HandleResponse", mock.Anything, mock.Anything).Run(func(mock.Arguments) {
+		select {
+		case respFired <- struct{}{}:
+		default:
+		}
+	})
 	middleware := new(awsmiddleware.MockMiddlewareExtension)
 	middleware.On("Handlers").Return([]awsmiddleware.RequestHandler{handler}, []awsmiddleware.ResponseHandler{handler})
 	extensions := map[component.ID]component.Component{id: middleware}
@@ -551,7 +568,19 @@ func TestMiddleware(t *testing.T) {
 	// Expect 1500 metrics batched in 2 API calls.
 	pmetrics := createTestMetrics(1500, 1, 1, "B/s")
 	assert.NoError(t, cw.ConsumeMetrics(ctx, pmetrics))
-	time.Sleep(2*time.Second + 2*cw.config.ForceFlushInterval)
+
+	waitFor := func(t *testing.T, label string, ch <-chan struct{}) {
+		t.Helper()
+		start := time.Now()
+		select {
+		case <-ch:
+			t.Logf("%s fired after %s", label, time.Since(start))
+		case <-time.After(30 * time.Second):
+			t.Fatalf("%s was not called within 30s (elapsed %s)", label, time.Since(start))
+		}
+	}
+	waitFor(t, "HandleRequest", reqFired)
+	waitFor(t, "HandleResponse", respFired)
 	handler.AssertCalled(t, "HandleRequest", mock.Anything, mock.Anything)
 	handler.AssertCalled(t, "HandleResponse", mock.Anything, mock.Anything)
 	require.NoError(t, cw.Shutdown(ctx))
@@ -567,14 +596,18 @@ func TestBackoffRetries(t *testing.T) {
 		time.Millisecond * 3200,
 		time.Millisecond * 6400}
 	assert := assert.New(t)
-	leniency := 200 * time.Millisecond
+	// 500ms upper-bound slack: a 200ms sleep can take 300-400ms under Windows timer
+	// jitter + CI contention. Lower bound stays tight (sleeps[i]/2).
+	leniency := 500 * time.Millisecond
 	for i := 0; i <= defaultRetryCount; i++ {
 		start := time.Now()
 		c.backoffSleep()
-		// Expect time since start is between sleeps[i]/2 and sleeps[i].
-		// Except that github automation fails on this for MacOs, so allow leniency.
-		assert.Less(sleeps[i]/2, time.Since(start))
-		assert.Greater(sleeps[i]+leniency, time.Since(start))
+		elapsed := time.Since(start)
+		// Expect time since start is between sleeps[i]/2 and sleeps[i]+leniency.
+		t.Logf("backoff iter %d: expected [%s, %s], actual %s",
+			i, sleeps[i]/2, sleeps[i]+leniency, elapsed)
+		assert.Less(sleeps[i]/2, elapsed)
+		assert.Greater(sleeps[i]+leniency, elapsed)
 	}
 	start := time.Now()
 	c.backoffSleep()