optimize histogram reservoir performance

Author: dashpole

CHANGELOG.md

@@ -40,6 +40,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - `WithInstrumentationAttributes` in `go.opentelemetry.io/otel/log` synchronously de-duplicates the passed attributes instead of delegating it to the returned `LoggerOption`. (#7266)
 - `Distinct` in `go.opentelemetry.io/otel/attribute` is no longer guaranteed to uniquely identify an attribute set. Collisions between `Distinct` values for different Sets are possible with extremely high cardinality (billions of series per instrument), but are highly unlikely. (#7175)
 - The default `TranslationStrategy` in `go.opentelemetry.io/exporters/prometheus` is changed from `otlptranslator.NoUTF8EscapingWithSuffixes` to `otlptranslator.UnderscoreEscapingWithSuffixes`. (#7421)
+- Improve the concurrent performance of `HistogramReservoir` in `go.opentelemetry.io/otel/sdk/metric/exemplar` by 10x. (#7443)
 
 <!-- Released section -->
 <!-- Don't change this section unless doing release -->

sdk/metric/exemplar/fixed_size_reservoir_test.go

@@ -45,8 +45,15 @@ func TestNewFixedSizeReservoirSamplingCorrectness(t *testing.T) {
 	}
 
 	var sum float64
-	for _, m := range r.measurements {
-		sum += m.Value.Float64()
+	for _, val := range r.measurements {
+		loaded := val.Load()
+		if loaded == nil {
+			continue
+		}
+		m := loaded.(*measurement)
+		if m != nil {
+			sum += m.Value.Float64()
+		}
 	}
 	mean := sum / float64(sampleSize)
 

sdk/metric/exemplar/histogram_reservoir.go

@@ -69,10 +69,5 @@ func (r *HistogramReservoir) Offer(ctx context.Context, t time.Time, v Value, a
 		panic("unknown value type")
 	}
 
-	idx := sort.SearchFloat64s(r.bounds, n)
-	m := newMeasurement(ctx, t, v, a)
-
-	r.mu.Lock()
-	defer r.mu.Unlock()
-	r.store(idx, m)
+	r.store(sort.SearchFloat64s(r.bounds, n), newMeasurement(ctx, t, v, a))
 }

sdk/metric/exemplar/storage.go

@@ -6,6 +6,7 @@ package exemplar // import "go.opentelemetry.io/otel/sdk/metric/exemplar"
 import (
 	"context"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"go.opentelemetry.io/otel/attribute"
@@ -19,15 +20,18 @@ type storage struct {
 	//
 	// This does not use []metricdata.Exemplar because it potentially would
 	// require an allocation for trace and span IDs in the hot path of Offer.
-	measurements []measurement
+	measurements []atomic.Value
 }
 
 func newStorage(n int) *storage {
-	return &storage{measurements: make([]measurement, n)}
+	return &storage{measurements: make([]atomic.Value, n)}
 }
 
-func (r *storage) store(idx int, m measurement) {
-	r.measurements[idx] = m
+func (r *storage) store(idx int, m *measurement) {
+	old := r.measurements[idx].Swap(m)
+	if old != nil {
+		mPool.Put(old)
+	}
 }
 
 // Collect returns all the held exemplars.
@@ -38,7 +42,12 @@ func (r *storage) Collect(dest *[]Exemplar) {
 	defer r.mu.Unlock()
 	*dest = reset(*dest, len(r.measurements), len(r.measurements))
 	var n int
-	for _, m := range r.measurements {
+	for _, val := range r.measurements {
+		loaded := val.Load()
+		if loaded == nil {
+			continue
+		}
+		m := loaded.(*measurement)
 		if !m.valid {
 			continue
 		}
@@ -58,20 +67,26 @@ type measurement struct {
 	// Value is the value of the measurement.
 	Value Value
 	// SpanContext is the SpanContext active when a measurement was made.
-	SpanContext trace.SpanContext
+	Ctx context.Context
 
 	valid bool
 }
 
+var mPool = sync.Pool{
+	New: func() any {
+		return &measurement{}
+	},
+}
+
 // newMeasurement returns a new non-empty Measurement.
-func newMeasurement(ctx context.Context, ts time.Time, v Value, droppedAttr []attribute.KeyValue) measurement {
-	return measurement{
-		FilteredAttributes: droppedAttr,
-		Time:               ts,
-		Value:              v,
-		SpanContext:        trace.SpanContextFromContext(ctx),
-		valid:              true,
-	}
+func newMeasurement(ctx context.Context, ts time.Time, v Value, droppedAttr []attribute.KeyValue) *measurement {
+	m := mPool.Get().(*measurement)
+	m.FilteredAttributes = droppedAttr
+	m.Time = ts
+	m.Value = v
+	m.Ctx = ctx
+	m.valid = true
+	return m
 }
 
 // exemplar returns m as an [Exemplar].
@@ -80,15 +95,16 @@ func (m measurement) exemplar(dest *Exemplar) {
 	dest.Time = m.Time
 	dest.Value = m.Value
 
-	if m.SpanContext.HasTraceID() {
-		traceID := m.SpanContext.TraceID()
+	sc := trace.SpanContextFromContext(m.Ctx)
+	if sc.HasTraceID() {
+		traceID := sc.TraceID()
 		dest.TraceID = traceID[:]
 	} else {
 		dest.TraceID = dest.TraceID[:0]
 	}
 
-	if m.SpanContext.HasSpanID() {
-		spanID := m.SpanContext.SpanID()
+	if sc.HasSpanID() {
+		spanID := sc.SpanID()
 		dest.SpanID = spanID[:]
 	} else {
 		dest.SpanID = dest.SpanID[:0]