diff --git a/packages/aws-durable-execution-sdk-python-otel/README.md b/packages/aws-durable-execution-sdk-python-otel/README.md index 8e9ed0816..d9c503187 100644 --- a/packages/aws-durable-execution-sdk-python-otel/README.md +++ b/packages/aws-durable-execution-sdk-python-otel/README.md @@ -401,11 +401,28 @@ each durable span in the same invocation. ### Status attributes `durable.invocation.status` uses `RETRYING` when the core plugin hook reports -`InvocationStatus.RETRY`. The core enum remains unchanged. Operation spans use -OTel `OK` only for `SUCCEEDED`, `ERROR` when error details are delivered, and -`UNSET` for other outcomes without error details, including `FAILED`, -`CANCELLED`, `TIMED_OUT`, and `STOPPED`. The original durable operation status -remains in `durable.operation.status`. +`InvocationStatus.RETRY`, with OTel `UNSET`. The core enum remains unchanged. +Operation spans use the following mapping in both views: + +| Durable operation outcome | Error details | OTel status | Exception event | +| --- | --- | --- | --- | +| `SUCCEEDED` | Absent | `OK` | None | +| `FAILED`, `TIMED_OUT`, `STOPPED` | Absent | `ERROR` | None | +| `CANCELLED` and other non-success outcomes | Absent | `UNSET` | None | +| Any outcome | Present | `ERROR` | Available diagnostics | + +The original durable operation status remains in `durable.operation.status`. +Failure status does not fabricate an exception when diagnostics are absent, +including for failed step attempts. A caught operation failure leaves the +enclosing Invocation and Workflow spans successful if the handler succeeds. + +Terminal operation and attempt exception events use the same normalized end +timestamp as their span, including delayed or replayed completion delivery. +Execution view retains durable timing and normalizes an end at or before its +start; Invocation view retains its invocation-local timestamp ordering. A +checkpointless context with no durable start uses its completion time as the +lower bound in Execution view. When an end timestamp is absent, the current time +is chosen once for both the event and span end. ### Invocation context isolation 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 index 872a7bedf..aab273056 100644 --- 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 @@ -40,16 +40,19 @@ def error_object_exception_attributes(error: ErrorObject) -> dict[str, Any]: return attributes -def record_error_object_exception(span: Any, error: ErrorObject | None) -> None: +def record_error_object_exception( + span: Any, error: ErrorObject | None, *, timestamp: int +) -> 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. + A failed outcome without an error object has no diagnostic event. Terminal + callers supply the normalized span-end timestamp so delayed delivery does + not place an event outside the span. """ 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), + timestamp=timestamp, ) 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 356990ede..265c01074 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 @@ -892,16 +892,25 @@ def on_operation_end(self, info: OperationEndInfo) -> None: span_id_override=span_id_override, ) + end_timestamp = to_otel_timestamp( + end_time or datetime.datetime.now(datetime.UTC) + ) if info.error: span.set_status(StatusCode.ERROR, info.error.message or "") - record_error_object_exception(span, info.error) + record_error_object_exception(span, info.error, timestamp=end_timestamp) + elif info.status in { + OperationStatus.FAILED, + OperationStatus.TIMED_OUT, + OperationStatus.STOPPED, + }: + span.set_status(StatusCode.ERROR) elif info.status is OperationStatus.SUCCEEDED: span.set_status(StatusCode.OK) self._note_parent_end(info.parent_id, end_time) popped = self._pop_span(info.operation_id) if popped is not None: - popped.end(end_time=_to_otel_timestamp(end_time)) + popped.end(end_time=end_timestamp) def _start_span( self, @@ -1024,18 +1033,20 @@ def on_user_function_end(self, info: UserFunctionEndInfo) -> None: and info.outcome is not UserFunctionOutcome.INCOMPLETE ): span.set_attributes(self._operation_attributes(info)) + end_timestamp = to_otel_timestamp( + end_time or datetime.datetime.now(datetime.UTC) + ) if info.outcome is UserFunctionOutcome.FAILED: span.set_status( StatusCode.ERROR, info.error.message if info.error else "" ) - record_error_object_exception(span, info.error) + record_error_object_exception(span, info.error, timestamp=end_timestamp) else: span.set_status(StatusCode.OK) - end_time = ensure_end_after_start(info.start_time, info.end_time) popped = self._pop_span(key) if popped is not None: - popped.end(end_time=_to_otel_timestamp(end_time)) + popped.end(end_time=end_timestamp) # Restore the enclosing context by releasing the scope this user # function attached, so the parent operation (or, at the top level, the 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 4eab6c7c3..d54b97980 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 @@ -563,23 +563,22 @@ def _start_span( return span def _end_span( - self, operation_id: str | None, end_timestamp: datetime.datetime | None = None - ): + self, operation_id: str | None, end_timestamp: int | None = None + ) -> None: """End and unregister the active span for an operation ID. Args: operation_id: Durable operation ID, or ``None`` for the invocation span. - end_timestamp: Optional durable end timestamp to use as the span end - time. When omitted, OpenTelemetry uses the current time. + end_timestamp: Normalized nanoseconds; omit for the next ordered time. """ logger.debug("Ending OTel span: operation_id=%s", operation_id) with self._operation_spans_lock: span = self._operation_spans.pop(operation_id, None) if span: - # the span is not going to be populated if it has the same end_time and start_time - end_time = self._next_ordered_timestamp(end_timestamp) - span.end(end_time=end_time) + if end_timestamp is None: + end_timestamp = self._next_ordered_timestamp() + span.end(end_time=end_timestamp) logger.debug("Ended OTel span: %s", span) # ------------------------------------------------------------------ @@ -963,13 +962,20 @@ def on_operation_end(self, info: OperationEndInfo) -> None: else: span.set_attributes(self._extract_attributes(info)) + end_timestamp = self._next_ordered_timestamp(info.end_time) if info.error: span.set_status(StatusCode.ERROR, info.error.message or "") - record_error_object_exception(span, info.error) + record_error_object_exception(span, info.error, timestamp=end_timestamp) + elif info.status in { + OperationStatus.FAILED, + OperationStatus.TIMED_OUT, + OperationStatus.STOPPED, + }: + span.set_status(StatusCode.ERROR) elif info.status is OperationStatus.SUCCEEDED: span.set_status(StatusCode.OK) - self._end_span(info.operation_id, info.end_time) + self._end_span(info.operation_id, end_timestamp) def on_user_function_start(self, info: UserFunctionStartInfo) -> None: """Called when a context or step operation starts user code. @@ -1084,17 +1090,15 @@ def on_user_function_end(self, info: UserFunctionEndInfo) -> None: and info.outcome is not UserFunctionOutcome.INCOMPLETE ): span.set_attributes(self._extract_attributes(info)) + end_timestamp = self._next_ordered_timestamp(info.end_time) if info.outcome is UserFunctionOutcome.FAILED: span.set_status( StatusCode.ERROR, info.error.message if info.error else "" ) - record_error_object_exception(span, info.error) + record_error_object_exception(span, info.error, timestamp=end_timestamp) else: span.set_status(StatusCode.OK) - end_timestamp = info.end_time - if end_timestamp is not None and end_timestamp == info.start_time: - end_timestamp += datetime.timedelta(microseconds=1) self._end_span(span_key, end_timestamp) # Restore the enclosing context by releasing the scope this user # function attached. Code that runs after this operation (e.g. between diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_failure_bounds_int.py b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_failure_bounds_int.py new file mode 100644 index 000000000..79dd2eda4 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-otel/tests/e2e/test_terminal_failure_bounds_int.py @@ -0,0 +1,134 @@ +"""Caught operation failures preserve telemetry through real suspension/replay.""" + +import json +from typing import Any + +import pytest +from aws_durable_execution_sdk_python import DurableContext, durable_execution +from aws_durable_execution_sdk_python.config import ChildConfig, Duration +from aws_durable_execution_sdk_python.exceptions import ChildContextError, InvokeError +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + InvocationStartInfo, + OperationEndInfo, +) +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from opentelemetry.trace import StatusCode + +from aws_durable_execution_sdk_python_otel import ( + ExecutionOtelPluginFactory, + InvocationOtelPluginFactory, + OtelPluginConfig, +) + + +@pytest.mark.parametrize( + "view", [ExecutionOtelPluginFactory, InvocationOtelPluginFactory] +) +@pytest.mark.parametrize("failure_kind", ["child", "virtual", "invoke"]) +@pytest.mark.parametrize("resume", [False, True]) +def test_caught_failure_keeps_events_bounded_and_workflow_successful( + view: type[ExecutionOtelPluginFactory] | type[InvocationOtelPluginFactory], + failure_kind: str, + resume: bool, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("DURABLE_EXECUTION_PLUGINS", raising=False) + provider = TracerProvider() + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + factory = view( + OtelPluginConfig( + tracer_provider=provider, + enrich_logger=False, + context_extractor=lambda _: None, + ) + ) + saved_calls: list[str] = [] + failure_calls: list[str] = [] + observations: list[OperationEndInfo] = [] + + class Observe(DurableInstrumentationPlugin): + def on_operation_end(self, info: OperationEndInfo) -> None: + if info.name == "caught-failure": + observations.append(info) + + class ObserveFactory: + def create_plugin(self, _info: InvocationStartInfo) -> Observe: + return Observe() + + def saved_step(_context: Any) -> str: + saved_calls.append("saved") + return "saved" + + def failing_child(_context: DurableContext) -> str: + failure_calls.append("failed") + raise ValueError("actual child failure") + + def failing_function(_event: Any, _context: Any) -> None: + failure_calls.append("failed") + raise ValueError("actual invoke failure") + + def body(_event: Any, context: DurableContext) -> str: + saved = context.step(saved_step, name="saved-step") + try: + if failure_kind == "invoke": + context.invoke("failing-target", {}, name="caught-failure") + else: + context.run_in_child_context( + failing_child, + name="caught-failure", + config=ChildConfig(is_virtual=failure_kind == "virtual"), + ) + except (ChildContextError, InvokeError): + if resume: + context.wait(Duration.from_seconds(1), name="replay-wait") + return saved + raise AssertionError("Expected the original operation failure") + + try: + with DurableFunctionTestRunner( + handler=durable_execution(body, plugins=[factory, ObserveFactory()]), + poll_interval=0.01, + ) as runner: + runner.register_function("failing-target", failing_function) + result = runner.run(input="{}", execution_timeout=15) + assert result.status.value == "SUCCEEDED" + assert result.result is not None and json.loads(result.result) == "saved" + assert saved_calls == ["saved"] + # Virtual contexts intentionally execute on replay; checkpointed child + # failures and invoke outcomes must return their stored failure instead. + assert len(failure_calls) == (2 if resume and failure_kind == "virtual" else 1) + assert observations + if failure_kind == "virtual": + assert all(info.start_time is None for info in observations) + spans = exporter.get_finished_spans() + failed = [ + span + for span in spans + if span.name == "caught-failure" + and span.attributes is not None + and span.attributes.get("durable.operation.status") == "FAILED" + ] + assert failed + for span in failed: + assert span.status.status_code is StatusCode.ERROR + (event,) = span.events + assert span.start_time is not None and span.end_time is not None + assert span.start_time <= event.timestamp == span.end_time + assert event.attributes is not None + assert event.attributes["exception.message"] in { + "actual child failure", + "actual invoke failure", + } + invocations = [span for span in spans if span.name == "Invocation"] + # An invoke can add its own suspension before the explicit replay wait. + assert len(invocations) >= (2 if resume else 1) + assert all(span.status.status_code is StatusCode.OK for span in invocations) + (workflow,) = [span for span in spans if span.name == "Workflow"] + assert workflow.status.status_code is StatusCode.OK + finally: + provider.shutdown() diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py index 27547c76c..09a132dc7 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py @@ -6,7 +6,7 @@ import threading import time from concurrent.futures import ThreadPoolExecutor -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from types import SimpleNamespace import opentelemetry.context as otel_context @@ -2061,6 +2061,38 @@ def test_incomplete_attempt_is_marked_when_invocation_ends(): assert attempt.attributes["durable.span.truncated_at_invocation_boundary"] is True +def test_invocation_cleanup_keeps_future_attempt_bounded() -> None: + """Cleanup remains ordered when an attempt started ahead of the local clock.""" + plugin, exporter = _create_plugin() + start_info = UserFunctionStartInfo( + operation_id="future-attempt", + operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="future-attempt", + parent_id=None, + start_time=datetime.now(UTC) + timedelta(days=1), + is_replayed=False, + status=OperationStatus.STARTED, + is_replay_children=False, + attempt=1, + ) + plugin.on_invocation_start(_invocation_start_info()) + plugin.on_user_function_start(start_info) + plugin.on_user_function_end( + UserFunctionEndInfo.from_start_info( + start_info, error=None, outcome=UserFunctionOutcome.INCOMPLETE + ) + ) + plugin.on_invocation_end(_invocation_end_info(status=InvocationStatus.PENDING)) + + spans = {span.name: span for span in exporter.get_finished_spans()} + attempt = spans["future-attempt attempt 1"] + invocation = spans["Invocation"] + assert attempt.start_time is not None and attempt.end_time is not None + assert invocation.end_time is not None + assert attempt.start_time < attempt.end_time < invocation.end_time + + def test_workflow_span_name_is_configurable(): """The Workflow span name can be overridden via constructor kwarg.""" exporter = InMemorySpanExporter() 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 0b4bcc324..cbf47a88a 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 @@ -1,6 +1,7 @@ """Shared status regressions through both public plugin lifecycles.""" from collections.abc import Iterator +from dataclasses import replace from datetime import UTC, datetime, timedelta from typing import Any @@ -69,13 +70,7 @@ def instrumented_view(request: pytest.FixtureRequest) -> Iterator[InstrumentedVi @pytest.mark.parametrize( "status", - [ - OperationStatus.SUCCEEDED, - OperationStatus.FAILED, - OperationStatus.CANCELLED, - OperationStatus.TIMED_OUT, - OperationStatus.STOPPED, - ], + list(OperationStatus), ) @pytest.mark.parametrize("with_error", [False, True]) @pytest.mark.parametrize("started_here", [False, True]) @@ -129,6 +124,8 @@ def test_operation_status_and_exception( expected = ( StatusCode.ERROR if with_error + or status + in {OperationStatus.FAILED, OperationStatus.TIMED_OUT, OperationStatus.STOPPED} else ( StatusCode.OK if status is OperationStatus.SUCCEEDED else StatusCode.UNSET ) @@ -144,6 +141,124 @@ def test_operation_status_and_exception( assert not span.events +@pytest.mark.parametrize( + ("start_time", "expected_durable_end"), + [ + (None, END + timedelta(microseconds=1)), + (START, END), + (END, END + timedelta(microseconds=1)), + (END + timedelta(seconds=1), END + timedelta(seconds=1, microseconds=1)), + ], +) +@pytest.mark.parametrize("end_time", [END, None]) +def test_terminal_exception_uses_normalized_span_end( + instrumented_view: InstrumentedView, + start_time: datetime | None, + expected_durable_end: datetime, + end_time: datetime | None, +) -> None: + """Execution view preserves durable ends, including delayed completions.""" + plugin, exporter = instrumented_view + _start_invocation(plugin) + before_callback = datetime.now(UTC) + plugin.on_operation_end( + replace( + _operation_end_info( + ErrorObject("stored failure", "ValueError", None, None) + ), + operation_type=OperationType.CONTEXT, + sub_type=OperationSubType.RUN_IN_CHILD_CONTEXT, + start_time=start_time, + end_time=end_time, + ) + ) + after_callback = datetime.now(UTC) + _end_invocation(plugin) + span = next(s for s in exporter.get_finished_spans() if s.name == "wait") + (event,) = span.events + assert span.start_time is not None and span.end_time is not None + assert span.start_time <= event.timestamp == span.end_time + if isinstance(plugin, ExecutionOtelPlugin): + if end_time is None: + assert ( + int(before_callback.timestamp() * 1_000_000_000) + <= span.end_time + <= int(after_callback.timestamp() * 1_000_000_000) + ) + else: + assert span.end_time == int( + expected_durable_end.timestamp() * 1_000_000_000 + ) + + +@pytest.mark.parametrize("with_error", [False, True]) +@pytest.mark.parametrize( + ("end_time", "expected_durable_end"), + [ + (START - timedelta(seconds=1), START + timedelta(microseconds=1)), + (START, START + timedelta(microseconds=1)), + (END, END), + (None, None), + ], +) +def test_failed_attempt_exception_uses_normalized_span_end( + instrumented_view: InstrumentedView, + with_error: bool, + end_time: datetime | None, + expected_durable_end: datetime | None, +) -> None: + """An attempt failure reports its outcome without fabricating diagnostics.""" + plugin, exporter = instrumented_view + _start_invocation(plugin) + operation: dict[str, Any] = dict( + operation_id="attempt", + operation_type=OperationType.STEP, + sub_type=OperationSubType.STEP, + name="attempt", + parent_id=None, + start_time=START, + is_replayed=False, + status=OperationStatus.STARTED, + is_replay_children=False, + attempt=1, + ) + plugin.on_user_function_start(UserFunctionStartInfo(**operation)) + before_callback = datetime.now(UTC) + plugin.on_user_function_end( + UserFunctionEndInfo( + **operation, + outcome=UserFunctionOutcome.FAILED, + end_time=end_time, + error=ErrorObject("stored failure", "ValueError", None, None) + if with_error + else None, + ) + ) + after_callback = datetime.now(UTC) + _end_invocation(plugin) + span = next( + s for s in exporter.get_finished_spans() if s.name == "attempt attempt 1" + ) + assert span.status.status_code is StatusCode.ERROR + assert span.start_time is not None and span.end_time is not None + if isinstance(plugin, ExecutionOtelPlugin): + if expected_durable_end is None: + assert ( + int(before_callback.timestamp() * 1_000_000_000) + <= span.end_time + <= int(after_callback.timestamp() * 1_000_000_000) + ) + else: + assert span.end_time == int( + expected_durable_end.timestamp() * 1_000_000_000 + ) + if with_error: + (event,) = span.events + assert span.start_time <= event.timestamp == span.end_time + else: + assert not span.events + + def test_retry_then_resume_exports_retrying( instrumented_view: InstrumentedView, ) -> None: