diff --git a/.changelog/5437.fixed b/.changelog/5437.fixed new file mode 100644 index 00000000000..c916f57fb72 --- /dev/null +++ b/.changelog/5437.fixed @@ -0,0 +1 @@ +`opentelemetry-sdk`: make methods on `FixedSizeExemplarReservoirABC` thread safe diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/exemplar/exemplar_reservoir.py b/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/exemplar/exemplar_reservoir.py index afe2dcba38a..858dc0aa7b6 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/exemplar/exemplar_reservoir.py +++ b/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, diff --git a/opentelemetry-sdk/tests/metrics/test_exemplarreservoir.py b/opentelemetry-sdk/tests/metrics/test_exemplarreservoir.py index 9c5bf69bdb3..ce196f68306 100644 --- a/opentelemetry-sdk/tests/metrics/test_exemplarreservoir.py +++ b/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} + )