diff --git a/asap-precompute-go/precompute.go b/asap-precompute-go/precompute.go index 9a0758764..68a654d68 100644 --- a/asap-precompute-go/precompute.go +++ b/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) { diff --git a/asap-precompute-go/sketches/ddsketch.go b/asap-precompute-go/sketches/ddsketch.go index 83a08d72f..afcbfe4fb 100644 --- a/asap-precompute-go/sketches/ddsketch.go +++ b/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 diff --git a/opentelemetry-collector-contrib-patch/processor/ddsketchprocessor/config_translate.go b/opentelemetry-collector-contrib-patch/processor/ddsketchprocessor/config_translate.go index 4a661cea4..5c9aa6752 100644 --- a/opentelemetry-collector-contrib-patch/processor/ddsketchprocessor/config_translate.go +++ b/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 diff --git a/opentelemetry-collector-contrib-patch/processor/ddsketchprocessor/processor.go b/opentelemetry-collector-contrib-patch/processor/ddsketchprocessor/processor.go index d7de96a97..4db36bf7b 100644 --- a/opentelemetry-collector-contrib-patch/processor/ddsketchprocessor/processor.go +++ b/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 {