feat(collector): decode SDK row-sampled points, route to ApplyAdmittedOccurrence

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

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)

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

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)

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)
 

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