diff --git a/packages/aws-durable-execution-sdk-python-otel/README.md b/packages/aws-durable-execution-sdk-python-otel/README.md index 5e78f2a9c..bf30b2733 100644 --- a/packages/aws-durable-execution-sdk-python-otel/README.md +++ b/packages/aws-durable-execution-sdk-python-otel/README.md @@ -297,6 +297,15 @@ The resolved decision is applied to Workflow, Invocation, operation, and attempt spans. This avoids independently querying stateful or ratio-based samplers for 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`. + ### Log Correlation When `enrich_logger=True` (the default), the plugin installs a logging filter on 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 1a46cafcb..dba9bda80 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 @@ -43,6 +43,7 @@ InvocationStartInfo, OperationEndInfo, OperationStartInfo, + OperationStatus, OperationType, UserFunctionEndInfo, UserFunctionOutcome, @@ -607,7 +608,9 @@ def on_invocation_end(self, info: InvocationEndInfo) -> None: if self._invocation_span is not None: self._invocation_span.set_attribute( "durable.invocation.status", - info.status.value if info.status else "", + "RETRYING" + if info.status is InvocationStatus.RETRY + else info.status.value, ) if info.status in (InvocationStatus.SUCCEEDED, InvocationStatus.PENDING): self._invocation_span.set_status(StatusCode.OK) @@ -726,7 +729,7 @@ def on_operation_end(self, info: OperationEndInfo) -> None: span.record_exception( Exception(info.error.message or info.error.type or "Unknown error") ) - else: + elif info.status is OperationStatus.SUCCEEDED: span.set_status(StatusCode.OK) self._note_parent_end(info.parent_id, end_time) 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 012f17f2f..1c4f74a34 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 @@ -14,6 +14,7 @@ InvocationStartInfo, OperationEndInfo, OperationStartInfo, + OperationStatus, OperationType, UserFunctionEndInfo, UserFunctionOutcome, @@ -676,7 +677,10 @@ def on_invocation_end(self, info: InvocationEndInfo) -> None: invocation_span = self._get_span(None) if invocation_span: invocation_span.set_attribute( - "durable.invocation.status", info.status.value + "durable.invocation.status", + "RETRYING" + if info.status is InvocationStatus.RETRY + else info.status.value, ) # Span status mapping: SUCCEEDED/PENDING -> OK, FAILED -> ERROR, # RETRY -> UNSET. RETRY is left UNSET because the plugin interface @@ -780,7 +784,7 @@ def on_operation_end(self, info: OperationEndInfo) -> None: span.record_exception( Exception(info.error.message or info.error.type or "Unknown error") ) - else: + elif info.status is OperationStatus.SUCCEEDED: span.set_status(StatusCode.OK) self._end_span(info.operation_id, info.end_time) diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py index 5c3cc3eca..12edcc0ae 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py @@ -1123,7 +1123,9 @@ def test_invocation_span_status_kind_and_attributes(status, expected_code): invocation = {s.name: s for s in exporter.get_finished_spans()}["Invocation"] assert invocation.kind is trace.SpanKind.INTERNAL assert invocation.attributes is not None - assert invocation.attributes["durable.invocation.status"] == status.value + assert invocation.attributes["durable.invocation.status"] == ( + "RETRYING" if status is InvocationStatus.RETRY else status.value + ) assert invocation.attributes["durable.invocation.first"] is True assert invocation.status.status_code is expected_code 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 f108ef728..1894d2b48 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 @@ -563,7 +563,11 @@ def test_invocation_span_status_reflects_execution_status( invocation = next(s for s in spans if s.name == "Invocation") attributes = invocation.attributes assert attributes is not None - assert attributes["durable.invocation.status"] == invocation_status.value + assert attributes["durable.invocation.status"] == ( + "RETRYING" + if invocation_status is InvocationStatus.RETRY + else invocation_status.value + ) assert invocation.status.status_code is expected_span_status 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 new file mode 100644 index 000000000..d1224a305 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_status_mapping.py @@ -0,0 +1,172 @@ +"""Shared status regressions through both public plugin lifecycles.""" + +from collections.abc import Iterator +from datetime import UTC, datetime, timedelta +from typing import Any + +import pytest +from aws_durable_execution_sdk_python.lambda_service import ( + ErrorObject, + OperationStatus, + OperationSubType, +) +from aws_durable_execution_sdk_python.plugin import ( + InvocationEndInfo, + InvocationStartInfo, + InvocationStatus, + OperationEndInfo, + OperationStartInfo, + OperationType, +) +from opentelemetry import context +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 ( + ExecutionOtelPlugin, + InvocationOtelPlugin, + OtelPluginConfig, +) + + +START = datetime(2026, 10, 1, tzinfo=UTC) +END = START + timedelta(seconds=1) +ARN = ( + "arn:aws:lambda:us-west-2:123456789012:function:status:1/durable-execution/test/id" +) + + +InstrumentedView = tuple[ + ExecutionOtelPlugin | InvocationOtelPlugin, InMemorySpanExporter +] + + +@pytest.fixture(params=[ExecutionOtelPlugin, InvocationOtelPlugin]) +def instrumented_view(request: pytest.FixtureRequest) -> Iterator[InstrumentedView]: + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + plugin = request.param( + OtelPluginConfig( + tracer_provider=provider, + context_extractor=lambda _: None, + enrich_logger=False, + ) + ) + original = context.get_current() + yield plugin, exporter + assert context.get_current() == original + provider.shutdown() + + +@pytest.mark.parametrize( + "status", + [ + OperationStatus.SUCCEEDED, + OperationStatus.FAILED, + OperationStatus.CANCELLED, + OperationStatus.TIMED_OUT, + OperationStatus.STOPPED, + ], +) +@pytest.mark.parametrize("with_error", [False, True]) +@pytest.mark.parametrize("started_here", [False, True]) +def test_operation_status_and_exception( + instrumented_view: InstrumentedView, + status: OperationStatus, + with_error: bool, + started_here: bool, +) -> None: + """A delivered terminal hook must not label an unsuccessful operation OK.""" + plugin, exporter = instrumented_view + invocation: dict[str, Any] = dict( + request_id="first", + execution_arn=ARN, + execution_start_time=START, + is_first_invocation=True, + ) + plugin.on_invocation_start(InvocationStartInfo(**invocation)) + operation: dict[str, Any] = dict( + operation_id="wait", + name="wait", + parent_id=None, + operation_type=OperationType.WAIT, + sub_type=OperationSubType.WAIT, + start_time=START, + is_replayed=False, + ) + if started_here: + plugin.on_operation_start( + OperationStartInfo(**operation, status=OperationStatus.STARTED) + ) + error = ( + ErrorObject( + message="deadline exceeded", + type="TimeoutError", + data=None, + stack_trace=None, + ) + if with_error + else None + ) + plugin.on_operation_end( + OperationEndInfo(**operation, end_time=END, status=status, error=error) + ) + plugin.on_invocation_end( + InvocationEndInfo(**invocation, status=InvocationStatus.SUCCEEDED) + ) + span = next(s for s in exporter.get_finished_spans() if s.name == "wait") + assert span.attributes is not None + assert span.attributes["durable.operation.status"] == status.value + expected = ( + StatusCode.ERROR + if with_error + else ( + StatusCode.OK if status is OperationStatus.SUCCEEDED else StatusCode.UNSET + ) + ) + assert span.status.status_code is expected + if with_error: + assert span.status.description == "deadline exceeded" + assert len(span.events) == 1 + assert span.events[0].name == "exception" + assert span.events[0].attributes is not None + assert span.events[0].attributes["exception.message"] == "deadline exceeded" + else: + assert not span.events + + +def test_retry_then_resume_exports_retrying( + instrumented_view: InstrumentedView, +) -> None: + """Telemetry normalizes RETRY without changing the core enum or next invocation.""" + plugin, exporter = instrumented_view + original = context.get_current() + for index, status in enumerate( + [InvocationStatus.RETRY, InvocationStatus.SUCCEEDED] + ): + invocation: dict[str, Any] = dict( + request_id=f"request-{index}", + execution_arn=ARN, + execution_start_time=START, + is_first_invocation=index == 0, + ) + plugin.on_invocation_start(InvocationStartInfo(**invocation)) + plugin.on_invocation_end(InvocationEndInfo(**invocation, status=status)) + assert context.get_current() == original + spans = [s for s in exporter.get_finished_spans() if s.name == "Invocation"] + assert [ + s.attributes["durable.invocation.status"] + for s in spans + if s.attributes is not None + ] == [ + "RETRYING", + "SUCCEEDED", + ] + assert [s.status.status_code for s in spans] == [StatusCode.UNSET, StatusCode.OK] + assert InvocationStatus.RETRY.value == "RETRY" + 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