diff --git a/.github/workflows/datadog-go.yaml b/.github/workflows/datadog-go.yaml index 0460c1fe..853e90c6 100644 --- a/.github/workflows/datadog-go.yaml +++ b/.github/workflows/datadog-go.yaml @@ -35,8 +35,13 @@ jobs: go-version: 1.14 - runs-on: macos-latest go-version: 1.15 + # These fail on macOS 26+ because of this bug https://github.com/golang/go/issues/68678 - runs-on: macos-latest go-version: 1.21 + - runs-on: macos-latest + go-version: 1.22 + - runs-on: macos-latest + go-version: 1.23 # No official linux/arm64 Go binaries before 1.16 - runs-on: ubuntu-24.04-arm go-version: 1.13 diff --git a/statsd/aggregator.go b/statsd/aggregator.go index ed18f8f5..aa2e36d2 100644 --- a/statsd/aggregator.go +++ b/statsd/aggregator.go @@ -1,10 +1,10 @@ package statsd import ( - "strings" "sync" "sync/atomic" "time" + "unsafe" ) type ( @@ -14,30 +14,61 @@ type ( bufferedMetricMap map[string]*bufferedMetric ) -type countShard struct { +const ( + // cacheLineSize is the size shards are padded to so that adjacent shards never + // share a cache line. Apple silicon and some other arm64 cores use 128-byte + // lines, so we pad to 128 to cover every platform we run on. Without this, + // several shards pack into one line and an RLock on one shard (which writes the + // mutex's reader count) bounces the neighbouring shards' mutexes between cores. + cacheLineSize = 128 + smallContextBufferSize = 512 + largeContextBufferSize = 4 * 1024 +) + +// The *Inner structs hold the actual shard state; the outer shard types embed +// them and append trailing padding so each shard fills a whole cache line (see +// cacheLineSize). Sizing the padding off the inner struct keeps the math +// correct automatically when fields are added or removed. + +type countShardInner struct { sync.RWMutex counts countsMap } -type gaugeShard struct { +type countShard struct { + countShardInner + _ [cacheLineSize - unsafe.Sizeof(countShardInner{})%cacheLineSize]byte +} + +type gaugeShardInner struct { sync.RWMutex gauges gaugesMap } -type setShard struct { +type gaugeShard struct { + gaugeShardInner + _ [cacheLineSize - unsafe.Sizeof(gaugeShardInner{})%cacheLineSize]byte +} + +type setShardInner struct { sync.RWMutex sets setsMap } +type setShard struct { + setShardInner + _ [cacheLineSize - unsafe.Sizeof(setShardInner{})%cacheLineSize]byte +} + type aggregator struct { nbContextGauge uint64 nbContextCount uint64 nbContextSet uint64 shardsCount int - countShards []*countShard - gaugeShards []*gaugeShard - setShards []*setShard + countShards []countShard + gaugeShards []gaugeShard + setShards []setShard histograms bufferedMetricContexts distributions bufferedMetricContexts @@ -60,20 +91,15 @@ func newAggregator(c *ClientEx, maxSamplesPerContext int64, shardsCount int) *ag agg := &aggregator{ client: c, shardsCount: shardsCount, - countShards: make([]*countShard, shardsCount), - gaugeShards: make([]*gaugeShard, shardsCount), - setShards: make([]*setShard, shardsCount), + countShards: make([]countShard, shardsCount), + gaugeShards: make([]gaugeShard, shardsCount), + setShards: make([]setShard, shardsCount), histograms: newBufferedContexts(newHistogramMetric, maxSamplesPerContext), distributions: newBufferedContexts(newDistributionMetric, maxSamplesPerContext), timings: newBufferedContexts(newTimingMetric, maxSamplesPerContext), closed: make(chan struct{}), stopChannelMode: make(chan struct{}), } - for i := 0; i < shardsCount; i++ { - agg.countShards[i] = &countShard{counts: countsMap{}} - agg.gaugeShards[i] = &gaugeShard{gauges: gaugesMap{}} - agg.setShards[i] = &setShard{sets: setsMap{}} - } return agg } @@ -155,10 +181,15 @@ func (a *aggregator) flushMetrics() []metric { // We reset the values to avoid sending 'zero' values for metrics not // sampled during this flush interval - for _, shard := range a.setShards { + for i := range a.setShards { + shard := &a.setShards[i] shard.Lock() sets := shard.sets - shard.sets = setsMap{} + if len(sets) == 0 { + shard.Unlock() + continue + } + shard.sets = nil shard.Unlock() for _, s := range sets { metrics = append(metrics, s.flushUnsafe()...) @@ -166,10 +197,15 @@ func (a *aggregator) flushMetrics() []metric { atomic.AddUint64(&a.nbContextSet, uint64(len(sets))) } - for _, shard := range a.gaugeShards { + for i := range a.gaugeShards { + shard := &a.gaugeShards[i] shard.Lock() gauges := shard.gauges - shard.gauges = gaugesMap{} + if len(gauges) == 0 { + shard.Unlock() + continue + } + shard.gauges = nil shard.Unlock() for _, g := range gauges { metrics = append(metrics, g.flushUnsafe()) @@ -177,10 +213,15 @@ func (a *aggregator) flushMetrics() []metric { atomic.AddUint64(&a.nbContextGauge, uint64(len(gauges))) } - for _, shard := range a.countShards { + for i := range a.countShards { + shard := &a.countShards[i] shard.Lock() counts := shard.counts - shard.counts = countsMap{} + if len(counts) == 0 { + shard.Unlock() + continue + } + shard.counts = nil shard.Unlock() for _, c := range counts { metrics = append(metrics, c.flushUnsafe()) @@ -195,69 +236,122 @@ func (a *aggregator) flushMetrics() []metric { return metrics } -// getContext returns the context for a metric name, tags, and cardinality. -// -// The context is the metric name, tags, and cardinality separated by separator symbols. -// It is not intended to be used as a metric name but as a unique key to aggregate -func getContext(name string, tags []string, cardinality Cardinality) string { - c, _ := getContextAndTags(name, tags, cardinality) - return c -} - -// getContextAndTags returns the context and tags for a metric name, tags, and cardinality. -// -// See getContext for usage for context -// The tags are the tags separated by a separator symbol and can be re-used to pass down to the writer -func getContextAndTags(name string, tags []string, cardinality Cardinality) (string, string) { - cardString := cardinality.String() +func getContextLength(name string, tags []string, cardString string) int { if len(tags) == 0 { if cardString == "" { - return name, "" + return len(name) } - return name + nameSeparatorSymbol + cardString, "" + return len(name) + len(nameSeparatorSymbol) + len(cardString) } n := len(name) + len(nameSeparatorSymbol) + len(tagSeparatorSymbol)*(len(tags)-1) - for _, s := range tags { - n += len(s) + for _, tag := range tags { + n += len(tag) } - var cardStringLen = 0 if cardString != "" { n += len(cardString) + len(cardSeparatorSymbol) - cardStringLen = len(cardString) + len(cardSeparatorSymbol) } + return n +} + +func appendContextAndHash(buf []byte, name string, tags []string, cardString string) ([]byte, uint32) { + h := init32 - var sb strings.Builder - sb.Grow(n) - sb.WriteString(name) - sb.WriteString(nameSeparatorSymbol) + buf, h = appendString32(buf, h, name) + if len(tags) == 0 { + if cardString == "" { + return buf, h + } + buf, h = appendString32(buf, h, nameSeparatorSymbol) + buf, h = appendString32(buf, h, cardString) + return buf, h + } + + buf, h = appendString32(buf, h, nameSeparatorSymbol) if cardString != "" { - sb.WriteString(cardString) - sb.WriteString(cardSeparatorSymbol) + buf, h = appendString32(buf, h, cardString) + buf, h = appendString32(buf, h, cardSeparatorSymbol) } - sb.WriteString(tags[0]) - for _, s := range tags[1:] { - sb.WriteString(tagSeparatorSymbol) - sb.WriteString(s) + buf, h = appendString32(buf, h, tags[0]) + for _, tag := range tags[1:] { + buf, h = appendString32(buf, h, tagSeparatorSymbol) + buf, h = appendString32(buf, h, tag) } + return buf, h +} + +func appendContext(buf []byte, name string, tags []string, cardString string) ([]byte, int) { + tagsStart := -1 - s := sb.String() + buf = append(buf, name...) + if len(tags) == 0 { + if cardString == "" { + return buf, tagsStart + } + buf = append(buf, nameSeparatorSymbol...) + buf = append(buf, cardString...) + return buf, tagsStart + } - return s, s[len(name)+len(nameSeparatorSymbol)+cardStringLen:] + buf = append(buf, nameSeparatorSymbol...) + if cardString != "" { + buf = append(buf, cardString...) + buf = append(buf, cardSeparatorSymbol...) + } + tagsStart = len(buf) + buf = append(buf, tags[0]...) + for _, tag := range tags[1:] { + buf = append(buf, tagSeparatorSymbol...) + buf = append(buf, tag...) + } + return buf, tagsStart } -func getShardIndex(shardsCount int, context string) int { +func getShardIndexFromHash(shardsCount int, contextHash uint32) int { if shardsCount <= 1 { return 0 } - return int(hashString32(context) % uint32(shardsCount)) + return int(contextHash % uint32(shardsCount)) } func (a *aggregator) count(name string, value int64, tags []string, cardinality Cardinality) error { - context := getContext(name, tags, cardinality) - shard := a.countShards[getShardIndex(a.shardsCount, context)] + if len(tags) == 0 && cardinality == CardinalityNotSet { + contextHash := uint32(0) + if a.shardsCount > 1 { + contextHash = hashString32(name) + } + return a.countWithStringContext(name, contextHash, name, value) + } + + cardString := cardinality.String() + contextLen := getContextLength(name, tags, cardString) + if contextLen <= smallContextBufferSize { + var contextBuffer [smallContextBufferSize]byte + return a.countWithContextBuffer(contextBuffer[:0], name, value, tags, cardinality, cardString) + } + return a.countWithLargeContextBuffer(contextLen, name, value, tags, cardinality, cardString) +} + +// countWithLargeContextBuffer keeps the 4 KiB stack array out of count's frame +// so the common small-context path is not penalized by the larger frame. +func (a *aggregator) countWithLargeContextBuffer(contextLen int, name string, value int64, tags []string, cardinality Cardinality, cardString string) error { + if contextLen <= largeContextBufferSize { + var contextBuffer [largeContextBufferSize]byte + return a.countWithContextBuffer(contextBuffer[:0], name, value, tags, cardinality, cardString) + } + return a.countWithContextBuffer(make([]byte, 0, contextLen), name, value, tags, cardinality, cardString) +} + +func (a *aggregator) countWithContextBuffer(contextBuffer []byte, name string, value int64, tags []string, cardinality Cardinality, cardString string) error { + contextHash := uint32(0) + if a.shardsCount > 1 { + contextBuffer, contextHash = appendContextAndHash(contextBuffer, name, tags, cardString) + } else { + contextBuffer, _ = appendContext(contextBuffer, name, tags, cardString) + } + shard := &a.countShards[getShardIndexFromHash(a.shardsCount, contextHash)] shard.RLock() - if count, found := shard.counts[context]; found { + if count, found := shard.counts[string(contextBuffer)]; found { count.sample(value) shard.RUnlock() return nil @@ -266,6 +360,36 @@ func (a *aggregator) count(name string, value int64, tags []string, cardinality metric := newCountMetric(name, value, tags, cardinality) + shard.Lock() + // Check if another goroutines hasn't created the value between the RUnlock and 'Lock' + if count, found := shard.counts[string(contextBuffer)]; found { + count.sample(value) + shard.Unlock() + return nil + } + + if shard.counts == nil { + shard.counts = countsMap{} + } + context := string(contextBuffer) + shard.counts[context] = metric + shard.Unlock() + return nil +} + +// handles the no-tags/no-cardinality fast path where the context key is the metric name itself. +func (a *aggregator) countWithStringContext(context string, contextHash uint32, name string, value int64) error { + shard := &a.countShards[getShardIndexFromHash(a.shardsCount, contextHash)] + shard.RLock() + if count, found := shard.counts[context]; found { + count.sample(value) + shard.RUnlock() + return nil + } + shard.RUnlock() + + metric := newCountMetric(name, value, nil, CardinalityNotSet) + shard.Lock() // Check if another goroutines hasn't created the value between the RUnlock and 'Lock' if count, found := shard.counts[context]; found { @@ -274,16 +398,52 @@ func (a *aggregator) count(name string, value int64, tags []string, cardinality return nil } + if shard.counts == nil { + shard.counts = countsMap{} + } shard.counts[context] = metric shard.Unlock() return nil } func (a *aggregator) gauge(name string, value float64, tags []string, cardinality Cardinality) error { - context := getContext(name, tags, cardinality) - shard := a.gaugeShards[getShardIndex(a.shardsCount, context)] + if len(tags) == 0 && cardinality == CardinalityNotSet { + contextHash := uint32(0) + if a.shardsCount > 1 { + contextHash = hashString32(name) + } + return a.gaugeWithStringContext(name, contextHash, name, value) + } + + cardString := cardinality.String() + contextLen := getContextLength(name, tags, cardString) + if contextLen <= smallContextBufferSize { + var contextBuffer [smallContextBufferSize]byte + return a.gaugeWithContextBuffer(contextBuffer[:0], name, value, tags, cardinality, cardString) + } + return a.gaugeWithLargeContextBuffer(contextLen, name, value, tags, cardinality, cardString) +} + +// gaugeWithLargeContextBuffer keeps the 4 KiB stack array out of gauge's frame +// so the common small-context path is not penalized by the larger frame. +func (a *aggregator) gaugeWithLargeContextBuffer(contextLen int, name string, value float64, tags []string, cardinality Cardinality, cardString string) error { + if contextLen <= largeContextBufferSize { + var contextBuffer [largeContextBufferSize]byte + return a.gaugeWithContextBuffer(contextBuffer[:0], name, value, tags, cardinality, cardString) + } + return a.gaugeWithContextBuffer(make([]byte, 0, contextLen), name, value, tags, cardinality, cardString) +} + +func (a *aggregator) gaugeWithContextBuffer(contextBuffer []byte, name string, value float64, tags []string, cardinality Cardinality, cardString string) error { + contextHash := uint32(0) + if a.shardsCount > 1 { + contextBuffer, contextHash = appendContextAndHash(contextBuffer, name, tags, cardString) + } else { + contextBuffer, _ = appendContext(contextBuffer, name, tags, cardString) + } + shard := &a.gaugeShards[getShardIndexFromHash(a.shardsCount, contextHash)] shard.RLock() - if gauge, found := shard.gauges[context]; found { + if gauge, found := shard.gauges[string(contextBuffer)]; found { gauge.sample(value) shard.RUnlock() return nil @@ -292,6 +452,35 @@ func (a *aggregator) gauge(name string, value float64, tags []string, cardinalit gauge := newGaugeMetric(name, value, tags, cardinality) + shard.Lock() + // Check if another goroutines hasn't created the value between the 'RUnlock' and 'Lock' + if gauge, found := shard.gauges[string(contextBuffer)]; found { + gauge.sample(value) + shard.Unlock() + return nil + } + if shard.gauges == nil { + shard.gauges = gaugesMap{} + } + context := string(contextBuffer) + shard.gauges[context] = gauge + shard.Unlock() + return nil +} + +// handles the no-tags/no-cardinality fast path where the context key is the metric name itself. +func (a *aggregator) gaugeWithStringContext(context string, contextHash uint32, name string, value float64) error { + shard := &a.gaugeShards[getShardIndexFromHash(a.shardsCount, contextHash)] + shard.RLock() + if gauge, found := shard.gauges[context]; found { + gauge.sample(value) + shard.RUnlock() + return nil + } + shard.RUnlock() + + gauge := newGaugeMetric(name, value, nil, CardinalityNotSet) + shard.Lock() // Check if another goroutines hasn't created the value between the 'RUnlock' and 'Lock' if gauge, found := shard.gauges[context]; found { @@ -299,16 +488,52 @@ func (a *aggregator) gauge(name string, value float64, tags []string, cardinalit shard.Unlock() return nil } + if shard.gauges == nil { + shard.gauges = gaugesMap{} + } shard.gauges[context] = gauge shard.Unlock() return nil } func (a *aggregator) set(name string, value string, tags []string, cardinality Cardinality) error { - context := getContext(name, tags, cardinality) - shard := a.setShards[getShardIndex(a.shardsCount, context)] + if len(tags) == 0 && cardinality == CardinalityNotSet { + contextHash := uint32(0) + if a.shardsCount > 1 { + contextHash = hashString32(name) + } + return a.setWithStringContext(name, contextHash, name, value) + } + + cardString := cardinality.String() + contextLen := getContextLength(name, tags, cardString) + if contextLen <= smallContextBufferSize { + var contextBuffer [smallContextBufferSize]byte + return a.setWithContextBuffer(contextBuffer[:0], name, value, tags, cardinality, cardString) + } + return a.setWithLargeContextBuffer(contextLen, name, value, tags, cardinality, cardString) +} + +// setWithLargeContextBuffer keeps the 4 KiB stack array out of set's frame so +// the common small-context path is not penalized by the larger frame. +func (a *aggregator) setWithLargeContextBuffer(contextLen int, name string, value string, tags []string, cardinality Cardinality, cardString string) error { + if contextLen <= largeContextBufferSize { + var contextBuffer [largeContextBufferSize]byte + return a.setWithContextBuffer(contextBuffer[:0], name, value, tags, cardinality, cardString) + } + return a.setWithContextBuffer(make([]byte, 0, contextLen), name, value, tags, cardinality, cardString) +} + +func (a *aggregator) setWithContextBuffer(contextBuffer []byte, name string, value string, tags []string, cardinality Cardinality, cardString string) error { + contextHash := uint32(0) + if a.shardsCount > 1 { + contextBuffer, contextHash = appendContextAndHash(contextBuffer, name, tags, cardString) + } else { + contextBuffer, _ = appendContext(contextBuffer, name, tags, cardString) + } + shard := &a.setShards[getShardIndexFromHash(a.shardsCount, contextHash)] shard.RLock() - if set, found := shard.sets[context]; found { + if set, found := shard.sets[string(contextBuffer)]; found { set.sample(value) shard.RUnlock() return nil @@ -317,6 +542,35 @@ func (a *aggregator) set(name string, value string, tags []string, cardinality C metric := newSetMetric(name, value, tags, cardinality) + shard.Lock() + // Check if another goroutines hasn't created the value between the 'RUnlock' and 'Lock' + if set, found := shard.sets[string(contextBuffer)]; found { + set.sample(value) + shard.Unlock() + return nil + } + if shard.sets == nil { + shard.sets = setsMap{} + } + context := string(contextBuffer) + shard.sets[context] = metric + shard.Unlock() + return nil +} + +// handles the no-tags/no-cardinality fast path where the context key is the metric name itself. +func (a *aggregator) setWithStringContext(context string, contextHash uint32, name string, value string) error { + shard := &a.setShards[getShardIndexFromHash(a.shardsCount, contextHash)] + shard.RLock() + if set, found := shard.sets[context]; found { + set.sample(value) + shard.RUnlock() + return nil + } + shard.RUnlock() + + metric := newSetMetric(name, value, nil, CardinalityNotSet) + shard.Lock() // Check if another goroutines hasn't created the value between the 'RUnlock' and 'Lock' if set, found := shard.sets[context]; found { @@ -324,6 +578,9 @@ func (a *aggregator) set(name string, value string, tags []string, cardinality C shard.Unlock() return nil } + if shard.sets == nil { + shard.sets = setsMap{} + } shard.sets[context] = metric shard.Unlock() return nil diff --git a/statsd/aggregator_benchmark_test.go b/statsd/aggregator_benchmark_test.go index 269c6ebb..bd037cc2 100644 --- a/statsd/aggregator_benchmark_test.go +++ b/statsd/aggregator_benchmark_test.go @@ -7,22 +7,52 @@ import ( // Prevent compiler from optimizing away function calls var benchErr error +var benchAggregator *aggregator +var benchMetrics []metric +func benchmarkMetricNames(n int) []string { + names := make([]string, n) + for i := 0; i < n; i++ { + names[i] = fmt.Sprintf("metric.%d", i) + } + return names +} + +// Pre-creates contexts so benchmarks can measure the steady-state update path +// separately from first-sample insertion. +func benchmarkPopulateAggregator(a *aggregator, names []string, tags []string) { + for _, name := range names { + _ = a.count(name, 1, tags, CardinalityLow) + _ = a.gauge(name, 10.0, tags, CardinalityLow) + _ = a.set(name, "val", tags, CardinalityLow) + } +} + +// Measures the cost of creating an aggregator at different shard counts. +func BenchmarkAggregatorConstruct(b *testing.B) { + for _, shards := range []int{1, 32, 256} { + b.Run(fmt.Sprintf("Shards_%d", shards), func(b *testing.B) { + b.ReportAllocs() + for i := 0; i < b.N; i++ { + benchAggregator = newAggregator(nil, 0, shards) + } + }) + } +} + +// Measures concurrent first-sample insertion and updates across many contexts. func BenchmarkAggregatorSharding(b *testing.B) { shardCounts := []int{1, 2, 3, 4, 5, 6, 8, 16, 32, 64, 128, 256} - // Pre-generate metric names to avoid measuring fmt.Sprintf performance const numMetrics = 100000 - metricNames := make([]string, numMetrics) - for i := 0; i < numMetrics; i++ { - metricNames[i] = fmt.Sprintf("metric.%d", i) - } + metricNames := benchmarkMetricNames(numMetrics) + tags := []string{"tag:1", "tag:2"} for _, shards := range shardCounts { b.Run(fmt.Sprintf("Shards_%d", shards), func(b *testing.B) { a := newAggregator(nil, 0, shards) - tags := []string{"tag:1", "tag:2"} + b.ReportAllocs() b.ResetTimer() b.RunParallel(func(pb *testing.PB) { var err error @@ -42,3 +72,74 @@ func BenchmarkAggregatorSharding(b *testing.B) { } } } + +// Measures the steady-state mixed update path across shard counts and tagset +// sizes. +func BenchmarkAggregatorSampleExistingByTagCount(b *testing.B) { + shardCounts := []int{1, 2, 4, 8, 16, 32, 64, 128, 256} + + const numMetrics = 100000 + metricNames := benchmarkMetricNames(numMetrics) + + for _, shards := range shardCounts { + for _, tagCount := range benchmarkTagCounts { + tags := benchGenerateSizedTags(tagCount, benchTagByteLen) + b.Run(fmt.Sprintf("Shards_%d_Tags_%d", shards, tagCount), func(b *testing.B) { + a := newAggregator(nil, 0, shards) + benchmarkPopulateAggregator(a, metricNames, tags) + + b.ReportAllocs() + b.ResetTimer() + b.RunParallel(func(pb *testing.PB) { + var err error + i := 0 + for pb.Next() { + name := metricNames[i%numMetrics] + i++ + err = a.count(name, 1, tags, CardinalityLow) + err = a.gauge(name, 10.0, tags, CardinalityLow) + err = a.set(name, "val", tags, CardinalityLow) + } + benchErr = err + }) + if benchErr != nil { + b.Fatal(benchErr) + } + }) + } + } +} + +// Measures flush overhead when no shard contains count, gauge, or set contexts +// to isolate the cost of scanning/resetting idle shards. +func BenchmarkAggregatorFlushEmpty(b *testing.B) { + for _, shards := range []int{1, 32, 256} { + b.Run(fmt.Sprintf("Shards_%d", shards), func(b *testing.B) { + a := newAggregator(nil, 0, shards) + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + benchMetrics = a.flushMetrics() + } + }) + } +} + +// Measures the sparse-flush case: one context is inserted for each simple +// metric type, then all shards are flushed. This is the hot path most affected +// by lazy map allocation and reset. +func BenchmarkAggregatorFlushSparse(b *testing.B) { + const shards = 256 + tags := []string{"tag:1", "tag:2"} + a := newAggregator(nil, 0, shards) + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + _ = a.count("metric.same", 1, tags, CardinalityLow) + _ = a.gauge("metric.same", 10.0, tags, CardinalityLow) + _ = a.set("metric.same", "val", tags, CardinalityLow) + benchMetrics = a.flushMetrics() + } +} diff --git a/statsd/aggregator_test.go b/statsd/aggregator_test.go index 896f3ba2..6f77ddcf 100644 --- a/statsd/aggregator_test.go +++ b/statsd/aggregator_test.go @@ -1,11 +1,11 @@ package statsd import ( - "fmt" "sort" "strings" "sync" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -13,7 +13,8 @@ import ( func getAllCounts(a *aggregator) countsMap { counts := countsMap{} - for _, shard := range a.countShards { + for i := range a.countShards { + shard := &a.countShards[i] shard.RLock() for k, v := range shard.counts { counts[k] = v @@ -25,7 +26,8 @@ func getAllCounts(a *aggregator) countsMap { func getAllGauges(a *aggregator) gaugesMap { gauges := gaugesMap{} - for _, shard := range a.gaugeShards { + for i := range a.gaugeShards { + shard := &a.gaugeShards[i] shard.RLock() for k, v := range shard.gauges { gauges[k] = v @@ -37,7 +39,8 @@ func getAllGauges(a *aggregator) gaugesMap { func getAllSets(a *aggregator) setsMap { sets := setsMap{} - for _, shard := range a.setShards { + for i := range a.setShards { + shard := &a.setShards[i] shard.RLock() for k, v := range shard.sets { sets[k] = v @@ -419,63 +422,6 @@ func TestAggregatorTagsCopy(t *testing.T) { } } -func TestGetContextAndTags(t *testing.T) { - tests := []struct { - testName string - name string - tags []string - wantContext string - wantTags string - }{ - { - testName: "no tags", - name: "name", - tags: nil, - wantContext: "name:low", - wantTags: "", - }, - { - testName: "one tag", - name: "name", - tags: []string{"tag1"}, - wantContext: "name:low|tag1", - wantTags: "tag1", - }, - { - testName: "two tags", - name: "name", - tags: []string{"tag1", "tag2"}, - wantContext: "name:low|tag1,tag2", - wantTags: "tag1,tag2", - }, - } - for _, test := range tests { - t.Run(test.testName, func(t *testing.T) { - gotContext, gotTags := getContextAndTags(test.name, test.tags, CardinalityLow) - assert.Equal(t, test.wantContext, gotContext) - assert.Equal(t, test.wantTags, gotTags) - }) - } -} - -func BenchmarkGetContext(b *testing.B) { - name := "test.metric" - tags := []string{"tag:tag", "foo:bar"} - for i := 0; i < b.N; i++ { - getContext(name, tags, CardinalityLow) - } - b.ReportAllocs() -} - -func BenchmarkGetContextNoTags(b *testing.B) { - name := "test.metric" - var tags []string - for i := 0; i < b.N; i++ { - getContext(name, tags, CardinalityLow) - } - b.ReportAllocs() -} - func TestAggregatorCardinalitySeparation(t *testing.T) { a := newAggregator(nil, 0, 8) tags := []string{"env:prod", "service:api"} @@ -546,6 +492,39 @@ func TestAggregatorCardinalitySeparation(t *testing.T) { assert.Contains(t, highSetValues, "value2") } +func TestAggregatorSampleNoTagsWithCardinality(t *testing.T) { + a := newAggregator(nil, 0, 8) + + for i := 0; i < 2; i++ { + a.gauge("gaugeTest", 21, nil, CardinalityLow) + gauges := getAllGauges(a) + assert.Len(t, gauges, 1) + assert.Contains(t, gauges, "gaugeTest:low") + + a.count("countTest", 21, nil, CardinalityLow) + counts := getAllCounts(a) + assert.Len(t, counts, 1) + assert.Contains(t, counts, "countTest:low") + + a.set("setTest", "value1", nil, CardinalityLow) + sets := getAllSets(a) + assert.Len(t, sets, 1) + assert.Contains(t, sets, "setTest:low") + + a.histogram("histogramTest", 21, nil, 1, CardinalityLow) + assert.Len(t, a.histograms.values, 1) + assert.Contains(t, a.histograms.values, "histogramTest:low") + + a.distribution("distributionTest", 21, nil, 1, CardinalityLow) + assert.Len(t, a.distributions.values, 1) + assert.Contains(t, a.distributions.values, "distributionTest:low") + + a.timing("timingTest", 21, nil, 1, CardinalityLow) + assert.Len(t, a.timings.values, 1) + assert.Contains(t, a.timings.values, "timingTest:low") + } +} + func TestAggregatorCardinalityPreservation(t *testing.T) { a := newAggregator(nil, 0, 8) tags := []string{"env:prod"} @@ -675,57 +654,267 @@ func TestAggregatorCardinalityEmptyVsNonEmpty(t *testing.T) { assert.Equal(t, float64(20), lowCard.fvalue) } -func TestAggregatorCardinalityContextGeneration(t *testing.T) { - // Test that getContextAndTags generates correct contexts for different cardinalities - tests := []struct { - name string - tags []string - cardinality Cardinality - wantContext string - wantTags string +// Verifies that count, gauge, and set correctly forward tags through all three +// context-buffer allocation paths: the small stack buffer (≤512 B), the large +// stack buffer (≤4 KiB), and heap allocation (>4 KiB) +func TestAggregatorContextBufferTiers(t *testing.T) { + // Each tag is sized so that len("m") + len(":") + len(tag) lands in the + // target tier: small ≤512, large 513–4096, heap ≥4097. + smallTag := strings.Repeat("x", 10) // total 12 + largeTag := strings.Repeat("x", smallContextBufferSize+1) // total 514 + heapTag := strings.Repeat("x", largeContextBufferSize+1) // total 4098 + + tiers := []struct { + name string + tags []string }{ - { - name: "test.metric", - tags: []string{"env:prod"}, - cardinality: CardinalityNotSet, - wantContext: "test.metric:env:prod", - wantTags: "env:prod", - }, - { - name: "test.metric", - tags: []string{"env:prod"}, - cardinality: CardinalityLow, - wantContext: "test.metric:low|env:prod", - wantTags: "env:prod", - }, - { - name: "test.metric", - tags: []string{"env:prod"}, - cardinality: CardinalityHigh, - wantContext: "test.metric:high|env:prod", - wantTags: "env:prod", - }, - { - name: "test.metric", - tags: []string{"env:prod", "service:api"}, - cardinality: CardinalityOrchestrator, - wantContext: "test.metric:orchestrator|env:prod,service:api", - wantTags: "env:prod,service:api", - }, - { - name: "test.metric", - tags: []string{}, - cardinality: CardinalityLow, - wantContext: "test.metric:low", - wantTags: "", - }, + {"small", []string{smallTag}}, + {"large", []string{largeTag}}, + {"heap", []string{heapTag}}, } - for _, test := range tests { - t.Run(fmt.Sprintf("%s_%s", test.name, test.cardinality.String()), func(t *testing.T) { - gotContext, gotTags := getContextAndTags(test.name, test.tags, test.cardinality) - assert.Equal(t, test.wantContext, gotContext) - assert.Equal(t, test.wantTags, gotTags) + for _, tier := range tiers { + tier := tier + t.Run("count/"+tier.name, func(t *testing.T) { + a := newAggregator(nil, 0, 8) + require.NoError(t, a.count("m", 3, tier.tags, CardinalityNotSet)) + require.NoError(t, a.count("m", 7, tier.tags, CardinalityNotSet)) + metrics := a.flushMetrics() + require.Len(t, metrics, 1) + assert.Equal(t, "m", metrics[0].name) + assert.Equal(t, int64(10), metrics[0].ivalue) + assert.Equal(t, tier.tags, metrics[0].tags) + }) + t.Run("gauge/"+tier.name, func(t *testing.T) { + a := newAggregator(nil, 0, 8) + require.NoError(t, a.gauge("m", 1.0, tier.tags, CardinalityNotSet)) + require.NoError(t, a.gauge("m", 2.0, tier.tags, CardinalityNotSet)) + metrics := a.flushMetrics() + require.Len(t, metrics, 1) + assert.Equal(t, "m", metrics[0].name) + assert.Equal(t, 2.0, metrics[0].fvalue) + assert.Equal(t, tier.tags, metrics[0].tags) + }) + t.Run("set/"+tier.name, func(t *testing.T) { + a := newAggregator(nil, 0, 8) + require.NoError(t, a.set("m", "a", tier.tags, CardinalityNotSet)) + require.NoError(t, a.set("m", "b", tier.tags, CardinalityNotSet)) + metrics := a.flushMetrics() + require.Len(t, metrics, 2) + for _, m := range metrics { + assert.Equal(t, "m", m.name) + assert.Equal(t, tier.tags, m.tags) + } }) } } + +// Verifies the StringContext fast path taken when a metric has no tags and +// CardinalityNotSet. The context key must be the bare metric name, aggregation +// semantics must match the regular path, and flushed metrics must carry nil +// tags and CardinalityNotSet. +func TestAggregatorNoTagsFastPath(t *testing.T) { + a := newAggregator(nil, 0, 8) + + // Count: values accumulate + require.NoError(t, a.count("my.count", 3, nil, CardinalityNotSet)) + require.NoError(t, a.count("my.count", 7, nil, CardinalityNotSet)) + counts := getAllCounts(a) + assert.Len(t, counts, 1) + assert.Contains(t, counts, "my.count") + + // Gauge: last value wins + require.NoError(t, a.gauge("my.gauge", 1.0, nil, CardinalityNotSet)) + require.NoError(t, a.gauge("my.gauge", 2.0, nil, CardinalityNotSet)) + gauges := getAllGauges(a) + assert.Len(t, gauges, 1) + assert.Contains(t, gauges, "my.gauge") + + // Set: unique values tracked + require.NoError(t, a.set("my.set", "a", nil, CardinalityNotSet)) + require.NoError(t, a.set("my.set", "a", nil, CardinalityNotSet)) + require.NoError(t, a.set("my.set", "b", nil, CardinalityNotSet)) + sets := getAllSets(a) + assert.Len(t, sets, 1) + assert.Contains(t, sets, "my.set") + + metrics := a.flushMetrics() + + var gotCount, gotGauge, gotSet *metric + for i := range metrics { + m := &metrics[i] + switch m.name { + case "my.count": + gotCount = m + case "my.gauge": + gotGauge = m + case "my.set": + gotSet = m + } + } + + require.NotNil(t, gotCount) + assert.Equal(t, int64(10), gotCount.ivalue) + assert.Nil(t, gotCount.tags) + assert.Equal(t, CardinalityNotSet, gotCount.cardinality) + + require.NotNil(t, gotGauge) + assert.Equal(t, float64(2.0), gotGauge.fvalue) + assert.Nil(t, gotGauge.tags) + assert.Equal(t, CardinalityNotSet, gotGauge.cardinality) + + require.NotNil(t, gotSet) + assert.Nil(t, gotSet.tags) + assert.Equal(t, CardinalityNotSet, gotSet.cardinality) +} + +// Checks that the no-tags fast path and the regular (tagged) path produce +// separate context keys and do not collide. +func TestAggregatorNoTagsFastPathIsolation(t *testing.T) { + a := newAggregator(nil, 0, 8) + tags := []string{"env:prod"} + + require.NoError(t, a.count("my.count", 1, nil, CardinalityNotSet)) + require.NoError(t, a.count("my.count", 2, tags, CardinalityNotSet)) + require.NoError(t, a.count("my.count", 4, tags, CardinalityLow)) + + counts := getAllCounts(a) + assert.Len(t, counts, 3) + assert.Contains(t, counts, "my.count") + assert.Contains(t, counts, "my.count:env:prod") + assert.Contains(t, counts, "my.count:low|env:prod") +} + +// Verifies the double-check locking in the StringContext fast path under +// concurrent writers. +func TestAggregatorNoTagsFastPathConcurrency(t *testing.T) { + a := newAggregator(nil, 0, 8) + + const goroutines = 20 + const samplesEach = 100 + + var wg sync.WaitGroup + wg.Add(goroutines) + for i := 0; i < goroutines; i++ { + go func() { + defer wg.Done() + for j := 0; j < samplesEach; j++ { + a.count("concurrent.count", 1, nil, CardinalityNotSet) + a.gauge("concurrent.gauge", float64(j), nil, CardinalityNotSet) + a.set("concurrent.set", "v", nil, CardinalityNotSet) + } + }() + } + wg.Wait() + + counts := getAllCounts(a) + assert.Len(t, counts, 1) + assert.Contains(t, counts, "concurrent.count") + + gauges := getAllGauges(a) + assert.Len(t, gauges, 1) + assert.Contains(t, gauges, "concurrent.gauge") + + sets := getAllSets(a) + assert.Len(t, sets, 1) + assert.Contains(t, sets, "concurrent.set") + + metrics := a.flushMetrics() + var gotCount *metric + for i := range metrics { + if metrics[i].name == "concurrent.count" { + gotCount = &metrics[i] + } + } + require.NotNil(t, gotCount) + assert.Equal(t, int64(goroutines*samplesEach), gotCount.ivalue) +} + +func TestContextHelpersNoTagsNoCardinality(t *testing.T) { + name := "test.metric" + + assert.Equal(t, len(name), getContextLength(name, nil, "")) + + withHash, hash := appendContextAndHash(nil, name, nil, "") + assert.Equal(t, []byte(name), withHash) + assert.Equal(t, hashString32(name), hash) + + context, tagsStart := appendContext(nil, name, nil, "") + assert.Equal(t, []byte(name), context) + assert.Equal(t, -1, tagsStart) +} + +func TestAggregatorContextBufferSecondCheckLocking(t *testing.T) { + const goroutines = 128 + + tags := []string{"env:prod"} + + runConcurrentMetricInsertions := func(t *testing.T, lockShard func(a *aggregator), unlockShard func(a *aggregator), sample func(a *aggregator, i int), assertAfter func(t *testing.T, a *aggregator)) { + t.Helper() + a := newAggregator(nil, 0, 1) + lockShard(a) + start := make(chan struct{}) + var wg sync.WaitGroup + wg.Add(goroutines) + for i := 0; i < goroutines; i++ { + i := i + go func() { + defer wg.Done() + <-start + sample(a, i) + }() + } + close(start) + time.Sleep(20 * time.Millisecond) + unlockShard(a) + wg.Wait() + assertAfter(t, a) + } + + t.Run("count", func(t *testing.T) { + runConcurrentMetricInsertions(t, func(a *aggregator) { + a.countShards[0].RLock() + }, func(a *aggregator) { + a.countShards[0].RUnlock() + }, func(a *aggregator, _ int) { + require.NoError(t, a.count("race.count", 1, tags, CardinalityLow)) + }, func(t *testing.T, a *aggregator) { + counts := getAllCounts(a) + require.Len(t, counts, 1) + m, found := counts["race.count:low|env:prod"] + require.True(t, found) + assert.Equal(t, int64(goroutines), m.value) + }) + }) + + t.Run("gauge", func(t *testing.T) { + runConcurrentMetricInsertions(t, func(a *aggregator) { + a.gaugeShards[0].RLock() + }, func(a *aggregator) { + a.gaugeShards[0].RUnlock() + }, func(a *aggregator, i int) { + require.NoError(t, a.gauge("race.gauge", float64(i), tags, CardinalityLow)) + }, func(t *testing.T, a *aggregator) { + gauges := getAllGauges(a) + require.Len(t, gauges, 1) + _, found := gauges["race.gauge:low|env:prod"] + require.True(t, found) + }) + }) + + t.Run("set", func(t *testing.T) { + runConcurrentMetricInsertions(t, func(a *aggregator) { + a.setShards[0].RLock() + }, func(a *aggregator) { + a.setShards[0].RUnlock() + }, func(a *aggregator, i int) { + require.NoError(t, a.set("race.set", string(rune('a'+(i%26))), tags, CardinalityLow)) + }, func(t *testing.T, a *aggregator) { + sets := getAllSets(a) + require.Len(t, sets, 1) + _, found := sets["race.set:low|env:prod"] + require.True(t, found) + }) + }) +} + diff --git a/statsd/buffered_metric_context.go b/statsd/buffered_metric_context.go index 85cab2a1..76ac7901 100644 --- a/statsd/buffered_metric_context.go +++ b/statsd/buffered_metric_context.go @@ -55,32 +55,85 @@ func (bc *bufferedMetricContexts) flush(metrics []metric) []metric { } func (bc *bufferedMetricContexts) sample(name string, value float64, tags []string, rate float64, cardinality Cardinality) error { - keepingSample := shouldSample(rate, bc.random, &bc.randomLock) - - // If we don't keep the sample, return early. If we do keep the sample - // we end up storing the *first* observed sampling rate in the metric. - // This is the *wrong* behavior but it's the one we had before and the alternative would increase lock contention too - // much with the current code. - // TODO: change this behavior in the future, probably by introducing thread-local storage and lockless stuctures. - // If this code is removed, also remove the observed sampling rate in the metric and fix `bufferedMetric.flushUnsafe()` - if !keepingSample { + // For user-sampled metrics, return early when the sample is dropped. If we + // keep it, the metric stores the first observed sampling rate. This matches + // the existing behavior; correcting it would require tracking later rates + // without adding contention on this hot path. + // TODO: change this behavior in the future, probably by introducing + // thread-local storage and lockless structures. If this early return is + // removed, also remove the observed sampling rate in bufferedMetric and fix + // bufferedMetric.flushUnsafe. + if rate < 1 && !shouldSample(rate, bc.random, &bc.randomLock) { return nil } - context, stringTags := getContextAndTags(name, tags, cardinality) + if len(tags) == 0 && cardinality == CardinalityNotSet { + bc.mutex.RLock() + v := bc.values[name] + bc.mutex.RUnlock() + if v != nil { + v.maybeKeepSample(value, bc.random, &bc.randomLock) + return nil + } + + // Create it if it wasn't found + bc.mutex.Lock() + // It might have been created by another goroutine since last call + v = bc.values[name] + if v == nil { + // If we might keep a sample that we should have skipped, but that should not drastically affect performances. + bc.values[name] = bc.newMetric(name, value, "", rate, cardinality) + // We added a new value, we need to unlock the mutex and quit + bc.mutex.Unlock() + return nil + } + bc.mutex.Unlock() + + // Now we can keep the sample. + v.maybeKeepSample(value, bc.random, &bc.randomLock) + + return nil + } + + cardString := cardinality.String() + contextLen := getContextLength(name, tags, cardString) + if contextLen <= smallContextBufferSize { + var contextBuffer [smallContextBufferSize]byte + return bc.sampleWithContextBuffer(contextBuffer[:0], name, value, tags, rate, cardinality, cardString) + } + return bc.sampleWithLargeContextBuffer(contextLen, name, value, tags, rate, cardinality, cardString) +} + +// sampleWithLargeContextBuffer keeps the 4 KiB stack array out of sample's +// frame so the common small-context path is not penalized by the larger frame. +func (bc *bufferedMetricContexts) sampleWithLargeContextBuffer(contextLen int, name string, value float64, tags []string, rate float64, cardinality Cardinality, cardString string) error { + if contextLen <= largeContextBufferSize { + var contextBuffer [largeContextBufferSize]byte + return bc.sampleWithContextBuffer(contextBuffer[:0], name, value, tags, rate, cardinality, cardString) + } + return bc.sampleWithContextBuffer(make([]byte, 0, contextLen), name, value, tags, rate, cardinality, cardString) +} + +func (bc *bufferedMetricContexts) sampleWithContextBuffer(contextBuffer []byte, name string, value float64, tags []string, rate float64, cardinality Cardinality, cardString string) error { + contextBuffer, tagsStart := appendContext(contextBuffer, name, tags, cardString) var v *bufferedMetric bc.mutex.RLock() - v, _ = bc.values[context] + v, _ = bc.values[string(contextBuffer)] bc.mutex.RUnlock() // Create it if it wasn't found if v == nil { bc.mutex.Lock() // It might have been created by another goroutine since last call - v, _ = bc.values[context] + v, _ = bc.values[string(contextBuffer)] if v == nil { // If we might keep a sample that we should have skipped, but that should not drastically affect performances. + context := string(contextBuffer) + stringTags := "" + if tagsStart >= 0 { + stringTags = context[tagsStart:] + } bc.values[context] = bc.newMetric(name, value, stringTags, rate, cardinality) // We added a new value, we need to unlock the mutex and quit bc.mutex.Unlock() @@ -89,12 +142,8 @@ func (bc *bufferedMetricContexts) sample(name string, value float64, tags []stri bc.mutex.Unlock() } - // Now we can keep the sample or skip it - if keepingSample { - v.maybeKeepSample(value, bc.random, &bc.randomLock) - } else { - v.skipSample() - } + // Now we can keep the sample. + v.maybeKeepSample(value, bc.random, &bc.randomLock) return nil } diff --git a/statsd/buffered_metric_context_test.go b/statsd/buffered_metric_context_test.go new file mode 100644 index 00000000..9ce0481f --- /dev/null +++ b/statsd/buffered_metric_context_test.go @@ -0,0 +1,74 @@ +package statsd + +import ( + "math/rand" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestBufferedMetricContextsSampleNoTagsPaths(t *testing.T) { + t.Run("drops_metric_when_sampling_rejects", func(t *testing.T) { + contexts := newBufferedContexts(newHistogramMetric, 0) + contexts.random = rand.New(rand.NewSource(1)) + + require.NoError(t, contexts.sample("metric.drop", 1, nil, -1, CardinalityNotSet)) + assert.Empty(t, contexts.values) + }) + + t.Run("creates_and_reuses_metric_no_tags", func(t *testing.T) { + contexts := newBufferedContexts(newHistogramMetric, 0) + + require.NoError(t, contexts.sample("metric.no_tags", 1, nil, 1, CardinalityNotSet)) + require.NoError(t, contexts.sample("metric.no_tags", 2, nil, 1, CardinalityNotSet)) + + v := contexts.values["metric.no_tags"] + require.NotNil(t, v) + assert.Equal(t, int64(2), v.storedSamples) + assert.Equal(t, []float64{1, 2}, v.data[:v.storedSamples]) + }) +} + +func TestBufferedMetricContextsSampleNoTagsSecondCheckLocking(t *testing.T) { + const goroutines = 128 + contexts := newBufferedContexts(newHistogramMetric, 0) + contexts.mutex.RLock() + + start := make(chan struct{}) + var wg sync.WaitGroup + wg.Add(goroutines) + for i := 0; i < goroutines; i++ { + i := i + go func() { + defer wg.Done() + <-start + require.NoError(t, contexts.sample("metric.concurrent", float64(i), nil, 1, CardinalityNotSet)) + }() + } + close(start) + time.Sleep(20 * time.Millisecond) + contexts.mutex.RUnlock() + wg.Wait() + + v := contexts.values["metric.concurrent"] + require.NotNil(t, v) + assert.Equal(t, int64(goroutines), v.storedSamples) +} + + +func TestBufferedMetricContextsLargeContextBuffers(t *testing.T) { + contexts := newBufferedContexts(newHistogramMetric, 0) + + largeTag := strings.Repeat("x", smallContextBufferSize+1) + heapTag := strings.Repeat("x", largeContextBufferSize+1) + + require.NoError(t, contexts.sample("metric.large", 1, []string{largeTag}, 1, CardinalityLow)) + require.NoError(t, contexts.sample("metric.heap", 2, []string{heapTag}, 1, CardinalityLow)) + + assert.Contains(t, contexts.values, "metric.large:low|"+largeTag) + assert.Contains(t, contexts.values, "metric.heap:low|"+heapTag) +} diff --git a/statsd/context_benchmark_test.go b/statsd/context_benchmark_test.go new file mode 100644 index 00000000..a80c5322 --- /dev/null +++ b/statsd/context_benchmark_test.go @@ -0,0 +1,256 @@ +package statsd + +import ( + "fmt" + "strings" + "testing" +) + +var benchContextErr error + +func benchGenerateTags(n int) []string { + tags := make([]string, n) + for i := range tags { + tags[i] = fmt.Sprintf("tag%d:tag%d", i, i) + } + return tags +} + +// Length of a single "key:value" tag used by the tag-count sweeps which covers +// the P99 tag-value pair's length. +const benchTagByteLen = 55 + +// The set of tag counts swept by the *ByTagCount benchmarks. +var benchmarkTagCounts = []int{5, 10, 40, 60, 100} + +// Returns n distinct tags that are each exactly byteLen bytes long, shaped as +// a realistic "key:value" pair. +func benchGenerateSizedTags(n, byteLen int) []string { + tags := make([]string, n) + for i := range tags { + key := fmt.Sprintf("key.%d:", i) + if len(key) > byteLen { + key = key[:byteLen] + } + tags[i] = key + strings.Repeat("v", byteLen-len(key)) + } + return tags +} + +func benchmarkContextCases() []struct { + name string + metricName string + tags []string + cardinality Cardinality +} { + return []struct { + name string + metricName string + tags []string + cardinality Cardinality + }{ + { + name: "NoTags_NotSet", + metricName: "test.metric", + cardinality: CardinalityNotSet, + }, + { + name: "NoTags_Low", + metricName: "test.metric", + cardinality: CardinalityLow, + }, + { + name: "OneTag_NotSet", + metricName: "test.metric", + tags: []string{"env:prod"}, + cardinality: CardinalityNotSet, + }, + { + name: "TwoTags_NotSet", + metricName: "test.metric", + tags: []string{"env:prod", "service:api"}, + cardinality: CardinalityNotSet, + }, + { + name: "OneTag_Low", + metricName: "test.metric", + tags: []string{"env:prod"}, + cardinality: CardinalityLow, + }, + { + name: "TwoTags_Low", + metricName: "test.metric", + tags: []string{"env:prod", "service:api"}, + cardinality: CardinalityLow, + }, + { + name: "FiveTags_Low", + metricName: "test.metric", + tags: []string{"env:prod", "service:api", "version:2026.06.09", "region:us-east-1", "endpoint:/checkout"}, + cardinality: CardinalityLow, + }, + { + name: "FiveTags_Orchestrator", + metricName: "test.metric", + tags: []string{"env:prod", "service:api", "version:2026.06.09", "region:us-east-1", "pod_name:checkout-api-7f7c8d6f98-b6l9x"}, + cardinality: CardinalityOrchestrator, + }, + { + name: "Tags100_Low", + metricName: "test.metric", + tags: benchGenerateTags(100), + cardinality: CardinalityLow, + }, + { + name: "Tags100_NotSet", + metricName: "test.metric", + tags: benchGenerateTags(100), + cardinality: CardinalityNotSet, + }, + { + name: "Tags100_High", + metricName: "test.metric", + tags: benchGenerateTags(100), + cardinality: CardinalityHigh, + }, + } +} + +func BenchmarkAggregatorSampleExistingByMetricType(b *testing.B) { + const ( + shards = 256 + numMetrics = 100000 + ) + metricNames := benchmarkMetricNames(numMetrics) + tags := []string{"tag:1", "tag:2"} + + benchmarks := []struct { + name string + populate func(*aggregator, string) error + sample func(*aggregator, string) error + }{ + { + name: "Count", + populate: func(a *aggregator, name string) error { + return a.count(name, 1, tags, CardinalityLow) + }, + sample: func(a *aggregator, name string) error { + return a.count(name, 1, tags, CardinalityLow) + }, + }, + { + name: "Gauge", + populate: func(a *aggregator, name string) error { + return a.gauge(name, 10.0, tags, CardinalityLow) + }, + sample: func(a *aggregator, name string) error { + return a.gauge(name, 10.0, tags, CardinalityLow) + }, + }, + { + name: "Set", + populate: func(a *aggregator, name string) error { + return a.set(name, "initial", tags, CardinalityLow) + }, + sample: func(a *aggregator, name string) error { + return a.set(name, "existing", tags, CardinalityLow) + }, + }, + } + + for _, bm := range benchmarks { + bm := bm + b.Run(bm.name, func(b *testing.B) { + a := newAggregator(nil, 0, shards) + for _, name := range metricNames { + if err := bm.populate(a, name); err != nil { + b.Fatal(err) + } + } + + b.ReportAllocs() + b.ResetTimer() + b.RunParallel(func(pb *testing.PB) { + var err error + i := 0 + for pb.Next() { + name := metricNames[i%numMetrics] + i++ + err = bm.sample(a, name) + } + benchContextErr = err + }) + if benchContextErr != nil { + b.Fatal(benchContextErr) + } + }) + } +} + +func BenchmarkBufferedMetricContextsSampleExistingByShape(b *testing.B) { + const numMetrics = 100000 + metricNames := benchmarkMetricNames(numMetrics) + + for _, tc := range benchmarkContextCases() { + tc := tc + b.Run(tc.name, func(b *testing.B) { + contexts := newBufferedContexts(newHistogramMetric, 1) + for _, name := range metricNames { + if err := contexts.sample(name, 1.0, tc.tags, 1.0, tc.cardinality); err != nil { + b.Fatal(err) + } + } + + b.ReportAllocs() + b.ResetTimer() + b.RunParallel(func(pb *testing.PB) { + var err error + i := 0 + for pb.Next() { + name := metricNames[i%numMetrics] + i++ + err = contexts.sample(name, float64(i), tc.tags, 1.0, tc.cardinality) + } + benchContextErr = err + }) + if benchContextErr != nil { + b.Fatal(benchContextErr) + } + }) + } +} + +// Measures the histogram/distribution/timing sample path as the number of tags +// grows, exercising the inline, large-stack, and heap context buffers. +func BenchmarkBufferedMetricContextsSampleExistingByTagCount(b *testing.B) { + const numMetrics = 100000 + metricNames := benchmarkMetricNames(numMetrics) + + for _, tagCount := range benchmarkTagCounts { + tags := benchGenerateSizedTags(tagCount, benchTagByteLen) + b.Run(fmt.Sprintf("Tags_%d", tagCount), func(b *testing.B) { + contexts := newBufferedContexts(newHistogramMetric, 1) + for _, name := range metricNames { + if err := contexts.sample(name, 1.0, tags, 1.0, CardinalityLow); err != nil { + b.Fatal(err) + } + } + + b.ReportAllocs() + b.ResetTimer() + b.RunParallel(func(pb *testing.PB) { + var err error + i := 0 + for pb.Next() { + name := metricNames[i%numMetrics] + i++ + err = contexts.sample(name, float64(i), tags, 1.0, CardinalityLow) + } + benchContextErr = err + }) + if benchContextErr != nil { + b.Fatal(benchContextErr) + } + }) + } +} diff --git a/statsd/fnv1a.go b/statsd/fnv1a.go index 03dc8a07..717fd13b 100644 --- a/statsd/fnv1a.go +++ b/statsd/fnv1a.go @@ -37,3 +37,10 @@ func addString32(h uint32, s string) uint32 { return h } + +// appendString32 appends s to b while adding its bytes to the precomputed hash +// value h. It is useful for building a byte representation and its hash in one +// pass. +func appendString32(b []byte, h uint32, s string) ([]byte, uint32) { + return append(b, s...), addString32(h, s) +} diff --git a/statsd/metrics.go b/statsd/metrics.go index ea78730e..776393db 100644 --- a/statsd/metrics.go +++ b/statsd/metrics.go @@ -194,9 +194,6 @@ func (s *bufferedMetric) maybeKeepSample(v float64, rand *rand.Rand, randLock *s } } -func (s *bufferedMetric) skipSample() { - atomic.AddInt64(&s.totalSamples, 1) -} func (s *bufferedMetric) flushUnsafe() metric { totalSamples := atomic.LoadInt64(&s.totalSamples)