[processor/tailsampling] fix benchmarks (open-telemetry#43947)

Several of the benchmarks were failing to run, this fixes them all for
future use.

Author: csmarchbanks

processor/tailsamplingprocessor/internal/idbatcher/id_batcher_test.go

@@ -49,18 +49,27 @@ func TestMinBufferedChannels(t *testing.T) {
 func BenchmarkConcurrentEnqueue(b *testing.B) {
 	ids := generateSequentialIDs(1)
 	batcher, err := New(10, 100, uint64(4*runtime.NumCPU()))
-	defer batcher.Stop()
 	require.NoError(b, err, "Failed to create Batcher")
 
 	ticker := time.NewTicker(100 * time.Millisecond)
-	defer ticker.Stop()
+	var wg sync.WaitGroup
+	defer func() {
+		batcher.Stop()
+		wg.Wait()
+		ticker.Stop()
+	}()
 	ticked := &atomic.Int64{}
 	received := &atomic.Int64{}
+	wg.Add(1)
 	go func() {
+		defer wg.Done()
 		for range ticker.C {
-			batch, _ := batcher.CloseCurrentAndTakeFirstBatch()
+			batch, more := batcher.CloseCurrentAndTakeFirstBatch()
 			ticked.Add(1)
 			received.Add(int64(len(batch)))
+			if !more {
+				return
+			}
 		}
 	}()
 

processor/tailsamplingprocessor/internal/sampling/util_test.go

@@ -63,6 +63,7 @@ func TestSetAttrOnScopeSpans_Many(t *testing.T) {
 
 func BenchmarkSetAttrOnScopeSpans(b *testing.B) {
 	for b.Loop() {
+		b.StopTimer()
 		traces := ptrace.NewTraces()
 
 		for range 5 {
@@ -86,6 +87,5 @@ func BenchmarkSetAttrOnScopeSpans(b *testing.B) {
 
 		b.StartTimer()
 		SetAttrOnScopeSpans(traceData, "test.attr", "value")
-		b.StopTimer()
 	}
 }

processor/tailsamplingprocessor/processor_benchmarks_test.go

@@ -4,13 +4,13 @@
 package tailsamplingprocessor
 
 import (
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/consumer/consumertest"
-	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.opentelemetry.io/collector/processor/processortest"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor/internal/metadata"
@@ -31,14 +31,16 @@ func BenchmarkSampling(b *testing.B) {
 	defer func() {
 		require.NoError(b, tsp.Shutdown(b.Context()))
 	}()
-	metrics := &policyMetrics{}
+	metrics := newPolicyMetrics(len(cfg.PolicyCfgs))
 	sampleBatches := make([]*samplingpolicy.TraceData, 0, len(batches))
 
-	for range batches {
+	for _, batch := range batches {
+		spanCount := &atomic.Int64{}
+		spanCount.Store(int64(batch.SpanCount()))
 		sampleBatches = append(sampleBatches, &samplingpolicy.TraceData{
-			ArrivalTime: time.Now(),
-			// SpanCount:       spanCount,
-			ReceivedBatches: ptrace.NewTraces(),
+			ArrivalTime:     time.Now(),
+			SpanCount:       spanCount,
+			ReceivedBatches: batch,
 		})
 	}