From 4d09926503ca2cf2e97bfa202f01b23833b75b38 Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Tue, 20 Oct 2026 22:39:05 +0000 Subject: [PATCH 1/5] test(conformance): add lifecycle log stability diagnostics --- .../scripts/check_lifecycle_log_stability.py | 183 +++++++++++++++ .../test_check_lifecycle_log_stability.py | 215 ++++++++++++++++++ 2 files changed, 398 insertions(+) create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/scripts/check_lifecycle_log_stability.py create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/tests/test_check_lifecycle_log_stability.py diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/scripts/check_lifecycle_log_stability.py b/packages/aws-durable-execution-sdk-python-conformance-tests/scripts/check_lifecycle_log_stability.py new file mode 100644 index 000000000..90d101c91 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/scripts/check_lifecycle_log_stability.py @@ -0,0 +1,183 @@ +#!/usr/bin/env python3 +"""Repeat upstream requirement 10-1 against an already deployed plugin stack. + +Requires the installed aws-durable-execution-conformance-tests runner. Each +sample uses its unchanged history, result, and log assertions. Raw CloudWatch +pages and invocation responses are retained so ordering failures can be traced +to their original event timestamps and ingestion times. +""" + +from __future__ import annotations + +import argparse +import json +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from typing import Any + + +def _load_runner() -> tuple[Any, Any, Any, Path]: + # Keep the upstream runner optional for the SDK's ordinary unit tests. + from aws_durable_execution_conformance_tests.clients import AwsClients + from aws_durable_execution_conformance_tests.config import TESTS_DIR + from aws_durable_execution_conformance_tests.sam import Invoker + from aws_durable_execution_conformance_tests.validate import validate_description + + return AwsClients, Invoker, validate_description, TESTS_DIR / "plugin" / "10-1.yaml" + + +def _write_json(path: Path, value: Any) -> None: + path.write_text(json.dumps(value, indent=2, default=str) + "\n", encoding="utf-8") + + +class RecordingLogsClient: + """Forward requests unchanged and persist every page before validation.""" + + def __init__(self, client: Any, path: Path) -> None: + self.client = client + self.path = path + + def filter_log_events(self, **kwargs: Any) -> dict[str, Any]: + record: dict[str, Any] = {"request": kwargs} + try: + response = self.client.filter_log_events(**kwargs) + record["response"] = response + return response + except Exception as exc: + record["error"] = f"{type(exc).__name__}: {exc}" + raise + finally: + with self.path.open("a", encoding="utf-8") as output: + output.write(json.dumps(record, default=str) + "\n") + + +class RecordingInvoker: + """Keep the execution ARN even when upstream returns a failed result.""" + + def __init__(self, invoker: Any, path: Path) -> None: + self.invoker = invoker + self.stack_name = invoker.stack_name + self.path = path + self.execution_arn: str | None = None + + def invoke(self, **kwargs: Any) -> Any: + result = self.invoker.invoke(**kwargs) + self.path.write_text(result.output, encoding="utf-8") + try: + response = json.loads(result.output) + except ValueError: + pass # Upstream reports malformed invocation responses itself. + else: + if isinstance(response, dict): + self.execution_arn = response.get("DurableExecutionArn") + return result + + +def _run_sample( + number: int, + args: argparse.Namespace, + clients: Any, + invoker: Any, + runner: tuple[Any, Any, Any, Path], +) -> dict[str, Any]: + clients_type, _, validate_description, requirement = runner + sample_dir = args.output_dir / f"sample-{number:04d}" + sample_dir.mkdir() + events_dir = sample_dir / "events" + events_dir.mkdir() + recording_invoker = RecordingInvoker(invoker, sample_dir / "invocation.json") + recording_clients = clients_type( + { + **dict(clients), + "logs": RecordingLogsClient( + clients["logs"], sample_dir / "cloudwatch-pages.jsonl" + ), + } + ) + result: dict[str, Any] = {"sample": number, "passed": False, "errors": []} + try: + validated = validate_description( + function_name="PluginInvocationLifecycle", + description_id="10-1", + test_file=str(requirement), + invoker=recording_invoker, + tmp_dir=str(events_dir), + region=args.region, + aws_clients=recording_clients, + output_dir=str(sample_dir / "history"), + log_poll_timeout_seconds=args.log_poll_timeout_seconds, + ) + result.update( + passed=validated.passed, + errors=validated.errors, + placeholders=validated.placeholders, + ) + except Exception as exc: + # Keep sampling after a failed request, without losing earlier pages. + result["errors"] = [f"{type(exc).__name__}: {exc}"] + result["execution_arn"] = recording_invoker.execution_arn + _write_json(sample_dir / "result.json", result) + print(f"Sample {number}: {'PASSED' if result['passed'] else 'FAILED'}", flush=True) + return result + + +def _positive_int(value: str) -> int: + number = int(value) + if number < 1: + raise argparse.ArgumentTypeError("must be at least 1") + return number + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--stack-name", required=True) + parser.add_argument("--region", default="us-west-2") + parser.add_argument("--samples", type=_positive_int, default=40) + parser.add_argument("--workers", type=int, choices=range(1, 17), default=4) + parser.add_argument("--output-dir", type=Path, required=True) + parser.add_argument( + "--log-poll-timeout-seconds", + type=_positive_int, + default=120, + help="Maximum upstream polling budget per sample; passing logs normally " + "finish after the runner's 20-second observation window.", + ) + args = parser.parse_args(argv) + runner = _load_runner() + clients_type, invoker_type, _, requirement = runner + if not requirement.is_file(): + parser.error(f"Upstream requirement not found: {requirement}") + if args.output_dir.exists(): + parser.error("--output-dir must be new to preserve previous diagnostic results") + args.output_dir.mkdir(parents=True) + clients = clients_type.create(args.region) + invoker = invoker_type( + stack_name=args.stack_name, + region=args.region, + lambda_client=clients["lambda"], + cfn_client=clients["cloudformation"], + ) + with ThreadPoolExecutor(max_workers=min(args.workers, args.samples)) as executor: + results = list( + executor.map( + lambda number: _run_sample(number, args, clients, invoker, runner), + range(1, args.samples + 1), + ) + ) + passed = sum(result["passed"] for result in results) + summary = { + "stack_name": args.stack_name, + "region": args.region, + "requirement": "10-1", + "samples": args.samples, + "passed": passed, + "failed": args.samples - passed, + "results": results, + } + _write_json(args.output_dir / "summary.json", summary) + print(f"Lifecycle log checks: {passed}/{args.samples} passed", flush=True) + return 0 if passed == args.samples else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/tests/test_check_lifecycle_log_stability.py b/packages/aws-durable-execution-sdk-python-conformance-tests/tests/test_check_lifecycle_log_stability.py new file mode 100644 index 000000000..9b3140f7c --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/tests/test_check_lifecycle_log_stability.py @@ -0,0 +1,215 @@ +"""Boundary tests for repeated upstream lifecycle validation and its artifacts.""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + + +SCRIPTS_DIR = Path(__file__).resolve().parents[1] / "scripts" +if str(SCRIPTS_DIR) not in sys.path: + sys.path.insert(0, str(SCRIPTS_DIR)) + +import check_lifecycle_log_stability as stability # noqa: E402 + + +@pytest.fixture +def runner(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> SimpleNamespace: + requirement = tmp_path / "10-1.yaml" + requirement.write_text("ExpectedLogs: []\n") + event = { + "eventId": "original-event-id", + "logStreamName": "original-stream", + "timestamp": 10, + "ingestionTime": 20, + "message": '{"hook": "invocation-start"}', + } + logs = Mock() + logs.filter_log_events.return_value = {"events": [event], "nextToken": "next-page"} + + class Clients(dict): + @classmethod + def create(cls, region: str) -> Clients: + return cls(logs=logs, **{"lambda": Mock(), "cloudformation": Mock()}) + + class Invoker: + def __init__(self, **kwargs: object) -> None: + self.stack_name = kwargs["stack_name"] + + def invoke(self, **kwargs: object) -> SimpleNamespace: + sample = Path(str(kwargs["event_file_path"])).parents[1].name + return SimpleNamespace( + output=json.dumps({"DurableExecutionArn": f"arn:execution:{sample}"}) + ) + + calls: list[dict] = [] + fail_samples: set[str] = set() + raise_samples: set[str] = set() + + def validate(**kwargs: object) -> SimpleNamespace: + calls.append(kwargs) + event_dir = Path(str(kwargs["tmp_dir"])) + sample = event_dir.parent.name + event_file = event_dir / "10-1_event.json" + event_file.write_text('{"Input": "test"}') + kwargs["invoker"].invoke(event_file_path=str(event_file)) + kwargs["aws_clients"]["logs"].filter_log_events( + logGroupName="/aws/lambda/lifecycle", + filterPattern=sample, + startTime=1, + ) + if sample in raise_samples: + raise RuntimeError("CloudWatch request failed after the first page") + history_dir = Path(str(kwargs["output_dir"])) + history_dir.mkdir() + (history_dir / "10-1.json").write_text('{"ExecutionStatus": "SUCCEEDED"}') + failed = sample in fail_samples + return SimpleNamespace( + passed=not failed, + errors=["log records out of order"] if failed else [], + placeholders={"INPUT_1": "test"}, + ) + + monkeypatch.setattr( + stability, "_load_runner", lambda: (Clients, Invoker, validate, requirement) + ) + return SimpleNamespace( + calls=calls, + event=event, + fail_samples=fail_samples, + raise_samples=raise_samples, + requirement=requirement, + ) + + +def _arguments(output: Path, samples: int = 4) -> list[str]: + return [ + "--stack-name", + "existing-stack", + "--region", + "us-west-2", + "--samples", + str(samples), + "--workers", + "2", + "--output-dir", + str(output), + "--log-poll-timeout-seconds", + "60", + ] + + +def test_repeats_exact_upstream_requirement_with_isolated_artifacts( + tmp_path: Path, runner: SimpleNamespace +) -> None: + output = tmp_path / "results" + + assert stability.main(_arguments(output)) == 0 + + assert len(runner.calls) == 4 + assert len({call["tmp_dir"] for call in runner.calls}) == 4 + assert len({call["output_dir"] for call in runner.calls}) == 4 + for call in runner.calls: + assert call["function_name"] == "PluginInvocationLifecycle" + assert call["description_id"] == "10-1" + assert call["test_file"] == str(runner.requirement) + assert call["log_poll_timeout_seconds"] == 60 + summary = json.loads((output / "summary.json").read_text()) + assert (summary["samples"], summary["passed"], summary["failed"]) == (4, 4, 0) + assert len({result["execution_arn"] for result in summary["results"]}) == 4 + assert all( + (path / "history" / "10-1.json").exists() for path in output.glob("sample-*") + ) + + +def test_failure_keeps_original_cloudwatch_data_and_arn_and_returns_nonzero( + tmp_path: Path, runner: SimpleNamespace +) -> None: + output = tmp_path / "results" + runner.fail_samples.add("sample-0002") + + assert stability.main(_arguments(output)) == 1 + + summary = json.loads((output / "summary.json").read_text()) + assert (summary["passed"], summary["failed"]) == (3, 1) + failed = json.loads((output / "sample-0002" / "result.json").read_text()) + assert failed["errors"] == ["log records out of order"] + assert failed["execution_arn"] == "arn:execution:sample-0002" + assert json.loads((output / "sample-0002" / "invocation.json").read_text()) == { + "DurableExecutionArn": failed["execution_arn"] + } + page = json.loads((output / "sample-0002" / "cloudwatch-pages.jsonl").read_text()) + assert page["response"] == {"events": [runner.event], "nextToken": "next-page"} + assert page["request"]["filterPattern"] == "sample-0002" + + +def test_unexpected_sample_error_preserves_pages_and_remaining_samples( + tmp_path: Path, runner: SimpleNamespace +) -> None: + output = tmp_path / "results" + runner.raise_samples.add("sample-0001") + + assert stability.main(_arguments(output)) == 1 + + summary = json.loads((output / "summary.json").read_text()) + assert len(runner.calls) == 4 + assert summary["failed"] == 1 + failed = summary["results"][0] + assert failed["execution_arn"] == "arn:execution:sample-0001" + assert failed["errors"] == [ + "RuntimeError: CloudWatch request failed after the first page" + ] + assert (output / "sample-0001" / "cloudwatch-pages.jsonl").is_file() + + +def test_records_each_page_and_failed_request_without_changing_response( + tmp_path: Path, +) -> None: + page = {"events": [{"timestamp": 3, "ingestionTime": 4}], "nextToken": "next"} + client = Mock() + client.filter_log_events.side_effect = [page, RuntimeError("unavailable")] + output = tmp_path / "pages.jsonl" + recorder = stability.RecordingLogsClient(client, output) + + assert recorder.filter_log_events(logGroupName="logs") is page + with pytest.raises(RuntimeError, match="unavailable"): + recorder.filter_log_events(logGroupName="logs", nextToken="next") + + records = [json.loads(line) for line in output.read_text().splitlines()] + assert records[0]["response"] == page + assert records[1] == { + "request": {"logGroupName": "logs", "nextToken": "next"}, + "error": "RuntimeError: unavailable", + } + + +@pytest.mark.parametrize( + "extra", [["--samples", "0"], ["--workers", "0"], ["--workers", "17"]] +) +def test_rejects_invalid_sample_or_worker_counts( + tmp_path: Path, extra: list[str] +) -> None: + with pytest.raises(SystemExit) as caught: + stability.main([*_arguments(tmp_path / "results"), *extra]) + assert caught.value.code == 2 + + +def test_existing_results_are_not_overwritten( + tmp_path: Path, runner: SimpleNamespace +) -> None: + output = tmp_path / "results" + output.mkdir() + evidence = output / "summary.json" + evidence.write_text("original evidence") + + with pytest.raises(SystemExit) as caught: + stability.main(_arguments(output)) + + assert caught.value.code == 2 + assert not runner.calls + assert evidence.read_text() == "original evidence" From d98df785e286aaed01017e20b89e695abd2c60ce Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Tue, 20 Oct 2026 22:39:21 +0000 Subject: [PATCH 2/5] test: add lifecycle logging stability check --- .github/workflows/conformance-tests.yml | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/.github/workflows/conformance-tests.yml b/.github/workflows/conformance-tests.yml index b3d70c205..6d40e280e 100644 --- a/.github/workflows/conformance-tests.yml +++ b/.github/workflows/conformance-tests.yml @@ -20,6 +20,11 @@ on: required: false default: us-west-2 type: string + lifecycle_log_samples: + description: "Repeat plugin lifecycle log validation this many times (0 disables)" + required: false + default: "0" + type: string env: AWS_REGION: ${{ github.event.inputs.region || 'us-west-2' }} @@ -140,6 +145,7 @@ jobs: echo "SUITE_SLUG=$(echo '${{ matrix.suite }}' | tr '_' '-')" >> "$GITHUB_ENV" - name: Run conformance suite + id: conformance run: | # Stable per-suite stack name (no run id): every workflow run updates # the same stack, and --no-cleanup leaves it deployed afterwards so @@ -156,6 +162,17 @@ jobs: --report junit \ --report-file report-${{ matrix.suite }} + - name: Check lifecycle log stability + if: ${{ !cancelled() && matrix.suite == 'plugin' && github.event_name == 'workflow_dispatch' && inputs.lifecycle_log_samples != '0' && steps.conformance.outcome != 'skipped' }} + env: + LIFECYCLE_LOG_SAMPLES: ${{ inputs.lifecycle_log_samples }} + run: | + python scripts/check_lifecycle_log_stability.py \ + --stack-name conformance-tests-conf-py-plugin \ + --region "$AWS_REGION" \ + --samples "$LIFECYCLE_LOG_SAMPLES" \ + --output-dir lifecycle-log-stability + - name: Upload conformance report if: always() uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 @@ -164,4 +181,5 @@ jobs: path: | packages/aws-durable-execution-sdk-python-conformance-tests/report-${{ matrix.suite }}.xml packages/aws-durable-execution-sdk-python-conformance-tests/history-${{ matrix.suite }}/ + packages/aws-durable-execution-sdk-python-conformance-tests/lifecycle-log-stability/ if-no-files-found: warn From f66e90abe4f6ffee0804bda34bb985bf1cbf74bc Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Tue, 20 Oct 2026 22:41:10 +0000 Subject: [PATCH 3/5] fix: align plugin lifecycle logging transport --- .../plugin/plugin_invocation_lifecycle.py | 7 +- .../template_plugin.yaml | 3 + .../e2e/test_plugin_invocation_lifecycle.py | 89 +++++++++++++++++++ 3 files changed, 97 insertions(+), 2 deletions(-) create mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/tests/e2e/test_plugin_invocation_lifecycle.py diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_lifecycle.py b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_lifecycle.py index ddd226736..9bf94fd10 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_lifecycle.py +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/handlers/plugin/plugin_invocation_lifecycle.py @@ -6,7 +6,7 @@ running line via the SDK-provided step context logger (mirrors handler 1-7). """ -import json +import logging from typing import Any from aws_durable_execution_sdk_python.context import ( @@ -28,7 +28,10 @@ def _emit(record: dict[str, Any], execution_arn: str | None) -> None: # execution. Omit the field when the ARN is unset (never invent a value). if execution_arn: record = {"durableExecutionArn": execution_arn, **record} - print(json.dumps(record), flush=True) + # Use the same logging transport as the step context logger. Mixing stdout + # and Lambda's logging sink can invert their CloudWatch timestamps even + # though invocation-start runs synchronously before the handler. + logging.getLogger().info("Plugin invocation lifecycle", extra=record) class LifecyclePlugin(DurableInstrumentationPlugin): diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/template_plugin.yaml b/packages/aws-durable-execution-sdk-python-conformance-tests/template_plugin.yaml index 9f388d607..cbebfc10c 100644 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/template_plugin.yaml +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/template_plugin.yaml @@ -36,6 +36,9 @@ Resources: CodeUri: lambda-build/ Handler: plugin.plugin_invocation_lifecycle.handler Description: Plugin invocation lifecycle hooks (start and end) + LoggingConfig: + LogFormat: JSON + ApplicationLogLevel: INFO Role: Fn::GetAtt: - DurableFunctionRole diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/tests/e2e/test_plugin_invocation_lifecycle.py b/packages/aws-durable-execution-sdk-python-conformance-tests/tests/e2e/test_plugin_invocation_lifecycle.py new file mode 100644 index 000000000..cea8d69e5 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-conformance-tests/tests/e2e/test_plugin_invocation_lifecycle.py @@ -0,0 +1,89 @@ +"""Exercise lifecycle log ordering through the SDK and its local runner.""" + +from __future__ import annotations + +import json +import logging +import sys +from pathlib import Path + +import pytest +from aws_durable_execution_sdk_python.config import Duration +from aws_durable_execution_sdk_python.context import DurableContext +from aws_durable_execution_sdk_python.execution import ( + InvocationStatus, + durable_execution, +) +from aws_durable_execution_sdk_python_testing import DurableFunctionTestRunner + + +HANDLERS_DIR = Path(__file__).resolve().parents[2] / "handlers" +sys.path.insert(0, str(HANDLERS_DIR)) + +from plugin.plugin_invocation_lifecycle import LifecyclePlugin, greet, handler # noqa: E402 + + +def _lifecycle_records(caplog: pytest.LogCaptureFixture) -> list[logging.LogRecord]: + return [ + record + for record in caplog.records + if getattr(record, "plugin", None) == "CONFPLUGIN" + or record.getMessage().startswith("Greeting step running for:") + ] + + +def test_lifecycle_and_greeting_share_structured_logging( + caplog: pytest.LogCaptureFixture, capsys: pytest.CaptureFixture[str] +) -> None: + caplog.set_level(logging.INFO) + with DurableFunctionTestRunner(handler=handler) as runner: + result = runner.run(input=json.dumps("Ada"), execution_timeout=10) + + assert result.status is InvocationStatus.SUCCEEDED + assert json.loads(result.result) == "Hello, Ada!" + start, greeting, end = _lifecycle_records(caplog) + assert start.plugin == end.plugin == "CONFPLUGIN" + assert start.hook == "invocation-start" + assert start.first is True + assert greeting.getMessage() == "Greeting step running for: Ada" + assert end.hook == "invocation-end" + assert end.status == "SUCCEEDED" + assert start.durableExecutionArn + assert start.durableExecutionArn == greeting.executionArn + assert start.durableExecutionArn == end.durableExecutionArn + assert start.name == greeting.name == end.name + assert capsys.readouterr().out == "" + + +def test_replay_logs_hooks_without_repeating_the_greeting( + caplog: pytest.LogCaptureFixture, capsys: pytest.CaptureFixture[str] +) -> None: + @durable_execution(plugins=[LifecyclePlugin()]) + def handler_with_wait(event: str, context: DurableContext) -> str: + result = context.step(greet(event)) + context.wait(Duration.from_seconds(10)) + return result + + caplog.set_level(logging.INFO) + with DurableFunctionTestRunner(handler=handler_with_wait) as runner: + result = runner.run(input=json.dumps("Ada"), execution_timeout=30) + + assert result.status is InvocationStatus.SUCCEEDED + assert json.loads(result.result) == "Hello, Ada!" + records = _lifecycle_records(caplog) + assert [getattr(record, "hook", "greeting") for record in records] == [ + "invocation-start", + "greeting", + "invocation-end", + "invocation-start", + "invocation-end", + ] + assert records[0].first is True + assert records[2].status == "PENDING" + assert records[3].first is False + assert records[4].status == "SUCCEEDED" + plugin_arns = { + record.durableExecutionArn for record in records if hasattr(record, "hook") + } + assert plugin_arns == {records[1].executionArn} + assert capsys.readouterr().out == "" From c29f9e70c770b4a452effd569025df9d53cbdf9c Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Tue, 20 Oct 2026 22:44:14 +0000 Subject: [PATCH 4/5] test: enable lifecycle sampling on labeled PRs --- .github/workflows/conformance-tests.yml | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/.github/workflows/conformance-tests.yml b/.github/workflows/conformance-tests.yml index 6d40e280e..a513d9beb 100644 --- a/.github/workflows/conformance-tests.yml +++ b/.github/workflows/conformance-tests.yml @@ -9,6 +9,7 @@ on: branches: ["main"] pull_request: branches: ["main"] + types: [opened, synchronize, reopened, labeled] paths: - "packages/aws-durable-execution-sdk-python/**" - "packages/aws-durable-execution-sdk-python-conformance-tests/**" @@ -47,7 +48,10 @@ jobs: name: discover conformance suites # 1. Run for non-PR events, such as scheduled runs and manual invocations # 2. For PRs, run only same-repo branches from non-Dependabot actors; forked PRs or Dependabot should not access secrets. - if: github.event_name != 'pull_request' || (github.actor != 'dependabot[bot]' && github.event.pull_request.head.repo.full_name == github.repository) + if: >- + github.event_name != 'pull_request' || + (github.actor != 'dependabot[bot]' && github.event.pull_request.head.repo.full_name == github.repository && + (github.event.action != 'labeled' || github.event.label.name == 'test:plugin-log-stability')) # CodeBuild-hosted runner in the Python testing account (Lambda compute, # Amazon Linux python3.13 image). Fork PRs are already excluded by the `if:` # above, so the plain label is safe here. The label must match the CodeBuild @@ -163,9 +167,14 @@ jobs: --report-file report-${{ matrix.suite }} - name: Check lifecycle log stability - if: ${{ !cancelled() && matrix.suite == 'plugin' && github.event_name == 'workflow_dispatch' && inputs.lifecycle_log_samples != '0' && steps.conformance.outcome != 'skipped' }} + # A labeled same-repo PR uses the existing pull_request OIDC trust. Manual + # dispatch remains useful on main, where the test role also permits it. + if: >- + ${{ !cancelled() && matrix.suite == 'plugin' && steps.conformance.outcome != 'skipped' && + ((github.event_name == 'workflow_dispatch' && inputs.lifecycle_log_samples != '0') || + (github.event_name == 'pull_request' && contains(github.event.pull_request.labels.*.name, 'test:plugin-log-stability'))) }} env: - LIFECYCLE_LOG_SAMPLES: ${{ inputs.lifecycle_log_samples }} + LIFECYCLE_LOG_SAMPLES: ${{ inputs.lifecycle_log_samples || '40' }} run: | python scripts/check_lifecycle_log_stability.py \ --stack-name conformance-tests-conf-py-plugin \ From 91076a7a8afc7403244b58a9d8103179145f18a8 Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Tue, 20 Oct 2026 23:17:41 +0000 Subject: [PATCH 5/5] chore: remove lifecycle sampling experiment --- .github/workflows/conformance-tests.yml | 29 +-- .../scripts/check_lifecycle_log_stability.py | 183 --------------- .../test_check_lifecycle_log_stability.py | 215 ------------------ 3 files changed, 1 insertion(+), 426 deletions(-) delete mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/scripts/check_lifecycle_log_stability.py delete mode 100644 packages/aws-durable-execution-sdk-python-conformance-tests/tests/test_check_lifecycle_log_stability.py diff --git a/.github/workflows/conformance-tests.yml b/.github/workflows/conformance-tests.yml index a513d9beb..b3d70c205 100644 --- a/.github/workflows/conformance-tests.yml +++ b/.github/workflows/conformance-tests.yml @@ -9,7 +9,6 @@ on: branches: ["main"] pull_request: branches: ["main"] - types: [opened, synchronize, reopened, labeled] paths: - "packages/aws-durable-execution-sdk-python/**" - "packages/aws-durable-execution-sdk-python-conformance-tests/**" @@ -21,11 +20,6 @@ on: required: false default: us-west-2 type: string - lifecycle_log_samples: - description: "Repeat plugin lifecycle log validation this many times (0 disables)" - required: false - default: "0" - type: string env: AWS_REGION: ${{ github.event.inputs.region || 'us-west-2' }} @@ -48,10 +42,7 @@ jobs: name: discover conformance suites # 1. Run for non-PR events, such as scheduled runs and manual invocations # 2. For PRs, run only same-repo branches from non-Dependabot actors; forked PRs or Dependabot should not access secrets. - if: >- - github.event_name != 'pull_request' || - (github.actor != 'dependabot[bot]' && github.event.pull_request.head.repo.full_name == github.repository && - (github.event.action != 'labeled' || github.event.label.name == 'test:plugin-log-stability')) + if: github.event_name != 'pull_request' || (github.actor != 'dependabot[bot]' && github.event.pull_request.head.repo.full_name == github.repository) # CodeBuild-hosted runner in the Python testing account (Lambda compute, # Amazon Linux python3.13 image). Fork PRs are already excluded by the `if:` # above, so the plain label is safe here. The label must match the CodeBuild @@ -149,7 +140,6 @@ jobs: echo "SUITE_SLUG=$(echo '${{ matrix.suite }}' | tr '_' '-')" >> "$GITHUB_ENV" - name: Run conformance suite - id: conformance run: | # Stable per-suite stack name (no run id): every workflow run updates # the same stack, and --no-cleanup leaves it deployed afterwards so @@ -166,22 +156,6 @@ jobs: --report junit \ --report-file report-${{ matrix.suite }} - - name: Check lifecycle log stability - # A labeled same-repo PR uses the existing pull_request OIDC trust. Manual - # dispatch remains useful on main, where the test role also permits it. - if: >- - ${{ !cancelled() && matrix.suite == 'plugin' && steps.conformance.outcome != 'skipped' && - ((github.event_name == 'workflow_dispatch' && inputs.lifecycle_log_samples != '0') || - (github.event_name == 'pull_request' && contains(github.event.pull_request.labels.*.name, 'test:plugin-log-stability'))) }} - env: - LIFECYCLE_LOG_SAMPLES: ${{ inputs.lifecycle_log_samples || '40' }} - run: | - python scripts/check_lifecycle_log_stability.py \ - --stack-name conformance-tests-conf-py-plugin \ - --region "$AWS_REGION" \ - --samples "$LIFECYCLE_LOG_SAMPLES" \ - --output-dir lifecycle-log-stability - - name: Upload conformance report if: always() uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 @@ -190,5 +164,4 @@ jobs: path: | packages/aws-durable-execution-sdk-python-conformance-tests/report-${{ matrix.suite }}.xml packages/aws-durable-execution-sdk-python-conformance-tests/history-${{ matrix.suite }}/ - packages/aws-durable-execution-sdk-python-conformance-tests/lifecycle-log-stability/ if-no-files-found: warn diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/scripts/check_lifecycle_log_stability.py b/packages/aws-durable-execution-sdk-python-conformance-tests/scripts/check_lifecycle_log_stability.py deleted file mode 100644 index 90d101c91..000000000 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/scripts/check_lifecycle_log_stability.py +++ /dev/null @@ -1,183 +0,0 @@ -#!/usr/bin/env python3 -"""Repeat upstream requirement 10-1 against an already deployed plugin stack. - -Requires the installed aws-durable-execution-conformance-tests runner. Each -sample uses its unchanged history, result, and log assertions. Raw CloudWatch -pages and invocation responses are retained so ordering failures can be traced -to their original event timestamps and ingestion times. -""" - -from __future__ import annotations - -import argparse -import json -from concurrent.futures import ThreadPoolExecutor -from pathlib import Path -from typing import Any - - -def _load_runner() -> tuple[Any, Any, Any, Path]: - # Keep the upstream runner optional for the SDK's ordinary unit tests. - from aws_durable_execution_conformance_tests.clients import AwsClients - from aws_durable_execution_conformance_tests.config import TESTS_DIR - from aws_durable_execution_conformance_tests.sam import Invoker - from aws_durable_execution_conformance_tests.validate import validate_description - - return AwsClients, Invoker, validate_description, TESTS_DIR / "plugin" / "10-1.yaml" - - -def _write_json(path: Path, value: Any) -> None: - path.write_text(json.dumps(value, indent=2, default=str) + "\n", encoding="utf-8") - - -class RecordingLogsClient: - """Forward requests unchanged and persist every page before validation.""" - - def __init__(self, client: Any, path: Path) -> None: - self.client = client - self.path = path - - def filter_log_events(self, **kwargs: Any) -> dict[str, Any]: - record: dict[str, Any] = {"request": kwargs} - try: - response = self.client.filter_log_events(**kwargs) - record["response"] = response - return response - except Exception as exc: - record["error"] = f"{type(exc).__name__}: {exc}" - raise - finally: - with self.path.open("a", encoding="utf-8") as output: - output.write(json.dumps(record, default=str) + "\n") - - -class RecordingInvoker: - """Keep the execution ARN even when upstream returns a failed result.""" - - def __init__(self, invoker: Any, path: Path) -> None: - self.invoker = invoker - self.stack_name = invoker.stack_name - self.path = path - self.execution_arn: str | None = None - - def invoke(self, **kwargs: Any) -> Any: - result = self.invoker.invoke(**kwargs) - self.path.write_text(result.output, encoding="utf-8") - try: - response = json.loads(result.output) - except ValueError: - pass # Upstream reports malformed invocation responses itself. - else: - if isinstance(response, dict): - self.execution_arn = response.get("DurableExecutionArn") - return result - - -def _run_sample( - number: int, - args: argparse.Namespace, - clients: Any, - invoker: Any, - runner: tuple[Any, Any, Any, Path], -) -> dict[str, Any]: - clients_type, _, validate_description, requirement = runner - sample_dir = args.output_dir / f"sample-{number:04d}" - sample_dir.mkdir() - events_dir = sample_dir / "events" - events_dir.mkdir() - recording_invoker = RecordingInvoker(invoker, sample_dir / "invocation.json") - recording_clients = clients_type( - { - **dict(clients), - "logs": RecordingLogsClient( - clients["logs"], sample_dir / "cloudwatch-pages.jsonl" - ), - } - ) - result: dict[str, Any] = {"sample": number, "passed": False, "errors": []} - try: - validated = validate_description( - function_name="PluginInvocationLifecycle", - description_id="10-1", - test_file=str(requirement), - invoker=recording_invoker, - tmp_dir=str(events_dir), - region=args.region, - aws_clients=recording_clients, - output_dir=str(sample_dir / "history"), - log_poll_timeout_seconds=args.log_poll_timeout_seconds, - ) - result.update( - passed=validated.passed, - errors=validated.errors, - placeholders=validated.placeholders, - ) - except Exception as exc: - # Keep sampling after a failed request, without losing earlier pages. - result["errors"] = [f"{type(exc).__name__}: {exc}"] - result["execution_arn"] = recording_invoker.execution_arn - _write_json(sample_dir / "result.json", result) - print(f"Sample {number}: {'PASSED' if result['passed'] else 'FAILED'}", flush=True) - return result - - -def _positive_int(value: str) -> int: - number = int(value) - if number < 1: - raise argparse.ArgumentTypeError("must be at least 1") - return number - - -def main(argv: list[str] | None = None) -> int: - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--stack-name", required=True) - parser.add_argument("--region", default="us-west-2") - parser.add_argument("--samples", type=_positive_int, default=40) - parser.add_argument("--workers", type=int, choices=range(1, 17), default=4) - parser.add_argument("--output-dir", type=Path, required=True) - parser.add_argument( - "--log-poll-timeout-seconds", - type=_positive_int, - default=120, - help="Maximum upstream polling budget per sample; passing logs normally " - "finish after the runner's 20-second observation window.", - ) - args = parser.parse_args(argv) - runner = _load_runner() - clients_type, invoker_type, _, requirement = runner - if not requirement.is_file(): - parser.error(f"Upstream requirement not found: {requirement}") - if args.output_dir.exists(): - parser.error("--output-dir must be new to preserve previous diagnostic results") - args.output_dir.mkdir(parents=True) - clients = clients_type.create(args.region) - invoker = invoker_type( - stack_name=args.stack_name, - region=args.region, - lambda_client=clients["lambda"], - cfn_client=clients["cloudformation"], - ) - with ThreadPoolExecutor(max_workers=min(args.workers, args.samples)) as executor: - results = list( - executor.map( - lambda number: _run_sample(number, args, clients, invoker, runner), - range(1, args.samples + 1), - ) - ) - passed = sum(result["passed"] for result in results) - summary = { - "stack_name": args.stack_name, - "region": args.region, - "requirement": "10-1", - "samples": args.samples, - "passed": passed, - "failed": args.samples - passed, - "results": results, - } - _write_json(args.output_dir / "summary.json", summary) - print(f"Lifecycle log checks: {passed}/{args.samples} passed", flush=True) - return 0 if passed == args.samples else 1 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/packages/aws-durable-execution-sdk-python-conformance-tests/tests/test_check_lifecycle_log_stability.py b/packages/aws-durable-execution-sdk-python-conformance-tests/tests/test_check_lifecycle_log_stability.py deleted file mode 100644 index 9b3140f7c..000000000 --- a/packages/aws-durable-execution-sdk-python-conformance-tests/tests/test_check_lifecycle_log_stability.py +++ /dev/null @@ -1,215 +0,0 @@ -"""Boundary tests for repeated upstream lifecycle validation and its artifacts.""" - -from __future__ import annotations - -import json -import sys -from pathlib import Path -from types import SimpleNamespace -from unittest.mock import Mock - -import pytest - - -SCRIPTS_DIR = Path(__file__).resolve().parents[1] / "scripts" -if str(SCRIPTS_DIR) not in sys.path: - sys.path.insert(0, str(SCRIPTS_DIR)) - -import check_lifecycle_log_stability as stability # noqa: E402 - - -@pytest.fixture -def runner(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> SimpleNamespace: - requirement = tmp_path / "10-1.yaml" - requirement.write_text("ExpectedLogs: []\n") - event = { - "eventId": "original-event-id", - "logStreamName": "original-stream", - "timestamp": 10, - "ingestionTime": 20, - "message": '{"hook": "invocation-start"}', - } - logs = Mock() - logs.filter_log_events.return_value = {"events": [event], "nextToken": "next-page"} - - class Clients(dict): - @classmethod - def create(cls, region: str) -> Clients: - return cls(logs=logs, **{"lambda": Mock(), "cloudformation": Mock()}) - - class Invoker: - def __init__(self, **kwargs: object) -> None: - self.stack_name = kwargs["stack_name"] - - def invoke(self, **kwargs: object) -> SimpleNamespace: - sample = Path(str(kwargs["event_file_path"])).parents[1].name - return SimpleNamespace( - output=json.dumps({"DurableExecutionArn": f"arn:execution:{sample}"}) - ) - - calls: list[dict] = [] - fail_samples: set[str] = set() - raise_samples: set[str] = set() - - def validate(**kwargs: object) -> SimpleNamespace: - calls.append(kwargs) - event_dir = Path(str(kwargs["tmp_dir"])) - sample = event_dir.parent.name - event_file = event_dir / "10-1_event.json" - event_file.write_text('{"Input": "test"}') - kwargs["invoker"].invoke(event_file_path=str(event_file)) - kwargs["aws_clients"]["logs"].filter_log_events( - logGroupName="/aws/lambda/lifecycle", - filterPattern=sample, - startTime=1, - ) - if sample in raise_samples: - raise RuntimeError("CloudWatch request failed after the first page") - history_dir = Path(str(kwargs["output_dir"])) - history_dir.mkdir() - (history_dir / "10-1.json").write_text('{"ExecutionStatus": "SUCCEEDED"}') - failed = sample in fail_samples - return SimpleNamespace( - passed=not failed, - errors=["log records out of order"] if failed else [], - placeholders={"INPUT_1": "test"}, - ) - - monkeypatch.setattr( - stability, "_load_runner", lambda: (Clients, Invoker, validate, requirement) - ) - return SimpleNamespace( - calls=calls, - event=event, - fail_samples=fail_samples, - raise_samples=raise_samples, - requirement=requirement, - ) - - -def _arguments(output: Path, samples: int = 4) -> list[str]: - return [ - "--stack-name", - "existing-stack", - "--region", - "us-west-2", - "--samples", - str(samples), - "--workers", - "2", - "--output-dir", - str(output), - "--log-poll-timeout-seconds", - "60", - ] - - -def test_repeats_exact_upstream_requirement_with_isolated_artifacts( - tmp_path: Path, runner: SimpleNamespace -) -> None: - output = tmp_path / "results" - - assert stability.main(_arguments(output)) == 0 - - assert len(runner.calls) == 4 - assert len({call["tmp_dir"] for call in runner.calls}) == 4 - assert len({call["output_dir"] for call in runner.calls}) == 4 - for call in runner.calls: - assert call["function_name"] == "PluginInvocationLifecycle" - assert call["description_id"] == "10-1" - assert call["test_file"] == str(runner.requirement) - assert call["log_poll_timeout_seconds"] == 60 - summary = json.loads((output / "summary.json").read_text()) - assert (summary["samples"], summary["passed"], summary["failed"]) == (4, 4, 0) - assert len({result["execution_arn"] for result in summary["results"]}) == 4 - assert all( - (path / "history" / "10-1.json").exists() for path in output.glob("sample-*") - ) - - -def test_failure_keeps_original_cloudwatch_data_and_arn_and_returns_nonzero( - tmp_path: Path, runner: SimpleNamespace -) -> None: - output = tmp_path / "results" - runner.fail_samples.add("sample-0002") - - assert stability.main(_arguments(output)) == 1 - - summary = json.loads((output / "summary.json").read_text()) - assert (summary["passed"], summary["failed"]) == (3, 1) - failed = json.loads((output / "sample-0002" / "result.json").read_text()) - assert failed["errors"] == ["log records out of order"] - assert failed["execution_arn"] == "arn:execution:sample-0002" - assert json.loads((output / "sample-0002" / "invocation.json").read_text()) == { - "DurableExecutionArn": failed["execution_arn"] - } - page = json.loads((output / "sample-0002" / "cloudwatch-pages.jsonl").read_text()) - assert page["response"] == {"events": [runner.event], "nextToken": "next-page"} - assert page["request"]["filterPattern"] == "sample-0002" - - -def test_unexpected_sample_error_preserves_pages_and_remaining_samples( - tmp_path: Path, runner: SimpleNamespace -) -> None: - output = tmp_path / "results" - runner.raise_samples.add("sample-0001") - - assert stability.main(_arguments(output)) == 1 - - summary = json.loads((output / "summary.json").read_text()) - assert len(runner.calls) == 4 - assert summary["failed"] == 1 - failed = summary["results"][0] - assert failed["execution_arn"] == "arn:execution:sample-0001" - assert failed["errors"] == [ - "RuntimeError: CloudWatch request failed after the first page" - ] - assert (output / "sample-0001" / "cloudwatch-pages.jsonl").is_file() - - -def test_records_each_page_and_failed_request_without_changing_response( - tmp_path: Path, -) -> None: - page = {"events": [{"timestamp": 3, "ingestionTime": 4}], "nextToken": "next"} - client = Mock() - client.filter_log_events.side_effect = [page, RuntimeError("unavailable")] - output = tmp_path / "pages.jsonl" - recorder = stability.RecordingLogsClient(client, output) - - assert recorder.filter_log_events(logGroupName="logs") is page - with pytest.raises(RuntimeError, match="unavailable"): - recorder.filter_log_events(logGroupName="logs", nextToken="next") - - records = [json.loads(line) for line in output.read_text().splitlines()] - assert records[0]["response"] == page - assert records[1] == { - "request": {"logGroupName": "logs", "nextToken": "next"}, - "error": "RuntimeError: unavailable", - } - - -@pytest.mark.parametrize( - "extra", [["--samples", "0"], ["--workers", "0"], ["--workers", "17"]] -) -def test_rejects_invalid_sample_or_worker_counts( - tmp_path: Path, extra: list[str] -) -> None: - with pytest.raises(SystemExit) as caught: - stability.main([*_arguments(tmp_path / "results"), *extra]) - assert caught.value.code == 2 - - -def test_existing_results_are_not_overwritten( - tmp_path: Path, runner: SimpleNamespace -) -> None: - output = tmp_path / "results" - output.mkdir() - evidence = output / "summary.json" - evidence.write_text("original evidence") - - with pytest.raises(SystemExit) as caught: - stability.main(_arguments(output)) - - assert caught.value.code == 2 - assert not runner.calls - assert evidence.read_text() == "original evidence"