From 9a7a2ea2ccbc7f1e83fcc90e8dffaa0062f28a9d Mon Sep 17 00:00:00 2001 From: Soumyajit Ghosh Date: Sat, 5 Sep 2026 09:42:00 +0530 Subject: [PATCH 1/4] sdk: catch and log unhandled exceptions in MultiSpanProcessor Fixes #5624 Per the OpenTelemetry specification: 1. OnEnd(Span) and OnStart(Span) MUST NOT throw an exception. 2. OpenTelemetry implementations MUST NOT throw unhandled exceptions at runtime. Previously, SynchronousMultiSpanProcessor and ConcurrentMultiSpanProcessor did not catch exceptions raised by underlying SpanProcessor instances during on_start, _on_ending, on_end, shutdown, and force_flush. An unhandled exception in on_end during context manager exit (__exit__) would escape and replace or suppress real application exceptions, or crash the application during normal execution. Additionally, a failure in an earlier processor would prevent subsequent processors from being invoked. Wrap underlying processor calls in try-except blocks, logging exceptions with logger.exception so that application execution is never disrupted and all registered processors are reliably called. Signed-off-by: Soumyajit Ghosh Signed-off-by: SOUMYAJIT GHOSH <23051387@kiit.ac.in> --- .../src/opentelemetry/sdk/trace/__init__.py | 51 +++++++++--- .../tests/trace/test_span_processor.py | 83 +++++++++++++++++++ 2 files changed, 122 insertions(+), 12 deletions(-) diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py index 6f47f213bff..fa5432e8447 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py @@ -178,7 +178,10 @@ def on_start( parent_context: context_api.Context | None = None, ) -> None: for sp in self._span_processors: - sp.on_start(span, parent_context=parent_context) + try: + sp.on_start(span, parent_context=parent_context) + except Exception: # pylint: disable=broad-exception-caught + logger.exception("Exception while calling SpanProcessor.on_start.") def _on_ending(self, span: "Span") -> None: for sp in self._span_processors: @@ -190,12 +193,18 @@ def _on_ending(self, span: "Span") -> None: def on_end(self, span: "ReadableSpan") -> None: for sp in self._span_processors: - sp.on_end(span) + try: + sp.on_end(span) + except Exception: # pylint: disable=broad-exception-caught + logger.exception("Exception while calling SpanProcessor.on_end.") def shutdown(self) -> None: """Sequentially shuts down all underlying span processors.""" for sp in self._span_processors: - sp.shutdown() + try: + sp.shutdown() + except Exception: # pylint: disable=broad-exception-caught + logger.exception("Exception while calling SpanProcessor.shutdown.") def force_flush(self, timeout_millis: int = 30000) -> bool: """Sequentially calls force_flush on all underlying @@ -218,7 +227,11 @@ def force_flush(self, timeout_millis: int = 30000) -> bool: if current_time_ns >= deadline_ns: return False - if sp.force_flush((deadline_ns - current_time_ns) // 1000000) is False: + try: + if sp.force_flush((deadline_ns - current_time_ns) // 1000000) is False: + all_flushed = False + except Exception: # pylint: disable=broad-exception-caught + logger.exception("Exception while calling SpanProcessor.force_flush.") all_flushed = False return all_flushed @@ -275,10 +288,16 @@ 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) + futures.append(future) + except Exception: # pylint: disable=broad-exception-caught + logger.exception("Exception while submitting task to SpanProcessor executor.") for future in futures: - future.result() + try: + future.result() + except Exception: # pylint: disable=broad-exception-caught + logger.exception("Exception while executing SpanProcessor task.") def on_start( self, @@ -318,19 +337,27 @@ 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) - futures.append(future) + try: + future = self._executor.submit(sp.force_flush, timeout_millis) + futures.append(future) + except Exception: # pylint: disable=broad-exception-caught + logger.exception("Exception while submitting task to SpanProcessor executor.") timeout_sec = timeout_millis / 1e3 done_futures, not_done_futures = concurrent.futures.wait(futures, timeout_sec) if not_done_futures: return False + all_flushed = True for future in done_futures: - if future.result() is False: - return False + try: + if future.result() is False: + all_flushed = False + except Exception: # pylint: disable=broad-exception-caught + logger.exception("Exception while executing SpanProcessor force_flush.") + all_flushed = False - return True + return all_flushed class EventBase(abc.ABC): diff --git a/opentelemetry-sdk/tests/trace/test_span_processor.py b/opentelemetry-sdk/tests/trace/test_span_processor.py index f2595b9669d..ef5c555a9c6 100644 --- a/opentelemetry-sdk/tests/trace/test_span_processor.py +++ b/opentelemetry-sdk/tests/trace/test_span_processor.py @@ -472,6 +472,89 @@ def test_on_shutdown(self): for mock_processor in mocks: mock_processor.shutdown.assert_called_once_with() + def test_on_end_exception_does_not_raise(self): + multi_processor = self.create_multi_span_processor() + raising_proc = mock.Mock(spec=trace.SpanProcessor) + raising_proc.on_end.side_effect = RuntimeError("processor failed") + normal_proc = mock.Mock(spec=trace.SpanProcessor) + + multi_processor.add_span_processor(raising_proc) + multi_processor.add_span_processor(normal_proc) + + span = self.create_default_span() + multi_processor.on_end(span) + + raising_proc.on_end.assert_called_once_with(span) + normal_proc.on_end.assert_called_once_with(span) + multi_processor.shutdown() + + def test_on_start_exception_does_not_raise(self): + multi_processor = self.create_multi_span_processor() + raising_proc = mock.Mock(spec=trace.SpanProcessor) + raising_proc.on_start.side_effect = RuntimeError("processor failed") + normal_proc = mock.Mock(spec=trace.SpanProcessor) + + multi_processor.add_span_processor(raising_proc) + multi_processor.add_span_processor(normal_proc) + + span = self.create_default_span() + multi_processor.on_start(span) + + raising_proc.on_start.assert_called_once_with(span, parent_context=None) + normal_proc.on_start.assert_called_once_with(span, parent_context=None) + multi_processor.shutdown() + + def test_on_ending_exception_does_not_raise(self): + multi_processor = self.create_multi_span_processor() + raising_proc = mock.Mock(spec=trace.SpanProcessor) + # pylint: disable=protected-access + raising_proc._on_ending.side_effect = RuntimeError("processor failed") + normal_proc = mock.Mock(spec=trace.SpanProcessor) + + multi_processor.add_span_processor(raising_proc) + multi_processor.add_span_processor(normal_proc) + + span = self.create_default_span() + # pylint: disable=protected-access + multi_processor._on_ending(span) + + # pylint: disable=protected-access + raising_proc._on_ending.assert_called_once_with(span) + normal_proc._on_ending.assert_called_once_with(span) + multi_processor.shutdown() + + def test_shutdown_exception_does_not_raise(self): + multi_processor = self.create_multi_span_processor() + raising_proc = mock.Mock(spec=trace.SpanProcessor) + raising_proc.shutdown.side_effect = RuntimeError("processor failed") + normal_proc = mock.Mock(spec=trace.SpanProcessor) + + multi_processor.add_span_processor(raising_proc) + multi_processor.add_span_processor(normal_proc) + + multi_processor.shutdown() + + raising_proc.shutdown.assert_called_once_with() + normal_proc.shutdown.assert_called_once_with() + + def test_on_end_exception_does_not_replace_application_exception(self): + tracer_provider = trace.TracerProvider() + tracer = tracer_provider.get_tracer(__name__) + + multi_processor = self.create_multi_span_processor() + raising_proc = mock.Mock(spec=trace.SpanProcessor) + raising_proc.on_end.side_effect = RuntimeError("processor failed") + multi_processor.add_span_processor(raising_proc) + + tracer_provider.add_span_processor(multi_processor) + + with self.assertRaises(ValueError) as ctx: + with tracer.start_as_current_span("app_span"): + raise ValueError("application failed") + + self.assertEqual(str(ctx.exception), "application failed") + multi_processor.shutdown() + def test_force_flush(self): multi_processor = self.create_multi_span_processor() From 8133f59750e80b3d44e475297b8ffde2be735744 Mon Sep 17 00:00:00 2001 From: Soumyajit Ghosh Date: Sat, 5 Sep 2026 09:43:17 +0530 Subject: [PATCH 2/4] changelog: add entry for PR #5626 Signed-off-by: SOUMYAJIT GHOSH <23051387@kiit.ac.in> --- .changelog/5626.fixed | 1 + 1 file changed, 1 insertion(+) create mode 100644 .changelog/5626.fixed diff --git a/.changelog/5626.fixed b/.changelog/5626.fixed new file mode 100644 index 00000000000..97356127fec --- /dev/null +++ b/.changelog/5626.fixed @@ -0,0 +1 @@ +`opentelemetry-sdk`: catch and log unhandled exceptions in MultiSpanProcessor From c43a4588dcdc5600d9101b88a474b256c6546f2c Mon Sep 17 00:00:00 2001 From: Soumyajit Ghosh Date: Wed, 9 Sep 2026 20:42:43 +0530 Subject: [PATCH 3/4] test(sdk): add tests for force_flush exception and executor submit failure - Cover force_flush exception handling in MultiSpanProcessorTestBase ensuring it returns False without propagating the exception. - Cover ConcurrentMultiSpanProcessor submission failure in _submit_and_await and force_flush when ThreadPoolExecutor is shut down. - Set all_flushed = False in ConcurrentMultiSpanProcessor.force_flush if submit raises. Signed-off-by: Soumyajit Ghosh --- .../src/opentelemetry/sdk/trace/__init__.py | 3 +- .../tests/trace/test_span_processor.py | 38 +++++++++++++++++++ 2 files changed, 40 insertions(+), 1 deletion(-) diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py index fa5432e8447..f1295a0ef92 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py @@ -335,6 +335,7 @@ def force_flush(self, timeout_millis: int = 30000) -> bool: True if all span processors flushed their spans within the given timeout, False otherwise. """ + all_flushed = True futures = [] for sp in self._span_processors: try: @@ -342,13 +343,13 @@ def force_flush(self, timeout_millis: int = 30000) -> bool: futures.append(future) except Exception: # pylint: disable=broad-exception-caught logger.exception("Exception while submitting task to SpanProcessor executor.") + all_flushed = False timeout_sec = timeout_millis / 1e3 done_futures, not_done_futures = concurrent.futures.wait(futures, timeout_sec) if not_done_futures: return False - all_flushed = True for future in done_futures: try: if future.result() is False: diff --git a/opentelemetry-sdk/tests/trace/test_span_processor.py b/opentelemetry-sdk/tests/trace/test_span_processor.py index ef5c555a9c6..7d8515977e6 100644 --- a/opentelemetry-sdk/tests/trace/test_span_processor.py +++ b/opentelemetry-sdk/tests/trace/test_span_processor.py @@ -555,6 +555,23 @@ def test_on_end_exception_does_not_replace_application_exception(self): self.assertEqual(str(ctx.exception), "application failed") multi_processor.shutdown() + def test_force_flush_exception_does_not_raise_and_returns_false(self): + multi_processor = self.create_multi_span_processor() + raising_proc = mock.Mock(spec=trace.SpanProcessor) + raising_proc.force_flush.side_effect = RuntimeError("processor failed") + normal_proc = mock.Mock(spec=trace.SpanProcessor) + normal_proc.force_flush.return_value = True + + multi_processor.add_span_processor(raising_proc) + multi_processor.add_span_processor(normal_proc) + + flushed = multi_processor.force_flush(1000) + + self.assertFalse(flushed) + raising_proc.force_flush.assert_called_once() + normal_proc.force_flush.assert_called_once() + multi_processor.shutdown() + def test_force_flush(self): multi_processor = self.create_multi_span_processor() @@ -838,3 +855,24 @@ def child(conn): with tracer.start_as_current_span("main process after fork span"): pass assert exporter.get_finished_spans()[-1].name == "main process after fork span" + + def test_executor_submit_exception_does_not_raise(self): + multi_processor = trace.ConcurrentMultiSpanProcessor(2) + # Shut down the executor so that subsequent submit calls raise RuntimeError + # pylint: disable=protected-access + multi_processor._executor.shutdown(wait=True) + + mock_processor = mock.Mock(spec=trace.SpanProcessor) + multi_processor.add_span_processor(mock_processor) + + span = self.create_default_span() + # Verify that all callbacks handle executor submission failures gracefully + multi_processor.on_start(span) + # pylint: disable=protected-access + multi_processor._on_ending(span) + multi_processor.on_end(span) + multi_processor.shutdown() + + # force_flush must return False when submission fails and not raise + flushed = multi_processor.force_flush(100) + self.assertFalse(flushed) From 15e3dfd7bae0d1d524abcaed2355f819787c9465 Mon Sep 17 00:00:00 2001 From: Soumyajit Ghosh Date: Mon, 19 Oct 2026 09:05:21 +0530 Subject: [PATCH 4/4] fix(sdk): inspect completed futures during force_flush on timeout Signed-off-by: Soumyajit Ghosh --- .../src/opentelemetry/sdk/trace/__init__.py | 2 +- .../tests/trace/test_span_processor.py | 26 +++++++++++++++++++ 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py index f1295a0ef92..3b85cce09a1 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py @@ -348,7 +348,7 @@ def force_flush(self, timeout_millis: int = 30000) -> bool: timeout_sec = timeout_millis / 1e3 done_futures, not_done_futures = concurrent.futures.wait(futures, timeout_sec) if not_done_futures: - return False + all_flushed = False for future in done_futures: try: diff --git a/opentelemetry-sdk/tests/trace/test_span_processor.py b/opentelemetry-sdk/tests/trace/test_span_processor.py index 7d8515977e6..db289676df5 100644 --- a/opentelemetry-sdk/tests/trace/test_span_processor.py +++ b/opentelemetry-sdk/tests/trace/test_span_processor.py @@ -876,3 +876,29 @@ def test_executor_submit_exception_does_not_raise(self): # force_flush must return False when submission fails and not raise flushed = multi_processor.force_flush(100) self.assertFalse(flushed) + + def test_force_flush_timeout_still_inspects_completed_failures(self): + multi_processor = trace.ConcurrentMultiSpanProcessor(2) + raising_proc = mock.Mock(spec=trace.SpanProcessor) + raising_proc.force_flush.side_effect = RuntimeError("flush failed") + + blocking_proc = mock.Mock(spec=trace.SpanProcessor) + block_event = Event() + + def _slow_flush(timeout_millis=None): + block_event.wait(timeout=2.0) + return True + + blocking_proc.force_flush.side_effect = _slow_flush + + multi_processor.add_span_processor(raising_proc) + multi_processor.add_span_processor(blocking_proc) + + try: + with self.assertLogs("opentelemetry.sdk.trace", level="ERROR") as cm: + flushed = multi_processor.force_flush(timeout_millis=50) + self.assertFalse(flushed) + self.assertTrue(any("Exception while executing SpanProcessor force_flush." in msg for msg in cm.output)) + finally: + block_event.set() + multi_processor.shutdown()