diff --git a/.changelog/15.fixed b/.changelog/15.fixed new file mode 100644 index 00000000000..d94fe66dc79 --- /dev/null +++ b/.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 diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py index 3e2b8a263a4..aa8ebfb8d94 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 @@ -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: diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py index ad8f57840a0..360b7e5d0f7 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 @@ -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() diff --git a/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py b/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py index 02337757073..0adbd0a450d 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, ) @@ -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): diff --git a/opentelemetry-sdk/tests/trace/export/test_export.py b/opentelemetry-sdk/tests/trace/export/test_export.py index d67aceb19f8..ed90f673bef 100644 --- a/opentelemetry-sdk/tests/trace/export/test_export.py +++ b/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"} )