perf(ddsketch): recycle sketch state across windows via a pool

asap-precompute-go/precompute.go

@@ -104,8 +104,20 @@ type FrequencyEntry struct {
 // SketchFactory constructs an empty Sketch of the type owned by a
 // specific Precompute instance. Phase 2 keeps construction
 // per-Precompute rather than registry-based to avoid global state.
+//
+// When the host wires a sketch pool (see SketchSink), the factory is
+// the pool's Get side: it returns a recycled, already-Reset sketch
+// when one is available and a fresh one otherwise.
 type SketchFactory func() Sketch
 
+// SketchSink receives a series' sketch at flush time, after its
+// envelope has been serialized and the Precompute is done with it.
+// It is the symmetric Put side of a host-provided sketch pool: a
+// pooled adapter Resets the sketch and returns it for reuse next
+// window. When no sink is installed the sketch is simply dropped and
+// garbage-collected (the prior behavior).
+type SketchSink func(Sketch)
+
 // SketchObserver is implemented by the per-shim glue that knows
 // how to feed a raw observation (Float / Hash / Bytes) into the
 // concrete sketch. The runtime calls Observe on this interface
@@ -203,6 +215,11 @@ type Precompute interface {
 	// with Observe; the runtime stores the function pointer atomically.
 	// See LatencyObserver godoc for semantics.
 	SetLatencyObserver(fn LatencyObserver)
+	// SetSketchSink installs (or replaces) the flush-time sketch sink
+	// that recycles a series' sketch once its envelope is serialized.
+	// Pass nil to disable (sketches are dropped/GC'd). Safe to call
+	// concurrently; the runtime stores the function pointer atomically.
+	SetSketchSink(fn SketchSink)
 	// Shutdown flushes any in-progress state; intended for the
 	// shim's Shutdown path to run a final Tick before returning.
 	Shutdown(ctx context.Context) error
@@ -227,6 +244,7 @@ type precompute struct {
 	sketchType      SketchType
 	closed          atomic.Bool
 	latencyObserver atomic.Pointer[LatencyObserver]
+	sketchSink      atomic.Pointer[SketchSink]
 }
 
 // New constructs a Precompute given an initial config, a sketch
@@ -377,18 +395,23 @@ func (p *precompute) finishRotate(closed []*seriesEntry, rng [2]uint64, nowMs ui
 		return nil
 	}
 	cfg := p.activeConfig()
+	sink := p.sketchSink.Load()
 	envelopes := make([]*SketchEnvelope, 0, len(closed))
 	for _, entry := range closed {
 		env, err := p.serializeSeries(entry, cfg, rng)
-		if err != nil {
-			// Best-effort: skip this series and continue. Real
-			// shims log via their host's logger; the Layer-3
-			// runtime is host-neutral and has no logger.
-			continue
-		}
-		if env != nil {
+		if err == nil && env != nil {
 			envelopes = append(envelopes, env)
 		}
+		// On err we skip the envelope (best-effort: host-neutral
+		// runtime has no logger) but still recycle the sketch.
+		//
+		// The entry is detached from the live window (rotateLocked
+		// replaced the map), so once its envelope is serialized
+		// nothing else references the sketch — hand it to the pool.
+		if sink != nil && entry != nil && entry.Sketch != nil {
+			(*sink)(entry.Sketch)
+			entry.Sketch = nil
+		}
 	}
 	p.stats.OutputEnvelopes.Add(uint64(len(envelopes)))
 	p.stats.LastTickMs.Store(nowMs)
@@ -516,6 +539,15 @@ func (p *precompute) SetLatencyObserver(fn LatencyObserver) {
 	p.latencyObserver.Store(&fn)
 }
 
+// SetSketchSink implements Precompute.SetSketchSink.
+func (p *precompute) SetSketchSink(fn SketchSink) {
+	if fn == nil {
+		p.sketchSink.Store(nil)
+		return
+	}
+	p.sketchSink.Store(&fn)
+}
+
 // Shutdown implements Precompute.Shutdown.
 func (p *precompute) Shutdown(ctx context.Context) error {
 	if !p.closed.CompareAndSwap(false, true) {

asap-precompute-go/sketches/ddsketch.go

@@ -135,11 +135,16 @@ func (w *DDSketchWrapper) Merge(other precompute.Sketch) error {
 	return w.sk.Merge(o.sk)
 }
 
-// Reset zeros the sketch in place by replacing it with a fresh
-// DDSketch of the same alpha. Window rotation calls this when the
-// runtime decides to recycle entries.
+// Reset zeros the sketch IN PLACE, preserving the bucket store's
+// backing-array capacity so a wrapper recycled through a sketch pool
+// across windows does not re-allocate its store. Previously this
+// reallocated via NewDDSketch, which defeated pooling.
 func (w *DDSketchWrapper) Reset() {
-	w.sk = ddsketch.NewDDSketch(w.alpha)
+	if w.sk == nil {
+		w.sk = ddsketch.NewDDSketch(w.alpha)
+		return
+	}
+	w.sk.Clear()
 }
 
 // Quantile returns the q-th rank value as a float64; (0, false) from

opentelemetry-collector-contrib-patch/processor/ddsketchprocessor/config_translate.go

@@ -21,14 +21,30 @@ func getOrCreate(batch map[string]precompute.Precompute, metricName string, cfg
 		return pc
 	}
 	pc := precompute.New(toPrecomputeConfig(cfg, metricName), func() precompute.Sketch {
+		// Draw from the processor's cross-window sketch pool when
+		// available; the pool's New makes a fresh wrapper at the
+		// configured relative_accuracy. The throwaway-batch path
+		// (proc==nil, used by tests) always makes a fresh sketch.
+		if proc != nil {
+			return proc.sketchPool.Get().(precompute.Sketch)
+		}
 		return sketches.NewDDSketchWrapper(cfg.RelativeAccuracy)
 	}, sketches.DDSketchObserver{})
-	// Wire the per-Observe latency histogram if the shim has one.
-	// Closes Phase 2.11B gap #3 — every observation routed through
-	// this Precompute now contributes to asap_processor_observe_seconds
-	// on the deployed /metrics endpoint.
-	if proc != nil && proc.observeLatency != nil {
-		pc.SetLatencyObserver(proc.recordObserveLatency)
+	if proc != nil {
+		// Recycle each series' sketch back to the pool once its
+		// envelope is serialized at flush; Reset() does the in-place
+		// Clear that keeps the bucket-store capacity for reuse.
+		pc.SetSketchSink(func(s precompute.Sketch) {
+			s.Reset()
+			proc.sketchPool.Put(s)
+		})
+		// Wire the per-Observe latency histogram if the shim has one.
+		// Closes Phase 2.11B gap #3 — every observation routed through
+		// this Precompute now contributes to asap_processor_observe_seconds
+		// on the deployed /metrics endpoint.
+		if proc.observeLatency != nil {
+			pc.SetLatencyObserver(proc.recordObserveLatency)
+		}
 	}
 	batch[metricName] = pc
 	return pc

opentelemetry-collector-contrib-patch/processor/ddsketchprocessor/processor.go

@@ -24,6 +24,7 @@ import (
 	"go.uber.org/zap"
 
 	precompute "github.com/ProjectASAP/asap-precompute-go"
+	"github.com/ProjectASAP/asap-precompute-go/sketches"
 )
 
 // ddsketchProcessor is the Layer-4 OTel shim. State is keyed per
@@ -48,17 +49,27 @@ type ddsketchProcessor struct {
 	mu          sync.Mutex
 	precomputes map[string]precompute.Precompute
 
+	// sketchPool recycles DDSketch wrappers across windows. Every
+	// sketch in one processor shares the configured relative_accuracy,
+	// so a single type+shape pool is correctness-safe. The flush-time
+	// sink Reset()s a sketch (in-place Clear, keeping bucket-store
+	// capacity) before Put, so Get returns a zeroed-but-warm sketch
+	// that skips the per-window store re-allocation + regrow.
+	sketchPool sync.Pool
+
 	stopCh        chan struct{}
 	doneCh        chan struct{}
 	windowStarted atomic.Bool
 }
 
 func newProcessor(cfg *Config, logger *zap.Logger, next consumer.Metrics) *ddsketchProcessor {
-	return &ddsketchProcessor{
+	p := &ddsketchProcessor{
 		cfg: cfg, logger: logger, nextConsumer: next,
 		precomputes: make(map[string]precompute.Precompute),
 		stopCh:      make(chan struct{}), doneCh: make(chan struct{}),
 	}
+	p.sketchPool.New = func() any { return sketches.NewDDSketchWrapper(cfg.RelativeAccuracy) }
+	return p
 }
 
 func (p *ddsketchProcessor) Capabilities() consumer.Capabilities {