diff --git a/.changelog/5554.fixed b/.changelog/5554.fixed new file mode 100644 index 00000000000..a026f2e3236 --- /dev/null +++ b/.changelog/5554.fixed @@ -0,0 +1 @@ +`opentelemetry-sdk`: fix `MetricReader` `preferred_aggregation` silently producing no data points when it maps an asynchronous instrument to a histogram aggregation 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 4007919d4b5..e8dd86009d1 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py @@ -37,6 +37,8 @@ from opentelemetry.sdk.metrics._internal.aggregation import ( AggregationTemporality, DefaultAggregation, + ExplicitBucketHistogramAggregation, + ExponentialBucketHistogramAggregation, ) from opentelemetry.sdk.metrics._internal.exceptions import MetricsTimeoutError from opentelemetry.sdk.metrics._internal.instrument import ( @@ -65,6 +67,19 @@ _logger = getLogger(__name__) +# Histogram aggregations only implement the DELTA instrument temporality path, +# so asynchronous instruments (always CUMULATIVE) would silently produce no +# data points. `MetricReaderStorage` applies the same check to views. +_ASYNCHRONOUS_INSTRUMENT_CLASSES = ( + ObservableCounter, + ObservableUpDownCounter, + ObservableGauge, +) +_ASYNCHRONOUS_INCOMPATIBLE_AGGREGATIONS = ( + ExplicitBucketHistogramAggregation, + ExponentialBucketHistogramAggregation, +) + class MetricExportResult(Enum): """Result of exporting a metric @@ -270,6 +285,16 @@ def __init__( if preferred_aggregation is not None: for typ, aggregation in preferred_aggregation.items(): + if typ in _ASYNCHRONOUS_INSTRUMENT_CLASSES and isinstance( + aggregation, _ASYNCHRONOUS_INCOMPATIBLE_AGGREGATIONS + ): + _logger.warning( + "Instrument class %s and aggregation %s will produce semantic errors when matched, " + "the preferred aggregation has not been applied.", + typ.__name__, + type(aggregation).__name__, + ) + continue if typ is Counter: self._instrument_class_aggregation[_Counter] = aggregation elif typ is UpDownCounter: diff --git a/opentelemetry-sdk/tests/metrics/test_metric_reader.py b/opentelemetry-sdk/tests/metrics/test_metric_reader.py index 7a5f4037f6a..ff05c8b5e0c 100644 --- a/opentelemetry-sdk/tests/metrics/test_metric_reader.py +++ b/opentelemetry-sdk/tests/metrics/test_metric_reader.py @@ -7,7 +7,13 @@ from unittest import TestCase from unittest.mock import patch -from opentelemetry.sdk.metrics import Counter, Histogram, ObservableGauge +from opentelemetry.sdk.metrics import ( + Counter, + Histogram, + ObservableCounter, + ObservableGauge, + ObservableUpDownCounter, +) from opentelemetry.sdk.metrics import _Gauge as _SDKGauge from opentelemetry.sdk.metrics._internal.instrument import ( _Counter, @@ -26,6 +32,8 @@ from opentelemetry.sdk.metrics.view import ( Aggregation, DefaultAggregation, + ExplicitBucketHistogramAggregation, + ExponentialBucketHistogramAggregation, LastValueAggregation, ) @@ -126,6 +134,38 @@ def test_configure_aggregation(self): LastValueAggregation, ) + def test_configure_aggregation_asynchronous_histogram_incompatibility(self): + for instrument_class, internal_class in ( + (ObservableCounter, _ObservableCounter), + (ObservableUpDownCounter, _ObservableUpDownCounter), + (ObservableGauge, _ObservableGauge), + ): + for aggregation in ( + ExplicitBucketHistogramAggregation(), + ExponentialBucketHistogramAggregation(), + ): + with self.subTest( + instrument=instrument_class.__name__, + aggregation=type(aggregation).__name__, + ): + with self.assertLogs(level="WARNING") as log: + dummy_metric_reader = DummyMetricReader(preferred_aggregation={instrument_class: aggregation}) + self.assertIn("will produce semantic errors when matched", log.output[0]) + # The incompatible aggregation is ignored, so the instrument + # keeps producing data through its default aggregation. + self.assertIsInstance( + dummy_metric_reader._instrument_class_aggregation[internal_class], + DefaultAggregation, + ) + + def test_configure_aggregation_synchronous_histogram_allowed(self): + aggregation = ExplicitBucketHistogramAggregation() + dummy_metric_reader = DummyMetricReader(preferred_aggregation={Counter: aggregation}) + self.assertIs( + dummy_metric_reader._instrument_class_aggregation[_Counter], + aggregation, + ) + # pylint: disable=no-self-use def test_force_flush(self): with patch.object(DummyMetricReader, "collect") as mock_collect: