opentelemetry-sdk: add support for metric producers

.changelog/5419.added

@@ -0,0 +1 @@
+`opentelemetry-sdk`: add support for the `MetricProducer` interface

opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py

@@ -6,7 +6,7 @@
 import os
 import weakref
 from abc import ABC, abstractmethod
-from collections.abc import Callable, Iterable
+from collections.abc import Callable, Iterable, Sequence
 from enum import Enum
 from logging import getLogger
 from os import environ, linesep
@@ -55,7 +55,10 @@
     _ObservableUpDownCounter,
     _UpDownCounter,
 )
-from opentelemetry.sdk.metrics._internal.point import MetricsData
+from opentelemetry.sdk.metrics._internal.point import (
+    MetricsData,
+    ScopeMetrics,
+)
 from opentelemetry.semconv._incubating.attributes.otel_attributes import (
     OtelComponentTypeValues,
 )
@@ -178,6 +181,34 @@ def force_flush(self, timeout_millis: float = 10_000) -> bool:
         return True
 
 
+class MetricProducer(ABC):
+    """Interface bridging third-party metric sources into a :class:`MetricReader`.
+
+    An implementation is registered on a ``MetricReader`` (via its
+    ``metric_producers`` argument) and its metrics are collected alongside the
+    SDK's own internal state whenever the reader collects. See the OpenTelemetry
+    metrics SDK specification on
+    `MetricProducer <https://opentelemetry.io/docs/specs/otel/metrics/sdk/#metricproducer>`__.
+    """
+
+    @abstractmethod
+    def produce(
+        self, timeout_millis: float = 10_000
+    ) -> Iterable[ScopeMetrics]:
+        """Returns the producer's metrics as an iterable of ``ScopeMetrics``.
+
+        A producer SHOULD emit a single ``InstrumentationScope`` that identifies
+        the producer itself.
+
+        Args:
+            timeout_millis: Amount of time in milliseconds before the produce
+                operation should time out.
+
+        Returns:
+            An iterable of :class:`~opentelemetry.sdk.metrics.export.ScopeMetrics`.
+        """
+
+
 class MetricReader(ABC):
     # pylint: disable=too-many-branches,broad-exception-raised
     """
@@ -207,6 +238,9 @@ class MetricReader(ABC):
             default aggregations. The aggregation defined here will be
             overridden by an aggregation defined by a view that is not
             `DefaultAggregation`.
+        metric_producers: A sequence of `MetricProducer` instances that bridge
+            third-party metric sources. Their metrics are collected alongside
+            the SDK's own metrics whenever this reader collects.
 
     .. document protected _receive_metrics which is a intended to be overridden by subclass
     .. automethod:: _receive_metrics
@@ -221,8 +255,12 @@ def __init__(
         ]
         | None = None,
         *,
+        metric_producers: Sequence[MetricProducer] = (),
         otel_component_type: OtelComponentTypeValues | None = None,
     ) -> None:
+        self._metric_producers: tuple[MetricProducer, ...] = tuple(
+            metric_producers
+        )
         self._collect: (
             Callable[
                 [
@@ -432,10 +470,13 @@ def __init__(
             type, opentelemetry.sdk.metrics.view.Aggregation
         ]
         | None = None,
+        *,
+        metric_producers: Sequence[MetricProducer] = (),
     ) -> None:
         super().__init__(
             preferred_temporality=preferred_temporality,
             preferred_aggregation=preferred_aggregation,
+            metric_producers=metric_producers,
         )
         self._lock = RLock()
         self._metrics_data: MetricsData | None = None
@@ -478,11 +519,14 @@ def __init__(
         exporter: MetricExporter,
         export_interval_millis: float | None = None,
         export_timeout_millis: float | None = None,
+        *,
+        metric_producers: Sequence[MetricProducer] = (),
     ) -> None:
         # PeriodicExportingMetricReader defers to exporter for configuration
         super().__init__(
             preferred_temporality=exporter._preferred_temporality,
             preferred_aggregation=exporter._preferred_aggregation,
+            metric_producers=metric_producers,
             otel_component_type=OtelComponentTypeValues.PERIODIC_METRIC_READER,
         )
 

opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/measurement_consumer.py

@@ -5,6 +5,7 @@
 
 from abc import ABC, abstractmethod
 from collections.abc import Iterable, Mapping
+from logging import getLogger
 from threading import Lock
 from time import time_ns
 
@@ -17,7 +18,14 @@
 from opentelemetry.sdk.metrics._internal.metric_reader_storage import (
     MetricReaderStorage,
 )
-from opentelemetry.sdk.metrics._internal.point import MetricsData
+from opentelemetry.sdk.metrics._internal.point import (
+    MetricsData,
+    ResourceMetrics,
+    ScopeMetrics,
+)
+from opentelemetry.sdk.resources import Resource
+
+_logger = getLogger(__name__)
 
 
 class MeasurementConsumer(ABC):
@@ -133,8 +141,83 @@ def collect(
 
             result = self._reader_storages[metric_reader].collect()
 
+            if producer_scope_metrics := self._collect_from_producers(
+                metric_reader, deadline_ns
+            ):
+                result = self._merge_producer_metrics(
+                    result, producer_scope_metrics, self._sdk_config.resource
+                )
+
         return result
 
+    @staticmethod
+    def _collect_from_producers(
+        metric_reader: "opentelemetry.sdk.metrics.export.MetricReader",
+        deadline_ns: float,
+    ) -> list[ScopeMetrics]:
+        """Collect ScopeMetrics from the reader's MetricProducers.
+
+        Called while holding ``self._lock`` so ``produce()`` calls are
+        serialized. A producer that fails or times out is isolated so
+        metrics collected by the SDK are not dropped.
+        """
+        producer_scope_metrics: list[ScopeMetrics] = []
+        # pylint: disable-next=protected-access
+        for producer in metric_reader._metric_producers:
+            remaining_millis = (deadline_ns - time_ns()) / 1e6
+            if remaining_millis <= 0:
+                _logger.warning(
+                    "Timed out collecting from metric producers, "
+                    "skipping remaining producers."
+                )
+                break
+
+            try:
+                scopes = list(
+                    producer.produce(timeout_millis=remaining_millis)
+                )
+            # pylint: disable-next=broad-except
+            except Exception:
+                _logger.exception(
+                    "Metric producer %s failed to produce metrics, skipping.",
+                    producer,
+                )
+                continue
+
+            producer_scope_metrics.extend(scopes)
+
+        return producer_scope_metrics
+
+    @staticmethod
+    def _merge_producer_metrics(
+        result: MetricsData | None,
+        producer_scope_metrics: list[ScopeMetrics],
+        resource: Resource,
+    ) -> MetricsData:
+        if result is not None and result.resource_metrics:
+            sdk_resource_metrics = result.resource_metrics[0]
+            merged = ResourceMetrics(
+                resource=sdk_resource_metrics.resource,
+                scope_metrics=[
+                    *sdk_resource_metrics.scope_metrics,
+                    *producer_scope_metrics,
+                ],
+                schema_url=sdk_resource_metrics.schema_url,
+            )
+            return MetricsData(
+                resource_metrics=[merged, *result.resource_metrics[1:]]
+            )
+
+        return MetricsData(
+            resource_metrics=[
+                ResourceMetrics(
+                    resource=resource,
+                    scope_metrics=producer_scope_metrics,
+                    schema_url=resource.schema_url,
+                )
+            ]
+        )
+
     def add_metric_reader(
         self, metric_reader: "opentelemetry.sdk.metrics.MetricReader"
     ) -> None:

opentelemetry-sdk/src/opentelemetry/sdk/metrics/export/__init__.py

@@ -10,6 +10,7 @@
     InMemoryMetricReader,
     MetricExporter,
     MetricExportResult,
+    MetricProducer,
     MetricReader,
     PeriodicExportingMetricReader,
 )
@@ -39,6 +40,7 @@
     "InMemoryMetricReader",
     "MetricExporter",
     "MetricExportResult",
+    "MetricProducer",
     "MetricReader",
     "PeriodicExportingMetricReader",
     "DataPointT",

opentelemetry-sdk/tests/metrics/test_measurement_consumer.py

@@ -30,7 +30,7 @@ def test_parent(self, _):
 
     def test_creates_metric_reader_storages(self, MockMetricReaderStorage):
         """It should create one MetricReaderStorage per metric reader passed in the SdkConfiguration"""
-        reader_mocks = [Mock() for _ in range(5)]
+        reader_mocks = [Mock(_metric_producers=[]) for _ in range(5)]
         SynchronousMeasurementConsumer(
             SdkConfiguration(
                 exemplar_filter=Mock(),
@@ -44,7 +44,7 @@ def test_creates_metric_reader_storages(self, MockMetricReaderStorage):
     def test_measurements_passed_to_each_reader_storage(
         self, MockMetricReaderStorage
     ):
-        reader_mocks = [Mock() for _ in range(5)]
+        reader_mocks = [Mock(_metric_producers=[]) for _ in range(5)]
         reader_storage_mocks = [Mock() for _ in range(5)]
         MockMetricReaderStorage.side_effect = reader_storage_mocks
 
@@ -66,7 +66,7 @@ def test_measurements_passed_to_each_reader_storage(
 
     def test_collect_passed_to_reader_stage(self, MockMetricReaderStorage):
         """Its collect() method should defer to the underlying MetricReaderStorage"""
-        reader_mocks = [Mock() for _ in range(5)]
+        reader_mocks = [Mock(_metric_producers=[]) for _ in range(5)]
         reader_storage_mocks = [Mock() for _ in range(5)]
         MockMetricReaderStorage.side_effect = reader_storage_mocks
 
@@ -86,7 +86,7 @@ def test_collect_passed_to_reader_stage(self, MockMetricReaderStorage):
     def test_collect_calls_async_instruments(self, MockMetricReaderStorage):
         """Its collect() method should invoke async instruments and pass measurements to the
         corresponding metric reader storage"""
-        reader_mock = Mock()
+        reader_mock = Mock(_metric_producers=[])
         reader_storage_mock = Mock()
         MockMetricReaderStorage.return_value = reader_storage_mock
         consumer = SynchronousMeasurementConsumer(
@@ -117,7 +117,7 @@ def test_collect_calls_async_instruments(self, MockMetricReaderStorage):
         self.assertFalse(reader_storage_mock.consume_measurement.call_args[1])
 
     def test_collect_timeout(self, MockMetricReaderStorage):
-        reader_mock = Mock()
+        reader_mock = Mock(_metric_producers=[])
         reader_storage_mock = Mock()
         MockMetricReaderStorage.return_value = reader_storage_mock
         consumer = SynchronousMeasurementConsumer(
@@ -151,7 +151,7 @@ def sleep_1(*args, **kwargs):
     def test_collect_deadline(
         self, mock_time_ns, mock_callback_options, MockMetricReaderStorage
     ):
-        reader_mock = Mock()
+        reader_mock = Mock(_metric_producers=[])
         reader_storage_mock = Mock()
         MockMetricReaderStorage.return_value = reader_storage_mock
         consumer = SynchronousMeasurementConsumer(