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

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

Update measurement_consumer.py

Removed return values

Author: JP-MY

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

@@ -580,3 +580,31 @@ def get_meter(
                     self._measurement_consumer,
                 )
             return self._meters[info]
+
+    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:
+                raise ValueError(
+                    f"MetricReader {metric_reader} has been registered already!"
+                )
+            self._measurement_consumer.add_metric_reader(metric_reader)
+            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:
+                raise ValueError(
+                    f"MetricReader {metric_reader} has not been registered!"
+                )
+            self._measurement_consumer.remove_metric_reader(metric_reader)
+            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

@@ -143,3 +143,33 @@ 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._sdk_config.metric_readers += type(
+                self._sdk_config.metric_readers
+            )((metric_reader,))
+            self._reader_storages[metric_reader] = MetricReaderStorage(
+                self._sdk_config,
+                metric_reader._instrument_class_temporality,
+                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, None)
+            self._sdk_config.metric_readers = type(
+                self._sdk_config.metric_readers
+            )(
+                (
+                    reader
+                    for reader in self._sdk_config.metric_readers
+                    if reader is not metric_reader
+                )
+            )

opentelemetry-sdk/tests/metrics/test_metrics.py

@@ -13,9 +13,8 @@
 # limitations under the License.
 
 # pylint: disable=protected-access,no-self-use
-
 import weakref
-from logging import WARNING
+from logging import DEBUG, WARNING
 from time import sleep
 from typing import Iterable, Sequence
 from unittest.mock import MagicMock, Mock, patch
@@ -36,6 +35,7 @@
 )
 from opentelemetry.sdk.metrics._internal import SynchronousMeasurementConsumer
 from opentelemetry.sdk.metrics.export import (
+    InMemoryMetricReader,
     Metric,
     MetricExporter,
     MetricExportResult,
@@ -426,6 +426,36 @@ def test_consume_measurement_gauge(self, mock_sync_measurement_consumer):
 
         sync_consumer_instance.consume_measurement.assert_called()
 
+    def test_addition_of_metric_reader(self):
+        # Suppress warnings for calling collect on an unregistered metric reader
+        with self.assertLogs(
+            "opentelemetry.sdk.metrics._internal.export", DEBUG
+        ):
+            reader = InMemoryMetricReader()
+            meter_provider = MeterProvider()
+            meter = meter_provider.get_meter(__name__)
+            counter = meter.create_counter("counter")
+            counter.add(1)
+            self.assertIsNone(reader.get_metrics_data())
+
+            meter_provider.add_metric_reader(reader)
+            counter.add(1)
+            self.assertIsNotNone(reader.get_metrics_data())
+
+            with self.assertRaises(ValueError) as cm:
+                meter_provider.add_metric_reader(reader)
+                self.assertIn(
+                    "has been registered already!", str(cm.exception)
+                )
+
+            meter_provider.remove_metric_reader(reader)
+            counter.add(1)
+            self.assertIsNone(reader.get_metrics_data())
+
+            with self.assertRaises(ValueError) as cm:
+                meter_provider.remove_metric_reader(reader)
+                self.assertIn("has not been registered!", str(cm.exception))
+
 
 class TestMeter(TestCase):
     def setUp(self):