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 4 commits into
base: main
Choose a base branch
from
+189 −13
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
4 commits
Select commit Hold shift + click to select a range
2565d2d
sdk: catch and log unhandled exceptions in MultiSpanProcessor
somuai Sep 5, 2026
542c380
changelog: add entry for PR #5626
somuai Sep 5, 2026
a517a73
test(sdk): add tests for force_flush exception and executor submit fa…
somuai Sep 9, 2026
f186a47
fix(sdk): inspect completed futures during force_flush on timeout
somuai Oct 19, 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
54 changes: 41 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 @@ -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:
Expand All @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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)

@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 @@ -316,21 +335,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.

@somuai somuai Oct 19, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

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

Agreed @herin049. The span processor exception is raised on future.result() in the result evaluation loop below, not at submit(). The try/except around submit() was originally added to guard against executor shutdown at interpreter exit (RuntimeError: cannot schedule new futures after shutdown). Rebased cleanly on upstream main, updated force_flush to inspect completed futures on timeout, and all 48 unit tests pass cleanly.

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 = False

for future in done_futures:

@somuai somuai Oct 19, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

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

Resolved in commit f186a47: removed early return on non-empty not_done_futures, marked all_flushed = False, and continue processing done_futures so completed futures are inspected and any exceptions are logged. Added regression test test_force_flush_timeout_still_inspects_completed_failures.

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
147 changes: 147 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 @@ -472,6 +472,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 @@ -755,3 +855,50 @@ 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)

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()