From 98ead9612c0074a4f7aab56820ee30ca36a5729f Mon Sep 17 00:00:00 2001 From: Diego Hurtado Date: Sun, 19 Jul 2026 12:16:44 -0500 Subject: [PATCH 1/3] Serialize SimpleSpanProcessor export and enforce BSP export timeout SimpleSpanProcessor.on_end previously called span_exporter.export with no synchronization, so two threads ending sampled spans could invoke export concurrently, which the specification forbids. Guard the export call with a per-processor lock so concurrent on_end calls are serialized. The batch processor read OTEL_BSP_EXPORT_TIMEOUT into configuration but never applied it, so a hung exporter could block the worker thread and force_flush indefinitely. Run each export on a dedicated single-worker executor and await it with the configured timeout as a deadline; on timeout the batch worker stops waiting, records the failure, and moves on instead of blocking forever. A non-positive timeout disables the deadline. This affects both BatchSpanProcessor and BatchLogRecordProcessor via the shared BatchProcessor. Add tests: SimpleSpanProcessor concurrent on_end never overlaps export, and a hung batch export is bounded by the export timeout. --- .changelog/4555.fixed | 3 + .changelog/4556.fixed | 3 + .../sdk/_shared_internal/__init__.py | 71 ++++++++++++++++++- .../sdk/trace/export/__init__.py | 12 +++- .../shared_internal/test_batch_processor.py | 46 ++++++++++++ .../tests/trace/export/test_export.py | 60 ++++++++++++++++ 6 files changed, 190 insertions(+), 5 deletions(-) create mode 100644 .changelog/4555.fixed create mode 100644 .changelog/4556.fixed diff --git a/.changelog/4555.fixed b/.changelog/4555.fixed new file mode 100644 index 0000000000..a28f440aba --- /dev/null +++ b/.changelog/4555.fixed @@ -0,0 +1,3 @@ +`BatchSpanProcessor`/`BatchLogRecordProcessor`: enforce `OTEL_BSP_EXPORT_TIMEOUT` +as a real deadline on the export call so a hung exporter can no longer block the +batch worker (and `force_flush`) indefinitely diff --git a/.changelog/4556.fixed b/.changelog/4556.fixed new file mode 100644 index 0000000000..9b94c5d522 --- /dev/null +++ b/.changelog/4556.fixed @@ -0,0 +1,3 @@ +`SimpleSpanProcessor`: serialize calls to the exporter with a lock so concurrent +`on_end` calls can no longer invoke `SpanExporter.export` concurrently, as +required by the specification diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py index 976cd38c20..690af51287 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py @@ -12,6 +12,8 @@ import time import weakref from abc import abstractmethod +from concurrent.futures import ThreadPoolExecutor +from concurrent.futures import TimeoutError as FutureTimeoutError from typing import ( Generic, Protocol, @@ -97,9 +99,21 @@ def __init__( self._schedule_delay_millis = schedule_delay_millis self._schedule_delay = schedule_delay_millis / 1e3 self._max_export_batch_size = max_export_batch_size - # Not used. No way currently to pass timeout to export. - # TODO(https://github.com/open-telemetry/opentelemetry-python/issues/4555): figure out what this should do. + # The maximum time a single export call is allowed to take. Exporters + # are synchronous and Python offers no way to cancel a running call, so + # the export is run on a dedicated single worker thread and awaited with + # this deadline. If the deadline is exceeded the batch worker stops + # waiting and moves on instead of blocking forever. See _run_export for + # the tradeoffs of this approach. self._export_timeout_millis = export_timeout_millis + self._export_timeout = export_timeout_millis / 1e3 + # Single worker so exports are never run concurrently (the spec forbids + # concurrent Export calls for the same exporter); the _export_lock + # further serializes callers into this executor. + self._export_executor = ThreadPoolExecutor( + max_workers=1, + thread_name_prefix=f"OtelBatch{exporting}Export", + ) # Deque is thread safe. self._queue = collections.deque([], max_queue_size) self._worker_thread = threading.Thread( @@ -138,6 +152,12 @@ def _at_fork_reinit(self): self._export_lock = threading.Lock() self._worker_awaken = threading.Event() self._queue.clear() + # The executor's worker thread does not survive a fork, so replace it + # with a fresh one in the child process. + self._export_executor = ThreadPoolExecutor( + max_workers=1, + thread_name_prefix=f"OtelBatch{self._exporting}Export", + ) self._worker_thread = threading.Thread( name=f"OtelBatch{self._exporting}RecordProcessor", target=self.worker, @@ -178,11 +198,51 @@ def _export(self, batch_strategy: BatchExportStrategy) -> None: batch = [self._queue.pop() for _ in range(count)] # Record on submission to the exporter. self._metrics.finish_items(count) + export_timed_out = False try: - self._exporter.export(batch) + self._run_export(batch) + except FutureTimeoutError: + export_timed_out = True + _logger.warning( + "Timed out (after %sms) while exporting %s. Export was " + "abandoned; the export may still be running in the " + "background.", + self._export_timeout_millis, + self._exporting, + ) except Exception: # pylint: disable=broad-exception-caught _logger.exception("Exception while exporting %s.", self._exporting) detach(token) + # If an export timed out we stop draining the queue for this + # cycle instead of piling more work onto a stuck exporter. + if export_timed_out: + break + + def _run_export(self, batch: list[Telemetry]) -> None: + """Run a single export call bounded by the configured timeout. + + The export is submitted to a dedicated single-threaded executor and + awaited with the export timeout as a deadline. This guarantees the + batch worker (and callers of force_flush) cannot be blocked + indefinitely by a hung exporter, satisfying the spec requirement that + Export MUST NOT block indefinitely. + + A non-positive timeout is treated as "no deadline" (block until the + export returns), matching the previous behaviour and Go/Java where a + zero timeout disables the deadline. + + Raises ``concurrent.futures.TimeoutError`` if the export does not + complete within the deadline. On timeout the export call itself is + abandoned but keeps running in the background on the executor thread + (Python offers no way to cancel a running synchronous call); the next + export will queue behind it on the single executor thread, which + naturally back-pressures against a permanently stuck exporter. + """ + timeout = self._export_timeout if self._export_timeout > 0 else None + future = self._export_executor.submit(self._exporter.export, batch) + # result() re-raises any exception from the export call, and raises + # FutureTimeoutError if the deadline is exceeded. + future.result(timeout=timeout) def emit(self, data: Telemetry) -> None: if self._shutdown: @@ -222,6 +282,11 @@ def shutdown(self, timeout_millis: int = 30000): # and set shutdown_is_occuring to prevent further export calls. It's possible that a single export # call is ongoing and the thread isn't finished. In this case we will return instead of waiting on # the thread to finish. + # Release the export executor without waiting: if an export is hung we + # must not block shutdown on it (that is the whole point of the export + # timeout). Pending-but-unstarted work is cancelled; a running export is + # left to finish (or leak) on its daemon-like executor thread. + self._export_executor.shutdown(wait=False, cancel_futures=True) # TODO: Fix force flush so the timeout is used https://github.com/open-telemetry/opentelemetry-python/issues/4568. def force_flush(self, timeout_millis: int | None = None) -> bool: diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py index 525334173d..97f71e680b 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py @@ -5,6 +5,7 @@ import collections.abc import logging import sys +import threading import typing from enum import Enum from os import environ, linesep @@ -100,6 +101,11 @@ def __init__( ): self.span_exporter = span_exporter self._shutdown = False + # Serializes calls to the exporter. The spec requires that Export MUST + # NOT be called concurrently for the same exporter, so concurrent + # on_end calls (e.g. two threads ending sampled spans) must not invoke + # span_exporter.export at the same time. + self._export_lock = threading.Lock() self._metrics = create_processor_metrics( "traces", OtelComponentTypeValues.SIMPLE_SPAN_PROCESSOR, @@ -124,7 +130,10 @@ def on_end(self, span: ReadableSpan) -> None: # Record on submission to the exporter. self._metrics.finish_items(1) try: - self.span_exporter.export((span,)) + # Hold the lock across the export call so that concurrent on_end + # calls cannot invoke the exporter concurrently. + with self._export_lock: + self.span_exporter.export((span,)) # pylint: disable=broad-exception-caught except Exception: logger.exception("Exception while exporting Span.") @@ -176,7 +185,6 @@ def __init__( if max_export_batch_size is None: max_export_batch_size = BatchSpanProcessor._default_max_export_batch_size() - # Not used. No way currently to pass timeout to export. if export_timeout_millis is None: export_timeout_millis = BatchSpanProcessor._default_export_timeout_millis() diff --git a/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py b/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py index 0b287d6e9f..4453e9a5d3 100644 --- a/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py +++ b/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py @@ -230,6 +230,52 @@ def test_shutdown_allows_1_export_to_finish(self, batch_processor_class, telemet assert exporter.sleep_interrupted is True assert 2 == exporter.num_export_calls + def test_hung_export_is_bounded_by_export_timeout( + self, batch_processor_class, telemetry + ): + # An exporter whose export() call would hang effectively forever. + export_released = threading.Event() + + class HangingExporter: + def __init__(self): + self.export_started = threading.Event() + + def export(self, _): + self.export_started.set() + # Would block ~forever if the deadline were not enforced. + export_released.wait(60) + + def shutdown(self): + export_released.set() + + exporter = HangingExporter() + processor = batch_processor_class( + exporter, + max_queue_size=15, + max_export_batch_size=15, + schedule_delay_millis=30000, + # Short deadline: a hung export must be abandoned after this. + export_timeout_millis=200, + ) + try: + processor._batch_processor.emit(telemetry) + before = time.time() + # force_flush triggers _export; the hung export must not block it + # for longer than the configured timeout (plus a small margin). + processor.force_flush() + elapsed = time.time() - before + # The export was actually entered. + assert exporter.export_started.is_set() + # Bounded, not infinite: well under the 60s the exporter would hang. + assert elapsed < 5, ( + f"force_flush blocked for {elapsed}s despite a 200ms export " + "timeout" + ) + finally: + # Release the hung export thread so the process can exit cleanly. + export_released.set() + processor.shutdown() + class TestCommonFuncs(unittest.TestCase): def test_duplicate_logs_filter_works(self): diff --git a/opentelemetry-sdk/tests/trace/export/test_export.py b/opentelemetry-sdk/tests/trace/export/test_export.py index 6f7a303ae0..b9f46e1d9f 100644 --- a/opentelemetry-sdk/tests/trace/export/test_export.py +++ b/opentelemetry-sdk/tests/trace/export/test_export.py @@ -129,6 +129,66 @@ def test_simple_span_processor_not_sampled(self): self.assertListEqual([], spans_names_list) + def test_export_is_serialized_across_concurrent_on_end(self): + """Concurrent on_end calls must never invoke export() concurrently. + + The spec requires that Export MUST NOT be called concurrently for the + same exporter. This exporter records whether it is ever entered by more + than one thread at a time. + """ + + class ConcurrencyDetectingExporter(export.SpanExporter): + def __init__(self): + self._active = 0 + self._active_lock = threading.Lock() + self.max_concurrency = 0 + self.export_count = 0 + + def export(self, spans): + with self._active_lock: + self._active += 1 + self.export_count += 1 + self.max_concurrency = max( + self.max_concurrency, self._active + ) + # Sleep outside the lock so genuine overlap would be observed + # by another thread bumping _active before we decrement. + time.sleep(0.005) + with self._active_lock: + self._active -= 1 + return export.SpanExportResult.SUCCESS + + def shutdown(self): + pass + + exporter = ConcurrencyDetectingExporter() + span_processor = export.SimpleSpanProcessor(exporter) + tracer_provider = trace.TracerProvider() + tracer_provider.add_span_processor(span_processor) + tracer = tracer_provider.get_tracer(__name__) + + num_threads = 8 + spans_per_thread = 5 + barrier = threading.Barrier(num_threads) + + def end_spans(): + barrier.wait() + for _ in range(spans_per_thread): + with tracer.start_as_current_span("concurrent"): + pass + + threads = [ + threading.Thread(target=end_spans) for _ in range(num_threads) + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + span_processor.shutdown() + self.assertEqual(exporter.export_count, num_threads * spans_per_thread) + self.assertEqual(exporter.max_concurrency, 1) + @mock.patch.dict("os.environ", {OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED: "true"}) def test_metrics(self): metric_reader = InMemoryMetricReader() From dc5c58a89fb27e169eddc50ec6e3db18b788ba11 Mon Sep 17 00:00:00 2001 From: Diego Hurtado Date: Sun, 19 Jul 2026 12:21:53 -0500 Subject: [PATCH 2/3] Preserve instrumentation suppression across the export executor thread The batch export now runs on a separate ThreadPoolExecutor thread, but ThreadPoolExecutor.submit does not copy the caller's contextvars into the worker thread. As a result the _SUPPRESS_INSTRUMENTATION_KEY attached by the batch worker was not visible to the exporter, so an auto-instrumented transport used by the exporter could emit telemetry that feeds back into the processor. Capture the current context on the worker thread and run the export inside it via contextvars.Context.run, restoring suppression during export. Add a regression test asserting the exporter observes _SUPPRESS_INSTRUMENTATION_KEY as True when driven through the BatchProcessor. --- .../sdk/_shared_internal/__init__.py | 13 ++++++- .../shared_internal/test_batch_processor.py | 39 +++++++++++++++++++ 2 files changed, 51 insertions(+), 1 deletion(-) diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py index 690af51287..80da70ab97 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py @@ -4,6 +4,7 @@ from __future__ import annotations import collections +import contextvars import enum import inspect import logging @@ -239,7 +240,17 @@ def _run_export(self, batch: list[Telemetry]) -> None: naturally back-pressures against a permanently stuck exporter. """ timeout = self._export_timeout if self._export_timeout > 0 else None - future = self._export_executor.submit(self._exporter.export, batch) + # The export runs on a separate executor thread, and + # ThreadPoolExecutor does not copy the caller's contextvars into it. + # Capture the current context here (on the caller/worker thread, where + # _SUPPRESS_INSTRUMENTATION_KEY has been attached) and run the export + # inside it, so instrumentation suppression is preserved during export + # and the exporter's own network calls do not generate telemetry that + # would feed back into the processor. + ctx = contextvars.copy_context() + future = self._export_executor.submit( + ctx.run, self._exporter.export, batch + ) # result() re-raises any exception from the export call, and raises # FutureTimeoutError if the deadline is exceeded. future.result(timeout=timeout) diff --git a/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py b/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py index 4453e9a5d3..6f0df54048 100644 --- a/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py +++ b/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py @@ -19,6 +19,10 @@ from opentelemetry._logs import ( LogRecord, ) +from opentelemetry.context import ( + _SUPPRESS_INSTRUMENTATION_KEY, + get_value, +) from opentelemetry.sdk._logs import ( ReadWriteLogRecord, ) @@ -276,6 +280,41 @@ def shutdown(self): export_released.set() processor.shutdown() + def test_export_runs_with_instrumentation_suppressed( + self, batch_processor_class, telemetry + ): + # The export now runs on a separate executor thread. The batch worker + # attaches _SUPPRESS_INSTRUMENTATION_KEY before exporting; the exporter + # must still observe it as True (otherwise the exporter's own network + # calls would be instrumented and fed back into the processor). + + class SuppressionRecordingExporter: + def __init__(self): + self.suppressed_during_export = None + + def export(self, _): + self.suppressed_during_export = get_value( + _SUPPRESS_INSTRUMENTATION_KEY + ) + + def shutdown(self): + pass + + exporter = SuppressionRecordingExporter() + processor = batch_processor_class( + exporter, + max_queue_size=15, + max_export_batch_size=15, + schedule_delay_millis=30000, + export_timeout_millis=500, + ) + try: + processor._batch_processor.emit(telemetry) + processor.force_flush() + assert exporter.suppressed_during_export is True + finally: + processor.shutdown() + class TestCommonFuncs(unittest.TestCase): def test_duplicate_logs_filter_works(self): From 78d08d14f92edecc79bb871808794b28d05368b0 Mon Sep 17 00:00:00 2001 From: Diego Hurtado Date: Wed, 22 Jul 2026 08:37:36 -0500 Subject: [PATCH 3/3] Rename changelog fragment to match PR number --- .changelog/{4555.fixed => 15.fixed} | 3 +++ .changelog/4556.fixed | 3 --- .../sdk/_shared_internal/__init__.py | 4 +--- .../shared_internal/test_batch_processor.py | 17 ++++------------- .../tests/trace/export/test_export.py | 8 ++------ 5 files changed, 10 insertions(+), 25 deletions(-) rename .changelog/{4555.fixed => 15.fixed} (52%) delete mode 100644 .changelog/4556.fixed diff --git a/.changelog/4555.fixed b/.changelog/15.fixed similarity index 52% rename from .changelog/4555.fixed rename to .changelog/15.fixed index a28f440aba..d94fe66dc7 100644 --- a/.changelog/4555.fixed +++ b/.changelog/15.fixed @@ -1,3 +1,6 @@ `BatchSpanProcessor`/`BatchLogRecordProcessor`: enforce `OTEL_BSP_EXPORT_TIMEOUT` as a real deadline on the export call so a hung exporter can no longer block the batch worker (and `force_flush`) indefinitely +`SimpleSpanProcessor`: serialize calls to the exporter with a lock so concurrent +`on_end` calls can no longer invoke `SpanExporter.export` concurrently, as +required by the specification diff --git a/.changelog/4556.fixed b/.changelog/4556.fixed deleted file mode 100644 index 9b94c5d522..0000000000 --- a/.changelog/4556.fixed +++ /dev/null @@ -1,3 +0,0 @@ -`SimpleSpanProcessor`: serialize calls to the exporter with a lock so concurrent -`on_end` calls can no longer invoke `SpanExporter.export` concurrently, as -required by the specification diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py index 80da70ab97..dc9bf2339c 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py @@ -248,9 +248,7 @@ def _run_export(self, batch: list[Telemetry]) -> None: # and the exporter's own network calls do not generate telemetry that # would feed back into the processor. ctx = contextvars.copy_context() - future = self._export_executor.submit( - ctx.run, self._exporter.export, batch - ) + future = self._export_executor.submit(ctx.run, self._exporter.export, batch) # result() re-raises any exception from the export call, and raises # FutureTimeoutError if the deadline is exceeded. future.result(timeout=timeout) diff --git a/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py b/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py index 6f0df54048..aef93507d3 100644 --- a/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py +++ b/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py @@ -234,9 +234,7 @@ def test_shutdown_allows_1_export_to_finish(self, batch_processor_class, telemet assert exporter.sleep_interrupted is True assert 2 == exporter.num_export_calls - def test_hung_export_is_bounded_by_export_timeout( - self, batch_processor_class, telemetry - ): + def test_hung_export_is_bounded_by_export_timeout(self, batch_processor_class, telemetry): # An exporter whose export() call would hang effectively forever. export_released = threading.Event() @@ -271,18 +269,13 @@ def shutdown(self): # The export was actually entered. assert exporter.export_started.is_set() # Bounded, not infinite: well under the 60s the exporter would hang. - assert elapsed < 5, ( - f"force_flush blocked for {elapsed}s despite a 200ms export " - "timeout" - ) + assert elapsed < 5, f"force_flush blocked for {elapsed}s despite a 200ms export timeout" finally: # Release the hung export thread so the process can exit cleanly. export_released.set() processor.shutdown() - def test_export_runs_with_instrumentation_suppressed( - self, batch_processor_class, telemetry - ): + def test_export_runs_with_instrumentation_suppressed(self, batch_processor_class, telemetry): # The export now runs on a separate executor thread. The batch worker # attaches _SUPPRESS_INSTRUMENTATION_KEY before exporting; the exporter # must still observe it as True (otherwise the exporter's own network @@ -293,9 +286,7 @@ def __init__(self): self.suppressed_during_export = None def export(self, _): - self.suppressed_during_export = get_value( - _SUPPRESS_INSTRUMENTATION_KEY - ) + self.suppressed_during_export = get_value(_SUPPRESS_INSTRUMENTATION_KEY) def shutdown(self): pass diff --git a/opentelemetry-sdk/tests/trace/export/test_export.py b/opentelemetry-sdk/tests/trace/export/test_export.py index b9f46e1d9f..d64ee494e7 100644 --- a/opentelemetry-sdk/tests/trace/export/test_export.py +++ b/opentelemetry-sdk/tests/trace/export/test_export.py @@ -148,9 +148,7 @@ def export(self, spans): with self._active_lock: self._active += 1 self.export_count += 1 - self.max_concurrency = max( - self.max_concurrency, self._active - ) + self.max_concurrency = max(self.max_concurrency, self._active) # Sleep outside the lock so genuine overlap would be observed # by another thread bumping _active before we decrement. time.sleep(0.005) @@ -177,9 +175,7 @@ def end_spans(): with tracer.start_as_current_span("concurrent"): pass - threads = [ - threading.Thread(target=end_spans) for _ in range(num_threads) - ] + threads = [threading.Thread(target=end_spans) for _ in range(num_threads)] for thread in threads: thread.start() for thread in threads: