diff --git a/.github/workflows/cloud-tests.yml b/.github/workflows/cloud-tests.yml index 80b5051ed..91e9f0a27 100644 --- a/.github/workflows/cloud-tests.yml +++ b/.github/workflows/cloud-tests.yml @@ -102,6 +102,19 @@ jobs: hatch run examples:build hatch run examples:generate-sam-template + - name: Ensure MicroVM image + # The worker runs in the MicroVM, not in the function. So one image + # serves every function runtime. Only the Python 3.13 job builds the + # image, deploys the MicroVM examples, and runs their tests. The other + # jobs pass no image, so their stacks leave the MicroVM examples out. + if: matrix.python-version == '3.13' + env: + TEST_ACCOUNT_ID: ${{ secrets.TEST_ACCOUNT_ID }} + # The script warns when the functions use a non-default endpoint, + # because the worker sends its callbacks to the default endpoint. + LAMBDA_ENDPOINT: ${{ github.event.inputs.lambda_endpoint || vars.TEST_LAMBDA_ENDPOINT }} + run: hatch run examples:ensure-microvm-image --output "$RUNNER_TEMP/microvm-image.json" + - name: Deploy all examples with SAM id: deploy env: @@ -111,6 +124,7 @@ jobs: STACK_NAME: ${{ matrix.python-prefix }}-python-examples FUNCTION_NAME_PREFIX: ${{ matrix.python-prefix }}- GH_TOKEN: ${{ github.token }} + MICROVM_EXAMPLES: ${{ matrix.python-version == '3.13' }} run: | echo "Building SAM application" sam build --template-file packages/aws-durable-execution-sdk-python-examples/template.generated.json @@ -154,9 +168,25 @@ jobs: echo "Using custom Lambda endpoint override" fi + # The MicroVM examples deploy only with an image. Without one, the + # template leaves them out, and the name map omits their tests. + NAME_MAP_ARGS=() + if [ "$MICROVM_EXAMPLES" = "true" ]; then + MICROVM_IMAGE_FILE="$RUNNER_TEMP/microvm-image.json" + MICROVM_IMAGE_ARN=$(python3 -c 'import json, sys; print(json.load(open(sys.argv[1]))["imageArn"])' "$MICROVM_IMAGE_FILE") + MICROVM_EXECUTION_ROLE_ARN=$(python3 -c 'import json, sys; print(json.load(open(sys.argv[1]))["executionRoleArn"])' "$MICROVM_IMAGE_FILE") + PARAMETER_OVERRIDES+=( + MicrovmImageArn="$MICROVM_IMAGE_ARN" + MicrovmExecutionRoleArn="$MICROVM_EXECUTION_ROLE_ARN" + ) + NAME_MAP_ARGS+=(--include-microvm) + echo "Deploying the MicroVM examples with image $MICROVM_IMAGE_ARN" + fi + PYTEST_FUNCTION_NAME_MAP=$( python packages/aws-durable-execution-sdk-python-examples/scripts/generate_function_name_map.py \ - --function-name-prefix "$FUNCTION_NAME_PREFIX" + --function-name-prefix "$FUNCTION_NAME_PREFIX" \ + "${NAME_MAP_ARGS[@]}" ) # TODO: Remove this one-time cleanup after existing CI stacks adopt the diff --git a/.gitignore b/.gitignore index c6b0d1df5..13ccb7ad3 100644 --- a/.gitignore +++ b/.gitignore @@ -34,6 +34,7 @@ dist/ **/build/ **/*.zip .aws-sam/ +.microvm-image-build/ **/template.generated.json .env diff --git a/packages/aws-durable-execution-sdk-python-examples/examples-catalog.json b/packages/aws-durable-execution-sdk-python-examples/examples-catalog.json index 219ccb96a..454522a11 100644 --- a/packages/aws-durable-execution-sdk-python-examples/examples-catalog.json +++ b/packages/aws-durable-execution-sdk-python-examples/examples-catalog.json @@ -796,6 +796,31 @@ "ExecutionTimeout": 300 }, "path": "./src/error_handling/catch_typed_step_error.py" + }, + { + "name": "MicroVM Run Job", + "description": "Runs one job in a Lambda MicroVM and returns the result that the MicroVM reports", + "handler": "microvm_run_job.handler", + "integration": true, + "usesMicrovm": true, + "durableConfig": { + "RetentionPeriodInDays": 7, + "ExecutionTimeout": 900 + }, + "path": "./src/microvm/microvm_run_job.py" + }, + { + "name": "MicroVM Session", + "description": "Runs two jobs in one Lambda MicroVM, with a durable wait between them", + "handler": "microvm_session.handler", + "integration": true, + "usesMicrovm": true, + "timeout": 120, + "durableConfig": { + "RetentionPeriodInDays": 7, + "ExecutionTimeout": 900 + }, + "path": "./src/microvm/microvm_session.py" } ] } diff --git a/packages/aws-durable-execution-sdk-python-examples/microvm-image/Dockerfile b/packages/aws-durable-execution-sdk-python-examples/microvm-image/Dockerfile new file mode 100644 index 000000000..6d279c0a3 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-examples/microvm-image/Dockerfile @@ -0,0 +1,13 @@ +FROM public.ecr.aws/docker/library/python:3.13-alpine +WORKDIR /app +# scripts/ensure_microvm_image.py puts the worker wheel, built from this +# repo, in wheels/. pip installs boto3 from PyPI, at the versions that +# constraints.txt pins. +COPY constraints.txt wheels/ /tmp/build/ +RUN pip install --no-cache-dir --root-user-action=ignore \ + --constraint /tmp/build/constraints.txt /tmp/build/*.whl \ + && rm -rf /tmp/build +COPY app.py . +EXPOSE 8080 +# Lambda snapshots the running process after the ready hook. +CMD ["python", "-u", "app.py"] diff --git a/packages/aws-durable-execution-sdk-python-examples/microvm-image/app.py b/packages/aws-durable-execution-sdk-python-examples/microvm-image/app.py new file mode 100644 index 000000000..4b1ccbf04 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-examples/microvm-image/app.py @@ -0,0 +1,78 @@ +"""The application in the MicroVM image that the MicroVM examples run. + +Lambda starts this process when it builds the image, and snapshots it. A +MicroVM launched from the image resumes the snapshot. The worker then receives +each job, runs it, and reports the result to the durable function's callback. + +scripts/ensure_microvm_image.py builds the image from this file, the +Dockerfile, and the worker wheel. The cloud tests pass the image ARN to the +example functions. +""" + +import hashlib +import logging +from pathlib import Path +from typing import Any + +from aws_durable_execution_sdk_python_microvm_worker import ( + MicrovmJobContext, + start_microvm_worker, +) + + +# Only the MicroVM's own jobs write here. A session's jobs run in one MicroVM. +# So a later job of the session reads what an earlier job wrote. +SESSION_DIR = Path("/tmp/session") + + +def count_words(job_input: dict[str, Any], context: MicrovmJobContext) -> dict: + """Serve the run-job example. A job that names no route runs this handler.""" + text: str = job_input["text"] + return { + "wordCount": len(text.split()), + "sha256": hashlib.sha256(text.encode()).hexdigest(), + "microvmId": context.microvm_id, + } + + +def _session_file(key: str) -> Path: + # A key with a directory part could name a file outside SESSION_DIR. So + # the handlers accept only a plain file name. + if not key or Path(key).name != key or key in {".", ".."}: + msg = f"invalid key: {key!r}" + raise ValueError(msg) + return SESSION_DIR / key + + +def write_value(job_input: dict[str, Any], context: MicrovmJobContext) -> dict: + """Serve the session example's first job. It writes a file.""" + path = _session_file(job_input["key"]) + SESSION_DIR.mkdir(parents=True, exist_ok=True) + path.write_text(job_input["value"], encoding="utf-8") + return {"microvmId": context.microvm_id} + + +def read_value(job_input: dict[str, Any], context: MicrovmJobContext) -> dict: + """Serve the session example's second job. It reads the file.""" + path = _session_file(job_input["key"]) + return { + "microvmId": context.microvm_id, + "value": path.read_text(encoding="utf-8"), + } + + +def main() -> None: + # The worker logs to the standard library logger. INFO makes its lines + # reach stdout. + logging.basicConfig(level=logging.INFO) + worker = start_microvm_worker( + handler=count_words, + routes={"/write": write_value, "/read": read_value}, + ) + # start_microvm_worker serves in a daemon thread and returns. So the + # process waits here. + worker.wait() + + +if __name__ == "__main__": + main() diff --git a/packages/aws-durable-execution-sdk-python-examples/microvm-image/constraints.txt b/packages/aws-durable-execution-sdk-python-examples/microvm-image/constraints.txt new file mode 100644 index 000000000..097387eb2 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-examples/microvm-image/constraints.txt @@ -0,0 +1,15 @@ +# The image installs the worker wheel and its dependencies from PyPI. +# scripts/ensure_microvm_image.py hashes this file into the image name. +# These pins fix the installed versions. So an image with a given name always +# holds the same dependency versions. A change to a pin builds a new image. +# +# The worker needs boto3 1.43.35 or later, the first release with the +# lambda-microvms client. These are the versions that the repo's examples +# environment resolved on 2026-10-22. +boto3==1.43.118 +botocore==1.43.118 +jmespath==1.1.0 +python-dateutil==2.9.0.post0 +s3transfer==0.19.3 +six==1.17.0 +urllib3==2.8.0 diff --git a/packages/aws-durable-execution-sdk-python-examples/scripts/build_examples.py b/packages/aws-durable-execution-sdk-python-examples/scripts/build_examples.py index bbdd26c4b..b30450a63 100644 --- a/packages/aws-durable-execution-sdk-python-examples/scripts/build_examples.py +++ b/packages/aws-durable-execution-sdk-python-examples/scripts/build_examples.py @@ -20,6 +20,9 @@ def build_examples() -> None: runtime_packages = [ packages_dir / "aws-durable-execution-sdk-python", + # The MicroVM examples import the extras. The examples package does + # not depend on the extras, so pip installs the extras only from here. + packages_dir / "aws-durable-execution-sdk-python-extras", packages_dir / "aws-durable-execution-sdk-python-otel", packages_dir / "aws-durable-execution-sdk-python-testing", ] diff --git a/packages/aws-durable-execution-sdk-python-examples/scripts/ensure_microvm_image.py b/packages/aws-durable-execution-sdk-python-examples/scripts/ensure_microvm_image.py new file mode 100644 index 000000000..dc9cfc601 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-examples/scripts/ensure_microvm_image.py @@ -0,0 +1,565 @@ +#!/usr/bin/env python3 +"""Build the MicroVM image that the MicroVM examples run, or reuse it. + +The script builds the worker wheel from this repo. It zips the wheel with +microvm-image/app.py, the Dockerfile, and constraints.txt. It uploads the zip +and calls CreateMicrovmImage. + +The image name ends with a hash of the inputs: the worker's source and +pyproject.toml, app.py, the Dockerfile, constraints.txt, and the +CreateMicrovmImage settings. So a run with unchanged inputs finds the image +and skips the build. A change to any input builds a new image. + +It writes the image ARN and the MicroVM execution role ARN as JSON to the file +named by --output. The cloud tests pass both to the SAM template, which sets +them as environment variables of the MicroVM examples. + +The resource names come from the language testing stack in the test account. +That stack names them after the account and Region. Environment variables +override them for another account: + MICROVM_ARTIFACT_BUCKET the bucket for the image code zip + MICROVM_IMAGE_BUILD_ROLE_ARN the role that the image build assumes + MICROVM_EXECUTION_ROLE_ARN the role that a running MicroVM assumes + +Usage: + TEST_ACCOUNT_ID=123456789012 AWS_REGION=us-west-2 \\ + hatch run examples:ensure-microvm-image --output image.json + +This is a port of ensure-microvm-image.ts in the JavaScript SDK examples. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import shutil +import subprocess +import sys +import time +import zipfile +from collections.abc import Callable, Iterable +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Protocol + +import boto3 +from botocore.exceptions import ClientError + + +EXAMPLES_DIR = Path(__file__).resolve().parent.parent +IMAGE_DIR = EXAMPLES_DIR / "microvm-image" +OUT_DIR = EXAMPLES_DIR / ".microvm-image-build" +WORKER_DIR = EXAMPLES_DIR.parent / "aws-durable-execution-sdk-python-microvm-worker" +IMAGE_FILES = ("Dockerfile", "app.py", "constraints.txt") +IMAGE_NAME_PREFIX = "sdk-python-examples" +HASH_LENGTH = 12 +POLL_INTERVAL_SECONDS = 10.0 +BUILD_TIMEOUT_SECONDS = 20 * 60.0 + +# GetMicrovmImage reports an image as missing for a short time after +# CreateMicrovmImage returns. A JavaScript test run saw this for one 10-second +# poll. A create in that time would fail on the duplicate name. So the script +# waits this long before it creates again. +CREATE_VISIBILITY_SECONDS = 60.0 + +# The worker answers the ready hook at image build and the run hook at each +# launch. The service requires the ready hook whenever a lifecycle hook is +# enabled. The resume hook lets a session's worker accept jobs again after it +# suspended its own MicroVM. On the suspend hook the worker refuses new jobs, +# and on the terminate hook it fails the callbacks of running jobs. A +# terminate hook gets 10 seconds, because the worker reports before it +# answers. +IMAGE_HOOKS: dict[str, Any] = { + "port": 8080, + "microvmHooks": { + "run": "ENABLED", + "runTimeoutInSeconds": 10, + "resume": "ENABLED", + "resumeTimeoutInSeconds": 10, + "suspend": "ENABLED", + "suspendTimeoutInSeconds": 10, + "terminate": "ENABLED", + "terminateTimeoutInSeconds": 10, + }, + "microvmImageHooks": {"ready": "ENABLED", "readyTimeoutInSeconds": 60}, +} + + +class ImageService(Protocol): + """The image operations that :func:`ensure_image` needs.""" + + def get(self) -> dict[str, Any] | None: + """Return the GetMicrovmImage response, or None for a missing image.""" + + def create(self) -> None: + """Upload the code and call CreateMicrovmImage.""" + + def delete(self) -> None: + """Call DeleteMicrovmImage.""" + + +def ensure_image( + service: ImageService, + name: str, + *, + timeout_seconds: float = BUILD_TIMEOUT_SECONDS, + poll_interval_seconds: float = POLL_INTERVAL_SECONDS, + now: Callable[[], float] = time.monotonic, + sleep: Callable[[float], None] = time.sleep, + log: Callable[[str], None] = print, +) -> str: + """Wait until the image has an active version, and create it when needed. + + Returns the active version. + + It acts on each state that GetMicrovmImage reports: + + - CREATED or UPDATED with an active version: the image is ready. + - Missing or DELETED: it creates the image. + - CREATING, UPDATING, or DELETING: it waits. + - CREATE_FAILED: it deletes the image, then creates it again. It does this + once per run, because a second failure most likely has the same cause. + - DELETE_FAILED, UPDATE_FAILED, or another state: it raises. Only an + administrator of the account can clear these states. + + The image name carries a hash of the inputs. So a failed build fails every + later run with the same inputs, until the failed image is deleted. The + delete and the recreate let a run recover from a failure that does not + repeat, such as a timeout of the ready hook, without an administrator. + """ + log_group = f"/aws/lambda-microvms/{name}" + deadline = now() + timeout_seconds + deleted_failed_image = False + created_after_delete = False + creates = 0 + last_create_at: float | None = None + first_poll = True + + while True: + image = service.get() + state = (image or {}).get("state") or "NOT_FOUND" + version = (image or {}).get("latestActiveImageVersion") + + if state in {"CREATED", "UPDATED"}: + if version: + log( + f"Reusing MicroVM image {name}, version {version}" + if first_poll + else f"MicroVM image {name} is {state}, version {version}" + ) + return version + elif state in {"NOT_FOUND", "DELETED"}: + create_is_recent = ( + last_create_at is not None + and now() - last_create_at < CREATE_VISIBILITY_SECONDS + ) + if not create_is_recent: + # One create at the start and one after a failed build. + if creates >= 2: + msg = ( + f"MicroVM image {name} is still {state} after " + f"{creates} CreateMicrovmImage calls" + ) + raise RuntimeError(msg) + log(f"Creating MicroVM image {name}") + service.create() + creates += 1 + last_create_at = now() + created_after_delete = deleted_failed_image + elif state in {"CREATING", "UPDATING", "DELETING"}: + pass + elif state == "CREATE_FAILED": + # GetMicrovmImage may still report the failed image right after + # DeleteMicrovmImage returns. So the state counts as a second + # failure only after the rebuild started. + if deleted_failed_image and created_after_delete: + msg = ( + f"MicroVM image {name} failed to build again after a " + f"rebuild. Read the build logs in the {log_group} log group." + ) + raise RuntimeError(msg) + if not deleted_failed_image: + log( + f"MicroVM image {name} is CREATE_FAILED. Deleting it to " + "build it again. The build logs are in the " + f"{log_group} log group." + ) + service.delete() + deleted_failed_image = True + last_create_at = None + else: + msg = ( + f"MicroVM image {name} is {state}. The script does not clear " + "this state. An administrator of the test account must delete " + f"the image. Read the logs in the {log_group} log group." + ) + raise RuntimeError(msg) + + first_poll = False + if now() > deadline: + msg = ( + f"MicroVM image {name} is still {state} after " + f"{timeout_seconds / 60:g} minutes" + ) + raise RuntimeError(msg) + log(f"Waiting for MicroVM image {name}: {state}") + sleep(poll_interval_seconds) + + +def _error_code(error: ClientError) -> str: + return error.response.get("Error", {}).get("Code", "") + + +def _error_message(error: ClientError) -> str: + return error.response.get("Error", {}).get("Message", "") + + +@dataclass +class BotoImageService: + """The :class:`ImageService` that calls S3 and the MicroVMs API. + + ``microvms`` is a boto3 ``lambda-microvms`` client, and ``s3`` a boto3 S3 + client. Tests pass fakes with the same methods. + """ + + microvms: Any + s3: Any + name: str + image_arn: str + bucket: str + key: str + zip_bytes: bytes + settings: dict[str, Any] + log: Callable[[str], None] = print + + def get(self) -> dict[str, Any] | None: + try: + return self.microvms.get_microvm_image(imageIdentifier=self.image_arn) + except ClientError as error: + if _error_code(error) == "ResourceNotFoundException": + return None + raise + + def create(self) -> None: + try: + self.s3.put_object(Bucket=self.bucket, Key=self.key, Body=self.zip_bytes) + except ClientError as error: + if _error_code(error) == "NoSuchBucket": + msg = ( + f"The bucket {self.bucket} does not exist. The MicroVM " + "examples need the MicroVM resources of the language " + "testing stack. For another account, set " + "MICROVM_ARTIFACT_BUCKET, MICROVM_IMAGE_BUILD_ROLE_ARN, " + "and MICROVM_EXECUTION_ROLE_ARN." + ) + raise RuntimeError(msg) from error + raise + try: + self.microvms.create_microvm_image( + **create_image_request(self.name, self.bucket, self.key, self.settings) + ) + except Exception as error: + # Two runs can build the same image at the same time. CI runs the + # build only in the Python 3.13 job. All runs of that job share + # one serialized concurrency group, cloud-tests-Py313, with + # queue: max. So two CI runs never build at the same time. So + # this case covers a manual run that overlaps a CI run. + # + # The service rejects the second create with a ValidationException, + # because the name exists. The image is then the one this run + # needs, because its name carries the hash of the inputs. So the + # run waits for it. + # + # GetMicrovmImage can still report the new image as missing. So + # the error message is the first signal, and the image state the + # second. + name_exists = ( + isinstance(error, ClientError) + and _error_code(error) == "ValidationException" + and "already exists" in _error_message(error) + ) + if name_exists or self.get() is not None: + self.log(f"MicroVM image {self.name} was created by another run") + return + raise + + def delete(self) -> None: + try: + self.microvms.delete_microvm_image(imageIdentifier=self.image_arn) + except Exception: + # Two runs can see the same failed image and delete it at the + # same time. The later delete then fails, because the image is + # gone or is being deleted. The poll loop handles both states. So + # the run continues when the image is missing or DELETING after + # the error. + # + # CI runs the build only in the Python 3.13 job. All runs of that + # job share one serialized concurrency group, cloud-tests-Py313, + # with queue: max. So this case covers a manual run that overlaps + # a CI run. + image = self.get() + if image is None or image.get("state") == "DELETING": + self.log(f"MicroVM image {self.name} was deleted by another run") + return + raise + + +def create_image_request( + name: str, bucket: str, key: str, settings: dict[str, Any] +) -> dict[str, Any]: + """Return the CreateMicrovmImage parameters, in boto3 names.""" + return { + "name": name, + "description": "The MicroVM image that the durable execution SDK examples run", + "codeArtifact": {"uri": f"s3://{bucket}/{key}"}, + **settings, + } + + +def image_settings( + region: str, account: str, build_role_arn: str | None = None +) -> dict[str, Any]: + """Return the CreateMicrovmImage settings other than the name and code.""" + return { + "baseImageArn": f"arn:aws:lambda:{region}:aws:microvm-image:al2023-1", + "buildRoleArn": build_role_arn + or f"arn:aws:iam::{account}:role/microvm-image-build", + # The build pulls the Python base image from Amazon ECR Public, and + # boto3 from PyPI. + "egressNetworkConnectors": [ + f"arn:aws:lambda:{region}:aws:network-connector:" + "aws-network-connector:INTERNET_EGRESS" + ], + "hooks": IMAGE_HOOKS, + } + + +def _is_source_file(path: Path) -> bool: + # Local test runs leave bytecode next to the source. CI runs do not. So + # the hash skips it, and both compute the same name. + return ( + path.is_file() + and "__pycache__" not in path.parts + and path.suffix not in {".pyc", ".pyo"} + and path.name != ".DS_Store" + ) + + +def hash_inputs( + worker_dir: Path, image_dir: Path, settings: dict[str, Any] +) -> list[tuple[str, bytes]]: + """Return the (name, bytes) pairs that the content hash covers. + + The pairs cover the worker's src/ tree and pyproject.toml, the image + files, and the settings as JSON with sorted keys. They do not cover the + wheel. Two builds of the same source can produce wheels that differ byte + for byte. Two runners would then compute different names, and reuse would + fail. + + The Dockerfile names a floating python:3.13-alpine tag, and the hash does + not cover the image behind the tag. Lambda snapshots the image at build + time. So a tag change takes effect at the next build. + """ + inputs: list[tuple[str, bytes]] = [] + src_dir = worker_dir / "src" + for path in sorted(src_dir.rglob("*")): + if _is_source_file(path): + relative = path.relative_to(worker_dir).as_posix() + inputs.append((f"worker/{relative}", path.read_bytes())) + inputs.append( + ("worker/pyproject.toml", (worker_dir / "pyproject.toml").read_bytes()) + ) + inputs.extend( + (f"image/{file_name}", (image_dir / file_name).read_bytes()) + for file_name in IMAGE_FILES + ) + settings_json = json.dumps(settings, sort_keys=True, separators=(",", ":")) + inputs.append(("settings.json", settings_json.encode())) + return inputs + + +def content_hash(inputs: Iterable[tuple[str, bytes]]) -> str: + """Return the first 12 hex characters of a SHA-256 over the inputs. + + Each entry adds its name, its length, and its bytes. The length separates + one entry from the next. So no two different input sets give the same + byte stream. + """ + digest = hashlib.sha256() + for name, data in sorted(inputs): + digest.update(name.encode()) + digest.update(b"\0") + digest.update(str(len(data)).encode()) + digest.update(b"\0") + digest.update(data) + return digest.hexdigest()[:HASH_LENGTH] + + +def build_worker_wheel(worker_dir: Path, wheel_dir: Path) -> Path: + """Build the worker wheel from the repo, without its dependencies.""" + subprocess.run( + [ + sys.executable, + "-m", + "pip", + "wheel", + "--quiet", + "--no-deps", + "--wheel-dir", + str(wheel_dir), + str(worker_dir), + ], + check=True, + ) + wheels = sorted(wheel_dir.glob("*.whl")) + if len(wheels) != 1: + msg = f"Expected one worker wheel in {wheel_dir}, found {len(wheels)}" + raise RuntimeError(msg) + return wheels[0] + + +def _add_to_zip(archive: zipfile.ZipFile, name: str, data: bytes) -> None: + # A fixed timestamp and mode keep the zip free of local file metadata. + info = zipfile.ZipInfo(name, date_time=(1980, 1, 1, 0, 0, 0)) + info.external_attr = 0o644 << 16 + info.compress_type = zipfile.ZIP_DEFLATED + archive.writestr(info, data) + + +@dataclass(frozen=True) +class ImageArtifact: + """The image code zip and the content hash of its inputs.""" + + zip_path: Path + zip_bytes: bytes + content_hash: str + + +def build_artifact( + settings: dict[str, Any], + *, + worker_dir: Path = WORKER_DIR, + image_dir: Path = IMAGE_DIR, + out_dir: Path = OUT_DIR, + build_wheel: Callable[[Path, Path], Path] = build_worker_wheel, +) -> ImageArtifact: + """Build the image code zip in ``out_dir``, and hash its inputs. + + The zip holds the Dockerfile at its root, app.py, constraints.txt, and + wheels/. CreateMicrovmImage reads the Dockerfile from the + root. + """ + shutil.rmtree(out_dir, ignore_errors=True) + wheel_dir = out_dir / "wheels" + wheel_dir.mkdir(parents=True) + wheel = build_wheel(worker_dir, wheel_dir) + + zip_path = out_dir / "microvm-image.zip" + with zipfile.ZipFile(zip_path, "w") as archive: + for file_name in IMAGE_FILES: + _add_to_zip(archive, file_name, (image_dir / file_name).read_bytes()) + _add_to_zip(archive, f"wheels/{wheel.name}", wheel.read_bytes()) + + # The zip is read once here. A rebuild uploads the same bytes, even if + # another run has replaced the file since. + return ImageArtifact( + zip_path=zip_path, + zip_bytes=zip_path.read_bytes(), + content_hash=content_hash(hash_inputs(worker_dir, image_dir, settings)), + ) + + +def warn_on_custom_lambda_endpoint( + region: str, environ: dict[str, str] | None = None +) -> bool: + """Warn when the functions use a Lambda endpoint other than the default. + + The worker in the MicroVM sends its callbacks to the default endpoint of + the Region. So an execution on another endpoint never receives them, and + the MicroVM tests time out. Returns True when it warned. + """ + endpoint = (environ if environ is not None else os.environ).get("LAMBDA_ENDPOINT") + if not endpoint: + return False + default_endpoint = f"https://lambda.{region}.amazonaws.com" + if re.sub(r"/+$", "", endpoint) == default_endpoint: + return False + # The value is not printed, because CI passes it as a secret. + print( + f"WARNING: LAMBDA_ENDPOINT is not {default_endpoint}. The worker in " + "the MicroVM sends its callbacks to the default endpoint. So the " + "MicroVM examples will time out.", + file=sys.stderr, + ) + return True + + +def _require_env(name: str) -> str: + value = os.environ.get(name) + if not value: + msg = f"{name} is not set" + raise SystemExit(msg) + return value + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="Build the MicroVM image for the examples, or reuse it" + ) + parser.add_argument( + "--output", + required=True, + type=Path, + help="Write {imageArn, executionRoleArn} as JSON to this file", + ) + args = parser.parse_args(argv) + + region = _require_env("AWS_REGION") + account = _require_env("TEST_ACCOUNT_ID") + # An unset GitHub secret or variable reaches the step as an empty string. + # So an empty override counts as unset. + bucket = ( + os.environ.get("MICROVM_ARTIFACT_BUCKET") + or f"microvm-image-artifacts-{account}-{region}" + ) + execution_role_arn = ( + os.environ.get("MICROVM_EXECUTION_ROLE_ARN") + or f"arn:aws:iam::{account}:role/microvm-execution" + ) + settings = image_settings( + region, account, os.environ.get("MICROVM_IMAGE_BUILD_ROLE_ARN") + ) + + warn_on_custom_lambda_endpoint(region) + + artifact = build_artifact(settings) + name = f"{IMAGE_NAME_PREFIX}-{artifact.content_hash}" + # GetMicrovmImage accepts only the image ARN, not the name. + image_arn = f"arn:aws:lambda:{region}:{account}:microvm-image:{name}" + print(f"MicroVM image {name}, zip {len(artifact.zip_bytes)} bytes") + + service = BotoImageService( + microvms=boto3.client("lambda-microvms", region_name=region), + s3=boto3.client("s3", region_name=region), + name=name, + image_arn=image_arn, + bucket=bucket, + key=f"examples/{name}.zip", + zip_bytes=artifact.zip_bytes, + settings=settings, + ) + ensure_image(service, name) + + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text( + json.dumps({"imageArn": image_arn, "executionRoleArn": execution_role_arn}) + ) + print(f"Wrote {args.output}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/packages/aws-durable-execution-sdk-python-examples/scripts/generate_function_name_map.py b/packages/aws-durable-execution-sdk-python-examples/scripts/generate_function_name_map.py index e963bae0d..513c137cd 100644 --- a/packages/aws-durable-execution-sdk-python-examples/scripts/generate_function_name_map.py +++ b/packages/aws-durable-execution-sdk-python-examples/scripts/generate_function_name_map.py @@ -2,21 +2,40 @@ import argparse import json +from typing import Any -from generate_sam_template import load_catalog, to_logical_id +from generate_sam_template import load_catalog, to_logical_id, uses_microvm -def generate_function_name_map(function_name_prefix: str) -> dict[str, str]: - """Generate a test-name to qualified Lambda function-name map.""" - catalog = load_catalog() +def build_function_name_map( + examples: list[dict[str, Any]], + function_name_prefix: str, + include_microvm: bool = False, +) -> dict[str, str]: + """Map each example name to its qualified Lambda function name. + + The stack deploys a MicroVM example only when CI passes an image. So the + map leaves a MicroVM example out unless ``include_microvm`` is true. The + conftest skips a test whose example is missing from the map. + """ return { example["name"]: ( f"{function_name_prefix}{to_logical_id(example['handler'])}:$LATEST" ) - for example in catalog["examples"] + for example in examples + if include_microvm or not uses_microvm(example) } +def generate_function_name_map( + function_name_prefix: str, include_microvm: bool = False +) -> dict[str, str]: + """Generate a test-name to qualified Lambda function-name map.""" + return build_function_name_map( + load_catalog()["examples"], function_name_prefix, include_microvm + ) + + def main() -> int: parser = argparse.ArgumentParser( description="Generate pytest cloud function-name mapping" @@ -26,11 +45,21 @@ def main() -> int: required=True, help="Prefix used by the deployed SAM stack for Lambda function names", ) + parser.add_argument( + "--include-microvm", + action="store_true", + help=( + 'Include the examples with "usesMicrovm": true. Pass it only when ' + "the stack was deployed with MicrovmImageArn." + ), + ) args = parser.parse_args() print( json.dumps( - generate_function_name_map(args.function_name_prefix), + generate_function_name_map( + args.function_name_prefix, include_microvm=args.include_microvm + ), separators=(",", ":"), ) ) diff --git a/packages/aws-durable-execution-sdk-python-examples/scripts/generate_sam_template.py b/packages/aws-durable-execution-sdk-python-examples/scripts/generate_sam_template.py index 1089024e8..3c1e63ffe 100644 --- a/packages/aws-durable-execution-sdk-python-examples/scripts/generate_sam_template.py +++ b/packages/aws-durable-execution-sdk-python-examples/scripts/generate_sam_template.py @@ -9,6 +9,12 @@ DEFAULT_FUNCTION_NAME_PREFIX = "DurablePythonExample-" DEFAULT_DURABLE_LOGGING_CONFIG: dict[str, str] = {"LogFormat": "JSON"} LOG_RETENTION_DAYS = 7 +MICROVM_CONDITION = "HasMicrovmImage" + + +def uses_microvm(example: dict[str, Any]) -> bool: + """Return true when the catalog example launches a MicroVM.""" + return example.get("usesMicrovm") is True def load_catalog() -> dict[str, Any]: @@ -67,10 +73,48 @@ def build_template(examples: list[dict[str, Any]]) -> dict[str, Any]: "Type": "String", "Description": "ARN of the ADOT Python Lambda layer.", }, + # A deploy can pass both parameters while no example uses them. + # CloudFormation rejects an override of an undeclared parameter. + # So the template always declares both. + "MicrovmImageArn": { + "Type": "String", + "Default": "", + "Description": ( + "ARN of the MicroVM image for the MicroVM examples. " + "Empty leaves the MicroVM examples out of the stack." + ), + }, + "MicrovmExecutionRoleArn": { + "Type": "String", + "Default": "", + "Description": ( + "ARN of the IAM role that the MicroVMs of the MicroVM " + "examples run with." + ), + }, }, "Resources": {}, } + # A MicroVM example launches the image named by MicrovmImageArn. Without an + # image and a role, every invocation of its function would fail. So the + # function and its log group exist only when both parameters are set. + # CloudFormation linters flag an unused condition. So the template declares + # the condition only when an example uses it. + if any(uses_microvm(example) for example in examples): + template["Conditions"] = { + MICROVM_CONDITION: { + "Fn::And": [ + {"Fn::Not": [{"Fn::Equals": [{"Ref": "MicrovmImageArn"}, ""]}]}, + { + "Fn::Not": [ + {"Fn::Equals": [{"Ref": "MicrovmExecutionRoleArn"}, ""]} + ] + }, + ] + } + } + for example in examples: logical_id = to_logical_id(example["handler"]) properties: dict[str, Any] = { @@ -99,15 +143,23 @@ def build_template(examples: list[dict[str, Any]]) -> dict[str, Any]: if "tracing" in example: properties["Tracing"] = example["tracing"] - if "environment" in example: - properties["Environment"] = {"Variables": example["environment"]} + if "timeout" in example: + properties["Timeout"] = example["timeout"] + + variables: dict[str, Any] = dict(example.get("environment", {})) + if uses_microvm(example): + # SAM merges these variables with the Globals variables. + variables["MICROVM_IMAGE_ARN"] = {"Ref": "MicrovmImageArn"} + variables["MICROVM_EXECUTION_ROLE_ARN"] = {"Ref": "MicrovmExecutionRoleArn"} + if variables: + properties["Environment"] = {"Variables": variables} - template["Resources"][logical_id] = { + function_resource: dict[str, Any] = { "Type": "AWS::Serverless::Function", "DependsOn": [f"{logical_id}LogGroup"], "Properties": properties, } - template["Resources"][f"{logical_id}LogGroup"] = { + log_group_resource: dict[str, Any] = { "Type": "AWS::Logs::LogGroup", "Properties": { "LogGroupName": { @@ -116,6 +168,14 @@ def build_template(examples: list[dict[str, Any]]) -> dict[str, Any]: "RetentionInDays": LOG_RETENTION_DAYS, }, } + if uses_microvm(example): + # SAM copies the condition of a function to the resources that it + # generates for the function. + function_resource["Condition"] = MICROVM_CONDITION + log_group_resource["Condition"] = MICROVM_CONDITION + + template["Resources"][logical_id] = function_resource + template["Resources"][f"{logical_id}LogGroup"] = log_group_resource return template diff --git a/packages/aws-durable-execution-sdk-python-examples/src/microvm/microvm_run_job.py b/packages/aws-durable-execution-sdk-python-examples/src/microvm/microvm_run_job.py new file mode 100644 index 000000000..c058d76ca --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-examples/src/microvm/microvm_run_job.py @@ -0,0 +1,57 @@ +"""Run one job in a Lambda MicroVM and return the result that it reports. + +The MicroVM runs the image that scripts/ensure_microvm_image.py builds from +microvm-image/. The image's default handler counts the words of the text and +hashes it. +""" + +import os +from typing import Any + +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 durable_execution +from aws_durable_execution_sdk_python_extras.microvm import MicrovmConfig, microvm + + +def _require_microvm_env() -> tuple[str, str]: + """Return the image ARN and the MicroVM execution role ARN. + + The SAM template sets both variables on every example with + ``"usesMicrovm": true``. scripts/build_examples.py copies every file under + src/ into one directory. So a shared helper module would need a different + import in Lambda than in the local tests. So each MicroVM example defines + this function itself. + """ + image_arn = os.environ.get("MICROVM_IMAGE_ARN") + execution_role_arn = os.environ.get("MICROVM_EXECUTION_ROLE_ARN") + if not image_arn or not execution_role_arn: + msg = ( + "MICROVM_IMAGE_ARN and MICROVM_EXECUTION_ROLE_ARN must be set. " + "Build the image with scripts/ensure_microvm_image.py." + ) + raise RuntimeError(msg) + return image_arn, execution_role_arn + + +@durable_execution +def handler(event: dict[str, Any], context: DurableContext) -> Any: + image_arn, execution_role_arn = _require_microvm_env() + + # The operation launches a MicroVM from the image and waits on a durable + # callback. The input is small, so it goes in the MicroVM's run hook. The + # worker in the MicroVM runs the job and completes the callback. The + # operation then terminates the MicroVM. + return microvm( + context, + "count-words", + job_input={"text": event["text"]}, + config=MicrovmConfig( + image_identifier=image_arn, + execution_role_arn=execution_role_arn, + timeout=Duration.from_minutes(5), + # A MicroVM that fails to boot fails the operation after this + # time, instead of after the timeout. + heartbeat_timeout=Duration.from_seconds(60), + ), + ) diff --git a/packages/aws-durable-execution-sdk-python-examples/src/microvm/microvm_session.py b/packages/aws-durable-execution-sdk-python-examples/src/microvm/microvm_session.py new file mode 100644 index 000000000..761e67d6e --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-examples/src/microvm/microvm_session.py @@ -0,0 +1,96 @@ +"""Run two jobs in one Lambda MicroVM, with a durable wait between them. + +The MicroVM runs the image that scripts/ensure_microvm_image.py builds from +microvm-image/. The image's routes /write and /read write and read a file in +the MicroVM. +""" + +import os +from typing import Any + +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 durable_execution +from aws_durable_execution_sdk_python_extras.microvm import ( + MicrovmInvokeOptions, + MicrovmSession, + MicrovmSessionConfig, + microvm_session, +) + + +def _require_microvm_env() -> tuple[str, str]: + """Return the image ARN and the MicroVM execution role ARN. + + The SAM template sets both variables on every example with + ``"usesMicrovm": true``. scripts/build_examples.py copies every file under + src/ into one directory. So a shared helper module would need a different + import in Lambda than in the local tests. So each MicroVM example defines + this function itself. + """ + image_arn = os.environ.get("MICROVM_IMAGE_ARN") + execution_role_arn = os.environ.get("MICROVM_EXECUTION_ROLE_ARN") + if not image_arn or not execution_role_arn: + msg = ( + "MICROVM_IMAGE_ARN and MICROVM_EXECUTION_ROLE_ARN must be set. " + "Build the image with scripts/ensure_microvm_image.py." + ) + raise RuntimeError(msg) + return image_arn, execution_role_arn + + +def _job_options(path: str) -> MicrovmInvokeOptions: + return MicrovmInvokeOptions( + path=path, + timeout=Duration.from_minutes(5), + # A job that the MicroVM stops answering fails after this time, + # instead of after the timeout. + heartbeat_timeout=Duration.from_seconds(30), + ) + + +@durable_execution +def handler(event: dict[str, Any], context: DurableContext) -> Any: + image_arn, execution_role_arn = _require_microvm_env() + key: str = event["key"] + + def write_then_read(vm: MicrovmSession, session_context: DurableContext) -> Any: + # Each vm.invoke sends one job over HTTP to a route of the app in the + # MicroVM. + written = vm.invoke( + session_context, + "write", + job_input={"key": key, "value": f"written-by-{key}"}, + options=_job_options("/write"), + ) + + # The wait ends this invocation. The MicroVM keeps running. So the + # next job finds the file that the first job wrote. + session_context.wait(Duration.from_seconds(5), name="pause") + + read = vm.invoke( + session_context, + "read", + job_input={"key": key}, + options=_job_options("/read"), + ) + + return { + "sessionMicrovmId": vm.microvm_id, + "writtenBy": written["microvmId"], + "readBy": read["microvmId"], + "value": read["value"], + } + + # The session launches one MicroVM, and terminates it when the handler + # returns. + return microvm_session( + context, + "file-session", + config=MicrovmSessionConfig( + image_identifier=image_arn, + execution_role_arn=execution_role_arn, + timeout=Duration.from_minutes(10), + ), + handler=write_then_read, + ) diff --git a/packages/aws-durable-execution-sdk-python-examples/test/README.md b/packages/aws-durable-execution-sdk-python-examples/test/README.md index 4e48b54ab..c46958e8b 100644 --- a/packages/aws-durable-execution-sdk-python-examples/test/README.md +++ b/packages/aws-durable-execution-sdk-python-examples/test/README.md @@ -133,6 +133,26 @@ def test_my_example(durable_runner): ## Configuration +### Catalog Keys for Deployment +`scripts/generate_sam_template.py` reads these optional keys from each entry in `examples-catalog.json`: +- `timeout` - Lambda function timeout in seconds. Without it, the function uses the global timeout of 60 seconds. +- `usesMicrovm` - `true` marks an example that launches a MicroVM. The template then gives its function and log group the condition `HasMicrovmImage`. The condition holds only when the parameters `MicrovmImageArn` and `MicrovmExecutionRoleArn` are both set. The function receives them as `MICROVM_IMAGE_ARN` and `MICROVM_EXECUTION_ROLE_ARN`. So a deploy without an image leaves the example out of the stack. + +`scripts/generate_function_name_map.py` leaves out the `usesMicrovm` examples unless you pass `--include-microvm`. Pass it only when the stack was deployed with an image. A test whose example is missing from the map is skipped. + +### MicroVM Image +`microvm-image/` holds the app, the Dockerfile, and the dependency pins of the MicroVM image that the MicroVM examples run. `hatch run examples:ensure-microvm-image --output image.json` builds the image or reuses it. It needs `TEST_ACCOUNT_ID`, `AWS_REGION`, and AWS credentials for the test account. + +1. The script builds the worker wheel from `packages/aws-durable-execution-sdk-python-microvm-worker`, and zips it with the files in `microvm-image/`. +2. The image name is `sdk-python-examples-`. The hash covers the worker's source and `pyproject.toml`, the files in `microvm-image/`, and the `CreateMicrovmImage` settings. +3. So a run with unchanged inputs reuses the image. A change to any input builds a new image. +4. The script writes `{"imageArn": ..., "executionRoleArn": ...}` to `--output`. CI passes both to `sam deploy` as `MicrovmImageArn` and `MicrovmExecutionRoleArn`, in the Python 3.13 job only. + +`MICROVM_ARTIFACT_BUCKET`, `MICROVM_IMAGE_BUILD_ROLE_ARN`, and `MICROVM_EXECUTION_ROLE_ARN` override the resource names for another account. + +### Cloud-Only Tests +`@pytest.mark.durable_execution(..., cloud_only=True)` marks a test that needs a service the local runner cannot provide, such as a MicroVM. Local mode skips the test with the reason `cloud_only: runs only against the deployed function`. Cloud mode runs it. + ### Environment Variables (Cloud Mode) - `AWS_REGION` - AWS region for Lambda invocation (default: us-west-2) - `LAMBDA_ENDPOINT` - Optional Lambda endpoint URL for testing diff --git a/packages/aws-durable-execution-sdk-python-examples/test/conftest.py b/packages/aws-durable-execution-sdk-python-examples/test/conftest.py index 61963eac1..e41d7fa9b 100644 --- a/packages/aws-durable-execution-sdk-python-examples/test/conftest.py +++ b/packages/aws-durable-execution-sdk-python-examples/test/conftest.py @@ -179,6 +179,24 @@ def __exit__(self, exc_type, exc_val, exc_tb): return None +CLOUD_ONLY_SKIP_REASON = ( + "cloud_only: runs only against the deployed function (--runner-mode=cloud)" +) + + +def cloud_only_skip_reason( + marker_kwargs: dict[str, Any], runner_mode: str +) -> str | None: + """Return the skip reason for a cloud-only test in local mode, else None. + + A test marked ``cloud_only=True`` depends on a service that the local + runner cannot provide, such as a MicroVM. So the local run skips it. + """ + if marker_kwargs.get("cloud_only") and runner_mode != RunnerMode.CLOUD: + return CLOUD_ONLY_SKIP_REASON + return None + + @pytest.fixture def durable_runner(request): """Pytest fixture that provides a test runner based on configuration. @@ -207,6 +225,8 @@ def test_hello_world(durable_runner): with durable_runner: result = durable_runner.run(input="test", timeout=10) assert result.status == InvocationStatus.SUCCEEDED + + Marker keyword ``cloud_only=True`` skips the test in local mode. """ # Get marker with test configuration marker = request.node.get_closest_marker("durable_execution") @@ -219,6 +239,10 @@ def test_hello_world(durable_runner): # Get runner mode from CLI option runner_mode: str = request.config.getoption("--runner-mode") + skip_reason = cloud_only_skip_reason(marker.kwargs, runner_mode) + if skip_reason: + pytest.skip(skip_reason) + logger.info("Running test in %s mode", runner_mode.upper()) # Create appropriate runner diff --git a/packages/aws-durable-execution-sdk-python-examples/test/microvm/microvm_helpers.py b/packages/aws-durable-execution-sdk-python-examples/test/microvm/microvm_helpers.py new file mode 100644 index 000000000..e5f6914ff --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-examples/test/microvm/microvm_helpers.py @@ -0,0 +1,66 @@ +"""Checks for the cloud tests of the MicroVM examples.""" + +import os +import time +from collections.abc import Callable +from typing import Any + +from botocore.exceptions import ClientError + + +ENDED_STATES = frozenset({"TERMINATING", "TERMINATED"}) +POLL_INTERVAL_SECONDS = 2.0 +TIMEOUT_SECONDS = 30.0 + + +def expect_microvm_terminated( + microvm_id: str, + *, + client: Any = None, + sleep: Callable[[float], None] = time.sleep, + clock: Callable[[], float] = time.monotonic, +) -> None: + """Assert that the operation terminated the MicroVM. + + 1. A TerminateMicrovm call that fails does not fail the operation. The + operation logs the error and returns the job's result. + 2. So the operation tree shows a successful terminate step either way. + 3. So this check reads the MicroVM state with GetMicrovm instead. + + TerminateMicrovm returns before the state changes. The API does not + document how soon GetMicrovm reports the change. So the check reads the + state every 2 seconds, until it is TERMINATING or TERMINATED, for up to + 30 seconds. A MicroVM that GetMicrovm does not find is not running either. + A MicroVM in another state after 30 seconds fails the test. + + The default client is created here, not at import. So the test modules + import without AWS credentials, and the local run can skip them. + """ + if client is None: + import boto3 + + client = boto3.client( + "lambda-microvms", region_name=os.environ.get("AWS_REGION", "us-west-2") + ) + + deadline = clock() + TIMEOUT_SECONDS + while True: + try: + state = client.get_microvm(microvmIdentifier=microvm_id).get("state") + except ClientError as error: + if ( + error.response.get("Error", {}).get("Code") + == "ResourceNotFoundException" + ): + return + raise + if state in ENDED_STATES: + return + if clock() >= deadline: + msg = ( + f"MicroVM {microvm_id} is {state} {TIMEOUT_SECONDS:.0f} seconds " + "after the execution ended. The operation's TerminateMicrovm " + "call failed or did not take effect." + ) + raise AssertionError(msg) + sleep(POLL_INTERVAL_SECONDS) diff --git a/packages/aws-durable-execution-sdk-python-examples/test/microvm/test_microvm_catalog.py b/packages/aws-durable-execution-sdk-python-examples/test/microvm/test_microvm_catalog.py new file mode 100644 index 000000000..eac08869c --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-examples/test/microvm/test_microvm_catalog.py @@ -0,0 +1,35 @@ +"""Tests for the catalog entries of the MicroVM examples.""" + +import json +from pathlib import Path + + +CATALOG_PATH = Path(__file__).resolve().parents[2] / "examples-catalog.json" + +# The request step of an HTTP delivery retries for up to retry_window, 60 +# seconds by default. It needs at least 11 seconds left in the invocation +# when it starts. So the extras README asks for a function timeout of at +# least retry_window plus 11 seconds. +HTTP_DELIVERY_MIN_TIMEOUT_SECONDS = 60 + 11 + + +def catalog_entry(name): + examples = json.loads(CATALOG_PATH.read_text(encoding="utf-8"))["examples"] + return next(example for example in examples if example["name"] == name) + + +def test_the_session_function_timeout_covers_the_http_retry_window(): + # Every session job goes over HTTP. + entry = catalog_entry("MicroVM Session") + + assert entry["usesMicrovm"] is True + assert entry["timeout"] >= HTTP_DELIVERY_MIN_TIMEOUT_SECONDS + + +def test_the_run_job_function_uses_the_global_timeout(): + # The input is small, so the job goes in the run hook. The run hook + # delivery adds no Lambda time for the boot. + entry = catalog_entry("MicroVM Run Job") + + assert entry["usesMicrovm"] is True + assert "timeout" not in entry diff --git a/packages/aws-durable-execution-sdk-python-examples/test/microvm/test_microvm_env.py b/packages/aws-durable-execution-sdk-python-examples/test/microvm/test_microvm_env.py new file mode 100644 index 000000000..0d17deb3b --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-examples/test/microvm/test_microvm_env.py @@ -0,0 +1,43 @@ +"""Tests for the MicroVM settings that the MicroVM examples read.""" + +import pytest + + +# The examples import the extras. So an environment without the extras skips +# this module. +pytest.importorskip("aws_durable_execution_sdk_python_extras") + +from src.microvm import microvm_run_job, microvm_session # noqa: E402 + + +EXAMPLES = [microvm_run_job, microvm_session] +IMAGE_ARN = "arn:aws:lambda:us-west-2:123456789012:microvm-image:img" +ROLE_ARN = "arn:aws:iam::123456789012:role/microvm-execution" + + +@pytest.mark.parametrize("example", EXAMPLES, ids=lambda module: module.__name__) +def test_returns_the_image_and_the_role(example, monkeypatch): + monkeypatch.setenv("MICROVM_IMAGE_ARN", IMAGE_ARN) + monkeypatch.setenv("MICROVM_EXECUTION_ROLE_ARN", ROLE_ARN) + + assert example._require_microvm_env() == (IMAGE_ARN, ROLE_ARN) + + +@pytest.mark.parametrize("example", EXAMPLES, ids=lambda module: module.__name__) +@pytest.mark.parametrize("missing", ["MICROVM_IMAGE_ARN", "MICROVM_EXECUTION_ROLE_ARN"]) +@pytest.mark.parametrize("unset", [True, False], ids=["unset", "empty"]) +def test_names_both_variables_when_one_is_missing(example, missing, unset, monkeypatch): + monkeypatch.setenv("MICROVM_IMAGE_ARN", IMAGE_ARN) + monkeypatch.setenv("MICROVM_EXECUTION_ROLE_ARN", ROLE_ARN) + if unset: + monkeypatch.delenv(missing) + else: + monkeypatch.setenv(missing, "") + + with pytest.raises(RuntimeError) as error: + example._require_microvm_env() + + assert str(error.value) == ( + "MICROVM_IMAGE_ARN and MICROVM_EXECUTION_ROLE_ARN must be set. " + "Build the image with scripts/ensure_microvm_image.py." + ) diff --git a/packages/aws-durable-execution-sdk-python-examples/test/microvm/test_microvm_helpers.py b/packages/aws-durable-execution-sdk-python-examples/test/microvm/test_microvm_helpers.py new file mode 100644 index 000000000..3d8446307 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-examples/test/microvm/test_microvm_helpers.py @@ -0,0 +1,104 @@ +"""Tests for expect_microvm_terminated with a fake client and a fake clock.""" + +from typing import Any + +import pytest +from botocore.exceptions import ClientError + +from test.microvm.microvm_helpers import ( + POLL_INTERVAL_SECONDS, + TIMEOUT_SECONDS, + expect_microvm_terminated, +) + + +def client_error(code: str) -> ClientError: + return ClientError({"Error": {"Code": code, "Message": code}}, "GetMicrovm") + + +class FakeClock: + """A monotonic clock that advances only when the check sleeps.""" + + def __init__(self) -> None: + self.now = 1000.0 + self.sleeps: list[float] = [] + + def __call__(self) -> float: + return self.now + + def sleep(self, seconds: float) -> None: + self.sleeps.append(seconds) + self.now += seconds + + +class FakeMicrovmsClient: + """Answers GetMicrovm from a list of states or exceptions, in order. + + After the list ends, it repeats the last answer. + """ + + def __init__(self, answers: list[Any]) -> None: + self._answers = answers + self.requests: list[dict[str, Any]] = [] + + def get_microvm(self, **request: Any) -> dict[str, Any]: + self.requests.append(request) + index = min(len(self.requests), len(self._answers)) - 1 + answer = self._answers[index] + if isinstance(answer, BaseException): + raise answer + return {"microvmId": request["microvmIdentifier"], "state": answer} + + +def check(client: FakeMicrovmsClient, clock: FakeClock) -> None: + expect_microvm_terminated("mvm-1", client=client, sleep=clock.sleep, clock=clock) + + +@pytest.mark.parametrize("state", ["TERMINATING", "TERMINATED"]) +def test_returns_at_once_for_an_ended_microvm(state): + client = FakeMicrovmsClient([state]) + clock = FakeClock() + + check(client, clock) + + assert client.requests == [{"microvmIdentifier": "mvm-1"}] + assert clock.sleeps == [] + + +def test_returns_for_a_microvm_that_get_microvm_does_not_find(): + client = FakeMicrovmsClient([client_error("ResourceNotFoundException")]) + clock = FakeClock() + + check(client, clock) + + assert clock.sleeps == [] + + +def test_polls_every_two_seconds_until_the_microvm_ends(): + client = FakeMicrovmsClient(["RUNNING", "RUNNING", "TERMINATING"]) + clock = FakeClock() + + check(client, clock) + + assert len(client.requests) == 3 + assert clock.sleeps == [POLL_INTERVAL_SECONDS, POLL_INTERVAL_SECONDS] + + +def test_fails_for_a_microvm_that_still_runs_after_30_seconds(): + client = FakeMicrovmsClient(["RUNNING"]) + clock = FakeClock() + + with pytest.raises(AssertionError, match="MicroVM mvm-1 is RUNNING 30 seconds"): + check(client, clock) + + # The check reads the state at 0, 2, ..., 30 seconds. + assert len(client.requests) == TIMEOUT_SECONDS / POLL_INTERVAL_SECONDS + 1 + assert sum(clock.sleeps) == TIMEOUT_SECONDS + + +def test_raises_another_get_microvm_error(): + client = FakeMicrovmsClient([client_error("AccessDeniedException")]) + clock = FakeClock() + + with pytest.raises(ClientError, match="AccessDeniedException"): + check(client, clock) diff --git a/packages/aws-durable-execution-sdk-python-examples/test/microvm/test_microvm_run_job.py b/packages/aws-durable-execution-sdk-python-examples/test/microvm/test_microvm_run_job.py new file mode 100644 index 000000000..ced2bd717 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-examples/test/microvm/test_microvm_run_job.py @@ -0,0 +1,75 @@ +"""Cloud test for the MicroVM run-job example. + +The local runner cannot launch a MicroVM. So the test is cloud-only. +""" + +import hashlib + +import pytest +from aws_durable_execution_sdk_python.execution import InvocationStatus +from aws_durable_execution_sdk_python.lambda_service import ( + OperationStatus, + OperationType, +) + + +# The examples package does not depend on the extras. So an environment +# without the extras skips this module instead of failing to import the +# example. +pytest.importorskip("aws_durable_execution_sdk_python_extras") + +from src.microvm import microvm_run_job # noqa: E402 +from test.conftest import deserialize_operation_payload # noqa: E402 +from test.microvm.microvm_helpers import expect_microvm_terminated # noqa: E402 + + +TEXT = "the quick brown fox jumps over the lazy dog" + + +@pytest.mark.example +@pytest.mark.durable_execution( + handler=microvm_run_job.handler, + lambda_function_name="MicroVM Run Job", + cloud_only=True, +) +def test_runs_the_job_in_a_microvm_and_returns_its_result(durable_runner): + with durable_runner: + # A synchronous Invoke returns only when the execution ends. The cloud + # runner's boto3 client reads for at most 60 seconds, and it sends the + # Invoke again after a read timeout. Each resend starts a new + # execution, which launches its own MicroVM. So the test starts the + # execution asynchronously and polls for its result. + execution_arn = durable_runner.run_async(input={"text": TEXT}, timeout=300) + result = durable_runner.wait_for_result( + execution_arn=execution_arn, timeout=300 + ) + + assert result.status is InvocationStatus.SUCCEEDED, result.error + output = deserialize_operation_payload(result.result) + assert output["wordCount"] == 9 + assert output["sha256"] == hashlib.sha256(TEXT.encode()).hexdigest() + assert isinstance(output["microvmId"], str) + assert output["microvmId"] + + expect_microvm_terminated(output["microvmId"]) + + # The operation runs its durable operations in one child context named + # after the call. The input is small, so the launch step delivers the job + # in the run hook, and no request step runs. + context = result.get_context("count-words") + assert context.operation_type is OperationType.CONTEXT + assert context.status is OperationStatus.SUCCEEDED + assert [ + (child.name, child.operation_type, child.status) + for child in context.child_operations + ] == [ + ("count-words.callback", OperationType.CALLBACK, OperationStatus.SUCCEEDED), + ("count-words.launch", OperationType.STEP, OperationStatus.SUCCEEDED), + ("count-words.terminate", OperationType.STEP, OperationStatus.SUCCEEDED), + ] + + launch = deserialize_operation_payload( + context.get_step("count-words.launch").result + ) + assert launch["delivery"] == "run-hook" + assert launch["microvmId"] == output["microvmId"] diff --git a/packages/aws-durable-execution-sdk-python-examples/test/microvm/test_microvm_session.py b/packages/aws-durable-execution-sdk-python-examples/test/microvm/test_microvm_session.py new file mode 100644 index 000000000..0ed20eb6e --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-examples/test/microvm/test_microvm_session.py @@ -0,0 +1,112 @@ +"""Cloud test for the MicroVM session example. + +The local runner cannot launch a MicroVM. So the test is cloud-only. +""" + +import uuid + +import pytest +from aws_durable_execution_sdk_python.execution import InvocationStatus +from aws_durable_execution_sdk_python.lambda_service import ( + OperationStatus, + OperationType, +) + + +# The examples package does not depend on the extras. So an environment +# without the extras skips this module instead of failing to import the +# example. +pytest.importorskip("aws_durable_execution_sdk_python_extras") + +from src.microvm import microvm_session # noqa: E402 +from test.conftest import deserialize_operation_payload # noqa: E402 +from test.microvm.microvm_helpers import expect_microvm_terminated # noqa: E402 + + +def tree(operations): + """Return (name, type, status, children) for each operation, in order.""" + return [ + ( + operation.name, + operation.operation_type, + operation.status, + tree(getattr(operation, "child_operations", None) or []), + ) + for operation in operations + ] + + +SUCCEEDED = OperationStatus.SUCCEEDED + + +@pytest.mark.example +@pytest.mark.durable_execution( + handler=microvm_session.handler, + lambda_function_name="MicroVM Session", + cloud_only=True, +) +def test_runs_both_jobs_in_the_same_microvm(durable_runner): + key = str(uuid.uuid4()) + + with durable_runner: + # A synchronous Invoke returns only when the execution ends. The cloud + # runner's boto3 client reads for at most 60 seconds, and it sends the + # Invoke again after a read timeout. Each resend starts a new + # execution, which launches its own MicroVM. So the test starts the + # execution asynchronously and polls for its result. + execution_arn = durable_runner.run_async(input={"key": key}, timeout=300) + result = durable_runner.wait_for_result( + execution_arn=execution_arn, timeout=300 + ) + + assert result.status is InvocationStatus.SUCCEEDED, result.error + output = deserialize_operation_payload(result.result) + # The second job read the file that the first job wrote. So both jobs ran + # in the session's MicroVM. + assert output["value"] == f"written-by-{key}" + assert output["writtenBy"] == output["sessionMicrovmId"] + assert output["readBy"] == output["sessionMicrovmId"] + + expect_microvm_terminated(output["sessionMicrovmId"]) + + # The session runs its durable operations in one child context. Each job + # runs in its own child context. Every session job goes over HTTP. So + # each job has a request step, and the launch result records "http". + assert tree(result.operations) == [ + ( + "file-session", + OperationType.CONTEXT, + SUCCEEDED, + [ + ("file-session.session", OperationType.STEP, SUCCEEDED, []), + ("file-session.launch", OperationType.STEP, SUCCEEDED, []), + ( + "write", + OperationType.CONTEXT, + SUCCEEDED, + [ + ("write.callback", OperationType.CALLBACK, SUCCEEDED, []), + ("write.request", OperationType.STEP, SUCCEEDED, []), + ], + ), + ("pause", OperationType.WAIT, SUCCEEDED, []), + ( + "read", + OperationType.CONTEXT, + SUCCEEDED, + [ + ("read.callback", OperationType.CALLBACK, SUCCEEDED, []), + ("read.request", OperationType.STEP, SUCCEEDED, []), + ], + ), + ("file-session.terminate", OperationType.STEP, SUCCEEDED, []), + ], + ) + ] + + session = result.get_context("file-session") + launch = deserialize_operation_payload( + session.get_step("file-session.launch").result + ) + assert launch["delivery"] == "http" + assert launch["microvmId"] == output["sessionMicrovmId"] diff --git a/packages/aws-durable-execution-sdk-python-examples/test/test_cloud_only.py b/packages/aws-durable-execution-sdk-python-examples/test/test_cloud_only.py new file mode 100644 index 000000000..f115ec376 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-examples/test/test_cloud_only.py @@ -0,0 +1,23 @@ +"""Tests for the cloud_only keyword of the durable_execution marker.""" + +from test.conftest import ( + CLOUD_ONLY_SKIP_REASON, + RunnerMode, + cloud_only_skip_reason, +) + + +def test_cloud_only_test_is_skipped_in_local_mode(): + assert ( + cloud_only_skip_reason({"cloud_only": True}, RunnerMode.LOCAL) + == CLOUD_ONLY_SKIP_REASON + ) + + +def test_cloud_only_test_runs_in_cloud_mode(): + assert cloud_only_skip_reason({"cloud_only": True}, RunnerMode.CLOUD) is None + + +def test_test_without_cloud_only_runs_in_local_mode(): + assert cloud_only_skip_reason({}, RunnerMode.LOCAL) is None + assert cloud_only_skip_reason({"cloud_only": False}, RunnerMode.LOCAL) is None diff --git a/packages/aws-durable-execution-sdk-python-examples/test/test_ensure_microvm_image.py b/packages/aws-durable-execution-sdk-python-examples/test/test_ensure_microvm_image.py new file mode 100644 index 000000000..a277962fa --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-examples/test/test_ensure_microvm_image.py @@ -0,0 +1,791 @@ +"""Tests for scripts/ensure_microvm_image.py. + +The tests make no AWS calls. A fake image service and a fake clock drive +ensure_image(). Fake clients drive BotoImageService. +""" + +import importlib.util +import json +import shutil +import sys +import zipfile +from pathlib import Path + +import pytest +from botocore.exceptions import ClientError + + +SCRIPT_PATH = ( + Path(__file__).resolve().parents[1] / "scripts" / "ensure_microvm_image.py" +) +SPEC = importlib.util.spec_from_file_location("ensure_microvm_image", SCRIPT_PATH) +assert SPEC is not None +assert SPEC.loader is not None +ensure_microvm_image = importlib.util.module_from_spec(SPEC) +# dataclasses reads the module from sys.modules. +sys.modules[SPEC.name] = ensure_microvm_image +SPEC.loader.exec_module(ensure_microvm_image) + +NAME = "sdk-python-examples-test" +MISSING = None + + +class FakeImageService: + """A scripted image service. + + get() returns the states in ``queue`` one per call, and repeats the last + one. create() and delete() replace the queue with the states that the + scenario gives for that call. ``MISSING`` stands for a missing image. + """ + + def __init__(self, queue, after_create=None, after_delete=None): + self.queue = list(queue) + self.after_create = after_create or (lambda _creates: ["CREATING", "CREATED"]) + self.after_delete = after_delete or (lambda: ["DELETING", MISSING]) + self.creates = 0 + self.deletes = 0 + self.gets = 0 + + def get(self): + self.gets += 1 + state = self.queue.pop(0) if len(self.queue) > 1 else self.queue[0] + if state is MISSING: + return None + image = {"state": state} + if state in {"CREATED", "UPDATED"}: + image["latestActiveImageVersion"] = "1.0" + return image + + def create(self): + self.creates += 1 + self.queue = list(self.after_create(self.creates)) + + def delete(self): + self.deletes += 1 + self.queue = list(self.after_delete()) + + +class FakeClock: + """A clock whose sleep() advances now() at once.""" + + def __init__(self): + self.time = 0.0 + + def now(self): + return self.time + + def sleep(self, seconds): + self.time += seconds + + +def run(service): + clock = FakeClock() + messages = [] + result = ensure_microvm_image.ensure_image( + service, + NAME, + now=clock.now, + sleep=clock.sleep, + log=messages.append, + ) + return result, messages, clock + + +def run_failing(service, match): + clock = FakeClock() + with pytest.raises(RuntimeError, match=match): + ensure_microvm_image.ensure_image( + service, NAME, now=clock.now, sleep=clock.sleep, log=lambda _m: None + ) + return clock + + +# ensure_image + + +def test_reuses_an_image_that_is_ready(): + service = FakeImageService(["CREATED"]) + + result, messages, _ = run(service) + + assert result == "1.0" + assert service.creates == 0 + assert messages[0].startswith("Reusing") + + +def test_creates_a_missing_image_and_waits_for_it(): + service = FakeImageService([MISSING]) + + result, _, _ = run(service) + + assert result == "1.0" + assert service.creates == 1 + + +def test_creates_the_image_after_a_delete_in_progress_finishes(): + service = FakeImageService(["DELETING", "DELETING", "DELETING", MISSING]) + + result, messages, _ = run(service) + + assert result == "1.0" + assert service.creates == 1 + assert not any(message.startswith("Reusing") for message in messages) + + +def test_treats_a_deleted_record_as_a_missing_image(): + service = FakeImageService(["DELETED"]) + + result, _, _ = run(service) + + assert result == "1.0" + assert service.creates == 1 + + +def test_deletes_a_failed_image_from_an_earlier_run_and_builds_it_again(): + service = FakeImageService(["CREATE_FAILED"]) + + result, _, _ = run(service) + + assert result == "1.0" + assert service.deletes == 1 + assert service.creates == 1 + + +def test_rebuilds_a_failed_build_once_then_raises_with_the_log_group(): + service = FakeImageService( + [MISSING], after_create=lambda _creates: ["CREATING", "CREATE_FAILED"] + ) + + run_failing(service, f"/aws/lambda-microvms/{NAME}") + + assert service.creates == 2 + assert service.deletes == 1 + + +def test_waits_while_a_deleted_failed_image_is_still_reported(): + service = FakeImageService( + ["CREATE_FAILED"], + after_delete=lambda: ["CREATE_FAILED", "DELETING", MISSING], + ) + + result, _, _ = run(service) + + assert result == "1.0" + assert service.deletes == 1 + assert service.creates == 1 + + +@pytest.mark.parametrize("state", ["DELETE_FAILED", "UPDATE_FAILED", "SOMETHING_NEW"]) +def test_raises_on_a_state_it_cannot_clear_without_a_delete_or_a_create(state): + service = FakeImageService([state]) + + run_failing(service, f"is {state}") + + assert service.creates == 0 + assert service.deletes == 0 + + +def test_does_not_create_again_while_a_new_image_is_not_visible_yet(): + service = FakeImageService( + [MISSING], + after_create=lambda _creates: [MISSING, MISSING, "CREATING", "CREATED"], + ) + + result, _, _ = run(service) + + assert result == "1.0" + assert service.creates == 1 + + +def test_creates_again_after_the_visibility_window(): + # The image stays missing for 7 polls of 10 seconds after the first + # create. The window is 60 seconds. So the second create comes at the + # first poll after the window. + service = FakeImageService( + [MISSING], + after_create=lambda creates: ( + [MISSING] * 7 + ["CREATING", "CREATED"] + if creates == 1 + else ["CREATING", "CREATED"] + ), + ) + + result, _, clock = run(service) + + assert result == "1.0" + assert service.creates == 2 + assert clock.time >= ensure_microvm_image.CREATE_VISIBILITY_SECONDS + + +def test_rebuilds_at_once_when_the_build_fails_inside_the_visibility_window(): + # The first create runs at 0 s. The build fails at 10 s, inside the + # 60-second window. The delete finishes at 30 s. The delete clears the + # create time. So the rebuild starts at 30 s, not at 60 s. + clock = FakeClock() + create_times = [] + + def after_create(creates): + create_times.append(clock.now()) + return ["CREATE_FAILED"] if creates == 1 else ["CREATING", "CREATED"] + + service = FakeImageService([MISSING], after_create=after_create) + + result = ensure_microvm_image.ensure_image( + service, NAME, now=clock.now, sleep=clock.sleep, log=lambda _m: None + ) + + assert result == "1.0" + assert service.deletes == 1 + assert create_times == [0.0, 30.0] + + +def test_raises_when_the_image_stays_missing_after_two_creates(): + service = FakeImageService([MISSING], after_create=lambda _creates: [MISSING]) + + run_failing(service, "after 2 CreateMicrovmImage calls") + + assert service.creates == 2 + + +def test_raises_when_the_build_does_not_finish_in_time(): + service = FakeImageService(["CREATING"]) + + clock = run_failing(service, "still CREATING after 20 minutes") + + assert clock.time > 20 * 60 + + +def test_polls_every_ten_seconds(): + service = FakeImageService(["CREATING", "CREATING", "CREATED"]) + + _, _, clock = run(service) + + assert service.gets == 3 + assert clock.time == 20 + + +# BotoImageService + + +def client_error(code, message="error"): + return ClientError({"Error": {"Code": code, "Message": message}}, "Operation") + + +class FakeMicrovmsClient: + """A lambda-microvms client whose methods run ``handler`` with the call.""" + + def __init__(self, handler): + self.handler = handler + self.calls = [] + + def _call(self, operation, kwargs): + self.calls.append((operation, kwargs)) + return self.handler(operation) + + def get_microvm_image(self, **kwargs): + return self._call("GetMicrovmImage", kwargs) + + def create_microvm_image(self, **kwargs): + return self._call("CreateMicrovmImage", kwargs) + + def delete_microvm_image(self, **kwargs): + return self._call("DeleteMicrovmImage", kwargs) + + +class FakeS3Client: + def __init__(self, error=None): + self.error = error + self.puts = [] + + def put_object(self, **kwargs): + self.puts.append(kwargs) + if self.error: + raise self.error + return {} + + +def raise_(error): + raise error + + +SETTINGS = ensure_microvm_image.image_settings("us-west-2", "123456789012") +IMAGE_ARN = f"arn:aws:lambda:us-west-2:123456789012:microvm-image:{NAME}" + + +def boto_service(microvms_handler, s3_error=None): + microvms = FakeMicrovmsClient(microvms_handler) + s3 = FakeS3Client(s3_error) + service = ensure_microvm_image.BotoImageService( + microvms=microvms, + s3=s3, + name=NAME, + image_arn=IMAGE_ARN, + bucket="test-bucket", + key=f"examples/{NAME}.zip", + zip_bytes=b"zip", + settings=SETTINGS, + log=lambda _message: None, + ) + return service, microvms, s3 + + +def test_get_returns_none_for_a_missing_image(): + service, microvms, _ = boto_service( + lambda _op: raise_(client_error("ResourceNotFoundException")) + ) + + assert service.get() is None + assert microvms.calls == [("GetMicrovmImage", {"imageIdentifier": IMAGE_ARN})] + + +def test_get_rethrows_other_errors(): + service, _, _ = boto_service( + lambda _op: raise_(client_error("AccessDeniedException")) + ) + + with pytest.raises(ClientError): + service.get() + + +def test_create_uploads_the_zip_and_sends_the_request(): + service, microvms, s3 = boto_service(lambda _op: {}) + + service.create() + + assert s3.puts == [ + {"Bucket": "test-bucket", "Key": f"examples/{NAME}.zip", "Body": b"zip"} + ] + assert microvms.calls == [ + ( + "CreateMicrovmImage", + { + "name": NAME, + "description": ( + "The MicroVM image that the durable execution SDK examples run" + ), + "codeArtifact": {"uri": f"s3://test-bucket/examples/{NAME}.zip"}, + "baseImageArn": "arn:aws:lambda:us-west-2:aws:microvm-image:al2023-1", + "buildRoleArn": "arn:aws:iam::123456789012:role/microvm-image-build", + "egressNetworkConnectors": [ + "arn:aws:lambda:us-west-2:aws:network-connector:" + "aws-network-connector:INTERNET_EGRESS" + ], + "hooks": { + "port": 8080, + "microvmHooks": { + "run": "ENABLED", + "runTimeoutInSeconds": 10, + "resume": "ENABLED", + "resumeTimeoutInSeconds": 10, + "suspend": "ENABLED", + "suspendTimeoutInSeconds": 10, + "terminate": "ENABLED", + "terminateTimeoutInSeconds": 10, + }, + "microvmImageHooks": { + "ready": "ENABLED", + "readyTimeoutInSeconds": 60, + }, + }, + }, + ) + ] + + +def test_create_request_matches_the_botocore_model(): + import botocore.session + + model = botocore.session.get_session().get_service_model("lambda-microvms") + request = ensure_microvm_image.create_image_request( + NAME, "test-bucket", "key.zip", SETTINGS + ) + + # The parameter validator raises on an unknown or mistyped member. + from botocore.validate import ParamValidator + + report = ParamValidator().validate( + request, model.operation_model("CreateMicrovmImage").input_shape + ) + assert not report.has_errors(), report.generate_report() + + +def test_carries_on_when_another_run_deleted_the_image_first(): + service, microvms, _ = boto_service( + lambda _op: raise_(client_error("ResourceNotFoundException")) + ) + + service.delete() + + assert [call[0] for call in microvms.calls] == [ + "DeleteMicrovmImage", + "GetMicrovmImage", + ] + + +def test_carries_on_when_another_run_is_deleting_the_image(): + def handler(operation): + if operation == "DeleteMicrovmImage": + raise client_error("ConflictException", "delete in progress") + return {"state": "DELETING"} + + service, _, _ = boto_service(handler) + + service.delete() + + +def test_rethrows_a_delete_error_when_the_failed_image_is_still_there(): + def handler(operation): + if operation == "DeleteMicrovmImage": + raise client_error("AccessDeniedException", "access denied") + return {"state": "CREATE_FAILED"} + + service, _, _ = boto_service(handler) + + with pytest.raises(ClientError, match="access denied"): + service.delete() + + +def test_treats_a_duplicate_name_as_a_build_by_another_run(): + def handler(operation): + if operation == "CreateMicrovmImage": + raise client_error( + "ValidationException", + f"A MicroVM image with the name '{NAME}' already exists in " + "this account", + ) + raise client_error("ResourceNotFoundException") + + service, _, _ = boto_service(handler) + + service.create() + + +def test_treats_a_create_error_as_a_build_by_another_run_when_the_image_exists(): + def handler(operation): + if operation == "CreateMicrovmImage": + raise client_error("ConflictException", "conflict") + return {"state": "CREATING"} + + service, _, _ = boto_service(handler) + + service.create() + + +def test_rethrows_a_create_error_when_no_image_exists(): + def handler(operation): + if operation == "CreateMicrovmImage": + raise client_error("ServiceQuotaExceededException", "quota exceeded") + raise client_error("ResourceNotFoundException") + + service, _, _ = boto_service(handler) + + with pytest.raises(ClientError, match="quota exceeded"): + service.create() + + +def test_names_the_override_variables_when_the_bucket_is_missing(): + service, microvms, _ = boto_service( + lambda _op: {}, s3_error=client_error("NoSuchBucket") + ) + + with pytest.raises( + RuntimeError, match=r"test-bucket does not exist.*MICROVM_ARTIFACT_BUCKET" + ): + service.create() + assert microvms.calls == [] + + +# Settings, endpoint warning + + +def test_settings_use_the_build_role_override(): + settings = ensure_microvm_image.image_settings( + "us-east-1", "111122223333", "arn:aws:iam::111122223333:role/other" + ) + + assert settings["buildRoleArn"] == "arn:aws:iam::111122223333:role/other" + assert settings["baseImageArn"] == ( + "arn:aws:lambda:us-east-1:aws:microvm-image:al2023-1" + ) + + +@pytest.mark.parametrize( + ("endpoint", "warns"), + [ + (None, False), + ("", False), + ("https://lambda.us-west-2.amazonaws.com", False), + ("https://lambda.us-west-2.amazonaws.com/", False), + ("https://beta.lambda.example.com", True), + ], +) +def test_warns_only_on_a_custom_lambda_endpoint(endpoint, warns, capsys): + environ = {} if endpoint is None else {"LAMBDA_ENDPOINT": endpoint} + + assert ( + ensure_microvm_image.warn_on_custom_lambda_endpoint("us-west-2", environ) + is warns + ) + stderr = capsys.readouterr().err + assert ("WARNING" in stderr) is warns + # CI passes the endpoint as a secret. So the warning omits it. + assert "example.com" not in stderr + + +# Content hash and zip + + +@pytest.fixture +def inputs(tmp_path): + """Copies of the real worker and image inputs, which a test can change.""" + worker_dir = tmp_path / "worker" + shutil.copytree( + ensure_microvm_image.WORKER_DIR / "src", + worker_dir / "src", + ignore=shutil.ignore_patterns("__pycache__", "*.pyc"), + ) + shutil.copy(ensure_microvm_image.WORKER_DIR / "pyproject.toml", worker_dir) + image_dir = tmp_path / "image" + shutil.copytree(ensure_microvm_image.IMAGE_DIR, image_dir) + return worker_dir, image_dir + + +def hash_of(worker_dir, image_dir, settings=SETTINGS): + return ensure_microvm_image.content_hash( + ensure_microvm_image.hash_inputs(worker_dir, image_dir, settings) + ) + + +def fake_wheel_builder(counter): + """Build a fake wheel whose bytes differ on each build.""" + + def build(worker_dir, wheel_dir): + counter.append(worker_dir) + wheel = wheel_dir / "worker-0.1.0-py3-none-any.whl" + wheel.write_bytes(f"build {len(counter)}".encode()) + return wheel + + return build + + +def test_two_builds_of_the_same_inputs_give_the_same_hash(inputs, tmp_path): + worker_dir, image_dir = inputs + builds = [] + + first, second = ( + ensure_microvm_image.build_artifact( + SETTINGS, + worker_dir=worker_dir, + image_dir=image_dir, + out_dir=tmp_path / "out", + build_wheel=fake_wheel_builder(builds), + ) + for _ in range(2) + ) + + # The wheel bytes differ between the builds. The hash covers the inputs, + # not the wheel. So the hash is the same. + assert len(builds) == 2 + assert first.zip_bytes != second.zip_bytes + assert first.content_hash == second.content_hash + assert len(first.content_hash) == 12 + int(first.content_hash, 16) + + +def test_hash_ignores_bytecode(inputs): + worker_dir, image_dir = inputs + before = hash_of(worker_dir, image_dir) + package_dir = next((worker_dir / "src").iterdir()) + (package_dir / "__pycache__").mkdir() + (package_dir / "__pycache__" / "worker.cpython-313.pyc").write_bytes(b"x") + (package_dir / "stray.pyc").write_bytes(b"x") + + assert hash_of(worker_dir, image_dir) == before + + +def _worker_source(worker_dir): + return next((worker_dir / "src").rglob("worker.py")) + + +@pytest.mark.parametrize( + "change", + [ + "worker source", + "new worker file", + "worker pyproject", + "app", + "dockerfile", + "constraints", + ], +) +def test_changing_an_input_changes_the_hash(inputs, change): + worker_dir, image_dir = inputs + before = hash_of(worker_dir, image_dir) + targets = { + "worker source": _worker_source(worker_dir), + "worker pyproject": worker_dir / "pyproject.toml", + "app": image_dir / "app.py", + "dockerfile": image_dir / "Dockerfile", + "constraints": image_dir / "constraints.txt", + } + if change == "new worker file": + (_worker_source(worker_dir).parent / "extra.py").write_text("x = 1\n") + else: + path = targets[change] + path.write_bytes(path.read_bytes() + b"\n# changed\n") + + assert hash_of(worker_dir, image_dir) != before + + +def test_changing_a_setting_changes_the_hash(inputs): + worker_dir, image_dir = inputs + other_role = ensure_microvm_image.image_settings( + "us-west-2", "123456789012", "arn:aws:iam::123456789012:role/other" + ) + other_region = ensure_microvm_image.image_settings("us-east-1", "123456789012") + + hashes = { + hash_of(worker_dir, image_dir, settings) + for settings in (SETTINGS, other_role, other_region) + } + + assert len(hashes) == 3 + + +def test_hash_does_not_depend_on_settings_key_order(inputs): + worker_dir, image_dir = inputs + reordered = json.loads(json.dumps(SETTINGS)) + reordered = dict(reversed(list(reordered.items()))) + + assert hash_of(worker_dir, image_dir, reordered) == hash_of(worker_dir, image_dir) + + +def test_hash_separates_entry_names_from_bytes(): + assert ensure_microvm_image.content_hash( + [("a", b"bc")] + ) != ensure_microvm_image.content_hash([("ab", b"c")]) + + +def test_zip_holds_the_dockerfile_at_its_root(inputs, tmp_path): + worker_dir, image_dir = inputs + + artifact = ensure_microvm_image.build_artifact( + SETTINGS, + worker_dir=worker_dir, + image_dir=image_dir, + out_dir=tmp_path / "out", + build_wheel=fake_wheel_builder([]), + ) + + with zipfile.ZipFile(artifact.zip_path) as archive: + assert sorted(archive.namelist()) == [ + "Dockerfile", + "app.py", + "constraints.txt", + "wheels/worker-0.1.0-py3-none-any.whl", + ] + assert archive.read("Dockerfile") == (image_dir / "Dockerfile").read_bytes() + + +# main + + +def test_main_writes_the_image_and_role_arns(monkeypatch, tmp_path): + monkeypatch.setenv("AWS_REGION", "us-west-2") + monkeypatch.setenv("TEST_ACCOUNT_ID", "123456789012") + monkeypatch.setenv("MICROVM_EXECUTION_ROLE_ARN", "") + monkeypatch.delenv("LAMBDA_ENDPOINT", raising=False) + monkeypatch.setattr( + ensure_microvm_image, + "build_artifact", + lambda settings: ensure_microvm_image.ImageArtifact( + zip_path=tmp_path / "z.zip", zip_bytes=b"zip", content_hash="0123456789ab" + ), + ) + created = [] + + def fake_client(service_name, region_name): + assert region_name == "us-west-2" + return service_name + + monkeypatch.setattr(ensure_microvm_image.boto3, "client", fake_client) + + def fake_ensure_image(service, name): + created.append((service.bucket, service.key, service.image_arn, name)) + return "1" + + monkeypatch.setattr(ensure_microvm_image, "ensure_image", fake_ensure_image) + output = tmp_path / "out" / "image.json" + + assert ensure_microvm_image.main(["--output", str(output)]) == 0 + + name = "sdk-python-examples-0123456789ab" + image_arn = f"arn:aws:lambda:us-west-2:123456789012:microvm-image:{name}" + assert created == [ + ( + "microvm-image-artifacts-123456789012-us-west-2", + f"examples/{name}.zip", + image_arn, + name, + ) + ] + # An empty override counts as unset. + assert json.loads(output.read_text()) == { + "imageArn": image_arn, + "executionRoleArn": "arn:aws:iam::123456789012:role/microvm-execution", + } + + +OVERRIDES = [ + "MICROVM_ARTIFACT_BUCKET", + "MICROVM_IMAGE_BUILD_ROLE_ARN", + "MICROVM_EXECUTION_ROLE_ARN", +] + + +@pytest.mark.parametrize("empty", OVERRIDES) +def test_main_treats_an_empty_override_as_unset(empty, monkeypatch, tmp_path): + # GitHub passes an unset secret to a step as an empty string. So main() + # must use the default resource name for an empty override. + monkeypatch.setenv("AWS_REGION", "us-west-2") + monkeypatch.setenv("TEST_ACCOUNT_ID", "123456789012") + monkeypatch.delenv("LAMBDA_ENDPOINT", raising=False) + for variable in OVERRIDES: + monkeypatch.delenv(variable, raising=False) + monkeypatch.setenv(empty, "") + monkeypatch.setattr( + ensure_microvm_image, + "build_artifact", + lambda settings: ensure_microvm_image.ImageArtifact( + zip_path=tmp_path / "z.zip", zip_bytes=b"zip", content_hash="0123456789ab" + ), + ) + monkeypatch.setattr( + ensure_microvm_image.boto3, "client", lambda service_name, region_name: None + ) + services = [] + monkeypatch.setattr( + ensure_microvm_image, + "ensure_image", + lambda service, name: services.append(service), + ) + output = tmp_path / "image.json" + + assert ensure_microvm_image.main(["--output", str(output)]) == 0 + + [service] = services + assert service.bucket == "microvm-image-artifacts-123456789012-us-west-2" + assert service.settings["buildRoleArn"] == ( + "arn:aws:iam::123456789012:role/microvm-image-build" + ) + assert json.loads(output.read_text())["executionRoleArn"] == ( + "arn:aws:iam::123456789012:role/microvm-execution" + ) + + +def test_main_requires_the_account(monkeypatch, tmp_path): + monkeypatch.setenv("AWS_REGION", "us-west-2") + monkeypatch.delenv("TEST_ACCOUNT_ID", raising=False) + + with pytest.raises(SystemExit, match="TEST_ACCOUNT_ID is not set"): + ensure_microvm_image.main(["--output", str(tmp_path / "x.json")]) diff --git a/packages/aws-durable-execution-sdk-python-examples/test/test_generate_function_name_map.py b/packages/aws-durable-execution-sdk-python-examples/test/test_generate_function_name_map.py new file mode 100644 index 000000000..b9d8153b1 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-examples/test/test_generate_function_name_map.py @@ -0,0 +1,57 @@ +"""Tests for the cloud test function-name map.""" + +import importlib.util +import sys +from pathlib import Path + + +SCRIPTS_DIR = Path(__file__).resolve().parents[1] / "scripts" +# The script imports generate_sam_template as a top-level module. +if str(SCRIPTS_DIR) not in sys.path: + sys.path.insert(0, str(SCRIPTS_DIR)) + +SPEC = importlib.util.spec_from_file_location( + "generate_function_name_map", SCRIPTS_DIR / "generate_function_name_map.py" +) +assert SPEC is not None +assert SPEC.loader is not None +generate_function_name_map = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(generate_function_name_map) + + +EXAMPLES = [ + {"name": "Hello World", "handler": "hello_world.handler"}, + { + "name": "MicroVM Run Job", + "handler": "microvm_run_job.handler", + "usesMicrovm": True, + }, +] + + +def test_map_leaves_out_microvm_examples_by_default(): + assert generate_function_name_map.build_function_name_map(EXAMPLES, "Py313-") == { + "Hello World": "Py313-HelloWorld:$LATEST" + } + + +def test_map_includes_microvm_examples_on_request(): + assert generate_function_name_map.build_function_name_map( + EXAMPLES, "Py313-", include_microvm=True + ) == { + "Hello World": "Py313-HelloWorld:$LATEST", + "MicroVM Run Job": "Py313-MicrovmRunJob:$LATEST", + } + + +def test_map_covers_every_catalog_example_without_microvm(): + catalog = generate_function_name_map.load_catalog()["examples"] + expected = { + example["name"] + for example in catalog + if not generate_function_name_map.uses_microvm(example) + } + + assert set(generate_function_name_map.generate_function_name_map("P-")) == ( + expected + ) diff --git a/packages/aws-durable-execution-sdk-python-examples/test/test_generate_sam_template.py b/packages/aws-durable-execution-sdk-python-examples/test/test_generate_sam_template.py index 5f8ef3de1..05016e875 100644 --- a/packages/aws-durable-execution-sdk-python-examples/test/test_generate_sam_template.py +++ b/packages/aws-durable-execution-sdk-python-examples/test/test_generate_sam_template.py @@ -54,3 +54,90 @@ def test_build_template_adds_default_json_logging_to_durable_functions(): assert resources["PlainLambda"]["Properties"]["LoggingConfig"] == { "LogGroup": {"Ref": "PlainLambdaLogGroup"}, } + + +MICROVM_EXAMPLE = { + "name": "MicroVM Example", + "handler": "microvm_example.handler", + "description": "A MicroVM example", + "usesMicrovm": True, + "durableConfig": {"RetentionPeriodInDays": 7, "ExecutionTimeout": 900}, +} +PLAIN_EXAMPLE = { + "name": "Hello World", + "handler": "hello_world.handler", + "description": "A durable example", + "durableConfig": {"RetentionPeriodInDays": 7, "ExecutionTimeout": 300}, +} + + +def test_build_template_declares_microvm_parameters_with_empty_defaults(): + template = generate_sam_template.build_template([PLAIN_EXAMPLE]) + + parameters = template["Parameters"] + assert parameters["MicrovmImageArn"]["Default"] == "" + assert parameters["MicrovmImageArn"]["Type"] == "String" + assert parameters["MicrovmExecutionRoleArn"]["Default"] == "" + assert parameters["MicrovmExecutionRoleArn"]["Type"] == "String" + + +def test_build_template_omits_condition_without_microvm_example(): + template = generate_sam_template.build_template([PLAIN_EXAMPLE]) + + assert "Conditions" not in template + + +def test_build_template_requires_image_and_role_for_microvm_condition(): + template = generate_sam_template.build_template([PLAIN_EXAMPLE, MICROVM_EXAMPLE]) + + assert template["Conditions"] == { + "HasMicrovmImage": { + "Fn::And": [ + {"Fn::Not": [{"Fn::Equals": [{"Ref": "MicrovmImageArn"}, ""]}]}, + {"Fn::Not": [{"Fn::Equals": [{"Ref": "MicrovmExecutionRoleArn"}, ""]}]}, + ] + } + } + + +def test_build_template_sets_condition_on_microvm_function_and_log_group(): + template = generate_sam_template.build_template([MICROVM_EXAMPLE]) + + resources = template["Resources"] + assert set(resources) == {"MicrovmExample", "MicrovmExampleLogGroup"} + assert resources["MicrovmExample"]["Condition"] == "HasMicrovmImage" + assert resources["MicrovmExampleLogGroup"]["Condition"] == "HasMicrovmImage" + + +def test_build_template_sets_microvm_environment_variables(): + template = generate_sam_template.build_template( + [{**MICROVM_EXAMPLE, "environment": {"EXTRA": "value"}}] + ) + + assert template["Resources"]["MicrovmExample"]["Properties"]["Environment"] == { + "Variables": { + "EXTRA": "value", + "MICROVM_IMAGE_ARN": {"Ref": "MicrovmImageArn"}, + "MICROVM_EXECUTION_ROLE_ARN": {"Ref": "MicrovmExecutionRoleArn"}, + } + } + + +def test_build_template_leaves_example_without_flag_unconditional(): + template = generate_sam_template.build_template( + [{**PLAIN_EXAMPLE, "usesMicrovm": False}] + ) + + resources = template["Resources"] + assert "Condition" not in resources["HelloWorld"] + assert "Condition" not in resources["HelloWorldLogGroup"] + assert "Environment" not in resources["HelloWorld"]["Properties"] + assert "Timeout" not in resources["HelloWorld"]["Properties"] + assert "Conditions" not in template + + +def test_build_template_sets_timeout_from_catalog(): + template = generate_sam_template.build_template([{**PLAIN_EXAMPLE, "timeout": 120}]) + + assert template["Resources"]["HelloWorld"]["Properties"]["Timeout"] == 120 + assert template["Globals"]["Function"]["Timeout"] == 60 diff --git a/packages/aws-durable-execution-sdk-python-examples/test/test_microvm_image_app.py b/packages/aws-durable-execution-sdk-python-examples/test/test_microvm_image_app.py new file mode 100644 index 000000000..e07dec3a1 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-examples/test/test_microvm_image_app.py @@ -0,0 +1,69 @@ +"""Tests for the handlers of microvm-image/app.py. + +The app imports the MicroVM worker. The examples package does not depend on +the worker. So an environment without the worker skips these tests. The +dev-examples environment is one such environment. +""" + +import hashlib +import importlib.util +from pathlib import Path +from types import SimpleNamespace + +import pytest + + +pytest.importorskip("aws_durable_execution_sdk_python_microvm_worker") + +APP_PATH = Path(__file__).resolve().parents[1] / "microvm-image" / "app.py" +SPEC = importlib.util.spec_from_file_location("microvm_image_app", APP_PATH) +assert SPEC is not None +assert SPEC.loader is not None +app = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(app) + +CONTEXT = SimpleNamespace(microvm_id="mv-1") + + +def test_count_words_returns_the_count_hash_and_microvm_id(): + text = "the quick brown fox\njumps" + + assert app.count_words({"text": text}, CONTEXT) == { + "wordCount": 5, + "sha256": hashlib.sha256(text.encode()).hexdigest(), + "microvmId": "mv-1", + } + + +def test_read_returns_what_write_wrote(monkeypatch, tmp_path): + monkeypatch.setattr(app, "SESSION_DIR", tmp_path / "session") + + assert app.write_value({"key": "k1", "value": "v1"}, CONTEXT) == { + "microvmId": "mv-1" + } + assert app.read_value({"key": "k1"}, CONTEXT) == { + "microvmId": "mv-1", + "value": "v1", + } + + +@pytest.mark.parametrize("key", ["", ".", "..", "../x", "a/b", "/etc/passwd"]) +def test_rejects_a_key_that_is_not_a_plain_file_name(monkeypatch, tmp_path, key): + monkeypatch.setattr(app, "SESSION_DIR", tmp_path / "session") + + with pytest.raises(ValueError, match="invalid key"): + app.write_value({"key": key, "value": "v"}, CONTEXT) + + +def test_the_worker_accepts_the_handlers(): + # The worker rejects an async handler when it starts. This builds the + # listener with the app's handlers, without a server. + from aws_durable_execution_sdk_python_microvm_worker import ( + MicrovmWorkerListener, + ) + + listener = MicrovmWorkerListener.create( + handler=app.count_words, + routes={"/write": app.write_value, "/read": app.read_value}, + ) + listener.close() diff --git a/pyproject.toml b/pyproject.toml index f4e735bd9..621244358 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -155,9 +155,18 @@ typecheck = "mypy --install-types --non-interactive packages/aws-durable-executi js-examples = "python .github/scripts/js_examples/run.py {args}" [tool.hatch.envs.dev-examples] +# The MicroVM examples import the extras. test-pypi-examples is meant to run +# the examples against the packages on PyPI. The extras are not published +# there. So the examples package does not depend on the extras. The +# environments that run the examples from this checkout add the extras as a +# workspace member instead. test-pypi-examples sets no workspace. So it +# inherits the default members and installs the extras from this checkout +# too. The MicroVM test modules call pytest.importorskip. So an environment +# without the extras skips them. workspace.members = [ "packages/aws-durable-execution-sdk-python", "packages/aws-durable-execution-sdk-python-examples", + "packages/aws-durable-execution-sdk-python-extras", "packages/aws-durable-execution-sdk-python-otel", "packages/aws-durable-execution-sdk-python-testing", ] @@ -234,13 +243,15 @@ test = "pytest packages/aws-durable-execution-sdk-python-examples/test {args}" workspace.members = [ "packages/aws-durable-execution-sdk-python", "packages/aws-durable-execution-sdk-python-examples", + "packages/aws-durable-execution-sdk-python-extras", "packages/aws-durable-execution-sdk-python-otel", ] [tool.hatch.envs.examples.scripts] generate-sam-template = "python packages/aws-durable-execution-sdk-python-examples/scripts/generate_sam_template.py" build = "python packages/aws-durable-execution-sdk-python-examples/scripts/build_examples.py" -clean = "rm -rf packages/aws-durable-execution-sdk-python-examples/build packages/aws-durable-execution-sdk-python-examples/template.generated.json .aws-sam" +ensure-microvm-image = "python packages/aws-durable-execution-sdk-python-examples/scripts/ensure_microvm_image.py {args}" +clean = "rm -rf packages/aws-durable-execution-sdk-python-examples/build packages/aws-durable-execution-sdk-python-examples/template.generated.json packages/aws-durable-execution-sdk-python-examples/.microvm-image-build .aws-sam" [tool.ruff] line-length = 88