diff --git a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/cms_observe_test.go b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/cms_observe_test.go index d96c2765..be3a18d0 100644 --- a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/cms_observe_test.go +++ b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/cms_observe_test.go @@ -42,7 +42,7 @@ func TestCountMinSketchRecordsFrequency(t *testing.T) { 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)) + sa.observe(am, float64(i), base+uint64(i), false, 0, 0) } // Core invariant: the value kind matches the observer. Pre-fix this is diff --git a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/emit_heap_test.go b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/emit_heap_test.go index 31600418..c90e4e91 100644 --- a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/emit_heap_test.go +++ b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/emit_heap_test.go @@ -78,7 +78,7 @@ func TestEmitHeap_CountSketchHeapRoundTrips(t *testing.T) { tick := uint64(0) for _, f := range feed { for i := 0; i < f.n; i++ { - sa.observe(map[string]string{"endpoint": f.ep}, 1.0, base+tick) + sa.observe(map[string]string{"endpoint": f.ep}, 1.0, base+tick, false, 0, 0) tick++ } } @@ -206,7 +206,7 @@ func TestEmitHeap_DeltaFrameAfterFirstWindow(t *testing.T) { tick := start for ep, n := range counts { for i := 0; i < n; i++ { - sa.observe(map[string]string{"endpoint": ep}, 1.0, tick) + sa.observe(map[string]string{"endpoint": ep}, 1.0, tick, false, 0, 0) tick++ } } @@ -287,7 +287,7 @@ func TestEmitHeap_NonHeapCountSketchUnchanged(t *testing.T) { base := uint64(time.Unix(1700000000, 0).UnixMilli()) for i := 0; i < 20; i++ { - sa.observe(map[string]string{"endpoint": "/a"}, 1.0, base+uint64(i)) + sa.observe(map[string]string{"endpoint": "/a"}, 1.0, base+uint64(i), false, 0, 0) } if sa.lastObserveErr != nil { t.Fatalf("observe errored: %v", sa.lastObserveErr) diff --git a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/fixes_test.go b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/fixes_test.go index 893827cc..0afac791 100644 --- a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/fixes_test.go +++ b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/fixes_test.go @@ -89,7 +89,7 @@ func TestSketchMaxSeriesBounds(t *testing.T) { base := uint64(time.Unix(1700000000, 0).UnixMilli()) // 5 distinct series; only 2 fit. for i := 0; i < 5; i++ { - sa.observe(map[string]string{"series": string(rune('a' + i))}, float64(i), base) + sa.observe(map[string]string{"series": string(rune('a' + i))}, float64(i), base, false, 0, 0) } snap := sa.pc.Stats().Snapshot() if snap.ActiveSeries > 2 { @@ -127,7 +127,7 @@ func TestCountSketchCountsAttributeSet(t *testing.T) { am := map[string]string{"zone": "z0"} base := uint64(time.Unix(1700000000, 0).UnixMilli()) for i := 0; i < n; i++ { - sa.observe(am, float64(100+i), base+uint64(i)) // values vary; count must not + sa.observe(am, float64(100+i), base+uint64(i), false, 0, 0) // values vary; count must not } if sa.lastObserveErr != nil { t.Fatalf("CountSketch observe errored: %v", sa.lastObserveErr) @@ -190,7 +190,7 @@ func TestDeltaTransmissionEmitsDeltaEncoding(t *testing.T) { // Window 1 -> PROTO_FULL (first snapshot for the series). for i := 0; i < 10; i++ { - sa.observe(am, float64(i), base+uint64(i)) + sa.observe(am, float64(i), base+uint64(i), false, 0, 0) } envs1 := sa.pc.Drain() if !hasEncoding(envs1, precompute.EncodingProtoFull) { @@ -199,7 +199,7 @@ func TestDeltaTransmissionEmitsDeltaEncoding(t *testing.T) { // Window 2 -> PROTO_DELTA (against the cached window-1 snapshot). for i := 0; i < 10; i++ { - sa.observe(am, float64(i), base+1000+uint64(i)) + sa.observe(am, float64(i), base+1000+uint64(i), false, 0, 0) } envs2 := sa.pc.Drain() if !hasEncoding(envs2, precompute.EncodingProtoDelta) { @@ -322,7 +322,7 @@ func TestControlPlaneAppliesConfig(t *testing.T) { // The aggregator must still observe cleanly after the in-place swap (state // preserved, no rebuild). sa := p.shards[0].sketchAggs["lat"] - sa.observe(map[string]string{"zone": "z0"}, 1, uint64(time.Now().UnixMilli())) + sa.observe(map[string]string{"zone": "z0"}, 1, uint64(time.Now().UnixMilli()), false, 0, 0) if sa.lastObserveErr != nil { t.Fatalf("observe after UpdateConfig errored: %v", sa.lastObserveErr) } diff --git a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/ingest.go b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/ingest.go index 7a8e9fc1..536dda1f 100644 --- a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/ingest.go +++ b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/ingest.go @@ -33,6 +33,38 @@ func getAttrMap(src pcommon.Map) map[string]string { func putAttrMap(m map[string]string) { attrMapPool.Put(m) } +// rowSampledAdmittedRowsKey / rowSampledRowsKey / rowSampledSamplePKey mirror +// the reserved attribute keys the SDK's OTLP wire transform stamps on a +// RowSampledSketchDataPoint (see otlpmetricgrpc/otlpmetrichttp's +// internal/transform/metricdata.go). A data point carrying these attrs is +// not a normal aggregated sample — it is one individually admitted RAW +// occurrence, already row-sampled at Record() time by an OTel SDK running +// AggregationRowSampledSketch (NitroSketch-style skip sampling). They MUST +// be stripped before the remaining attributes are used as series identity +// (gorilla key / AggregateBy grouping / emitted labels) — they are +// wire-transport metadata, not series-identifying dimensions. +const ( + rowSampledAdmittedRowsKey = "__asap_row_sampled_admitted_rows" + rowSampledRowsKey = "__asap_row_sampled_rows" + rowSampledSamplePKey = "__asap_row_sampled_sample_p" +) + +// extractRowSampledMeta reports whether dp is a row-sampled raw occurrence +// (signaled by the presence of rowSampledAdmittedRowsKey, which the SDK +// always stamps alongside the other two reserved keys) and, if so, decodes +// the admission bitmask and sample probability the SDK computed. +func extractRowSampledMeta(attrs pcommon.Map) (rowSampled bool, admittedRows uint64, sampleP float64) { + v, ok := attrs.Get(rowSampledAdmittedRowsKey) + if !ok { + return false, 0, 0 + } + admittedRows = uint64(v.Int()) + if p, ok := attrs.Get(rowSampledSamplePKey); ok { + sampleP = p.Double() + } + return true, admittedRows, sampleP +} + func (p *asapEdgeProcessor) shardForKey(key string) int { if len(p.shards) <= 1 { return 0 @@ -119,7 +151,13 @@ func (p *asapEdgeProcessor) consumeMetric(m pmetric.Metric) { for i := 0; i < dps.Len(); i++ { dp := dps.At(i) + rowSampled, admittedRows, sampleP := extractRowSampledMeta(dp.Attributes()) am := getAttrMap(dp.Attributes()) // shared decode (once) + if rowSampled { + delete(am, rowSampledAdmittedRowsKey) + delete(am, rowSampledRowsKey) + delete(am, rowSampledSamplePKey) + } key := gorilla.SeriesKey(name, am, p.coldExtLabels) val := numberValue(dp) ts := dp.Timestamp().AsTime() @@ -128,7 +166,12 @@ func (p *asapEdgeProcessor) consumeMetric(m pmetric.Metric) { tsMs := uint64(ts.UnixMilli()) sh := p.shards[p.shardForKey(key)] sh.mu.Lock() - if sh.cold != nil && coldArchive { + // A row-sampled data point is one individually admitted raw + // occurrence, not a normal aggregated sample — it is never + // cold-archived (the cold path expects real aggregate samples, and + // the whole point of SDK-side row sampling is fewer, not more, raw + // points reaching the collector). + if sh.cold != nil && coldArchive && !rowSampled { // The fragment encoder rekeys internally by (metric, attrs); the // shared SeriesKey above is kept for shard selection only. _ = sh.cold.AddSample(gorilla.TSDBSample{ @@ -139,7 +182,7 @@ func (p *asapEdgeProcessor) consumeMetric(m pmetric.Metric) { }) } if sa := sh.sketchAggs[name]; sa != nil { - sa.observe(am, val, tsMs) + sa.observe(am, val, tsMs, rowSampled, admittedRows, sampleP) } sh.mu.Unlock() putAttrMap(am) diff --git a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/review_fixes_test.go b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/review_fixes_test.go index 96dea662..325a243f 100644 --- a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/review_fixes_test.go +++ b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/review_fixes_test.go @@ -129,10 +129,10 @@ func TestWarmWindowAdmitsProcessingDelayedSample(t *testing.T) { // start is well-defined, then feed an earlier-but-in-window sample. base := uint64(time.Unix(1700000040, 0).UnixMilli()) // 40s into a 60s-aligned window am := map[string]string{"zone": "z0"} - sa.observe(am, 1, base) + sa.observe(am, 1, base, false, 0, 0) // A sample 30s earlier in event-time: older than a 2s grace, but inside the // 60s window. Must be admitted (no observe drop). - sa.observe(am, 2, base-30_000) + sa.observe(am, 2, base-30_000, false, 0, 0) if sa.lastObserveErr != nil { t.Fatalf("in-window-but-delayed sample dropped: %v", sa.lastObserveErr) } @@ -157,7 +157,7 @@ func TestObserveScratchPreservesCounts(t *testing.T) { // Re-create the map each iter (as the real ingest path does) so the // scratch reuse is the only thing carrying state across samples. am := map[string]string{"zone": "z0", "method": "GET"} - sa.observe(am, float64(i), base+uint64(i)) + sa.observe(am, float64(i), base+uint64(i), false, 0, 0) } if sa.lastObserveErr != nil { t.Fatalf("observe errored: %v", sa.lastObserveErr) @@ -207,7 +207,7 @@ func TestObserveScratchReusesBuffers(t *testing.T) { base := uint64(time.Unix(1700000000, 0).UnixMilli()) // Warm up so the scratch buffers reach steady-state capacity. for i := 0; i < 100; i++ { - sa.observe(am, 1, base+uint64(i)) + sa.observe(am, 1, base+uint64(i), false, 0, 0) } // Irreducible baseline: everything observe() must do that is NOT in our @@ -227,7 +227,7 @@ func TestObserveScratchReusesBuffers(t *testing.T) { }) got := testing.AllocsPerRun(500, func() { - sa.observe(am, 1, base) + sa.observe(am, 1, base, false, 0, 0) }) t.Logf("observe allocs/op = %v, irreducible baseline = %v", got, baseline) diff --git a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/row_sampled_test.go b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/row_sampled_test.go new file mode 100644 index 00000000..44f4f37e --- /dev/null +++ b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/row_sampled_test.go @@ -0,0 +1,190 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +package asapedgeprocessor + +import ( + "context" + "testing" + "time" + + precompute "github.com/ProjectASAP/asap-precompute-go" + "github.com/ProjectASAP/asap-precompute-go/sketches" + "go.opentelemetry.io/collector/component" + "go.opentelemetry.io/collector/pdata/pcommon" + "go.opentelemetry.io/collector/pdata/pmetric" + "go.opentelemetry.io/collector/processor" + "go.uber.org/zap" +) + +// TestSketchAggregatorObserve_RowSampled_RescalesByP drives observe() with +// rowSampled=true directly against a CMS aggregator and checks the emitted +// envelope's estimate reflects the SDK's 1/SampleP correction — proving +// ApplyAdmittedOccurrence, not InsertHash, was used. +func TestSketchAggregatorObserve_RowSampled_RescalesByP(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], sketchOpts{window: time.Hour}, zap.NewNop()) + if !ok { + t.Fatal("newSketchAggregator(CountMinSketch) returned ok=false") + } + + const n, p = 1000, 0.5 + am := map[string]string{"zone": "z0"} + base := uint64(time.Unix(1700000000, 0).UnixMilli()) + allRows := uint64(0b11111) // csmDims defaults to 5 rows + for i := 0; i < n; i++ { + sa.observe(am, 1, base+uint64(i), true, allRows, p) + } + if sa.lastObserveErr != nil { + t.Fatalf("row-sampled observe errored: %v", sa.lastObserveErr) + } + + 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)) + want := float64(n) / p + got := rebuilt.EstimateCount(key) + if rel := (got - want) / want; rel < -0.1 || rel > 0.1 { + t.Fatalf("EstimateCount = %v, want ~%v (n/p rescale)", got, want) + } +} + +// TestSketchAggregatorObserve_RowSampled_UnsupportedFamilyDrops verifies a +// row-sampled observation against a family with no *AtRows sketchlib +// primitive (DDSketch, obsKindFloat) is dropped rather than silently +// misapplied — this can only happen if the SDK's AggregationRouter and this +// collector's AggID disagreed about which family a PolicyFingerprint +// targets, and that must fail loud (a counted drop), not corrupt state. +func TestSketchAggregatorObserve_RowSampled_UnsupportedFamilyDrops(t *testing.T) { + cfg := &Config{ + ShardCount: 1, + WindowDuration: time.Hour, + Metrics: []MetricFamily{{Metric: "lat", Family: FamilyDDSketch}}, + Cold: ColdConfig{Enabled: false}, + } + if err := cfg.Validate(); err != nil { + t.Fatal(err) + } + sa, ok := newSketchAggregator("lat", &cfg.Metrics[0], sketchOpts{window: time.Hour}, zap.NewNop()) + if !ok { + t.Fatal("newSketchAggregator(DDSketch) returned ok=false") + } + + before := sa.droppedSamples.Load() + sa.observe(map[string]string{"zone": "z0"}, 1, uint64(time.Now().UnixMilli()), true, 0b1, 0.5) + if got := sa.droppedSamples.Load(); got != before+1 { + t.Fatalf("droppedSamples = %d, want %d (row-sampled obs against obsKindFloat must drop)", got, before+1) + } +} + +// TestConsumeMetrics_RowSampled_StripsReservedAttrsAndRescales drives the +// real OTLP ingest path (ConsumeMetrics): a Gauge data point carrying the 3 +// reserved row-sampled attrs must be (a) routed through +// ApplyAdmittedOccurrence with the correct 1/SampleP rescale and (b) have the +// reserved keys stripped before they ever reach series identity — they must +// not appear in any emitted output label. +func TestConsumeMetrics_RowSampled_StripsReservedAttrsAndRescales(t *testing.T) { + cfg := &Config{ + ShardCount: 1, + WindowDuration: time.Hour, + Metrics: []MetricFamily{{Metric: "conns", Family: FamilyCountMinSketch}}, + Cold: ColdConfig{Enabled: false}, + } + if err := cfg.Validate(); err != nil { + t.Fatal(err) + } + set := processor.Settings{TelemetrySettings: component.TelemetrySettings{Logger: zap.NewNop()}} + sink := &capMetrics{} + p, err := newProcessor(cfg, set, sink) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = p.Shutdown(context.Background()) }) + + const n, sampleP = 200, 0.4 + allRows := uint64(0b11111) + base := time.Unix(1700000000, 0) + md := pmetric.NewMetrics() + m := md.ResourceMetrics().AppendEmpty().ScopeMetrics().AppendEmpty().Metrics().AppendEmpty() + m.SetName("conns") + g := m.SetEmptyGauge() + for i := 0; i < n; i++ { + dp := g.DataPoints().AppendEmpty() + dp.Attributes().PutStr("zone", "z0") + dp.Attributes().PutInt(rowSampledAdmittedRowsKey, int64(allRows)) + dp.Attributes().PutInt(rowSampledRowsKey, 5) + dp.Attributes().PutDouble(rowSampledSamplePKey, sampleP) + dp.SetDoubleValue(1) + dp.SetTimestamp(pcommon.NewTimestampFromTime(base.Add(time.Duration(i) * time.Millisecond))) + } + if err := p.ConsumeMetrics(context.Background(), md); err != nil { + t.Fatalf("ConsumeMetrics: %v", err) + } + + sa := p.shards[0].sketchAggs["conns"] + if sa == nil { + t.Fatal("no sketch aggregator for \"conns\"") + } + if sa.lastObserveErr != nil { + t.Fatalf("row-sampled ingest errored: %v", sa.lastObserveErr) + } + + 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") + } + + // The series key must be built from {zone=z0} ONLY — the 3 reserved keys + // must never reach AttributesKey/series identity. + key := []byte(precompute.AttributesKey([]precompute.KeyValue{{Key: "zone", Value: "z0"}}, nil)) + want := float64(n) / sampleP + got := rebuilt.EstimateCount(key) + if rel := (got - want) / want; rel < -0.1 || rel > 0.1 { + t.Fatalf("EstimateCount = %v, want ~%v (n/sampleP rescale)", got, want) + } + + // A key built with any reserved attribute folded in must estimate ~0 — + // proof the reserved attrs never leaked into the series-identifying set. + poisoned := []byte(precompute.AttributesKey([]precompute.KeyValue{ + {Key: "zone", Value: "z0"}, + {Key: rowSampledAdmittedRowsKey, Value: "31"}, + }, nil)) + if got := rebuilt.EstimateCount(poisoned); got > float64(n)/sampleP*0.1 { + t.Fatalf("EstimateCount(poisoned key with reserved attr) = %v, want ~0 (reserved attrs leaked into series identity)", got) + } +} diff --git a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/warm_sketch.go b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/warm_sketch.go index 3a681a17..cfa0dd9a 100644 --- a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/warm_sketch.go +++ b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/warm_sketch.go @@ -838,7 +838,16 @@ func (s *sketchAggregator) attrKeyBytes(kv []precompute.KeyValue) []byte { // still allocated per sample — the keyed precompute entry point takes a // string and there is no exported zero-alloc keyed path — so that one // allocation remains. -func (s *sketchAggregator) observe(am map[string]string, val float64, tsMs uint64) { +// +// rowSampled/admittedRows/sampleP carry an SDK-side pre-decided row-admission +// bitmask (NitroSketch-style skip sampling, see AggregationRowSampledSketch +// in the SDK). When rowSampled is true, the ObservationValue built below is +// tagged so the CMS/CountSketch observer applies the bitmask verbatim via +// Sketch.ApplyAdmittedOccurrence instead of the plain insert path — see the +// precompute.ObservationValue.RowSampled doc for why this must not be +// re-derived collector-side. Every non-row-sampled caller passes +// rowSampled=false (admittedRows/sampleP ignored). +func (s *sketchAggregator) observe(am map[string]string, val float64, tsMs uint64, rowSampled bool, admittedRows uint64, sampleP float64) { // For the HLL / CMS item_label paths the item_label attribute is the sketch // subject (its value is hashed below), so it must NOT appear in the series // key or the emitted labels — project it out of the observation labels here. @@ -927,6 +936,26 @@ func (s *sketchAggregator) observe(am map[string]string, val float64, tsMs uint6 s.attrKeyScratch = append(s.attrKeyScratch[:0], am[s.itemLabel]...) obs.Value = precompute.BytesValue(s.attrKeyScratch) } + if rowSampled { + switch s.obsKind { + case obsKindBytesHash, obsKindKeyedFreq, obsKindKeyedItem: + obs.Value.RowSampled = true + obs.Value.AdmittedRows = admittedRows + obs.Value.SampleP = sampleP + default: + // obsKindFloat/obsKindItemHLL/obsKindItemCMS have no *AtRows + // sketchlib primitive (DDSketch/KLL/HLL aren't row-replicated + // matrices) — this can only happen if the SDK's AggregationRouter + // and this collector's AggID disagreed about which family a + // PolicyFingerprint targets. Drop rather than silently misapply + // an unrelated observer. + s.droppedSamples.Add(1) + if s.procDropCount != nil { + s.procDropCount.Add(1) + } + return + } + } if err := s.pc.ObserveKeyed(s.pcfg.SeriesKeyFor(obs), obs); err != nil { s.lastObserveErr = err // Always count the drop so it stays observable; the log is latched to diff --git a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/weight_mode_test.go b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/weight_mode_test.go index cff69e3f..19512779 100644 --- a/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/weight_mode_test.go +++ b/opentelemetry-collector-contrib-patch/processor/asapedgeprocessor/weight_mode_test.go @@ -41,7 +41,7 @@ func feedAdversarial(t *testing.T, sa *sketchAggregator) []asapmsgpack.HeapItem tick := uint64(0) for _, f := range adversarialFeed { for i := 0; i < f.n; i++ { - sa.observe(map[string]string{"endpoint": f.ep}, f.value, base+tick) + sa.observe(map[string]string{"endpoint": f.ep}, f.value, base+tick, false, 0, 0) tick++ } }