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()