From b64be2ad425e53948106dad5efb87774c7f5d943 Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Wed, 15 Apr 2026 17:11:20 -0400 Subject: [PATCH] =?UTF-8?q?feat:=20per-processor=20encoding=20config=20?= =?UTF-8?q?=E2=80=94=20MSGPACK=20wire=20format=20option=20(3/4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds an `encoding: msgpack` config option to countminsketchprocessor, countsketchprocessor, and hllprocessor so operators can opt into the cross-language MessagePack wire format as an alternative to the default sketchlib proto path. This makes real MessagePack traffic flow on the modified-OTLP data plane — ASAPQuery-backend already accepts it (PRs #9, pmetric constants in #157), sketchlib-go already emits it (PR #51), but no processor selected it until now. ## Scope Three of four sketch processors are wired in this PR: * `countminsketchprocessor` — calls `ws.cms.SerializeMsgpack()` when `encoding: msgpack`, tags the data point with `CountMinSketchEncodingMsgpack`. * `countsketchprocessor` — calls `ws.cs.SerializeMsgpack()`, tags with `CountSketchEncodingMsgpack`. * `hllprocessor` — calls `series.sketch.SerializeMsgpack()` via a new `serializeHLLSketch(sketch, enc)` helper that returns `(payload, encodingTag, err)`, tags with `HLLSketchEncodingMsgpack`. `ddsketchprocessor` is intentionally **deferred** — it uses `github.com/DataDog/sketches-go`, not sketchlib-go, so it can't call `SerializeMsgpack` directly. Enabling MSGPACK for ddsketchprocessor requires a conversion shim that walks the DataDog sketch's buckets, builds a sketchlib-go `DDSketchState` proto (the same shape sketchlib-go [PR #52](https://github.com/ProjectASAP/sketchlib-go/pull/52) introduced via `NewFromStateProtoBytes`), and calls `SerializeMsgpack` on the reconstructed sketchlib-go sketch. Tracked as a separate follow-up because (a) the conversion code is ~100 lines of bucket flattening, (b) DataDog's proto doesn't carry Sum/Min/Max so msgpack emission from that source is lossy, and (c) the long-term fix is a full ddsketchprocessor migration to sketchlib-go internally, which is a much bigger refactor. ## Delta transmission stays proto-only All three processors keep delta transmission on the proto path when `delta_transmission: true` is set. Sketchlib-go has `SerializeMsgpack` for full sketch state but no matching delta wire format — tracked upstream until sketchlib-go grows an `apply_delta` API parallel to its proto one. The net effect: `encoding: msgpack` + `delta_transmission: true` emits proto deltas for sparse windows and never falls back to msgpack-full for those. ## Config shape Each processor's `Config` gains: ```yaml encoding: msgpack # "proto" (default) or "msgpack" ``` New `SketchEncoding` string type + `EncodingProto` / `EncodingMsgpack` constants per processor. `Validate()` rejects unknown values with a clear error rather than silently falling back. ## Validation Same pre-existing `go.opentelemetry.io/collector/processor/selfmonitor` module-resolution issue as the earlier typed-DP refactor PRs blocks local `go build`. gofmt is clean on all 6 modified files. All API methods used (`SerializeMsgpack` on each sketch type, `*SketchEncodingMsgpack` on the pmetric patch) are already available: * sketchlib-go `CountMinSketch.SerializeMsgpack` — PR #51 * sketchlib-go `CountSketch.SerializeMsgpack` — PR #51 * sketchlib-go `HyperLogLog.SerializeMsgpack` — PR #51 * pmetric `CountMinSketchEncodingMsgpack` — PR #157 * pmetric `CountSketchEncodingMsgpack` — PR #157 * pmetric `HLLSketchEncodingMsgpack` — PR #157 ## Follow-ups * `ddsketchprocessor` MSGPACK option via a DataDog→sketchlib-go conversion shim (or a full internal migration). Tracked. * Delta msgpack wire format (requires sketchlib-go upstream work). * Full ddsketchprocessor migration from DataDog/sketches-go to sketchlib-go DDSketch — bigger refactor, not in this PR's scope. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../countminsketchprocessor/config.go | 42 +++++++++++++++++++ .../countminsketchprocessor/processor.go | 34 ++++++++++++--- .../processor/countsketchprocessor/config.go | 35 ++++++++++++++++ .../countsketchprocessor/processor.go | 17 +++++++- .../processor/hllprocessor/config.go | 32 ++++++++++++++ .../processor/hllprocessor/processor.go | 29 +++++++++++-- 6 files changed, 177 insertions(+), 12 deletions(-) diff --git a/opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/config.go b/opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/config.go index f2f6c205..de883c16 100644 --- a/opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/config.go +++ b/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 } diff --git a/opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/processor.go b/opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/processor.go index 9d2e57e7..6015ae41 100644 --- a/opentelemetry-collector-contrib-patch/processor/countminsketchprocessor/processor.go +++ b/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) diff --git a/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/config.go b/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/config.go index 7b17ae9d..cb51c440 100644 --- a/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/config.go +++ b/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 } diff --git a/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/processor.go b/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/processor.go index e0a93384..5d698cdf 100644 --- a/opentelemetry-collector-contrib-patch/processor/countsketchprocessor/processor.go +++ b/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) diff --git a/opentelemetry-collector-contrib-patch/processor/hllprocessor/config.go b/opentelemetry-collector-contrib-patch/processor/hllprocessor/config.go index d0c4b891..065c250e 100644 --- a/opentelemetry-collector-contrib-patch/processor/hllprocessor/config.go +++ b/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 } diff --git a/opentelemetry-collector-contrib-patch/processor/hllprocessor/processor.go b/opentelemetry-collector-contrib-patch/processor/hllprocessor/processor.go index 11252375..1c199bc5 100644 --- a/opentelemetry-collector-contrib-patch/processor/hllprocessor/processor.go +++ b/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 + } +}