Use sync.Map for exponential histogram aggregations

CHANGELOG.md

@@ -18,6 +18,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ### Fixed
 
+- Prevent underflowed measurements from updating sum, min, or max in exponential histograms in `go.opentelemetry.io/otel/sdk/metric`. (#8077)
 - The simple span and log processors record `otel.sdk.processor.{span,log}.processed` when the record is submitted to the exporter instead of after the export completes, and no longer set `error.type` from the export outcome, in `go.opentelemetry.io/otel/sdk/trace` and `go.opentelemetry.io/otel/sdk/log`. (#8705)
 
 <!-- Released section -->

sdk/metric/internal/aggregate/aggregate.go

@@ -142,12 +142,20 @@ func (b Builder[N]) ExponentialBucketHistogram(
 	maxSize, maxScale int32,
 	noMinMax, noSum bool,
 ) (Measure[N], ComputeAggregation) {
-	h := newExponentialHistogram[N](maxSize, maxScale, noMinMax, noSum, b.AggregationLimit, b.resFunc())
 	switch b.Temporality {
 	case metricdata.DeltaTemporality:
-		return b.filter(h.measure), h.delta
+		h := newDeltaExpoHistogram[N](maxSize, maxScale, noMinMax, noSum, b.AggregationLimit, b.resFunc())
+		return b.filter(h.measure), h.collect
 	default:
-		return b.filter(h.measure), h.cumulative
+		measure, collect := newCumulativeExpoHistogram[N](
+			maxSize,
+			maxScale,
+			noMinMax,
+			noSum,
+			b.AggregationLimit,
+			b.resFunc(),
+		)
+		return b.filter(measure), collect
 	}
 }
 

sdk/metric/internal/aggregate/atomic.go

@@ -142,6 +142,11 @@ func (s *atomicMinMax[N]) Update(val N) {
 	}
 }
 
+// reset resets the internal state, and is not safe to call concurrently.
+func (s *atomicMinMax[N]) reset() {
+	s.set.Store(false)
+}
+
 // hotColdWaitGroup is a synchronization primitive which enables lockless
 // writes for concurrent writers and enables a reader to acquire exclusive
 // access to a snapshot of state including only completed operations.
@@ -226,6 +231,7 @@ type limitedSyncMap[V any] struct {
 	aggLimit int
 	len      int
 	lenMux   sync.Mutex
+	overflow atomic.Bool
 }
 
 // LoadOrStoreAttr performs lookup using lazy.Distinct() on the hot path without
@@ -238,11 +244,13 @@ func (m *limitedSyncMap[V]) LoadOrStoreAttr(lazy lazyFilteredAttributes, newValu
 	if loaded {
 		return actual.(V)
 	}
-	// If the overflow set exists, assume we have already overflowed and don't
-	// bother with the slow path below.
-	actual, loaded = m.Load(overflowSet.Equivalent())
-	if loaded {
-		return actual.(V)
+	// If aggregation overflow has already happened due to exceeding the limit,
+	// any new attribute set will be aggregated into the overflow set.
+	if m.aggLimit > 0 && m.overflow.Load() {
+		actual, loaded = m.Load(overflowSet.Equivalent())
+		if loaded {
+			return actual.(V)
+		}
 	}
 	// Slow path: add a new attribute set.
 	m.lenMux.Lock()
@@ -260,6 +268,7 @@ func (m *limitedSyncMap[V]) LoadOrStoreAttr(lazy lazyFilteredAttributes, newValu
 	if m.aggLimit > 0 && m.len >= m.aggLimit-1 {
 		fltrAttr = overflowSet
 		distinct = overflowSet.Equivalent()
+		m.overflow.Store(true)
 	} else {
 		fltrAttr = lazy.Set()
 	}
@@ -273,6 +282,7 @@ func (m *limitedSyncMap[V]) LoadOrStoreAttr(lazy lazyFilteredAttributes, newValu
 func (m *limitedSyncMap[V]) Clear() {
 	m.lenMux.Lock()
 	defer m.lenMux.Unlock()
+	m.overflow.Store(false)
 	m.len = 0
 	m.Map.Clear()
 }

sdk/metric/internal/aggregate/atomic_test.go

@@ -296,6 +296,42 @@ func TestLimitedSyncMapLimit(t *testing.T) {
 	assert.Same(t, v7, v8, "Subsequent keys should return same overflow value")
 }
 
+func TestLimitedSyncMapOverflowAttributeBeforeLimit(t *testing.T) {
+	newValue := func(attribute.Set) any { return new(int) }
+
+	t.Run("NoLimit", func(t *testing.T) {
+		var m limitedSyncMap[any]
+		v1 := loadOrStore(&m, overflowSet, newValue)
+		assert.Equal(t, 1, m.Len())
+
+		attr := attribute.NewSet(attribute.String("key", "1"))
+		v2 := loadOrStore(&m, attr, newValue)
+		assert.Equal(t, 2, m.Len())
+		assert.NotSame(t, v1, v2)
+	})
+
+	t.Run("WithLimit", func(t *testing.T) {
+		m := limitedSyncMap[any]{aggLimit: 3}
+		v1 := loadOrStore(&m, overflowSet, newValue)
+		assert.Equal(t, 1, m.Len())
+
+		attr2 := attribute.NewSet(attribute.String("key", "2"))
+		v2 := loadOrStore(&m, attr2, newValue)
+		assert.Equal(t, 2, m.Len())
+		assert.NotSame(t, v1, v2)
+
+		attr3 := attribute.NewSet(attribute.String("key", "3"))
+		v3 := loadOrStore(&m, attr3, newValue)
+		assert.Equal(t, 2, m.Len())
+		assert.Same(t, v1, v3)
+
+		attr4 := attribute.NewSet(attribute.String("key", "4"))
+		v4 := loadOrStore(&m, attr4, newValue)
+		assert.Equal(t, 2, m.Len())
+		assert.Same(t, v3, v4)
+	})
+}
+
 func TestLimitedSyncMapConcurrentSafe(t *testing.T) {
 	m := limitedSyncMap[any]{aggLimit: 5}
 	newValue := func(attribute.Set) any { return 1 }