perf(precompute): prune snapshots by generation

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

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)
+				}
+			}
+		})
+	}
+}

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

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,