diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/exception_events.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/exception_events.py new file mode 100644 index 000000000..872a7bedf --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/exception_events.py @@ -0,0 +1,55 @@ +"""Record serialized errors without reconstructing application exceptions. + +Use a plain exception as the event carrier and explicitly supply the original +checkpointed type, message, and stack trace. Resolving the original class is +unnecessary and could run application imports or constructors during replay. +""" + +from __future__ import annotations + +from typing import Any + +from aws_durable_execution_sdk_python.lambda_service import ErrorObject + + +_DEFAULT_MESSAGE = "Unknown error" + + +def exception_from_error_object(error: ErrorObject) -> Exception: + """Create a plain event carrier; attributes preserve the original details.""" + message: str = error.message if error.message is not None else "" + if not error.type: + message = message or _DEFAULT_MESSAGE + return Exception(message) + + +def error_object_exception_attributes(error: ErrorObject) -> dict[str, Any]: + """Return explicit event attributes for a checkpointed error. + + OTel derives exception attributes from the supplied exception. The plain + carrier has neither the original runtime type nor its traceback, so the + checkpointed values are supplied explicitly. + """ + attributes: dict[str, Any] = {} + if error.type is not None: + attributes["exception.type"] = error.type + if error.message is not None: + attributes["exception.message"] = error.message + if error.stack_trace: + attributes["exception.stacktrace"] = "\n".join(error.stack_trace) + return attributes + + +def record_error_object_exception(span: Any, error: ErrorObject | None) -> None: + """Record a span exception that preserves the SDK's original error type. + + ``error`` may be ``None`` for a failed outcome that carried no metadata; a + generic exception is recorded in that case. + """ + if error is None: + span.record_exception(Exception(_DEFAULT_MESSAGE)) + return + span.record_exception( + exception_from_error_object(error), + attributes=error_object_exception_attributes(error), + ) diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py index c01ef6a7d..356990ede 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py @@ -87,6 +87,9 @@ ensure_end_after_start, ) from aws_durable_execution_sdk_python_otel.execution_root import ExecutionRoot +from aws_durable_execution_sdk_python_otel.exception_events import ( + record_error_object_exception, +) from aws_durable_execution_sdk_python_otel.execution_trace_context import ( ExecutionTraceContext, canonical_trace_id, @@ -891,9 +894,7 @@ def on_operation_end(self, info: OperationEndInfo) -> None: if info.error: span.set_status(StatusCode.ERROR, info.error.message or "") - span.record_exception( - Exception(info.error.message or info.error.type or "Unknown error") - ) + record_error_object_exception(span, info.error) elif info.status is OperationStatus.SUCCEEDED: span.set_status(StatusCode.OK) @@ -1027,13 +1028,7 @@ def on_user_function_end(self, info: UserFunctionEndInfo) -> None: span.set_status( StatusCode.ERROR, info.error.message if info.error else "" ) - span.record_exception( - Exception( - (info.error.message or info.error.type) - if info.error - else "Unknown error" - ) - ) + record_error_object_exception(span, info.error) else: span.set_status(StatusCode.OK) diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py index 1b0c39604..4eab6c7c3 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py @@ -55,6 +55,9 @@ from aws_durable_execution_sdk_python_otel.durable_parent_span import ( DurableParentSpan, ) +from aws_durable_execution_sdk_python_otel.exception_events import ( + record_error_object_exception, +) from aws_durable_execution_sdk_python_otel.execution_root import ExecutionRoot from aws_durable_execution_sdk_python_otel.execution_trace_context import ( ExecutionTraceContext, @@ -962,9 +965,7 @@ def on_operation_end(self, info: OperationEndInfo) -> None: if info.error: span.set_status(StatusCode.ERROR, info.error.message or "") - span.record_exception( - Exception(info.error.message or info.error.type or "Unknown error") - ) + record_error_object_exception(span, info.error) elif info.status is OperationStatus.SUCCEEDED: span.set_status(StatusCode.OK) @@ -1087,13 +1088,7 @@ def on_user_function_end(self, info: UserFunctionEndInfo) -> None: span.set_status( StatusCode.ERROR, info.error.message if info.error else "" ) - span.record_exception( - Exception( - (info.error.message or info.error.type) - if info.error - else "Unknown error" - ) - ) + record_error_object_exception(span, info.error) else: span.set_status(StatusCode.OK) diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py index d1224a305..0b4bcc324 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py @@ -17,6 +17,9 @@ OperationEndInfo, OperationStartInfo, OperationType, + UserFunctionEndInfo, + UserFunctionOutcome, + UserFunctionStartInfo, ) from opentelemetry import context from opentelemetry.sdk.trace import TracerProvider @@ -29,6 +32,9 @@ InvocationOtelPlugin, OtelPluginConfig, ) +from aws_durable_execution_sdk_python_otel.exception_events import ( + exception_from_error_object, +) START = datetime(2026, 10, 1, tzinfo=UTC) @@ -170,3 +176,256 @@ def test_retry_then_resume_exports_retrying( assert spans[0].context is not None assert spans[1].context is not None assert spans[0].context.trace_id == spans[1].context.trace_id + + +def _start_invocation(plugin: Any) -> None: + plugin.on_invocation_start( + InvocationStartInfo( + request_id="request-1", + execution_arn=ARN, + execution_start_time=START, + is_first_invocation=True, + ) + ) + + +def _end_invocation(plugin: Any) -> None: + plugin.on_invocation_end( + InvocationEndInfo( + request_id="request-1", + execution_arn=ARN, + execution_start_time=START, + is_first_invocation=True, + status=InvocationStatus.SUCCEEDED, + ) + ) + + +def _operation_end_info( + error: ErrorObject | None, *, is_replayed: bool = False +) -> OperationEndInfo: + return OperationEndInfo( + operation_id="wait", + name="wait", + parent_id=None, + operation_type=OperationType.WAIT, + sub_type=OperationSubType.WAIT, + start_time=START, + is_replayed=is_replayed, + end_time=END, + status=OperationStatus.FAILED, + error=error, + ) + + +def _run_failed_step_attempt( + plugin: Any, error: ErrorObject | None, *, is_replayed: bool = False +) -> None: + operation: dict[str, Any] = dict( + operation_id="step-1", + operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="step-1", + parent_id=None, + start_time=START, + is_replayed=is_replayed, + status=OperationStatus.STARTED, + is_replay_children=False, + attempt=1, + ) + plugin.on_user_function_start(UserFunctionStartInfo(**operation)) + plugin.on_user_function_end( + UserFunctionEndInfo( + **operation, + outcome=UserFunctionOutcome.FAILED, + end_time=END, + error=error, + ) + ) + + +def _exception_attributes(span: Any) -> Any: + assert len(span.events) == 1 + assert span.events[0].name == "exception" + assert span.events[0].attributes is not None + return span.events[0].attributes + + +CASES = [ + ("ValueError", "example failure"), + ("KeyError", "missing key"), + ("TimeoutError", "deadline exceeded"), + ("custompkg.NotARealError", "boom"), + ("ExceptionGroup", "group failure (1 sub-exception)"), + ("BaseExceptionGroup", "base group failure (1 sub-exception)"), +] + + +@pytest.mark.parametrize("is_replayed", [False, True]) +@pytest.mark.parametrize(("error_type", "message"), CASES) +def test_operation_exception_event_preserves_original_type( + instrumented_view: InstrumentedView, + error_type: str, + message: str, + is_replayed: bool, +) -> None: + """The operation exception event must keep the SDK-supplied error type.""" + plugin, exporter = instrumented_view + _start_invocation(plugin) + stack_trace = ["Traceback from checkpoint", f"{error_type}: {message}"] + plugin.on_operation_end( + _operation_end_info( + ErrorObject(message, error_type, None, stack_trace), + is_replayed=is_replayed, + ) + ) + _end_invocation(plugin) + if is_replayed and isinstance(plugin, ExecutionOtelPlugin): + # The execution view suppresses duplicate logical operation spans. + assert all(s.name != "wait" for s in exporter.get_finished_spans()) + return + span = next(s for s in exporter.get_finished_spans() if s.name == "wait") + attributes = _exception_attributes(span) + assert attributes["exception.type"] == error_type + assert attributes["exception.message"] == message + assert attributes["exception.stacktrace"] == "\n".join(stack_trace) + + +@pytest.mark.parametrize("is_replayed", [False, True]) +@pytest.mark.parametrize(("error_type", "message"), CASES) +def test_attempt_exception_event_preserves_original_type( + instrumented_view: InstrumentedView, + error_type: str, + message: str, + is_replayed: bool, +) -> None: + """A failed STEP attempt event must keep the SDK-supplied error type.""" + plugin, exporter = instrumented_view + _start_invocation(plugin) + stack_trace = ["Traceback from checkpoint", f"{error_type}: {message}"] + _run_failed_step_attempt( + plugin, + ErrorObject(message, error_type, None, stack_trace), + is_replayed=is_replayed, + ) + _end_invocation(plugin) + span = next( + s for s in exporter.get_finished_spans() if s.name == "step-1 attempt 1" + ) + attributes = _exception_attributes(span) + assert attributes["exception.type"] == error_type + assert attributes["exception.message"] == message + assert attributes["exception.stacktrace"] == "\n".join(stack_trace) + + +def test_replay_attempt_event_uses_stored_error_metadata( + instrumented_view: InstrumentedView, +) -> None: + """A replayed attempt must recover type/message/stack from checkpoint data.""" + plugin, exporter = instrumented_view + _start_invocation(plugin) + stack_trace = [ + "Traceback (most recent call last):", + ' File "handler.py", line 1, in handler', + "ValueError: example failure", + ] + _run_failed_step_attempt( + plugin, + ErrorObject("example failure", "ValueError", None, stack_trace), + is_replayed=True, + ) + _end_invocation(plugin) + span = next( + s for s in exporter.get_finished_spans() if s.name == "step-1 attempt 1" + ) + attributes = _exception_attributes(span) + assert attributes["exception.type"] == "ValueError" + assert attributes["exception.message"] == "example failure" + assert attributes["exception.stacktrace"] == "\n".join(stack_trace) + + +def test_failed_operation_without_type_falls_back( + instrumented_view: InstrumentedView, +) -> None: + """Without type metadata the event still carries the original message.""" + plugin, exporter = instrumented_view + _start_invocation(plugin) + plugin.on_operation_end( + _operation_end_info(ErrorObject("no type here", None, None, None)) + ) + _end_invocation(plugin) + span = next(s for s in exporter.get_finished_spans() if s.name == "wait") + attributes = _exception_attributes(span) + assert attributes["exception.message"] == "no type here" + assert attributes["exception.type"] == "Exception" + + +@pytest.mark.parametrize("error_type", [None, ""]) +@pytest.mark.parametrize("message", [None, ""]) +def test_exception_from_error_object_uses_fallback_message( + error_type: str | None, + message: str | None, +) -> None: + """A missing or empty type with an empty message uses the default.""" + error = ErrorObject(message, error_type, None, None) + assert str(exception_from_error_object(error)) == "Unknown error" + + +def test_failed_operation_without_type_or_message_uses_fallback( + instrumented_view: InstrumentedView, +) -> None: + """Without type or message metadata the event uses the default message.""" + plugin, exporter = instrumented_view + _start_invocation(plugin) + plugin.on_operation_end(_operation_end_info(ErrorObject(None, None, None, None))) + _end_invocation(plugin) + span = next(s for s in exporter.get_finished_spans() if s.name == "wait") + attributes = _exception_attributes(span) + assert attributes["exception.message"] == "Unknown error" + assert attributes["exception.type"] == "Exception" + + +class ConstructorSideEffectError(Exception): + constructor_calls = 0 + + def __init__(self, message: str) -> None: + type(self).constructor_calls += 1 + super().__init__(message) + + +@pytest.mark.parametrize("is_replayed", [False, True]) +@pytest.mark.parametrize("event_kind", ["operation", "attempt"]) +def test_exception_event_does_not_repeat_custom_constructor( + instrumented_view: InstrumentedView, + monkeypatch: pytest.MonkeyPatch, + is_replayed: bool, + event_kind: str, +) -> None: + """Exporting stored error metadata must not rerun application code.""" + monkeypatch.setattr(ConstructorSideEffectError, "constructor_calls", 0) + original = ConstructorSideEffectError("custom failure") + error_type = f"{__name__}.ConstructorSideEffectError" + stack_trace = ["Traceback from checkpoint", f"{error_type}: {original}"] + error = ErrorObject(str(original), error_type, None, stack_trace) + plugin, exporter = instrumented_view + _start_invocation(plugin) + if event_kind == "operation": + plugin.on_operation_end(_operation_end_info(error, is_replayed=is_replayed)) + span_name = "wait" + else: + _run_failed_step_attempt(plugin, error, is_replayed=is_replayed) + span_name = "step-1 attempt 1" + _end_invocation(plugin) + assert ConstructorSideEffectError.constructor_calls == 1 + if ( + is_replayed + and event_kind == "operation" + and isinstance(plugin, ExecutionOtelPlugin) + ): + assert all(s.name != span_name for s in exporter.get_finished_spans()) + return + span = next(s for s in exporter.get_finished_spans() if s.name == span_name) + attributes = _exception_attributes(span) + assert attributes["exception.type"] == error_type + assert attributes["exception.message"] == str(original) + assert attributes["exception.stacktrace"] == "\n".join(stack_trace)