add concurrent safe test for histogram reservoir

Author: dashpole

sdk/metric/exemplar/fixed_size_reservoir_test.go

@@ -58,3 +58,14 @@ func TestNewFixedSizeReservoirSamplingCorrectness(t *testing.T) {
 	// ensuring no bias in our random sampling algorithm.
 	assert.InDelta(t, 1/mean, intensity, 0.02) // Within 5σ.
 }
+
+func TestFixedSizeReservoirConcurrentSafe(t *testing.T) {
+	t.Run("Int64", reservoirConcurrentSafeTest[int64](func(n int) (ReservoirProvider, int) {
+		return FixedSizeReservoirProvider(n), n
+	}))
+
+	t.Run("Float64", reservoirConcurrentSafeTest[float64](func(n int) (ReservoirProvider, int) {
+		return FixedSizeReservoirProvider(n), n
+	}))
+
+}

sdk/metric/exemplar/histogram_reservoir_test.go

@@ -15,3 +15,15 @@ func TestHist(t *testing.T) {
 		return HistogramReservoirProvider(bounds), len(bounds)
 	}))
 }
+
+func TestHistogramReservoirConcurrentSafe(t *testing.T) {
+	bounds := []float64{0, 100}
+	t.Run("Int64", reservoirConcurrentSafeTest[int64](func(int) (ReservoirProvider, int) {
+		return HistogramReservoirProvider(bounds), len(bounds)
+	}))
+
+	t.Run("Float64", reservoirConcurrentSafeTest[float64](func(int) (ReservoirProvider, int) {
+		return HistogramReservoirProvider(bounds), len(bounds)
+	}))
+
+}

sdk/metric/exemplar/reservoir_test.go

@@ -4,6 +4,8 @@
 package exemplar
 
 import (
+	"context"
+	"sync"
 	"testing"
 	"time"
 
@@ -144,3 +146,64 @@ func ReservoirTest[N int64 | float64](f factory) func(*testing.T) {
 		})
 	}
 }
+
+func reservoirConcurrentSafeTest[N int64 | float64](f factory) func(*testing.T) {
+	return func(t *testing.T) {
+		t.Helper()
+		rp, n := f(1)
+		if n < 1 {
+			t.Skip("skipping, reservoir capacity less than 1:", n)
+		}
+		r := rp(*attribute.EmptySet())
+
+		var wg sync.WaitGroup
+
+		for i := range 5 {
+			wg.Add(1)
+			go func() {
+				ctx, ts, val, attrs := generateOfferInputs[N](t, i+1)
+				r.Offer(ctx, ts, val, attrs)
+				wg.Done()
+			}()
+		}
+		for range 2 {
+			wg.Add(1)
+			var dest []Exemplar
+			r.Collect(&dest)
+			for _, e := range dest {
+				validateExemplar[N](t, e)
+			}
+			wg.Done()
+		}
+		wg.Wait()
+	}
+}
+
+func generateOfferInputs[N int64 | float64](t *testing.T, i int) (context.Context, time.Time, Value, []attribute.KeyValue) {
+	sc := trace.NewSpanContext(trace.SpanContextConfig{
+		TraceID:    trace.TraceID([16]byte{byte(i)}),
+		SpanID:     trace.SpanID([8]byte{byte(i)}),
+		TraceFlags: trace.FlagsSampled,
+	})
+	ctx := trace.ContextWithSpanContext(t.Context(), sc)
+	ts := time.Unix(int64(i), int64(i))
+	val := NewValue(N(i))
+	attrs := []attribute.KeyValue{attribute.Int("i", i)}
+	return ctx, ts, val, attrs
+}
+
+func validateExemplar[N int64 | float64](t *testing.T, e Exemplar) {
+	i := 0
+	switch e.Value.Type() {
+	case Int64ValueType:
+		i = int(e.Value.Int64())
+	case Float64ValueType:
+		i = int(e.Value.Float64())
+	}
+	ctx, ts, _, attrs := generateOfferInputs[N](t, i)
+	sc := trace.SpanContextFromContext(ctx)
+	assert.Equal(t, sc.TraceID(), e.TraceID)
+	assert.Equal(t, sc.SpanID(), e.SpanID)
+	assert.Equal(t, ts, e.Time)
+	assert.Equal(t, attrs, e.FilteredAttributes)
+}