From e579430de948e631947bf3e0635d8ba44a4dab7c Mon Sep 17 00:00:00 2001 From: Diego Hurtado Date: Wed, 22 Jul 2026 01:32:44 -0500 Subject: [PATCH] Add per-MetricReader default cardinality limit Add an optional cardinality_limit parameter to MetricReader and the concrete readers (PeriodicExportingMetricReader, InMemoryMetricReader). When set, the limit is threaded through the SDK pipeline (SynchronousMeasurementConsumer -> MetricReaderStorage -> _ViewInstrumentMatch) so that every metric stream produced through that reader's pipeline overflows into the single otel.metric.overflow series at the reader's limit instead of the base default. When unset, the base default cardinality limit still applies. A non-positive limit is rejected with ValueError. --- .changelog/44.added | 3 ++ .../_internal/_view_instrument_match.py | 5 +- .../sdk/metrics/_internal/export/__init__.py | 20 +++++++ .../metrics/_internal/measurement_consumer.py | 3 ++ .../_internal/metric_reader_storage.py | 4 ++ .../test_metric_cardinality_limit.py | 52 ++++++++++++++++++- .../metrics/test_measurement_consumer.py | 5 +- 7 files changed, 88 insertions(+), 4 deletions(-) create mode 100644 .changelog/44.added diff --git a/.changelog/44.added b/.changelog/44.added new file mode 100644 index 00000000000..a8b8c759b67 --- /dev/null +++ b/.changelog/44.added @@ -0,0 +1,3 @@ +Add a per-`MetricReader` `cardinality_limit` option that overrides the base +default cardinality limit for the metric streams produced through that reader's +pipeline diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/_view_instrument_match.py b/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/_view_instrument_match.py index d8d21085baa..85f31f7e333 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/_view_instrument_match.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/_view_instrument_match.py @@ -39,13 +39,16 @@ def __init__( view: View, instrument: _Instrument, instrument_class_aggregation: dict[type, Aggregation], + cardinality_limit: int | None = None, ): self._view = view self._instrument = instrument self._attributes_aggregation: dict[frozenset, _Aggregation] = {} self._lock = Lock() self._instrument_class_aggregation = instrument_class_aggregation - self._cardinality_limit = _DEFAULT_CARDINALITY_LIMIT + if cardinality_limit is None: + cardinality_limit = _DEFAULT_CARDINALITY_LIMIT + self._cardinality_limit = cardinality_limit self._name = self._view._name or self._instrument.name self._description = ( self._view._description or self._instrument.description diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py index aee356f78d2..38899f6cc56 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py @@ -207,6 +207,12 @@ class MetricReader(ABC): default aggregations. The aggregation defined here will be overridden by an aggregation defined by a view that is not `DefaultAggregation`. + cardinality_limit: The default maximum number of distinct attribute + sets aggregated per metric stream produced through this reader's + pipeline. Once the limit is reached, additional attribute sets are + folded into a single overflow series identified by the attribute + ``otel.metric.overflow=true``. When ``None`` (the default), the + SDK's base default cardinality limit is used. .. document protected _receive_metrics which is a intended to be overridden by subclass .. automethod:: _receive_metrics @@ -222,6 +228,7 @@ def __init__( | None = None, *, otel_component_type: OtelComponentTypeValues | None = None, + cardinality_limit: int | None = None, ) -> None: self._collect: ( Callable[ @@ -323,6 +330,13 @@ def __init__( else: raise Exception(f"Invalid instrument class found {typ}") + if cardinality_limit is not None and cardinality_limit <= 0: + raise ValueError( + f"cardinality_limit must be a positive integer, " + f"got {cardinality_limit}" + ) + self._cardinality_limit = cardinality_limit + self._otel_component_type = ( otel_component_type.value if otel_component_type @@ -432,10 +446,13 @@ def __init__( type, opentelemetry.sdk.metrics.view.Aggregation ] | None = None, + *, + cardinality_limit: int | None = None, ) -> None: super().__init__( preferred_temporality=preferred_temporality, preferred_aggregation=preferred_aggregation, + cardinality_limit=cardinality_limit, ) self._lock = RLock() self._metrics_data: MetricsData | None = None @@ -478,12 +495,15 @@ def __init__( exporter: MetricExporter, export_interval_millis: float | None = None, export_timeout_millis: float | None = None, + *, + cardinality_limit: int | None = None, ) -> None: # PeriodicExportingMetricReader defers to exporter for configuration super().__init__( preferred_temporality=exporter._preferred_temporality, preferred_aggregation=exporter._preferred_aggregation, otel_component_type=OtelComponentTypeValues.PERIODIC_METRIC_READER, + cardinality_limit=cardinality_limit, ) # This lock is held whenever calling self._exporter.export() to prevent concurrent diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/measurement_consumer.py b/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/measurement_consumer.py index b17475bc551..747a77b0af7 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/measurement_consumer.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/measurement_consumer.py @@ -58,6 +58,7 @@ def __init__( sdk_config, reader._instrument_class_temporality, reader._instrument_class_aggregation, + reader._cardinality_limit, ) for reader in metric_readers } @@ -150,6 +151,8 @@ def add_metric_reader( metric_reader._instrument_class_temporality, # pylint: disable-next=protected-access metric_reader._instrument_class_aggregation, + # pylint: disable-next=protected-access + metric_reader._cardinality_limit, ) self._reader_storages = new_reader_storages diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/metric_reader_storage.py b/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/metric_reader_storage.py index 7c409430bf9..c00f59ad158 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/metric_reader_storage.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/metric_reader_storage.py @@ -54,6 +54,7 @@ def __init__( sdk_config: SdkConfiguration, instrument_class_temporality: dict[type, AggregationTemporality], instrument_class_aggregation: dict[type, Aggregation], + cardinality_limit: int | None = None, ) -> None: self._lock = RLock() self._sdk_config = sdk_config @@ -62,6 +63,7 @@ def __init__( ] = {} self._instrument_class_temporality = instrument_class_temporality self._instrument_class_aggregation = instrument_class_aggregation + self._cardinality_limit = cardinality_limit def _get_or_init_view_instrument_match( self, instrument: _Instrument @@ -93,6 +95,7 @@ def _get_or_init_view_instrument_match( instrument_class_aggregation=( self._instrument_class_aggregation ), + cardinality_limit=self._cardinality_limit, ) ) self._instrument_view_instrument_matches[instrument] = ( @@ -258,6 +261,7 @@ def _handle_view_instrument_match( instrument_class_aggregation=( self._instrument_class_aggregation ), + cardinality_limit=self._cardinality_limit, ) for ( diff --git a/opentelemetry-sdk/tests/metrics/integration_test/test_metric_cardinality_limit.py b/opentelemetry-sdk/tests/metrics/integration_test/test_metric_cardinality_limit.py index d694fff7257..2e688302fa3 100644 --- a/opentelemetry-sdk/tests/metrics/integration_test/test_metric_cardinality_limit.py +++ b/opentelemetry-sdk/tests/metrics/integration_test/test_metric_cardinality_limit.py @@ -13,8 +13,11 @@ class TestCardinalityLimit(TestCase): @staticmethod - def _record_distinct_attribute_sets(count): - reader = InMemoryMetricReader() + def _record_distinct_attribute_sets(count, cardinality_limit=None): + if cardinality_limit is None: + reader = InMemoryMetricReader() + else: + reader = InMemoryMetricReader(cardinality_limit=cardinality_limit) meter_provider = MeterProvider(metric_readers=[reader]) meter = meter_provider.get_meter("testmeter") counter = meter.create_counter("testcounter") @@ -73,3 +76,48 @@ def test_no_measurement_dropped_during_overflow(self): sum(data_point.value for data_point in data_points), total_measurements, ) + + def test_reader_cardinality_limit_overflows_at_reader_limit(self): + # A reader configured with a small cardinality limit overflows at that + # limit, independently of the base default. + reader_limit = 10 + data_points = self._record_distinct_attribute_sets( + reader_limit + 20, cardinality_limit=reader_limit + ) + + self.assertEqual(len(data_points), reader_limit) + + overflow_points = [ + data_point + for data_point in data_points + if dict(data_point.attributes) == _OVERFLOW_ATTRIBUTES + ] + self.assertEqual(len(overflow_points), 1) + + def test_reader_cardinality_limit_no_overflow_below_reader_limit(self): + # Below the reader's own limit no overflow series is produced. + reader_limit = 10 + data_points = self._record_distinct_attribute_sets( + reader_limit - 1, cardinality_limit=reader_limit + ) + + self.assertEqual(len(data_points), reader_limit - 1) + self.assertNotIn( + _OVERFLOW_ATTRIBUTES, + [dict(data_point.attributes) for data_point in data_points], + ) + + def test_reader_cardinality_limit_unset_falls_back_to_default(self): + # When the reader does not set a cardinality limit, the base default + # applies (no regression to the base behavior). + data_points = self._record_distinct_attribute_sets( + _DEFAULT_CARDINALITY_LIMIT + 100 + ) + + self.assertEqual(len(data_points), _DEFAULT_CARDINALITY_LIMIT) + + def test_reader_cardinality_limit_rejects_non_positive(self): + for invalid in (0, -1): + with self.subTest(cardinality_limit=invalid): + with self.assertRaises(ValueError): + InMemoryMetricReader(cardinality_limit=invalid) diff --git a/opentelemetry-sdk/tests/metrics/test_measurement_consumer.py b/opentelemetry-sdk/tests/metrics/test_measurement_consumer.py index d09368c75dc..7b005b1e907 100644 --- a/opentelemetry-sdk/tests/metrics/test_measurement_consumer.py +++ b/opentelemetry-sdk/tests/metrics/test_measurement_consumer.py @@ -226,13 +226,15 @@ def test_concurrent_changes_to_metric_readers(self): iteration_timeout_error = "Timed out waiting for iteration to start" mutation_timeout_error = "Timed out waiting for mutation to be done" + initial_reader = MagicMock() + initial_reader._cardinality_limit = None consumer = SynchronousMeasurementConsumer( SdkConfiguration( exemplar_filter=MagicMock(), resource=MagicMock(), views=MagicMock(), ), - metric_readers=[MagicMock()], + metric_readers=[initial_reader], ) def _hooked_iter(iterable): @@ -296,6 +298,7 @@ def add_and_remove_readers(): if not iteration_started.wait(timeout): failure = iteration_timeout_error reader = MagicMock() + reader._cardinality_limit = None consumer.add_metric_reader(reader) consumer.remove_metric_reader(reader) mutation_done.set()