Add Prometheus Exporter for Exemplars

CHANGELOG.md

@@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4251](https://github.com/open-telemetry/opentelemetry-python/pull/4251))
 - Fix recursion error with sdk disabled and handler added to root logger
   ([#4259](https://github.com/open-telemetry/opentelemetry-python/pull/4259))
+- sdk: Add exemplars to the Prometheus exporter
+  ([#4178](https://github.com/open-telemetry/opentelemetry-python/pull/4178))
 
 ## Version 1.28.0/0.49b0 (2024-11-05)
 

exporter/opentelemetry-exporter-otlp-proto-common/src/opentelemetry/exporter/otlp/proto/common/_internal/metrics_encoder/__init__.py

@@ -347,7 +347,7 @@ def _encode_exemplars(sdk_exemplars: List[Exemplar]) -> List[pb2.Exemplar]:
     Converts a list of SDK Exemplars into a list of protobuf Exemplars.
 
     Args:
-        sdk_exemplars (list): The list of exemplars from the OpenTelemetry SDK.
+        sdk_exemplars: The list of exemplars from the OpenTelemetry SDK.
 
     Returns:
         list: A list of protobuf exemplars.

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

@@ -67,7 +67,7 @@
 from json import dumps
 from logging import getLogger
 from os import environ
-from typing import Deque, Dict, Iterable, Sequence, Tuple, Union
+from typing import Deque, Dict, Iterable, Optional, Sequence, Tuple, Union
 
 from prometheus_client import start_http_server
 from prometheus_client.core import (
@@ -78,6 +78,7 @@
     InfoMetricFamily,
 )
 from prometheus_client.core import Metric as PrometheusMetric
+from prometheus_client.samples import Exemplar as PrometheusExemplar
 
 from opentelemetry.exporter.prometheus._mapping import (
     map_unit,
@@ -90,12 +91,15 @@
 )
 from opentelemetry.sdk.metrics import (
     Counter,
+    Exemplar,
     ObservableCounter,
     ObservableGauge,
     ObservableUpDownCounter,
     UpDownCounter,
 )
-from opentelemetry.sdk.metrics import Histogram as HistogramInstrument
+from opentelemetry.sdk.metrics import (
+    Histogram as HistogramInstrument,
+)
 from opentelemetry.sdk.metrics.export import (
     AggregationTemporality,
     Gauge,
@@ -105,6 +109,7 @@
     MetricsData,
     Sum,
 )
+from opentelemetry.trace import format_span_id, format_trace_id
 from opentelemetry.util.types import Attributes
 
 _logger = getLogger(__name__)
@@ -114,16 +119,32 @@
 
 
 def _convert_buckets(
-    bucket_counts: Sequence[int], explicit_bounds: Sequence[float]
-) -> Sequence[Tuple[str, int]]:
+    bucket_counts: Sequence[int],
+    explicit_bounds: Sequence[float],
+    exemplars: Optional[Sequence[PrometheusExemplar]] = None,
+) -> Sequence[Tuple[str, int, Optional[Exemplar]]]:
     buckets = []
     total_count = 0
+    previous_bound = float("-inf")
+
+    exemplars = list(reversed(exemplars or []))
+    exemplar = exemplars.pop() if exemplars else None
+
     for upper_bound, count in zip(
         chain(explicit_bounds, ["+Inf"]),
         bucket_counts,
     ):
         total_count += count
-        buckets.append((f"{upper_bound}", total_count))
+        current_exemplar = None
+        upper_bound_f = float(upper_bound)
+        while exemplar and previous_bound <= exemplar.value < upper_bound_f:
+            if current_exemplar is None:
+                # Assign the exemplar to the current bucket if it's the first valid one found
+                current_exemplar = exemplar
+            exemplar = exemplars.pop() if exemplars else None
+        previous_bound = upper_bound_f
+
+        buckets.append((f"{upper_bound}", total_count, current_exemplar))
 
     return buckets
 
@@ -238,6 +259,10 @@ def _translate_to_prometheus(
             for number_data_point in metric.data.data_points:
                 label_keys = []
                 label_values = []
+                exemplars = [
+                    self._convert_exemplar(ex)
+                    for ex in number_data_point.exemplars
+                ]
 
                 for key, value in sorted(number_data_point.attributes.items()):
                     label_keys.append(sanitize_attribute(key))
@@ -263,6 +288,7 @@ def _translate_to_prometheus(
                                 number_data_point.explicit_bounds
                             ),
                             "sum": number_data_point.sum,
+                            "exemplars": exemplars,
                         }
                     )
                 else:
@@ -351,7 +377,9 @@ def _translate_to_prometheus(
                     ].add_metric(
                         labels=label_values,
                         buckets=_convert_buckets(
-                            value["bucket_counts"], value["explicit_bounds"]
+                            value["bucket_counts"],
+                            value["explicit_bounds"],
+                            value["exemplars"],
                         ),
                         sum_value=value["sum"],
                     )
@@ -380,6 +408,37 @@ def _create_info_metric(
         info.add_metric(labels=list(attributes.keys()), value=attributes)
         return info
 
+    def _convert_exemplar(self, exemplar_data: Exemplar) -> PrometheusExemplar:
+        """
+        Converts the SDK exemplar into a Prometheus Exemplar, including proper time conversion.
+
+        Parameters:
+        - value (float): The value associated with the exemplar.
+        - exemplar_data (ExemplarData): An OpenTelemetry exemplar data object containing attributes and timing information.
+
+        Returns:
+        - Exemplar: A Prometheus Exemplar object with correct labeling and timing.
+        """
+        labels = {
+            sanitize_attribute(key): str(value)
+            for key, value in exemplar_data.filtered_attributes.items()
+        }
+
+        # Add trace_id and span_id to labels only if they are valid and not None
+        if (
+            exemplar_data.trace_id is not None
+            and exemplar_data.span_id is not None
+        ):
+            labels["trace_id"] = format_trace_id(exemplar_data.trace_id)
+            labels["span_id"] = format_span_id(exemplar_data.span_id)
+
+        # Convert time from nanoseconds to seconds
+        timestamp_seconds = exemplar_data.time_unix_nano / 1e9
+        prom_exemplar = PrometheusExemplar(
+            labels, exemplar_data.value, timestamp_seconds
+        )
+        return prom_exemplar
+
 
 class _AutoPrometheusMetricReader(PrometheusMetricReader):
     """Thin wrapper around PrometheusMetricReader used for the opentelemetry_metrics_exporter entry point.

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

@@ -22,12 +22,15 @@
     GaugeMetricFamily,
     InfoMetricFamily,
 )
+from prometheus_client.openmetrics.exposition import (
+    generate_latest as openmetrics_generate_latest,
+)
 
 from opentelemetry.exporter.prometheus import (
     PrometheusMetricReader,
     _CustomCollector,
 )
-from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.metrics import Exemplar, MeterProvider
 from opentelemetry.sdk.metrics.export import (
     AggregationTemporality,
     Histogram,
@@ -44,9 +47,12 @@
     _generate_sum,
     _generate_unsupported_metric,
 )
+from opentelemetry.trace import format_span_id, format_trace_id
 
 
 class TestPrometheusMetricReader(TestCase):
+    # pylint: disable=too-many-public-methods
+
     def setUp(self):
         self._mock_registry_register = Mock()
         self._registry_register_patch = patch(
@@ -55,7 +61,10 @@ def setUp(self):
         )
 
     def verify_text_format(
-        self, metric: Metric, expect_prometheus_text: str
+        self,
+        metric: Metric,
+        expect_prometheus_text: str,
+        openmetrics_generator: bool = False,
     ) -> None:
         metrics_data = MetricsData(
             resource_metrics=[
@@ -75,7 +84,11 @@ def verify_text_format(
 
         collector = _CustomCollector(disable_target_info=True)
         collector.add_metrics_data(metrics_data)
-        result_bytes = generate_latest(collector)
+        result_bytes = (
+            openmetrics_generate_latest(collector)
+            if openmetrics_generator
+            else generate_latest(collector)
+        )
         result = result_bytes.decode("utf-8")
         self.assertEqual(result, expect_prometheus_text)
 
@@ -131,6 +144,67 @@ def test_histogram_to_prometheus(self):
             ),
         )
 
+    def test_histogram_with_exemplar_to_prometheus(self):
+        span_id = 10217189687419569865
+        trace_id = 67545097771067222548457157018666467027
+        metric = Metric(
+            name="test@name",
+            description="foo",
+            unit="s",
+            data=Histogram(
+                data_points=[
+                    HistogramDataPoint(
+                        attributes={"histo": 1},
+                        start_time_unix_nano=1641946016139533244,
+                        time_unix_nano=1641946016139533244,
+                        exemplars=[
+                            Exemplar(
+                                {"filtered": "banana"},
+                                305.0,
+                                1641946016139533244,
+                                span_id,
+                                trace_id,
+                            ),
+                            # Will be ignored as part of the same buckets
+                            Exemplar(
+                                {"filtered": "banana"},
+                                298.0,
+                                1641946016139533400,
+                                span_id,
+                                trace_id,
+                            ),
+                        ],
+                        count=6,
+                        sum=579.0,
+                        bucket_counts=[1, 3, 2],
+                        explicit_bounds=[123.0, 456.0],
+                        min=1,
+                        max=457,
+                    )
+                ],
+                aggregation_temporality=AggregationTemporality.DELTA,
+            ),
+        )
+        span_str = format_span_id(span_id)
+        trace_str = format_trace_id(trace_id)
+        self.verify_text_format(
+            metric,
+            dedent(
+                f"""\
+                # HELP test_name_seconds foo
+                # TYPE test_name_seconds histogram
+                # UNIT test_name_seconds seconds
+                test_name_seconds_bucket{{histo="1",le="123.0"}} 1.0
+                test_name_seconds_bucket{{histo="1",le="456.0"}} 4.0 # {{filtered="banana",span_id="{span_str}",trace_id="{trace_str}"}} 305.0 1641946016.1395333
+                test_name_seconds_bucket{{histo="1",le="+Inf"}} 6.0
+                test_name_seconds_count{{histo="1"}} 6.0
+                test_name_seconds_sum{{histo="1"}} 579.0
+                # EOF
+                """
+            ),
+            openmetrics_generator=True,
+        )
+
     def test_monotonic_sum_to_prometheus(self):
         labels = {"environment@": "staging", "os": "Windows"}
         metric = _generate_sum(