Skip to content

fix(sdk): catch and log unhandled exceptions in MultiSpanProcessor #5626

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
somuai wants to merge 3 commits into
base: main
Choose a base branch
from
Open

fix(sdk): catch and log unhandled exceptions in MultiSpanProcessor #5626

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
2cd1c55
sdk: catch and log unhandled exceptions in MultiSpanProcessor
somuai Sep 5, 2026
66d8693
changelog: add entry for PR #5626
somuai Sep 5, 2026
2c8cd34
test(sdk): add tests for force_flush exception and executor submit fa…
somuai Sep 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .changelog/5626.fixed
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
`opentelemetry-sdk`: catch and log unhandled exceptions in MultiSpanProcessor
57 changes: 44 additions & 13 deletions opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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)

@herin049 herin049 Oct 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same issue here.

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,
Expand Down Expand Up @@ -303,21 +325,30 @@ 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:
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

@herin049 herin049 Oct 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The exception potentially raised from the span processor won't be raised here, it'll be raised on the result below. Likewise the call to submit should never raise, unless the executor is shutdown which indicates a bug in the implementation.

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

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):
Expand Down
121 changes: 121 additions & 0 deletions opentelemetry-sdk/tests/trace/test_span_processor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -321,6 +321,106 @@ 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_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()

Expand Down Expand Up @@ -557,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)