oss_11_m2_per_reader_cardinality_limit

.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

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

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

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
 

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 (

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)

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