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