oss_14_processor_export_lock_timeout

.changelog/15.fixed

@@ -0,0 +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

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
@@ -12,6 +13,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 +100,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(
@@ -140,6 +155,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,
@@ -179,12 +200,20 @@ def _export(self, batch_strategy: BatchExportStrategy) -> None:
                         self._max_export_batch_size,
                         len(self._queue),
                     )
-                    self._exporter.export(
-                        [
-                            # Oldest records are at the back, so pop from there.
-                            self._queue.pop()
-                            for _ in range(count)
-                        ]
+                    batch = [
+                        # Oldest records are at the back, so pop from there.
+                        self._queue.pop()
+                        for _ in range(count)
+                    ]
+                    self._run_export(batch)
+                except FutureTimeoutError as err:
+                    error = err
+                    _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 as err:  # pylint: disable=broad-exception-caught
                     error = err
@@ -194,6 +223,46 @@ def _export(self, batch_strategy: BatchExportStrategy) -> None:
                 finally:
                     self._metrics.finish_items(count, error)
                 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 isinstance(error, FutureTimeoutError):
+                    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
+        # 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)
 
     def emit(self, data: Telemetry) -> None:
         if self._shutdown:
@@ -238,6 +307,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:

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
@@ -103,6 +104,11 @@ def __init__(
         meter_provider: MeterProvider | None = None,
     ):
         self.span_exporter = span_exporter
+        # 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,
@@ -126,7 +132,10 @@ def on_end(self, span: ReadableSpan) -> None:
         token = attach(set_value(_SUPPRESS_INSTRUMENTATION_KEY, True))
         error: Exception | None = None
         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 as err:
             error = err
@@ -183,7 +192,6 @@ def __init__(
                 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()

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,
 )
@@ -244,6 +248,87 @@ def test_shutdown_allows_1_export_to_finish(
         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()
+
+    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):

opentelemetry-sdk/tests/trace/export/test_export.py

@@ -136,6 +136,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"}
     )