fix(sdk): handle unmapped aggregations defensively in MetricReaderStorage.collect

.changelog/5622.fixed

@@ -0,0 +1 @@
+`opentelemetry-sdk`: reject views with unsupported aggregations in `_check_view_instrument_compatibility` with a warning, and skip unmapped aggregations in `MetricReaderStorage.collect` to prevent `UnboundLocalError`.

opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/metric_reader_storage.py

@@ -16,8 +16,12 @@
 from opentelemetry.sdk.metrics._internal.aggregation import (
     Aggregation,
     AggregationTemporality,
+    DefaultAggregation,
+    DropAggregation,
     ExplicitBucketHistogramAggregation,
     ExponentialBucketHistogramAggregation,
+    LastValueAggregation,
+    SumAggregation,
     _DropAggregation,
     _ExplicitBucketHistogramAggregation,
     _ExponentialBucketHistogramAggregation,
@@ -171,11 +175,12 @@ def collect(self) -> MetricsData | None:
                             data_points=data_points,
                             aggregation_temporality=aggregation_temporality,
                         )
+                    else:
+                        continue
 
                     metrics.append(
                         Metric(
                             # pylint: disable=protected-access
-                            # pylint: disable=possibly-used-before-assignment
                             name=view_instrument_match._name,
                             description=view_instrument_match._description,
                             unit=view_instrument_match._instrument.unit,
@@ -245,9 +250,25 @@ def _check_view_instrument_compatibility(view: View, instrument: _Instrument) ->
         object should be created, `false` otherwise.
         """
 
-        result = True
-
         # pylint: disable=protected-access
+        if isinstance(view._aggregation, Aggregation) and not isinstance(
+            view._aggregation,
+            (
+                DefaultAggregation,
+                DropAggregation,
+                ExplicitBucketHistogramAggregation,
+                ExponentialBucketHistogramAggregation,
+                LastValueAggregation,
+                SumAggregation,
+            ),
+        ):
+            _logger.warning(
+                "Unsupported aggregation %s for instrument %s",
+                type(view._aggregation).__name__,
+                instrument.name,
+            )
+            return False
+
         if isinstance(instrument, Asynchronous) and isinstance(
             view._aggregation,
             (
@@ -260,6 +281,6 @@ def _check_view_instrument_compatibility(view: View, instrument: _Instrument) ->
                 view,
                 instrument,
             )
-            result = False
+            return False
 
-        return result
+        return True

opentelemetry-sdk/tests/metrics/test_metric_reader_storage.py

@@ -28,6 +28,7 @@
 )
 from opentelemetry.sdk.metrics.export import AggregationTemporality
 from opentelemetry.sdk.metrics.view import (
+    Aggregation,
     DefaultAggregation,
     DropAggregation,
     ExplicitBucketHistogramAggregation,
@@ -753,3 +754,91 @@ def test_view_instrument_match_conflict_8(self):
             "will cause conflicting metrics",
             log.records[0].message,
         )
+
+    def test_collect_skips_unsupported_aggregation(self):
+        unsupported_match = Mock(
+            _aggregation=Mock(),
+            _name="unsupported_metric",
+            _description="description",
+            _instrument=Mock(unit="1"),
+        )
+        unsupported_match.collect.return_value = [Mock()]
+
+        valid_point = Mock()
+        valid_match = Mock(
+            _aggregation=_LastValueAggregation({}, Mock(), instrument_is_synchronous=False),
+            _name="valid_metric",
+            _description="description",
+            _instrument=Mock(unit="1"),
+        )
+        valid_match.collect.return_value = [valid_point]
+
+        instrument1 = Mock(name="instrument1")
+        instrument2 = Mock(name="instrument2")
+        storage = MetricReaderStorage(
+            SdkConfiguration(
+                exemplar_filter=Mock(),
+                resource=Mock(),
+                views=(),
+            ),
+            MagicMock(**{"__getitem__.return_value": AggregationTemporality.CUMULATIVE}),
+            MagicMock(**{"__getitem__.return_value": DefaultAggregation()}),
+        )
+        storage._instrument_view_instrument_matches[instrument1] = [unsupported_match]
+        storage._instrument_view_instrument_matches[instrument2] = [valid_match]
+
+        result = storage.collect()
+
+        self.assertIsNotNone(result)
+        self.assertEqual(len(result.resource_metrics[0].scope_metrics[0].metrics), 1)
+        self.assertEqual(result.resource_metrics[0].scope_metrics[0].metrics[0].name, "valid_metric")
+
+    def test_unsupported_aggregation_view_not_applied(self):
+        counter = _ObservableCounter(
+            "test_counter",
+            Mock(),
+            [Mock()],
+            unit="unit",
+            description="description",
+        )
+
+        class CustomAggregation(Aggregation):
+            def _create_aggregation(
+                self,
+                instrument,
+                explicit_bucket_boundaries,
+                exemplar_reservoir_factory,
+                max_scale,
+            ):
+                return Mock()
+
+        metric_reader_storage = MetricReaderStorage(
+            SdkConfiguration(
+                exemplar_filter=Mock(),
+                resource=Mock(),
+                views=(
+                    View(
+                        instrument_name="test_counter",
+                        aggregation=CustomAggregation(),
+                    ),
+                ),
+            ),
+            MagicMock(**{"__getitem__.return_value": AggregationTemporality.CUMULATIVE}),
+            MagicMock(**{"__getitem__.return_value": DefaultAggregation()}),
+        )
+
+        with self.assertLogs(
+            "opentelemetry.sdk.metrics._internal.metric_reader_storage",
+            level=WARNING,
+        ) as log:
+            metric_reader_storage.consume_measurement(Measurement(1, time_ns(), counter, Context()))
+
+        self.assertEqual(len(log.records), 1)
+        self.assertIn(
+            "Unsupported aggregation CustomAggregation for instrument test_counter",
+            log.records[0].message,
+        )
+        self.assertIs(
+            metric_reader_storage._instrument_view_instrument_matches[counter][0]._view,
+            _DEFAULT_VIEW,
+        )

opentelemetry-sdk/tests/metrics/test_periodic_exporting_metric_reader.py

@@ -245,6 +245,7 @@ def test_exporter_temporality_preference(self):
             },
         )
         pmr = PeriodicExportingMetricReader(exporter)
+        self.addCleanup(pmr.shutdown)
         for key, value in pmr._instrument_class_temporality.items():
             if key is not _Counter:
                 self.assertEqual(value, AggregationTemporality.CUMULATIVE)
@@ -258,6 +259,7 @@ def test_exporter_aggregation_preference(self):
             },
         )
         pmr = PeriodicExportingMetricReader(exporter)
+        self.addCleanup(pmr.shutdown)
         for key, value in pmr._instrument_class_aggregation.items():
             if key is not _Counter:
                 self.assertTrue(isinstance(value, DefaultAggregation))