From 2565d2d8c7971231b51657f3cfaf4fa9f34e2dac 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 6f47f213bf..fa5432e844 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 f2595b9669..ef5c555a9c 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 542c380347e4cd9acabb55e25be7b1297f8b1bd8 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 0000000000..97356127fe --- /dev/null +++ b/.changelog/5626.fixed @@ -0,0 +1 @@ +`opentelemetry-sdk`: catch and log unhandled exceptions in MultiSpanProcessor From a517a73b5d2c45935d17200fa522c805941a66e0 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 fa5432e844..f1295a0ef9 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 ef5c555a9c..7d8515977e 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 f186a47be5ef7c59559722b263080c10210d3b99 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 f1295a0ef9..3b85cce09a 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 7d8515977e..db289676df 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()