From 1f155e19dde6353a034182a4e9802efcb7553da6 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 | 17 +++++++ .../metrics/_internal/measurement_consumer.py | 3 ++ .../_internal/metric_reader_storage.py | 4 ++ .../test_metric_cardinality_limit.py | 44 ++++++++++++++++++- .../metrics/test_measurement_consumer.py | 5 ++- 7 files changed, 77 insertions(+), 4 deletions(-) create mode 100644 .changelog/44.added diff --git a/.changelog/44.added b/.changelog/44.added new file mode 100644 index 0000000000..a8b8c759b6 --- /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 2c9d5312cf..1575df6081 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 if not isinstance(self._view._aggregation, DefaultAggregation): 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 4007919d4b..39f68973f5 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py @@ -197,6 +197,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 @@ -208,6 +214,7 @@ def __init__( preferred_aggregation: dict[type, opentelemetry.sdk.metrics.view.Aggregation] | None = None, *, otel_component_type: OtelComponentTypeValues | None = None, + cardinality_limit: int | None = None, ) -> None: self._collect: ( Callable[ @@ -287,6 +294,10 @@ 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, got {cardinality_limit}") + self._cardinality_limit = cardinality_limit + self._otel_component_type = otel_component_type.value if otel_component_type else type(self).__qualname__ self._metrics = create_metric_reader_metrics( self._otel_component_type, @@ -382,10 +393,13 @@ def __init__( self, preferred_temporality: dict[type, AggregationTemporality] | None = None, preferred_aggregation: dict[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 @@ -428,12 +442,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 a332418b4e..d335c872cb 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/measurement_consumer.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/measurement_consumer.py @@ -54,6 +54,7 @@ def __init__( sdk_config, reader._instrument_class_temporality, reader._instrument_class_aggregation, + reader._cardinality_limit, ) for reader in metric_readers } @@ -128,6 +129,8 @@ def add_metric_reader(self, metric_reader: "opentelemetry.sdk.metrics.MetricRead 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 8d105b0a7e..0d8d66ded0 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,12 +54,14 @@ 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 self._instrument_view_instrument_matches: dict[_Instrument, list[_ViewInstrumentMatch]] = {} 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) -> list[_ViewInstrumentMatch]: # Optimistically get the relevant views for the given instrument. Once set for a given @@ -85,6 +87,7 @@ def _get_or_init_view_instrument_match(self, instrument: _Instrument) -> list[_V view=_DEFAULT_VIEW, instrument=instrument, instrument_class_aggregation=(self._instrument_class_aggregation), + cardinality_limit=self._cardinality_limit, ) ) self._instrument_view_instrument_matches[instrument] = view_instrument_matches @@ -222,6 +225,7 @@ def _handle_view_instrument_match( view=view, instrument=instrument, instrument_class_aggregation=(self._instrument_class_aggregation), + cardinality_limit=self._cardinality_limit, ) for existing_view_instrument_matches in self._instrument_view_instrument_matches.values(): 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 afa5c27cfc..e1e5803219 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") @@ -62,3 +65,40 @@ 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 7633327d41..00689cdd85 100644 --- a/opentelemetry-sdk/tests/metrics/test_measurement_consumer.py +++ b/opentelemetry-sdk/tests/metrics/test_measurement_consumer.py @@ -197,13 +197,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): @@ -265,6 +267,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()