Updates to ensure consistent updates to all metrics readers

Author: JP-MY

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

@@ -584,12 +584,27 @@ def get_meter(
     def add_metric_reader(
         self, metric_reader: "opentelemetry.sdk.metrics.export.MetricReader"
     ) -> None:
-        with self._lock:
+        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._lock:
+        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

@@ -15,7 +15,6 @@
 # pylint: disable=unused-import
 
 from abc import ABC, abstractmethod
-from logging import getLogger
 from threading import Lock
 from time import time_ns
 from typing import Iterable, List, Mapping, Optional
@@ -32,8 +31,6 @@
 )
 from opentelemetry.sdk.metrics._internal.point import Metric
 
-_logger = getLogger(__name__)
-
 
 class MeasurementConsumer(ABC):
     @abstractmethod
@@ -152,8 +149,6 @@ def add_metric_reader(
     ) -> None:
         """Registers a new metric reader."""
         with self._lock:
-            if metric_reader in self._reader_storages:
-                _logger.warning("'%s' already registered!", metric_reader)
             self._sdk_config.metric_readers += type(
                 self._sdk_config.metric_readers
             )((metric_reader,))
@@ -162,17 +157,13 @@ def add_metric_reader(
                 metric_reader._instrument_class_temporality,
                 metric_reader._instrument_class_aggregation,
             )
-            metric_reader._set_collect_callback(self.collect)
 
     def remove_metric_reader(
         self, metric_reader: "opentelemetry.sdk.metrics.MetricReader"
     ) -> None:
         """Unregisters the given metric reader."""
         with self._lock:
-            if metric_reader not in self._reader_storages:
-                _logger.warning("'%s' has not been registered!", metric_reader)
             self._reader_storages.pop(metric_reader, None)
-            metric_reader._set_collect_callback(None)
             self._sdk_config.metric_readers = type(
                 self._sdk_config.metric_readers
             )(

opentelemetry-sdk/tests/metrics/test_metrics.py

@@ -442,23 +442,19 @@ def test_addition_of_metric_reader(self):
             counter.add(1)
             self.assertIsNotNone(reader.get_metrics_data())
 
-            with self.assertLogs(
-                "opentelemetry.sdk.metrics._internal.measurement_consumer",
-                WARNING,
-            ) as logger:
+            with self.assertRaises(ValueError) as cm:
                 meter_provider.add_metric_reader(reader)
-                self.assertIn("already registered!", logger.output[0])
+                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.assertLogs(
-                "opentelemetry.sdk.metrics._internal.measurement_consumer",
-                WARNING,
-            ) as logger:
+            with self.assertRaises(ValueError) as cm:
                 meter_provider.remove_metric_reader(reader)
-                self.assertIn("has not been registered!", logger.output[0])
+                self.assertIn("has not been registered!", str(cm.exception))
 
 
 class TestMeter(TestCase):