diff --git a/internal/tags/tags.go b/internal/tags/tags.go index c492d640..da5b5189 100644 --- a/internal/tags/tags.go +++ b/internal/tags/tags.go @@ -450,6 +450,54 @@ func SerializeTags(name string, tags map[string]string) string { } } +const ( + fnvOffset64a = 14695981039346656037 + fnvOffset64b = 0x27220a5774762123 // an arbitrary, independent offset basis + fnvPrime64 = 1099511628211 +) + +// hashString folds s into the running FNV-1a hash h. +func hashString(h uint64, s string) uint64 { + for i := 0; i < len(s); i++ { + h ^= uint64(s[i]) + h *= fnvPrime64 + } + return h +} + +// HashNameTags128 returns a 128-bit (as two independent uint64 halves), +// order-independent hash of name and tags, suitable as a memoization cache +// key: equal (name, tags) pairs hash to the same value regardless of the +// tags map's iteration order. Unlike Serialize, it never allocates and +// doesn't require the tags to be sorted. +// +// At 128 bits, two different (name, tags) pairs colliding is negligible for +// any realistic number of distinct metrics (the birthday bound is n²/2^129, +// which stays negligible even at cardinalities far beyond what a real +// service would produce) - so callers may treat a match on both halves as +// authoritative without also retaining a copy of the original name/tags to +// confirm it, which matters for a cache that (like scopeCache) never evicts. +func HashNameTags128(name string, tags map[string]string) (hi, lo uint64) { + hi = hashString(fnvOffset64a, name) + lo = hashString(fnvOffset64b, name) + var chi, clo uint64 + for k, v := range tags { + if k == "" || v == "" { + continue + } + ph := hashString(fnvOffset64a, k) + ph = hashString(ph, "=") + ph = hashString(ph, v) + chi ^= ph // order-independent: XOR doesn't care what order pairs arrive in + + pl := hashString(fnvOffset64b, k) + pl = hashString(pl, "=") + pl = hashString(pl, v) + clo ^= pl + } + return hi ^ chi, lo ^ clo +} + // ReplaceChars replaces any invalid chars ([.:|]) in value s with '_'. func ReplaceChars(s string) string { var buf []byte // lazily allocated diff --git a/internal/tags/tags_test.go b/internal/tags/tags_test.go index f923f475..5f8add14 100644 --- a/internal/tags/tags_test.go +++ b/internal/tags/tags_test.go @@ -1031,6 +1031,50 @@ func TestParseTags(t *testing.T) { } } +/////////////////////////////////////////////////////////////////// +// HashNameTags128 Tests + +func TestHashNameTags128OrderIndependent(t *testing.T) { + tags1 := map[string]string{"region_code": "us-east-1", "ride_type": "standard"} + tags2 := map[string]string{"ride_type": "standard", "region_code": "us-east-1"} // same content, different map + hi1, lo1 := HashNameTags128("name", tags1) + hi2, lo2 := HashNameTags128("name", tags2) + if hi1 != hi2 || lo1 != lo2 { + t.Error("hash should not depend on map iteration order") + } +} + +func TestHashNameTags128Differ(t *testing.T) { + baseHi, baseLo := HashNameTags128("name", map[string]string{"region_code": "us-east-1"}) + cases := map[string][2]uint64{} + set := func(label string, hi, lo uint64) { cases[label] = [2]uint64{hi, lo} } + + hi, lo := HashNameTags128("other", map[string]string{"region_code": "us-east-1"}) + set("different name", hi, lo) + hi, lo = HashNameTags128("name", map[string]string{"region_code": "us-west-2"}) + set("different value", hi, lo) + hi, lo = HashNameTags128("name", map[string]string{"other_key": "us-east-1"}) + set("different key", hi, lo) + hi, lo = HashNameTags128("name", map[string]string{"region_code": "us-east-1", "extra": "x"}) + set("extra tag", hi, lo) + hi, lo = HashNameTags128("name", nil) + set("no tags", hi, lo) + + for label, h := range cases { + if h[0] == baseHi && h[1] == baseLo { + t.Errorf("%s: expected a different hash, got the same value", label) + } + } +} + +func TestHashNameTags128IgnoresEmptyKeyValue(t *testing.T) { + hiA, loA := HashNameTags128("name", map[string]string{"k": "v"}) + hiB, loB := HashNameTags128("name", map[string]string{"k": "v", "": "invalid_key", "invalid_value": ""}) + if hiA != hiB || loA != loB { + t.Error("empty-key/empty-value tags should not affect the hash") + } +} + func TestParseTagSet(t *testing.T) { for _, x := range parseTagsTests { s, set := ParseTagSet(x.Stat) diff --git a/stats.go b/stats.go index 9f167bd8..178850f2 100644 --- a/stats.go +++ b/stats.go @@ -340,12 +340,60 @@ type statStore struct { gauges sync.Map timers sync.Map + cache scopeCache + mu sync.RWMutex statGenerators []StatGenerator sink Sink } +// scopeCache memoizes tag-based Scope/Counter/Gauge/Timer lookups so that +// repeated calls with identical name+tags skip re-deriving the +// joinScopes/MergeTags/Serialize (or, at the root, SerializeTags) path once a +// handle has already been created for that combination. +// +// Entries are never evicted: gostats already retains one Counter/Gauge/Timer +// per unique serialized name+tags forever (see statStore.counters/gauges/ +// timers), so this cache's growth is bounded by the same real metric +// cardinality a caller already commits to by using tags at all, not by +// anything new. Since entries are permanent, keys are the 128-bit +// tagspkg.HashNameTags128 hash rather than a 64-bit hash plus a retained +// copy of the original name/tags for collision verification: at 128 bits a +// collision is negligible for any realistic cardinality (see +// HashNameTags128), and skipping the verification copy avoids pinning a +// map[string]string (~200+ bytes, several times the size of the Counter/ +// Gauge/Timer it would be guarding) in memory forever for every distinct +// metric ever seen. +// +// The zero value is ready to use. +type scopeCache struct { + scopes sync.Map + counters sync.Map + gauges sync.Map + timers sync.Map + milliTimers sync.Map +} + +// hash128 is the two-uint64 cache key scopeCache uses (see its doc comment +// for why 128 bits is trusted without a verification copy). +type hash128 struct { + hi, lo uint64 +} + +// cachedOrCreate returns the memoized value for name+tags in m, or calls +// create and memoizes the result on a miss. +func cachedOrCreate[T any](m *sync.Map, name string, tags map[string]string, create func() T) T { + hi, lo := tagspkg.HashNameTags128(name, tags) + k := hash128{hi, lo} + if v, ok := m.Load(k); ok { + return v.(T) + } + val := create() + m.Store(k, val) + return val +} + var ReservedTagWords = map[string]bool{"asg": true, "az": true, "backend": true, "canary": true, "host": true, "period": true, "region": true, "shard": true, "window": true, "source": true, "project": true, "facet": true, "envoyservice": true} func (s *statStore) validateTags(tags map[string]string) { @@ -410,12 +458,14 @@ func (s *statStore) Store() Store { } func (s *statStore) Scope(name string) Scope { - return newSubScope(s, name, nil) + return s.ScopeWithTags(name, nil) } func (s *statStore) ScopeWithTags(name string, tags map[string]string) Scope { s.validateTags(tags) - return newSubScope(s, name, tags) + return cachedOrCreate(&s.cache.scopes, name, tags, func() Scope { + return newSubScope(s, name, tags) + }) } func (s *statStore) newCounter(serializedName string) *counter { @@ -435,7 +485,9 @@ func (s *statStore) NewCounter(name string) Counter { func (s *statStore) NewCounterWithTags(name string, tags map[string]string) Counter { s.validateTags(tags) - return s.newCounter(tagspkg.SerializeTags(name, tags)) + return cachedOrCreate(&s.cache.counters, name, tags, func() Counter { + return s.newCounter(tagspkg.SerializeTags(name, tags)) + }) } func (s *statStore) newCounterWithTagSet(name string, tags tagspkg.TagSet) Counter { @@ -472,7 +524,9 @@ func (s *statStore) NewGauge(name string) Gauge { func (s *statStore) NewGaugeWithTags(name string, tags map[string]string) Gauge { s.validateTags(tags) - return s.newGauge(tagspkg.SerializeTags(name, tags)) + return cachedOrCreate(&s.cache.gauges, name, tags, func() Gauge { + return s.newGauge(tagspkg.SerializeTags(name, tags)) + }) } func (s *statStore) newGaugeWithTagSet(name string, tags tagspkg.TagSet) Gauge { @@ -507,7 +561,9 @@ func (s *statStore) NewMilliTimer(name string) Timer { func (s *statStore) NewMilliTimerWithTags(name string, tags map[string]string) Timer { s.validateTags(tags) - return s.newTimer(tagspkg.SerializeTags(name, tags), time.Millisecond) + return cachedOrCreate(&s.cache.milliTimers, name, tags, func() Timer { + return s.newTimer(tagspkg.SerializeTags(name, tags), time.Millisecond) + }) } func (s *statStore) NewTimer(name string) Timer { @@ -516,7 +572,9 @@ func (s *statStore) NewTimer(name string) Timer { func (s *statStore) NewTimerWithTags(name string, tags map[string]string) Timer { s.validateTags(tags) - return s.newTimer(tagspkg.SerializeTags(name, tags), time.Microsecond) + return cachedOrCreate(&s.cache.timers, name, tags, func() Timer { + return s.newTimer(tagspkg.SerializeTags(name, tags), time.Microsecond) + }) } func (s *statStore) newTimerWithTagSet(name string, tags tagspkg.TagSet, base time.Duration) Timer { @@ -549,6 +607,7 @@ type subScope struct { registry *statStore name string tags tagspkg.TagSet // read-only and may be shared by multiple subScopes + cache scopeCache } func newSubScope(registry *statStore, name string, tags map[string]string) *subScope { @@ -561,11 +620,13 @@ func (s *subScope) Scope(name string) Scope { func (s *subScope) ScopeWithTags(name string, tags map[string]string) Scope { s.registry.validateTags(tags) - return &subScope{ - registry: s.registry, - name: joinScopes(s.name, name), - tags: s.tags.MergeTags(tags), - } + return cachedOrCreate(&s.cache.scopes, name, tags, func() Scope { + return &subScope{ + registry: s.registry, + name: joinScopes(s.name, name), + tags: s.tags.MergeTags(tags), + } + }) } func (s *subScope) Store() Store { @@ -577,7 +638,9 @@ func (s *subScope) NewCounter(name string) Counter { } func (s *subScope) NewCounterWithTags(name string, tags map[string]string) Counter { - return s.registry.newCounterWithTagSet(joinScopes(s.name, name), s.tags.MergeTags(tags)) + return cachedOrCreate(&s.cache.counters, name, tags, func() Counter { + return s.registry.newCounterWithTagSet(joinScopes(s.name, name), s.tags.MergeTags(tags)) + }) } func (s *subScope) NewPerInstanceCounter(name string, tags map[string]string) Counter { @@ -590,7 +653,9 @@ func (s *subScope) NewGauge(name string) Gauge { } func (s *subScope) NewGaugeWithTags(name string, tags map[string]string) Gauge { - return s.registry.newGaugeWithTagSet(joinScopes(s.name, name), s.tags.MergeTags(tags)) + return cachedOrCreate(&s.cache.gauges, name, tags, func() Gauge { + return s.registry.newGaugeWithTagSet(joinScopes(s.name, name), s.tags.MergeTags(tags)) + }) } func (s *subScope) NewPerInstanceGauge(name string, tags map[string]string) Gauge { @@ -603,7 +668,9 @@ func (s *subScope) NewTimer(name string) Timer { } func (s *subScope) NewTimerWithTags(name string, tags map[string]string) Timer { - return s.registry.newTimerWithTagSet(joinScopes(s.name, name), s.tags.MergeTags(tags), time.Microsecond) + return cachedOrCreate(&s.cache.timers, name, tags, func() Timer { + return s.registry.newTimerWithTagSet(joinScopes(s.name, name), s.tags.MergeTags(tags), time.Microsecond) + }) } func (s *subScope) NewPerInstanceTimer(name string, tags map[string]string) Timer { @@ -616,7 +683,9 @@ func (s *subScope) NewMilliTimer(name string) Timer { } func (s *subScope) NewMilliTimerWithTags(name string, tags map[string]string) Timer { - return s.registry.newTimerWithTagSet(joinScopes(s.name, name), s.tags.MergeTags(tags), time.Millisecond) + return cachedOrCreate(&s.cache.milliTimers, name, tags, func() Timer { + return s.registry.newTimerWithTagSet(joinScopes(s.name, name), s.tags.MergeTags(tags), time.Millisecond) + }) } func (s *subScope) NewPerInstanceMilliTimer(name string, tags map[string]string) Timer { diff --git a/stats_cache_test.go b/stats_cache_test.go new file mode 100644 index 00000000..f44c9ce6 --- /dev/null +++ b/stats_cache_test.go @@ -0,0 +1,154 @@ +package stats + +import ( + "sync" + "testing" +) + +// Ensure repeated calls with equal, but distinct, tags maps return the exact +// same Counter/Gauge/Timer/Scope, and that the value recorded through a +// cache hit is identical to going through the slow path directly. + +func TestSubScopeCounterCacheHit(t *testing.T) { + store := NewStore(&testStatSink{}, false) + scope := store.Scope("app") + + tags1 := map[string]string{"region_code": "us-east-1", "ride_type": "standard"} + tags2 := map[string]string{"ride_type": "standard", "region_code": "us-east-1"} // different map, same content + + c1 := scope.NewCounterWithTags("attempt", tags1) + c2 := scope.NewCounterWithTags("attempt", tags2) + if c1 != c2 { + t.Fatal("expected the same Counter for equal tags regardless of map identity/order") + } + + c1.Add(2) + c2.Add(3) // should land on the same counter + if v := c1.(*counter).String(); v != "5" { + t.Fatalf("got: %s want: 5", v) + } +} + +func TestSubScopeCounterCacheDifferentTags(t *testing.T) { + store := NewStore(&testStatSink{}, false) + scope := store.Scope("app") + + c1 := scope.NewCounterWithTags("attempt", map[string]string{"region_code": "us-east-1"}) + c2 := scope.NewCounterWithTags("attempt", map[string]string{"region_code": "us-west-2"}) + if c1 == c2 { + t.Fatal("different tag values must not collide in the cache") + } +} + +func TestSubScopeGaugeAndTimerCacheHit(t *testing.T) { + store := NewStore(&testStatSink{}, false) + scope := store.Scope("app") + + tags := map[string]string{"graph_name": "supply_estimate"} + g1 := scope.NewGaugeWithTags("queue_depth", tags) + g2 := scope.NewGaugeWithTags("queue_depth", map[string]string{"graph_name": "supply_estimate"}) + if g1 != g2 { + t.Fatal("expected the same Gauge for equal tags") + } + + tm1 := scope.NewTimerWithTags("latency", tags) + tm2 := scope.NewTimerWithTags("latency", map[string]string{"graph_name": "supply_estimate"}) + if tm1 != tm2 { + t.Fatal("expected the same Timer for equal tags") + } + + mt1 := scope.NewMilliTimerWithTags("latency_ms", tags) + mt2 := scope.NewMilliTimerWithTags("latency_ms", map[string]string{"graph_name": "supply_estimate"}) + if mt1 != mt2 { + t.Fatal("expected the same milli Timer for equal tags") + } +} + +func TestSubScopeScopeCacheHit(t *testing.T) { + store := NewStore(&testStatSink{}, false) + scope := store.Scope("app") + + child1 := scope.Scope("nested") + child2 := scope.Scope("nested") + if child1 != child2 { + t.Fatal("expected the same child Scope for repeated Scope() calls with the same name") + } + + tagged1 := scope.ScopeWithTags("waypoint_bonus", map[string]string{"graph_name": "supply_estimate"}) + tagged2 := scope.ScopeWithTags("waypoint_bonus", map[string]string{"graph_name": "supply_estimate"}) + if tagged1 != tagged2 { + t.Fatal("expected the same child Scope for repeated ScopeWithTags() calls with equal tags") + } + + // A cached child scope must itself still be a working, cached Scope. + c1 := child1.NewCounter("hits") + c2 := child2.NewCounter("hits") + if c1 != c2 { + t.Fatal("counters created via a cached child scope should also be cached/identical") + } +} + +func TestStoreRootCounterCacheHit(t *testing.T) { + store := NewStore(&testStatSink{}, false) + + tags := map[string]string{"region_code": "us-east-1"} + c1 := store.NewCounterWithTags("attempt", tags) + c2 := store.NewCounterWithTags("attempt", map[string]string{"region_code": "us-east-1"}) + if c1 != c2 { + t.Fatal("expected the same Counter for equal tags at the store root") + } +} + +func TestStoreRootScopeCacheHit(t *testing.T) { + store := NewStore(&testStatSink{}, false) + + s1 := store.Scope("app") + s2 := store.Scope("app") + if s1 != s2 { + t.Fatal("expected the same root child Scope for repeated Scope() calls with the same name") + } +} + +// Regression check: caching must not change what actually gets flushed. +func TestScopeCacheFlushOutputUnchanged(t *testing.T) { + sink := &testStatSink{} + store := NewStore(sink, false) + scope := store.Scope("app").ScopeWithTags("child", map[string]string{"k": "v"}) + + scope.NewCounterWithTags("attempt", map[string]string{"region_code": "us-east-1"}).Add(1) + scope.NewCounterWithTags("attempt", map[string]string{"region_code": "us-east-1"}).Add(1) // cache hit + store.Flush() + + const want = "app.child.attempt.__k=v.__region_code=us-east-1:2|c\n" + if sink.record != want { + t.Fatalf("got: %q want: %q", sink.record, want) + } +} + +// Concurrent callers hitting the same cache entries must not race and must +// converge on a single created instance (run with -race). +func TestScopeCacheConcurrentSafety(t *testing.T) { + store := NewStore(&testStatSink{}, false) + scope := store.Scope("app") + + const n = 64 + results := make([]Counter, n) + var wg sync.WaitGroup + start := make(chan struct{}) + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + <-start + results[i] = scope.NewCounterWithTags("attempt", map[string]string{"region_code": "us-east-1"}) + }(i) + } + close(start) + wg.Wait() + + for i := 1; i < n; i++ { + if results[i] != results[0] { + t.Fatalf("expected all concurrent callers to converge on the same Counter, index %d differed", i) + } + } +}