sdk/metric: Support specifying cardinality limits per instrument kinds

CHANGELOG.md

@@ -17,6 +17,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - Support attributes with empty value (`attribute.EMPTY`) in `go.opentelemetry.io/otel/sdk/metric/metricdata/metricdatatest`. (#8038)
 - Add support for per-series start time tracking for cumulative metrics in `go.opentelemetry.io/otel/sdk/metric`.
   Set `OTEL_GO_X_PER_SERIES_START_TIMESTAMPS=true` to enable. (#8060)
+- Add `WithCardinalityLimitSelector` for metric reader for configuring cardinality limits specific to the instrument kind. (#7855)
 
 ### Changed
 

sdk/metric/config.go

@@ -160,12 +160,14 @@ func WithExemplarFilter(filter exemplar.Filter) Option {
 	})
 }
 
-// WithCardinalityLimit sets the cardinality limit for the MeterProvider.
+// WithCardinalityLimit sets the global cardinality limit for the MeterProvider.
 //
 // The cardinality limit is the hard limit on the number of metric datapoints
 // that can be collected for a single instrument in a single collect cycle.
 //
 // Setting this to a zero or negative value means no limit is applied.
+// This value applies to all instrument kinds, but can be overridden per kind by
+// the reader's cardinality limit selector (see [WithCardinalityLimitSelector]).
 func WithCardinalityLimit(limit int) Option {
 	// For backward compatibility, the environment variable `OTEL_GO_X_CARDINALITY_LIMIT`
 	// can also be used to set this value.

sdk/metric/config_test.go

@@ -20,13 +20,14 @@ import (
 )
 
 type reader struct {
-	producer          sdkProducer
-	externalProducers []Producer
-	temporalityFunc   TemporalitySelector
-	aggregationFunc   AggregationSelector
-	collectFunc       func(context.Context, *metricdata.ResourceMetrics) error
-	forceFlushFunc    func(context.Context) error
-	shutdownFunc      func(context.Context) error
+	producer                 sdkProducer
+	externalProducers        []Producer
+	temporalityFunc          TemporalitySelector
+	aggregationFunc          AggregationSelector
+	cardinalityLimitSelector CardinalityLimitSelector
+	collectFunc              func(context.Context, *metricdata.ResourceMetrics) error
+	forceFlushFunc           func(context.Context) error
+	shutdownFunc             func(context.Context) error
 }
 
 const envVarResourceAttributes = "OTEL_RESOURCE_ATTRIBUTES"
@@ -45,6 +46,13 @@ func (r *reader) temporality(kind InstrumentKind) metricdata.Temporality {
 	return r.temporalityFunc(kind)
 }
 
+func (r *reader) cardinalityLimit(kind InstrumentKind) (int, bool) {
+	if r.cardinalityLimitSelector != nil {
+		return r.cardinalityLimitSelector(kind)
+	}
+	return 0, true
+}
+
 func (r *reader) Collect(ctx context.Context, rm *metricdata.ResourceMetrics) error {
 	return r.collectFunc(ctx, rm)
 }

sdk/metric/manual_reader.go

@@ -32,8 +32,9 @@ type ManualReader struct {
 	isShutdown        bool
 	externalProducers atomic.Value
 
-	temporalitySelector TemporalitySelector
-	aggregationSelector AggregationSelector
+	temporalitySelector      TemporalitySelector
+	aggregationSelector      AggregationSelector
+	cardinalityLimitSelector CardinalityLimitSelector
 
 	inst *observ.Instrumentation
 }
@@ -45,8 +46,9 @@ var _ = map[Reader]struct{}{&ManualReader{}: {}}
 func NewManualReader(opts ...ManualReaderOption) *ManualReader {
 	cfg := newManualReaderConfig(opts)
 	r := &ManualReader{
-		temporalitySelector: cfg.temporalitySelector,
-		aggregationSelector: cfg.aggregationSelector,
+		temporalitySelector:      cfg.temporalitySelector,
+		aggregationSelector:      cfg.aggregationSelector,
+		cardinalityLimitSelector: cfg.cardinalityLimitSelector,
 	}
 	r.externalProducers.Store(cfg.producers)
 
@@ -89,6 +91,11 @@ func (mr *ManualReader) aggregation(
 	return mr.aggregationSelector(kind)
 }
 
+// cardinalityLimit returns the cardinality limit for kind.
+func (mr *ManualReader) cardinalityLimit(kind InstrumentKind) (int, bool) {
+	return mr.cardinalityLimitSelector(kind)
+}
+
 // Shutdown closes any connections and frees any resources used by the reader.
 //
 // This method is safe to call concurrently.
@@ -179,16 +186,18 @@ func (r *ManualReader) MarshalLog() any {
 
 // manualReaderConfig contains configuration options for a ManualReader.
 type manualReaderConfig struct {
-	temporalitySelector TemporalitySelector
-	aggregationSelector AggregationSelector
-	producers           []Producer
+	temporalitySelector      TemporalitySelector
+	aggregationSelector      AggregationSelector
+	cardinalityLimitSelector CardinalityLimitSelector
+	producers                []Producer
 }
 
 // newManualReaderConfig returns a manualReaderConfig configured with options.
 func newManualReaderConfig(opts []ManualReaderOption) manualReaderConfig {
 	cfg := manualReaderConfig{
-		temporalitySelector: DefaultTemporalitySelector,
-		aggregationSelector: DefaultAggregationSelector,
+		temporalitySelector:      DefaultTemporalitySelector,
+		aggregationSelector:      DefaultAggregationSelector,
+		cardinalityLimitSelector: defaultCardinalityLimitSelector,
 	}
 	for _, opt := range opts {
 		cfg = opt.applyManual(cfg)

sdk/metric/periodic_reader.go

@@ -26,17 +26,19 @@ const (
 
 // periodicReaderConfig contains configuration options for a PeriodicReader.
 type periodicReaderConfig struct {
-	interval  time.Duration
-	timeout   time.Duration
-	producers []Producer
+	interval                 time.Duration
+	timeout                  time.Duration
+	producers                []Producer
+	cardinalityLimitSelector CardinalityLimitSelector
 }
 
 // newPeriodicReaderConfig returns a periodicReaderConfig configured with
 // options.
 func newPeriodicReaderConfig(options []PeriodicReaderOption) periodicReaderConfig {
 	c := periodicReaderConfig{
-		interval: envDuration(envInterval, defaultInterval),
-		timeout:  envDuration(envTimeout, defaultTimeout),
+		interval:                 envDuration(envInterval, defaultInterval),
+		timeout:                  envDuration(envTimeout, defaultTimeout),
+		cardinalityLimitSelector: defaultCardinalityLimitSelector,
 	}
 	for _, o := range options {
 		c = o.applyPeriodic(c)
@@ -111,12 +113,13 @@ func NewPeriodicReader(exporter Exporter, options ...PeriodicReaderOption) *Peri
 		context.Background(),
 	)
 	r := &PeriodicReader{
-		interval: conf.interval,
-		timeout:  conf.timeout,
-		exporter: exporter,
-		flushCh:  make(chan chan error),
-		cancel:   cancel,
-		done:     make(chan struct{}),
+		interval:                 conf.interval,
+		timeout:                  conf.timeout,
+		exporter:                 exporter,
+		flushCh:                  make(chan chan error),
+		cancel:                   cancel,
+		done:                     make(chan struct{}),
+		cardinalityLimitSelector: conf.cardinalityLimitSelector,
 		rmPool: sync.Pool{
 			New: func() any {
 				return &metricdata.ResourceMetrics{}
@@ -170,6 +173,8 @@ type PeriodicReader struct {
 
 	rmPool sync.Pool
 
+	cardinalityLimitSelector CardinalityLimitSelector
+
 	inst *observ.Instrumentation
 }
 
@@ -222,6 +227,11 @@ func (r *PeriodicReader) aggregation(
 	return r.exporter.Aggregation(kind)
 }
 
+// cardinalityLimit returns the cardinality limit for kind.
+func (r *PeriodicReader) cardinalityLimit(kind InstrumentKind) (int, bool) {
+	return r.cardinalityLimitSelector(kind)
+}
+
 // collectAndExport gather all metric data related to the periodicReader r from
 // the SDK and exports it with r's exporter.
 func (r *PeriodicReader) collectAndExport(ctx context.Context) error {

sdk/metric/pipeline.go

@@ -395,9 +395,7 @@ func (i *inserter[N]) cachedAggregator(
 		b.Filter = stream.AttributeFilter
 		// A value less than or equal to zero will disable the aggregation
 		// limits for the builder (an all the created aggregates).
-		// cardinalityLimit will be 0 by default if unset (or
-		// unrecognized input). Use that value directly.
-		b.AggregationLimit = i.pipeline.cardinalityLimit
+		b.AggregationLimit = i.getCardinalityLimit(kind)
 		in, out, err := i.aggregateFunc(b, stream.Aggregation, kind)
 		if err != nil {
 			return aggVal[N]{0, nil, err}
@@ -419,6 +417,18 @@ func (i *inserter[N]) cachedAggregator(
 	return cv.Measure, cv.ID, cv.Err
 }
 
+// getCardinalityLimit returns the cardinality limit for the given instrument kind.
+// When the reader's selector returns fallback = true, the pipeline's global
+// limit is used, then the default if global is unset. When fallback is false,
+// the selector's limit is used (0 or less means unlimited).
+func (i *inserter[N]) getCardinalityLimit(kind InstrumentKind) int {
+	limit, fallback := i.pipeline.reader.cardinalityLimit(kind)
+	if fallback {
+		return i.pipeline.cardinalityLimit
+	}
+	return limit
+}
+
 // logConflict validates if an instrument with the same case-insensitive name
 // as id has already been created. If that instrument conflicts with id, a
 // warning is logged.