Addressed PR comments

Author: JP-MY

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

@@ -599,6 +599,7 @@ def add_metric_reader(
                     f"MetricReader {metric_reader} has been registered already!"
                 )
             self._measurement_consumer.add_metric_reader(metric_reader)
+            # pylint: disable-next=protected-access
             metric_reader._set_collect_callback(
                 self._measurement_consumer.collect
             )
@@ -614,6 +615,7 @@ def remove_metric_reader(
                     f"MetricReader {metric_reader} has not been registered!"
                 )
             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,16 +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 weakref import WeakSet
+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
@@ -88,7 +87,7 @@ def consume_measurement(self, measurement: Measurement) -> None:
             )
         )
         with self._lock:
-            reader_storages = WeakSet(self._reader_storages.values())
+            reader_storages = weakref.WeakSet(self._reader_storages.values())
         for reader_storage in reader_storages:
             reader_storage.consume_measurement(
                 measurement, should_sample_exemplar
@@ -154,7 +153,9 @@ def add_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,
             )
 

opentelemetry-sdk/tests/metrics/test_measurement_consumer.py

@@ -14,6 +14,7 @@
 
 # pylint: disable=invalid-name,no-self-use
 
+from threading import Event, Thread
 from time import sleep
 from unittest import TestCase
 from unittest.mock import MagicMock, Mock, patch
@@ -193,3 +194,89 @@ def sleep_1(*args, **kwargs):
             callback_options_time_call,
             10000,
         )
+
+
+class TestSynchronousMeasurementConsumerConcurrency(TestCase):
+    def test_concurrent_changes_to_metric_readers(self):
+        timeout = 1
+        failure = None
+        iteration_started = Event()
+        mutation_done = Event()
+        iteration_timeout_error = "Timed out waiting for iteration to start"
+        mutation_timeout_error = "Timed out waiting for mutation to be done"
+
+        consumer = SynchronousMeasurementConsumer(
+            SdkConfiguration(
+                exemplar_filter=MagicMock(),
+                resource=MagicMock(),
+                views=MagicMock(),
+            ),
+            metric_readers=[MagicMock()],
+        )
+
+        def _hooked_iter(iterable):
+            nonlocal failure
+
+            iterable = iter(iterable)
+            iteration_started.set()
+            if not mutation_done.wait(timeout):
+                failure = mutation_timeout_error
+            yield next(iterable, None)
+            yield from iterable
+
+        class HookedDict(dict):
+            def __iter__(self):
+                return _hooked_iter(super().__iter__())
+
+            def keys(self):
+                return _hooked_iter(super().keys())
+
+            def values(self):
+                return _hooked_iter(super().values())
+
+            def items(self):
+                return _hooked_iter(super().items())
+
+        with patch.object(
+            # pylint: disable-next=protected-access
+            consumer, "_reader_storages", HookedDict(consumer._reader_storages)
+        ):
+
+            def mutate():
+                """Directly mutate _reader_storages after iteration starts"""
+                nonlocal failure
+                if not iteration_started.wait(timeout):
+                    failure = iteration_timeout_error
+                #pylint: disable-next=protected-access
+                consumer._reader_storages.clear()
+
+            # Verify that test setup works (direct mutation with no synchronization fails)
+            with self.assertRaises(RuntimeError) as cm:
+                t = Thread(target=mutate)
+                t.start()
+                try:
+                    consumer.consume_measurement(MagicMock())
+                finally:
+                    t.join()
+            self.assertEqual(
+                "dictionary changed size during iteration", str(cm.exception)
+            )
+
+            def add_and_remove_readers():
+                """Modifies _reader_storages after iteration starts"""
+                nonlocal failure
+                if not iteration_started.wait(timeout):
+                    failure = iteration_timeout_error
+                reader = MagicMock()
+                consumer.add_metric_reader(reader)
+                consumer.remove_metric_reader(reader)
+
+            # Verify the API calls do not attempt concurrent modification of reader storages
+            t = Thread(target=add_and_remove_readers)
+            t.start()
+            try:
+                consumer.add_metric_reader(MagicMock())
+                consumer.consume_measurement(MagicMock())
+            finally:
+                t.join()
+            self.assertEqual(mutation_timeout_error, failure)

opentelemetry-sdk/tests/metrics/test_metric_reader_storage.py

@@ -280,7 +280,6 @@ def test_race_collect_with_new_instruments(self):
             SdkConfiguration(
                 exemplar_filter=Mock(),
                 resource=Mock(),
-                metric_readers=(),
                 views=(View(instrument_name="test"),),
             ),
             MagicMock(

opentelemetry-sdk/tests/metrics/test_metrics.py

@@ -429,34 +429,33 @@ def test_consume_measurement_gauge(self, mock_sync_measurement_consumer):
         sync_consumer_instance.consume_measurement.assert_called()
 
     def test_addition_of_metric_reader(self):
+        logger_name = "opentelemetry.sdk.metrics._internal.export"
+
+        reader = InMemoryMetricReader()
+        meter_provider = MeterProvider()
+        meter = meter_provider.get_meter(__name__)
+        counter = meter.create_counter("counter")
+        counter.add(1)
         # 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)
+        with self.assertLogs(logger_name, DEBUG):
             self.assertIsNone(reader.get_metrics_data())
 
-            meter_provider.add_metric_reader(reader)
-            counter.add(1)
-            self.assertIsNotNone(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)
-                )
+        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)
+        meter_provider.remove_metric_reader(reader)
+        counter.add(1)
+        with self.assertLogs(logger_name, DEBUG):
             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))
+        with self.assertRaises(ValueError) as cm:
+            meter_provider.remove_metric_reader(reader)
+            self.assertIn("has not been registered!", str(cm.exception))
 
 
 class TestMeterConcurrency(ConcurrencyTestBase, TestCase):