diff --git a/CHANGELOG.md b/CHANGELOG.md index f9d4d34555b..735e2cf40c9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm - Add `error.type` attribute to `http.client.request.duration` for transport failures in `otelhttp`. (#8801) - Add examples for prometheus compatibility document. (#8716) +- Add support for `cardinality_limits` in `PeriodicMetricReader` in `otelconf`. (#8885) - Add `Resource` method to `SDK` in `go.opentelemetry.io/contrib/otelconf/x` to expose the resolved SDK resource from declarative configuration. (#8912) - Add `go.opentelemetry.io/contrib/detectors/hetzner` — a new resource detector for Hetzner Cloud servers, ported from `processor/resourcedetectionprocessor/internal/hetzner` in `opentelemetry-collector-contrib`. Detects `cloud.provider`, `cloud.platform`, `cloud.region`, `cloud.availability_zone`, `host.id`, and `host.name`. (#8962) diff --git a/otelconf/metric.go b/otelconf/metric.go index 4a9a845e556..f13e715d8f5 100644 --- a/otelconf/metric.go +++ b/otelconf/metric.go @@ -78,6 +78,10 @@ func metricReader(ctx context.Context, r MetricReader) (sdkmetric.Reader, error) if r.Periodic.Timeout != nil { opts = append(opts, sdkmetric.WithTimeout(time.Duration(*r.Periodic.Timeout)*time.Millisecond)) } + + if r.Periodic.CardinalityLimits != nil { + opts = append(opts, sdkmetric.WithCardinalityLimitSelector(cardinalityLimitSelector(r.Periodic.CardinalityLimits))) + } return periodicExporter(ctx, r.Periodic.Exporter, opts...) } @@ -91,6 +95,49 @@ func pullReader(_ context.Context, _ PullMetricExporter) (sdkmetric.Reader, erro return nil, newErrInvalid("no valid metric exporter") } +// cardinalityLimitSelector returns a CardinalityLimitSelector for the given CardinalityLimits config. +// Note: this function is intentionally duplicated in otelconf/x because the two packages define +// their own CardinalityLimits types. Changes here must be mirrored there. +func cardinalityLimitSelector(cl *CardinalityLimits) sdkmetric.CardinalityLimitSelector { + return func(ik sdkmetric.InstrumentKind) (int, bool) { + switch ik { + case sdkmetric.InstrumentKindCounter: + if cl.Counter != nil { + return *cl.Counter, false + } + case sdkmetric.InstrumentKindUpDownCounter: + if cl.UpDownCounter != nil { + return *cl.UpDownCounter, false + } + case sdkmetric.InstrumentKindHistogram: + if cl.Histogram != nil { + return *cl.Histogram, false + } + case sdkmetric.InstrumentKindObservableCounter: + if cl.ObservableCounter != nil { + return *cl.ObservableCounter, false + } + case sdkmetric.InstrumentKindObservableUpDownCounter: + if cl.ObservableUpDownCounter != nil { + return *cl.ObservableUpDownCounter, false + } + case sdkmetric.InstrumentKindObservableGauge: + if cl.ObservableGauge != nil { + return *cl.ObservableGauge, false + } + case sdkmetric.InstrumentKindGauge: + if cl.Gauge != nil { + return *cl.Gauge, false + } + } + if cl.Default != nil { + return *cl.Default, false + } + // fallback=true; defer to the SDK/provider global cardinality limit + return 0, true + } +} + func periodicExporter(ctx context.Context, exporter PushMetricExporter, opts ...sdkmetric.PeriodicReaderOption) (sdkmetric.Reader, error) { exportersConfigured := 0 var exportFunc func() (sdkmetric.Reader, error) diff --git a/otelconf/metric_test.go b/otelconf/metric_test.go index 59dc6cda5b0..7b5b098575d 100644 --- a/otelconf/metric_test.go +++ b/otelconf/metric_test.go @@ -765,6 +765,66 @@ func TestReader(t *testing.T) { }, wantErrT: newErrInvalid("no valid metric exporter"), }, + { + name: "periodic/console-exporter-with-cardinality-limits", + reader: MetricReader{ + Periodic: &PeriodicMetricReader{ + CardinalityLimits: &CardinalityLimits{ + Counter: ptr(100), + UpDownCounter: ptr(200), + Histogram: ptr(300), + ObservableCounter: ptr(400), + ObservableUpDownCounter: ptr(500), + ObservableGauge: ptr(600), + Gauge: ptr(700), + }, + Exporter: PushMetricExporter{ + Console: &ConsoleMetricExporter{}, + }, + }, + }, + wantReader: sdkmetric.NewPeriodicReader( + consoleExporter, + sdkmetric.WithCardinalityLimitSelector(func(ik sdkmetric.InstrumentKind) (int, bool) { + switch ik { + case sdkmetric.InstrumentKindCounter: + return 100, false + case sdkmetric.InstrumentKindUpDownCounter: + return 200, false + case sdkmetric.InstrumentKindHistogram: + return 300, false + case sdkmetric.InstrumentKindObservableCounter: + return 400, false + case sdkmetric.InstrumentKindObservableUpDownCounter: + return 500, false + case sdkmetric.InstrumentKindObservableGauge: + return 600, false + case sdkmetric.InstrumentKindGauge: + return 700, false + } + return 0, true + }), + ), + }, + { + name: "periodic/console-exporter-with-default-cardinality-limit", + reader: MetricReader{ + Periodic: &PeriodicMetricReader{ + CardinalityLimits: &CardinalityLimits{ + Default: ptr(50), + }, + Exporter: PushMetricExporter{ + Console: &ConsoleMetricExporter{}, + }, + }, + }, + wantReader: sdkmetric.NewPeriodicReader( + consoleExporter, + sdkmetric.WithCardinalityLimitSelector(func(sdkmetric.InstrumentKind) (int, bool) { + return 50, false + }), + ), + }, { name: "periodic/console-exporter", reader: MetricReader{ @@ -820,6 +880,126 @@ func TestReader(t *testing.T) { } } +func TestCardinalityLimitSelector(t *testing.T) { + allKinds := []sdkmetric.InstrumentKind{ + sdkmetric.InstrumentKindCounter, + sdkmetric.InstrumentKindUpDownCounter, + sdkmetric.InstrumentKindHistogram, + sdkmetric.InstrumentKindObservableCounter, + sdkmetric.InstrumentKindObservableUpDownCounter, + sdkmetric.InstrumentKindObservableGauge, + sdkmetric.InstrumentKindGauge, + } + + t.Run("per-kind limits", func(t *testing.T) { + cl := &CardinalityLimits{ + Counter: ptr(100), + UpDownCounter: ptr(200), + Histogram: ptr(300), + ObservableCounter: ptr(400), + ObservableUpDownCounter: ptr(500), + ObservableGauge: ptr(600), + Gauge: ptr(700), + } + sel := cardinalityLimitSelector(cl) + expected := map[sdkmetric.InstrumentKind]int{ + sdkmetric.InstrumentKindCounter: 100, + sdkmetric.InstrumentKindUpDownCounter: 200, + sdkmetric.InstrumentKindHistogram: 300, + sdkmetric.InstrumentKindObservableCounter: 400, + sdkmetric.InstrumentKindObservableUpDownCounter: 500, + sdkmetric.InstrumentKindObservableGauge: 600, + sdkmetric.InstrumentKindGauge: 700, + } + for _, ik := range allKinds { + limit, fallback := sel(ik) + assert.Equal(t, expected[ik], limit) + assert.False(t, fallback) + } + }) + + t.Run("default limit used when kind not set", func(t *testing.T) { + cl := &CardinalityLimits{ + Default: ptr(50), + } + sel := cardinalityLimitSelector(cl) + for _, ik := range allKinds { + limit, fallback := sel(ik) + assert.Equal(t, 50, limit) + assert.False(t, fallback) + } + }) + + t.Run("per-kind overrides default", func(t *testing.T) { + cl := &CardinalityLimits{ + Default: ptr(50), + Counter: ptr(100), + } + sel := cardinalityLimitSelector(cl) + limit, fallback := sel(sdkmetric.InstrumentKindCounter) + assert.Equal(t, 100, limit) + assert.False(t, fallback) + + limit, fallback = sel(sdkmetric.InstrumentKindGauge) + assert.Equal(t, 50, limit) + assert.False(t, fallback) + }) + + t.Run("fallback to provider when no limit set", func(t *testing.T) { + cl := &CardinalityLimits{} + sel := cardinalityLimitSelector(cl) + for _, ik := range allKinds { + limit, fallback := sel(ik) + assert.Equal(t, 0, limit) + assert.True(t, fallback) + } + }) +} + +// TestMetricReaderCardinalityLimitsWired verifies that CardinalityLimits set on +// a PeriodicMetricReader are actually wired into the returned SDK reader. +// It records 3 distinct attribute sets with a per-kind limit of 1; the SDK +// must produce exactly 1 data point (only the overflow bucket) rather than +// 3, which would happen if the selector were never registered. With limit=1 +// the overflow slot consumes the entire limit, so no normal data points fit. +func TestMetricReaderCardinalityLimitsWired(t *testing.T) { + ctx := t.Context() + + reader, err := metricReader(ctx, MetricReader{ + Periodic: &PeriodicMetricReader{ + CardinalityLimits: &CardinalityLimits{ + Counter: ptr(1), + }, + Exporter: PushMetricExporter{ + Console: &ConsoleMetricExporter{}, + }, + }, + }) + require.NoError(t, err) + + mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + defer func() { assert.NoError(t, mp.Shutdown(t.Context())) }() + + counter, err := mp.Meter("test").Int64Counter("cardinality.wiring.test") + require.NoError(t, err) + + // Record 3 distinct attribute sets; with limit=1 the SDK must emit only + // 1 data point: the overflow bucket (the limit counts the overflow slot + // itself, so no "normal" data points fit alongside it). + counter.Add(ctx, 1, metric.WithAttributes(attribute.Int("k", 1))) + counter.Add(ctx, 1, metric.WithAttributes(attribute.Int("k", 2))) + counter.Add(ctx, 1, metric.WithAttributes(attribute.Int("k", 3))) + + pr := reader.(*sdkmetric.PeriodicReader) + var rm metricdata.ResourceMetrics + require.NoError(t, pr.Collect(ctx, &rm)) + + require.Len(t, rm.ScopeMetrics, 1) + require.Len(t, rm.ScopeMetrics[0].Metrics, 1) + dataPoints := rm.ScopeMetrics[0].Metrics[0].Data.(metricdata.Sum[int64]).DataPoints + assert.Len(t, dataPoints, 1) +} + func TestView(t *testing.T) { testCases := []struct { name string diff --git a/otelconf/x/metric.go b/otelconf/x/metric.go index 6316010ff58..53bb2a6e096 100644 --- a/otelconf/x/metric.go +++ b/otelconf/x/metric.go @@ -86,6 +86,10 @@ func metricReader(ctx context.Context, r MetricReader) (sdkmetric.Reader, error) if r.Periodic.Timeout != nil { opts = append(opts, sdkmetric.WithTimeout(time.Duration(*r.Periodic.Timeout)*time.Millisecond)) } + + if r.Periodic.CardinalityLimits != nil { + opts = append(opts, sdkmetric.WithCardinalityLimitSelector(cardinalityLimitSelector(r.Periodic.CardinalityLimits))) + } return periodicExporter(ctx, r.Periodic.Exporter, opts...) } @@ -95,6 +99,49 @@ func metricReader(ctx context.Context, r MetricReader) (sdkmetric.Reader, error) return nil, newErrInvalid("no valid metric reader") } +// cardinalityLimitSelector returns a CardinalityLimitSelector for the given CardinalityLimits config. +// Note: this function is intentionally duplicated from otelconf because the two packages define +// their own CardinalityLimits types. Changes here must be mirrored there. +func cardinalityLimitSelector(cl *CardinalityLimits) sdkmetric.CardinalityLimitSelector { + return func(ik sdkmetric.InstrumentKind) (int, bool) { + switch ik { + case sdkmetric.InstrumentKindCounter: + if cl.Counter != nil { + return *cl.Counter, false + } + case sdkmetric.InstrumentKindUpDownCounter: + if cl.UpDownCounter != nil { + return *cl.UpDownCounter, false + } + case sdkmetric.InstrumentKindHistogram: + if cl.Histogram != nil { + return *cl.Histogram, false + } + case sdkmetric.InstrumentKindObservableCounter: + if cl.ObservableCounter != nil { + return *cl.ObservableCounter, false + } + case sdkmetric.InstrumentKindObservableUpDownCounter: + if cl.ObservableUpDownCounter != nil { + return *cl.ObservableUpDownCounter, false + } + case sdkmetric.InstrumentKindObservableGauge: + if cl.ObservableGauge != nil { + return *cl.ObservableGauge, false + } + case sdkmetric.InstrumentKindGauge: + if cl.Gauge != nil { + return *cl.Gauge, false + } + } + if cl.Default != nil { + return *cl.Default, false + } + // fallback=true; defer to the SDK/provider global cardinality limit + return 0, true + } +} + func pullReader(ctx context.Context, exporter PullMetricExporter) (sdkmetric.Reader, error) { if exporter.PrometheusDevelopment != nil { return prometheusReader(ctx, exporter.PrometheusDevelopment) diff --git a/otelconf/x/metric_test.go b/otelconf/x/metric_test.go index 17e98d6da42..4209704445e 100644 --- a/otelconf/x/metric_test.go +++ b/otelconf/x/metric_test.go @@ -833,6 +833,66 @@ func TestReader(t *testing.T) { }, wantErrT: newErrInvalid("no valid metric exporter"), }, + { + name: "periodic/console-exporter-with-cardinality-limits", + reader: MetricReader{ + Periodic: &PeriodicMetricReader{ + CardinalityLimits: &CardinalityLimits{ + Counter: ptr(100), + UpDownCounter: ptr(200), + Histogram: ptr(300), + ObservableCounter: ptr(400), + ObservableUpDownCounter: ptr(500), + ObservableGauge: ptr(600), + Gauge: ptr(700), + }, + Exporter: PushMetricExporter{ + Console: &ConsoleMetricExporter{}, + }, + }, + }, + wantReader: sdkmetric.NewPeriodicReader( + consoleExporter, + sdkmetric.WithCardinalityLimitSelector(func(ik sdkmetric.InstrumentKind) (int, bool) { + switch ik { + case sdkmetric.InstrumentKindCounter: + return 100, false + case sdkmetric.InstrumentKindUpDownCounter: + return 200, false + case sdkmetric.InstrumentKindHistogram: + return 300, false + case sdkmetric.InstrumentKindObservableCounter: + return 400, false + case sdkmetric.InstrumentKindObservableUpDownCounter: + return 500, false + case sdkmetric.InstrumentKindObservableGauge: + return 600, false + case sdkmetric.InstrumentKindGauge: + return 700, false + } + return 0, true + }), + ), + }, + { + name: "periodic/console-exporter-with-default-cardinality-limit", + reader: MetricReader{ + Periodic: &PeriodicMetricReader{ + CardinalityLimits: &CardinalityLimits{ + Default: ptr(50), + }, + Exporter: PushMetricExporter{ + Console: &ConsoleMetricExporter{}, + }, + }, + }, + wantReader: sdkmetric.NewPeriodicReader( + consoleExporter, + sdkmetric.WithCardinalityLimitSelector(func(sdkmetric.InstrumentKind) (int, bool) { + return 50, false + }), + ), + }, { name: "periodic/console-exporter", reader: MetricReader{ @@ -899,6 +959,127 @@ func TestReader(t *testing.T) { } } +func TestCardinalityLimitSelector(t *testing.T) { + allKinds := []sdkmetric.InstrumentKind{ + sdkmetric.InstrumentKindCounter, + sdkmetric.InstrumentKindUpDownCounter, + sdkmetric.InstrumentKindHistogram, + sdkmetric.InstrumentKindObservableCounter, + sdkmetric.InstrumentKindObservableUpDownCounter, + sdkmetric.InstrumentKindObservableGauge, + sdkmetric.InstrumentKindGauge, + } + + t.Run("per-kind limits", func(t *testing.T) { + cl := &CardinalityLimits{ + Counter: ptr(100), + UpDownCounter: ptr(200), + Histogram: ptr(300), + ObservableCounter: ptr(400), + ObservableUpDownCounter: ptr(500), + ObservableGauge: ptr(600), + Gauge: ptr(700), + } + sel := cardinalityLimitSelector(cl) + expected := map[sdkmetric.InstrumentKind]int{ + sdkmetric.InstrumentKindCounter: 100, + sdkmetric.InstrumentKindUpDownCounter: 200, + sdkmetric.InstrumentKindHistogram: 300, + sdkmetric.InstrumentKindObservableCounter: 400, + sdkmetric.InstrumentKindObservableUpDownCounter: 500, + sdkmetric.InstrumentKindObservableGauge: 600, + sdkmetric.InstrumentKindGauge: 700, + } + for _, ik := range allKinds { + limit, fallback := sel(ik) + assert.Equal(t, expected[ik], limit) + assert.False(t, fallback) + } + }) + + t.Run("default limit used when kind not set", func(t *testing.T) { + cl := &CardinalityLimits{ + Default: ptr(50), + } + sel := cardinalityLimitSelector(cl) + for _, ik := range allKinds { + limit, fallback := sel(ik) + assert.Equal(t, 50, limit) + assert.False(t, fallback) + } + }) + + t.Run("per-kind overrides default", func(t *testing.T) { + cl := &CardinalityLimits{ + Default: ptr(50), + Counter: ptr(100), + } + sel := cardinalityLimitSelector(cl) + limit, fallback := sel(sdkmetric.InstrumentKindCounter) + assert.Equal(t, 100, limit) + assert.False(t, fallback) + + limit, fallback = sel(sdkmetric.InstrumentKindGauge) + assert.Equal(t, 50, limit) + assert.False(t, fallback) + }) + + t.Run("fallback to provider when no limit set", func(t *testing.T) { + cl := &CardinalityLimits{} + sel := cardinalityLimitSelector(cl) + for _, ik := range allKinds { + limit, fallback := sel(ik) + assert.Equal(t, 0, limit) + assert.True(t, fallback) + } + }) +} + +// TestMetricReaderCardinalityLimitsWired verifies that CardinalityLimits set on +// a PeriodicMetricReader are actually wired into the returned SDK reader. +// It records 3 distinct attribute sets with a per-kind limit of 1; the SDK +// must produce exactly 1 data point (only the overflow bucket) rather than +// 3, which would happen if the selector were never registered. With limit=1 +// the overflow slot consumes the entire limit, so no normal data points fit. +func TestMetricReaderCardinalityLimitsWired(t *testing.T) { + ctx := t.Context() + + reader, err := metricReader(ctx, MetricReader{ + Periodic: &PeriodicMetricReader{ + CardinalityLimits: &CardinalityLimits{ + Counter: ptr(1), + }, + Exporter: PushMetricExporter{ + Console: &ConsoleMetricExporter{}, + }, + }, + }) + require.NoError(t, err) + + mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + defer func() { assert.NoError(t, mp.Shutdown(t.Context())) }() + + counter, err := mp.Meter("test").Int64Counter("cardinality.wiring.test") + require.NoError(t, err) + + // Record 3 distinct attribute sets; with limit=1 the SDK must emit only + // 1 data point: the overflow bucket (the limit counts the overflow slot + // itself, so no "normal" data points fit alongside it). + counter.Add(ctx, 1, metric.WithAttributes(attribute.Int("k", 1))) + counter.Add(ctx, 1, metric.WithAttributes(attribute.Int("k", 2))) + counter.Add(ctx, 1, metric.WithAttributes(attribute.Int("k", 3))) + + pr := reader.(*sdkmetric.PeriodicReader) + var rm metricdata.ResourceMetrics + require.NoError(t, pr.Collect(ctx, &rm)) + + require.Len(t, rm.ScopeMetrics, 1) + require.Len(t, rm.ScopeMetrics[0].Metrics, 1) + dataPoints := rm.ScopeMetrics[0].Metrics[0].Data.(metricdata.Sum[int64]).DataPoints + // limit=1: all measurements overflow into 1 overflow bucket (not 3 unconstrained). + assert.Len(t, dataPoints, 1) +} + func TestView(t *testing.T) { testCases := []struct { name string