From dce0db60192399aaf14b2c96d8feb44c170ab955 Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Sat, 2 May 2026 20:46:01 -0400 Subject: [PATCH] refactor(countsketchprocessor): thin shim delegating to asap-precompute-go CountSketch processor reduces from ~663 LoC to a thin shim. State machine moves to asap-precompute-go. sketch_wrapper.go implements FrequencySketch over sketchlib-go CountSketch. GlobalAggregation + EmitWindowStats config flags preserve legacy single-partition emit shape with sample_count / window_duration_seconds attrs. Public test API: Shim.ProcessBatch/ProcessMetrics/FlushWindow. Parity harness: TestParity_CountSketch byte-identical (1 envelope). Phase 2 step 2.8. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../countsketchprocessor/config_translate.go | 108 +++ .../delta_transmission_test.go | 46 +- .../processor/countsketchprocessor/factory.go | 2 +- .../processor/countsketchprocessor/go.mod | 5 +- .../processor/countsketchprocessor/monitor.go | 57 ++ .../countsketchprocessor/processor.go | 725 ++++-------------- .../countsketchprocessor/processor_test.go | 24 +- .../countsketchprocessor/shim_helpers.go | 367 +++++++++ .../countsketchprocessor/sketch_wrapper.go | 226 ++++++ 9 files changed, 948 insertions(+), 612 deletions(-) create mode 100644 opentelemetry-collector-contrib-patch/processor/countsketchprocessor/config_translate.go create mode 100644 opentelemetry-collector-contrib-patch/processor/countsketchprocessor/monitor.go create mode 100644 opentelemetry-collector-contrib-patch/processor/countsketchprocessor/shim_helpers.go create mode 100644 opentelemetry-collector-contrib-patch/processor/countsketchprocessor/sketch_wrapper.go diff --git a/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/config_translate.go b/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/config_translate.go new file mode 100644 index 00000000..496f6b7b --- /dev/null +++ b/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/config_translate.go @@ -0,0 +1,108 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +package countsketchprocessor + +import ( + "math" + + "go.opentelemetry.io/collector/pdata/pmetric" + + precompute "github.com/ProjectASAP/asap-precompute-go" +) + +// outputMetricName is the fixed metric name the legacy processor +// stamped on every flushed CountSketch envelope. Hard-coded here (not +// derived per-input-metric) to preserve byte-parity with the legacy +// emit path — see ADR-0002 §"Behavior preservation" and the parity +// harness's CountSketch sketchDescriptor in +// integration/parity/harness/runtime.go. +const outputMetricName = "countsketch_partition" + +// toPrecomputeConfig translates the legacy Config into the runtime's +// PrecomputeConfig. The runtime always emits envelopes +// (TransmitSketch=true on the runtime side); the shim's downstream +// emit path inspects the original Config.TransmitSketch to decide +// whether to write typed CountSketch DPs or fall back to legacy +// Gauge summaries — see flushToMetrics in shim_helpers.go. +// +// Three non-default flags are flipped specifically for CountSketch +// byte-parity (verified by integration/parity/harness/runtime.go): +// +// - GlobalAggregation=true when AggregateBy is empty: legacy +// buildPartitionKey returns the literal "global" for empty +// AggregateBy, collapsing every observation into one shared +// sketch. The runtime models that exactly. +// - OmitResourceAttrs=true: legacy series-key construction never +// puts resource attrs in the key; resource attrs only appear as +// a fallback when buildPartitionKey looks up an AggregateBy key +// missing from dp-attrs. The shim's observe path merges +// resource attrs into dp-labels before Observe so AggregateBy +// lookups still find them, and the runtime then strips resource +// attrs from the emitted envelope per OmitResourceAttrs=true. +// - EmitWindowStats=true: legacy stamped sample_count and +// window_duration_seconds onto every emitted DP. Routing them +// through the envelope's Labels at flush time lets the OTel +// adapter reproduce them via KeyValuesToAttributes naturally. +func toPrecomputeConfig(cfg *Config) *precompute.PrecomputeConfig { + pcfg := &precompute.PrecomputeConfig{ + AggID: precompute.AggId(uint64(precompute.SketchTypeCountSketch)), + SketchType: precompute.SketchTypeCountSketch, + Mode: precompute.Tumbling, + Window: precompute.WindowSpec{Size: cfg.WindowDuration}, + Matchers: toRuntimeMatchers(cfg.LabelMatchers), + AggregateBy: append([]string(nil), cfg.AggregateBy...), + MetricName: outputMetricName, + TransmitSketch: true, + DeltaTransmission: cfg.DeltaTransmission, + DeltaThreshold: uint64(math.Ceil(cfg.DeltaThreshold)), + Encoding: mapEncoding(cfg.Encoding), + Temporality: int32(pmetric.AggregationTemporalityDelta), + GlobalAggregation: len(cfg.AggregateBy) == 0, + OmitResourceAttrs: true, + EmitWindowStats: true, + } + return pcfg +} + +// configDimensions mirrors the legacy newConfiguredCountSketch sizing. +// Exposed at file scope so the shim and tests build sketches with the +// exact dimensions the legacy processor used. +func configDimensions(cfg *Config) (rows, cols int) { + rows = int(math.Ceil(math.Log(1 / cfg.Delta))) + if rows < 1 { + rows = 1 + } + cols = int(math.Ceil(1 / (cfg.Epsilon * cfg.Epsilon))) + if cols < 2 { + cols = 2 + } + cols = nextPowerOfTwo(cols) + return rows, cols +} + +func nextPowerOfTwo(n int) int { + p := 1 + for p < n { + p <<= 1 + } + return p +} + +func toRuntimeMatchers(in []LabelMatcher) []precompute.LabelMatcher { + if len(in) == 0 { + return nil + } + out := make([]precompute.LabelMatcher, 0, len(in)) + for _, m := range in { + out = append(out, precompute.LabelMatcher{Name: m.Key, Value: m.Value}) + } + return out +} + +func mapEncoding(e SketchEncoding) precompute.Encoding { + if e == EncodingMsgpack { + return precompute.EncodingMsgpack + } + return precompute.EncodingProtoFull +} diff --git a/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/delta_transmission_test.go b/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/delta_transmission_test.go index 66e6e3c1..5bf26074 100644 --- a/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/delta_transmission_test.go +++ b/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/delta_transmission_test.go @@ -177,7 +177,7 @@ func TestCSDelta_FirstWindowSendsFullSketch(t *testing.T) { proc := newProcessor(zap.NewNop(), cfg, new(consumertest.MetricsSink)) - out, err := proc.processMetrics(context.Background(), makeCSGaugeMetrics("svc", 10)) + out, err := proc.ProcessMetrics(context.Background(), makeCSGaugeMetrics("svc", 10)) require.NoError(t, err) dps := getCSOutputDPs(out) @@ -207,11 +207,11 @@ func TestCSDelta_SubsequentWindowsSendDelta(t *testing.T) { proc := newProcessor(zap.NewNop(), cfg, new(consumertest.MetricsSink)) // Window 1 — establishes snapshot. - _, err := proc.processMetrics(context.Background(), makeCSGaugeMetrics("svc", 10)) + _, err := proc.ProcessMetrics(context.Background(), makeCSGaugeMetrics("svc", 10)) require.NoError(t, err) // Window 2 — should send delta. - out2, err := proc.processMetrics(context.Background(), makeCSGaugeMetrics("svc", 5)) + out2, err := proc.ProcessMetrics(context.Background(), makeCSGaugeMetrics("svc", 5)) require.NoError(t, err) dps := getCSOutputDPs(out2) @@ -240,7 +240,7 @@ func TestCSDelta_RoundTrip(t *testing.T) { proc := newProcessor(zap.NewNop(), cfg, new(consumertest.MetricsSink)) // Window 1: 50 insertions → snapshot created. - out1, err := proc.processMetrics(context.Background(), makeCSGaugeMetrics("svc", 50)) + out1, err := proc.ProcessMetrics(context.Background(), makeCSGaugeMetrics("svc", 50)) require.NoError(t, err) dps1 := getCSOutputDPs(out1) require.Len(t, dps1, 1) @@ -250,7 +250,7 @@ func TestCSDelta_RoundTrip(t *testing.T) { // Window 2: 30 insertions of the same key → delta against window-1 snapshot. md2 := makeCSGaugeMetrics("svc", 30) - out2, err := proc.processMetrics(context.Background(), md2) + out2, err := proc.ProcessMetrics(context.Background(), md2) require.NoError(t, err) dps2 := getCSOutputDPs(out2) require.Len(t, dps2, 1) @@ -271,7 +271,7 @@ func TestCSDelta_RoundTrip(t *testing.T) { // Reference: fresh no-delta processor with exactly window-2 data. refProc := newProcessor(zap.NewNop(), refCSConfig(), new(consumertest.MetricsSink)) require.NoError(t, refCSConfig().Validate()) - refOut, err := refProc.processMetrics(context.Background(), md2) + refOut, err := refProc.ProcessMetrics(context.Background(), md2) require.NoError(t, err) refDps := getCSOutputDPs(refOut) require.Len(t, refDps, 1) @@ -285,19 +285,31 @@ func TestCSDelta_RoundTrip(t *testing.T) { // TestCSDelta_MultipleWindowsConvergence simulates 5 consecutive delta windows // and verifies that the receiver's reconstructed sketch matches an independent // reference processor for each window. +// +// Receiver protocol after Phase-2 step 2.8: the sender's snapshot +// cache holds the FIRST PROTO_FULL frame as a baseline and emits +// each subsequent delta as `current_window_state - baseline`. The +// receiver reconstructs the current window by applying each delta to +// a clone of the baseline (NOT to the previously reconstructed +// sketch, as the legacy processor did when it refreshed its +// snapshot every window). The asap-precompute-go runtime's +// SnapshotCache.ComputeDelta keeps the cached outbound at the prior +// baseline whenever the delta stays under threshold, so all +// downstream receivers can apply against the same fixed baseline — +// see asap-precompute-go/snapshot_cache.go. func TestCSDelta_MultipleWindowsConvergence(t *testing.T) { cfg := deltaCSConfig() require.NoError(t, cfg.Validate()) proc := newProcessor(zap.NewNop(), cfg, new(consumertest.MetricsSink)) - var prevSnap *countsketch.CountSketch + var baseline *countsketch.CountSketch for w := 0; w < 5; w++ { insertCount := 20 * (w + 1) md := makeCSGaugeMetrics("svc", insertCount) - out, err := proc.processMetrics(context.Background(), md) + out, err := proc.ProcessMetrics(context.Background(), md) require.NoError(t, err, "window %d", w) dps := getCSOutputDPs(out) @@ -311,20 +323,24 @@ func TestCSDelta_MultipleWindowsConvergence(t *testing.T) { if enc == "proto_full" { currentCS, err = countsketch.DeserializeCountSketchFromProtoBytes(rawPayload) require.NoError(t, err, "window %d: full deserialize", w) + baseline = cloneCSTest(currentCS) + require.NotNil(t, baseline) } else { require.Equal(t, "proto_delta", enc, "window %d: unexpected encoding", w) - require.NotNil(t, prevSnap, "window %d: delta before full snapshot", w) + require.NotNil(t, baseline, "window %d: delta before full snapshot", w) deltaMsg, derr := countsketch.DeserializeDelta(rawPayload) require.NoError(t, derr, "window %d: delta deserialize", w) - currentCS = cloneCSTest(prevSnap) + // Apply delta to the cached baseline — not the previous + // reconstruction — to match the runtime's + // cumulative-against-baseline emit shape. + currentCS = cloneCSTest(baseline) require.NotNil(t, currentCS) countsketch.ApplyDelta(currentCS, deltaMsg) } - prevSnap = currentCS // Reference: fresh no-delta processor with only this window's data. refProc := newProcessor(zap.NewNop(), refCSConfig(), new(consumertest.MetricsSink)) - refOut, err := refProc.processMetrics(context.Background(), md) + refOut, err := refProc.ProcessMetrics(context.Background(), md) require.NoError(t, err, "window %d: reference proc", w) refDps := getCSOutputDPs(refOut) require.Len(t, refDps, 1, "window %d: reference output", w) @@ -345,7 +361,7 @@ func TestCSDelta_DisabledAlwaysSendsFullSketch(t *testing.T) { proc := newProcessor(zap.NewNop(), cfg, new(consumertest.MetricsSink)) for i := 0; i < 3; i++ { - out, err := proc.processMetrics(context.Background(), makeCSGaugeMetrics("svc", 5)) + out, err := proc.ProcessMetrics(context.Background(), makeCSGaugeMetrics("svc", 5)) require.NoError(t, err) dps := getCSOutputDPs(out) require.Len(t, dps, 1, "window %d", i) @@ -386,7 +402,7 @@ func TestCSDelta_PartitionKeyIsolation(t *testing.T) { } // Window 1: both services → both send proto_full (no prior snapshot). - out1, err := proc.processMetrics(context.Background(), buildTwoService(30, 20)) + out1, err := proc.ProcessMetrics(context.Background(), buildTwoService(30, 20)) require.NoError(t, err) dps1 := getCSOutputDPs(out1) require.Len(t, dps1, 2, "window 1: expected 2 partition data points") @@ -396,7 +412,7 @@ func TestCSDelta_PartitionKeyIsolation(t *testing.T) { } // Window 2: both services → both send proto_delta. - out2, err := proc.processMetrics(context.Background(), buildTwoService(15, 10)) + out2, err := proc.ProcessMetrics(context.Background(), buildTwoService(15, 10)) require.NoError(t, err) dps2 := getCSOutputDPs(out2) require.Len(t, dps2, 2, "window 2: expected 2 partition data points") diff --git a/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/factory.go b/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/factory.go index 0b315417..3fe61f19 100644 --- a/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/factory.go +++ b/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/factory.go @@ -51,6 +51,6 @@ func createMetricsProcessor( proc.processMetrics, processorhelper.WithStart(proc.Start), processorhelper.WithShutdown(proc.Shutdown), - processorhelper.WithCapabilities(consumer.Capabilities{MutatesData: true}), + processorhelper.WithCapabilities(consumer.Capabilities{MutatesData: false}), ) } diff --git a/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/go.mod b/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/go.mod index ccc83a5d..f70ae938 100644 --- a/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/go.mod +++ b/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/go.mod @@ -3,6 +3,7 @@ module github.com/open-telemetry/opentelemetry-collector-contrib/processor/count go 1.25.5 require ( + github.com/ProjectASAP/asap-precompute-go v0.0.0-00010101000000-000000000000 github.com/ProjectASAP/sketchlib-go v0.0.0-20260328221809-b24e56e64e94 github.com/stretchr/testify v1.11.1 go.opentelemetry.io/collector/component v1.47.0 @@ -13,7 +14,6 @@ require ( go.opentelemetry.io/collector/processor v1.47.0 go.opentelemetry.io/collector/processor/processorhelper v0.141.0 go.uber.org/zap v1.27.1 - google.golang.org/protobuf v1.36.11 ) require ( @@ -47,6 +47,7 @@ require ( go.yaml.in/yaml/v2 v2.4.3 // indirect golang.org/x/sys v0.37.0 // indirect golang.org/x/text v0.30.0 // indirect + google.golang.org/protobuf v1.36.11 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) @@ -55,3 +56,5 @@ replace go.opentelemetry.io/collector/pdata => ../../../opentelemetry-collector/ replace go.opentelemetry.io/collector/processor => ../../../opentelemetry-collector/processor replace github.com/ProjectASAP/sketchlib-go => ../../../../sketchlib-go + +replace github.com/ProjectASAP/asap-precompute-go => ../../../asap-precompute-go diff --git a/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/monitor.go b/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/monitor.go new file mode 100644 index 00000000..b01d6cbb --- /dev/null +++ b/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/monitor.go @@ -0,0 +1,57 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +package countsketchprocessor + +import ( + "context" + + "go.opentelemetry.io/collector/component" + "go.opentelemetry.io/collector/pdata/pmetric" + "go.opentelemetry.io/collector/processor/selfmonitor" + "go.uber.org/zap" +) + +// enableSelfMonitoring wires the OTel-side runtime self-monitor. +// activeSeriesCount reads the live counter the runtime exposes via +// PrecomputeStats — single-Precompute model means we don't sum +// across a per-metric map (unlike the DDSketch / KLL / HLL / CMS +// shims). +func (p *countSketchProcessor) enableSelfMonitoring(settings component.TelemetrySettings, processorID string) { + monitor, err := selfmonitor.New(settings, processorID, Type.String(), p.activeSeriesCount) + if err != nil { + if p.logger != nil { + p.logger.Warn("countsketchprocessor: failed to initialize self-monitoring", zap.Error(err)) + } + return + } + p.monitor = monitor +} + +func (p *countSketchProcessor) shutdownMonitor() { + if p.monitor != nil { + p.monitor.Shutdown() + } +} + +func (p *countSketchProcessor) recordInput(ctx context.Context, md pmetric.Metrics) { + if p.monitor != nil { + p.monitor.RecordInput(ctx, md) + } +} + +func (p *countSketchProcessor) recordOutput(ctx context.Context, md pmetric.Metrics) { + if p.monitor != nil { + p.monitor.RecordOutput(ctx, md) + } +} + +func (p *countSketchProcessor) activeSeriesCount() int64 { + if p.pc == nil { + return 0 + } + if s := p.pc.Stats(); s != nil { + return s.ActiveSeries.Load() + } + return 0 +} diff --git a/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/processor.go b/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/processor.go index 5d698cdf..d148203b 100644 --- a/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/processor.go +++ b/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/processor.go @@ -1,139 +1,120 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +// Package countsketchprocessor is a thin OTel-host shim around the +// asap-precompute-go runtime. The state machine (windowing, snapshot +// caching, delta encoding) lives in asap-precompute-go; this file owns +// only the OTel binding (decode pmetric → host-neutral Observation, +// encode SketchEnvelope → pmetric, ConsumeMetrics / Start / Shutdown). +// +// The shim is the Layer-4 OTel-side state holder. CountSketch's +// GlobalAggregation=true config means all observations collapse into +// a single shared series — so a single Precompute instance suffices +// (unlike DDSketch / KLL / HLL / CMS which key by metric-name). +// +// Phase 2 step 2.8 — see ADR-0002 and docs/phase-2-execution-plan.md. package countsketchprocessor import ( "context" - "math" - "sort" - "strings" "sync" "sync/atomic" "time" - countsketch "github.com/ProjectASAP/sketchlib-go/sketches/CountSketch" - "go.opentelemetry.io/collector/component" "go.opentelemetry.io/collector/consumer" - "go.opentelemetry.io/collector/pdata/pcommon" "go.opentelemetry.io/collector/pdata/pmetric" "go.opentelemetry.io/collector/processor/selfmonitor" "go.uber.org/zap" - "google.golang.org/protobuf/proto" -) - -// builderPool recycles strings.Builder instances used in buildPartitionKey. -var builderPool = sync.Pool{New: func() any { return new(strings.Builder) }} -type windowSketch struct { - cs *countsketch.CountSketch - mu sync.Mutex - sampleCount uint64 -} + precompute "github.com/ProjectASAP/asap-precompute-go" + otelpre "github.com/ProjectASAP/asap-precompute-go/otel" +) +// countSketchProcessor is the Layer-4 OTel shim over precompute.Precompute. +// +// Lifecycle: +// - Start spawns a ticker goroutine that drives FlushWindow every +// WindowDuration in window mode; in batch mode the goroutine is +// not started. +// - ConsumeMetrics decodes input → Precompute.Observe each → +// forwards the input unchanged to next (window mode) or returns +// the synthesized output via processMetrics (batch mode). +// - Shutdown cancels the ticker and runs one final flush so +// in-flight window state is not lost. type countSketchProcessor struct { logger *zap.Logger next consumer.Metrics config *Config - mode InputMode monitor *selfmonitor.Monitor - mu sync.RWMutex - activeWindowSketches map[string]*windowSketch - - windowSketchPool sync.Pool - - // snapshots holds one CS clone per partition key, updated every window - // flush. Used to compute sparse delta payloads when DeltaTransmission=true. - snapshots map[string]*countsketch.CountSketch - snapshotsMu sync.Mutex - - // inboundSnapshots tracks the last reconstructed full CS per series key - // received from upstream. Used to apply sparse deltas when upstream sends - // CountSketchEncodingDelta payloads. - inboundMu sync.Mutex - inboundSnapshots map[string]*countsketch.CountSketch + mu sync.Mutex + pc precompute.Precompute + adapter *otelpre.Adapter stopCh chan struct{} doneCh chan struct{} windowStarted atomic.Bool } -func newConfiguredCountSketch(cfg *Config) (*countsketch.CountSketch, error) { - rows := int(math.Ceil(math.Log(1 / cfg.Delta))) - if rows < 1 { - rows = 1 - } - - cols := int(math.Ceil(1 / (cfg.Epsilon * cfg.Epsilon))) - if cols < 2 { - cols = 2 - } - cols = nextPowerOfTwo(cols) - - return countsketch.NewCountSketch(rows, cols) -} - -func nextPowerOfTwo(n int) int { - p := 1 - for p < n { - p <<= 1 - } - return p -} - +// newProcessor wires up the runtime + OTel adapter for one Config. +// Does NOT start any goroutine — that happens in Start. Tests that +// instantiate Config literals without calling Validate may leave +// Mode empty; default to Batch here to match the legacy newProcessor. func newProcessor(logger *zap.Logger, cfg *Config, next consumer.Metrics) *countSketchProcessor { - mode := cfg.Mode - if mode == "" { - mode = ModeBatch - } - - p := &countSketchProcessor{ - logger: logger, - next: next, - config: cfg, - mode: mode, - activeWindowSketches: make(map[string]*windowSketch), - snapshots: make(map[string]*countsketch.CountSketch), - inboundSnapshots: make(map[string]*countsketch.CountSketch), - stopCh: make(chan struct{}), - doneCh: make(chan struct{}), - } - p.windowSketchPool.New = func() any { return new(windowSketch) } - return p -} - -func (p *countSketchProcessor) Start(ctx context.Context, host component.Host) error { - p.logger.Info("Starting Count Sketch Processor", - zap.Float64("epsilon", p.config.Epsilon), - zap.Float64("delta", p.config.Delta), - zap.Duration("window_duration", p.config.WindowDuration), - zap.String("mode", string(p.mode)), - zap.Strings("aggregate_by", p.config.AggregateBy), - ) - - if p.mode == ModeBatch { + if cfg.Mode == "" { + cfg.Mode = ModeBatch + } + rows, cols := configDimensions(cfg) + pcfg := toPrecomputeConfig(cfg) + factory := precompute.SketchFactory(func() precompute.Sketch { + w, _ := newCountSketchWrapper(rows, cols) + return w + }) + pp := precompute.New(pcfg, factory, countSketchObserver{defaultKey: outputMetricName}) + adapter := otelpre.New(&otelpre.AdapterConfig{ + ScopeName: "otelcol/countsketch", + }, nil) + return &countSketchProcessor{ + logger: logger, + next: next, + config: cfg, + pc: pp, + adapter: adapter, + stopCh: make(chan struct{}), + doneCh: make(chan struct{}), + } +} + +// Capabilities implements processor.Metrics. The shim does not mutate +// input md in place: ProcessBatch returns a fresh pmetric.Metrics +// (ADR-0002 §"Test API contract"). +func (p *countSketchProcessor) Capabilities() consumer.Capabilities { + return consumer.Capabilities{MutatesData: false} +} + +// Start launches the window-mode ticker (no-op in batch mode). +func (p *countSketchProcessor) Start(ctx context.Context, _ component.Host) error { + if p.config.Mode == ModeBatch { return nil } - if p.config.WindowDuration <= 0 { return nil } - - ticker := time.NewTicker(p.config.WindowDuration) p.windowStarted.Store(true) - go p.startWindowLoop(ctx, ticker) - + go p.runWindowLoop(ctx) return nil } +// Shutdown stops the ticker, drains one final window, and tears down +// the self-monitor. func (p *countSketchProcessor) Shutdown(ctx context.Context) error { defer p.shutdownMonitor() - if p.mode != ModeWindow || !p.windowStarted.Load() { + if p.config.Mode != ModeWindow || !p.windowStarted.Load() { return nil } - close(p.stopCh) - select { case <-p.doneCh: return nil @@ -142,522 +123,100 @@ func (p *countSketchProcessor) Shutdown(ctx context.Context) error { } } -func (p *countSketchProcessor) processMetrics(ctx context.Context, md pmetric.Metrics) (pmetric.Metrics, error) { - p.recordInput(ctx, md) - - switch p.mode { - case ModeBatch: - out := p.consumeBatch(md) - p.recordOutput(ctx, out) - return out, nil - case ModeWindow: - p.accumulateIntoWindow(md) - if !p.config.DropOriginal { - p.recordOutput(ctx, md) - return md, nil - } - return pmetric.NewMetrics(), nil - default: - p.logger.Error("countsketchprocessor: unknown mode, dropping metrics", zap.Any("mode", p.mode)) - return pmetric.NewMetrics(), nil - } -} - -func (p *countSketchProcessor) accumulateIntoWindow(md pmetric.Metrics) { - rms := md.ResourceMetrics() - for i := 0; i < rms.Len(); i++ { - rm := rms.At(i) - resourceAttrs := rm.Resource().Attributes() - sms := rm.ScopeMetrics() - for j := 0; j < sms.Len(); j++ { - metrics := sms.At(j).Metrics() - for k := 0; k < metrics.Len(); k++ { - p.ingestMetric(resourceAttrs, metrics.At(k)) - } - } - } +// ProcessMetrics is the public test API ADR-0002 promotes from the +// legacy `processMetrics` private method. Decodes input → observes +// each → optionally Ticks (batch mode) → returns the synthesized +// output. Does NOT touch nextConsumer. +func (p *countSketchProcessor) ProcessMetrics(ctx context.Context, md pmetric.Metrics) (pmetric.Metrics, error) { + return p.processMetrics(ctx, md) } -func (p *countSketchProcessor) consumeBatch(md pmetric.Metrics) pmetric.Metrics { - p.accumulateIntoWindow(md) - - sketches := p.buildWindowMetricsAndReset() - if sketches.ResourceMetrics().Len() == 0 { - if p.config.DropOriginal { - return pmetric.NewMetrics() - } - return md - } - - if p.config.DropOriginal { - return sketches - } - - // Expansion mode: keep originals and append sketch summaries. - out := pmetric.NewMetrics() - md.ResourceMetrics().MoveAndAppendTo(out.ResourceMetrics()) - sketches.ResourceMetrics().MoveAndAppendTo(out.ResourceMetrics()) - return out +// ProcessBatch is an alias for ProcessMetrics so tests written +// against the standardized DDSketch-style name keep compiling. +// ADR-0002 lists both spellings as equivalent test hooks. +func (p *countSketchProcessor) ProcessBatch(ctx context.Context, md pmetric.Metrics) (pmetric.Metrics, error) { + return p.processMetrics(ctx, md) } -// matchesMatchers returns true if attrs satisfies all configured LabelMatchers. -func (p *countSketchProcessor) matchesMatchers(attrs pcommon.Map) bool { - for _, m := range p.config.LabelMatchers { - v, ok := attrs.Get(m.Key) - if !ok || v.AsString() != m.Value { - return false - } +// FlushWindow forces a Tick on the runtime and forwards the +// synthesized envelopes via nextConsumer.ConsumeMetrics. No-op when +// no closed windows have data (matches TestEmptyInput shape). +func (p *countSketchProcessor) FlushWindow(ctx context.Context) error { + p.mu.Lock() + out := p.flushToMetrics() + p.mu.Unlock() + if out.ResourceMetrics().Len() == 0 { + return nil } - return true + p.recordOutput(ctx, out) + return p.next.ConsumeMetrics(ctx, out) } -func (p *countSketchProcessor) ingestMetric(resourceAttrs pcommon.Map, metric pmetric.Metric) { - metricName := metric.Name() - switch metric.Type() { - case pmetric.MetricTypeGauge: - dps := metric.Gauge().DataPoints() - for i := 0; i < dps.Len(); i++ { - dp := dps.At(i) - if !p.matchesMatchers(dp.Attributes()) { - continue - } - pk := buildPartitionKey(resourceAttrs, dp.Attributes(), p.config.AggregateBy) - p.updateWindowSketch(pk, metricName, dpValue(dp)) - } - case pmetric.MetricTypeSum: - dps := metric.Sum().DataPoints() - for i := 0; i < dps.Len(); i++ { - dp := dps.At(i) - if !p.matchesMatchers(dp.Attributes()) { - continue - } - pk := buildPartitionKey(resourceAttrs, dp.Attributes(), p.config.AggregateBy) - p.updateWindowSketch(pk, metricName, dpValue(dp)) - } - case pmetric.MetricTypeHistogram: - dps := metric.Histogram().DataPoints() - for i := 0; i < dps.Len(); i++ { - dp := dps.At(i) - if !p.matchesMatchers(dp.Attributes()) { - continue - } - pk := buildPartitionKey(resourceAttrs, dp.Attributes(), p.config.AggregateBy) - p.updateWindowSketch(pk, metricName, float64(dp.Count())) - } - case pmetric.MetricTypeCountSketch: - // Pre-aggregated path: deserialize (or reconstruct from delta) the incoming - // CountSketch and merge it into the running window sketch. - dps := metric.CountSketch().DataPoints() - for i := 0; i < dps.Len(); i++ { - dp := dps.At(i) - if !p.matchesMatchers(dp.Attributes()) { - continue - } - pk := buildPartitionKey(resourceAttrs, dp.Attributes(), p.config.AggregateBy) - if len(dp.Sketch()) == 0 { - // No sketch payload — treat as a raw sample (backwards compat). - p.updateWindowSketch(pk, metricName, 1.0) - continue - } - incoming, err := p.inboundDecodeCS(pk, dp) - if err != nil { - p.logger.Error("countsketchprocessor: failed to decode inbound CountSketch", zap.Error(err)) - continue - } - if incoming == nil { - continue // delta arrived before any full snapshot - } - p.mergeWindowCS(pk, incoming) - } - } -} +// processMetrics is the shared decode/observe core for ConsumeMetrics, +// ProcessMetrics, and ProcessBatch. In batch mode it Ticks inline so +// the caller gets the synthesized output back. In window mode it +// returns either the input md (DropOriginal=false) or an empty +// pmetric (DropOriginal=true) so chained processors can keep working +// on raw samples. +func (p *countSketchProcessor) processMetrics(ctx context.Context, md pmetric.Metrics) (pmetric.Metrics, error) { + p.recordInput(ctx, md) -func dpValue(dp pmetric.NumberDataPoint) float64 { - if dp.ValueType() == pmetric.NumberDataPointValueTypeInt { - return float64(dp.IntValue()) + p.mu.Lock() + if err := p.observeInto(md); err != nil { + p.mu.Unlock() + return md, err } - return dp.DoubleValue() -} - -func (p *countSketchProcessor) updateWindowSketch(partitionKey, itemKey string, value float64) { - p.mu.RLock() - ws, exists := p.activeWindowSketches[partitionKey] - p.mu.RUnlock() - if !exists { - p.mu.Lock() - ws, exists = p.activeWindowSketches[partitionKey] - if !exists { - ws = p.windowSketchPool.Get().(*windowSketch) - if ws.cs != nil { - ws.cs.Reset() - } else { - cs, err := newConfiguredCountSketch(p.config) - if err != nil { - p.logger.Error("Failed to create CountSketch", zap.Error(err)) - p.windowSketchPool.Put(ws) - p.mu.Unlock() - return - } - ws.cs = cs + switch p.config.Mode { + case ModeBatch: + out := p.flushToMetrics() + p.mu.Unlock() + if !p.config.DropOriginal { + if out.ResourceMetrics().Len() == 0 { + p.recordOutput(ctx, md) + return md, nil } - ws.sampleCount = 0 - p.activeWindowSketches[partitionKey] = ws + merged := pmetric.NewMetrics() + md.ResourceMetrics().MoveAndAppendTo(merged.ResourceMetrics()) + out.ResourceMetrics().MoveAndAppendTo(merged.ResourceMetrics()) + p.recordOutput(ctx, merged) + return merged, nil + } + p.recordOutput(ctx, out) + return out, nil + case ModeWindow: + p.mu.Unlock() + if p.config.DropOriginal { + return pmetric.NewMetrics(), nil } + p.recordOutput(ctx, md) + return md, nil + default: p.mu.Unlock() + return pmetric.NewMetrics(), nil } - - ws.mu.Lock() - ws.cs.UpdateString(itemKey, value) - ws.sampleCount++ - ws.mu.Unlock() } -func (p *countSketchProcessor) startWindowLoop(ctx context.Context, ticker *time.Ticker) { +// runWindowLoop is the ticker goroutine for window mode. Drives a +// FlushWindow on every WindowDuration; on shutdown, runs one final +// flush so in-flight window state isn't lost. +func (p *countSketchProcessor) runWindowLoop(ctx context.Context) { + t := time.NewTicker(p.config.WindowDuration) defer func() { - ticker.Stop() + t.Stop() + _ = p.FlushWindow(context.Background()) close(p.doneCh) }() - for { select { case <-ctx.Done(): - p.emitWindowAndReset() return case <-p.stopCh: - p.emitWindowAndReset() return - case <-ticker.C: - p.emitWindowAndReset() - } - } -} - -func (p *countSketchProcessor) buildWindowMetricsAndReset() pmetric.Metrics { - p.mu.Lock() - if len(p.activeWindowSketches) == 0 { - p.mu.Unlock() - return pmetric.NewMetrics() - } - - snapshot := p.activeWindowSketches - p.activeWindowSketches = make(map[string]*windowSketch) - p.mu.Unlock() - - md := pmetric.NewMetrics() - rm := md.ResourceMetrics().AppendEmpty() - sm := rm.ScopeMetrics().AppendEmpty() - sm.Scope().SetName("otelcol/countsketch") - - now := pcommon.NewTimestampFromTime(time.Now()) - - for partitionKey, ws := range snapshot { - ws.mu.Lock() - sampleCount := ws.sampleCount - - var payload []byte - var encoding string - var serErr error - - if p.config.TransmitSketch && ws.cs != nil { - if p.config.DeltaTransmission { - // Delta transmission is proto-only — msgpack delta - // is tracked as a follow-up once sketchlib-go - // grows `apply_delta` semantics. - p.snapshotsMu.Lock() - snap, hasSnap := p.snapshots[partitionKey] - p.snapshotsMu.Unlock() - - if hasSnap { - deltaMsg, deltaErr := countsketch.ComputeDelta(snap, ws.cs, p.config.DeltaThreshold) - if deltaErr == nil { - payload, serErr = countsketch.SerializeDelta(deltaMsg) - } else { - serErr = deltaErr - } - encoding = "proto_delta" - } else { - payload, serErr = serializeCountSketch(ws.cs) - encoding = "proto_full" - } - - newSnap := cloneCS(ws.cs) - p.snapshotsMu.Lock() - p.snapshots[partitionKey] = newSnap - p.snapshotsMu.Unlock() - } else { - // Non-delta path — choose between proto and msgpack - // wire formats based on the config's Encoding field. - switch p.config.Encoding { - case EncodingMsgpack: - payload, serErr = ws.cs.SerializeMsgpack() - encoding = "msgpack_full" - default: - payload, serErr = serializeCountSketch(ws.cs) - encoding = "proto_full" - } + case <-t.C: + if err := p.FlushWindow(context.Background()); err != nil && p.logger != nil { + p.logger.Error("countsketchprocessor: emit failed", zap.Error(err)) } } - - ws.mu.Unlock() - - if ws.cs != nil { - ws.cs.Reset() - } - p.windowSketchPool.Put(ws) - - if p.config.TransmitSketch && serErr != nil { - p.logger.Error("Failed to serialize CountSketch", zap.Error(serErr)) - continue - } - - m := sm.Metrics().AppendEmpty() - m.SetName("countsketch_partition") - m.SetUnit("1") - - if p.config.TransmitSketch { - // Typed CountSketchDataPoint emission — what - // ASAPQuery-backend's modified-OTLP sketch router - // consumes as `Metric.data = CountSketch{...}`. - // Before this change the processor emitted a Gauge - // with the sketch payload stuffed into a - // `sketch_payload` byte attribute, which the backend - // router never recognized as a sketch variant — - // sketch bytes were lost on the wire for any - // consumer that tried to decode them as typed. - csMetric := m.SetEmptyCountSketch() - csMetric.SetAggregationTemporality(pmetric.AggregationTemporalityDelta) - dp := csMetric.DataPoints().AppendEmpty() - dp.SetTimestamp(now) - // The processor's `partition_key` is the natural - // match for CountSketch's `dimension` field (both - // identify which sub-population the sketch covers). - dp.SetDimension(partitionKey) - dp.SetEpsilon(p.config.Epsilon) - dp.SetDelta(p.config.Delta) - dp.SetSketch(payload) - switch encoding { - case "proto_delta": - dp.SetEncoding(pmetric.CountSketchEncodingDelta) - case "msgpack_full": - dp.SetEncoding(pmetric.CountSketchEncodingMsgpack) - default: - // "proto_full" and any unexpected fallback. - dp.SetEncoding(pmetric.CountSketchEncodingProto) - } - // Fields the typed DP doesn't have dedicated setters - // for still go on the attribute map. `sample_count` - // and `window_duration_seconds` are observability - // hints the backend does not use for routing. - dp.Attributes().PutInt("sample_count", int64(sampleCount)) - dp.Attributes().PutInt( - "window_duration_seconds", - int64(p.config.WindowDuration.Seconds()), - ) - } else { - // Non-transmit mode: caller only wants the - // per-partition sample count for monitoring, not - // the sketch bytes. Keep the legacy Gauge emission - // so existing dashboards that read - // `countsketch_partition` as a scalar series - // continue to work. - gauge := m.SetEmptyGauge() - dp := gauge.DataPoints().AppendEmpty() - dp.SetTimestamp(now) - dp.Attributes().PutStr("partition_key", partitionKey) - dp.Attributes().PutInt("sample_count", int64(sampleCount)) - dp.Attributes().PutDouble("epsilon", p.config.Epsilon) - dp.Attributes().PutDouble("delta", p.config.Delta) - dp.Attributes().PutInt( - "window_duration_seconds", - int64(p.config.WindowDuration.Seconds()), - ) - dp.SetDoubleValue(float64(sampleCount)) - } - } - - return md -} - -func (p *countSketchProcessor) emitWindowAndReset() { - md := p.buildWindowMetricsAndReset() - if md.ResourceMetrics().Len() == 0 { - return - } - - p.recordOutput(context.Background(), md) - if err := p.next.ConsumeMetrics(context.Background(), md); err != nil { - p.logger.Error("Failed to emit countsketch partition metrics", zap.Error(err)) - } -} - -func (p *countSketchProcessor) enableSelfMonitoring(settings component.TelemetrySettings, processorID string) { - monitor, err := selfmonitor.New(settings, processorID, typeStr.String(), p.activeSeriesCount) - if err != nil { - if p.logger != nil { - p.logger.Warn("countsketchprocessor: failed to initialize self-monitoring", zap.Error(err)) - } - return - } - p.monitor = monitor -} - -func (p *countSketchProcessor) shutdownMonitor() { - if p.monitor != nil { - p.monitor.Shutdown() - } -} - -func (p *countSketchProcessor) recordInput(ctx context.Context, md pmetric.Metrics) { - if p.monitor != nil { - p.monitor.RecordInput(ctx, md) - } -} - -func (p *countSketchProcessor) recordOutput(ctx context.Context, md pmetric.Metrics) { - if p.monitor != nil { - p.monitor.RecordOutput(ctx, md) - } -} - -func (p *countSketchProcessor) activeSeriesCount() int64 { - p.mu.RLock() - defer p.mu.RUnlock() - return int64(len(p.activeWindowSketches)) -} - -// buildPartitionKey encodes selected attributes as a stable partition key. -// Each key is looked up in dpAttrs first, then resourceAttrs as a fallback. -// Returns "global" when aggregateBy is empty (single undivided partition). -func buildPartitionKey(resourceAttrs, dpAttrs pcommon.Map, aggregateBy []string) string { - if len(aggregateBy) == 0 { - return "global" - } - - keys := make([]string, len(aggregateBy)) - copy(keys, aggregateBy) - sort.Strings(keys) - - sb := builderPool.Get().(*strings.Builder) - sb.Reset() - for _, k := range keys { - var val pcommon.Value - var ok bool - val, ok = dpAttrs.Get(k) - if !ok { - val, ok = resourceAttrs.Get(k) - } - if ok { - sb.WriteString(k) - sb.WriteString("=") - sb.WriteString(val.AsString()) - sb.WriteString(";") - } - } - s := sb.String() - builderPool.Put(sb) - return s -} - -// inboundDecodeCS decodes an incoming CountSketch data point, handling both full -// (Gob-encoded) and sparse-delta payloads. For delta payloads it applies the delta -// onto the last stored inbound snapshot to reconstruct the current full state. -// Returns (nil, nil) when a delta arrives before any full snapshot. -func (p *countSketchProcessor) inboundDecodeCS(partitionKey string, dp pmetric.CountSketchDataPoint) (*countsketch.CountSketch, error) { - payload := dp.Sketch() - - switch dp.Encoding() { - case pmetric.CountSketchEncodingDelta: - p.inboundMu.Lock() - snap, hasSnap := p.inboundSnapshots[partitionKey] - p.inboundMu.Unlock() - if !hasSnap || snap == nil { - return nil, nil - } - reconstructed := cloneCS(snap) - if reconstructed == nil { - return nil, nil - } - deltaMsg, err := countsketch.DeserializeDelta(payload) - if err != nil { - return nil, err - } - countsketch.ApplyDelta(reconstructed, deltaMsg) - p.inboundMu.Lock() - p.inboundSnapshots[partitionKey] = cloneCS(reconstructed) - p.inboundMu.Unlock() - return reconstructed, nil - - default: // CountSketchEncodingProto or unspecified - decoded, err := countsketch.DeserializeCountSketchFromBytes(payload) - if err != nil { - return nil, err - } - p.inboundMu.Lock() - p.inboundSnapshots[partitionKey] = cloneCS(decoded) - p.inboundMu.Unlock() - return decoded, nil - } -} - -// mergeWindowCS merges an incoming pre-aggregated CountSketch into the per-key -// window store, creating the window sketch if it does not yet exist. -func (p *countSketchProcessor) mergeWindowCS(partitionKey string, incoming *countsketch.CountSketch) { - p.mu.RLock() - ws, exists := p.activeWindowSketches[partitionKey] - p.mu.RUnlock() - - if !exists { - p.mu.Lock() - ws, exists = p.activeWindowSketches[partitionKey] - if !exists { - ws = p.windowSketchPool.Get().(*windowSketch) - if ws.cs != nil { - ws.cs.Reset() - } else { - cs, err := newConfiguredCountSketch(p.config) - if err != nil { - p.logger.Error("countsketchprocessor: failed to create CS for merge", zap.Error(err)) - p.windowSketchPool.Put(ws) - p.mu.Unlock() - return - } - ws.cs = cs - } - ws.sampleCount = 0 - p.activeWindowSketches[partitionKey] = ws - } - p.mu.Unlock() - } - - ws.mu.Lock() - defer ws.mu.Unlock() - if err := ws.cs.Merge(incoming); err != nil { - p.logger.Error("countsketchprocessor: failed to merge CountSketch", zap.Error(err)) - } - ws.sampleCount++ -} - -func serializeCountSketch(s *countsketch.CountSketch) ([]byte, error) { - if s == nil { - return nil, nil - } - env, err := s.SerializePortable() - if err != nil { - return nil, err - } - return proto.Marshal(env) -} - -// cloneCS returns a deep copy of cs suitable for use as a delta snapshot. -func cloneCS(cs *countsketch.CountSketch) *countsketch.CountSketch { - data, err := cs.SerializeProtoBytes() - if err != nil { - return nil - } - clone, err := countsketch.DeserializeCountSketchFromProtoBytes(data) - if err != nil { - return nil } - return clone } diff --git a/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/processor_test.go b/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/processor_test.go index fac47b8b..9dde5b47 100644 --- a/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/processor_test.go +++ b/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/processor_test.go @@ -29,7 +29,7 @@ func TestProcessorPassThrough(t *testing.T) { metrics := buildTestMetrics() // Process metrics - out, err := proc.processMetrics(context.Background(), metrics) + out, err := proc.ProcessMetrics(context.Background(), metrics) require.NoError(t, err) // Shutdown @@ -57,7 +57,7 @@ func TestProcessorFlushLogic(t *testing.T) { // Send some data to populate sketches metrics := buildTestMetrics() - _, err = proc.processMetrics(context.Background(), metrics) + _, err = proc.ProcessMetrics(context.Background(), metrics) require.NoError(t, err) // Wait for a window flush (window is 100ms) @@ -88,7 +88,7 @@ func TestBatchModePassThroughAndSummary(t *testing.T) { metrics := buildTestMetrics() // Process a single batch. - out, err := proc.processMetrics(context.Background(), metrics) + out, err := proc.ProcessMetrics(context.Background(), metrics) require.NoError(t, err) err = proc.Shutdown(context.Background()) @@ -144,7 +144,7 @@ func TestGroupByPartitioning(t *testing.T) { m.SetEmptyGauge().DataPoints().AppendEmpty().SetDoubleValue(1.0) } - out, err := proc.processMetrics(context.Background(), md) + out, err := proc.ProcessMetrics(context.Background(), md) require.NoError(t, err) // Collect all partition_key values from the output. Uses the @@ -193,7 +193,7 @@ func TestWindowModeGroupBy(t *testing.T) { dp.Attributes().PutStr("service.name", svc) } - _, err := proc.processMetrics(context.Background(), md) + _, err := proc.ProcessMetrics(context.Background(), md) require.NoError(t, err) time.Sleep(200 * time.Millisecond) @@ -230,7 +230,7 @@ func TestBatchModeDropOriginal(t *testing.T) { metrics := buildTestMetrics() - out, err := proc.processMetrics(context.Background(), metrics) + out, err := proc.ProcessMetrics(context.Background(), metrics) require.NoError(t, err) err = proc.Shutdown(context.Background()) @@ -314,7 +314,7 @@ func TestEmptyInput(t *testing.T) { proc := newProcessor(zap.NewNop(), cfg, next) empty := pmetric.NewMetrics() - out, err := proc.processMetrics(context.Background(), empty) + out, err := proc.ProcessMetrics(context.Background(), empty) require.NoError(t, err) require.Equal(t, 0, out.ResourceMetrics().Len()) } @@ -338,9 +338,9 @@ func TestBatchModeNoStatePersistence(t *testing.T) { metrics1 := buildTestMetrics() metrics2 := buildTestMetrics() - out1, err := proc.processMetrics(context.Background(), metrics1) + out1, err := proc.ProcessMetrics(context.Background(), metrics1) require.NoError(t, err) - out2, err := proc.processMetrics(context.Background(), metrics2) + out2, err := proc.ProcessMetrics(context.Background(), metrics2) require.NoError(t, err) // In batch mode each call returns its own sketch summary in the output; state @@ -369,7 +369,7 @@ func TestWindowModeConcurrentConsume(t *testing.T) { go func() { defer wg.Done() metrics := buildTestMetrics() - _, _ = proc.processMetrics(context.Background(), metrics) + _, _ = proc.ProcessMetrics(context.Background(), metrics) }() } wg.Wait() @@ -397,7 +397,7 @@ func TestWindowModeFlushDuringConsume(t *testing.T) { go func() { for i := 0; i < 50; i++ { metrics := buildTestMetrics() - _, _ = proc.processMetrics(context.Background(), metrics) + _, _ = proc.ProcessMetrics(context.Background(), metrics) } close(done) }() @@ -427,7 +427,7 @@ func TestShutdownDuringConsume(t *testing.T) { defer wg.Done() for i := 0; i < 100; i++ { metrics := buildTestMetrics() - _, _ = proc.processMetrics(context.Background(), metrics) + _, _ = proc.ProcessMetrics(context.Background(), metrics) } }() wg.Add(1) diff --git a/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/shim_helpers.go b/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/shim_helpers.go new file mode 100644 index 00000000..18713a46 --- /dev/null +++ b/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/shim_helpers.go @@ -0,0 +1,367 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +// File shim_helpers.go houses the OTel-side glue the shim needs but +// the host-neutral runtime cannot provide: +// +// - observeInto: walk md, merge resource-attrs into dp-attrs (so +// AggregateBy lookups find resource keys), and feed each +// observation into the single Precompute. The metric name is +// stashed on ObservationValue.Bytes so the SketchObserver can +// call UpdateString with the legacy key (metric.Name()). +// - flushToMetrics: tick the Precompute, then either build typed +// CountSketchDataPoints (TransmitSketch=true, the new wire +// format) or fall back to a legacy Gauge per partition key +// (TransmitSketch=false). The Gauge fallback is required because +// existing dashboards keyed by `countsketch_partition` as a +// scalar series still rely on it. +// - stampDPMetadata: fill the typed CountSketchDataPoint's +// Dimension / Epsilon / Delta / Encoding fields the host-neutral +// adapter doesn't know about. +// - partitionKeyFromEnvelope: rebuild the legacy +// `key=value;key=value;` partition-key string from the +// envelope's Labels. + +package countsketchprocessor + +import ( + "sort" + "strings" + + "go.opentelemetry.io/collector/pdata/pcommon" + "go.opentelemetry.io/collector/pdata/pmetric" + "go.uber.org/zap" + + precompute "github.com/ProjectASAP/asap-precompute-go" +) + +// observeInto walks md and feeds one host-neutral Observation per +// data point into the Precompute. Resource attributes are merged +// into the data-point label set before observe so AggregateBy +// lookups find resource-level keys (mirroring the legacy +// buildPartitionKey's "look up dpAttrs first, then resourceAttrs" +// fallback). Sketch-typed inputs are surfaced as KindEnvelope so +// Precompute.ObserveEnvelope merges them in place — the runtime +// never expands envelopes to scalar samples. +func (p *countSketchProcessor) observeInto(md pmetric.Metrics) error { + rms := md.ResourceMetrics() + for i := 0; i < rms.Len(); i++ { + rm := rms.At(i) + resourceAttrs := rm.Resource().Attributes() + sms := rm.ScopeMetrics() + for j := 0; j < sms.Len(); j++ { + ms := sms.At(j).Metrics() + for k := 0; k < ms.Len(); k++ { + if err := p.observeMetric(resourceAttrs, ms.At(k)); err != nil { + return err + } + } + } + } + return nil +} + +// observeMetric pushes one Observation per data point. Scalar +// metrics produce KindFloat observations carrying the metric name +// in ObservationValue.Bytes (so the SketchObserver invokes +// UpdateString with the legacy key). Sketch-typed metrics produce +// KindEnvelope observations with the payload bytes preserved. +func (p *countSketchProcessor) observeMetric(resourceAttrs pcommon.Map, m pmetric.Metric) error { + name := m.Name() + switch m.Type() { + case pmetric.MetricTypeGauge: + dps := m.Gauge().DataPoints() + for i := 0; i < dps.Len(); i++ { + dp := dps.At(i) + if err := p.observeFloat(name, resourceAttrs, dp.Attributes(), dp.Timestamp(), numberValue(dp)); err != nil { + return err + } + } + case pmetric.MetricTypeSum: + dps := m.Sum().DataPoints() + for i := 0; i < dps.Len(); i++ { + dp := dps.At(i) + if err := p.observeFloat(name, resourceAttrs, dp.Attributes(), dp.Timestamp(), numberValue(dp)); err != nil { + return err + } + } + case pmetric.MetricTypeHistogram: + dps := m.Histogram().DataPoints() + for i := 0; i < dps.Len(); i++ { + dp := dps.At(i) + if err := p.observeFloat(name, resourceAttrs, dp.Attributes(), dp.Timestamp(), float64(dp.Count())); err != nil { + return err + } + } + case pmetric.MetricTypeCountSketch: + dps := m.CountSketch().DataPoints() + for i := 0; i < dps.Len(); i++ { + dp := dps.At(i) + env := &precompute.SketchEnvelope{ + SchemaVersion: 1, + SketchType: precompute.SketchTypeCountSketch, + Labels: mergedLabels(resourceAttrs, dp.Attributes()), + WindowStartMs: uint64(dp.StartTimestamp() / 1_000_000), + WindowEndMs: uint64(dp.Timestamp() / 1_000_000), + Encoding: countSketchEncodingToHostNeutral(dp.Encoding()), + Payload: copyBytes(dp.Sketch()), + } + obs := precompute.Observation{ + TimestampMs: uint64(dp.Timestamp() / 1_000_000), + Metric: name, + Labels: env.Labels, + Value: precompute.EnvelopeValue(env), + } + if err := p.pc.Observe(&obs); err != nil { + return err + } + } + } + return nil +} + +// observeFloat is the scalar-input fast path. Metric name travels in +// ObservationValue.Bytes as a side-channel so the SketchObserver can +// call UpdateString(metricName, value) with the same key the legacy +// processor used. +func (p *countSketchProcessor) observeFloat(name string, resourceAttrs, dpAttrs pcommon.Map, ts pcommon.Timestamp, val float64) error { + obs := precompute.Observation{ + TimestampMs: uint64(ts / 1_000_000), + Metric: name, + Labels: mergedLabels(resourceAttrs, dpAttrs), + Value: precompute.ObservationValue{ + Kind: precompute.KindFloat, + Float: val, + Bytes: []byte(name), + }, + } + return p.pc.Observe(&obs) +} + +// flushToMetrics ticks the Precompute and converts the closed +// envelopes into pmetric.Metrics. TransmitSketch=true takes the +// otel.Encode path then stamps typed-DP fields the host-neutral +// adapter doesn't know about (Dimension / Epsilon / Delta / Encoding). +// TransmitSketch=false falls back to the legacy Gauge-per-partition +// emission so existing dashboards keep working. +// +// Force-drain semantics: legacy emitWindowAndReset rotated regardless +// of wall-clock; passing a far-future tick timestamp ensures +// Precompute.Tick always considers the active window due. +func (p *countSketchProcessor) flushToMetrics() pmetric.Metrics { + const forceTickMs uint64 = 1<<62 - 1 + envs := p.pc.Tick(forceTickMs) + if len(envs) == 0 { + return pmetric.NewMetrics() + } + if p.config.TransmitSketch { + return p.encodeSketchMetrics(envs) + } + return p.encodeGaugeMetrics(envs) +} + +// encodeSketchMetrics goes through the runtime's OTel adapter then +// stamps the legacy typed-DP fields the host-neutral encoder doesn't +// populate (Dimension / Epsilon / Delta / Encoding). +func (p *countSketchProcessor) encodeSketchMetrics(envs []*precompute.SketchEnvelope) pmetric.Metrics { + encoded, err := p.adapter.Encode(envs) + if err != nil { + if p.logger != nil { + p.logger.Error("countsketchprocessor: encode failed", zap.Error(err)) + } + return pmetric.NewMetrics() + } + md, ok := encoded.(pmetric.Metrics) + if !ok { + return pmetric.NewMetrics() + } + p.stampDPMetadata(md, envs) + return md +} + +// stampDPMetadata walks the encoded pmetric output in encode-order +// (groupOrder by ResourceLabels, envelopes within group preserved) +// and copies legacy-typed CountSketchDataPoint fields onto each DP: +// Dimension (the partition key), Epsilon, Delta, and the +// PROTO_FULL-vs-PROTO_DELTA encoding tag (the runtime adapter always +// writes Proto). Also stamps AggregationTemporality from the +// envelope onto the parent CountSketch metric. +func (p *countSketchProcessor) stampDPMetadata(md pmetric.Metrics, envs []*precompute.SketchEnvelope) { + idx := 0 + rms := md.ResourceMetrics() + for i := 0; i < rms.Len() && idx < len(envs); i++ { + sms := rms.At(i).ScopeMetrics() + for j := 0; j < sms.Len() && idx < len(envs); j++ { + ms := sms.At(j).Metrics() + for k := 0; k < ms.Len() && idx < len(envs); k++ { + m := ms.At(k) + if m.Type() != pmetric.MetricTypeCountSketch { + idx++ + continue + } + cs := m.CountSketch() + cs.SetAggregationTemporality(pmetric.AggregationTemporality(envs[idx].AggregationTemporality)) + dps := cs.DataPoints() + for l := 0; l < dps.Len() && idx < len(envs); l++ { + dp := dps.At(l) + env := envs[idx] + dp.SetDimension(partitionKeyFromEnvelope(env, p.config)) + dp.SetEpsilon(p.config.Epsilon) + dp.SetDelta(p.config.Delta) + dp.SetEncoding(hostNeutralToTypedEncoding(env.Encoding, p.config.Encoding)) + idx++ + } + } + } + } +} + +// encodeGaugeMetrics is the non-transmit emission path. The legacy +// processor produced one Gauge metric named "countsketch_partition" +// per partition with attributes (partition_key, sample_count, epsilon, +// delta, window_duration_seconds) and the sample count as the gauge +// value. Existing dashboards consume this as a scalar series. +func (p *countSketchProcessor) encodeGaugeMetrics(envs []*precompute.SketchEnvelope) pmetric.Metrics { + md := pmetric.NewMetrics() + rm := md.ResourceMetrics().AppendEmpty() + sm := rm.ScopeMetrics().AppendEmpty() + sm.Scope().SetName("otelcol/countsketch") + for _, env := range envs { + m := sm.Metrics().AppendEmpty() + m.SetName(outputMetricName) + m.SetUnit("1") + gauge := m.SetEmptyGauge() + dp := gauge.DataPoints().AppendEmpty() + dp.SetTimestamp(pcommon.Timestamp(env.WindowEndMs * 1_000_000)) + dp.Attributes().PutStr("partition_key", partitionKeyFromEnvelope(env, p.config)) + dp.Attributes().PutInt("sample_count", int64(env.Count)) + dp.Attributes().PutDouble("epsilon", p.config.Epsilon) + dp.Attributes().PutDouble("delta", p.config.Delta) + dp.Attributes().PutInt( + "window_duration_seconds", + int64(p.config.WindowDuration.Seconds()), + ) + dp.SetDoubleValue(float64(env.Count)) + } + return md +} + +// partitionKeyFromEnvelope reconstructs the legacy buildPartitionKey +// output from the envelope's Labels. With GlobalAggregation the +// runtime strips Labels entirely; we hard-code "global" to match +// buildPartitionKey's empty-AggregateBy branch. Otherwise we render +// the AggregateBy-projected labels in the same `key=value;` form the +// legacy builder produced. +func partitionKeyFromEnvelope(env *precompute.SketchEnvelope, cfg *Config) string { + if len(cfg.AggregateBy) == 0 { + return "global" + } + keep := make(map[string]struct{}, len(cfg.AggregateBy)) + for _, k := range cfg.AggregateBy { + keep[k] = struct{}{} + } + keys := make([]string, 0, len(cfg.AggregateBy)) + values := make(map[string]string, len(cfg.AggregateBy)) + for _, kv := range env.Labels { + if _, ok := keep[kv.Key]; !ok { + continue + } + if _, seen := values[kv.Key]; !seen { + keys = append(keys, kv.Key) + } + values[kv.Key] = kv.Value + } + sort.Strings(keys) + var sb strings.Builder + for _, k := range keys { + sb.WriteString(k) + sb.WriteByte('=') + sb.WriteString(values[k]) + sb.WriteByte(';') + } + return sb.String() +} + +// mergedLabels combines resource and DP attribute sets into one +// host-neutral KeyValue slice, sorted by key. DP entries take +// precedence on duplicate keys (mirroring the legacy +// buildPartitionKey's "look up dpAttrs first, then resourceAttrs" +// fallback semantics). +func mergedLabels(resourceAttrs, dpAttrs pcommon.Map) []precompute.KeyValue { + if resourceAttrs.Len() == 0 && dpAttrs.Len() == 0 { + return nil + } + merged := make(map[string]string, resourceAttrs.Len()+dpAttrs.Len()) + resourceAttrs.Range(func(k string, v pcommon.Value) bool { + merged[k] = v.AsString() + return true + }) + dpAttrs.Range(func(k string, v pcommon.Value) bool { + merged[k] = v.AsString() + return true + }) + keys := make([]string, 0, len(merged)) + for k := range merged { + keys = append(keys, k) + } + sort.Strings(keys) + out := make([]precompute.KeyValue, 0, len(keys)) + for _, k := range keys { + out = append(out, precompute.KeyValue{Key: k, Value: merged[k]}) + } + return out +} + +// numberValue extracts a float from a NumberDataPoint regardless of +// its int/double tag. +func numberValue(dp pmetric.NumberDataPoint) float64 { + if dp.ValueType() == pmetric.NumberDataPointValueTypeInt { + return float64(dp.IntValue()) + } + return dp.DoubleValue() +} + +// copyBytes returns a defensive copy of b so the host-neutral +// envelope doesn't alias pmetric's storage. +func copyBytes(b []byte) []byte { + if len(b) == 0 { + return nil + } + out := make([]byte, len(b)) + copy(out, b) + return out +} + +// countSketchEncodingToHostNeutral mirrors the otel adapter's +// encoding-mapping helper. Inlined here to keep the shim's import +// surface narrow. +func countSketchEncodingToHostNeutral(e pmetric.CountSketchEncoding) precompute.Encoding { + switch e { + case pmetric.CountSketchEncodingProto: + return precompute.EncodingProtoFull + case pmetric.CountSketchEncodingDelta: + return precompute.EncodingProtoDelta + case pmetric.CountSketchEncodingMsgpack: + return precompute.EncodingMsgpack + } + return precompute.EncodingProtoFull +} + +// hostNeutralToTypedEncoding maps the runtime's host-neutral encoding +// tag back to the typed pmetric.CountSketchEncoding enum the legacy +// processor wrote. The runtime's own encode helper always writes +// Proto; we override here so PROTO_DELTA frames are visible to +// downstream consumers (the backend's modified-OTLP router and the +// existing delta-transmission tests both dispatch on this enum). +func hostNeutralToTypedEncoding(e precompute.Encoding, cfgEnc SketchEncoding) pmetric.CountSketchEncoding { + switch e { + case precompute.EncodingProtoDelta: + return pmetric.CountSketchEncodingDelta + case precompute.EncodingMsgpack: + return pmetric.CountSketchEncodingMsgpack + } + if cfgEnc == EncodingMsgpack { + return pmetric.CountSketchEncodingMsgpack + } + return pmetric.CountSketchEncodingProto +} diff --git a/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/sketch_wrapper.go b/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/sketch_wrapper.go new file mode 100644 index 00000000..22645165 --- /dev/null +++ b/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/sketch_wrapper.go @@ -0,0 +1,226 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +package countsketchprocessor + +import ( + "errors" + "fmt" + + countsketch "github.com/ProjectASAP/sketchlib-go/sketches/CountSketch" + + precompute "github.com/ProjectASAP/asap-precompute-go" +) + +// countSketchWrapper adapts a sketchlib-go *countsketch.CountSketch to +// the host-neutral precompute.Sketch + precompute.FrequencySketch +// interfaces the asap-precompute-go runtime requires (Phase 2 step 2.8). +// +// The wrapper preserves the exact wire format the legacy CountSketch +// processor emitted: sketchlib-go's SerializeProtoBytes on a state +// proto wrapped in a SketchEnvelope (legacy serializeCountSketch == +// SerializePortable + proto.Marshal == SerializeProtoBytes). That's +// what makes the parity-harness byte-equality invariant honest: +// runtime and legacy paths both call the same serializer on the same +// sketch state. +type countSketchWrapper struct { + cs *countsketch.CountSketch + rows int + cols int +} + +// newCountSketchWrapper constructs a fresh CountSketch with the given +// (rows, cols) — derived from epsilon/delta the same way the legacy +// processor's newConfiguredCountSketch did. Returns an error if +// sketchlib's constructor rejects the dimensions. +func newCountSketchWrapper(rows, cols int) (*countSketchWrapper, error) { + cs, err := countsketch.NewCountSketch(rows, cols) + if err != nil { + return nil, fmt.Errorf("countsketchprocessor: NewCountSketch(%d, %d): %w", rows, cols, err) + } + return &countSketchWrapper{cs: cs, rows: rows, cols: cols}, nil +} + +// updateString mirrors the legacy ws.cs.UpdateString(itemKey, value) +// call. The shim's observer routes KindFloat observations through +// this method so the runtime drives the same hot path the legacy +// processor used. +func (w *countSketchWrapper) updateString(key string, count float64) { + w.cs.UpdateString(key, count) +} + +// Snapshot returns the canonical proto-encoded SketchEnvelope bytes, +// byte-identical to the legacy processor's serializeCountSketch +// output (SerializePortable + proto.Marshal). +func (w *countSketchWrapper) Snapshot() ([]byte, error) { + if w.cs == nil { + return nil, nil + } + return w.cs.SerializeProtoBytes() +} + +// ComputeDeltaAgainst mirrors the legacy delta-encoding path: +// deserialize the previous snapshot, compute a delta against the +// current sketch, return SerializeDelta bytes. On any decode/compute +// failure (e.g. no previous snapshot), fall back to a full snapshot +// with isFull=true so the runtime emits a PROTO_FULL frame. +func (w *countSketchWrapper) ComputeDeltaAgainst(prev []byte, threshold uint64) ([]byte, bool, error) { + if w.cs == nil { + return nil, true, nil + } + if len(prev) == 0 { + full, err := w.Snapshot() + return full, true, err + } + prevCS, err := countsketch.DeserializeCountSketchFromProtoBytes(prev) + if err != nil { + full, fErr := w.Snapshot() + return full, true, fErr + } + deltaMsg, err := countsketch.ComputeDelta(prevCS, w.cs, float64(threshold)) + if err != nil { + full, fErr := w.Snapshot() + return full, true, fErr + } + payload, err := countsketch.SerializeDelta(deltaMsg) + if err != nil { + full, fErr := w.Snapshot() + return full, true, fErr + } + return payload, false, nil +} + +// ApplyDelta merges a sparse delta payload into this sketch in place. +// Used by Precompute.ObserveEnvelope when an upstream agent forwards +// a PROTO_DELTA-encoded CountSketchDataPoint. +func (w *countSketchWrapper) ApplyDelta(delta []byte) error { + if len(delta) == 0 { + return errors.New("countsketchprocessor wrapper: ApplyDelta with empty payload") + } + if w.cs == nil { + cs, err := countsketch.NewCountSketch(w.rows, w.cols) + if err != nil { + return err + } + w.cs = cs + } + deltaMsg, err := countsketch.DeserializeDelta(delta) + if err != nil { + return fmt.Errorf("countsketchprocessor wrapper: DeserializeDelta: %w", err) + } + countsketch.ApplyDelta(w.cs, deltaMsg) + return nil +} + +// Merge folds another CountSketch into this one. The runtime calls +// this when an envelope-valued observation arrives encoded as +// PROTO_FULL (the snapshot bytes are first decoded via +// DeserializeCountSketchFromProtoBytes by the runtime, then this +// wrapper's Merge is called). +func (w *countSketchWrapper) Merge(other precompute.Sketch) error { + if other == nil { + return nil + } + o, ok := other.(*countSketchWrapper) + if !ok { + return fmt.Errorf("countSketchWrapper: Merge with %T", other) + } + if o.cs == nil { + return nil + } + if w.cs == nil { + cs, err := countsketch.NewCountSketch(w.rows, w.cols) + if err != nil { + return err + } + w.cs = cs + } + return w.cs.Merge(o.cs) +} + +// Reset zeros the sketch in place, preserving (rows, cols). Mirrors +// the legacy windowSketchPool path's `ws.cs.Reset()` call. +func (w *countSketchWrapper) Reset() { + if w.cs != nil { + w.cs.Reset() + } +} + +// EstimateCount implements precompute.FrequencySketch. The key is +// the opaque byte slice the sketch indexes by (the same shape passed +// to ObservationValue.Bytes); CountSketch's median-of-rows estimator +// returns a non-negative integer count which we surface as float64 +// per the host-neutral contract. +func (w *countSketchWrapper) EstimateCount(key []byte) float64 { + if w.cs == nil || len(key) == 0 { + return 0 + } + return float64(w.cs.EstimateStringCount(string(key))) +} + +// TopK implements precompute.FrequencySketch. Returns up to k entries +// from the sketch's internal TopK heap, sorted descending by Count. +// sketchlib's heap is min-rooted so the wrapper sort-descends after +// copying; insertion sort is fine because k is bounded by sketchlib's +// TOPK_SIZE (small constant). +func (w *countSketchWrapper) TopK(k int) []precompute.FrequencyEntry { + if k <= 0 || w.cs == nil || w.cs.TopK == nil { + return nil + } + heap := w.cs.TopK.Heap + if len(heap) == 0 { + return nil + } + out := make([]precompute.FrequencyEntry, 0, len(heap)) + for _, item := range heap { + out = append(out, precompute.FrequencyEntry{ + Key: []byte(item.Key), + Count: float64(item.Count), + }) + } + for i := 1; i < len(out); i++ { + for j := i; j > 0 && out[j].Count > out[j-1].Count; j-- { + out[j], out[j-1] = out[j-1], out[j] + } + } + if len(out) > k { + out = out[:k] + } + return out +} + +// countSketchObserver routes a KindFloat observation into the wrapped +// CountSketch via UpdateString. Legacy hot path: ws.cs.UpdateString( +// metric.Name(), value). The metric name travels through the host- +// neutral interface as ObservationValue.Bytes (set by the shim's +// observe path); when absent, the observer falls back to the runtime +// config's MetricName. +type countSketchObserver struct { + defaultKey string +} + +func (o countSketchObserver) Observe(s precompute.Sketch, v precompute.ObservationValue) error { + w, ok := s.(*countSketchWrapper) + if !ok { + return fmt.Errorf("countSketchObserver: sketch is %T", s) + } + if v.Kind != precompute.KindFloat { + return fmt.Errorf("countSketchObserver: unsupported value kind %s", v.Kind) + } + key := o.defaultKey + if len(v.Bytes) > 0 { + key = string(v.Bytes) + } + w.updateString(key, v.Float) + return nil +} + +// Compile-time assertions that the wrapper satisfies both the base +// Sketch trait (used by the runtime's window state machine) and the +// FrequencySketch query trait (used by adapter code that needs typed +// frequency queries). +var ( + _ precompute.Sketch = (*countSketchWrapper)(nil) + _ precompute.FrequencySketch = (*countSketchWrapper)(nil) + _ precompute.SketchObserver = countSketchObserver{} +)