go/opentelemetry-go
1
0
Fork 0
mirror of https://github.com/open-telemetry/opentelemetry-go.git synced 2026-06-19 21:45:50 +02:00
Code Issues Releases Activity

Cleanup limitedSyncMap, and add tests (#8302)

Browse Source 
Making some cleanups to the existing limitedSyncMap before
https://github.com/open-telemetry/opentelemetry-go/pull/8276.

### Changes

limitedSyncMap uses generics to eliminate the need to use type
assertions in the aggregators themselves, and moves all type assertions
to be internal to the limitedSyncMap implementation

Testing

* Added two new tests for limitedSyncMap: a ConcurrentSafe test, and a
test that ensures the overflow limit is correctly enforced after it is
Cleared.

Co-authored-by: Robert Pająk <pellared@hotmail.com>
Co-authored-by: Tyler Yahn <MrAlias@users.noreply.github.com>
This commit is contained in:
David Ashpole
2026-06-09 14:03:38 -04:00
committed by GitHub
parent 3a1412d2b3
commit 41d99e7eb1
5 changed files with 153 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sdk/metric/internal/aggregate/atomic.go
+8 -8
View File
@@ -221,23 +221,23 @@ func (l *hotColdWaitGroup) swapHotAndWait() uint64 {
// limitedSyncMap is a sync.Map which enforces the aggregation limit on
// attribute sets and provides a Len() function.
type limitedSyncMap struct {
type limitedSyncMap[V any] struct {
sync.Map
aggLimit int
len int
lenMux sync.Mutex
}
func (m *limitedSyncMap) LoadOrStoreAttr(fltrAttr attribute.Set, newValue func(attribute.Set) any) any {
func (m *limitedSyncMap[V]) LoadOrStoreAttr(fltrAttr attribute.Set, newValue func(attribute.Set) V) V {
actual, loaded := m.Load(fltrAttr.Equivalent())
if loaded {
return actual
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
return actual.(V)
}
// Slow path: add a new attribute set.
m.lenMux.Lock()
@@ -248,7 +248,7 @@ func (m *limitedSyncMap) LoadOrStoreAttr(fltrAttr attribute.Set, newValue func(a
// concurrently.
actual, loaded = m.Load(fltrAttr.Equivalent())
if loaded {
return actual
return actual.(V)
}
if m.aggLimit > 0 && m.len >= m.aggLimit-1 {
@@ -258,17 +258,17 @@ func (m *limitedSyncMap) LoadOrStoreAttr(fltrAttr attribute.Set, newValue func(a
if !loaded {
m.len++
}
return actual
return actual.(V)
}
func (m *limitedSyncMap) Clear() {
func (m *limitedSyncMap[V]) Clear() {
m.lenMux.Lock()
defer m.lenMux.Unlock()
m.len = 0
m.Map.Clear()
}
func (m *limitedSyncMap) Len() int {
func (m *limitedSyncMap[V]) Len() int {
m.lenMux.Lock()
defer m.lenMux.Unlock()
return m.len
sdk/metric/internal/aggregate/atomic_test.go
+125
View File
@@ -10,6 +10,8 @@ import (
"testing"
"github.com/stretchr/testify/assert"
"go.opentelemetry.io/otel/attribute"
)
func TestAtomicSumAddFloatConcurrentSafe(t *testing.T) {
@@ -236,3 +238,126 @@ func benchmarkAtomicMinMax[N int64 | float64](b *testing.B) {
})
})
}
func TestLimitedSyncMapLimit(t *testing.T) {
m := limitedSyncMap[any]{aggLimit: 3}
newValue := func(attribute.Set) any { return new(int) }
attr1 := attribute.NewSet(attribute.String("key", "1"))
attr2 := attribute.NewSet(attribute.String("key", "2"))
attr3 := attribute.NewSet(attribute.String("key", "3"))
attr4 := attribute.NewSet(attribute.String("key", "4"))
// Add first (normal)
v1 := m.LoadOrStoreAttr(attr1, newValue)
assert.Equal(t, 1, m.Len())
// Add second (normal)
v2 := m.LoadOrStoreAttr(attr2, newValue)
assert.Equal(t, 2, m.Len())
// Add third (overflow)
v3 := m.LoadOrStoreAttr(attr3, newValue)
assert.Equal(t, 3, m.Len()) // Overflow counts as the 3rd entry
assert.NotSame(t, v1, v3)
assert.NotSame(t, v2, v3)
// Add fourth (overflow) - should return same overflow value
v4 := m.LoadOrStoreAttr(attr4, newValue)
assert.Same(t, v3, v4)
// Clear the map. Should be able to add new keys up to limit again.
m.Clear()
assert.Equal(t, 0, m.Len())
attr5 := attribute.NewSet(attribute.String("key", "5"))
attr6 := attribute.NewSet(attribute.String("key", "6"))
attr7 := attribute.NewSet(attribute.String("key", "7"))
attr8 := attribute.NewSet(attribute.String("key", "8"))
v5 := m.LoadOrStoreAttr(attr5, newValue)
assert.Equal(t, 1, m.Len())
v6 := m.LoadOrStoreAttr(attr6, newValue)
assert.Equal(t, 2, m.Len())
assert.NotSame(t, v5, v6, "Different keys should return different values")
v7 := m.LoadOrStoreAttr(attr7, newValue)
assert.Equal(t, 3, m.Len()) // Overflow counts as 3rd entry
assert.NotSame(t, v5, v7, "Overflow should be different from normal values")
assert.NotSame(t, v6, v7, "Overflow should be different from normal values")
v8 := m.LoadOrStoreAttr(attr8, newValue)
assert.Same(t, v7, v8, "Subsequent keys should return same overflow value")
}
func TestLimitedSyncMapConcurrentSafe(t *testing.T) {
m := limitedSyncMap[any]{aggLimit: 5}
newValue := func(attribute.Set) any { return 1 }
attr := attribute.NewSet(attribute.String("k", "v"))
var wg sync.WaitGroup
// 100 routines trying to read/write the same key
for range 100 {
wg.Go(func() {
m.LoadOrStoreAttr(attr, newValue)
})
}
wg.Wait()
assert.Equal(t, 1, m.Len())
// 10 routines trying to read/write DIFFERENT keys exceeding limit
var wg2 sync.WaitGroup
attrs := []attribute.Set{
attribute.NewSet(attribute.String("k", "1")),
attribute.NewSet(attribute.String("k", "2")),
attribute.NewSet(attribute.String("k", "3")),
attribute.NewSet(attribute.String("k", "4")),
attribute.NewSet(attribute.String("k", "5")),
attribute.NewSet(attribute.String("k", "6")),
attribute.NewSet(attribute.String("k", "7")),
attribute.NewSet(attribute.String("k", "8")),
attribute.NewSet(attribute.String("k", "9")),
attribute.NewSet(attribute.String("k", "10")),
}
for _, a := range attrs {
attrCopy := a
wg2.Go(func() {
m.LoadOrStoreAttr(attrCopy, newValue)
})
}
wg2.Wait()
// Map should be at limit (5)
assert.Equal(t, 5, m.Len())
}
func BenchmarkSyncMap(b *testing.B) {
attr := attribute.NewSet(attribute.String("key", "value"))
newValue := func(attribute.Set) any { return 1 }
b.Run("limitedSyncMap/LoadOrStoreNoClear", func(b *testing.B) {
m := limitedSyncMap[any]{aggLimit: 10}
b.ResetTimer()
for i := 0; i < b.N; i++ {
m.LoadOrStoreAttr(attr, newValue)
}
})
b.Run("limitedSyncMap/LoadOrStoreWithClear", func(b *testing.B) {
m := limitedSyncMap[any]{aggLimit: 10}
b.ResetTimer()
for i := 0; i < b.N; i++ {
m.Clear()
m.LoadOrStoreAttr(attr, newValue)
}
})
b.Run("limitedSyncMap/OnlyClear", func(b *testing.B) {
m := limitedSyncMap[any]{aggLimit: 10}
b.ResetTimer()
for i := 0; i < b.N; i++ {
m.Clear()
}
})
}
sdk/metric/internal/aggregate/histogram.go
+8 -8
View File
@@ -97,7 +97,7 @@ func (b *histogramPointCounters[N]) mergeIntoAndReset( // nolint:revive // Inten
// unused attribute sets do not report in subsequent collect() calls.
type deltaHistogram[N int64 | float64] struct {
hcwg hotColdWaitGroup
hotColdValMap [2]limitedSyncMap
hotColdValMap [2]limitedSyncMap[*histogramPoint[N]]
start time.Time
noMinMax bool
@@ -114,7 +114,7 @@ func (s *deltaHistogram[N]) measure(
) {
hotIdx := s.hcwg.start()
defer s.hcwg.done(hotIdx)
h := s.hotColdValMap[hotIdx].LoadOrStoreAttr(fltrAttr, func(attr attribute.Set) any {
h := s.hotColdValMap[hotIdx].LoadOrStoreAttr(fltrAttr, func(attr attribute.Set) *histogramPoint[N] {
r := s.newRes(attr)
_, isDrop := r.(*dropRes[N])
hPt := &histogramPoint[N]{
@@ -131,7 +131,7 @@ func (s *deltaHistogram[N]) measure(
histogramPointCounters: histogramPointCounters[N]{counts: make([]atomic.Uint64, len(s.bounds)+1)},
}
return hPt
}).(*histogramPoint[N])
})
// This search will return an index in the range [0, len(s.bounds)], where
// it will return len(s.bounds) if value is greater than the last element
@@ -171,7 +171,7 @@ func newDeltaHistogram[N int64 | float64](
noSum: noSum,
bounds: b,
newRes: r,
hotColdValMap: [2]limitedSyncMap{
hotColdValMap: [2]limitedSyncMap[*histogramPoint[N]]{
{aggLimit: limit},
{aggLimit: limit},
},
@@ -249,7 +249,7 @@ func (s *deltaHistogram[N]) collect(
// to reading. Unlike deltaHistogram, this maintains a single map so that the
// preserved attribute sets do not change when collect() is called.
type cumulativeHistogram[N int64 | float64] struct {
values limitedSyncMap
values limitedSyncMap[*hotColdHistogramPoint[N]]
start time.Time
noMinMax bool
@@ -278,7 +278,7 @@ func newCumulativeHistogram[N int64 | float64](
noSum: noSum,
bounds: b,
newRes: r,
values: limitedSyncMap{aggLimit: limit},
values: limitedSyncMap[*hotColdHistogramPoint[N]]{aggLimit: limit},
}
}
@@ -288,7 +288,7 @@ func (s *cumulativeHistogram[N]) measure(
fltrAttr attribute.Set,
droppedAttr []attribute.KeyValue,
) {
h := s.values.LoadOrStoreAttr(fltrAttr, func(attr attribute.Set) any {
h := s.values.LoadOrStoreAttr(fltrAttr, func(attr attribute.Set) *hotColdHistogramPoint[N] {
r := s.newRes(attr)
_, isDrop := r.(*dropRes[N])
hPt := &hotColdHistogramPoint[N]{
@@ -313,7 +313,7 @@ func (s *cumulativeHistogram[N]) measure(
},
}
return hPt
}).(*hotColdHistogramPoint[N])
})
// This search will return an index in the range [0, len(s.bounds)], where
// it will return len(s.bounds) if value is greater than the last element
sdk/metric/internal/aggregate/lastvalue.go
+6 -6
View File
@@ -24,7 +24,7 @@ type lastValuePoint[N int64 | float64] struct {
// lastValueMap summarizes a set of measurements as the last one made.
type lastValueMap[N int64 | float64] struct {
newRes func(attribute.Set) FilteredExemplarReservoir[N]
values limitedSyncMap
values limitedSyncMap[*lastValuePoint[N]]
}
func (s *lastValueMap[N]) measure(
@@ -33,7 +33,7 @@ func (s *lastValueMap[N]) measure(
fltrAttr attribute.Set,
droppedAttr []attribute.KeyValue,
) {
lv := s.values.LoadOrStoreAttr(fltrAttr, func(attr attribute.Set) any {
lv := s.values.LoadOrStoreAttr(fltrAttr, func(attr attribute.Set) *lastValuePoint[N] {
r := s.newRes(attr)
_, isDrop := r.(*dropRes[N])
p := &lastValuePoint[N]{
@@ -44,7 +44,7 @@ func (s *lastValueMap[N]) measure(
}
p.value.Store(value)
return p
}).(*lastValuePoint[N])
})
lv.value.Store(value)
if !lv.dropExemplars {
@@ -61,12 +61,12 @@ func newDeltaLastValue[N int64 | float64](
start: now(),
hotColdValMap: [2]lastValueMap[N]{
{
values: limitedSyncMap{aggLimit: limit},
newRes: r,
values: limitedSyncMap[*lastValuePoint[N]]{aggLimit: limit},
},
{
values: limitedSyncMap{aggLimit: limit},
newRes: r,
values: limitedSyncMap[*lastValuePoint[N]]{aggLimit: limit},
},
},
}
@@ -148,8 +148,8 @@ func newCumulativeLastValue[N int64 | float64](
) *cumulativeLastValue[N] {
return &cumulativeLastValue[N]{
lastValueMap: lastValueMap[N]{
values: limitedSyncMap{aggLimit: limit},
newRes: r,
values: limitedSyncMap[*lastValuePoint[N]]{aggLimit: limit},
},
start: now(),
}
sdk/metric/internal/aggregate/sum.go
+6 -6
View File
@@ -21,8 +21,8 @@ type sumValue[N int64 | float64] struct {
}
type sumValueMap[N int64 | float64] struct {
values limitedSyncMap
newRes func(attribute.Set) FilteredExemplarReservoir[N]
values limitedSyncMap[*sumValue[N]]
}
func (s *sumValueMap[N]) measure(
@@ -31,7 +31,7 @@ func (s *sumValueMap[N]) measure(
fltrAttr attribute.Set,
droppedAttr []attribute.KeyValue,
) {
sv := s.values.LoadOrStoreAttr(fltrAttr, func(attr attribute.Set) any {
sv := s.values.LoadOrStoreAttr(fltrAttr, func(attr attribute.Set) *sumValue[N] {
r := s.newRes(attr)
_, isDrop := r.(*dropRes[N])
return &sumValue[N]{
@@ -40,7 +40,7 @@ func (s *sumValueMap[N]) measure(
startTime: now(),
dropExemplars: isDrop,
}
}).(*sumValue[N])
})
sv.n.add(value)
// It is possible for collection to race with measurement and observe the
// exemplar in the batch of metrics after the add() for cumulative sums.
@@ -63,12 +63,12 @@ func newDeltaSum[N int64 | float64](
start: now(),
hotColdValMap: [2]sumValueMap[N]{
{
values: limitedSyncMap{aggLimit: limit},
newRes: r,
values: limitedSyncMap[*sumValue[N]]{aggLimit: limit},
},
{
values: limitedSyncMap{aggLimit: limit},
newRes: r,
values: limitedSyncMap[*sumValue[N]]{aggLimit: limit},
},
},
}
@@ -140,8 +140,8 @@ func newCumulativeSum[N int64 | float64](
monotonic: monotonic,
start: now(),
sumValueMap: sumValueMap[N]{
values: limitedSyncMap{aggLimit: limit},
newRes: r,
values: limitedSyncMap[*sumValue[N]]{aggLimit: limit},
},
}
}