fix(asapedgeprocessor): feed CountMinSketch KindBytes so it records frequencies

opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/cms_observe_test.go

@@ -0,0 +1,79 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package asapedgeprocessor
+
+import (
+	"testing"
+	"time"
+
+	precompute "github.com/ProjectASAP/asap-precompute-go"
+	"github.com/ProjectASAP/asap-precompute-go/sketches"
+	"go.uber.org/zap"
+)
+
+// TestCountMinSketchRecordsFrequency guards the fused CountMinSketch path.
+// CMSObserver only accepts KindBytes (it hashes the encoded attribute key to
+// count series cardinality, not the numeric value). The fused observe used to
+// hand it precompute.FloatValue for every family, so CMSObserver.Observe
+// rejected every sample and the (then-swallowed) error left the sketch empty —
+// an envelope was still emitted, so the existing presence-only test stayed
+// green while nothing was recorded.
+//
+// This asserts both halves of the fix: ObserveKeyed no longer errors for CMS,
+// and the emitted envelope, once reconstructed, actually estimates the inserted
+// key's frequency.
+func TestCountMinSketchRecordsFrequency(t *testing.T) {
+	cfg := &Config{
+		ShardCount:     1,
+		WindowDuration: time.Hour,
+		Metrics:        []MetricFamily{{Metric: "m", Family: FamilyCountMinSketch}},
+		Cold:           ColdConfig{Enabled: false},
+	}
+	if err := cfg.Validate(); err != nil {
+		t.Fatal(err)
+	}
+	sa, ok := newSketchAggregator("m", &cfg.Metrics[0], time.Hour, zap.NewNop())
+	if !ok {
+		t.Fatal("newSketchAggregator(CountMinSketch) returned ok=false")
+	}
+
+	const n = 40
+	am := map[string]string{"zone": "z0"}
+	base := uint64(time.Unix(1700000000, 0).UnixMilli())
+	for i := 0; i < n; i++ {
+		sa.observe(am, float64(i), base+uint64(i))
+	}
+
+	// Core invariant: the value kind matches the observer. Pre-fix this is
+	// non-nil ("CMSObserver: expected KindBytes, got KindFloat").
+	if sa.lastObserveErr != nil {
+		t.Fatalf("CMS observe errored — value-kind regressed: %v", sa.lastObserveErr)
+	}
+
+	// Semantic check: reconstruct the emitted CMS and confirm it estimates the
+	// inserted attribute key's frequency (~n), not zero.
+	envs := sa.pc.Drain()
+	rows, cols := csmDims(&cfg.Metrics[0])
+	rebuilt := sketches.NewCMSWrapper(rows, cols, false)
+	gotEnvelope := false
+	for _, env := range envs {
+		if env.SketchType != precompute.SketchTypeCountMinSketch || len(env.Payload) == 0 {
+			continue
+		}
+		if err := rebuilt.ApplyDelta(env.Payload); err != nil {
+			t.Fatalf("ApplyDelta(payload): %v", err)
+		}
+		gotEnvelope = true
+	}
+	if !gotEnvelope {
+		t.Fatal("no CountMinSketch envelope with a non-empty payload was emitted")
+	}
+
+	key := []byte(precompute.AttributesKey([]precompute.KeyValue{{Key: "zone", Value: "z0"}}, nil))
+	// CMS never underestimates; with a single distinct key there is nothing to
+	// collide with, so the estimate should be the exact insert count.
+	if got := rebuilt.EstimateCount(key); got < float64(n) {
+		t.Fatalf("EstimateCount(zone=z0)=%v, want >= %d (frequency not recorded)", got, n)
+	}
+}

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

@@ -124,7 +124,7 @@ func newProcessor(cfg *Config, set processor.Settings, next consumer.Metrics) (*
 			sh.sumAggs[name] = newSumAggregator(fam.AggregateBy)
 		}
 		for name, fam := range p.sketchMetrics {
-			if sa, ok := newSketchAggregator(name, fam, cfg.WindowDuration); ok {
+			if sa, ok := newSketchAggregator(name, fam, cfg.WindowDuration, p.logger); ok {
 				sh.sketchAggs[name] = sa
 			}
 		}

opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/sample_p_path_test.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"go.opentelemetry.io/collector/confmap"
+	"go.uber.org/zap"
 )
 
 // samplePProbe is implemented by the sampling-aware sketch wrappers
@@ -104,7 +105,7 @@ func TestFusedSketchBuildAppliesSampleP(t *testing.T) {
 			if err := cfg.Validate(); err != nil {
 				t.Fatal(err)
 			}
-			sa, ok := newSketchAggregator("m", &cfg.Metrics[0], time.Hour)
+			sa, ok := newSketchAggregator("m", &cfg.Metrics[0], time.Hour, zap.NewNop())
 			if !ok {
 				t.Fatalf("family %s: newSketchAggregator returned not-ok", tc.family)
 			}

opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/warm_sketch.go

@@ -10,6 +10,7 @@ import (
 	oteladapter "github.com/ProjectASAP/asap-precompute-go/otel"
 	"github.com/ProjectASAP/asap-precompute-go/sketches"
 	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.uber.org/zap"
 )
 
 // sketchAggregator wraps a precompute.Precompute for one sketch-family metric.
@@ -26,6 +27,19 @@ type sketchAggregator struct {
 	// so the built sampling probability is observable (e.g. in tests) without
 	// reaching into precompute internals.
 	factory precompute.SketchFactory
+	// valueAsKey routes the observation as a KindBytes attribute key instead of
+	// the numeric float. CountMinSketch counts attribute-set cardinality and its
+	// observer requires KindBytes (sketches.CMSObserver); every other wired
+	// family observes the numeric value as KindFloat. Feeding CMS a KindFloat
+	// makes the observer reject every sample, leaving an empty sketch.
+	valueAsKey bool
+	logger     *zap.Logger
+	// lastObserveErr is the most recent ObserveKeyed result (nil when the last
+	// sample recorded cleanly). The observe error used to be discarded, which
+	// hid exactly the CMS KindBytes mismatch above; it is now retained (and
+	// logged once) so a value-kind regression is visible instead of silent.
+	lastObserveErr   error
+	loggedObserveErr bool
 }
 
 // fnv64 derives a stable per-metric AggID (matches the standalone sketch
@@ -42,7 +56,10 @@ func fnv64(s string) uint64 {
 // newSketchAggregator builds the aggregator for fam, or (nil,false) if the
 // family isn't wired yet. DDSketch (observes the number value) is wired;
 // KLL/HLL/CS/CMS follow with their per-family params + observe-subject.
-func newSketchAggregator(metric string, fam *MetricFamily, window time.Duration) (*sketchAggregator, bool) {
+func newSketchAggregator(metric string, fam *MetricFamily, window time.Duration, logger *zap.Logger) (*sketchAggregator, bool) {
+	if logger == nil {
+		logger = zap.NewNop()
+	}
 	var (
 		st       precompute.SketchType
 		factory  precompute.SketchFactory
@@ -103,10 +120,12 @@ func newSketchAggregator(metric string, fam *MetricFamily, window time.Duration)
 		Temporality:    int32(pmetric.AggregationTemporalityDelta),
 	}
 	return &sketchAggregator{
-		pc:      precompute.New(pcfg, factory, observer),
-		pcfg:    pcfg,
-		enc:     &oteladapter.AdapterConfig{MetricSuffix: "_" + string(fam.Family), DropOriginal: true},
-		factory: factory,
+		pc:         precompute.New(pcfg, factory, observer),
+		pcfg:       pcfg,
+		enc:        &oteladapter.AdapterConfig{MetricSuffix: "_" + string(fam.Family), DropOriginal: true},
+		factory:    factory,
+		valueAsKey: fam.Family == FamilyCountMinSketch,
+		logger:     logger,
 	}, true
 }
 
@@ -134,12 +153,31 @@ func kvFromMap(am map[string]string) []precompute.KeyValue {
 // observe feeds one sample. The precompute key is built once here from the
 // shared decoded attrs and passed via ObserveKeyed (no internal re-key).
 func (s *sketchAggregator) observe(am map[string]string, val float64, tsMs uint64) {
+	kv := kvFromMap(am)
 	obs := &precompute.Observation{
 		TimestampMs: tsMs,
-		Labels:      kvFromMap(am),
+		Labels:      kv,
 		Value:       precompute.FloatValue(val),
 	}
-	_ = s.pc.ObserveKeyed(s.pcfg.SeriesKeyFor(obs), obs)
+	if s.valueAsKey {
+		// CountMinSketch's observer consumes KindBytes: it hashes the encoded
+		// attribute key (matching the standalone countminsketchprocessor's
+		// AttributesKey(labels, nil)) to count series cardinality, not the numeric
+		// value. AggregateBy grouping is applied separately by SeriesKeyFor below,
+		// so the inserted key is the full attribute set (nil), identical to the
+		// standalone shim.
+		obs.Value = precompute.BytesValue([]byte(precompute.AttributesKey(kv, nil)))
+	}
+	if err := s.pc.ObserveKeyed(s.pcfg.SeriesKeyFor(obs), obs); err != nil {
+		s.lastObserveErr = err
+		if !s.loggedObserveErr {
+			s.loggedObserveErr = true
+			s.logger.Warn("asap_edge: sketch observe dropped sample",
+				zap.String("metric", s.pcfg.MetricName), zap.Error(err))
+		}
+		return
+	}
+	s.lastObserveErr = nil
 }
 
 // flush force-rotates the window (Drain, regardless of wall-clock — the