diff --git a/asap-precompute-go/precompute.go b/asap-precompute-go/precompute.go index 732c8424..ca864203 100644 --- a/asap-precompute-go/precompute.go +++ b/asap-precompute-go/precompute.go @@ -861,19 +861,13 @@ func (p *precompute) finishRotate(closed []*seriesEntry, rng [2]uint64, nowMs ui cfg := p.activeConfig() sink := p.sketchSink.Load() envelopes := make([]*SketchEnvelope, 0, len(closed)) - // closedKeys collects every series key in the just-closed window so the - // snapshot cache can prune entries for keys that did NOT reappear this - // window (P1-2: the outbound/inbound maps would otherwise grow forever, - // retaining a snapshot copy for every series key ever seen). Built only - // when the delta path is active (the only consumer of the cache) and a - // cache is present. - var closedKeys map[string]struct{} + var cacheGeneration uint64 if cfg != nil && cfg.DeltaTransmission && p.snapshotCache != nil { - closedKeys = make(map[string]struct{}, len(closed)) + cacheGeneration = p.snapshotCache.BeginGeneration() } for _, entry := range closed { - if closedKeys != nil && entry != nil { - closedKeys[cfg.SeriesKeyForEntry(entry.ResourceLabels, entry.Labels)] = struct{}{} + if cacheGeneration != 0 && entry != nil { + p.snapshotCache.Touch(entry.seriesKey, cacheGeneration) } env, err := p.serializeSeries(entry, cfg, rng) if err == nil && env != nil { @@ -897,8 +891,8 @@ func (p *precompute) finishRotate(closed []*seriesEntry, rng [2]uint64, nowMs ui // window. A series that vanished (never observed again this window) no // longer needs its cached outbound/inbound snapshot, and keeping it // would pin agent memory for the lifetime of the process (P1-2). - if closedKeys != nil { - p.snapshotCache.RetainKeys(closedKeys) + if cacheGeneration != 0 { + p.snapshotCache.EndGeneration(cacheGeneration) } p.stats.OutputEnvelopes.Add(uint64(len(envelopes))) // LastEmittedEnvelopes is a snapshot (not a running total) of the @@ -918,7 +912,10 @@ func (p *precompute) serializeSeries(entry *seriesEntry, cfg *PrecomputeConfig, // through cfg.SeriesKeyForEntry guarantees the snapshot-cache // lookup in the delta path agrees with the observe-time bucket // regardless of the OmitResourceAttrs / GlobalAggregation flags. - seriesKey := cfg.SeriesKeyForEntry(entry.ResourceLabels, entry.Labels) + seriesKey := entry.seriesKey + if seriesKey == "" { + seriesKey = cfg.SeriesKeyForEntry(entry.ResourceLabels, entry.Labels) + } var ( payload []byte isFull bool diff --git a/asap-precompute-go/rotation_bench_test.go b/asap-precompute-go/rotation_bench_test.go new file mode 100644 index 00000000..25fb5757 --- /dev/null +++ b/asap-precompute-go/rotation_bench_test.go @@ -0,0 +1,31 @@ +package precompute + +import ( + "strconv" + "testing" + "time" +) + +func BenchmarkDeltaRotationAtCardinality(b *testing.B) { + for _, cardinality := range []int{1_000, 10_000, 100_000} { + b.Run(strconv.Itoa(cardinality), func(b *testing.B) { + b.ReportAllocs() + for iteration := 0; iteration < b.N; iteration++ { + b.StopTimer() + cfg := &PrecomputeConfig{AggID: 1, SketchType: SketchTypeDDSketch, + Mode: Tumbling, Window: WindowSpec{Size: time.Second}, DeltaTransmission: true} + p := New(cfg, newFakeFactory(), &fakeObserver{}) + for i := 0; i < cardinality; i++ { + obs := &Observation{TimestampMs: 1, Labels: []KeyValue{{Key: "k", Value: strconv.Itoa(i)}}, Value: FloatValue(1)} + if err := p.Observe(obs); err != nil { + b.Fatal(err) + } + } + b.StartTimer() + if got := len(p.Tick(1_000)); got != cardinality { + b.Fatalf("envelopes=%d", got) + } + } + }) + } +} diff --git a/asap-precompute-go/snapshot_cache.go b/asap-precompute-go/snapshot_cache.go index 9e5f42eb..97470434 100644 --- a/asap-precompute-go/snapshot_cache.go +++ b/asap-precompute-go/snapshot_cache.go @@ -17,16 +17,20 @@ import ( // interchangeable (a remote sender's snapshot is not what we'd emit // locally). type SnapshotCache struct { - mu sync.RWMutex - outbound map[string][]byte - inbound map[string][]byte + mu sync.RWMutex + outbound map[string][]byte + inbound map[string][]byte + generation uint64 + seen map[string]uint64 } // NewSnapshotCache constructs an empty cache. func NewSnapshotCache() *SnapshotCache { return &SnapshotCache{ - outbound: make(map[string][]byte), - inbound: make(map[string][]byte), + outbound: make(map[string][]byte), + inbound: make(map[string][]byte), + generation: 1, + seen: make(map[string]uint64), } } @@ -43,6 +47,7 @@ func (c *SnapshotCache) CacheOutbound(seriesKey string, payload []byte) (firstTi cp := make([]byte, len(payload)) copy(cp, payload) c.outbound[seriesKey] = cp + c.seen[seriesKey] = c.generation return !existed } @@ -62,6 +67,49 @@ func (c *SnapshotCache) CacheInbound(seriesKey string, payload []byte) { cp := make([]byte, len(payload)) copy(cp, payload) c.inbound[seriesKey] = cp + c.seen[seriesKey] = c.generation +} + +// BeginGeneration advances the reusable liveness epoch used by rotation. +func (c *SnapshotCache) BeginGeneration() uint64 { + c.mu.Lock() + defer c.mu.Unlock() + c.generation++ + return c.generation +} + +// Touch marks a retained series without allocating a per-rotation keep set. +func (c *SnapshotCache) Touch(seriesKey string, generation uint64) { + c.mu.Lock() + defer c.mu.Unlock() + if generation == c.generation { + c.seen[seriesKey] = generation + } +} + +// EndGeneration removes snapshots not touched by the closed generation. Cache +// writes racing with serialization mark themselves in the current generation. +func (c *SnapshotCache) EndGeneration(generation uint64) { + c.mu.Lock() + defer c.mu.Unlock() + if generation != c.generation { + return + } + for key := range c.outbound { + if c.seen[key] != generation { + delete(c.outbound, key) + } + } + for key := range c.inbound { + if c.seen[key] != generation { + delete(c.inbound, key) + } + } + for key, seen := range c.seen { + if seen != generation { + delete(c.seen, key) + } + } } // GetInbound returns the cached upstream snapshot or nil. @@ -217,6 +265,7 @@ func (c *SnapshotCache) Delete(seriesKey string) { defer c.mu.Unlock() delete(c.outbound, seriesKey) delete(c.inbound, seriesKey) + delete(c.seen, seriesKey) } // Reset clears all cached state (used in tests and on shutdown). @@ -225,6 +274,8 @@ func (c *SnapshotCache) Reset() { defer c.mu.Unlock() c.outbound = make(map[string][]byte) c.inbound = make(map[string][]byte) + c.seen = make(map[string]uint64) + c.generation++ } // LenOutbound returns the number of cached outbound snapshots; for diff --git a/asap-precompute-go/window.go b/asap-precompute-go/window.go index 137fbbbe..f68c2384 100644 --- a/asap-precompute-go/window.go +++ b/asap-precompute-go/window.go @@ -809,6 +809,8 @@ func (w *windowState) rotateSlidingLocked(nowMs uint64, cfg *PrecomputeConfig) ( // labels. The merge below folds in this pane's state. dst = &seriesEntry{ Sketch: w.sketchFactory(), + seriesKey: key, + heapIndex: -1, ResourceLabels: src.ResourceLabels, Labels: src.Labels, LastSeenMs: src.LastSeenMs,