From 2cd1c55db4f41b246f6fa5302279aaf362493bf3 Mon Sep 17 00:00:00 2001 From: Soumyajit Ghosh Date: Sat, 5 Sep 2026 09:42:00 +0530 Subject: [PATCH 1/3] 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 | 56 ++++++++++--- .../tests/trace/test_span_processor.py | 83 +++++++++++++++++++ 2 files changed, 126 insertions(+), 13 deletions(-) diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py index 14d6573d0cf..66243afd34d 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py @@ -177,21 +177,33 @@ 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: # pylint: disable=protected-access - sp._on_ending(span) + try: + sp._on_ending(span) + except Exception: # pylint: disable=broad-exception-caught + logger.exception("Exception while calling SpanProcessor._on_ending.") 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 @@ -214,7 +226,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 @@ -269,10 +285,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, @@ -305,19 +327,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 4f0fbee60a0..d9477fe6104 100644 --- a/opentelemetry-sdk/tests/trace/test_span_processor.py +++ b/opentelemetry-sdk/tests/trace/test_span_processor.py @@ -321,6 +321,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 66d86932d5c487443c49eb998caff4d3c3c92f8d Mon Sep 17 00:00:00 2001 From: Soumyajit Ghosh Date: Sat, 5 Sep 2026 09:43:17 +0530 Subject: [PATCH 2/3] 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 2c8cd343045f364f0b51955c69e4708aa74ebbb9 Mon Sep 17 00:00:00 2001 From: Soumyajit Ghosh Date: Wed, 9 Sep 2026 20:42:43 +0530 Subject: [PATCH 3/3] 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 66243afd34d..5109d6d26d3 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py @@ -325,6 +325,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: @@ -332,13 +333,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 d9477fe6104..178a7263233 100644 --- a/opentelemetry-sdk/tests/trace/test_span_processor.py +++ b/opentelemetry-sdk/tests/trace/test_span_processor.py @@ -404,6 +404,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() @@ -640,3 +657,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)