opentelemetry-sdk: make methods on FixedSizeExemplarReservoirABC thread safe

.changelog/5437.fixed

@@ -0,0 +1 @@
+`opentelemetry-sdk`: make methods on `FixedSizeExemplarReservoirABC` thread safe

opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/exemplar/exemplar_reservoir.py

@@ -5,6 +5,7 @@
 from collections import defaultdict
 from collections.abc import Callable, Mapping, Sequence
 from random import randrange
+from threading import Lock
 from typing import (
     Any,
 )
@@ -25,6 +26,11 @@ class ExemplarReservoir(ABC):
         The constructor MUST accept ``**kwargs`` that may be set from aggregation
         parameters.
 
+    Note:
+        All methods MUST be safe to call concurrently. Measurements are offered
+        from application threads while a metric reader collects from its own
+        thread.
+
     Reference:
         https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/sdk.md#exemplarreservoir
     """
@@ -151,6 +157,7 @@ def __init__(self, size: int, **kwargs) -> None:
         self._reservoir_storage: Mapping[int, ExemplarBucket] = defaultdict(
             ExemplarBucket
         )
+        self._lock = Lock()
 
     def collect(self, point_attributes: Attributes) -> list[Exemplar]:
         """Returns accumulated Exemplars and also resets the reservoir for the next
@@ -164,15 +171,16 @@ def collect(self, point_attributes: Attributes) -> list[Exemplar]:
             exemplars contain the attributes that were filtered out by the aggregator,
             but recorded alongside the original measurement.
         """
-        exemplars = [
-            e
-            for e in (
-                bucket.collect(point_attributes)
-                for _, bucket in sorted(self._reservoir_storage.items())
-            )
-            if e is not None
-        ]
-        self._reset()
+        with self._lock:
+            exemplars = [
+                e
+                for e in (
+                    bucket.collect(point_attributes)
+                    for _, bucket in sorted(self._reservoir_storage.items())
+                )
+                if e is not None
+            ]
+            self._reset()
         return exemplars
 
     def offer(
@@ -190,17 +198,18 @@ def offer(
             attributes: Measurement attributes
             context: Measurement context
         """
-        try:
-            index = self._find_bucket_index(
-                value, time_unix_nano, attributes, context
-            )
-
-            self._reservoir_storage[index].offer(
-                value, time_unix_nano, attributes, context
-            )
-        except BucketIndexError:
-            # Ignore invalid bucket index
-            pass
+        with self._lock:
+            try:
+                index = self._find_bucket_index(
+                    value, time_unix_nano, attributes, context
+                )
+
+                self._reservoir_storage[index].offer(
+                    value, time_unix_nano, attributes, context
+                )
+            except BucketIndexError:
+                # Ignore invalid bucket index
+                pass
 
     @abstractmethod
     def _find_bucket_index(
@@ -279,21 +288,6 @@ def __init__(self, boundaries: Sequence[float], **kwargs) -> None:
         super().__init__(len(boundaries) + 1, **kwargs)
         self._boundaries: Sequence[float] = boundaries
 
-    def offer(
-        self,
-        value: int | float,
-        time_unix_nano: int,
-        attributes: Attributes,
-        context: Context,
-    ) -> None:
-        """Offers a measurement to be sampled."""
-        index = self._find_bucket_index(
-            value, time_unix_nano, attributes, context
-        )
-        self._reservoir_storage[index].offer(
-            value, time_unix_nano, attributes, context
-        )
-
     def _find_bucket_index(
         self,
         value: int | float,

opentelemetry-sdk/tests/metrics/test_exemplarreservoir.py

@@ -1,6 +1,7 @@
 # Copyright The OpenTelemetry Authors
 # SPDX-License-Identifier: Apache-2.0
 
+from itertools import chain, count, islice
 from time import time_ns
 from unittest import TestCase
 
@@ -16,6 +17,7 @@
     SimpleFixedSizeExemplarReservoir,
 )
 from opentelemetry.sdk.metrics._internal.view import _default_reservoir_factory
+from opentelemetry.test.concurrency_test import ConcurrencyTestBase
 from opentelemetry.trace import SpanContext, TraceFlags
 
 
@@ -159,3 +161,52 @@ def test_explicit_histogram_aggregation(self):
         self.assertEqual(
             exemplar_reservoir, AlignedHistogramBucketExemplarReservoir
         )
+
+
+class TestExemplarReservoirConcurrency(ConcurrencyTestBase):
+    """Every reservoir method must be safe to call concurrently: measurements
+    are offered from application threads while a metric reader collects from
+    its own thread.
+    """
+
+    NUM_THREADS = 50
+    ITERATIONS = 200
+
+    def _run_concurrently(self, reservoir):
+        threads = count()
+        values = count(1)
+
+        def worker():
+            if next(threads) % 2:
+                return [
+                    exemplar
+                    for _ in range(self.ITERATIONS)
+                    for exemplar in reservoir.collect({})
+                ]
+
+            for value in islice(values, self.ITERATIONS):
+                reservoir.offer(value, value, {"v": value}, Context())
+            return []
+
+        collected = self.run_with_many_threads(worker, self.NUM_THREADS)
+        return list(chain(*collected, reservoir.collect({})))
+
+    def test_offer_and_collect_are_mutually_exclusive(self):
+        for name, build_reservoir in (
+            (
+                "simple_fixed_size",
+                lambda: SimpleFixedSizeExemplarReservoir(size=4),
+            ),
+            (
+                "aligned_histogram_bucket",
+                lambda: AlignedHistogramBucketExemplarReservoir(
+                    [10.0, 20.0, 30.0]
+                ),
+            ),
+        ):
+            with self.subTest(reservoir=name):
+                for exemplar in self._run_concurrently(build_reservoir()):
+                    self.assertEqual(exemplar.value, exemplar.time_unix_nano)
+                    self.assertEqual(
+                        exemplar.filtered_attributes, {"v": exemplar.value}
+                    )