New APIs to add/remove metric readers at run-time (open-telemetry#4863)

* New APIs to add/remove metric readers at run-time

* Updated typehints for `_collect` callback.
Added Changelog entry for the public API

* Updated tests and fixed lint

---------

Co-authored-by: Riccardo Magliocchetti <riccardo.magliocchetti@gmail.com>
Co-authored-by: Emídio Neto <9735060+emdneto@users.noreply.github.com>
Co-authored-by: Aaron Abbott <aaronabbott@google.com>

Author: 4 people

.changelog/4863.added

@@ -0,0 +1 @@
+`opentelemetry-sdk`: add `add_metric_reader` / `remove_metric_reader` public APIs to register / unregister metric readers at runtime.

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

@@ -490,11 +490,12 @@ def __init__(
                 )
             ),
             resource=resource,
-            metric_readers=metric_readers,
             views=views,
         )
+        self._metric_readers = metric_readers
         self._measurement_consumer = SynchronousMeasurementConsumer(
-            sdk_config=self._sdk_config
+            sdk_config=self._sdk_config,
+            metric_readers=metric_readers,
         )
         disabled = environ.get(OTEL_SDK_DISABLED, "")
         self._disabled = disabled.lower().strip() == "true"
@@ -509,7 +510,7 @@ def __init__(
             _meter_configurator or _default_meter_configurator
         )
 
-        for metric_reader in self._sdk_config.metric_readers:
+        for metric_reader in self._metric_readers:
             with self._all_metric_readers_lock:
                 if metric_reader in self._all_metric_readers:
                     # pylint: disable=broad-exception-raised
@@ -560,7 +561,7 @@ def force_flush(self, timeout_millis: float = 10_000) -> bool:
 
         metric_reader_error = {}
 
-        for metric_reader in self._sdk_config.metric_readers:
+        for metric_reader in self._metric_readers:
             current_ts = time_ns()
             try:
                 if current_ts >= deadline_ns:
@@ -605,7 +606,7 @@ def _shutdown():
 
         metric_reader_error = {}
 
-        for metric_reader in self._sdk_config.metric_readers:
+        for metric_reader in self._metric_readers:
             current_ts = time_ns()
             try:
                 if current_ts >= deadline_ns:
@@ -675,3 +676,36 @@ def get_meter(
                     ),
                 )
             return self._meters[instrumentation_scope]
+
+    def add_metric_reader(
+        self, metric_reader: "opentelemetry.sdk.metrics.export.MetricReader"
+    ) -> None:
+        with self._all_metric_readers_lock:
+            if metric_reader in self._all_metric_readers:
+                _logger.warning(
+                    "MetricReader '%s' has been registered already!",
+                    metric_reader,
+                )
+                return
+            self._measurement_consumer.add_metric_reader(metric_reader)
+            # pylint: disable-next=protected-access
+            metric_reader._set_collect_callback(
+                self._measurement_consumer.collect
+            )
+            self._all_metric_readers.add(metric_reader)
+
+    def remove_metric_reader(
+        self,
+        metric_reader: "opentelemetry.sdk.metrics.export.MetricReader",
+    ) -> None:
+        with self._all_metric_readers_lock:
+            if metric_reader not in self._all_metric_readers:
+                _logger.warning(
+                    "MetricReader '%s' has not been registered!", metric_reader
+                )
+                return
+            self._measurement_consumer.remove_metric_reader(metric_reader)
+            # pylint: disable-next=protected-access
+            metric_reader._set_collect_callback(None)
+            metric_reader.shutdown()
+            self._all_metric_readers.remove(metric_reader)

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

@@ -223,13 +223,16 @@ def __init__(
         *,
         otel_component_type: OtelComponentTypeValues | None = None,
     ) -> None:
-        self._collect: Callable[
-            [
-                opentelemetry.sdk.metrics.export.MetricReader,
-                AggregationTemporality,
-            ],
-            Iterable[opentelemetry.sdk.metrics.export.Metric],
-        ] = None
+        self._collect: (
+            Callable[
+                [
+                    opentelemetry.sdk.metrics.export.MetricReader,
+                    AggregationTemporality,
+                ],
+                Iterable[opentelemetry.sdk.metrics.export.Metric],
+            ]
+            | None
+        ) = None
 
         self._instrument_class_temporality = {
             _Counter: AggregationTemporality.CUMULATIVE,
@@ -373,7 +376,8 @@ def _set_collect_callback(
                 AggregationTemporality,
             ],
             MetricsData,
-        ],
+        ]
+        | None,
     ) -> None:
         """This function is internal to the SDK. It should not be called or overridden by users"""
         self._collect = func

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

@@ -3,15 +3,15 @@
 
 # pylint: disable=unused-import
 
+import weakref
 from abc import ABC, abstractmethod
-from collections.abc import Mapping
+from collections.abc import Iterable, Mapping
 from threading import Lock
 from time import time_ns
 
 # This kind of import is needed to avoid Sphinx errors.
 import opentelemetry.sdk.metrics
 import opentelemetry.sdk.metrics._internal.instrument
-import opentelemetry.sdk.metrics._internal.sdk_configuration
 from opentelemetry.metrics._internal.instrument import CallbackOptions
 from opentelemetry.sdk.metrics._internal.exceptions import MetricsTimeoutError
 from opentelemetry.sdk.metrics._internal.measurement import Measurement
@@ -48,10 +48,10 @@ class SynchronousMeasurementConsumer(MeasurementConsumer):
     def __init__(
         self,
         sdk_config: "opentelemetry.sdk.metrics._internal.sdk_configuration.SdkConfiguration",
+        metric_readers: Iterable["opentelemetry.sdk.metrics.MetricReader"],
     ) -> None:
         self._lock = Lock()
         self._sdk_config = sdk_config
-        # should never be mutated
         self._reader_storages: Mapping[
             opentelemetry.sdk.metrics.export.MetricReader, MetricReaderStorage
         ] = {
@@ -60,7 +60,7 @@ def __init__(
                 reader._instrument_class_temporality,
                 reader._instrument_class_aggregation,
             )
-            for reader in sdk_config.metric_readers
+            for reader in metric_readers
         }
         self._async_instruments: list[
             opentelemetry.sdk.metrics._internal.instrument._Asynchronous
@@ -75,7 +75,9 @@ def consume_measurement(self, measurement: Measurement) -> None:
                 measurement.context,
             )
         )
-        for reader_storage in self._reader_storages.values():
+        with self._lock:
+            reader_storages = weakref.WeakSet(self._reader_storages.values())
+        for reader_storage in reader_storages:
             reader_storage.consume_measurement(
                 measurement, should_sample_exemplar
             )
@@ -132,3 +134,23 @@ def collect(
             result = self._reader_storages[metric_reader].collect()
 
         return result
+
+    def add_metric_reader(
+        self, metric_reader: "opentelemetry.sdk.metrics.MetricReader"
+    ) -> None:
+        """Registers a new metric reader."""
+        with self._lock:
+            self._reader_storages[metric_reader] = MetricReaderStorage(
+                self._sdk_config,
+                # pylint: disable-next=protected-access
+                metric_reader._instrument_class_temporality,
+                # pylint: disable-next=protected-access
+                metric_reader._instrument_class_aggregation,
+            )
+
+    def remove_metric_reader(
+        self, metric_reader: "opentelemetry.sdk.metrics.MetricReader"
+    ) -> None:
+        """Unregisters the given metric reader."""
+        with self._lock:
+            self._reader_storages.pop(metric_reader)

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

@@ -15,5 +15,4 @@
 class SdkConfiguration:
     exemplar_filter: "opentelemetry.sdk.metrics.ExemplarFilter"
     resource: "opentelemetry.sdk.resources.Resource"
-    metric_readers: Sequence["opentelemetry.sdk.metrics.export.MetricReader"]
     views: Sequence["opentelemetry.sdk.metrics.view.View"]

opentelemetry-sdk/tests/_configuration/test_meter_provider.py

@@ -108,7 +108,7 @@ def test_none_config_uses_supplied_resource(self):
 
     def test_none_config_no_readers(self):
         provider = create_meter_provider(None)
-        self.assertEqual(len(provider._sdk_config.metric_readers), 0)
+        self.assertEqual(len(provider._metric_readers), 0)
 
     def test_none_config_uses_trace_based_exemplar_filter(self):
         provider = create_meter_provider(None)
@@ -139,7 +139,7 @@ def test_none_config_does_not_read_interval_env_var(self):
         )
         with patch.dict(os.environ, {"OTEL_METRIC_EXPORT_INTERVAL": "999999"}):
             provider = create_meter_provider(config)
-        reader = provider._sdk_config.metric_readers[0]
+        reader = provider._metric_readers[0]
         self.assertIsInstance(reader, PeriodicExportingMetricReader)
         self.assertEqual(reader._export_interval_millis, 60000.0)
 
@@ -163,11 +163,6 @@ def test_configure_with_config_sets_global(self):
             arg = mock_set.call_args[0][0]
             self.assertIsInstance(arg, MeterProvider)
 
-    def test_empty_readers_list(self):
-        config = MeterProviderConfig(readers=[])
-        provider = create_meter_provider(config)
-        self.assertEqual(len(provider._sdk_config.metric_readers), 0)
-
 
 class TestCreateMetricReaders(unittest.TestCase):
     @staticmethod
@@ -189,7 +184,7 @@ def test_console_exporter(self):
             PushMetricExporterConfig(console=ConsoleMetricExporterConfig())
         )
         provider = create_meter_provider(config)
-        reader = provider._sdk_config.metric_readers[0]
+        reader = provider._metric_readers[0]
         self.assertIsInstance(reader, PeriodicExportingMetricReader)
         self.assertIsInstance(reader._exporter, ConsoleMetricExporter)
 
@@ -198,15 +193,15 @@ def test_periodic_reader_default_interval(self):
             PushMetricExporterConfig(console=ConsoleMetricExporterConfig())
         )
         provider = create_meter_provider(config)
-        reader = provider._sdk_config.metric_readers[0]
+        reader = provider._metric_readers[0]
         self.assertEqual(reader._export_interval_millis, 60000.0)
 
     def test_periodic_reader_default_timeout(self):
         config = self._make_periodic_config(
             PushMetricExporterConfig(console=ConsoleMetricExporterConfig())
         )
         provider = create_meter_provider(config)
-        reader = provider._sdk_config.metric_readers[0]
+        reader = provider._metric_readers[0]
         self.assertEqual(reader._export_timeout_millis, 30000.0)
 
     def test_periodic_reader_explicit_interval(self):
@@ -215,7 +210,7 @@ def test_periodic_reader_explicit_interval(self):
             interval=5000,
         )
         provider = create_meter_provider(config)
-        reader = provider._sdk_config.metric_readers[0]
+        reader = provider._metric_readers[0]
         self.assertEqual(reader._export_interval_millis, 5000.0)
 
     def test_periodic_reader_explicit_timeout(self):
@@ -224,7 +219,7 @@ def test_periodic_reader_explicit_timeout(self):
             timeout=10000,
         )
         provider = create_meter_provider(config)
-        reader = provider._sdk_config.metric_readers[0]
+        reader = provider._metric_readers[0]
         self.assertEqual(reader._export_timeout_millis, 10000.0)
 
     def test_otlp_http_missing_package_raises(self):
@@ -347,7 +342,7 @@ def test_pull_prometheus_creates_reader(self):
 
         mock_reader_cls.assert_called_once_with(disable_target_info=True)
         mock_start_server.assert_called_once_with(port=9090, addr="0.0.0.0")
-        self.assertEqual(len(provider._sdk_config.metric_readers), 1)
+        self.assertEqual(len(provider._metric_readers), 1)
 
     def test_pull_prometheus_defaults(self):
         mock_reader_cls = MagicMock()
@@ -375,7 +370,7 @@ def test_pull_prometheus_defaults(self):
 
         mock_reader_cls.assert_called_once_with(disable_target_info=False)
         mock_start_server.assert_called_once_with(port=9464, addr="localhost")
-        self.assertEqual(len(provider._sdk_config.metric_readers), 1)
+        self.assertEqual(len(provider._metric_readers), 1)
 
     def test_pull_prometheus_missing_package_raises(self):
         with patch.dict(
@@ -432,7 +427,7 @@ def test_pull_plugin_loads_via_entry_point(self):
                 ]
             )
             provider = create_meter_provider(config)
-        self.assertEqual(len(provider._sdk_config.metric_readers), 1)
+        self.assertEqual(len(provider._metric_readers), 1)
         mock_class.assert_called_once_with(port=8080)
         mock_entry_points.assert_called_once_with(
             group="opentelemetry_pull_metric_exporter",
@@ -547,7 +542,7 @@ def test_plugin_metric_exporter_loaded_via_entry_point(self):
                 PushMetricExporterConfig(my_custom_exporter={})
             )
             provider = create_meter_provider(config)
-        self.assertEqual(len(provider._sdk_config.metric_readers), 1)
+        self.assertEqual(len(provider._metric_readers), 1)
 
     def test_unknown_metric_exporter_raises_configuration_error(self):
         with patch(
@@ -581,7 +576,7 @@ def test_multiple_readers(self):
             ]
         )
         provider = create_meter_provider(config)
-        self.assertEqual(len(provider._sdk_config.metric_readers), 2)
+        self.assertEqual(len(provider._metric_readers), 2)
 
 
 class TestTemporalityAndAggregation(unittest.TestCase):
@@ -605,7 +600,7 @@ def _make_console_config(temporality=None, histogram_agg=None):
     @staticmethod
     def _get_exporter(config):
         provider = create_meter_provider(config)
-        return provider._sdk_config.metric_readers[0]._exporter
+        return provider._metric_readers[0]._exporter
 
     def test_default_temporality_is_cumulative(self):
         exporter = self._get_exporter(self._make_console_config())