diff --git a/.changelog/5835.fixed b/.changelog/5835.fixed new file mode 100644 index 0000000000..3e260213c3 --- /dev/null +++ b/.changelog/5835.fixed @@ -0,0 +1 @@ +Fix concurrent span/log processor shutdown at exit so that a failing processor callback no longer blocks healthy sibling processors and a Thread.start RuntimeError no longer double-invokes a callback. diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/__init__.py index 5b543d01e4..21ff442875 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/__init__.py @@ -9,6 +9,7 @@ import json import logging import os +import sys import threading import traceback import warnings @@ -87,6 +88,49 @@ _ENV_VALUE_UNSET = "" _logger = logging.getLogger(__name__) +_callback_log_reentry = threading.local() + + +def _safe_log_exception(logger, msg): # pylint: disable=redefined-outer-name + """Log an exception safely against recursive logging handlers. + + If we are already inside a diagnostic log dispatch on this thread, write + the record to stderr directly instead of going through the logging module + to prevent infinite feedback loops. Any exception raised by a user logging + handler during the outer dispatch is contained here. + """ + if getattr(_callback_log_reentry, "in_diag", False): + print(msg, file=sys.stderr) + traceback.print_exc(file=sys.stderr) + return + _callback_log_reentry.in_diag = True + try: + logger.exception(msg) + except Exception: # pylint: disable=broad-exception-caught + print(msg, file=sys.stderr) + traceback.print_exc(file=sys.stderr) + finally: + _callback_log_reentry.in_diag = False + + +def _safe_log_debug(logger, msg): # pylint: disable=redefined-outer-name + """Emit a DEBUG diagnostic without letting a user logging handler error escape. + + The inline fallback path must run even when an attached logging handler + raises; the DEBUG record is still attempted through the standard handler, + and any handler error falls back to stderr instead of aborting the + callback dispatch. + """ + if getattr(_callback_log_reentry, "in_diag", False): + print(msg, file=sys.stderr) + return + _callback_log_reentry.in_diag = True + try: + logger.debug(msg) # noqa: SHADOW + except Exception: # pylint: disable=broad-exception-caught + print(msg, file=sys.stderr) + finally: + _callback_log_reentry.in_diag = False class BytesEncoder(json.JSONEncoder): @@ -476,6 +520,49 @@ def force_flush(self, timeout_millis: int = 30000) -> bool: return all_flushed +class _CallbackAttempt: + """Per-dispatch at-most-once wrapper for a log record processor callback. + + See :class:`opentelemetry.sdk.trace._CallbackAttempt` for the rationale. + """ + + __slots__ = ("_callback", "_args", "_kwargs", "_lock", "_started", "_finished") + + def __init__(self, callback: Callable[..., None], args: tuple[Any, ...], kwargs: dict[str, Any]): + self._callback = callback + self._args = args + self._kwargs = kwargs + self._lock = threading.Lock() + self._started = False + self._finished = threading.Event() + + def run(self) -> None: + with self._lock: + if self._started: + return + self._started = True + # Mark that we are executing a processor callback on this thread. If a + # logging handler (e.g. LoggingHandler) synchronously calls back into + # on_emit, _submit_and_wait sees this flag and runs callbacks inline + # instead of submitting to the executor (whose workers are blocked + # inside the outer callback, which would deadlock). + reentry = getattr(_callback_log_reentry, "active", False) + _callback_log_reentry.active = True + try: + self._callback(*self._args, **self._kwargs) + except Exception: # pylint: disable=broad-exception-caught + _safe_log_exception(_logger, "Exception while calling LogRecordProcessor callback.") + finally: + _callback_log_reentry.active = reentry + self._finished.set() + + def wait_if_started(self) -> None: + with self._lock: + started = self._started + if started: + self._finished.wait() + + class ConcurrentMultiLogRecordProcessor(LogRecordProcessor): """Implementation of :class:`LogRecordProcessor` that forwards all received events to a list of log processors in parallel. @@ -506,12 +593,57 @@ def _submit_and_wait( *args: Any, **kwargs: Any, ): + # Reentry guard: if we are already running a callback on this thread, or + # we were dispatched from inside a real logging handler that holds its own + # lock across emit(), run the callbacks inline instead of through the + # executor. Submitting would either queue work no worker can pick up + # (workers blocked inside the outer callback) or block a re-entrant + # worker on the logging handler lock the dispatching thread still holds, + # and future.result() would deadlock in both cases. + if getattr(_callback_log_reentry, "active", False) or getattr(_callback_log_reentry, "in_bridge", False): + for lp in self._log_record_processors: + try: + callback = func(lp) + callback(*args, **kwargs) + except Exception: # pylint: disable=broad-exception-caught + _safe_log_exception( + _logger, + "Exception while re-entering LogRecordProcessor callback inline.", + ) + return + attempts = [] futures = [] for lp in self._log_record_processors: - future = self._executor.submit(func(lp), *args, **kwargs) - futures.append(future) + try: + callback = func(lp) + except Exception: # pylint: disable=broad-exception-caught + _safe_log_exception(_logger, "Exception while resolving LogRecordProcessor callback.") + continue + attempt = _CallbackAttempt(callback, args, kwargs) + attempts.append(attempt) + try: + futures.append(self._executor.submit(attempt.run)) + except RuntimeError: + # The executor may already be shut down when this runs from an + # atexit handler: CPython runs threading._shutdown() before the + # atexit queue, so the LoggerProvider.shutdown() runs. Run + # inline through the SAME attempt; the shared lock prevents + # double invocation even if the work item was already enqueued + # before Thread.start failed. + _safe_log_debug( + _logger, + "Executor is shut down; running LogRecordProcessor callback inline.", + ) + attempt.run() + except Exception: # pylint: disable=broad-exception-caught + _safe_log_exception(_logger, "Exception while submitting LogRecordProcessor callback.") for future in futures: - future.result() + try: + future.result() + except Exception: # pylint: disable=broad-exception-caught + _safe_log_exception(_logger, "Exception in LogRecordProcessor worker.") + for attempt in attempts: + attempt.wait_if_started() def on_emit(self, log_record: ReadWriteLogRecord) -> None: self._submit_and_wait(lambda lp: lp.on_emit, log_record) @@ -710,7 +842,17 @@ def emit(self, record: logging.LogRecord) -> None: """ logger = get_logger(record.name, logger_provider=self._logger_provider) if not isinstance(logger, NoOpLogger): - logger.emit(self._translate(record)) + # A real logging handler holds its own lock across emit(). If the + # concurrent multi-processor submits work to its executor and then + # blocks waiting on it, a worker that re-enters logging through this + # same handler would deadlock on that lock. Mark the bridge boundary + # so the multi-processor dispatches inline while we hold the handler + # lock, matching the re-entry guard for callbacks. + _callback_log_reentry.in_bridge = True + try: + logger.emit(self._translate(record)) + finally: + _callback_log_reentry.in_bridge = False def flush(self) -> None: """ diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py index 6f47f213bf..d552042427 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py @@ -8,6 +8,7 @@ import json import logging import os +import sys import threading import traceback import typing @@ -75,6 +76,49 @@ from opentelemetry.util._decorator import _agnosticcontextmanager logger = logging.getLogger(__name__) +_callback_log_reentry = threading.local() + + +def _safe_log_exception(logger, msg): # pylint: disable=redefined-outer-name + """Log an exception safely against recursive logging handlers. + + If we are already inside a diagnostic log dispatch on this thread, write + the record to stderr directly instead of going through the logging module + to prevent infinite feedback loops. Any exception raised by a user logging + handler during the outer dispatch is contained here. + """ + if getattr(_callback_log_reentry, "in_diag", False): + print(msg, file=sys.stderr) + traceback.print_exc(file=sys.stderr) + return + _callback_log_reentry.in_diag = True + try: + logger.exception(msg) # noqa: SHADOW + except Exception: # pylint: disable=broad-exception-caught + print(msg, file=sys.stderr) + traceback.print_exc(file=sys.stderr) + finally: + _callback_log_reentry.in_diag = False + + +def _safe_log_debug(logger, msg): # pylint: disable=redefined-outer-name + """Emit a DEBUG diagnostic without letting a user logging handler error escape. + + The inline fallback path must run even when an attached logging handler + raises; the DEBUG record is still attempted through the standard handler, + and any handler error falls back to stderr instead of aborting the + callback dispatch. + """ + if getattr(_callback_log_reentry, "in_diag", False): + print(msg, file=sys.stderr) + return + _callback_log_reentry.in_diag = True + try: + logger.debug(msg) # noqa: SHADOW + except Exception: # pylint: disable=broad-exception-caught + print(msg, file=sys.stderr) + finally: + _callback_log_reentry.in_diag = False _DEFAULT_OTEL_ATTRIBUTE_COUNT_LIMIT = 128 @@ -224,6 +268,58 @@ def force_flush(self, timeout_millis: int = 30000) -> bool: return all_flushed +class _CallbackAttempt: + """Per-dispatch at-most-once wrapper for a processor callback. + + The queue worker and the inline fallback (when ``executor.submit`` raises + ``RuntimeError``) share the same attempt. A lock-guarded claim ensures the + underlying callback runs at most once even when ``Thread.start`` fails + after the work item was already enqueued, which previously caused the + callback to run twice. Callback exceptions are contained and logged so a + faulty processor does not block healthy siblings or replace the caller's + own exception. + + This scopes to a single dispatch entry; it does not promise global + exactly-once delivery, retry, or bounded shutdown. + """ + + __slots__ = ("_callback", "_args", "_kwargs", "_lock", "_started", "_finished") + + def __init__(self, callback: Callable[..., None], args: tuple[Any, ...], kwargs: dict[str, Any]): + self._callback = callback + self._args = args + self._kwargs = kwargs + self._lock = threading.Lock() + self._started = False + self._finished = threading.Event() + + def run(self) -> None: + with self._lock: + if self._started: + return + self._started = True + # Mark that we are executing a processor callback on this thread. If a + # logging handler (e.g. LoggingHandler) synchronously calls back into + # on_emit/on_start, _submit_and_await sees this flag and runs callbacks + # inline instead of submitting to the executor (whose worker threads + # are blocked inside the outer callback, which would deadlock). + reentry = getattr(_callback_log_reentry, "active", False) + _callback_log_reentry.active = True + try: + self._callback(*self._args, **self._kwargs) + except Exception: # pylint: disable=broad-exception-caught + _safe_log_exception(logger, "Exception while calling SpanProcessor callback.") + finally: + _callback_log_reentry.active = reentry + self._finished.set() + + def wait_if_started(self) -> None: + with self._lock: + started = self._started + if started: + self._finished.wait() + + class ConcurrentMultiSpanProcessor(SpanProcessor): """Implementation of :class:`SpanProcessor` that forwards all received events to a list of span processors in parallel. @@ -273,12 +369,57 @@ def _submit_and_await( *args: Any, **kwargs: Any, ): + # Reentry guard: if we are already running a callback on this thread + # (e.g. a logging handler synchronously called back into on_start), + # run the remaining callbacks inline instead of through the executor. + # Submitting would queue work that no worker can pick up (workers are + # blocked inside the outer callback), and future.result() would + # deadlock. This is bounded: it runs synchronously on the calling + # thread and does not recurse through the executor. + if getattr(_callback_log_reentry, "active", False): + for sp in self._span_processors: + try: + callback = func(sp) + callback(*args, **kwargs) + except Exception: # pylint: disable=broad-exception-caught + _safe_log_exception( + logger, + "Exception while re-entering SpanProcessor callback inline.", + ) + return + attempts = [] futures = [] for sp in self._span_processors: - future = self._executor.submit(func(sp), *args, **kwargs) - futures.append(future) + try: + callback = func(sp) + except Exception: # pylint: disable=broad-exception-caught + _safe_log_exception(logger, "Exception while resolving SpanProcessor callback.") + continue + attempt = _CallbackAttempt(callback, args, kwargs) + attempts.append(attempt) + try: + futures.append(self._executor.submit(attempt.run)) + except RuntimeError: + # The executor may already be shut down when this runs from an + # atexit handler: CPython runs threading._shutdown() before the + # atexit queue, so the ThreadPoolExecutor is closed by the time + # TracerProvider.shutdown() runs. Run inline through the SAME + # attempt; the shared lock prevents double invocation even if + # the work item was already enqueued before Thread.start failed. + _safe_log_debug( + logger, + "Executor is shut down; running SpanProcessor callback inline.", + ) + attempt.run() + except Exception: # pylint: disable=broad-exception-caught + _safe_log_exception(logger, "Exception while submitting SpanProcessor callback.") for future in futures: - future.result() + try: + future.result() + except Exception: # pylint: disable=broad-exception-caught + _safe_log_exception(logger, "Exception in SpanProcessor worker.") + for attempt in attempts: + attempt.wait_if_started() def on_start( self, diff --git a/opentelemetry-sdk/tests/logs/test_multi_log_processor.py b/opentelemetry-sdk/tests/logs/test_multi_log_processor.py index e88854fc7c..6a5d22b961 100644 --- a/opentelemetry-sdk/tests/logs/test_multi_log_processor.py +++ b/opentelemetry-sdk/tests/logs/test_multi_log_processor.py @@ -3,12 +3,14 @@ # pylint:disable=protected-access,no-self-use,no-member +import io import logging +import sys import threading import time import unittest from abc import ABC, abstractmethod -from unittest.mock import Mock +from unittest.mock import Mock, patch from opentelemetry._logs import LogRecord, SeverityNumber from opentelemetry.sdk._logs._internal import ( @@ -251,3 +253,274 @@ def test_force_flush_processor_returns_none(self): for mock in mocks: self.assertEqual(1, mock.force_flush.call_count) multi_log_record_processor.shutdown() + + # pylint: disable=no-self-use + def test_shutdown_runs_inline_after_executor_shutdown(self): + # Simulate the atexit scenario: CPython runs threading._shutdown() + # before the atexit queue, so the ThreadPoolExecutor is closed by the + # time LoggerProvider.shutdown() runs. Underlying processors must + # still be shut down inline so buffered work is not dropped. + multi_log_record_processor = ConcurrentMultiLogRecordProcessor(2) + mock_processor = Mock(spec=LogRecordProcessor) + multi_log_record_processor.add_log_record_processor(mock_processor) + + # pylint: disable=protected-access + multi_log_record_processor._executor.shutdown(wait=True) + + multi_log_record_processor.shutdown() + mock_processor.shutdown.assert_called_once_with() + + def test_callback_exception_does_not_block_siblings(self): + # A LogRecordProcessor callback failure must not raise out of + # shutdown() and must not prevent healthy siblings from running. + multi_log_record_processor = ConcurrentMultiLogRecordProcessor(2) + raising_processor = Mock(spec=LogRecordProcessor) + raising_processor.shutdown.side_effect = RuntimeError("processor boom") + healthy_processor = Mock(spec=LogRecordProcessor) + multi_log_record_processor.add_log_record_processor(raising_processor) + multi_log_record_processor.add_log_record_processor(healthy_processor) + + with self.assertLogs("opentelemetry.sdk._logs._internal", level="ERROR"): + multi_log_record_processor.shutdown() + + raising_processor.shutdown.assert_called_once_with() + healthy_processor.shutdown.assert_called_once_with() + + def test_submit_runtime_error_does_not_double_invoke(self): + # If Thread.start fails AFTER the work item was enqueued, the inline + # fallback must not run the callback a second time. + multi_log_record_processor = ConcurrentMultiLogRecordProcessor(2) + processor_a = Mock(spec=LogRecordProcessor) + processor_b = Mock(spec=LogRecordProcessor) + multi_log_record_processor.add_log_record_processor(processor_a) + multi_log_record_processor.add_log_record_processor(processor_b) + + real_start = threading.Thread.start + calls = {"n": 0} + + def first_start_fails(target): + calls["n"] += 1 + if calls["n"] == 1: + raise RuntimeError("cannot start new thread") + return real_start(target) + + with patch.object(threading.Thread, "start", first_start_fails): + multi_log_record_processor.shutdown() + + self.assertEqual(processor_a.shutdown.call_count, 1) + self.assertEqual(processor_b.shutdown.call_count, 1) + + def test_callback_logger_handler_raise_does_not_crash(self): + # A logging handler attached to the OTel logger that raises must not + # escape out of shutdown() nor deadlock; reentry guard routes to stderr. + multi = ConcurrentMultiLogRecordProcessor(2) + raising = Mock(spec=LogRecordProcessor) + raising.shutdown.side_effect = RuntimeError("processor boom") + healthy = Mock(spec=LogRecordProcessor) + multi.add_log_record_processor(raising) + multi.add_log_record_processor(healthy) + + class _BoomHandler(logging.Handler): + def emit(self, record): + # Raises OUTWARD (no self-catch, raiseExceptions left default). + raise RuntimeError("handler boom") + + handler = _BoomHandler(level=logging.ERROR) + otel_logger = logging.getLogger("opentelemetry.sdk._logs._internal") + otel_logger.addHandler(handler) + try: + # Normal-executor path: faulty callback raises -> future.result() -> + # safe_log_exception -> outward handler raise must not escape. + multi.shutdown() + finally: + otel_logger.removeHandler(handler) + multi._executor.shutdown(wait=True) + + healthy.shutdown.assert_called_once_with() + + def test_logging_handler_raise_during_callback_resolution_is_isolated(self): + # If resolving a callback (attribute lookup) raises AND an attached + # handler raises on the resulting exception log, dispatch continues to + # healthy siblings. + multi = ConcurrentMultiLogRecordProcessor(2) + + class _ResolveBoom: + @property + def shutdown(self): + raise RuntimeError("resolve boom") + + healthy = Mock(spec=LogRecordProcessor) + multi.add_log_record_processor(_ResolveBoom()) + multi.add_log_record_processor(healthy) + + class _OutwardBoom(logging.Handler): + def emit(self, record): + raise RuntimeError("handler boom outward") + + otel_logger = logging.getLogger("opentelemetry.sdk._logs._internal") + old_level = otel_logger.level + otel_logger.setLevel(logging.ERROR) + handler = _OutwardBoom(level=logging.ERROR) + otel_logger.addHandler(handler) + try: + multi.shutdown() + finally: + otel_logger.removeHandler(handler) + otel_logger.setLevel(old_level) + multi._executor.shutdown(wait=True) + + healthy.shutdown.assert_called_once_with() + + def test_logging_handler_outward_raise_inline_fallback_is_isolated(self): + # A user logging handler that raises OUTWARD (no self-catch, with + # raiseExceptions left at its default) must not abort the inline + # fallback when the executor is already shut down. The executor-shutdown + # DEBUG diagnostic and per-callback exception log are wrapped so a + # handler error falls back to stderr; both faulty and healthy shut down. + multi = ConcurrentMultiLogRecordProcessor(2) + faulty = Mock(spec=LogRecordProcessor) + faulty.shutdown.side_effect = RuntimeError("processor fault") + healthy = Mock(spec=LogRecordProcessor) + multi.add_log_record_processor(faulty) + multi.add_log_record_processor(healthy) + + class _OutwardBoom(logging.Handler): + def emit(self, record): + raise RuntimeError("handler boom outward") + + otel_logger = logging.getLogger("opentelemetry.sdk._logs._internal") + old_level = otel_logger.level + handler = _OutwardBoom(level=logging.DEBUG) + otel_logger.setLevel(logging.DEBUG) + otel_logger.addHandler(handler) + # pylint: disable=protected-access + multi._executor.shutdown(wait=True) + try: + multi.shutdown() + finally: + otel_logger.removeHandler(handler) + otel_logger.setLevel(old_level) + multi._executor.shutdown(wait=True) + + healthy.shutdown.assert_called_once_with() + faulty.shutdown.assert_called_once_with() + + def test_submit_runtime_error_after_enqueue_does_not_double_invoke(self): + # When executor.submit raises RuntimeError AFTER the work item was + # actually enqueued, the inline fallback must not run the callback a + # second time. The shared _CallbackAttempt lock guarantees at-most-once, + # and an outward-raising user handler on the diagnostic log must not + # escape. Stderr must still contain the diagnostic record. + + multi = ConcurrentMultiLogRecordProcessor(2) + faulty = Mock(spec=LogRecordProcessor) + faulty.shutdown.side_effect = RuntimeError("processor fault") + healthy = Mock(spec=LogRecordProcessor) + multi.add_log_record_processor(faulty) + multi.add_log_record_processor(healthy) + + class _OutwardBoom(logging.Handler): + def emit(self, record): + raise RuntimeError("handler boom outward") + + otel_logger = logging.getLogger("opentelemetry.sdk._logs._internal") + old_level = otel_logger.level + otel_logger.setLevel(logging.DEBUG) + handler = _OutwardBoom(level=logging.DEBUG) + otel_logger.addHandler(handler) + + # pylint: disable=protected-access + orig_submit = multi._executor.submit + enqueued_futures = [] + + def patched(fn, *a, **kw): + fut = orig_submit(fn, *a, **kw) # actually enqueue + enqueued_futures.append(fut) + raise RuntimeError("cannot start new thread") + + multi._executor.submit = patched + stderr_buf = io.StringIO() + old_stderr = sys.stderr + sys.stderr = stderr_buf + try: + multi.shutdown() + for fut in enqueued_futures: + fut.result() + finally: + sys.stderr = old_stderr + otel_logger.removeHandler(handler) + otel_logger.setLevel(old_level) + multi._executor.shutdown(wait=True) + + assert len(enqueued_futures) == 2 + healthy.shutdown.assert_called_once_with() + faulty.shutdown.assert_called_once_with() + assert "Executor is shut down" in stderr_buf.getvalue() + + def test_real_logging_handler_reentry_runs_inline(self): + # Wire the REAL LoggingHandler bridge into a ConcurrentMultiLogRecordProcessor. + # When a child processor emits a Python log on a worker thread, the bridge + # re-enters on_emit on that same thread. The active reentry guard must run + # the re-entered callbacks inline instead of submitting to the (busy) executor, + # so the call returns without deadlock and a healthy sibling still receives + # both the outer and the re-entered record. + multi = ConcurrentMultiLogRecordProcessor(2) + bridge_logger = logging.getLogger("otel.bridge.reentry") + bridge_logger.propagate = False + state = {"fired": False} + + class _Reenter(LogRecordProcessor): + def on_emit(self, log_record): + if not state["fired"]: + state["fired"] = True + bridge_logger.error("reentered") + + def shutdown(self): + pass + + def force_flush(self, timeout_millis=30000): + return True + + healthy = Mock(spec=LogRecordProcessor) + multi.add_log_record_processor(_Reenter()) + multi.add_log_record_processor(healthy) + provider = LoggerProvider(multi_log_record_processor=multi, shutdown_on_exit=False) + bridge = LoggingHandler(logger_provider=provider) + bridge_logger.addHandler(bridge) + try: + # Must return promptly: no executor deadlock, no unbounded recursion. + bridge_logger.error("outer") + finally: + bridge_logger.removeHandler(bridge) + multi._executor.shutdown(wait=True) + + # healthy sibling saw the outer record and the inline re-entered record. + self.assertEqual(healthy.on_emit.call_count, 2) + + def test_real_logging_handler_raise_is_isolated(self): + # Attach the REAL LoggingHandler bridge to the OTel internal logger. A + # raising child processor must not let the handler error escape shutdown(), + # must not deadlock, and must not prevent a healthy sibling from shutting + # down. The Python logging framework contains the handler error; the SDK's + # own diagnostic fallback keeps the record off the recursive path. + multi = ConcurrentMultiLogRecordProcessor(2) + raising = Mock(spec=LogRecordProcessor) + raising.shutdown.side_effect = RuntimeError("processor boom") + healthy = Mock(spec=LogRecordProcessor) + multi.add_log_record_processor(raising) + multi.add_log_record_processor(healthy) + provider = LoggerProvider(multi_log_record_processor=multi, shutdown_on_exit=False) + bridge = LoggingHandler(logger_provider=provider) + otel_logger = logging.getLogger("opentelemetry.sdk._logs._internal") + otel_logger.addHandler(bridge) + old_raise = logging.raiseExceptions + logging.raiseExceptions = False + try: + # Must not raise and must not deadlock. + multi.shutdown() + finally: + logging.raiseExceptions = old_raise + otel_logger.removeHandler(bridge) + multi._executor.shutdown(wait=True) + + healthy.shutdown.assert_called_once_with() diff --git a/opentelemetry-sdk/tests/trace/test_span_processor.py b/opentelemetry-sdk/tests/trace/test_span_processor.py index f2595b9669..89c50fc3a3 100644 --- a/opentelemetry-sdk/tests/trace/test_span_processor.py +++ b/opentelemetry-sdk/tests/trace/test_span_processor.py @@ -3,8 +3,11 @@ import abc import gc +import io +import logging import multiprocessing import os +import sys import time import unittest import weakref @@ -718,6 +721,246 @@ def test_processor_gc(self): "The ConcurrentMultiSpanProcessor object created by this test wasn't garbage collected", ) + # pylint: disable=no-self-use + def test_shutdown_runs_inline_after_executor_shutdown(self): + # Simulate the atexit scenario: CPython runs threading._shutdown() + # before the atexit queue, so the ThreadPoolExecutor is closed by the + # time TracerProvider.shutdown() runs. The underlying processors must + # still be shut down inline so buffered work is not dropped. + multi_processor = trace.ConcurrentMultiSpanProcessor(2) + mock_processor = mock.Mock(spec=trace.SpanProcessor) + multi_processor.add_span_processor(mock_processor) + + # pylint: disable=protected-access + multi_processor._executor.shutdown(wait=True) + + multi_processor.shutdown() + mock_processor.shutdown.assert_called_once_with() + + def test_on_end_runs_inline_after_executor_shutdown(self): + multi_processor = trace.ConcurrentMultiSpanProcessor(2) + mock_processor = mock.Mock(spec=trace.SpanProcessor) + multi_processor.add_span_processor(mock_processor) + + # pylint: disable=protected-access + multi_processor._executor.shutdown(wait=True) + + span = self.create_default_span() + multi_processor.on_end(span) + mock_processor.on_end.assert_called_once_with(span) + + def test_callback_exception_does_not_block_siblings(self): + # Per OTel error-handling spec, a processor callback must not raise out + # of shutdown() and must not prevent healthy siblings from running. + multi_processor = trace.ConcurrentMultiSpanProcessor(2) + raising_processor = mock.Mock(spec=trace.SpanProcessor) + raising_processor.shutdown.side_effect = RuntimeError("processor boom") + healthy_processor = mock.Mock(spec=trace.SpanProcessor) + multi_processor.add_span_processor(raising_processor) + multi_processor.add_span_processor(healthy_processor) + + with self.assertLogs("opentelemetry.sdk.trace", level="ERROR"): + multi_processor.shutdown() + + raising_processor.shutdown.assert_called_once_with() + healthy_processor.shutdown.assert_called_once_with() + + def test_submit_runtime_error_does_not_double_invoke(self): + # If Thread.start fails AFTER the work item was enqueued, the inline + # fallback must not run the callback a second time: the queue worker + # will pick up the enqueued item. A shared claim per dispatch ensures + # at-most-once. + multi_processor = trace.ConcurrentMultiSpanProcessor(2) + processor_a = mock.Mock(spec=trace.SpanProcessor) + processor_b = mock.Mock(spec=trace.SpanProcessor) + multi_processor.add_span_processor(processor_a) + multi_processor.add_span_processor(processor_b) + + real_start = Thread.start + calls = {"n": 0} + + def first_start_fails(target): + calls["n"] += 1 + if calls["n"] == 1: + raise RuntimeError("cannot start new thread") + return real_start(target) + + with mock.patch.object(Thread, "start", first_start_fails): + multi_processor.shutdown() + + self.assertEqual(processor_a.shutdown.call_count, 1) + self.assertEqual(processor_b.shutdown.call_count, 1) + + def test_callback_exception_preserves_application_exception(self): + # A processor callback failure must not replace the caller's own + # exception when on_end runs on the caller's thread. + multi_processor = trace.ConcurrentMultiSpanProcessor(2) + faulty = mock.Mock(spec=trace.SpanProcessor) + faulty.on_end.side_effect = RuntimeError("telemetry fault") + healthy = mock.Mock(spec=trace.SpanProcessor) + multi_processor.add_span_processor(faulty) + multi_processor.add_span_processor(healthy) + + provider = trace.TracerProvider(active_span_processor=multi_processor, shutdown_on_exit=False) + app_error = ValueError("business fault") + try: + with self.assertRaises(ValueError) as caught: + with provider.get_tracer("test").start_as_current_span("x"): + raise app_error + self.assertIs(caught.exception, app_error) + finally: + # Tear down through the public contract instead of reaching into + # the executor's protected member; the processor was never shut down. + multi_processor.shutdown() + + def test_callback_logger_handler_raise_does_not_crash(self): + # A logging handler attached to the OTel logger that raises must not + # escape out of shutdown() nor deadlock; the reentry guard routes + # diagnostics to stderr and healthy siblings still run. + multi_processor = trace.ConcurrentMultiSpanProcessor(2) + raising_processor = mock.Mock(spec=trace.SpanProcessor) + raising_processor.shutdown.side_effect = RuntimeError("processor boom") + healthy = mock.Mock(spec=trace.SpanProcessor) + multi_processor.add_span_processor(raising_processor) + multi_processor.add_span_processor(healthy) + + class _BoomHandler(logging.Handler): + def emit(self, record): + # Raises OUTWARD (no self-catch, raiseExceptions left default). + # The safe diagnostic wrapper must contain it and fall back to stderr. + raise RuntimeError("handler boom") + + handler = _BoomHandler(level=logging.ERROR) + otel_logger = logging.getLogger("opentelemetry.sdk.trace") + otel_logger.addHandler(handler) + try: + # Normal-executor path: faulty callback raises -> future.result() -> + # safe_log_exception -> outward handler raise must not escape. + multi_processor.shutdown() + finally: + otel_logger.removeHandler(handler) + + healthy.shutdown.assert_called_once_with() + + def test_logging_handler_raise_during_callback_resolution_is_isolated(self): + # If resolving a callback (attribute lookup) raises AND an attached + # handler raises on the resulting exception log, dispatch must continue + # to healthy siblings. + multi_processor = trace.ConcurrentMultiSpanProcessor(2) + + class _ResolveBoom: + @property + def shutdown(self): + raise RuntimeError("resolve boom") + + healthy = mock.Mock(spec=trace.SpanProcessor) + multi_processor.add_span_processor(_ResolveBoom()) + multi_processor.add_span_processor(healthy) + + class _OutwardBoom(logging.Handler): + def emit(self, record): + raise RuntimeError("handler boom outward") + + otel_logger = logging.getLogger("opentelemetry.sdk.trace") + old_level = otel_logger.level + otel_logger.setLevel(logging.ERROR) + handler = _OutwardBoom(level=logging.ERROR) + otel_logger.addHandler(handler) + try: + multi_processor.shutdown() + finally: + otel_logger.removeHandler(handler) + otel_logger.setLevel(old_level) + + healthy.shutdown.assert_called_once_with() + + def test_logging_handler_outward_raise_inline_fallback_is_isolated(self): + # A user logging handler that raises OUTWARD (no self-catch, with + # raiseExceptions left at its default) must not abort the inline + # fallback when the executor is already shut down (the atexit path). + # The executor-shutdown DEBUG diagnostic and the per-callback exception + # log are wrapped so a handler error falls back to stderr instead of + # escaping; both the faulty and the healthy processor still shut down. + multi_processor = trace.ConcurrentMultiSpanProcessor(2) + faulty = mock.Mock(spec=trace.SpanProcessor) + faulty.shutdown.side_effect = RuntimeError("processor fault") + healthy = mock.Mock(spec=trace.SpanProcessor) + multi_processor.add_span_processor(faulty) + multi_processor.add_span_processor(healthy) + + class _OutwardBoom(logging.Handler): + def emit(self, record): + raise RuntimeError("handler boom outward") + + otel_logger = logging.getLogger("opentelemetry.sdk.trace") + old_level = otel_logger.level + handler = _OutwardBoom(level=logging.DEBUG) + otel_logger.setLevel(logging.DEBUG) + otel_logger.addHandler(handler) + # pylint: disable=protected-access + multi_processor._executor.shutdown(wait=True) + try: + # Must not raise despite the outward handler raise on the + # "Executor is shut down" DEBUG diagnostic. + multi_processor.shutdown() + finally: + otel_logger.removeHandler(handler) + otel_logger.setLevel(old_level) + + healthy.shutdown.assert_called_once_with() + faulty.shutdown.assert_called_once_with() + + def test_submit_runtime_error_after_enqueue_does_not_double_invoke(self): + # submit raises RuntimeError after enqueue: inline fallback must not + # double-invoke; outward DEBUG handler must not escape; stderr fallback + # must still record the diagnostic. + multi_processor = trace.ConcurrentMultiSpanProcessor(2) + faulty = mock.Mock(spec=trace.SpanProcessor) + faulty.shutdown.side_effect = RuntimeError("processor fault") + healthy = mock.Mock(spec=trace.SpanProcessor) + multi_processor.add_span_processor(faulty) + multi_processor.add_span_processor(healthy) + + class _OutwardBoom(logging.Handler): + def emit(self, record): + raise RuntimeError("handler boom outward") + + otel_logger = logging.getLogger("opentelemetry.sdk.trace") + old_level = otel_logger.level + otel_logger.setLevel(logging.DEBUG) + handler = _OutwardBoom(level=logging.DEBUG) + otel_logger.addHandler(handler) + + # pylint: disable=protected-access + orig_submit = multi_processor._executor.submit + enqueued_futures = [] + + def patched(fn, *a, **kw): + fut = orig_submit(fn, *a, **kw) # actually enqueue + enqueued_futures.append(fut) + raise RuntimeError("cannot start new thread") # then raise + + multi_processor._executor.submit = patched + stderr_buf = io.StringIO() + old_stderr = sys.stderr + sys.stderr = stderr_buf + try: + multi_processor.shutdown() + # Wait for all enqueued workers to finish before asserting. + for fut in enqueued_futures: + fut.result() + finally: + sys.stderr = old_stderr + otel_logger.removeHandler(handler) + otel_logger.setLevel(old_level) + multi_processor._executor.shutdown(wait=True) + + assert len(enqueued_futures) == 2 # one per processor, both enqueued + healthy.shutdown.assert_called_once_with() + faulty.shutdown.assert_called_once_with() + # Diagnostic stderr fallback must have recorded the shutdown message. + assert "Executor is shut down" in stderr_buf.getvalue() + @unittest.skipUnless(hasattr(os, "fork"), "needs *nix") def test_batch_span_processor_fork(self): multiprocessing_context = multiprocessing.get_context("fork")