diff --git a/.changelog/5622.fixed b/.changelog/5622.fixed new file mode 100644 index 0000000000..e7c5eab7d3 --- /dev/null +++ b/.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`. 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 d0ef125d79..b7ed96ced7 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 @@ -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 diff --git a/opentelemetry-sdk/tests/metrics/test_metric_reader_storage.py b/opentelemetry-sdk/tests/metrics/test_metric_reader_storage.py index 581e066216..7b30350033 100644 --- a/opentelemetry-sdk/tests/metrics/test_metric_reader_storage.py +++ b/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, + ) diff --git a/opentelemetry-sdk/tests/metrics/test_periodic_exporting_metric_reader.py b/opentelemetry-sdk/tests/metrics/test_periodic_exporting_metric_reader.py index 76ad37b8db..16fd3b07a8 100644 --- a/opentelemetry-sdk/tests/metrics/test_periodic_exporting_metric_reader.py +++ b/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))