diff --git a/.changelog/5568.fixed b/.changelog/5568.fixed new file mode 100644 index 00000000000..8f2d8b582cc --- /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 300f5c3cecc..18cb933673a 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 14d6573d0cf..6eb74e2ce58 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 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 00000000000..107b00404af --- /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)