diff --git a/packages/aws-durable-execution-sdk-python-otel/README.md b/packages/aws-durable-execution-sdk-python-otel/README.md index 187971ba5..4fdb0c1da 100644 --- a/packages/aws-durable-execution-sdk-python-otel/README.md +++ b/packages/aws-durable-execution-sdk-python-otel/README.md @@ -12,6 +12,9 @@ OpenTelemetry instrumentation plugin for the [AWS Durable Execution SDK for Pyth - **Provider Integration**: Use the global ADOT provider or supply an explicit SDK `TracerProvider` - **Execution Sampling**: Resolve sampling once per invocation and apply it consistently to Workflow, Invocation, operation, and attempt spans +Operation subtype attributes accept enum members or plain string labels. The +plugin preserves the label and uses the operation's explicit type independently. + ## Installation When using an ADOT or community OpenTelemetry Lambda layer: 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 2008615e1..f6e7e6bff 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 @@ -921,7 +921,9 @@ def _operation_attributes(self, info: Any) -> dict[str, Any]: if getattr(info, "operation_type", None) is not None: attributes["durable.operation.type"] = info.operation_type.value if getattr(info, "sub_type", None) is not None: - attributes["durable.operation.subtype"] = info.sub_type.value + attributes["durable.operation.subtype"] = getattr( + info.sub_type, "value", info.sub_type + ) # STEP user-function spans represent attempts, not durable operations. if ( not ( 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 4fc2bf02b..294ef2119 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 @@ -984,7 +984,9 @@ def _extract_attributes(self, info: Any) -> _SpanAttributes: if hasattr(info, "operation_type") and info.operation_type is not None: attributes["durable.operation.type"] = info.operation_type.value if hasattr(info, "sub_type") and info.sub_type is not None: - attributes["durable.operation.subtype"] = info.sub_type.value + attributes["durable.operation.subtype"] = getattr( + info.sub_type, "value", info.sub_type + ) # STEP user-function spans represent attempts, not durable operations. if ( not ( diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_subtype_compatibility.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_subtype_compatibility.py new file mode 100644 index 000000000..af8e3b92a --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_subtype_compatibility.py @@ -0,0 +1,96 @@ +"""Plugin hook metadata may come from enum- or string-based SDK subtypes.""" + +from datetime import UTC, datetime, timedelta + +import opentelemetry.context as otel_context +import pytest +from aws_durable_execution_sdk_python.lambda_service import ( + OperationStatus, + OperationSubType, +) +from aws_durable_execution_sdk_python.plugin import ( + InvocationEndInfo, + InvocationStartInfo, + InvocationStatus, + OperationEndInfo, + OperationStartInfo, + OperationType, +) +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 aws_durable_execution_sdk_python_otel import ( + ExecutionOtelPlugin, + InvocationOtelPlugin, + OtelPluginConfig, +) + + +@pytest.mark.parametrize("plugin_type", [ExecutionOtelPlugin, InvocationOtelPlugin]) +@pytest.mark.parametrize( + "sub_type,expected", + [ + (OperationSubType.WAIT, "Wait"), + ("Wait", "Wait"), + ("Library_CustomWait", "Library_CustomWait"), + (None, None), + ], +) +def test_hook_exports_enum_or_string_subtype_without_changing_type( + plugin_type, sub_type, expected +): + provider = TracerProvider() + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + plugin = plugin_type( + OtelPluginConfig( + tracer_provider=provider, + context_extractor=lambda _: None, + enrich_logger=False, + ) + ) + before = otel_context.get_current() + start = datetime.now(UTC) + invocation = { + "execution_arn": "arn:aws:lambda:us-west-2:123456789012:function:subtype:$LATEST", + "request_id": "request", + "execution_start_time": start, + "is_first_invocation": True, + } + operation = { + "operation_id": "target", + "operation_type": OperationType.WAIT, + "sub_type": sub_type, + "name": "target", + "parent_id": None, + "start_time": start, + "is_replayed": False, + } + try: + plugin.on_invocation_start(InvocationStartInfo(**invocation)) + plugin.on_operation_start( + OperationStartInfo(**operation, status=OperationStatus.STARTED) + ) + plugin.on_operation_end( + OperationEndInfo( + **operation, + status=OperationStatus.SUCCEEDED, + end_time=start + timedelta(seconds=1), + ) + ) + (span,) = [ + span for span in exporter.get_finished_spans() if span.name == "target" + ] + assert span.attributes["durable.operation.type"] == "WAIT" + assert span.attributes["durable.operation.status"] == "SUCCEEDED" + if expected is None: + assert "durable.operation.subtype" not in span.attributes + else: + assert span.attributes["durable.operation.subtype"] == expected + finally: + plugin.on_invocation_end( + InvocationEndInfo(**invocation, status=InvocationStatus.SUCCEEDED) + ) + provider.shutdown() + assert otel_context.get_current() == before diff --git a/packages/aws-durable-execution-sdk-python-testing/README.md b/packages/aws-durable-execution-sdk-python-testing/README.md index 58b21e793..3df614344 100644 --- a/packages/aws-durable-execution-sdk-python-testing/README.md +++ b/packages/aws-durable-execution-sdk-python-testing/README.md @@ -170,6 +170,13 @@ flow, pagination, and a map of the code. ## Documentation +History event projection can read enum members or plain string subtype metadata. +Parsing checkpoints and reconstructing typed operation results still requires +known core enum labels; unknown labels are rejected before checkpoint mutation. +Declared public core and testing subtype types remain unchanged. Support for +custom-subtype configuration, persistence and replay requires a coordinated future +core implementation; this preparation does not change those contracts. + ### Error Handling The testing framework implements AWS-compliant error responses that match the exact format expected by boto3 and AWS services. For detailed information about error response formats, exception types, and troubleshooting, see: diff --git a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py index 43026ec60..b7e8bd257 100644 --- a/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py +++ b/packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing/model.py @@ -8,7 +8,7 @@ import json from dataclasses import dataclass, replace from enum import Enum -from typing import Any +from typing import Any, cast from aws_durable_execution_sdk_python.execution import DurableExecutionInvocationOutput @@ -1330,7 +1330,11 @@ def create( @property def sub_type(self) -> str | None: - return self.operation.sub_type.value if self.operation.sub_type else None + if not self.operation.sub_type: + return None + return cast( + str, getattr(self.operation.sub_type, "value", self.operation.sub_type) + ) def get_retry_details(self) -> RetryDetails | None: if not self.operation.step_details or not self.operation_update: diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/e2e/subtype_history_int_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/subtype_history_int_test.py new file mode 100644 index 000000000..63dca4d64 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/e2e/subtype_history_int_test.py @@ -0,0 +1,60 @@ +"""Read subtype labels through the public cloud-history result API.""" + +from datetime import UTC, datetime + +import pytest +from aws_durable_execution_sdk_python.lambda_service import ( + OperationStatus, + OperationSubType, + OperationType, +) + +from aws_durable_execution_sdk_python_testing.model import ( + GetDurableExecutionHistoryResponse, + GetDurableExecutionResponse, +) +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestResult + + +@pytest.mark.parametrize("label", ["Step", None]) +def test_history_result_preserves_declared_enum_contract(label): + events = [] + for index, (name, sub_type) in enumerate( + [("known", "Step"), ("target", label)], start=1 + ): + event = { + "EventType": "StepSucceeded", + "EventId": index, + "EventTimestamp": datetime(2024, 1, 1, tzinfo=UTC), + "Id": name, + "Name": name, + "StepSucceededDetails": { + "Result": {"Payload": '"saved"', "Truncated": False} + }, + } + if sub_type is not None: + event["SubType"] = sub_type + events.append(event) + history = GetDurableExecutionHistoryResponse.from_dict({"Events": events}) + # Exercise the history DTO round trip as well as the public result factory. + history = GetDurableExecutionHistoryResponse.from_dict(history.to_dict()) + result = DurableFunctionTestResult.from_execution_history( + GetDurableExecutionResponse( + durable_execution_arn="execution", + durable_execution_name="execution", + function_arn="function", + status="SUCCEEDED", + start_timestamp=datetime(2024, 1, 1, tzinfo=UTC), + ), + history, + ) + assert len(result.operations) == 2 + assert result.get_step("known").sub_type is OperationSubType.STEP + target = result.get_step("target") + assert target.operation_type is OperationType.STEP + assert target.status is OperationStatus.SUCCEEDED + assert target.result == '"saved"' + if label == "Step": + assert target.sub_type is OperationSubType.STEP + elif label is None: + assert target.sub_type is None diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/subtype_compatibility_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/subtype_compatibility_test.py new file mode 100644 index 000000000..14cbcfe95 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/subtype_compatibility_test.py @@ -0,0 +1,112 @@ +"""Read string metadata without widening current checkpoint/result contracts.""" + +import datetime + +import pytest +from aws_durable_execution_sdk_python.lambda_service import ( + Operation, + OperationStatus, + OperationSubType, + OperationType, +) + +from aws_durable_execution_sdk_python_testing.model import ( + CheckpointDurableExecutionRequest, + Event, + EventCreationContext, + StartDurableExecutionInput, + events_to_operations, +) +from aws_durable_execution_sdk_python_testing.exceptions import ( + InvalidParameterValueException, +) + + +@pytest.mark.parametrize( + "sub_type,expected", + [ + (OperationSubType.STEP, "Step"), + ("Step", "Step"), + ("Library_CustomStep", "Library_CustomStep"), + (None, None), + ("", None), + ], +) +def test_history_projection_keeps_enum_or_plain_string_subtype(sub_type, expected): + operation = Operation( + operation_id="operation", + operation_type=OperationType.STEP, + status=OperationStatus.STARTED, + sub_type=sub_type, + start_timestamp=datetime.datetime(2024, 1, 1, tzinfo=datetime.UTC), + ) + context = EventCreationContext( + operation=operation, + event_id=1, + durable_execution_arn="execution", + start_durable_execution_input=StartDurableExecutionInput( + account_id="123456789012", + function_name="function", + function_qualifier="$LATEST", + execution_name="execution", + execution_timeout_seconds=30, + execution_retention_period_days=1, + ), + ) + event = Event.create_event_started(context) + if expected is None: + assert "SubType" not in event.to_dict() + else: + assert event.to_dict()["SubType"] == expected + assert Event.from_dict(event.to_dict()).sub_type == expected + if expected == "Library_CustomStep": + with pytest.raises(InvalidParameterValueException): + events_to_operations([event]) + return + (restored,) = events_to_operations([event]) + assert restored.operation_type is OperationType.STEP + assert getattr(restored.sub_type, "value", restored.sub_type) == expected + + +@pytest.mark.parametrize("sub_type", ["Step", None, ""]) +def test_checkpoint_request_preserves_existing_enum_contract(sub_type): + request = CheckpointDurableExecutionRequest.from_dict( + { + "CheckpointToken": "token", + "Updates": [ + { + "Id": "operation", + "Type": "STEP", + "Action": "START", + "SubType": sub_type, + } + ], + }, + durable_execution_arn="execution", + ) + assert request.updates is not None and len(request.updates) == 1 + update = request.updates[0] + assert update.operation_type is OperationType.STEP + if sub_type == "Step": + assert update.sub_type is OperationSubType.STEP + elif not sub_type: + assert update.sub_type is None + + +@pytest.mark.parametrize("sub_type", ["Library_CustomStep", 42, True]) +def test_checkpoint_request_rejects_unsupported_subtype(sub_type): + with pytest.raises(ValueError, match="not a valid OperationSubType"): + CheckpointDurableExecutionRequest.from_dict( + { + "CheckpointToken": "token", + "Updates": [ + { + "Id": "operation", + "Type": "STEP", + "Action": "START", + "SubType": sub_type, + } + ], + }, + durable_execution_arn="execution", + ) diff --git a/packages/aws-durable-execution-sdk-python-testing/tests/web/e2e/checkpoint_subtype_int_test.py b/packages/aws-durable-execution-sdk-python-testing/tests/web/e2e/checkpoint_subtype_int_test.py new file mode 100644 index 000000000..d55a6ef46 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-testing/tests/web/e2e/checkpoint_subtype_int_test.py @@ -0,0 +1,110 @@ +"""Unsupported wire subtypes must not consume checkpoint state or tokens.""" + +from threading import Event +from urllib.parse import quote + +import pytest + +from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner +from aws_durable_execution_sdk_python_testing.stores.filesystem import ( + FileSystemExecutionStore, +) +from aws_durable_execution_sdk_python_testing.stores.memory import ( + InMemoryExecutionStore, +) +from aws_durable_execution_sdk_python_testing.stores.sqlite import SQLiteExecutionStore +from aws_durable_execution_sdk_python_testing.web.handlers import ( + CheckpointDurableExecutionHandler, +) +from aws_durable_execution_sdk_python_testing.web.models import HTTPRequest +from aws_durable_execution_sdk_python_testing.web.routes import ( + CheckpointDurableExecutionRoute, + Route, +) + + +@pytest.mark.parametrize("store_kind", ["memory", "filesystem", "sqlite"]) +@pytest.mark.parametrize("label", ["Step", "Library_CustomStep"]) +def test_checkpoint_subtype_validation_precedes_mutation(store_kind, label, tmp_path): + if store_kind == "memory": + store = InMemoryExecutionStore() + elif store_kind == "filesystem": + store = FileSystemExecutionStore.create(tmp_path / "store") + else: + store = SQLiteExecutionStore.create_and_initialize(str(tmp_path / "store.db")) + + ready = Event() + release = Event() + inputs = [] + + def handler(event, context): + inputs.append(event) + ready.set() + release.wait(10) + return {"Status": "SUCCEEDED", "Result": '"done"'} + + runner = DurableFunctionTestRunner( + handler=handler, store=store, skip_time=False, execution_timeout=20 + ) + try: + arn = runner.run_async(input="{}") + assert ready.wait(3), "The real invocation must start before checkpointing" + token = inputs[0].checkpoint_token + before = store.load(arn) + before_sequence = before.token_sequence + before_ids = [operation.operation_id for operation in before.operations] + route = CheckpointDurableExecutionRoute.from_route( + Route.from_string( + "/2025-12-01/durable-executions/" + quote(arn, safe="") + "/checkpoint" + ) + ) + # Wire the actual endpoint to the runner's actual executor; the runner + # establishes the invocation gate, worker and checkpoint token itself. + endpoint = CheckpointDurableExecutionHandler(runner._executor) # noqa: SLF001 + response = endpoint.handle( + route, + HTTPRequest( + method="POST", + path=route, + headers={}, + query_params={}, + body={ + "CheckpointToken": token, + "Updates": [ + { + "Id": "target", + "Type": "STEP", + "Action": "START", + "Name": "target", + "SubType": label, + } + ], + }, + ), + ) + after = store.load(arn) + after_ids = [operation.operation_id for operation in after.operations] + if label == "Step": + assert response.status_code == 200 + assert after.token_sequence == before_sequence + 1 + assert after_ids == [*before_ids, "target"] + else: + assert response.status_code == 400 + assert "is not a valid OperationSubType" in str(response.body) + assert after.token_sequence == before_sequence + assert after_ids == before_ids + + retry = endpoint.handle( + route, + HTTPRequest( + method="POST", + path=route, + headers={}, + query_params={}, + body={"CheckpointToken": token, "Updates": []}, + ), + ) + assert retry.status_code == (400 if label == "Step" else 200) + finally: + runner.close() + release.set()