[processor/spanmetrics] Fix flaky test (open-telemetry#18024)

* Fix flaky test from observed logs assertion

Signed-off-by: albertteoh <[email protected]>

.chloggen/spanmetricsprocessor-fix-flaky-test.yaml

@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: spanmetricsprocessor
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Fix a flaky test caused by a race condition between WaitGroup completion and observed logs being written and flushed.
+
+# One or more tracking issues related to the change
+issues: [18014]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:

processor/spanmetricsprocessor/processor_test.go

@@ -174,8 +174,15 @@ func TestProcessorConcurrentShutdown(t *testing.T) {
  }()
  }
  wg.Wait()
- allLogs := observedLogs.All()
- require.NotEmpty(t, allLogs)
+
+ // Allow time for log observer to sync all logs emitted.
+ // Even though the WaitGroup has been given the "done" signal, there's still a potential race condition
+ // between the WaitGroup being unblocked and when the logs will be flushed.
+ var allLogs []observer.LoggedEntry
+ assert.Eventually(t, func() bool {
+ allLogs = observedLogs.All()
+ return len(allLogs) > 0
+ }, time.Second, time.Millisecond*10)
 
  // Starting spanmetricsprocessor...
  // Started spanmetricsprocessor...
@@ -288,13 +295,13 @@ func TestProcessorConsumeMetricsErrors(t *testing.T) {
  wg.Wait()
 
  // Allow time for log observer to sync all logs emitted.
- // Unfortunately, we can't tell the log observer to wait until all logs have been synced/received.
- // Core/Logger.Sync() does not appear to achieve the desired behavior of syncing observedLogs with the logger.
- time.Sleep(time.Millisecond)
-
- // Verify
- allLogs := observedLogs.All()
- require.NotEmpty(t, allLogs)
+ // Even though the WaitGroup has been given the "done" signal, there's still a potential race condition
+ // between the WaitGroup being unblocked and when the logs will be flushed.
+ var allLogs []observer.LoggedEntry
+ assert.Eventually(t, func() bool {
+  allLogs = observedLogs.All()
+  return len(allLogs) > 0
+ }, time.Second, time.Millisecond*10)
 
  assert.Equal(t, "Failed ConsumeMetrics", allLogs[0].Message)
 }
@@ -412,7 +419,7 @@ func TestMetricKeyCache(t *testing.T) {
  require.NoError(t, err)
  // 2 key was cached, 1 key was evicted and cleaned after the processing
  assert.Eventually(t, func() bool {
- return assert.Equal(t, DimensionsCacheSize, p.metricKeyToDimensions.Len())
+ return p.metricKeyToDimensions.Len() == DimensionsCacheSize
  }, 10*time.Second, time.Millisecond*100)
 
  // consume another batch of traces
@@ -421,7 +428,7 @@ func TestMetricKeyCache(t *testing.T) {
 
  // 2 key was cached, other keys were evicted and cleaned after the processing
  assert.Eventually(t, func() bool {
- return assert.Equal(t, DimensionsCacheSize, p.metricKeyToDimensions.Len())
+ return p.metricKeyToDimensions.Len() == DimensionsCacheSize
  }, 10*time.Second, time.Millisecond*100)
 }