Ignore reader preferred aggregations incompatible with async instruments

.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

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:

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: