opentelemetry-exporter-prometheus: add preferred_aggregation parameter to PrometheusMetricReader

.changelog/5411.added

@@ -0,0 +1 @@
+`opentelemetry-exporter-prometheus`: Add `preferred_aggregation` parameter to `PrometheusMetricReader`

exporter/opentelemetry-exporter-prometheus/src/opentelemetry/exporter/prometheus/__init__.py

@@ -52,7 +52,7 @@
 """
 
 from collections import deque
-from collections.abc import Callable, Iterable, Sequence
+from collections.abc import Callable, Iterable, Mapping, Sequence
 from itertools import chain
 from json import dumps
 from logging import getLogger
@@ -100,6 +100,7 @@
     MetricsData,
     Sum,
 )
+from opentelemetry.sdk.metrics.view import Aggregation
 from opentelemetry.sdk.util.instrumentation import InstrumentationScope
 from opentelemetry.semconv._incubating.attributes.otel_attributes import (
     OtelComponentTypeValues,
@@ -255,13 +256,19 @@ class PrometheusMetricReader(MetricReader):
         scope_info_enabled: Whether to include instrumentation scope labels on
             exported metrics. Scope labels are exported by default.
         prefix: Prefix added to exported Prometheus metric names.
+        preferred_aggregation: A mapping between instrument classes and
+            aggregation instances used to override the default aggregation of
+            the corresponding instrument classes. Classes not included in the
+            mapping retain their default aggregation. See
+            :class:`~opentelemetry.sdk.metrics.export.MetricReader` for details.
     """
 
     def __init__(
         self,
         disable_target_info: bool = False,
         prefix: str = "",
         scope_info_enabled: bool = True,
+        preferred_aggregation: Mapping[type, Aggregation] | None = None,
         *,
         registry: CollectorRegistry = REGISTRY,
     ) -> None:
@@ -274,6 +281,9 @@ def __init__(
                 ObservableUpDownCounter: AggregationTemporality.CUMULATIVE,
                 ObservableGauge: AggregationTemporality.CUMULATIVE,
             },
+            preferred_aggregation=dict(preferred_aggregation)
+            if preferred_aggregation is not None
+            else None,
             otel_component_type=OtelComponentTypeValues.PROMETHEUS_HTTP_TEXT_METRIC_EXPORTER,
         )
         self._collector = _CustomCollector(

exporter/opentelemetry-exporter-prometheus/tests/test_prometheus_exporter.py

@@ -21,6 +21,7 @@
     _CustomCollector,
 )
 from opentelemetry.metrics import NoOpMeterProvider
+from opentelemetry.sdk.metrics import Histogram as HistogramInstrument
 from opentelemetry.sdk.metrics import MeterProvider
 from opentelemetry.sdk.metrics.export import (
     AggregationTemporality,
@@ -31,6 +32,7 @@
     ResourceMetrics,
     ScopeMetrics,
 )
+from opentelemetry.sdk.metrics.view import ExplicitBucketHistogramAggregation
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.util.instrumentation import InstrumentationScope
 from opentelemetry.test.metrictestutil import (
@@ -920,3 +922,33 @@ def test_multiple_data_points_with_different_label_sets(self):
                 """
             ),
         )
+
+    def test_preferred_aggregation(self):
+        custom_aggregation = {
+            HistogramInstrument: ExplicitBucketHistogramAggregation(
+                boundaries=[1.0, 5.0, 10.0]
+            )
+        }
+        reader = PrometheusMetricReader(
+            preferred_aggregation=custom_aggregation,
+            registry=CollectorRegistry(),
+        )
+        provider = MeterProvider(
+            metric_readers=[reader], shutdown_on_exit=False
+        )
+        meter = provider.get_meter("test")
+        histogram = meter.create_histogram("test_histogram")
+        histogram.record(5)
+
+        result = list(reader._collector.collect())
+        self.assertTrue(result)
+        bucket_bounds = [
+            sample.labels["le"]
+            for metric_family in result
+            for sample in metric_family.samples
+            if "le" in sample.labels
+        ]
+        self.assertIn("1.0", bucket_bounds)
+        self.assertIn("5.0", bucket_bounds)
+        self.assertIn("10.0", bucket_bounds)
+        reader.shutdown()