From ccfdf2d7a1d3a2b0d089b44fc98a9dea615f94cc Mon Sep 17 00:00:00 2001 From: Dwin Gharibi Date: Sun, 23 Aug 2026 19:29:44 +0330 Subject: [PATCH 1/2] test(sdk): assert concurrent processors flush at interpreter exit The providers register shutdown with atexit, but concurrent.futures registers its cleanup through threading._register_atexit and CPython runs threading._shutdown() before the atexit queue, so the pool is already closed when the provider's shutdown runs. Drive both signals in a subprocess, since the defect only appears at real interpreter exit, and assert the telemetry is exported and no RuntimeError is reported. Unit tests close the pool explicitly and assert the work still runs inline, with a control asserting a healthy pool is unaffected. These tests fail against the current implementation. --- .../test_concurrent_processor_at_exit.py | 188 ++++++++++++++++++ 1 file changed, 188 insertions(+) create mode 100644 opentelemetry-sdk/tests/test_concurrent_processor_at_exit.py diff --git a/opentelemetry-sdk/tests/test_concurrent_processor_at_exit.py b/opentelemetry-sdk/tests/test_concurrent_processor_at_exit.py new file mode 100644 index 0000000000..107b00404a --- /dev/null +++ b/opentelemetry-sdk/tests/test_concurrent_processor_at_exit.py @@ -0,0 +1,188 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""Concurrent multi-processors must still flush during interpreter shutdown. + +The providers register their shutdown with `atexit.register`, but +`concurrent.futures` registers its own cleanup through +`threading._register_atexit`, and CPython runs `threading._shutdown()` *before* +the `atexit` queue. By the time the provider's shutdown runs the thread pool is +closed, so submitting to it raises RuntimeError and the underlying batch +processor is never shut down. +""" + +import subprocess +import sys +import textwrap +import unittest +from concurrent.futures import ThreadPoolExecutor + +from opentelemetry.sdk._logs import ConcurrentMultiLogRecordProcessor +from opentelemetry.sdk.trace import ConcurrentMultiSpanProcessor + +_SPAN_PROGRAM = """ + from opentelemetry.sdk.trace import TracerProvider, ConcurrentMultiSpanProcessor + from opentelemetry.sdk.trace.export import ( + BatchSpanProcessor, SpanExporter, SpanExportResult, + ) + + class Exporter(SpanExporter): + def export(self, spans): + for span in spans: + print("EXPORTED", span.name) + return SpanExportResult.SUCCESS + + def shutdown(self): + print("EXPORTER_SHUTDOWN") + + provider = TracerProvider(active_span_processor=ConcurrentMultiSpanProcessor(2)) + # long delay so only shutdown can flush this span + provider.add_span_processor(BatchSpanProcessor(Exporter(), schedule_delay_millis=600000)) + with provider.get_tracer(__name__).start_as_current_span("span-flushed-at-exit"): + pass +""" + +_LOG_PROGRAM = """ + from opentelemetry._logs import SeverityNumber + from opentelemetry.sdk._logs import LoggerProvider, ConcurrentMultiLogRecordProcessor + from opentelemetry.sdk._logs.export import ( + BatchLogRecordProcessor, LogExporter, LogExportResult, + ) + + class Exporter(LogExporter): + def export(self, batch): + for record in batch: + print("EXPORTED", record.log_record.body) + return LogExportResult.SUCCESS + + def force_flush(self, timeout_millis=30000): + return True + + def shutdown(self): + print("EXPORTER_SHUTDOWN") + + provider = LoggerProvider( + multi_log_record_processor=ConcurrentMultiLogRecordProcessor(2) + ) + provider.add_log_record_processor( + BatchLogRecordProcessor(Exporter(), schedule_delay_millis=600000) + ) + provider.get_logger(__name__).emit( + body="log-flushed-at-exit", severity_number=SeverityNumber.INFO + ) +""" + + +def _run(program): + return subprocess.run( + [sys.executable, "-c", textwrap.dedent(program)], + capture_output=True, + text=True, + timeout=120, + check=False, + ) + + +class TestFlushAtInterpreterExit(unittest.TestCase): + """Driven in a subprocess: the defect only appears at real interpreter exit.""" + + def test_spans_are_flushed_at_exit(self): + result = _run(_SPAN_PROGRAM) + self.assertIn("EXPORTED span-flushed-at-exit", result.stdout) + self.assertIn("EXPORTER_SHUTDOWN", result.stdout) + + def test_span_shutdown_does_not_raise_at_exit(self): + result = _run(_SPAN_PROGRAM) + self.assertNotIn("cannot schedule new futures", result.stderr) + + def test_logs_are_flushed_at_exit(self): + result = _run(_LOG_PROGRAM) + self.assertIn("EXPORTED log-flushed-at-exit", result.stdout) + self.assertIn("EXPORTER_SHUTDOWN", result.stdout) + + def test_log_shutdown_does_not_raise_at_exit(self): + result = _run(_LOG_PROGRAM) + self.assertNotIn("cannot schedule new futures", result.stderr) + + +class _RecordingSpanProcessor: + def __init__(self): + self.shutdown_called = False + self.flushed = False + + def on_start(self, span, parent_context=None): + pass + + def _on_ending(self, span): + pass + + def on_end(self, span): + pass + + def shutdown(self): + self.shutdown_called = True + + def force_flush(self, timeout_millis=30000): + self.flushed = True + return True + + +class _RecordingLogProcessor: + def __init__(self): + self.shutdown_called = False + self.flushed = False + + def on_emit(self, log_record): + pass + + def shutdown(self): + self.shutdown_called = True + + def force_flush(self, timeout_millis=30000): + self.flushed = True + return True + + +class TestDeadExecutorFallsBackInline(unittest.TestCase): + """With the pool already closed, work must run inline rather than be lost.""" + + def test_span_processor_shutdown_runs_inline(self): + multi = ConcurrentMultiSpanProcessor(2) + child = _RecordingSpanProcessor() + multi.add_span_processor(child) + multi._executor.shutdown() # pylint: disable=protected-access + multi.shutdown() + self.assertTrue(child.shutdown_called) + + def test_span_processor_force_flush_runs_inline(self): + multi = ConcurrentMultiSpanProcessor(2) + child = _RecordingSpanProcessor() + multi.add_span_processor(child) + multi._executor.shutdown() # pylint: disable=protected-access + self.assertTrue(multi.force_flush()) + self.assertTrue(child.flushed) + + def test_log_processor_shutdown_runs_inline(self): + multi = ConcurrentMultiLogRecordProcessor(2) + child = _RecordingLogProcessor() + multi.add_log_record_processor(child) + multi._executor.shutdown() # pylint: disable=protected-access + multi.shutdown() + self.assertTrue(child.shutdown_called) + + def test_log_processor_force_flush_runs_inline(self): + multi = ConcurrentMultiLogRecordProcessor(2) + child = _RecordingLogProcessor() + multi.add_log_record_processor(child) + multi._executor.shutdown() # pylint: disable=protected-access + self.assertTrue(multi.force_flush()) + self.assertTrue(child.flushed) + + def test_healthy_executor_is_still_used(self): + """Control: nothing should change while the pool is alive.""" + multi = ConcurrentMultiSpanProcessor(2) + child = _RecordingSpanProcessor() + multi.add_span_processor(child) + self.assertIsInstance(multi._executor, ThreadPoolExecutor) # pylint: disable=protected-access + multi.shutdown() + self.assertTrue(child.shutdown_called) From 11e63dcd4b6e2ad010669bd3976b8d16f1584386 Mon Sep 17 00:00:00 2001 From: Dwin Gharibi Date: Sun, 23 Aug 2026 19:29:44 +0330 Subject: [PATCH 2/2] fix(sdk): flush concurrent processors during interpreter shutdown TracerProvider and LoggerProvider register their shutdown with atexit, but concurrent.futures registers its own cleanup through threading._register_atexit and CPython runs threading._shutdown() before the atexit queue. The thread pool is therefore always closed by the time an atexit-driven provider shutdown reaches it, so submitting raised "RuntimeError: cannot schedule new futures after shutdown" and the underlying batch processor was never shut down. Every buffered span and log record was lost on every clean exit, reported only as an ignored atexit error. Fall back to running the callable inline when the pool refuses work. There is no concurrency worth preserving at that point in the process lifecycle. force_flush gets the same treatment and propagates a False result from an inline call. --- .changelog/5568.fixed | 1 + .../sdk/_logs/_internal/__init__.py | 16 +++++++++++++--- .../src/opentelemetry/sdk/trace/__init__.py | 18 +++++++++++++++--- 3 files changed, 29 insertions(+), 6 deletions(-) create mode 100644 .changelog/5568.fixed diff --git a/.changelog/5568.fixed b/.changelog/5568.fixed new file mode 100644 index 0000000000..8f2d8b582c --- /dev/null +++ b/.changelog/5568.fixed @@ -0,0 +1 @@ +`opentelemetry-sdk`: run `ConcurrentMultiSpanProcessor` and `ConcurrentMultiLogRecordProcessor` work inline once the thread pool is closed, so telemetry is still flushed at interpreter exit diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/__init__.py index 300f5c3cec..18cb933673 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/__init__.py @@ -467,8 +467,13 @@ def _submit_and_wait( ): futures = [] for lp in self._log_record_processors: - future = self._executor.submit(func(lp), *args, **kwargs) - futures.append(future) + try: + future = self._executor.submit(func(lp), *args, **kwargs) + except RuntimeError: + # See ConcurrentMultiSpanProcessor._submit_and_await. + func(lp)(*args, **kwargs) + else: + futures.append(future) for future in futures: future.result() @@ -491,7 +496,12 @@ def force_flush(self, timeout_millis: int = 30000) -> bool: """ futures = [] for lp in self._log_record_processors: - future = self._executor.submit(lp.force_flush, timeout_millis) + try: + future = self._executor.submit(lp.force_flush, timeout_millis) + except RuntimeError: + if lp.force_flush(timeout_millis) is False: + return False + continue futures.append(future) done_futures, not_done_futures = concurrent.futures.wait(futures, timeout_millis / 1e3) diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py index 14d6573d0c..6eb74e2ce5 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py @@ -269,8 +269,15 @@ def _submit_and_await( ): futures = [] for sp in self._span_processors: - future = self._executor.submit(func(sp), *args, **kwargs) - futures.append(future) + try: + future = self._executor.submit(func(sp), *args, **kwargs) + except RuntimeError: + # The interpreter is shutting down: concurrent.futures' own + # atexit hook has already run, so no new work can be scheduled. + # Run inline instead, otherwise shutdown and flush are lost. + func(sp)(*args, **kwargs) + else: + futures.append(future) for future in futures: future.result() @@ -305,7 +312,12 @@ def force_flush(self, timeout_millis: int = 30000) -> bool: """ futures = [] for sp in self._span_processors: - future = self._executor.submit(sp.force_flush, timeout_millis) + try: + future = self._executor.submit(sp.force_flush, timeout_millis) + except RuntimeError: + if sp.force_flush(timeout_millis) is False: + return False + continue futures.append(future) timeout_sec = timeout_millis / 1e3