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..2526c25e5 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/exception_events.py @@ -0,0 +1,108 @@ +"""Reconstruct exceptions from serialized error metadata for span events. + +The SDK checkpoints an ``ErrorObject`` (message/type/data/stack_trace) for a +failed operation or attempt. OpenTelemetry derives ``exception.type`` from the +runtime class passed to ``Span.record_exception``, so constructing a generic +``Exception`` collapses every failure to ``Exception`` and loses the original +type. These helpers rebuild an exception that carries the recorded type and +keep the original message and checkpointed stack trace. +""" + +from __future__ import annotations + +import builtins +import importlib +from typing import Any + +from aws_durable_execution_sdk_python.lambda_service import ErrorObject + + +_DEFAULT_MESSAGE = "Unknown error" + + +def _resolve_exception_class(type_name: str) -> type[BaseException] | None: + """Resolve a wire ErrorType to a real exception class, if available.""" + candidate: Any = getattr(builtins, type_name, None) + if isinstance(candidate, type) and issubclass(candidate, BaseException): + return candidate + if "." in type_name: + module_name, _, qualname = type_name.rpartition(".") + try: + module = importlib.import_module(module_name) + except ImportError: + return None + candidate = module + for part in qualname.split("."): + candidate = getattr(candidate, part, None) + if candidate is None: + return None + if isinstance(candidate, type) and issubclass(candidate, BaseException): + return candidate + return None + + +def _synthesize_exception_class(type_name: str) -> type[BaseException]: + """Build an exception class whose qualified name matches ``type_name``. + + Used when the original type is not importable. ``record_exception`` reads + ``__module__``/``__qualname__`` from the class, so a class carrying the + original name reproduces ``exception.type`` without rerunning user code. + """ + if "." in type_name: + module_name, _, qualname = type_name.rpartition(".") + else: + module_name, qualname = "builtins", type_name + return type(qualname, (Exception,), {"__module__": module_name}) + + +def exception_from_error_object(error: ErrorObject) -> BaseException: + """Rebuild an exception that preserves the SDK-recorded error type. + + Prefers a resolvable real class, then a synthesized class named after the + wire type, falling back to ``Exception`` only when no type is available. + """ + message: str = error.message if error.message is not None else "" + type_name: str | None = error.type + if not type_name: + return Exception(message or _DEFAULT_MESSAGE) + resolved: type[BaseException] | None = _resolve_exception_class(type_name) + cls: type[BaseException] = resolved or _synthesize_exception_class(type_name) + try: + return cls(message) + except Exception: # noqa: BLE001 - a class may reject positional args + exc: BaseException = cls.__new__(cls) + BaseException.__init__(exc, message) + return exc + + +def error_object_exception_attributes(error: ErrorObject) -> dict[str, Any]: + """Return explicit event attributes for a checkpointed error. + + OTel derives ``exception.message`` from ``str(exception)`` and auto-fills + ``exception.stacktrace`` from a live traceback; a rebuilt exception has + neither the exact message formatting (e.g. ``KeyError`` quotes its + argument) nor a traceback, so the recorded 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 2cc841a35..0fc5e053c 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 d6ab7f792..98b96bdbc 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..b33d14107 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 @@ -170,3 +173,160 @@ 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"), +] + + +@pytest.mark.parametrize(("error_type", "message"), CASES) +def test_operation_exception_event_preserves_original_type( + instrumented_view: InstrumentedView, error_type: str, message: str +) -> None: + """The operation exception event must keep the SDK-supplied error type.""" + plugin, exporter = instrumented_view + _start_invocation(plugin) + plugin.on_operation_end( + _operation_end_info(ErrorObject(message, error_type, 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.type"] == error_type + assert attributes["exception.message"] == message + + +@pytest.mark.parametrize(("error_type", "message"), CASES) +def test_attempt_exception_event_preserves_original_type( + instrumented_view: InstrumentedView, error_type: str, message: str +) -> None: + """A failed STEP attempt event must keep the SDK-supplied error type.""" + plugin, exporter = instrumented_view + _start_invocation(plugin) + _run_failed_step_attempt(plugin, ErrorObject(message, error_type, None, None)) + _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 + + +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"