refactor(countsketchprocessor): thin shim delegating to asap-precompute-go (Phase 2 step 2.8)

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

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")

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

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

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