feat: matrix aggregation (aggregate_by + label_matchers) for KLL, HLL, DDSketch processors

[zzylol]

Summary

• Adds aggregate_by (list of label keys) to KLL, HLL, and DDSketch processor configs: all data points sharing the same values for those labels are merged into one sketch per group; the output carries only those labels. Empty = existing per-series behavior (no change).
• Adds label_matchers (list of exact key=value filters): data points are included only if all matchers are satisfied. Empty = include all (no change).
• Adds cmd/kll/config-matrix.yaml example config documenting both SDK source modes.

Two source modes (same collector config handles both)

Mode	SDK sends	Collector accumulates
1 — sketch	Pre-aggregated KLLSketch / HLLSketch / DDSketch per series (fine granularity)	Merges incoming sketches cross-series by aggregate_by over the window
2 — raw	Raw Gauge float64 values	Inserts raw values from all matching series into one sketch per group

SDK configuration

Mode 1 — SDK pre-aggregates into sketches, collector merges cross-series

Configure a sketch aggregation view and a 1 s reader interval so each export carries one sketch per series:

provider := sdkmetric.NewMeterProvider(
    sdkmetric.WithReader(
        sdkmetric.NewPeriodicReader(exp, sdkmetric.WithInterval(time.Second)),
    ),
    sdkmetric.WithView(sdkmetric.NewView(
        sdkmetric.Instrument{Name: "benchmark.latency"},
        sdkmetric.Stream{Aggregation: sdkmetric.AggregationKLLSketch{K: 256}},
    )),
)
// Each series records with its own labels; collector aggregate_by merges them.
gauge.Record(ctx, value, metric.WithAttributes(
    attribute.String("region", "us-east"),
    attribute.String("env",    "prod"),
    attribute.String("server", "host-42"),  // dropped by aggregate_by
))

The SDK emits one KLLSketchDataPoint per series per reader interval, carrying the serialized sketch bytes + count/sum/min/max. The collector window accumulates these across series and time, then merges them into one sketch per aggregate_by group.

Mode 2 — SDK sends raw gauge samples, collector builds the sketch

Use the default LastValue aggregation (no view) and match the reader interval to the sample rate:

provider := sdkmetric.NewMeterProvider(
    sdkmetric.WithReader(
        sdkmetric.NewPeriodicReader(exp, sdkmetric.WithInterval(workerInterval)),
    ),
    // no WithView → gauge uses LastValue; collector receives raw float64 data points
)
gauge.Record(ctx, value, metric.WithAttributes(
    attribute.String("region", "us-east"),
    attribute.String("env",    "prod"),
    attribute.String("server", "host-42"),
))

Collector config

Output as quantile gauges (transmit_sketch: false, default)

The collector queries the merged sketch at the configured quantile points and emits one gauge metric per quantile per group:

processors:
  KLL:
    mode: window
    window_duration: 30s
    transmit_sketch: false
    quantiles: [0.5, 0.9, 0.99]
    label_matchers:
      - key: env
        value: prod
    aggregate_by: [region]

Output (Prometheus-scraped):

latency_p50{region="us-east"} 15.3
latency_p90{region="us-east"} 42.1
latency_p99{region="us-east"} 98.7

Output as sketch data points (transmit_sketch: true)

The collector forwards the merged sketch bytes as a KLLSketchDataPoint instead of emitting quantile gauges. Useful for multi-hop aggregation (e.g. sidecar → regional collector → global collector) where a downstream processor will do the final query:

processors:
  KLL:
    mode: window
    window_duration: 10s      # fine-grained window at edge
    transmit_sketch: true     # forward sketch bytes, not quantiles
    label_matchers:
      - key: env
        value: prod
    aggregate_by: [region]

Output: one KLLSketchDataPoint per aggregate_by group, with the serialized sketch embedded in the kll.sketch_payload attribute. A downstream KLL processor set to mode: window + transmit_sketch: false can receive and merge these further, then emit the final quantiles.

Test plan

• kllprocessor: TestAggregateByCollapsesSeries, TestLabelMatchersFilterGauge, TestAggregateByWithLabelMatchersWindowMode
• hllprocessor: TestHLLAggregateByCollapsesSeries, TestHLLLabelMatchersFilter
• ddsketchprocessor: TestDDAggregateByCollapsesSeries, TestDDLabelMatchersFilterGauge, TestDDAggregateByWindowModeDDSketchInput
• All existing tests continue to pass (no behavioral change when new fields are empty)

🤖 Generated with Claude Code