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