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

Author: JP-MY

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

@@ -501,11 +501,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"
@@ -520,7 +521,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
@@ -571,7 +572,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:
@@ -616,7 +617,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:
@@ -686,3 +687,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/measurement_consumer.py

@@ -14,15 +14,15 @@
 
 # pylint: disable=unused-import
 
+import weakref
 from abc import ABC, abstractmethod
 from threading import Lock
 from time import time_ns
-from typing import List, Mapping, Optional
+from typing import Iterable, List, Mapping, Optional
 
 # 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
@@ -59,10 +59,10 @@ class SynchronousMeasurementConsumer(MeasurementConsumer):
     def __init__(
         self,
         sdk_config: "opentelemetry.sdk.metrics._internal.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
         ] = {
@@ -71,7 +71,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
@@ -86,7 +86,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
             )
@@ -143,3 +145,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

@@ -26,5 +26,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

@@ -110,7 +110,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)
@@ -141,7 +141,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)
 
@@ -165,11 +165,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
@@ -191,7 +186,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)
 
@@ -200,15 +195,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):
@@ -217,7 +212,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):
@@ -226,7 +221,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):
@@ -326,7 +321,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):
@@ -350,7 +345,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())