From 5c3adc9dbc33baa54056684bd417571db0023d80 Mon Sep 17 00:00:00 2001 From: silanhe Date: Tue, 28 Jul 2026 20:59:42 +0000 Subject: [PATCH] feat(otel): add Workflow span to InvocationOtelPlugin InvocationOtelPlugin now emits a deterministic, execution-scoped Workflow root span (span ID derived from the execution ARN, INTERNAL, started at execution_start_time), exported once on a terminal invocation (SUCCEEDED->OK, FAILED->ERROR; non-terminal dropped unexported). Created unconditionally since this plugin has no default/owned tracer-provider distinction. Operation and attempt spans link to the Workflow span while remaining parented to the Invocation span. The workflow span name is configurable via a new workflow_span_name constructor kwarg. Span display names are capitalized in both plugins: "Workflow" (default) and "Invocation". --- .../execution_plugin.py | 2 +- .../invocation_plugin.py | 68 +++++++++- .../tests/test_execution_plugin.py | 16 +-- .../test_execution_plugin_integration.py | 4 +- .../tests/test_invocation_plugin.py | 125 ++++++++++++++++-- .../test_invocation_plugin_integration.py | 2 +- 6 files changed, 193 insertions(+), 24 deletions(-) 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 837fe6b9..1ae94d1e 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 @@ -276,7 +276,7 @@ def _start_invocation_span(self, info: InvocationStartInfo) -> None: if info.request_id: attributes["faas.invocation_id"] = info.request_id self._invocation_span = self._tracer.start_span( - name="invocation", + name="Invocation", kind=SpanKind.INTERNAL, attributes=attributes, context=parent_ctx, 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 d335b2b5..ddf486b8 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 @@ -40,13 +40,21 @@ ) from aws_durable_execution_sdk_python_otel.deterministic_id_generator import ( DeterministicIdGenerator, + derive_workflow_span_id, operation_id_to_span_id, ) from aws_durable_execution_sdk_python_otel.log_filter import install_log_filter +from aws_durable_execution_sdk_python_otel.otel_plugin_config import ( + DEFAULT_WORKFLOW_SPAN_NAME, +) logger = logging.getLogger(__name__) +_TERMINAL_INVOCATION_STATUSES = frozenset( + {InvocationStatus.SUCCEEDED, InvocationStatus.FAILED} +) + _SpanAttributes = dict[str, str | bool | int] @@ -88,6 +96,7 @@ def __init__( context_extractor: ContextExtractor | None = None, instrument_name: str = DEFAULT_INSTRUMENT_NAME, enrich_logger: bool = True, + workflow_span_name: str = DEFAULT_WORKFLOW_SPAN_NAME, ) -> None: """Initialize the plugin with an OpenTelemetry tracer provider. @@ -101,6 +110,7 @@ def __init__( OTel trace context onto every emitted log record. """ self._enrich_logger = enrich_logger + self._workflow_span_name = workflow_span_name self._context_extractor: ContextExtractor = ( context_extractor or xray_context_extractor ) @@ -132,6 +142,7 @@ def __init__( # per invocation status: self._execution_arn = "" self._extracted_context: Context | None = None + self._workflow_span: Span | None = None # Maps operation ID (None for root) to the active span. self._operation_spans: dict[str | None, Span] = {} self._operation_spans_lock = threading.RLock() @@ -279,6 +290,12 @@ def _start_span( if operation_id else None ) + # Operation and attempt spans link to the execution-scoped Workflow + # span (the invocation span itself, operation_id=None, does not). + if self._workflow_span is not None and operation_id is not None: + workflow_ctx = self._workflow_span.get_span_context() + if workflow_ctx and workflow_ctx.is_valid: + links = [*links, Link(context=workflow_ctx)] if parent_span is None: # root span parent_context = self._extracted_context @@ -329,12 +346,42 @@ 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) + self._start_workflow_span(info) + self._start_span( operation_id=None, - name="invocation", + name="Invocation", attributes=self._extract_attributes(info), ) + def _start_workflow_span(self, info: InvocationStartInfo) -> None: + """Create the deterministic, execution-scoped Workflow root span. + + The Workflow span is a parentless root keyed to a deterministic span ID + derived from the execution ARN, so every invocation of the same durable + execution contributes to one Workflow span. It is exported once, on a + terminal invocation. Operation and attempt spans link to it while + remaining parented to the invocation span. It is created unconditionally + -- InvocationOtelPlugin has no default/owned tracer-provider distinction, + so the span is emitted whether the provider is the ambient (ADOT/global) + one or an explicitly supplied one. + """ + if not self._execution_arn: + logger.warning("No execution ARN; skipping Workflow span creation") + return + self._id_generator.set_next_span_id( + derive_workflow_span_id(self._execution_arn) + ) + # Empty context => root span with no parent. _to_otel_timestamp falls + # back to now() when execution_start_time is None. + self._workflow_span = self._tracer.start_span( + name=self._workflow_span_name, + kind=SpanKind.INTERNAL, + attributes={"durable.execution.arn": self._execution_arn}, + start_time=_to_otel_timestamp(info.execution_start_time), + context=Context(), + ) + def on_invocation_end(self, info: InvocationEndInfo) -> None: """Called at the end of each invocation. Ends the invocation span and flushes.""" logger.debug("Durable invocation ended: %s", info) @@ -365,9 +412,28 @@ def on_invocation_end(self, info: InvocationEndInfo) -> None: # end the invocation span self._end_span(None) + # The Workflow span (execution view) is exported only on a terminal + # status; on non-terminal statuses its reference is dropped without + # ending it (so it is not exported yet). SUCCEEDED -> OK, FAILED -> ERROR; + # RETRY/PENDING are non-terminal and leave it unexported. + 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.status is InvocationStatus.FAILED: + self._workflow_span.set_status( + StatusCode.ERROR, info.error.message if info.error else "" + ) + elif info.status is InvocationStatus.SUCCEEDED: + self._workflow_span.set_status(StatusCode.OK) + self._workflow_span.end() + # Clear all per-invocation state to prevent leaks across warm Lambda reuses self._execution_arn = "" self._extracted_context = None + self._workflow_span = None with self._operation_spans_lock: self._operation_spans = {} 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 1dd00570..03a56969 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 @@ -126,10 +126,10 @@ def test_workflow_span_is_root_and_invocation_is_its_child(): spans = {s.name: s for s in exporter.get_finished_spans()} assert "Workflow" in spans - assert "invocation" in spans + assert "Invocation" in spans workflow = spans["Workflow"] - invocation = spans["invocation"] + invocation = spans["Invocation"] # Workflow is a root span (no parent) with the deterministic span ID. assert workflow.parent is None @@ -154,7 +154,7 @@ def test_workflow_span_dropped_on_non_terminal_status(): names = [s.name for s in exporter.get_finished_spans()] # Invocation span is always ended/exported; the Workflow span is dropped # (not ended) on a non-terminal status, so it must not be exported. - assert "invocation" in names + assert "Invocation" in names assert "Workflow" not in names @@ -193,7 +193,7 @@ def test_operation_parented_under_workflow_and_linked_to_invocation(): spans = {s.name: s for s in exporter.get_finished_spans()} workflow = spans["Workflow"] - invocation = spans["invocation"] + invocation = spans["Invocation"] operation = spans["wait-for-signal"] # Parented under the Workflow span (no parentId => Workflow fallback). @@ -266,8 +266,8 @@ def test_default_mode_creates_invocation_span(): spans = {s.name: s for s in exporter.get_finished_spans()} # The invocation span is now created even in default-provider mode. - assert "invocation" in spans - invocation = spans["invocation"] + assert "Invocation" in spans + invocation = spans["Invocation"] assert invocation.attributes["durable.execution.arn"] == EXECUTION_ARN assert invocation.attributes["durable.invocation.first"] is True @@ -285,7 +285,7 @@ def test_default_mode_invocation_span_parented_to_ambient_span(): otel_context.detach(token) ambient.end() - invocation = {s.name: s for s in exporter.get_finished_spans()}["invocation"] + invocation = {s.name: s for s in exporter.get_finished_spans()}["Invocation"] assert invocation.parent is not None assert invocation.parent.span_id == ambient.get_span_context().span_id @@ -335,7 +335,7 @@ def test_invocation_span_status_kind_and_attributes(status, expected_code): 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"] + 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 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 5a15a358..f3150429 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 @@ -172,7 +172,7 @@ def test_community_layer_full_lifecycle_is_workflow_rooted(): finished = exporter.get_finished_spans() spans = {s.name: s for s in finished} workflow = spans["Workflow"] - invocation = spans["invocation"] + invocation = spans["Invocation"] operation = spans[OP_NAME] attempt = spans[f"{OP_NAME} attempt 1"] @@ -229,7 +229,7 @@ def test_adot_layer_full_lifecycle_parents_to_ambient_span(): finished = exporter.get_finished_spans() spans = {s.name: s for s in finished} - invocation = spans["invocation"] + invocation = spans["Invocation"] operation = spans[OP_NAME] # Invocation span parents to the ambient ADOT span and carries the first flag. 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 116fa556..66e7b7ea 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 @@ -31,6 +31,7 @@ from opentelemetry.trace import SpanKind, StatusCode from aws_durable_execution_sdk_python_otel.deterministic_id_generator import ( + derive_workflow_span_id, operation_id_to_span_id, ) from aws_durable_execution_sdk_python_otel.invocation_plugin import InvocationOtelPlugin @@ -149,12 +150,15 @@ def test_invocation_start_and_end_emit_invocation_span(): plugin.on_invocation_end(_invocation_end_info()) spans = exporter.get_finished_spans() - assert [span.name for span in spans] == ["invocation"] - assert spans[0].kind is SpanKind.INTERNAL - assert spans[0].attributes["durable.execution.arn"] == EXECUTION_ARN - assert spans[0].attributes["durable.invocation.first"] is True + spans_by_name = {span.name: span for span in spans} + # Terminal invocation also exports the Workflow span. + assert set(spans_by_name) == {"Invocation", "Workflow"} + invocation = spans_by_name["Invocation"] + assert invocation.kind is SpanKind.INTERNAL + assert invocation.attributes["durable.execution.arn"] == EXECUTION_ARN + assert invocation.attributes["durable.invocation.first"] is True assert ( - spans[0].attributes["durable.invocation.status"] + invocation.attributes["durable.invocation.status"] == InvocationStatus.SUCCEEDED.value ) assert plugin._get_span(None) is None @@ -168,8 +172,8 @@ def test_invocation_span_records_subsequent_invocation(): plugin.on_invocation_end(_invocation_end_info()) spans = exporter.get_finished_spans() - assert len(spans) == 1 - assert spans[0].attributes["durable.invocation.first"] is False + invocation = next(s for s in spans if s.name == "Invocation") + assert invocation.attributes["durable.invocation.first"] is False @pytest.mark.parametrize( @@ -192,11 +196,11 @@ def test_invocation_span_status_reflects_execution_status( plugin.on_invocation_end(_invocation_end_info(invocation_status)) spans = exporter.get_finished_spans() - assert len(spans) == 1 - attributes = spans[0].attributes + 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 spans[0].status.status_code is expected_span_status + assert invocation.status.status_code is expected_span_status def test_operation_callbacks_emit_child_span_with_deterministic_span_id(): @@ -249,7 +253,7 @@ def test_operation_callbacks_emit_child_span_with_deterministic_span_id(): spans_by_name = {span.name: span for span in exporter.get_finished_spans()} assert all(span.kind is SpanKind.INTERNAL for span in spans_by_name.values()) wait_span = spans_by_name["wait-for-signal"] - invocation_span = spans_by_name["invocation"] + invocation_span = spans_by_name["Invocation"] assert wait_span.context.span_id == operation_id_to_span_id( EXECUTION_ARN, operation_id ) @@ -889,3 +893,102 @@ def test_nested_steps_restore_context_span_across_multiple_iterations(): ) # Between each inner step, the child-context span is current. assert trace.get_current_span().get_span_context().span_id == context_span_id + + +@pytest.mark.parametrize( + ("status", "expected_code"), + [ + (InvocationStatus.SUCCEEDED, StatusCode.OK), + (InvocationStatus.FAILED, StatusCode.ERROR), + ], +) +def test_workflow_span_exported_on_terminal(status, expected_code): + """A terminal invocation exports a deterministic Workflow root span.""" + plugin, exporter = _create_plugin() + plugin.on_invocation_start(_invocation_start_info()) + plugin.on_invocation_end(_invocation_end_info(status)) + + workflow = next(s for s in exporter.get_finished_spans() if s.name == "Workflow") + # Root span: no parent. + assert workflow.parent is None + assert workflow.kind is SpanKind.INTERNAL + # Deterministic span id derived from the execution ARN. + assert workflow.context.span_id == derive_workflow_span_id(EXECUTION_ARN) + assert workflow.attributes["durable.execution.arn"] == EXECUTION_ARN + assert workflow.attributes["durable.execution.status"] == status.value + assert workflow.status.status_code is expected_code + # Anchored to the execution start time. + assert workflow.start_time == int(START_TIME.timestamp() * 1_000_000_000) + + +@pytest.mark.parametrize("status", [InvocationStatus.PENDING, InvocationStatus.RETRY]) +def test_workflow_span_not_exported_on_non_terminal(status): + """Non-terminal invocations do not export (end) the Workflow span.""" + plugin, exporter = _create_plugin() + plugin.on_invocation_start(_invocation_start_info()) + plugin.on_invocation_end(_invocation_end_info(status)) + + names = [s.name for s in exporter.get_finished_spans()] + assert "Workflow" not in names + assert "Invocation" in names + + +def test_operation_span_links_to_workflow_span(): + """Operation spans link to the Workflow span while parented to invocation.""" + plugin, exporter = _create_plugin() + plugin.on_invocation_start(_invocation_start_info()) + operation_id = "wait-1" + plugin.on_operation_start( + OperationStartInfo( + operation_id=operation_id, + operation_type=OperationType.WAIT, + sub_type=OperationSubType.WAIT, + name="wait-for-signal", + parent_id=None, + start_time=START_TIME, + is_replayed=False, + status=OperationStatus.STARTED, + ) + ) + plugin.on_operation_end( + OperationEndInfo( + operation_id=operation_id, + operation_type=OperationType.WAIT, + sub_type=OperationSubType.WAIT, + name="wait-for-signal", + parent_id=None, + start_time=START_TIME, + is_replayed=False, + status=OperationStatus.SUCCEEDED, + end_time=END_TIME, + error=None, + ) + ) + plugin.on_invocation_end(_invocation_end_info()) + + spans_by_name = {s.name: s for s in exporter.get_finished_spans()} + op_span = spans_by_name["wait-for-signal"] + workflow_span_id = derive_workflow_span_id(EXECUTION_ARN) + linked_span_ids = {link.context.span_id for link in op_span.links} + assert workflow_span_id in linked_span_ids + # Still parented to the invocation span (not the Workflow span). + assert op_span.parent is not None + assert op_span.parent.span_id == spans_by_name["Invocation"].context.span_id + + +def test_workflow_span_name_is_configurable(): + """The Workflow span name can be overridden via constructor kwarg.""" + exporter = InMemorySpanExporter() + trace_provider = TracerProvider() + trace_provider.add_span_processor(SimpleSpanProcessor(exporter)) + plugin = InvocationOtelPlugin( + trace_provider=trace_provider, + context_extractor=lambda _: Context(), + workflow_span_name="MyExecution", + ) + plugin.on_invocation_start(_invocation_start_info()) + plugin.on_invocation_end(_invocation_end_info()) + + names = [s.name for s in exporter.get_finished_spans()] + assert "MyExecution" in names + assert "Workflow" not in names diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin_integration.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin_integration.py index 129e3e97..b316fa80 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin_integration.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin_integration.py @@ -155,7 +155,7 @@ def _run_step_lifecycle(plugin: InvocationOtelPlugin) -> None: def _assert_hierarchy(exporter: InMemorySpanExporter) -> None: spans = {s.name: s for s in exporter.get_finished_spans()} - invocation = spans["invocation"] + invocation = spans["Invocation"] operation = spans[OP_NAME] attempt = spans[f"{OP_NAME} attempt 1"]