oss_18_prometheus_openmetrics_compat

.changelog/35.fixed

@@ -0,0 +1,5 @@
+Prometheus exporter: improve Prometheus/OpenMetrics compatibility by merging
+attributes that sanitize to the same label, emitting exemplars on counters and
+histograms, adding a `without_counter_suffixes` option, dropping same-name
+families with conflicting types, and skipping scope attributes that collide
+with reserved `otel_scope_*` labels

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

@@ -70,7 +70,9 @@
     HistogramMetricFamily,
     InfoMetricFamily,
 )
+from prometheus_client.core import Exemplar as PrometheusExemplar
 from prometheus_client.core import Metric as PrometheusMetric
+from prometheus_client.samples import Sample
 
 from opentelemetry.exporter.prometheus._mapping import (
     map_unit,
@@ -100,10 +102,12 @@
     MetricsData,
     Sum,
 )
+from opentelemetry.sdk.metrics._internal.point import Exemplar as OTelExemplar
 from opentelemetry.sdk.util.instrumentation import InstrumentationScope
 from opentelemetry.semconv._incubating.attributes.otel_attributes import (
     OtelComponentTypeValues,
 )
+from opentelemetry.trace import format_span_id, format_trace_id
 from opentelemetry.util.types import Attributes, AttributeValue
 
 _logger = getLogger(__name__)
@@ -116,6 +120,56 @@
 _OTEL_SCOPE_SCHEMA_URL_LABEL = "otel_scope_schema_url"
 _OTEL_SCOPE_ATTR_PREFIX = "otel_scope_"
 
+_RESERVED_SCOPE_LABELS = frozenset(
+    {
+        _OTEL_SCOPE_NAME_LABEL,
+        _OTEL_SCOPE_VERSION_LABEL,
+        _OTEL_SCOPE_SCHEMA_URL_LABEL,
+    }
+)
+
+# Exemplar label names as defined by the Prometheus/OpenMetrics compatibility
+# specification.
+_EXEMPLAR_TRACE_ID_LABEL = "trace_id"
+_EXEMPLAR_SPAN_ID_LABEL = "span_id"
+
+
+def _convert_exemplar(
+    exemplar: OTelExemplar,
+) -> PrometheusExemplar:
+    """Map an OpenTelemetry exemplar to a ``prometheus_client`` exemplar.
+
+    The trace and span identifiers, when present, are exposed as the
+    ``trace_id`` and ``span_id`` labels as required by the Prometheus and
+    OpenMetrics compatibility specification.
+    """
+    labels: dict[str, str] = {}
+    if exemplar.trace_id is not None:
+        labels[_EXEMPLAR_TRACE_ID_LABEL] = format_trace_id(exemplar.trace_id)
+    if exemplar.span_id is not None:
+        labels[_EXEMPLAR_SPAN_ID_LABEL] = format_span_id(exemplar.span_id)
+    timestamp = None
+    if exemplar.time_unix_nano is not None:
+        timestamp = exemplar.time_unix_nano / 1e9
+    return PrometheusExemplar(
+        labels=labels,
+        value=exemplar.value,
+        timestamp=timestamp,
+    )
+
+
+def _first_exemplar(
+    exemplars: Sequence[OTelExemplar] | None,
+) -> PrometheusExemplar | None:
+    """Return the first exemplar of a data point converted for Prometheus.
+
+    Prometheus attaches at most one exemplar per sample, so only the first
+    OpenTelemetry exemplar is used.
+    """
+    if not exemplars:
+        return None
+    return _convert_exemplar(exemplars[0])
+
 
 def _convert_buckets(
     bucket_counts: Sequence[int], explicit_bounds: Sequence[float]
@@ -176,6 +230,8 @@ def _populate_counter_family(
     label_keys: Sequence[str],
     label_rows: Sequence[Sequence[str]],
     values: Sequence[float],
+    exemplars: Sequence[PrometheusExemplar | None],
+    without_counter_suffixes: bool,
 ) -> None:
     family_id = "|".join([per_metric_family_id, CounterMetricFamily.__name__])
     family = _get_or_create_family(
@@ -187,8 +243,15 @@ def _populate_counter_family(
         labels=label_keys,
         unit=unit,
     )
-    for label_values, value in zip(label_rows, values):
-        family.add_metric(labels=label_values, value=value)
+    for label_values, value, exemplar in zip(label_rows, values, exemplars):
+        family.add_metric(labels=label_values, value=value, exemplar=exemplar)
+    if without_counter_suffixes:
+        family.samples = [
+            sample._replace(name=sample.name[: -len("_total")])
+            if sample.name.endswith("_total")
+            else sample
+            for sample in family.samples
+        ]
 
 
 def _populate_gauge_family(
@@ -224,6 +287,7 @@ def _populate_histogram_family(
     label_keys: Sequence[str],
     label_rows: Sequence[Sequence[str]],
     values: Sequence[dict[str, Any]],
+    exemplars: Sequence[PrometheusExemplar | None],
 ) -> None:
     family_id = "|".join(
         [per_metric_family_id, HistogramMetricFamily.__name__]
@@ -237,12 +301,18 @@ def _populate_histogram_family(
         labels=label_keys,
         unit=unit,
     )
-    for label_values, value in zip(label_rows, values):
+    for label_values, value, exemplar in zip(label_rows, values, exemplars):
+        buckets = _convert_buckets(
+            value["bucket_counts"], value["explicit_bounds"]
+        )
+        if exemplar is not None and buckets:
+            # Prometheus attaches an exemplar to a single bucket; use the
+            # catch-all (+Inf) bucket so it is always present.
+            upper_bound, count = buckets[-1]
+            buckets[-1] = (upper_bound, count, exemplar)
         family.add_metric(
             labels=label_values,
-            buckets=_convert_buckets(
-                value["bucket_counts"], value["explicit_bounds"]
-            ),
+            buckets=buckets,
             sum_value=value["sum"],
         )
 
@@ -255,6 +325,8 @@ 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.
+        without_counter_suffixes: Whether to suppress the ``_total`` suffix
+            that is otherwise appended to counter (monotonic Sum) metrics.
     """
 
     def __init__(
@@ -264,6 +336,7 @@ def __init__(
         scope_info_enabled: bool = True,
         *,
         registry: CollectorRegistry = REGISTRY,
+        without_counter_suffixes: bool = False,
     ) -> None:
         super().__init__(
             preferred_temporality={
@@ -280,6 +353,7 @@ def __init__(
             disable_target_info=disable_target_info,
             prefix=prefix,
             scope_info_enabled=scope_info_enabled,
+            without_counter_suffixes=without_counter_suffixes,
         )
         self._registry = registry
         self._registry.register(self._collector)
@@ -312,13 +386,16 @@ def __init__(
         disable_target_info: bool = False,
         prefix: str = "",
         scope_info_enabled: bool = True,
+        without_counter_suffixes: bool = False,
     ):
         self._callback = None
         self._metrics_datas: deque[MetricsData] = deque()
         self._disable_target_info = disable_target_info
         self._scope_info_enabled = scope_info_enabled
+        self._without_counter_suffixes = without_counter_suffixes
         self._target_info = None
         self._prefix = prefix
+        self._metric_name_to_type: dict[str, str] = {}
 
     def add_metrics_data(self, metrics_data: MetricsData) -> None:
         """Add metrics to Prometheus data"""
@@ -334,6 +411,9 @@ def collect(self) -> Iterable[PrometheusMetric]:
             self._callback()
 
         metric_family_id_metric_family = {}
+        # Track the Prometheus type chosen for each metric name during this
+        # collection so conflicting-type families can be detected and dropped.
+        self._metric_name_to_type: dict[str, str] = {}
 
         if len(self._metrics_datas):
             if not self._disable_target_info:
@@ -381,14 +461,41 @@ def _translate_metric(
         metric_name = self._resolve_metric_name(metric.name)
         description = metric.description or ""
         unit = map_unit(metric.unit or "")
-        label_keys, label_rows, values = self._collect_data_points(
-            metric.data, scope_attrs
-        )
-        per_metric_family_id = "|".join((metric_name, description, unit))
 
         convert_sum_to_gauge = _should_convert_sum_to_gauge(metric)
 
         if isinstance(metric.data, Sum) and not convert_sum_to_gauge:
+            prometheus_type = "counter"
+        elif isinstance(metric.data, Gauge) or convert_sum_to_gauge:
+            prometheus_type = "gauge"
+        elif isinstance(metric.data, Histogram):
+            prometheus_type = "histogram"
+        else:
+            _logger.warning("Unsupported metric data. %s", type(metric.data))
+            return
+
+        # Prometheus does not allow two metric families that share a name but
+        # have conflicting types. Keep the first type seen for a name and drop
+        # any later metric of a different type, emitting a warning.
+        existing_type = self._metric_name_to_type.get(metric_name)
+        if existing_type is None:
+            self._metric_name_to_type[metric_name] = prometheus_type
+        elif existing_type != prometheus_type:
+            _logger.warning(
+                "Dropping metric '%s' with type '%s' because a metric with "
+                "the same name and conflicting type '%s' was already exported.",
+                metric_name,
+                prometheus_type,
+                existing_type,
+            )
+            return
+
+        label_keys, label_rows, values, exemplars = self._collect_data_points(
+            metric.data, scope_attrs
+        )
+        per_metric_family_id = "|".join((metric_name, description, unit))
+
+        if prometheus_type == "counter":
             _populate_counter_family(
                 registry=metric_family_id_metric_family,
                 per_metric_family_id=per_metric_family_id,
@@ -398,8 +505,10 @@ def _translate_metric(
                 label_keys=label_keys,
                 label_rows=label_rows,
                 values=values,
+                exemplars=exemplars,
+                without_counter_suffixes=self._without_counter_suffixes,
             )
-        elif isinstance(metric.data, Gauge) or convert_sum_to_gauge:
+        elif prometheus_type == "gauge":
             _populate_gauge_family(
                 registry=metric_family_id_metric_family,
                 per_metric_family_id=per_metric_family_id,
@@ -410,7 +519,7 @@ def _translate_metric(
                 label_rows=label_rows,
                 values=values,
             )
-        elif isinstance(metric.data, Histogram):
+        else:
             _populate_histogram_family(
                 registry=metric_family_id_metric_family,
                 per_metric_family_id=per_metric_family_id,
@@ -420,9 +529,8 @@ def _translate_metric(
                 label_keys=label_keys,
                 label_rows=label_rows,
                 values=values,
+                exemplars=exemplars,
             )
-        else:
-            _logger.warning("Unsupported metric data. %s", type(metric.data))
 
     def _build_scope_attrs(
         self, scope: InstrumentationScope
@@ -447,20 +555,48 @@ def _collect_data_points(
         self,
         metric_data: DataT,
         scope_attrs: dict[str, AttributeValue],
-    ) -> tuple[list[str], list[list[str]], list[float | dict[str, Any]]]:
+    ) -> tuple[
+        list[str],
+        list[list[str]],
+        list[float | dict[str, Any]],
+        list[PrometheusExemplar | None],
+    ]:
         keys: set[str] = set()
         rows: list[dict[str, str]] = []
         values: list[float | dict[str, Any]] = []
+        exemplars: list[PrometheusExemplar | None] = []
+
+        # Reserved scope labels are supplied by the scope and must never be
+        # overwritten by data point attributes that sanitize to the same name.
+        reserved_scope_labels = _RESERVED_SCOPE_LABELS & set(
+            scope_attrs.keys()
+        )
 
         for point in metric_data.data_points:
-            labels: dict[str, str] = {}
-            for key, value in chain(
-                scope_attrs.items(),
-                point.attributes.items(),
+            # Multiple original attribute keys may sanitize to the same
+            # Prometheus label. Accumulate the value of every colliding key so
+            # they can be merged instead of silently overwriting one another.
+            label_values_by_key: dict[str, dict[str, str]] = {}
+            for from_scope, (key, value) in chain(
+                ((True, item) for item in scope_attrs.items()),
+                ((False, item) for item in point.attributes.items()),
             ):
                 label = sanitize_attribute(key)
+                if not from_scope and label in reserved_scope_labels:
+                    # A data point attribute collides with a reserved
+                    # otel_scope_* label; skip it in favor of the scope value.
+                    continue
                 keys.add(label)
-                labels[label] = self._check_value(value)
+                label_values_by_key.setdefault(label, {})[key] = (
+                    self._check_value(value)
+                )
+
+            labels: dict[str, str] = {}
+            for label, original_key_values in label_values_by_key.items():
+                labels[label] = ";".join(
+                    original_key_values[original_key]
+                    for original_key in sorted(original_key_values)
+                )
             rows.append(labels)
 
             if isinstance(point, HistogramDataPoint):
@@ -474,13 +610,17 @@ def _collect_data_points(
             else:
                 values.append(point.value)
 
+            exemplars.append(
+                _first_exemplar(getattr(point, "exemplars", None))
+            )
+
         label_keys = sorted(keys)
         # Backfill missing labels with "" so every data point exposes the
         # full label set expected by the Prometheus family.
         label_rows = [
             [labels.get(k, "") for k in label_keys] for labels in rows
         ]
-        return label_keys, label_rows, values
+        return label_keys, label_rows, values, exemplars
 
     # pylint: disable=no-self-use
     def _check_value(self, value: int | float | str | Sequence) -> str: