From 797310ec3119f2fc6eeb19283c2b8a591d905ca1 Mon Sep 17 00:00:00 2001 From: Frank Chen <65260095+zhongkechen@users.noreply.github.com> Date: Fri, 7 Aug 2026 17:30:20 -0700 Subject: [PATCH 1/7] feat: auto-discover instrumentation plugins --- .../README.md | 13 + .../pyproject.toml | 4 + .../plugin_provider.py | 19 + .../tests/test_plugin_provider.py | 38 ++ .../README.md | 46 ++ .../__init__.py | 2 + .../exceptions.py | 4 + .../execution.py | 5 +- .../plugin.py | 11 + .../plugin_discovery.py | 209 ++++++++ .../tests/execution_test.py | 21 + .../tests/plugin_discovery_test.py | 469 ++++++++++++++++++ 12 files changed, 840 insertions(+), 1 deletion(-) create mode 100644 packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/plugin_provider.py create mode 100644 packages/aws-durable-execution-sdk-python-otel/tests/test_plugin_provider.py create mode 100644 packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin_discovery.py create mode 100644 packages/aws-durable-execution-sdk-python/tests/plugin_discovery_test.py diff --git a/packages/aws-durable-execution-sdk-python-otel/README.md b/packages/aws-durable-execution-sdk-python-otel/README.md index 5a38774a..889aa9e2 100644 --- a/packages/aws-durable-execution-sdk-python-otel/README.md +++ b/packages/aws-durable-execution-sdk-python-otel/README.md @@ -24,6 +24,19 @@ pip install aws-durable-execution-sdk-python-otel 3. Pass `InvocationOtelPlugin` to your handler's `plugins` list 4. Add X-Ray write permissions +Alternatively, install this package in the function artifact or a Lambda layer +and select either OTel plugin by entry-point name: + +```text +DURABLE_EXECUTION_PLUGINS=otel-invocation +DURABLE_EXECUTION_PLUGINS=otel-execution +``` + +`otel-invocation` creates `InvocationOtelPlugin`; `otel-execution` creates +`ExecutionOtelPlugin`. The SDK discovers the selected package entry point at +cold start, so the handler does not need to import or explicitly register the +plugin. + ### 1. ADOT Lambda Layer This plugin requires the [AWS Distro for OpenTelemetry (ADOT) Lambda layer](https://aws-otel.github.io/docs/getting-started/lambda) to export traces from your Lambda function. diff --git a/packages/aws-durable-execution-sdk-python-otel/pyproject.toml b/packages/aws-durable-execution-sdk-python-otel/pyproject.toml index bd65c93d..3de722d5 100644 --- a/packages/aws-durable-execution-sdk-python-otel/pyproject.toml +++ b/packages/aws-durable-execution-sdk-python-otel/pyproject.toml @@ -29,6 +29,10 @@ dependencies = [ "opentelemetry-propagator-aws-xray", ] +[project.entry-points."aws_durable_execution.plugins"] +otel-invocation = "aws_durable_execution_sdk_python_otel.plugin_provider:INVOCATION_OTEL_PLUGIN_PROVIDER" +otel-execution = "aws_durable_execution_sdk_python_otel.plugin_provider:EXECUTION_OTEL_PLUGIN_PROVIDER" + [project.optional-dependencies] # Instrumentation used by ExecutionOtelPlugin's auto-configured provider path. # Kept optional so the InvocationOtelPlugin (ADOT / global provider) install diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/plugin_provider.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/plugin_provider.py new file mode 100644 index 00000000..24f1e34f --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/plugin_provider.py @@ -0,0 +1,19 @@ +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPluginProvider, +) + +from aws_durable_execution_sdk_python_otel.execution_plugin import ( + ExecutionOtelPlugin, +) +from aws_durable_execution_sdk_python_otel.invocation_plugin import ( + InvocationOtelPlugin, +) + + +INVOCATION_OTEL_PLUGIN_PROVIDER = DurableInstrumentationPluginProvider( + plugin_type=InvocationOtelPlugin, factory=InvocationOtelPlugin +) + +EXECUTION_OTEL_PLUGIN_PROVIDER = DurableInstrumentationPluginProvider( + plugin_type=ExecutionOtelPlugin, factory=ExecutionOtelPlugin +) diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_plugin_provider.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_plugin_provider.py new file mode 100644 index 00000000..9e61abb5 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_plugin_provider.py @@ -0,0 +1,38 @@ +from aws_durable_execution_sdk_python.plugin import ( + DURABLE_INSTRUMENTATION_PLUGIN_API_VERSION, +) + +from aws_durable_execution_sdk_python_otel.execution_plugin import ( + ExecutionOtelPlugin, +) +from aws_durable_execution_sdk_python_otel.invocation_plugin import ( + InvocationOtelPlugin, +) +from aws_durable_execution_sdk_python_otel.plugin_provider import ( + EXECUTION_OTEL_PLUGIN_PROVIDER, + INVOCATION_OTEL_PLUGIN_PROVIDER, +) + + +def test_invocation_otel_plugin_provider_uses_current_plugin_api() -> None: + assert INVOCATION_OTEL_PLUGIN_PROVIDER.plugin_type is InvocationOtelPlugin + assert ( + INVOCATION_OTEL_PLUGIN_PROVIDER.plugin_api_version + == DURABLE_INSTRUMENTATION_PLUGIN_API_VERSION + ) + + +def test_invocation_otel_plugin_provider_creates_invocation_plugin() -> None: + assert isinstance(INVOCATION_OTEL_PLUGIN_PROVIDER.factory(), InvocationOtelPlugin) + + +def test_execution_otel_plugin_provider_uses_current_plugin_api() -> None: + assert EXECUTION_OTEL_PLUGIN_PROVIDER.plugin_type is ExecutionOtelPlugin + assert ( + EXECUTION_OTEL_PLUGIN_PROVIDER.plugin_api_version + == DURABLE_INSTRUMENTATION_PLUGIN_API_VERSION + ) + + +def test_execution_otel_plugin_provider_creates_execution_plugin() -> None: + assert isinstance(EXECUTION_OTEL_PLUGIN_PROVIDER.factory(), ExecutionOtelPlugin) diff --git a/packages/aws-durable-execution-sdk-python/README.md b/packages/aws-durable-execution-sdk-python/README.md index 419a49a5..d91b1c0c 100644 --- a/packages/aws-durable-execution-sdk-python/README.md +++ b/packages/aws-durable-execution-sdk-python/README.md @@ -27,6 +27,52 @@ Build reliable, long-running AWS Lambda workflows with checkpointed steps, waits | `aws-durable-execution-sdk-python` | Execution SDK for Lambda durable functions | [![PyPI - Version](https://img.shields.io/pypi/v/aws-durable-execution-sdk-python.svg)](https://pypi.org/project/aws-durable-execution-sdk-python) | | `aws-durable-execution-sdk-python-testing` | Local/cloud test runner and pytest helpers | [![PyPI - Version](https://img.shields.io/pypi/v/aws-durable-execution-sdk-python-testing.svg)](https://pypi.org/project/aws-durable-execution-sdk-python-testing) | +## Dynamic instrumentation plugins + +Instrumentation plugins can be selected at Lambda cold start without importing +them in the function artifact. Install a provider package in the function or a +Lambda layer, then set an ordered allow-list: + +```text +DURABLE_EXECUTION_PLUGINS=otel-invocation,example_audit +``` + +The SDK resolves those names from the `aws_durable_execution.plugins` Python +entry-point group when the decorated handler is initialized. An unset or blank +variable preserves the existing behavior. The decorator's `plugins` argument +remains supported; explicit plugins run first and take precedence when a +dynamic provider creates the same concrete plugin type. + +Provider packages expose a versioned factory: + +```python +from aws_durable_execution_sdk_python.plugin import ( + DurableInstrumentationPlugin, + DurableInstrumentationPluginProvider, +) + + +class AuditPlugin(DurableInstrumentationPlugin): + pass + + +AUDIT_PLUGIN_PROVIDER = DurableInstrumentationPluginProvider( + plugin_type=AuditPlugin, + factory=AuditPlugin, +) +``` + +Register the provider in the package's `pyproject.toml`: + +```toml +[project.entry-points."aws_durable_execution.plugins"] +example_audit = "example_audit:AUDIT_PLUGIN_PROVIDER" +``` + +Provider names must be unique across installed distributions. Missing, +ambiguous, incompatible, or invalid providers raise `PluginLoadError` during +handler initialization with the provider and distribution details. + ## 🚀 Quick Start Install the execution SDK: diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/__init__.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/__init__.py index 6871c5d3..a676908c 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/__init__.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/__init__.py @@ -28,6 +28,7 @@ ExecutionError, InvocationError, InvokeError, + PluginLoadError, StepError, ValidationError, WaitForConditionError, @@ -55,6 +56,7 @@ "InvocationError", "InvokeError", "ParallelBranch", + "PluginLoadError", "StepContext", "StepError", "ValidationError", diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/exceptions.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/exceptions.py index 46c23356..145edf6e 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/exceptions.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/exceptions.py @@ -62,6 +62,10 @@ class DurableExecutionsError(Exception): """Base class for Durable Executions exceptions""" +class PluginLoadError(DurableExecutionsError): + """A dynamically configured instrumentation plugin could not be loaded.""" + + class UnrecoverableError(DurableExecutionsError): """Base class for errors that terminate execution.""" diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/execution.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/execution.py index 7b280d46..e5b367c4 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/execution.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/execution.py @@ -31,6 +31,9 @@ DurableInstrumentationPlugin, PluginExecutor, ) +from aws_durable_execution_sdk_python.plugin_discovery import ( + load_configured_plugins, +) from aws_durable_execution_sdk_python.state import ExecutionState, ReplayStatus @@ -194,7 +197,7 @@ def durable_execution( stacklevel=2, # point the warning to the caller of durable_execution ) - plugin_executor = PluginExecutor(plugins) + plugin_executor = PluginExecutor(load_configured_plugins(plugins)) @plugin_executor.handle_durable_output def wrapper(event: Any, context: LambdaContext) -> MutableMapping[str, Any]: diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin.py index 197a9cf3..f22ac571 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin.py @@ -27,6 +27,8 @@ logger = logging.getLogger(__name__) +DURABLE_INSTRUMENTATION_PLUGIN_API_VERSION = 1 + def _extract_result(operation: Operation) -> str | None: if operation.step_details and operation.step_details.result is not None: @@ -261,6 +263,15 @@ def on_user_function_end(self, info: UserFunctionEndInfo) -> None: pass +@dataclass(frozen=True) +class DurableInstrumentationPluginProvider: + """Versioned factory exposed through the plugin entry-point group.""" + + plugin_type: type[DurableInstrumentationPlugin] + factory: Callable[[], DurableInstrumentationPlugin] + plugin_api_version: int = DURABLE_INSTRUMENTATION_PLUGIN_API_VERSION + + class PluginExecutor: def __init__(self, plugins: list[DurableInstrumentationPlugin] | None): self._plugins = plugins or [] diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin_discovery.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin_discovery.py new file mode 100644 index 00000000..ffb07260 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin_discovery.py @@ -0,0 +1,209 @@ +from __future__ import annotations + +import logging +import os +from collections.abc import Mapping, Sequence +from importlib import metadata + +from aws_durable_execution_sdk_python.__about__ import __version__ +from aws_durable_execution_sdk_python.exceptions import PluginLoadError +from aws_durable_execution_sdk_python.plugin import ( + DURABLE_INSTRUMENTATION_PLUGIN_API_VERSION, + DurableInstrumentationPlugin, + DurableInstrumentationPluginProvider, +) + + +logger = logging.getLogger(__name__) + +PLUGIN_ENTRY_POINT_GROUP = "aws_durable_execution.plugins" +PLUGIN_ENVIRONMENT_VARIABLE = "DURABLE_EXECUTION_PLUGINS" + + +def _parse_configured_plugin_names(environment: Mapping[str, str]) -> list[str]: + configured_plugins = environment.get(PLUGIN_ENVIRONMENT_VARIABLE) + if configured_plugins is None or not configured_plugins.strip(): + return [] + + plugin_names = [name.strip() for name in configured_plugins.split(",")] + if any(not name for name in plugin_names): + raise PluginLoadError( + f"{PLUGIN_ENVIRONMENT_VARIABLE} must contain non-empty, " + "comma-separated plugin names." + ) + + seen_names: set[str] = set() + for plugin_name in plugin_names: + if plugin_name in seen_names: + raise PluginLoadError( + f"{PLUGIN_ENVIRONMENT_VARIABLE} contains duplicate plugin name " + f"'{plugin_name}'." + ) + seen_names.add(plugin_name) + + return plugin_names + + +def _distribution_name(entry_point: metadata.EntryPoint) -> str: + distribution = getattr(entry_point, "dist", None) + if distribution is None: + return "unknown distribution" + return distribution.metadata.get("Name", "unknown distribution") + + +def _qualified_type_name(value: object) -> str: + value_type = type(value) + return f"{value_type.__module__}.{value_type.__qualname__}" + + +def _qualified_class_name(value_type: type[object]) -> str: + return f"{value_type.__module__}.{value_type.__qualname__}" + + +def _load_provider( + plugin_name: str, entry_point: metadata.EntryPoint +) -> DurableInstrumentationPluginProvider: + try: + provider = entry_point.load() + except Exception as error: + raise PluginLoadError( + f"Failed to load durable instrumentation plugin provider " + f"'{plugin_name}' from '{entry_point.value}' " + f"({_distribution_name(entry_point)}): {error}" + ) from error + + if not isinstance(provider, DurableInstrumentationPluginProvider): + raise PluginLoadError( + f"Durable instrumentation plugin entry point '{plugin_name}' must " + "resolve to DurableInstrumentationPluginProvider, but resolved to " + f"{_qualified_type_name(provider)}." + ) + + if provider.plugin_api_version != DURABLE_INSTRUMENTATION_PLUGIN_API_VERSION: + raise PluginLoadError( + f"Durable instrumentation plugin provider '{plugin_name}' declares " + f"plugin API version {provider.plugin_api_version}, but " + f"aws-durable-execution-sdk-python {__version__} supports plugin API " + f"version {DURABLE_INSTRUMENTATION_PLUGIN_API_VERSION}. Install " + "compatible SDK and plugin package versions." + ) + + declared_plugin_type: object = provider.plugin_type + if not isinstance(declared_plugin_type, type) or not issubclass( + declared_plugin_type, DurableInstrumentationPlugin + ): + declared_type_name = ( + _qualified_class_name(declared_plugin_type) + if isinstance(declared_plugin_type, type) + else _qualified_type_name(declared_plugin_type) + ) + raise PluginLoadError( + f"Durable instrumentation plugin provider '{plugin_name}' declares " + f"invalid plugin type {declared_type_name}; " + "expected a DurableInstrumentationPlugin subclass." + ) + + return provider + + +def _create_plugin( + plugin_name: str, + entry_point: metadata.EntryPoint, + provider: DurableInstrumentationPluginProvider, +) -> DurableInstrumentationPlugin: + try: + plugin = provider.factory() + except Exception as error: + raise PluginLoadError( + f"Failed to create durable instrumentation plugin '{plugin_name}' " + f"from '{entry_point.value}' ({_distribution_name(entry_point)}): " + f"{error}" + ) from error + + if type(plugin) is not provider.plugin_type: + raise PluginLoadError( + f"Durable instrumentation plugin provider '{plugin_name}' returned " + f"{_qualified_type_name(plugin)}; expected " + f"{_qualified_class_name(provider.plugin_type)}." + ) + + return plugin + + +def load_configured_plugins( + explicit_plugins: Sequence[DurableInstrumentationPlugin] | None, + *, + environment: Mapping[str, str] | None = None, +) -> list[DurableInstrumentationPlugin]: + """Combine explicit plugins with providers selected through the environment. + + Explicit plugins retain their order. Dynamically selected plugins follow in + configured order. When discovery creates a plugin whose concrete type is + already registered, the first registration wins, so explicit registration + takes precedence. + """ + + resolved_plugins = list(explicit_plugins or []) + resolved_environment = os.environ if environment is None else environment + plugin_names = _parse_configured_plugin_names(resolved_environment) + if not plugin_names: + return resolved_plugins + + try: + discovered_entry_points = list( + metadata.entry_points(group=PLUGIN_ENTRY_POINT_GROUP) + ) + except Exception as error: + raise PluginLoadError( + "Failed to inspect installed durable instrumentation plugin " + f"providers in entry-point group '{PLUGIN_ENTRY_POINT_GROUP}': {error}" + ) from error + + entry_points_by_name: dict[str, list[metadata.EntryPoint]] = {} + for entry_point in discovered_entry_points: + entry_points_by_name.setdefault(entry_point.name, []).append(entry_point) + + registered_types: dict[type[DurableInstrumentationPlugin], str] = { + type(plugin): "the decorator's plugins argument" for plugin in resolved_plugins + } + + for plugin_name in plugin_names: + matching_entry_points = entry_points_by_name.get(plugin_name, []) + if not matching_entry_points: + available_names = ", ".join(sorted(entry_points_by_name)) or "none" + raise PluginLoadError( + f"No durable instrumentation plugin provider named " + f"'{plugin_name}' was found in entry-point group " + f"'{PLUGIN_ENTRY_POINT_GROUP}'. Installed providers: " + f"{available_names}. Ensure the provider package is installed " + "in the function artifact or an attached Lambda layer." + ) + + if len(matching_entry_points) > 1: + distributions = ", ".join( + _distribution_name(entry_point) for entry_point in matching_entry_points + ) + raise PluginLoadError( + f"Multiple durable instrumentation plugin providers named " + f"'{plugin_name}' were found in entry-point group " + f"'{PLUGIN_ENTRY_POINT_GROUP}': {distributions}. Remove the " + "duplicate provider package." + ) + + entry_point = matching_entry_points[0] + provider = _load_provider(plugin_name, entry_point) + if existing_registration := registered_types.get(provider.plugin_type): + logger.warning( + "Skipping dynamically configured plugin '%s' because %s is " + "already registered by %s.", + plugin_name, + _qualified_class_name(provider.plugin_type), + existing_registration, + ) + continue + + plugin = _create_plugin(plugin_name, entry_point, provider) + resolved_plugins.append(plugin) + registered_types[provider.plugin_type] = f"dynamic provider '{plugin_name}'" + + return resolved_plugins diff --git a/packages/aws-durable-execution-sdk-python/tests/execution_test.py b/packages/aws-durable-execution-sdk-python/tests/execution_test.py index a03fe7f8..3334815f 100644 --- a/packages/aws-durable-execution-sdk-python/tests/execution_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/execution_test.py @@ -2908,6 +2908,27 @@ def on_operation_attempt_end(self, info): raise RuntimeError("plugin boom") +def test_durable_execution_loads_plugins_when_handler_is_initialized(): + """Configured plugins are resolved once while the decorator initializes.""" + explicit_plugin = _RecordingPlugin() + resolved_plugin = _RecordingPlugin() + + with ( + patch( + "aws_durable_execution_sdk_python.execution.load_configured_plugins", + return_value=[explicit_plugin, resolved_plugin], + ) as load_plugins, + pytest.warns(FutureWarning), + ): + + @durable_execution(plugins=[explicit_plugin]) + def test_handler(event: Any, context: DurableContext) -> dict: + return {"result": "success"} + + load_plugins.assert_called_once_with([explicit_plugin]) + assert callable(test_handler) + + def test_durable_execution_with_plugins_success(): """Test that plugins receive invocation start/end and execution end on success.""" mock_client = Mock(spec=DurableServiceClient) diff --git a/packages/aws-durable-execution-sdk-python/tests/plugin_discovery_test.py b/packages/aws-durable-execution-sdk-python/tests/plugin_discovery_test.py new file mode 100644 index 00000000..6e5b4129 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python/tests/plugin_discovery_test.py @@ -0,0 +1,469 @@ +from __future__ import annotations + +import logging +import os +from collections.abc import Callable +from typing import cast +from unittest.mock import Mock, patch + +import pytest + +from aws_durable_execution_sdk_python.exceptions import PluginLoadError +from aws_durable_execution_sdk_python.plugin import ( + DURABLE_INSTRUMENTATION_PLUGIN_API_VERSION, + DurableInstrumentationPlugin, + DurableInstrumentationPluginProvider, +) +from aws_durable_execution_sdk_python.plugin_discovery import ( + PLUGIN_ENTRY_POINT_GROUP, + PLUGIN_ENVIRONMENT_VARIABLE, + load_configured_plugins, +) + + +class _PluginA(DurableInstrumentationPlugin): + pass + + +class _PluginB(DurableInstrumentationPlugin): + pass + + +class _FakeDistribution: + def __init__(self, name: str) -> None: + self.metadata = {"Name": name} + + +class _FakeEntryPoint: + def __init__( + self, + name: str, + loaded_value: object, + *, + distribution_name: str | None = "test-plugin-package", + load_error: Exception | None = None, + ) -> None: + self.name = name + self.value = f"test_plugins:{name}" + self.dist = ( + _FakeDistribution(distribution_name) + if distribution_name is not None + else None + ) + self._loaded_value = loaded_value + self._load_error = load_error + + def load(self) -> object: + if self._load_error is not None: + raise self._load_error + return self._loaded_value + + +def _provider( + factory: Callable[[], object], + *, + plugin_type: type[DurableInstrumentationPlugin] = _PluginA, + plugin_api_version: int = DURABLE_INSTRUMENTATION_PLUGIN_API_VERSION, +) -> DurableInstrumentationPluginProvider: + return DurableInstrumentationPluginProvider( + plugin_type=plugin_type, + factory=cast(Callable[[], DurableInstrumentationPlugin], factory), + plugin_api_version=plugin_api_version, + ) + + +@pytest.mark.parametrize("configured_value", [None, "", " "]) +def test_unconfigured_discovery_preserves_explicit_plugins( + configured_value: str | None, +) -> None: + explicit_plugin = _PluginA() + environment = ( + {} + if configured_value is None + else {PLUGIN_ENVIRONMENT_VARIABLE: configured_value} + ) + + with patch( + "aws_durable_execution_sdk_python.plugin_discovery.metadata.entry_points" + ) as entry_points: + result = load_configured_plugins( + [explicit_plugin], + environment=environment, + ) + + assert result == [explicit_plugin] + entry_points.assert_not_called() + + +def test_discovery_uses_process_environment_by_default() -> None: + entry_point = _FakeEntryPoint("a", _provider(_PluginA)) + + with ( + patch.dict( + os.environ, + {PLUGIN_ENVIRONMENT_VARIABLE: "a"}, + clear=True, + ), + patch( + "aws_durable_execution_sdk_python.plugin_discovery.metadata.entry_points", + return_value=[entry_point], + ) as entry_points, + ): + result = load_configured_plugins(None) + + assert len(result) == 1 + assert isinstance(result[0], _PluginA) + entry_points.assert_called_once_with(group=PLUGIN_ENTRY_POINT_GROUP) + + +def test_discovery_preserves_configured_order() -> None: + factory_calls: list[str] = [] + + def create_a() -> _PluginA: + factory_calls.append("a") + return _PluginA() + + def create_b() -> _PluginB: + factory_calls.append("b") + return _PluginB() + + entry_points = [ + _FakeEntryPoint("b", _provider(create_b, plugin_type=_PluginB)), + _FakeEntryPoint("a", _provider(create_a)), + ] + + with patch( + "aws_durable_execution_sdk_python.plugin_discovery.metadata.entry_points", + return_value=entry_points, + ): + result = load_configured_plugins( + None, + environment={PLUGIN_ENVIRONMENT_VARIABLE: " a, b "}, + ) + + assert [type(plugin) for plugin in result] == [_PluginA, _PluginB] + assert factory_calls == ["a", "b"] + + +@pytest.mark.parametrize("configured_value", ["a,,b", ",a", "a,"]) +def test_discovery_rejects_empty_plugin_names(configured_value: str) -> None: + with pytest.raises( + PluginLoadError, + match="must contain non-empty, comma-separated plugin names", + ): + load_configured_plugins( + None, + environment={PLUGIN_ENVIRONMENT_VARIABLE: configured_value}, + ) + + +def test_discovery_rejects_duplicate_configured_names() -> None: + with pytest.raises( + PluginLoadError, + match="contains duplicate plugin name 'a'", + ): + load_configured_plugins( + None, + environment={PLUGIN_ENVIRONMENT_VARIABLE: "a,b,a"}, + ) + + +def test_discovery_reports_missing_provider_and_available_names() -> None: + entry_point = _FakeEntryPoint("available", _provider(_PluginA)) + + with ( + patch( + "aws_durable_execution_sdk_python.plugin_discovery.metadata.entry_points", + return_value=[entry_point], + ), + pytest.raises(PluginLoadError) as error, + ): + load_configured_plugins( + None, + environment={PLUGIN_ENVIRONMENT_VARIABLE: "missing"}, + ) + + assert "No durable instrumentation plugin provider named 'missing'" in str( + error.value + ) + assert "Installed providers: available" in str(error.value) + assert "Lambda layer" in str(error.value) + + +def test_discovery_reports_when_no_providers_are_installed() -> None: + with ( + patch( + "aws_durable_execution_sdk_python.plugin_discovery.metadata.entry_points", + return_value=[], + ), + pytest.raises(PluginLoadError, match="Installed providers: none"), + ): + load_configured_plugins( + None, + environment={PLUGIN_ENVIRONMENT_VARIABLE: "missing"}, + ) + + +def test_discovery_rejects_ambiguous_provider_name() -> None: + entry_points = [ + _FakeEntryPoint( + "duplicate", + _provider(_PluginA), + distribution_name="package-a", + ), + _FakeEntryPoint( + "duplicate", + _provider(_PluginB), + distribution_name="package-b", + ), + ] + + with ( + patch( + "aws_durable_execution_sdk_python.plugin_discovery.metadata.entry_points", + return_value=entry_points, + ), + pytest.raises(PluginLoadError) as error, + ): + load_configured_plugins( + None, + environment={PLUGIN_ENVIRONMENT_VARIABLE: "duplicate"}, + ) + + assert "Multiple durable instrumentation plugin providers" in str(error.value) + assert "package-a, package-b" in str(error.value) + + +def test_discovery_wraps_entry_point_enumeration_failure() -> None: + with ( + patch( + "aws_durable_execution_sdk_python.plugin_discovery.metadata.entry_points", + side_effect=RuntimeError("metadata unavailable"), + ), + pytest.raises(PluginLoadError, match="metadata unavailable"), + ): + load_configured_plugins( + None, + environment={PLUGIN_ENVIRONMENT_VARIABLE: "a"}, + ) + + +def test_discovery_wraps_provider_load_failure() -> None: + entry_point = _FakeEntryPoint( + "a", + _provider(_PluginA), + load_error=ImportError("missing dependency"), + ) + + with ( + patch( + "aws_durable_execution_sdk_python.plugin_discovery.metadata.entry_points", + return_value=[entry_point], + ), + pytest.raises(PluginLoadError) as error, + ): + load_configured_plugins( + None, + environment={PLUGIN_ENVIRONMENT_VARIABLE: "a"}, + ) + + assert "Failed to load durable instrumentation plugin provider 'a'" in str( + error.value + ) + assert "test-plugin-package" in str(error.value) + assert isinstance(error.value.__cause__, ImportError) + + +def test_discovery_rejects_invalid_provider_type() -> None: + entry_point = _FakeEntryPoint("a", _PluginA) + + with ( + patch( + "aws_durable_execution_sdk_python.plugin_discovery.metadata.entry_points", + return_value=[entry_point], + ), + pytest.raises( + PluginLoadError, + match="must resolve to DurableInstrumentationPluginProvider", + ), + ): + load_configured_plugins( + None, + environment={PLUGIN_ENVIRONMENT_VARIABLE: "a"}, + ) + + +def test_discovery_rejects_incompatible_plugin_api_version() -> None: + entry_point = _FakeEntryPoint( + "a", + _provider(_PluginA, plugin_api_version=99), + ) + + with ( + patch( + "aws_durable_execution_sdk_python.plugin_discovery.metadata.entry_points", + return_value=[entry_point], + ), + pytest.raises(PluginLoadError) as error, + ): + load_configured_plugins( + None, + environment={PLUGIN_ENVIRONMENT_VARIABLE: "a"}, + ) + + assert "declares plugin API version 99" in str(error.value) + assert ( + f"supports plugin API version {DURABLE_INSTRUMENTATION_PLUGIN_API_VERSION}" + in str(error.value) + ) + + +def test_discovery_rejects_invalid_declared_plugin_type() -> None: + provider = DurableInstrumentationPluginProvider( + plugin_type=cast(type[DurableInstrumentationPlugin], object), + factory=_PluginA, + ) + entry_point = _FakeEntryPoint("a", provider) + + with ( + patch( + "aws_durable_execution_sdk_python.plugin_discovery.metadata.entry_points", + return_value=[entry_point], + ), + pytest.raises( + PluginLoadError, + match="declares invalid plugin type builtins.object", + ), + ): + load_configured_plugins( + None, + environment={PLUGIN_ENVIRONMENT_VARIABLE: "a"}, + ) + + +def test_discovery_rejects_non_class_declared_plugin_type() -> None: + provider = DurableInstrumentationPluginProvider( + plugin_type=cast(type[DurableInstrumentationPlugin], _PluginA()), + factory=_PluginA, + ) + entry_point = _FakeEntryPoint("a", provider) + + with ( + patch( + "aws_durable_execution_sdk_python.plugin_discovery.metadata.entry_points", + return_value=[entry_point], + ), + pytest.raises( + PluginLoadError, + match="declares invalid plugin type .*_PluginA", + ), + ): + load_configured_plugins( + None, + environment={PLUGIN_ENVIRONMENT_VARIABLE: "a"}, + ) + + +def test_discovery_wraps_plugin_factory_failure() -> None: + def fail_factory() -> _PluginA: + raise RuntimeError("factory failed") + + entry_point = _FakeEntryPoint( + "a", + _provider(fail_factory), + distribution_name=None, + ) + + with ( + patch( + "aws_durable_execution_sdk_python.plugin_discovery.metadata.entry_points", + return_value=[entry_point], + ), + pytest.raises(PluginLoadError) as error, + ): + load_configured_plugins( + None, + environment={PLUGIN_ENVIRONMENT_VARIABLE: "a"}, + ) + + assert "Failed to create durable instrumentation plugin 'a'" in str(error.value) + assert "unknown distribution" in str(error.value) + assert isinstance(error.value.__cause__, RuntimeError) + + +def test_discovery_rejects_invalid_plugin_type() -> None: + entry_point = _FakeEntryPoint("a", _provider(lambda: object())) + + with ( + patch( + "aws_durable_execution_sdk_python.plugin_discovery.metadata.entry_points", + return_value=[entry_point], + ), + pytest.raises( + PluginLoadError, + match="expected .*_PluginA", + ), + ): + load_configured_plugins( + None, + environment={PLUGIN_ENVIRONMENT_VARIABLE: "a"}, + ) + + +def test_explicit_plugin_registration_takes_precedence( + caplog: pytest.LogCaptureFixture, +) -> None: + explicit_plugin = _PluginA() + factory = Mock(return_value=_PluginA()) + entry_point = _FakeEntryPoint("a", _provider(factory)) + + with ( + patch( + "aws_durable_execution_sdk_python.plugin_discovery.metadata.entry_points", + return_value=[entry_point], + ), + caplog.at_level( + logging.WARNING, + logger="aws_durable_execution_sdk_python.plugin_discovery", + ), + ): + result = load_configured_plugins( + [explicit_plugin], + environment={PLUGIN_ENVIRONMENT_VARIABLE: "a"}, + ) + + assert result == [explicit_plugin] + factory.assert_not_called() + assert "already registered by the decorator's plugins argument" in caplog.text + + +def test_first_dynamic_registration_wins_for_duplicate_plugin_type( + caplog: pytest.LogCaptureFixture, +) -> None: + first_factory = Mock(return_value=_PluginA()) + second_factory = Mock(return_value=_PluginA()) + entry_points = [ + _FakeEntryPoint("first", _provider(first_factory)), + _FakeEntryPoint("second", _provider(second_factory)), + ] + + with ( + patch( + "aws_durable_execution_sdk_python.plugin_discovery.metadata.entry_points", + return_value=entry_points, + ), + caplog.at_level( + logging.WARNING, + logger="aws_durable_execution_sdk_python.plugin_discovery", + ), + ): + result = load_configured_plugins( + None, + environment={PLUGIN_ENVIRONMENT_VARIABLE: "first,second"}, + ) + + assert len(result) == 1 + assert isinstance(result[0], _PluginA) + first_factory.assert_called_once_with() + second_factory.assert_not_called() + assert "already registered by dynamic provider 'first'" in caplog.text From 0a74dce9f8cc480899ec586dd9e9d3c993b5c0a0 Mon Sep 17 00:00:00 2001 From: Frank Chen <65260095+zhongkechen@users.noreply.github.com> Date: Mon, 10 Aug 2026 11:42:52 -0700 Subject: [PATCH 2/7] fix(otel): require plugin discovery core API --- packages/aws-durable-execution-sdk-python-otel/README.md | 2 +- packages/aws-durable-execution-sdk-python-otel/pyproject.toml | 2 +- pyproject.toml | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python-otel/README.md b/packages/aws-durable-execution-sdk-python-otel/README.md index aca6a207..4eb563aa 100644 --- a/packages/aws-durable-execution-sdk-python-otel/README.md +++ b/packages/aws-durable-execution-sdk-python-otel/README.md @@ -306,7 +306,7 @@ setups. ## Requirements - Python >= 3.11 -- `aws-durable-execution-sdk-python` >= 1.5.0 +- `aws-durable-execution-sdk-python` >= 1.8.0 - `opentelemetry-api` >= 1.20.0 - `opentelemetry-sdk` >= 1.20.0 - `opentelemetry-exporter-otlp` diff --git a/packages/aws-durable-execution-sdk-python-otel/pyproject.toml b/packages/aws-durable-execution-sdk-python-otel/pyproject.toml index dee495ff..9f9d4ff4 100644 --- a/packages/aws-durable-execution-sdk-python-otel/pyproject.toml +++ b/packages/aws-durable-execution-sdk-python-otel/pyproject.toml @@ -22,7 +22,7 @@ classifiers = [ "Programming Language :: Python :: Implementation :: PyPy", ] dependencies = [ - "aws-durable-execution-sdk-python>=1.5.0", + "aws-durable-execution-sdk-python>=1.8.0", "opentelemetry-api>=1.20.0", "opentelemetry-sdk>=1.20.0", "opentelemetry-exporter-otlp", diff --git a/pyproject.toml b/pyproject.toml index 740daf38..6ea8e7c1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -124,7 +124,7 @@ test = "pytest packages/aws-durable-execution-sdk-python-examples/test {args}" [tool.hatch.envs.test-pypi-otel] dependencies = [ - "aws-durable-execution-sdk-python", + "aws-durable-execution-sdk-python>=1.8.0", "opentelemetry-sdk>=1.20.0", "pytest", "pytest-cov", From 03666105a01a4838a2d8641360f6ac53bfd55304 Mon Sep 17 00:00:00 2001 From: Frank Chen <65260095+zhongkechen@users.noreply.github.com> Date: Mon, 10 Aug 2026 11:54:12 -0700 Subject: [PATCH 3/7] fix: require providers to pin plugin API version --- .../plugin_provider.py | 8 ++++++-- packages/aws-durable-execution-sdk-python/README.md | 4 ++++ .../src/aws_durable_execution_sdk_python/plugin.py | 2 +- .../tests/plugin_discovery_test.py | 10 ++++++++++ 4 files changed, 21 insertions(+), 3 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/plugin_provider.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/plugin_provider.py index 24f1e34f..c4be734c 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/plugin_provider.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/plugin_provider.py @@ -11,9 +11,13 @@ INVOCATION_OTEL_PLUGIN_PROVIDER = DurableInstrumentationPluginProvider( - plugin_type=InvocationOtelPlugin, factory=InvocationOtelPlugin + plugin_type=InvocationOtelPlugin, + factory=InvocationOtelPlugin, + plugin_api_version=1, ) EXECUTION_OTEL_PLUGIN_PROVIDER = DurableInstrumentationPluginProvider( - plugin_type=ExecutionOtelPlugin, factory=ExecutionOtelPlugin + plugin_type=ExecutionOtelPlugin, + factory=ExecutionOtelPlugin, + plugin_api_version=1, ) diff --git a/packages/aws-durable-execution-sdk-python/README.md b/packages/aws-durable-execution-sdk-python/README.md index d91b1c0c..bf7776f4 100644 --- a/packages/aws-durable-execution-sdk-python/README.md +++ b/packages/aws-durable-execution-sdk-python/README.md @@ -59,6 +59,7 @@ class AuditPlugin(DurableInstrumentationPlugin): AUDIT_PLUGIN_PROVIDER = DurableInstrumentationPluginProvider( plugin_type=AuditPlugin, factory=AuditPlugin, + plugin_api_version=1, ) ``` @@ -69,6 +70,9 @@ Register the provider in the package's `pyproject.toml`: example_audit = "example_audit:AUDIT_PLUGIN_PROVIDER" ``` +Set `plugin_api_version` to the literal API version the provider implements. +Update it only after verifying the provider against that API version. + Provider names must be unique across installed distributions. Missing, ambiguous, incompatible, or invalid providers raise `PluginLoadError` during handler initialization with the provider and distribution details. diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin.py index f22ac571..72cd6820 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/plugin.py @@ -269,7 +269,7 @@ class DurableInstrumentationPluginProvider: plugin_type: type[DurableInstrumentationPlugin] factory: Callable[[], DurableInstrumentationPlugin] - plugin_api_version: int = DURABLE_INSTRUMENTATION_PLUGIN_API_VERSION + plugin_api_version: int class PluginExecutor: diff --git a/packages/aws-durable-execution-sdk-python/tests/plugin_discovery_test.py b/packages/aws-durable-execution-sdk-python/tests/plugin_discovery_test.py index 6e5b4129..51200ba3 100644 --- a/packages/aws-durable-execution-sdk-python/tests/plugin_discovery_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/plugin_discovery_test.py @@ -72,6 +72,14 @@ def _provider( ) +def test_plugin_provider_requires_authored_api_version() -> None: + with pytest.raises(TypeError, match="plugin_api_version"): + DurableInstrumentationPluginProvider( + plugin_type=_PluginA, + factory=_PluginA, + ) # type: ignore[call-arg] + + @pytest.mark.parametrize("configured_value", [None, "", " "]) def test_unconfigured_discovery_preserves_explicit_plugins( configured_value: str | None, @@ -322,6 +330,7 @@ def test_discovery_rejects_invalid_declared_plugin_type() -> None: provider = DurableInstrumentationPluginProvider( plugin_type=cast(type[DurableInstrumentationPlugin], object), factory=_PluginA, + plugin_api_version=DURABLE_INSTRUMENTATION_PLUGIN_API_VERSION, ) entry_point = _FakeEntryPoint("a", provider) @@ -345,6 +354,7 @@ def test_discovery_rejects_non_class_declared_plugin_type() -> None: provider = DurableInstrumentationPluginProvider( plugin_type=cast(type[DurableInstrumentationPlugin], _PluginA()), factory=_PluginA, + plugin_api_version=DURABLE_INSTRUMENTATION_PLUGIN_API_VERSION, ) entry_point = _FakeEntryPoint("a", provider) From c4d0f27fc41c0d3421dde50c6b23a642a3f09bfa Mon Sep 17 00:00:00 2001 From: Frank Chen <65260095+zhongkechen@users.noreply.github.com> Date: Mon, 10 Aug 2026 12:03:16 -0700 Subject: [PATCH 4/7] test(otel): discover installed plugin entry points --- .../tests/test_plugin_provider.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_plugin_provider.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_plugin_provider.py index 9e61abb5..c47098b2 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_plugin_provider.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_plugin_provider.py @@ -1,6 +1,10 @@ from aws_durable_execution_sdk_python.plugin import ( DURABLE_INSTRUMENTATION_PLUGIN_API_VERSION, ) +from aws_durable_execution_sdk_python.plugin_discovery import ( + PLUGIN_ENVIRONMENT_VARIABLE, + load_configured_plugins, +) from aws_durable_execution_sdk_python_otel.execution_plugin import ( ExecutionOtelPlugin, @@ -36,3 +40,17 @@ def test_execution_otel_plugin_provider_uses_current_plugin_api() -> None: def test_execution_otel_plugin_provider_creates_execution_plugin() -> None: assert isinstance(EXECUTION_OTEL_PLUGIN_PROVIDER.factory(), ExecutionOtelPlugin) + + +def test_installed_otel_entry_points_load_both_plugin_types() -> None: + plugins = load_configured_plugins( + None, + environment={ + PLUGIN_ENVIRONMENT_VARIABLE: "otel-invocation,otel-execution", + }, + ) + + assert [type(plugin) for plugin in plugins] == [ + InvocationOtelPlugin, + ExecutionOtelPlugin, + ] From f984e02174db3b8e51276fb260e7017b26483152 Mon Sep 17 00:00:00 2001 From: Frank Chen <65260095+zhongkechen@users.noreply.github.com> Date: Mon, 10 Aug 2026 12:32:57 -0700 Subject: [PATCH 5/7] fix: align SDK version with plugin API release --- .../src/aws_durable_execution_sdk_python/__about__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/__about__.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/__about__.py index b7ac8fde..8ce3c043 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/__about__.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/__about__.py @@ -1,4 +1,4 @@ # SPDX-FileCopyrightText: 2025-present Amazon.com, Inc. or its affiliates. # # SPDX-License-Identifier: Apache-2.0 -__version__ = "1.7.0" +__version__ = "1.8.0" From 8bdd08ebc7a76bcefe0ace81269db88706eb6c70 Mon Sep 17 00:00:00 2001 From: Frank Chen <65260095+zhongkechen@users.noreply.github.com> Date: Mon, 10 Aug 2026 12:57:52 -0700 Subject: [PATCH 6/7] fix: align conformance tests with SDK version --- .../pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/pyproject.toml b/packages/aws-durable-execution-sdk-python-conformance-tests/pyproject.toml index 14f6bd9b..39ff3cab 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/pyproject.toml +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/pyproject.toml @@ -8,7 +8,7 @@ version = "0.0.0" description = "Cross-SDK conformance test handlers for the AWS Durable Execution SDK for Python, exercised by the aws-durable-execution-conformance-tests runner." requires-python = ">=3.11" dependencies = [ - "aws-durable-execution-sdk-python==1.7.0", + "aws-durable-execution-sdk-python==1.8.0", ] [tool.hatch.build.targets.wheel] From 3a7ef0a277d0d19a7288bd8ac8652a14a9234e16 Mon Sep 17 00:00:00 2001 From: Frank Chen <65260095+zhongkechen@users.noreply.github.com> Date: Mon, 10 Aug 2026 13:06:46 -0700 Subject: [PATCH 7/7] chore(release): bump otel to 0.4.0 --- .../src/aws_durable_execution_sdk_python_otel/__about__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/__about__.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/__about__.py index 9c3804e8..08043af9 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/__about__.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/__about__.py @@ -1,4 +1,4 @@ # SPDX-FileCopyrightText: 2025-present Amazon.com, Inc. or its affiliates. # # SPDX-License-Identifier: Apache-2.0 -__version__ = "0.3.0" +__version__ = "0.4.0"