diff --git a/internal/k8sCommon/k8sclient/kubernetes_utils_test.go b/internal/k8sCommon/k8sclient/kubernetes_utils_test.go
index dc66b5e90ba..ade6a851265 100644
--- a/internal/k8sCommon/k8sclient/kubernetes_utils_test.go
+++ b/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)
 }
diff --git a/internal/retryer/logthrottle.go b/internal/retryer/logthrottle.go
index 3a18851cef9..d2eff2b57e0 100644
--- a/internal/retryer/logthrottle.go
+++ b/internal/retryer/logthrottle.go
@@ -5,6 +5,7 @@ package retryer
 
 import (
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/aws/aws-sdk-go/aws/client"
@@ -15,6 +16,14 @@ 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 {
@@ -22,6 +31,8 @@ type LogThrottleRetryer struct {
 
 	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)
 			}
diff --git a/internal/retryer/logthrottle_test.go b/internal/retryer/logthrottle_test.go
index 06e3aae8c85..b15c88360f6 100644
--- a/internal/retryer/logthrottle_test.go
+++ b/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),
diff --git a/internal/state/statetest/manager_test.go b/internal/state/statetest/manager_test.go
index ba792a666b6..d75edd82edb 100644
--- a/internal/state/statetest/manager_test.go
+++ b/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()
 
diff --git a/plugins/inputs/logfile/logfile_test.go b/plugins/inputs/logfile/logfile_test.go
index 2a28e3cae70..84a04c6d417 100644
--- a/plugins/inputs/logfile/logfile_test.go
+++ b/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:
 	}
 
diff --git a/plugins/inputs/logfile/tail/tail.go b/plugins/inputs/logfile/tail/tail.go
index cfaaa321a4f..2fc61ea9083 100644
--- a/plugins/inputs/logfile/tail/tail.go
+++ b/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 {
diff --git a/plugins/inputs/logfile/tailersrc_test.go b/plugins/inputs/logfile/tailersrc_test.go
index 86bb53abc6d..6dec21e8530 100644
--- a/plugins/inputs/logfile/tailersrc_test.go
+++ b/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{
diff --git a/plugins/inputs/windows_event_log/wineventlog/wineventlog_test.go b/plugins/inputs/windows_event_log/wineventlog/wineventlog_test.go
index b94bbfca19b..6bfc4b06a22 100644
--- a/plugins/inputs/windows_event_log/wineventlog/wineventlog_test.go
+++ b/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
diff --git a/plugins/outputs/cloudwatch/cloudwatch_test.go b/plugins/outputs/cloudwatch/cloudwatch_test.go
index f59e7d57948..246d1326450 100644
--- a/plugins/outputs/cloudwatch/cloudwatch_test.go
+++ b/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()
diff --git a/plugins/outputs/cloudwatch/convert_otel_test.go b/plugins/outputs/cloudwatch/convert_otel_test.go
index 688b390e208..d4fdfce22e8 100644
--- a/plugins/outputs/cloudwatch/convert_otel_test.go
+++ b/plugins/outputs/cloudwatch/convert_otel_test.go
@@ -181,6 +181,7 @@ func checkDatum(
 	unit string,
 	numMetrics int,
 ) {
+	t.Helper()
 	assert.True(t, strings.HasPrefix(*d.MetricName, namePrefix))
 	assert.Equal(t, unit, *d.Unit)
 	if d.distribution != nil {
@@ -201,8 +202,14 @@ func checkDatum(
 		assert.Equal(t, metricValue, *d.Value)
 	}
 
-	// Assuming unit test does not take more than 1 s.
-	assert.Less(t, time.Since(*d.Timestamp), time.Second)
+	// 1-minute window: this checks the timestamp survived ConvertOtelMetrics unmodified,
+	// not wall-clock performance. Wide enough to absorb CI scheduling jitter while still
+	// catching real timestamp bugs (which drift by hours/years, not seconds).
+	elapsed := time.Since(*d.Timestamp)
+	if !assert.Less(t, elapsed, time.Minute,
+		"datum timestamp is unexpectedly stale: elapsed=%s", elapsed) {
+		t.Logf("checkDatum: numMetrics=%d unit=%s elapsed=%s", numMetrics, unit, elapsed)
+	}
 	for _, dim := range d.Dimensions {
 		assert.True(t, strings.HasPrefix(*dim.Name, keyPrefix))
 		assert.True(t, strings.HasPrefix(*dim.Value, valPrefix))
diff --git a/plugins/processors/awsapplicationsignals/internal/cardinalitycontrol/metrics_limiter_test.go b/plugins/processors/awsapplicationsignals/internal/cardinalitycontrol/metrics_limiter_test.go
index 6139a3919f7..9fdbc38e074 100644
--- a/plugins/processors/awsapplicationsignals/internal/cardinalitycontrol/metrics_limiter_test.go
+++ b/plugins/processors/awsapplicationsignals/internal/cardinalitycontrol/metrics_limiter_test.go
@@ -34,8 +34,13 @@ func TestAdmitAndRollup(t *testing.T) {
 	limiter := NewMetricsLimiter(config, logger)
 
 	admittedAttributes := map[string]pcommon.Map{}
+	// Use 10 DISTINCT keys so exactly-2-admitted is deterministic. With random keys
+	// (the old newLowCardinalityAttributes(100)), the stream could redraw a rejected
+	// key, bump its CMS frequency above the min, and get it promoted+admitted -- the
+	// top-K's intended rotation, but it made the exactly-2 assertion flaky (~7% on
+	// Windows, ~1% on Linux).
 	for i := 0; i < 10; i++ {
-		attr := newLowCardinalityAttributes(100)
+		attr := newFixedAttributes(i)
 		if ok, _ := limiter.Admit("latency", attr, emptyResourceAttributes); ok {
 			uniqKey, _ := attr.Get("RemoteOperation")
 			admittedAttributes[uniqKey.AsString()] = attr
diff --git a/plugins/processors/ec2tagger/ec2tagger_test.go b/plugins/processors/ec2tagger/ec2tagger_test.go
index 96362272635..55ecd99a5e9 100644
--- a/plugins/processors/ec2tagger/ec2tagger_test.go
+++ b/plugins/processors/ec2tagger/ec2tagger_test.go
@@ -674,8 +674,35 @@ func TestExistingAttributesNotOverwritten(t *testing.T) {
 	err := tagger.Start(context.Background(), componenttest.NewNopHost())
 	assert.Nil(t, err)
 
-	// Wait for tags and volumes to be retrieved
-	time.Sleep(time.Second)
+	// The tagger loads tags, volumes, and metadata asynchronously after Start(), so a
+	// fixed 1s sleep raced them under CI. Poll instead: dispatch a probe metric carrying
+	// the InstanceId the tagger keys off, and wait until the output picks up the
+	// async-loaded attributes (ImageId, tagKey2, VolumeId). The probe must use the real
+	// test attributes or the tagger drops it (no InstanceId -> no ResourceMetrics -> panic).
+	require.Eventually(t, func() bool {
+		probe := createTestMetrics([]map[string]string{{
+			"InstanceId": "i-100000",
+			"device":     device1,
+		}})
+		out, perr := tagger.processMetrics(context.Background(), probe)
+		if perr != nil || out.ResourceMetrics().Len() == 0 {
+			return false
+		}
+		sms := out.ResourceMetrics().At(0).ScopeMetrics()
+		if sms.Len() == 0 || sms.At(0).Metrics().Len() == 0 {
+			return false
+		}
+		metric := sms.At(0).Metrics().At(0)
+		if metric.Gauge().DataPoints().Len() == 0 {
+			return false
+		}
+		attrs := metric.Gauge().DataPoints().At(0).Attributes()
+		_, hasImg := attrs.Get("ImageId")
+		_, hasTag := attrs.Get(tagKey2)
+		_, hasVol := attrs.Get("VolumeId")
+		return hasImg && hasTag && hasVol
+	}, 15*time.Second, 50*time.Millisecond,
+		"tagger async metadata/tag/volume fetches did not complete within 15s")
 
 	// Create metrics with existing attributes that should not be overwritten
 	md := createTestMetrics([]map[string]string{