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 52c25123..b326ec77 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 @@ -46,6 +46,7 @@ Link, Span, SpanContext, + SpanKind, StatusCode, Tracer, ) @@ -141,7 +142,6 @@ def __init__(self, config: OtelPluginConfig | None = None) -> None: # Per-invocation state. self._execution_arn = "" self._extracted_context: Context | None = None - self._saved_invocation_context: Context | None = None self._workflow_span: Span | None = None self._invocation_span: Span | None = None self._operation_spans: dict[str, Span] = {} @@ -188,12 +188,7 @@ def get_current_span_context(self) -> SpanContext | None: # Links # ------------------------------------------------------------------ def _build_invocation_links(self) -> list[Link]: - """Link operation/attempt spans to the invocation span.""" - if self._use_default and self._saved_invocation_context is not None: - span = trace.get_current_span(self._saved_invocation_context) - ctx = span.get_span_context() - if ctx and ctx.is_valid: - return [Link(context=ctx)] + """Link operation/attempt spans to the durable invocation span.""" if self._invocation_span is not None: ctx = self._invocation_span.get_span_context() if ctx and ctx.is_valid: @@ -217,10 +212,6 @@ def on_invocation_start(self, info: InvocationStartInfo) -> None: self._extracted_context = self._context_extractor(info) self._id_generator.set_trace_id(self._execution_arn, info.execution_start_time) - # Capture the ambient context for link-building in default mode. - if self._use_default: - self._saved_invocation_context = otel_context.get_current() - self._start_workflow_span(info) # Create the Invocation span in both modes. In default-provider mode it # is parented to the ambient Lambda invocation span. @@ -259,10 +250,11 @@ def _start_invocation_span(self, info: InvocationStartInfo) -> None: if self._use_default: # Default-provider mode: parent the Invocation span to the ambient # Lambda invocation span (from the ADOT layer or other - # auto-instrumentation) captured at invocation start. - # Lambda semantic attributes belong to that ambient span, so carry - # only durable correlation attributes here. - parent_ctx = self._saved_invocation_context or otel_context.get_current() + # auto-instrumentation), which is still the active context here (the + # Workflow span is created with an empty context and not yet + # attached). Lambda semantic attributes belong to that ambient span, + # so carry only durable correlation attributes here. + parent_ctx = otel_context.get_current() attributes = { "durable.execution.arn": self._execution_arn, "durable.invocation.first": info.is_first_invocation, @@ -275,6 +267,7 @@ def _start_invocation_span(self, info: InvocationStartInfo) -> None: ) attributes = { "durable.execution.arn": self._execution_arn, + "durable.invocation.first": info.is_first_invocation, "faas.coldstart": self._is_cold_start, "cloud.provider": "aws", "cloud.platform": "aws_lambda", @@ -283,6 +276,7 @@ def _start_invocation_span(self, info: InvocationStartInfo) -> None: attributes["faas.invocation_id"] = info.request_id self._invocation_span = self._tracer.start_span( name="invocation", + kind=SpanKind.INTERNAL, attributes=attributes, context=parent_ctx, ) @@ -296,23 +290,38 @@ def on_invocation_end(self, info: InvocationEndInfo) -> None: # so they are not exported as if completed. _reset_state # clears the span map below. - # End the invocation span regardless of terminal status. + # End the invocation span regardless of terminal status. Record the + # invocation status and map it to a span status: SUCCEEDED/PENDING are OK + # (this invocation did its work, whether it completed or suspended), + # RETRY/FAILED are ERROR (this invocation failed). if self._invocation_span is not None: + self._invocation_span.set_attribute( + "durable.invocation.status", + info.status.value if info.status else "", + ) + if info.status in (InvocationStatus.SUCCEEDED, InvocationStatus.PENDING): + self._invocation_span.set_status(StatusCode.OK) + elif info.status in (InvocationStatus.RETRY, InvocationStatus.FAILED): + self._invocation_span.set_status( + StatusCode.ERROR, info.error.message if info.error else "" + ) self._invocation_span.end() - # The Workflow span is exported only on a terminal status; otherwise its - # reference is dropped without ending it. + # The Workflow span (execution view) is exported only on a terminal + # status; otherwise its reference is dropped without ending it. Its span + # status reflects the execution outcome: SUCCEEDED -> OK, FAILED -> ERROR + # (RETRY/PENDING are non-terminal and never reach here -> UNSET). if self._workflow_span is not None: if info.status in _TERMINAL_INVOCATION_STATUSES: self._workflow_span.set_attribute( "durable.execution.status", info.status.value if info.status else "", ) - if info.error: + if info.status is InvocationStatus.FAILED: self._workflow_span.set_status( - StatusCode.ERROR, info.error.message or "" + StatusCode.ERROR, info.error.message if info.error else "" ) - else: + elif info.status is InvocationStatus.SUCCEEDED: self._workflow_span.set_status(StatusCode.OK) self._workflow_span.end() @@ -327,7 +336,6 @@ def on_invocation_end(self, info: InvocationEndInfo) -> None: def _reset_state(self) -> None: self._execution_arn = "" self._extracted_context = None - self._saved_invocation_context = None self._workflow_span = None self._invocation_span = None with self._lock: 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 d0ab1f68..675bc027 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 @@ -350,12 +350,12 @@ def on_invocation_end(self, info: InvocationEndInfo) -> None: invocation_span.set_attribute( "durable.invocation.status", info.status.value ) - if info.status is InvocationStatus.FAILED: + if info.status in (InvocationStatus.SUCCEEDED, InvocationStatus.PENDING): + invocation_span.set_status(StatusCode.OK) + elif info.status in (InvocationStatus.RETRY, InvocationStatus.FAILED): invocation_span.set_status( StatusCode.ERROR, info.error.message if info.error else "" ) - elif info.status is InvocationStatus.SUCCEEDED: - invocation_span.set_status(StatusCode.OK) # end the invocation span self._end_span(None) 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 86d9da18..7cd544eb 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 @@ -317,3 +317,27 @@ def test_open_operation_span_not_exported_at_invocation_end(): exported = {s.name for s in exporter.get_finished_spans()} # The open operation span is NOT exported (never ended). assert "wait-for-signal" not in exported + + +@pytest.mark.parametrize( + ("status", "expected_code"), + [ + (InvocationStatus.PENDING, trace.StatusCode.OK), + (InvocationStatus.SUCCEEDED, trace.StatusCode.OK), + (InvocationStatus.RETRY, trace.StatusCode.ERROR), + (InvocationStatus.FAILED, trace.StatusCode.ERROR), + ], +) +def test_invocation_span_status_kind_and_attributes(status, expected_code): + """Invocation span is INTERNAL, carries status/first; SUCCEEDED/PENDING are + OK and RETRY/FAILED are ERROR.""" + plugin, exporter = _create_plugin() + plugin.on_invocation_start(_invocation_start_info()) + plugin.on_invocation_end(_invocation_end_info(status=status)) + + 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.first"] is True + assert invocation.status.status_code is expected_code diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin_integration.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin_integration.py index 86166376..5a15a358 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin_integration.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin_integration.py @@ -237,9 +237,10 @@ def test_adot_layer_full_lifecycle_parents_to_ambient_span(): assert invocation.parent.span_id == ambient.get_span_context().span_id assert invocation.attributes["durable.invocation.first"] is True - # Operation span still uses the deterministic id and links to the ambient span. + # Operation span still uses the deterministic id and links to the durable + # invocation span (which is itself parented to the ambient ADOT span). assert operation.context.span_id == operation_id_to_span_id(EXECUTION_ARN, OP_ID) - assert ambient.get_span_context().span_id in { + assert invocation.context.span_id in { link.context.span_id for link in operation.links } 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 58d22a68..8a243503 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 @@ -174,8 +174,8 @@ def test_invocation_span_records_subsequent_invocation(): @pytest.mark.parametrize( ("invocation_status", "expected_span_status"), [ - (InvocationStatus.PENDING, StatusCode.UNSET), - (InvocationStatus.RETRY, StatusCode.UNSET), + (InvocationStatus.PENDING, StatusCode.OK), + (InvocationStatus.RETRY, StatusCode.ERROR), (InvocationStatus.SUCCEEDED, StatusCode.OK), (InvocationStatus.FAILED, StatusCode.ERROR), ],