From d24e1a22d0763796e4d10e4bbed8a1d958d8ce4b Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 26 May 2026 05:48:42 -0600 Subject: [PATCH] fix(asapedgeprocessor): feed CountMinSketch KindBytes, not KindFloat MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The fused sketch observe handed precompute.FloatValue to every family, but sketches.CMSObserver only accepts KindBytes (it hashes the encoded attribute key to count series cardinality, not the numeric value). So for a metric with family=countminsketch, ObserveKeyed -> CMSObserver.Observe rejected every sample with "expected KindBytes, got Float". The error was discarded with `_ =`, and the series was still admitted (empty sketch), so an envelope shell was emitted while no frequency was ever recorded. Fix: - For the CMS family, build the observation value as BytesValue(AttributesKey(labels, nil)) — the full attribute key, matching the standalone countminsketchprocessor shim. AggregateBy grouping is applied separately by SeriesKeyFor, so the inserted key stays the full attribute set. - Stop swallowing the ObserveKeyed error: retain it on the aggregator and log once per aggregator. This is what hid the bug. - Add TestCountMinSketchRecordsFrequency, which reconstructs the emitted CMS and asserts the inserted key's estimated frequency (~N). It fails without the fix ("expected KindBytes, got Float") and passes with it. CountSketch is unaffected (its observer takes KindFloat); only CMS required KindBytes. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../asapedgeprocessor/cms_observe_test.go | 79 +++++++++++++++++++ .../processor/asapedgeprocessor/processor.go | 2 +- .../asapedgeprocessor/sample_p_path_test.go | 3 +- .../asapedgeprocessor/warm_sketch.go | 52 ++++++++++-- 4 files changed, 127 insertions(+), 9 deletions(-) create mode 100644 opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/cms_observe_test.go diff --git a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/cms_observe_test.go b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/cms_observe_test.go new file mode 100644 index 00000000..c674e8ab --- /dev/null +++ b/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) + } +} diff --git a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/processor.go b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/processor.go index 00629208..2883a36d 100644 --- a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/processor.go +++ b/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 } } diff --git a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/sample_p_path_test.go b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/sample_p_path_test.go index f9d51270..56fe048b 100644 --- a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/sample_p_path_test.go +++ b/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) } diff --git a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/warm_sketch.go b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/warm_sketch.go index 587d48d3..d8d5f4bf 100644 --- a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/warm_sketch.go +++ b/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