feat(countminsketchprocessor): emit typed CountMinSketchDataPoint

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

@@ -520,18 +520,47 @@ func (p *windowedCountMinSketchProcessor) buildWindowMetricsAndReset() pmetric.M
 		m.SetName(p.cfg.MetricName)
 		m.SetUnit("1")
 
-		gauge := m.SetEmptyGauge()
-		dp := gauge.DataPoints().AppendEmpty()
-		dp.SetTimestamp(now)
-
-		outputAttrs.CopyTo(dp.Attributes())
-		dp.Attributes().PutInt("rows", int64(rows))
-		dp.Attributes().PutInt("cols", int64(cols))
-		dp.Attributes().PutInt("sample_count", int64(sampleCount))
 		if p.cfg.TransmitSketch {
-			dp.Attributes().PutEmptyBytes("sketch_payload").FromRaw(payload)
-			dp.Attributes().PutStr("encoding", encoding)
+			// Typed CountMinSketchDataPoint emission — what
+			// ASAPQuery-backend's modified-OTLP sketch router
+			// consumes as `Metric.data = CountMinSketch{...}`.
+			// Before this change the processor emitted a Gauge
+			// with the sketch payload stuffed into a
+			// `sketch_payload` byte attribute, which the backend
+			// router never recognized as a sketch variant.
+			cmsMetric := m.SetEmptyCountMinSketch()
+			cmsMetric.SetAggregationTemporality(pmetric.AggregationTemporalityDelta)
+			dp := cmsMetric.DataPoints().AppendEmpty()
+			dp.SetTimestamp(now)
+			outputAttrs.CopyTo(dp.Attributes())
+			dp.SetSampleCount(uint64(sampleCount))
+			dp.SetRows(int32(rows))
+			dp.SetCols(int32(cols))
+			dp.SetSketch(payload)
+			// Map the internal encoding string onto the proto
+			// enum the backend expects. The encoding string only
+			// branches on delta vs full when delta transmission
+			// is on; otherwise it's always proto_full.
+			switch encoding {
+			case "proto_delta":
+				dp.SetEncoding(pmetric.CountMinSketchEncodingDelta)
+			default:
+				// "proto_full" and any unexpected fallback.
+				dp.SetEncoding(pmetric.CountMinSketchEncodingProto)
+			}
 		} else {
+			// Non-transmit mode: caller only wants the
+			// per-window sample count for monitoring, not the
+			// sketch bytes. Keep the legacy Gauge emission so
+			// existing dashboards that read `countmin` as a
+			// scalar series continue to work.
+			gauge := m.SetEmptyGauge()
+			dp := gauge.DataPoints().AppendEmpty()
+			dp.SetTimestamp(now)
+			outputAttrs.CopyTo(dp.Attributes())
+			dp.Attributes().PutInt("rows", int64(rows))
+			dp.Attributes().PutInt("cols", int64(cols))
+			dp.Attributes().PutInt("sample_count", int64(sampleCount))
 			dp.SetDoubleValue(float64(sampleCount))
 		}
 	}

opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/processor_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.uber.org/zap"
 )
@@ -180,7 +181,7 @@ func TestProcessor_TumblingWindow_Correctness(t *testing.T) {
 	// ==========================================
 	// 3. Verify Binary Payload (Gob Decode)
 	// ==========================================
-	payloadVal, ok := dps2[0].Attributes().Get("cms.sketch_payload")
+	payloadVal, ok := dps2[0].Attributes().Get("sketch_payload")
 	require.True(t, ok, "Sketch payload must exist in attributes")
 
 	rawBytes := payloadVal.Bytes().AsRaw()
@@ -210,18 +211,80 @@ func generateMetrics(serviceName string, count int) pmetric.Metrics {
 	return md
 }
 
-func getAllDataPoints(md pmetric.Metrics) []pmetric.NumberDataPoint {
-	var dps []pmetric.NumberDataPoint
+// cmsTestDataPoint is a test-only adapter that flattens either a
+// `CountMinSketchDataPoint` (the typed emission path) or a `NumberDataPoint`
+// (the legacy Gauge path kept for non-TransmitSketch mode) into a single
+// `pcommon.Map` so existing test assertions that read `sketch_payload`,
+// `encoding`, `sample_count`, `rows`, `cols` via `Attributes().Get(...)`
+// continue to compile without site-by-site rewrites.
+//
+// `doubleValue` is only populated for the non-TransmitSketch Gauge path —
+// `TestBatchModeQueryMetricsWhenTransmitSketchDisabled` is the single
+// test that reads it. For the typed-DP path it stays 0 and is unused.
+type cmsTestDataPoint struct {
+	attributes  pcommon.Map
+	doubleValue float64
+}
+
+func (d cmsTestDataPoint) Attributes() pcommon.Map { return d.attributes }
+func (d cmsTestDataPoint) DoubleValue() float64    { return d.doubleValue }
+
+// encodingToLegacyString mirrors the strings the processor used to
+// write into the `encoding` attribute, so tests that compare against
+// "proto_full" / "proto_delta" continue to work.
+func encodingToLegacyString(enc pmetric.CountMinSketchEncoding) string {
+	switch enc {
+	case pmetric.CountMinSketchEncodingProto:
+		return "proto_full"
+	case pmetric.CountMinSketchEncodingDelta:
+		return "proto_delta"
+	}
+	return "unknown"
+}
+
+// getAllDataPoints walks the output metrics and returns a flat slice of
+// test adapters, one per sketch data point (typed or legacy Gauge). When
+// the input metric is a typed `CountMinSketch`, its fields (sketch bytes,
+// encoding enum, sample_count, rows, cols) are synthesized into the
+// adapter's attribute map under their legacy string keys so the rest of
+// the test file can keep reading them with `Attributes().Get(...)`.
+func getAllDataPoints(md pmetric.Metrics) []cmsTestDataPoint {
+	var dps []cmsTestDataPoint
 	rms := md.ResourceMetrics()
 	for i := 0; i < rms.Len(); i++ {
 		sms := rms.At(i).ScopeMetrics()
 		for j := 0; j < sms.Len(); j++ {
 			ms := sms.At(j).Metrics()
 			for k := 0; k < ms.Len(); k++ {
 				m := ms.At(k)
-				pts := m.Gauge().DataPoints()
-				for l := 0; l < pts.Len(); l++ {
-					dps = append(dps, pts.At(l))
+				switch m.Type() {
+				case pmetric.MetricTypeCountMinSketch:
+					pts := m.CountMinSketch().DataPoints()
+					for l := 0; l < pts.Len(); l++ {
+						dp := pts.At(l)
+						attrs := pcommon.NewMap()
+						dp.Attributes().CopyTo(attrs)
+						// Inject the typed-DP fields under their
+						// legacy attribute names so existing tests
+						// keep working.
+						attrs.PutEmptyBytes("sketch_payload").FromRaw(dp.Sketch())
+						attrs.PutStr("encoding", encodingToLegacyString(dp.Encoding()))
+						attrs.PutInt("sample_count", int64(dp.SampleCount()))
+						attrs.PutInt("rows", int64(dp.Rows()))
+						attrs.PutInt("cols", int64(dp.Cols()))
+						dps = append(dps, cmsTestDataPoint{attributes: attrs})
+					}
+				case pmetric.MetricTypeGauge:
+					pts := m.Gauge().DataPoints()
+					for l := 0; l < pts.Len(); l++ {
+						dp := pts.At(l)
+						attrs := pcommon.NewMap()
+						dp.Attributes().CopyTo(attrs)
+						dps = append(dps, cmsTestDataPoint{
+							attributes:  attrs,
+							doubleValue: dp.DoubleValue(),
+						})
+					}
 				}
 			}
 		}
@@ -301,7 +364,7 @@ func TestBatchModeQueryMetricsWhenTransmitSketchDisabled(t *testing.T) {
 	dps := getAllDataPoints(out)
 	require.Len(t, dps, 1)
 	assert.Equal(t, 3.0, dps[0].DoubleValue())
-	_, hasPayload := dps[0].Attributes().Get("cms.sketch_payload")
+	_, hasPayload := dps[0].Attributes().Get("sketch_payload")
 	assert.False(t, hasPayload)
 }