feat: per-processor encoding config — MSGPACK wire format option (3/4)

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

@@ -21,6 +21,29 @@ const (
 	ModeWindow InputMode = "window"
 )
 
+// SketchEncoding selects the wire format for the serialized sketch bytes
+// carried in `CountMinSketchDataPoint.Sketch`. The corresponding
+// `CountMinSketchDataPoint.Encoding` enum value is written alongside
+// so the downstream consumer knows how to decode.
+//
+//   - "proto" (default) — sketchlib-go `SerializeProtoBytesFO`
+//     (sketchlib `CountMinState` proto). Tag = `CountMinSketchEncodingProto`
+//     or `CountMinSketchEncodingDelta` depending on DeltaTransmission.
+//   - "msgpack"           — sketchlib-go `SerializeMsgpack` (the
+//     cross-language wire format consumed by ASAPQuery-backend's
+//     `CountMinSketchAccumulator::from_msgpack_bytes`). Tag =
+//     `CountMinSketchEncodingMsgpack`. Delta transmission is currently
+//     proto-only, so when `encoding = msgpack` and
+//     `delta_transmission = true`, the processor still falls back to
+//     proto for per-window deltas until sketchlib-go grows a msgpack
+//     delta path.
+type SketchEncoding string
+
+const (
+	EncodingProto   SketchEncoding = "proto"
+	EncodingMsgpack SketchEncoding = "msgpack"
+)
+
 // LabelMatcher specifies an exact label key=value filter.
 // A data point matches only if the named label exists and its string value equals Value.
 type LabelMatcher struct {
@@ -45,6 +68,12 @@ type Config struct {
 	TransmitSketch       bool `mapstructure:"transmit_sketch"`
 	DropOriginal         bool `mapstructure:"drop_original"`
 
+	// Encoding controls the wire format of the sketch bytes written to
+	// `CountMinSketchDataPoint.Sketch` when `TransmitSketch = true`.
+	// See the [`SketchEncoding`] doc for the supported values. Defaults
+	// to "proto" for backwards compatibility.
+	Encoding SketchEncoding `mapstructure:"encoding"`
+
 	// WindowDuration is the time window to accumulate data before emitting a sketch (window mode only).
 	WindowDuration time.Duration `mapstructure:"window_duration"`
 
@@ -108,5 +137,18 @@ func (c *Config) Validate() error {
 		}
 	}
 
+	// Default Encoding to proto when unset. Accept both supported
+	// values; anything else is a config error rather than a silent
+	// fallback.
+	switch c.Encoding {
+	case "":
+		c.Encoding = EncodingProto
+	case EncodingProto, EncodingMsgpack:
+	default:
+		return fmt.Errorf(
+			"invalid encoding %q, must be %q or %q",
+			c.Encoding, EncodingProto, EncodingMsgpack)
+	}
+
 	return nil
 }

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

@@ -480,6 +480,13 @@ func (p *windowedCountMinSketchProcessor) buildWindowMetricsAndReset() pmetric.M
 
 		if p.cfg.TransmitSketch && p.cfg.DeltaTransmission {
 			// Delta path: compute sparse diff against the last snapshot.
+			// Delta transmission is proto-only today; the msgpack wire
+			// format doesn't yet carry deltas (tracked as a follow-up
+			// once sketchlib-go grows `apply_delta`). When
+			// `encoding: msgpack` is set AND delta is on, the
+			// processor falls through to proto delta for per-window
+			// diffs and still tags the encoding as delta so the
+			// consumer knows.
 			p.snapshotsMu.Lock()
 			snap, hasSnap := p.snapshots[aggregationKey]
 			p.snapshotsMu.Unlock()
@@ -504,8 +511,19 @@ func (p *windowedCountMinSketchProcessor) buildWindowMetricsAndReset() pmetric.M
 			p.snapshots[aggregationKey] = newSnap
 			p.snapshotsMu.Unlock()
 		} else if p.cfg.TransmitSketch {
-			payload, err = serializeCMS(ws.cms)
-			encoding = "proto_full"
+			// Non-delta path: emit a full sketch payload in the
+			// configured encoding. sketchlib-go exposes a parallel
+			// `SerializeMsgpack` that matches the cross-language wire
+			// format ASAPQuery-backend's
+			// `CountMinSketchAccumulator::from_msgpack_bytes` consumes.
+			switch p.cfg.Encoding {
+			case EncodingMsgpack:
+				payload, err = ws.cms.SerializeMsgpack()
+				encoding = "msgpack_full"
+			default:
+				payload, err = serializeCMS(ws.cms)
+				encoding = "proto_full"
+			}
 		}
 
 		ws.mu.Unlock()
@@ -537,13 +555,17 @@ func (p *windowedCountMinSketchProcessor) buildWindowMetricsAndReset() pmetric.M
 			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.
+			// Map the internal encoding string onto the proto enum
+			// the backend expects. `proto_delta` is the sparse-cell
+			// diff format (delta transmission); `msgpack_full` is
+			// the cross-language sketchlib-go msgpack wire format;
+			// everything else is the default sketchlib `CountMinState`
+			// proto.
 			switch encoding {
 			case "proto_delta":
 				dp.SetEncoding(pmetric.CountMinSketchEncodingDelta)
+			case "msgpack_full":
+				dp.SetEncoding(pmetric.CountMinSketchEncodingMsgpack)
 			default:
 				// "proto_full" and any unexpected fallback.
 				dp.SetEncoding(pmetric.CountMinSketchEncodingProto)

opentelemetry-collector-contrib-patch/processor/countsketchprocessor/config.go

@@ -73,8 +73,33 @@ type Config struct {
 	// DeltaThreshold is the minimum absolute cell change required to include a
 	// cell in the delta payload. Defaults to 1.0 when DeltaTransmission=true.
 	DeltaThreshold float64 `mapstructure:"delta_threshold"`
+
+	// Encoding selects the wire format for the `CountSketchDataPoint.Sketch`
+	// bytes. See `SketchEncoding` for supported values. Defaults to "proto".
+	Encoding SketchEncoding `mapstructure:"encoding"`
 }
 
+// SketchEncoding selects the wire format for the serialized sketch bytes
+// carried in `CountSketchDataPoint.Sketch`. The corresponding
+// `CountSketchDataPoint.Encoding` enum value is written alongside so the
+// downstream consumer knows how to decode.
+//
+//   - "proto" (default) — sketchlib-go `SerializeProtoBytes`
+//     (sketchlib `CountSketchState` proto). Tag =
+//     `CountSketchEncodingProto` or `CountSketchEncodingDelta` depending
+//     on DeltaTransmission.
+//   - "msgpack"           — sketchlib-go `SerializeMsgpack`. Tag =
+//     `CountSketchEncodingMsgpack`. Delta transmission is currently
+//     proto-only; `encoding = msgpack` + `delta_transmission = true`
+//     still falls back to proto deltas per window until sketchlib-go
+//     grows a msgpack delta path.
+type SketchEncoding string
+
+const (
+	EncodingProto   SketchEncoding = "proto"
+	EncodingMsgpack SketchEncoding = "msgpack"
+)
+
 var _ component.Config = (*Config)(nil)
 
 func (c *Config) Validate() error {
@@ -115,5 +140,15 @@ func (c *Config) Validate() error {
 		}
 	}
 
+	switch c.Encoding {
+	case "":
+		c.Encoding = EncodingProto
+	case EncodingProto, EncodingMsgpack:
+	default:
+		return fmt.Errorf(
+			"invalid encoding %q, must be %q or %q",
+			c.Encoding, EncodingProto, EncodingMsgpack)
+	}
+
 	return nil
 }

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

@@ -361,6 +361,9 @@ func (p *countSketchProcessor) buildWindowMetricsAndReset() pmetric.Metrics {
 
 		if p.config.TransmitSketch && ws.cs != nil {
 			if p.config.DeltaTransmission {
+				// Delta transmission is proto-only — msgpack delta
+				// is tracked as a follow-up once sketchlib-go
+				// grows `apply_delta` semantics.
 				p.snapshotsMu.Lock()
 				snap, hasSnap := p.snapshots[partitionKey]
 				p.snapshotsMu.Unlock()
@@ -383,8 +386,16 @@ func (p *countSketchProcessor) buildWindowMetricsAndReset() pmetric.Metrics {
 				p.snapshots[partitionKey] = newSnap
 				p.snapshotsMu.Unlock()
 			} else {
-				payload, serErr = serializeCountSketch(ws.cs)
-				encoding = "proto_full"
+				// Non-delta path — choose between proto and msgpack
+				// wire formats based on the config's Encoding field.
+				switch p.config.Encoding {
+				case EncodingMsgpack:
+					payload, serErr = ws.cs.SerializeMsgpack()
+					encoding = "msgpack_full"
+				default:
+					payload, serErr = serializeCountSketch(ws.cs)
+					encoding = "proto_full"
+				}
 			}
 		}
 
@@ -428,6 +439,8 @@ func (p *countSketchProcessor) buildWindowMetricsAndReset() pmetric.Metrics {
 			switch encoding {
 			case "proto_delta":
 				dp.SetEncoding(pmetric.CountSketchEncodingDelta)
+			case "msgpack_full":
+				dp.SetEncoding(pmetric.CountSketchEncodingMsgpack)
 			default:
 				// "proto_full" and any unexpected fallback.
 				dp.SetEncoding(pmetric.CountSketchEncodingProto)

opentelemetry-collector-contrib-patch/processor/hllprocessor/config.go

@@ -53,8 +53,29 @@ type Config struct {
 	// increased since the last snapshot are transmitted (max semantics).
 	// Requires TransmitSketch=true; has no effect in batch mode.
 	DeltaTransmission bool `mapstructure:"delta_transmission"`
+
+	// Encoding selects the wire format for the `HLLSketchDataPoint.Sketch`
+	// bytes. See `SketchEncoding` for supported values. Defaults to "proto".
+	Encoding SketchEncoding `mapstructure:"encoding"`
 }
 
+// SketchEncoding selects the wire format for the serialized HLL bytes
+// carried in `HLLSketchDataPoint.Sketch`.
+//
+//   - "proto" (default) — sketchlib-go `SerializeProtoBytes` →
+//     sketchlib `HyperLogLogState` proto. Tag =
+//     `HLLSketchEncodingProto` / `HLLSketchEncodingDelta`.
+//   - "msgpack"           — sketchlib-go `SerializeMsgpack`. Tag =
+//     `HLLSketchEncodingMsgpack`. Delta transmission is proto-only
+//     today; `encoding = msgpack` + `delta_transmission = true`
+//     still falls back to proto deltas per window.
+type SketchEncoding string
+
+const (
+	EncodingProto   SketchEncoding = "proto"
+	EncodingMsgpack SketchEncoding = "msgpack"
+)
+
 var _ component.Config = (*Config)(nil)
 
 func (c *Config) Validate() error {
@@ -70,5 +91,16 @@ func (c *Config) Validate() error {
 	}
 	// Sort AggregateBy so seriesKey always produces a consistent ordering.
 	sort.Strings(c.AggregateBy)
+
+	switch c.Encoding {
+	case "":
+		c.Encoding = EncodingProto
+	case EncodingProto, EncodingMsgpack:
+	default:
+		return fmt.Errorf(
+			"invalid encoding %q, must be %q or %q",
+			c.Encoding, EncodingProto, EncodingMsgpack)
+	}
+
 	return nil
 }

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

@@ -302,7 +302,7 @@ func (p *hllProcessor) processBatch(md pmetric.Metrics) error {
 				m.SetEmptyHLLSketch().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
 				hllMetrics[metricName] = m
 			}
-			payload, err := bs.sketch.SerializeProtoBytes()
+			payload, encodingTag, err := serializeHLLSketch(bs.sketch, p.cfg.Encoding)
 			if err != nil {
 				if p.logger != nil {
 					p.logger.Error("hllprocessor: failed to serialize sketch", zap.Error(err))
@@ -315,7 +315,7 @@ func (p *hllProcessor) processBatch(md pmetric.Metrics) error {
 			dp.SetCount(bs.count)
 			dp.SetCardinality(uint64(bs.sketch.EstimateCardinality()))
 			dp.SetSketch(payload)
-			dp.SetEncoding(pmetric.HLLSketchEncodingProto)
+			dp.SetEncoding(encodingTag)
 			dp.SetPrecision(uint32(hll.HLLPrecision))
 		}
 	} else {
@@ -614,7 +614,7 @@ func (p *hllProcessor) flushWindow(ctx context.Context) error {
 							p.snapshots[snapKey] = newSnap
 							p.snapshotsMu.Unlock()
 						} else {
-							payload, err := series.sketch.SerializeProtoBytes()
+							payload, encodingTag, err := serializeHLLSketch(series.sketch, p.cfg.Encoding)
 							if err != nil {
 								if p.logger != nil {
 									p.logger.Error("hllprocessor: failed to serialize sketch", zap.Error(err))
@@ -626,7 +626,7 @@ func (p *hllProcessor) flushWindow(ctx context.Context) error {
 								dp.SetCount(0)
 								dp.SetCardinality(uint64(series.sketch.EstimateCardinality()))
 								dp.SetSketch(payload)
-								dp.SetEncoding(pmetric.HLLSketchEncodingProto)
+								dp.SetEncoding(encodingTag)
 								dp.SetPrecision(uint32(hll.HLLPrecision))
 							}
 						}
@@ -755,3 +755,24 @@ func (p *hllProcessor) cardinalityMetricName(base string) string {
 	}
 	return base + "_hll_cardinality"
 }
+
+// serializeHLLSketch serializes an HLL sketch in the configured wire
+// format and returns the bytes along with the matching pmetric
+// encoding enum to write into `HLLSketchDataPoint.Encoding`. Bridges
+// the cross-language msgpack wire format (sketchlib-go `SerializeMsgpack`,
+// consumed by ASAPQuery-backend's
+// `HllSketchAccumulator::from_msgpack_bytes`) alongside the existing
+// sketchlib proto path.
+func serializeHLLSketch(
+	sketch *hll.HyperLogLog,
+	enc SketchEncoding,
+) ([]byte, pmetric.HLLSketchEncoding, error) {
+	switch enc {
+	case EncodingMsgpack:
+		payload, err := sketch.SerializeMsgpack()
+		return payload, pmetric.HLLSketchEncodingMsgpack, err
+	default:
+		payload, err := sketch.SerializeProtoBytes()
+		return payload, pmetric.HLLSketchEncodingProto, err
+	}
+}