feat: add metric name translation strategy support for the Prometheus exporter

CHANGELOG.md

@@ -26,6 +26,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#5076](https://github.com/open-telemetry/opentelemetry-python/pull/5076))
 - `opentelemetry-semantic-conventions`: use `X | Y` union annotation
   ([#5096](https://github.com/open-telemetry/opentelemetry-python/pull/5096))
+- `opentelemetry-exporter-prometheus`: add support for specifying metric name translation strategy 
+  ([#5118](https://github.com/open-telemetry/opentelemetry-python/pull/5118))
 
 
 ## Version 1.41.0/0.62b0 (2026-04-09)

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

@@ -63,6 +63,7 @@
 """
 
 from collections import deque
+from enum import Enum
 from itertools import chain
 from json import dumps
 from logging import getLogger
@@ -76,6 +77,7 @@
     GaugeMetricFamily,
     HistogramMetricFamily,
     InfoMetricFamily,
+    UnknownMetricFamily,
 )
 from prometheus_client.core import Metric as PrometheusMetric
 
@@ -116,6 +118,17 @@
 _TARGET_INFO_DESCRIPTION = "Target metadata"
 
 
+class TranslationStrategy(Enum):
+    """Controls how OpenTelemetry metric names are translated to Prometheus conventions."""
+
+    UNDERSCORE_ESCAPING_WITH_SUFFIXES = "underscore_escaping_with_suffixes"
+    UNDERSCORE_ESCAPING_WITHOUT_SUFFIXES = (
+        "underscore_escaping_without_suffixes"
+    )
+    NO_UTF8_ESCAPING_WITH_SUFFIXES = "no_utf8_escaping_with_suffixes"
+    NO_TRANSLATION = "no_translation"
+
+
 def _convert_buckets(
     bucket_counts: Sequence[int], explicit_bounds: Sequence[float]
 ) -> Sequence[Tuple[str, int]]:
@@ -135,7 +148,10 @@ class PrometheusMetricReader(MetricReader):
     """Prometheus metric exporter for OpenTelemetry."""
 
     def __init__(
-        self, disable_target_info: bool = False, prefix: str = ""
+        self,
+        disable_target_info: bool = False,
+        prefix: str = "",
+        translation_strategy: TranslationStrategy = TranslationStrategy.UNDERSCORE_ESCAPING_WITH_SUFFIXES,
     ) -> None:
         super().__init__(
             preferred_temporality={
@@ -149,7 +165,9 @@ def __init__(
             otel_component_type=OtelComponentTypeValues.PROMETHEUS_HTTP_TEXT_METRIC_EXPORTER,
         )
         self._collector = _CustomCollector(
-            disable_target_info=disable_target_info, prefix=prefix
+            disable_target_info=disable_target_info,
+            prefix=prefix,
+            translation_strategy=translation_strategy,
         )
         REGISTRY.register(self._collector)
         self._collector._callback = self.collect
@@ -176,12 +194,18 @@ class _CustomCollector:
     https://github.com/prometheus/client_python#custom-collectors
     """
 
-    def __init__(self, disable_target_info: bool = False, prefix: str = ""):
+    def __init__(
+        self,
+        disable_target_info: bool = False,
+        prefix: str = "",
+        translation_strategy: TranslationStrategy = TranslationStrategy.UNDERSCORE_ESCAPING_WITH_SUFFIXES,
+    ):
         self._callback = None
         self._metrics_datas: Deque[MetricsData] = deque()
         self._disable_target_info = disable_target_info
         self._target_info = None
         self._prefix = prefix
+        self._translation_strategy = translation_strategy
 
     def add_metrics_data(self, metrics_data: MetricsData) -> None:
         """Add metrics to Prometheus data"""
@@ -220,7 +244,7 @@ def collect(self) -> Iterable[PrometheusMetric]:
             if metric_family_id_metric_family:
                 yield from metric_family_id_metric_family.values()
 
-    # pylint: disable=too-many-locals,too-many-branches
+    # pylint: disable=too-many-locals,too-many-branches,too-many-statements
     def _translate_to_prometheus(
         self,
         metrics_data: MetricsData,
@@ -233,16 +257,30 @@ def _translate_to_prometheus(
                 for metric in scope_metrics.metrics:
                     metrics.append(metric)
 
+        _add_suffixes = self._translation_strategy in (
+            TranslationStrategy.UNDERSCORE_ESCAPING_WITH_SUFFIXES,
+            TranslationStrategy.NO_UTF8_ESCAPING_WITH_SUFFIXES,
+        )
+        _escape_names = self._translation_strategy in (
+            TranslationStrategy.UNDERSCORE_ESCAPING_WITH_SUFFIXES,
+            TranslationStrategy.UNDERSCORE_ESCAPING_WITHOUT_SUFFIXES,
+        )
+
         for metric in metrics:
             label_values_data_points = []
             values = []
 
             metric_name = metric.name
-            if self._prefix:
+            if (
+                self._translation_strategy
+                != TranslationStrategy.NO_TRANSLATION
+                and self._prefix
+            ):
                 metric_name = self._prefix + "_" + metric_name
-            metric_name = sanitize_full_name(metric_name)
+            if _escape_names:
+                metric_name = sanitize_full_name(metric_name)
             metric_description = metric.description or ""
-            metric_unit = map_unit(metric.unit)
+            metric_unit = map_unit(metric.unit) if _add_suffixes else ""
 
             # First pass: collect all unique label keys across all data points
             all_label_keys_set = set()
@@ -306,17 +344,25 @@ def _translate_to_prometheus(
                 isinstance(metric.data, Sum)
                 and not should_convert_sum_to_gauge
             ):
+                family_kwargs = {}
+                if _add_suffixes:
+                    family_class = CounterMetricFamily
+                    family_kwargs["unit"] = metric_unit
+                else:
+                    # The CounterMetricFamily always adds the "_total" suffix to
+                    # metric names. To avoid adding this suffix for Sums, we must
+                    # use the untyped (unknown) metric family.
+                    family_class = UnknownMetricFamily
                 metric_family_id = "|".join(
-                    [per_metric_family_id, CounterMetricFamily.__name__]
+                    [per_metric_family_id, family_class.__name__]
                 )
-
                 if metric_family_id not in metric_family_id_metric_family:
                     metric_family_id_metric_family[metric_family_id] = (
-                        CounterMetricFamily(
+                        family_class(
                             name=metric_name,
                             documentation=metric_description,
                             labels=all_label_keys,
-                            unit=metric_unit,
+                            **family_kwargs,
                         )
                     )
                 for label_values, value in zip(

exporter/opentelemetry-exporter-prometheus/test-requirements.txt

@@ -3,7 +3,7 @@ importlib-metadata==6.11.0
 iniconfig==2.0.0
 packaging==24.0
 pluggy==1.6.0
-prometheus_client==0.20.0
+prometheus_client==0.25.0
 py-cpuinfo==9.0.0
 pytest==7.4.4
 tomli==2.0.1

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

@@ -21,10 +21,12 @@
     CounterMetricFamily,
     GaugeMetricFamily,
     InfoMetricFamily,
+    UnknownMetricFamily,
 )
 
 from opentelemetry.exporter.prometheus import (
     PrometheusMetricReader,
+    TranslationStrategy,
     _CustomCollector,
 )
 from opentelemetry.metrics import NoOpMeterProvider
@@ -47,6 +49,36 @@
 )
 
 
+def _collect_metric(
+    metric: Metric,
+    translation_strategy: TranslationStrategy = TranslationStrategy.UNDERSCORE_ESCAPING_WITH_SUFFIXES,
+    prefix: str = "",
+) -> list:
+    metrics_data = MetricsData(
+        resource_metrics=[
+            ResourceMetrics(
+                resource=Mock(),
+                scope_metrics=[
+                    ScopeMetrics(
+                        scope=Mock(),
+                        metrics=[metric],
+                        schema_url="schema_url",
+                    )
+                ],
+                schema_url="schema_url",
+            )
+        ]
+    )
+    collector = _CustomCollector(
+        disable_target_info=True,
+        prefix=prefix,
+        translation_strategy=translation_strategy,
+    )
+    collector.add_metrics_data(metrics_data)
+    return list(collector.collect())
+
+
+# pylint: disable=too-many-public-methods
 class TestPrometheusMetricReader(TestCase):
     def setUp(self):
         self._mock_registry_register = Mock()
@@ -719,3 +751,82 @@ def test_multiple_data_points_with_different_label_sets(self):
                 """
             ),
         )
+
+    def test_translation_strategy(self):
+        cases = [
+            (
+                TranslationStrategy.UNDERSCORE_ESCAPING_WITH_SUFFIXES,
+                CounterMetricFamily,
+                "test_counter_seconds",
+                GaugeMetricFamily,
+                "test_gauge_seconds",
+            ),
+            (
+                TranslationStrategy.UNDERSCORE_ESCAPING_WITHOUT_SUFFIXES,
+                UnknownMetricFamily,
+                "test_counter",
+                GaugeMetricFamily,
+                "test_gauge",
+            ),
+            (
+                TranslationStrategy.NO_UTF8_ESCAPING_WITH_SUFFIXES,
+                CounterMetricFamily,
+                "test.counter_seconds",
+                GaugeMetricFamily,
+                "test.gauge_seconds",
+            ),
+            (
+                TranslationStrategy.NO_TRANSLATION,
+                UnknownMetricFamily,
+                "test.counter",
+                GaugeMetricFamily,
+                "test.gauge",
+            ),
+        ]
+        for (
+            strategy,
+            counter_cls,
+            counter_name,
+            gauge_cls,
+            gauge_name,
+        ) in cases:
+            with self.subTest(strategy=strategy):
+                counter_result = _collect_metric(
+                    _generate_sum("test.counter", 1, unit="s"), strategy
+                )
+                self.assertEqual(type(counter_result[0]), counter_cls)
+                self.assertEqual(counter_result[0].name, counter_name)
+
+                gauge_result = _collect_metric(
+                    _generate_gauge("test.gauge", 1, unit="s"), strategy
+                )
+                self.assertEqual(type(gauge_result[0]), gauge_cls)
+                self.assertEqual(gauge_result[0].name, gauge_name)
+
+    def test_translation_strategy_prefix(self):
+        cases = [
+            (
+                TranslationStrategy.UNDERSCORE_ESCAPING_WITH_SUFFIXES,
+                "myprefix_test_counter",
+            ),
+            (
+                TranslationStrategy.UNDERSCORE_ESCAPING_WITHOUT_SUFFIXES,
+                "myprefix_test_counter",
+            ),
+            (
+                TranslationStrategy.NO_UTF8_ESCAPING_WITH_SUFFIXES,
+                "myprefix_test.counter",
+            ),
+            (
+                TranslationStrategy.NO_TRANSLATION,
+                "test.counter",  # prefix is not applied
+            ),
+        ]
+        for strategy, expected_name in cases:
+            with self.subTest(strategy=strategy):
+                result = _collect_metric(
+                    _generate_sum("test.counter", 1, unit=""),
+                    strategy,
+                    prefix="myprefix",
+                )
+                self.assertEqual(result[0].name, expected_name)