From 04a3596de4842ecf1241694d9725980334a6d44c Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Wed, 2 Sep 2026 22:15:25 +0000 Subject: [PATCH] docs: add payload offloader examples and E2E coverage --- .github/workflows/e2e-tests.yml | 55 +++++- .gitignore | 1 + README.md | 5 + docs/advanced/configuration.md | 92 +++++++++ docs/wire-formats/payload-offloader.md | 156 +++++++++++++++ examples/README.md | 16 ++ examples/generate-template.py | 186 +++++++++++++++++- .../durable/examples/ExampleTemplate.java | 2 + .../FileSystemPayloadOffloaderExample.java | 67 +++++++ .../examples/CloudBasedIntegrationTest.java | 31 +++ examples/test_generate_template.py | 52 +++++ 11 files changed, 655 insertions(+), 8 deletions(-) create mode 100644 docs/wire-formats/payload-offloader.md create mode 100644 examples/src/main/java/software/amazon/lambda/durable/examples/general/FileSystemPayloadOffloaderExample.java create mode 100644 examples/test_generate_template.py diff --git a/.github/workflows/e2e-tests.yml b/.github/workflows/e2e-tests.yml index ae25b371a..b4d660edc 100644 --- a/.github/workflows/e2e-tests.yml +++ b/.github/workflows/e2e-tests.yml @@ -32,10 +32,44 @@ permissions: id-token: write # This is required for requesting the JWT contents: read # This is required for actions/checkout +env: + AWS_REGION: us-west-2 + FILESYSTEM_INFRASTRUCTURE_STACK_NAME: JavaSDKFileSystemPayloadE2EInfrastructureStack + jobs: + filesystem-infrastructure: + runs-on: ubuntu-latest + steps: + - name: Checkout repository + uses: actions/checkout@v7 + - name: Configure AWS credentials + uses: aws-actions/configure-aws-credentials@e6de054238d6b7531b4efff3b6587d9aade6a06c # v6.2.3 + with: + role-to-assume: "${{ secrets.TEST_ROLE_ARN }}" + role-session-name: java-language-sdk-test-infrastructure + aws-region: ${{ env.AWS_REGION }} + allowed-account-ids: ${{ secrets.TEST_ACCOUNT_ID }} + - name: Test SAM template generator + run: python3 -m unittest test_generate_template.py + working-directory: ./examples + - name: Generate persistent filesystem payload E2E infrastructure template + run: | + python3 generate-template.py \ + --file-system-infrastructure-only \ + --output filesystem-infrastructure-template.yaml + working-directory: ./examples + - name: Ensure persistent filesystem payload E2E infrastructure + run: | + aws cloudformation deploy \ + --template-file filesystem-infrastructure-template.yaml \ + --stack-name ${{ env.FILESYSTEM_INFRASTRUCTURE_STACK_NAME }} \ + --no-fail-on-empty-changeset \ + --tags Purpose=JavaSDKFileSystemPayloadE2E + working-directory: ./examples + e2e-tests: + needs: filesystem-infrastructure env: - AWS_REGION: us-west-2 E2E_TEST_PARALLELISM: 4 runs-on: ubuntu-latest strategy: @@ -71,6 +105,9 @@ jobs: - name: Generate SAM template run: python3 generate-template.py working-directory: ./examples + - name: Generate filesystem payload E2E SAM template + run: python3 generate-template.py --file-system-only --output filesystem-template.yaml + working-directory: ./examples - name: sam build env: MAVEN_OPTS: -DskipTests=true -Dmaven.test.skip=true @@ -78,6 +115,14 @@ jobs: sam build --debug --parameter-overrides \ 'ParameterKey=Architecture,ParameterValue=x86_64 ParameterKey=JavaVersion,ParameterValue=java${{ matrix.java }} ParameterKey=FunctionNamePrefix,ParameterValue=Java${{ matrix.java }}- ParameterKey=RoleArn,ParameterValue=${{ secrets.TEST_LAMBDA_EXECUTION_ROLE_ARN }}' working-directory: ./examples + - name: sam build filesystem payload E2E stack + env: + MAVEN_OPTS: -DskipTests=true -Dmaven.test.skip=true + run: | + sam build --debug --template-file filesystem-template.yaml --build-dir .aws-sam-filesystem \ + --parameter-overrides \ + 'ParameterKey=Architecture,ParameterValue=x86_64 ParameterKey=JavaVersion,ParameterValue=java${{ matrix.java }} ParameterKey=FunctionNamePrefix,ParameterValue=Java${{ matrix.java }}- ParameterKey=RoleArn,ParameterValue=${{ secrets.TEST_LAMBDA_EXECUTION_ROLE_ARN }} ParameterKey=FileSystemInfrastructureStackName,ParameterValue=${{ env.FILESYSTEM_INFRASTRUCTURE_STACK_NAME }}' + working-directory: ./examples - name: Clean up unmanaged Lambda log groups run: | # TODO: Remove this one-time migration cleanup after existing e2e stacks adopt managed log groups. @@ -89,12 +134,20 @@ jobs: --resolve-image-repos --resolve-s3 --parameter-overrides \ 'ParameterKey=Architecture,ParameterValue=x86_64 ParameterKey=JavaVersion,ParameterValue=java${{ matrix.java }} ParameterKey=FunctionNamePrefix,ParameterValue=Java${{ matrix.java }}- ParameterKey=RoleArn,ParameterValue=${{ secrets.TEST_LAMBDA_EXECUTION_ROLE_ARN }}' working-directory: ./examples + - name: sam deploy filesystem payload E2E stack + run: | + sam deploy --template-file .aws-sam-filesystem/template.yaml \ + --stack-name Java${{ matrix.java }}-JavaSDKFileSystemPayloadE2EStack \ + --resolve-s3 --parameter-overrides \ + 'ParameterKey=Architecture,ParameterValue=x86_64 ParameterKey=JavaVersion,ParameterValue=java${{ matrix.java }} ParameterKey=FunctionNamePrefix,ParameterValue=Java${{ matrix.java }}- ParameterKey=RoleArn,ParameterValue=${{ secrets.TEST_LAMBDA_EXECUTION_ROLE_ARN }} ParameterKey=FileSystemInfrastructureStackName,ParameterValue=${{ env.FILESYSTEM_INFRASTRUCTURE_STACK_NAME }}' + working-directory: ./examples - name: Record E2E log start time run: echo "E2E_LOG_START_TIME_MS=$(date +%s%3N)" >> "$GITHUB_ENV" - name: Cloud Based Integration Tests run: | mvn clean test -B \ -Dtest.cloud.enabled=true \ + -Dtest.filesystem.enabled=true \ -Dtest.aws.account='${{ secrets.TEST_ACCOUNT_ID }}' \ -Dtest=CloudBasedIntegrationTest \ -Dtest.function.name.prefix='Java${{ matrix.java }}-' \ diff --git a/.gitignore b/.gitignore index 378d24717..1addbb95d 100644 --- a/.gitignore +++ b/.gitignore @@ -42,6 +42,7 @@ __pycache__/ # SAM .aws-sam/ examples/template.yaml +examples/filesystem-template.yaml samconfig.toml samconfig.toml.bak diff --git a/README.md b/README.md index 766a71b02..f6fe20cdc 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,7 @@ Build resilient, long-running AWS Lambda functions that automatically checkpoint - **Replay Safety** – Functions deterministically resume from checkpoints after interruptions - **Type Safety** – Full generic type support for step results - **Data-Driven Concurrency** – Apply a function across a collection with `map()`, with per-item error isolation and configurable completion criteria +- **Payload Offloading** – Keep large serialized payloads in durable external storage while checkpoints retain compact references ## How It Works @@ -50,6 +51,9 @@ Your durable function extends `DurableHandler` and implements `handleReque ``` +Filesystem payload offloading is included in the core SDK artifact under +`software.amazon.lambda.durable.offload.filesystem`. + ### Your First Durable Function ```java @@ -111,6 +115,7 @@ See [Deploy Lambda durable functions with Infrastructure as Code](https://docs.a **Advanced Topics** - [Configuration](docs/advanced/configuration.md) - Customize SDK behaviour +- [Payload Offloading](docs/advanced/configuration.md#payload-offloading) - Store serialized payloads outside checkpoints - [Error Handling](docs/advanced/error-handling.md) - SDK exceptions for handling failures - [Logging](docs/advanced/logging.md) - How to use DurableLogger - [Migrating from 1.x to 2.x](docs/migration-1.x-to-2.x.md) - Upgrade guide for breaking changes since `v1.2.1` diff --git a/docs/advanced/configuration.md b/docs/advanced/configuration.md index bc8bf89d0..390c7a26c 100644 --- a/docs/advanced/configuration.md +++ b/docs/advanced/configuration.md @@ -17,7 +17,9 @@ public class OrderProcessor extends DurableHandler { return DurableConfig.builder() .withLambdaClientBuilder(lambdaClientBuilder) .withSerDes(new MyCustomSerDes()) // Custom serialization + .withPayloadOffloader(myPayloadOffloader) // Optional external payload storage .withExecutorService(Executors.newFixedThreadPool(10)) // Custom thread pool + .withPayloadOffloadExecutorService(payloadIoExecutor) // Blocking payload I/O .withLoggerConfig(LoggerConfig.withReplayLogging()) // Enable replay logs .build(); } @@ -33,13 +35,103 @@ public class OrderProcessor extends DurableHandler { |-----------------------------|-----------------------------------------|-------------------------------| | `withLambdaClientBuilder()` | Custom AWS Lambda client | Auto-configured Lambda client | | `withSerDes()` | Serializer for step results | Jackson with default settings | +| `withPayloadOffloader()` | External storage for serialized user payloads | Disabled | | `withExecutorService()` | Thread pool for user-defined operations | Cached daemon thread pool | +| `withPayloadOffloadExecutorService()` | Thread pool for blocking payload storage I/O | Inline on the calling thread | | `withLoggerConfig()` | Logger behavior configuration | Suppress logs during replay | | `withPollingStrategy()` | Backend polling strategy | Exponential backoff: 1s base, 2x rate, FULL jitter, 10s max | | `withCheckpointDelay()` | How often the SDK checkpoints updates | `Duration.ofSeconds(0)` (as soon as possible) | The `withExecutorService()` option configures the thread pool used for running user-defined operations. Internal SDK coordination (checkpoint batching, polling) runs on an SDK-managed thread pool. +### Payload offloading + +`SerDes` remains responsible for converting objects to serialized text. A `PayloadOffloader` runs after serialization +and decides whether that text remains inline or is stored externally. On replay, the SDK resolves the stored reference +before passing the serialized text back to `SerDes`. + +Filesystem payload offloading is included in the core SDK artifact. + +Configure a durable shared mount: + +```java +import java.nio.file.Path; +import software.amazon.lambda.durable.offload.filesystem.FileSystemPathEncoding; +import software.amazon.lambda.durable.offload.filesystem.FileSystemPayloadOffloader; +import software.amazon.lambda.durable.offload.filesystem.PayloadOffloadMode; +import software.amazon.lambda.durable.offload.filesystem.PreviewConfig; +import software.amazon.lambda.durable.offload.filesystem.PreviewField; +import software.amazon.lambda.durable.offload.filesystem.PreviewMode; + +var offloader = FileSystemPayloadOffloader.builder(Path.of("/mnt/efs")) + .storageMode(PayloadOffloadMode.OVERFLOW) + .pathEncoding(FileSystemPathEncoding.HASH) + .previewConfig(PreviewConfig.builder(PreviewMode.EXCLUDE_ALL) + .include(PreviewField.anywhere("id")) + .mask(PreviewField.anywhere("email")) + .build()) + .build(); + +return DurableConfig.builder() + .withSerDes(new JacksonSerDes()) + .withPayloadOffloader(offloader) + .build(); +``` + +`ALWAYS` writes every serialized payload to an immutable file. `OVERFLOW` keeps payloads inline until they approach the +configured checkpoint-envelope limit. Every envelope records producer ownership and a SHA-256 digest; loads validate +the owner, path, filename, and content. The configured base path and all ancestors must already exist; payload files are +direct children of that directory. `URI` includes a bounded readable entity prefix plus an owner digest, while `HASH` +uses only the fixed-length SHA-256 owner digest. The filesystem provider must support `SecureDirectoryStream`. + +The global offloader applies to root output, checkpointed step/invoke/child/map/parallel results, +wait-for-condition state, and serialized exception data. Direct Lambda input and externally submitted callback results +remain ordinary SerDes data. Chained invoke request payloads also remain normal Lambda JSON by default so standard +Lambda targets do not need SDK envelope or shared-storage support. Compatible durable callers and targets can +explicitly opt in to offloaded invoke requests. Operation configuration can override the offloader: + +```java +var stepConfig = StepConfig.builder() + .payloadOffloader(otherOffloader) + .build(); + +var inlineStepConfig = StepConfig.builder() + .payloadOffloader(PayloadOffloader.disabled()) + .build(); +``` + +The same `payloadOffloader(...)` option is available on `InvokeConfig`, `RunInChildContextConfig`, `MapConfig`, +`ParallelConfig`, `ParallelBranchConfig`, and `WaitForConditionConfig`. + +Transient storage failures can be retried explicitly: + +```java +var retryingOffloader = new RetryPayloadOffloader( + offloader, + RetryStrategies.fixedDelay(3, Duration.ofSeconds(1))); +``` + +Payload I/O runs inline by default. Configure `withPayloadOffloadExecutorService(...)` when blocking I/O should use a +dedicated pool; it must not be the user-operation executor. Calls made from SDK-managed user-operation threads execute +inline even when this executor is configured, preventing deadlock when separate executor wrappers share one bounded +backing pool. + +The SDK uses a versioned checkpoint envelope and continues to read payloads written by older SDK versions as raw +serialized text. Within one Lambda invocation, resolved storage data and deserialized objects use bounded weak caches +and concurrent identical loads share one in-flight operation. Garbage collection or eviction can cause a later reload. + +> **Do not use Lambda `/tmp` for durable payloads.** It is local to one execution environment and might not exist on +> replay. Use a shared durable filesystem such as EFS. S3 Files can have delayed synchronization and recent writes can +> be lost if the runtime crashes before the mount flushes; use it only when that durability tradeoff is acceptable. + +The SDK does not delete offloaded files. Configure storage lifecycle and retention separately, and keep the mounted +path accessible to every function environment that may replay or consume the payload. Treat each stored file reference +as a capability: restrict access to the shared base path and protect checkpoint/history data containing references with +the same controls as the payload itself. + +The versioned envelope, ownership, digest, filesystem, and chained-invoke contracts are defined in +[Payload offloader wire formats](../wire-formats/payload-offloader.md). + ### Dynamic plugin loading Dynamic plugin loading is an opt-in alternative to registering plugins in application code. Put provider JARs on the application class path, then set `DURABLE_EXECUTION_PLUGINS` to an ordered, comma-separated list of provider names: diff --git a/docs/wire-formats/payload-offloader.md b/docs/wire-formats/payload-offloader.md new file mode 100644 index 000000000..890d0ec16 --- /dev/null +++ b/docs/wire-formats/payload-offloader.md @@ -0,0 +1,156 @@ +# Payload offloader wire formats + +## Status and scope + +This document defines version 1 of the Java SDK payload-offloader formats. Other SDKs can implement the same contract +when exchanging offloaded chained-invoke payloads or results. + +Three independent formats are involved: + +1. a chained-invoke request frame selects the target handler's persisted SerDes-plus-offloader path; +2. a chained-invoke output frame distinguishes SDK codec data from ordinary Lambda result/error data; and +3. an SDK payload envelope stores inline data or an external reference. + +## Chained-invoke source frame + +Version 1 has two exact forms: + +```text +__durable_execution_chained_invoke_payload:1:value: +__durable_execution_chained_invoke_payload:1:null +``` + +The `value:` form appends the opaque SDK payload without additional encoding. The `null` form represents a null request +while preserving the protocol handshake. + +- Producers MUST use the frame only when `InvokeConfig.usePayloadOffloaderForPayload(true)` is enabled. +- Consumers MUST interpret the frame only when + `DurableConfig.withPayloadOffloaderForChainedInvokePayloads(true)` is enabled. +- Acceptance MUST be disabled by default. +- A recognized unsupported or malformed frame MUST fail closed. +- Without both opt-ins, invoke requests retain the ordinary Lambda serialized-input contract. + +The frame is application-controlled data, not authenticated service metadata. Enable target acceptance only when every +principal allowed to invoke the function may select the persisted payload path. + +## Chained-invoke output frame + +A compatible durable target returns non-null results and error data in one of these exact version 1 forms: + +```text +__durable_execution_chained_invoke_output:1:codec: +__durable_execution_chained_invoke_output:1:raw: +``` + +- `codec:` means the payload is an SDK payload envelope and the caller MUST resolve it through the payload codec. +- `raw:` means the payload is ordinary serialized data and the caller MUST pass it directly to SerDes. +- Null results or error data remain null and are not framed. +- Callers MUST interpret output frames only for invokes that enabled + `InvokeConfig.usePayloadOffloaderForPayload(true)`. +- A recognized unsupported or malformed output frame MUST fail closed. + +## SDK payload envelope + +The envelope is the concatenation of: + +```text +@aws-durable-payload:v1: +``` + +The JSON representation uses these fields: + +| Field | Meaning | +|---|---| +| `mode` | `INLINE` or `REFERENCE` | +| `data` | Serialized payload for inline mode | +| `reference` | External storage reference for reference mode | +| `preview` | Optional informational preview metadata | +| `ownerDurableExecutionArn` | Producing durable execution | +| `ownerEntityId` | Producing payload entity | +| `payloadDigest` | Lowercase SHA-256 digest of the exact UTF-8 serialized payload | +| `producerContext` | Exact SDK payload context used to create the reference | +| `requiresLoad` | Whether the producer's payload offloader must restore the serialized data | + +Exactly one of `data` and `reference` is non-null. Version 1 SDK envelopes bind all custom-offloader results to +`producerContext`, populate the ownership and digest fields, and require the nested context to match them. A recognized +version 1 envelope missing this metadata MUST fail closed. Custom-offloader results set `requiresLoad` to `true`, +including `INLINE` results whose data may use an offloader-specific encoding. SDK-created inline envelopes that only +escape the reserved marker set it to `false`; consumers MUST use their inline `data` directly and MUST NOT pass it to a +configured offloader. Every version 1 envelope MUST include a boolean `requiresLoad` value; a missing or non-boolean +value is invalid. A `REFERENCE` envelope with `requiresLoad` set to `false` is also invalid. + +Legacy checkpoint strings without the versioned prefix are passed directly to the configured SerDes. + +## Ownership and cross-execution boundaries + +For ordinary checkpoint replay, `ownerDurableExecutionArn` and `ownerEntityId` MUST match the current payload context. +A different owner is accepted only where values legitimately cross executions: + +- the input of an invoked durable execution; or +- the result of a chained invoke operation. + +The declared producer remains authoritative for resolving the filesystem path. Consumers MUST NOT rewrite ownership to +the receiving execution. The SDK passes `producerContext`, rather than the consuming operation context, to +`PayloadOffloader.load` so context-keyed custom storage remains usable across executions. + +## Filesystem payloads + +The filesystem offloader stores the exact UTF-8 bytes produced by SerDes. + +### Path construction + +The configured base path and all of its ancestors MUST already exist. Payload files are direct children of that base +path. The owner prefix binds the producing execution and entity: + +- `URI` mode uses a bounded escaped entity prefix followed by the lowercase SHA-256 digest of + `ownerDurableExecutionArn + NUL + ownerEntityId`. +- `HASH` mode uses only that fixed-length SHA-256 owner digest. + +Payload filenames use: + +```text +--.payload +``` + +The unique suffix has no payload semantics. Each serialization publishes a new immutable file using `CREATE_NEW`; an +existing payload file is never replaced. + +### Validation + +Consumers MUST: + +- normalize the reference as a direct child of the configured base path; +- derive the expected filename prefix from the declared producer and payload digest; +- read without following symbolic links; +- keep secure directory handles open through file access; and +- verify the restored bytes against `payloadDigest`. + +The Java implementation traverses to the pre-provisioned base path through `SecureDirectoryStream` handles with +symbolic-link following disabled, then keeps the base-directory handle open through file access. This avoids pathname +races and temporary-file rename requirements while remaining compatible with EFS and S3 Files providers that support +normal `CREATE_NEW` writes. + +The external file reference is a capability, not an authentication credential. Producers and consumers MUST restrict +access to the shared base path and protect references with the same controls as the payload. + +### Structured previews + +`preview` is informational and is not used to restore the payload. The built-in preview builder supports: + +- include-all and exclude-all defaults; +- include, exclude, and mask selectors; +- exact path and field-name-at-any-depth matching; +- configurable mask text; and +- a serialized UTF-8 byte budget. + +Exact path selectors use dots as segment separators. Escape a literal dot as `\.` and a literal backslash as `\\`. +Field-name-at-any-depth selectors always treat their names literally. Preview metadata is snapshotted when the envelope +is created; arrays are normalized to immutable lists, and custom preview generators must return JSON-compatible values. + +The complete reference envelope, including preview data, must remain within the configured checkpoint-envelope limit. + +## Versioning + +The chained-invoke frame version and SDK payload-envelope version are independent. Incompatible changes require a new +version. Producers MUST NOT emit a version until the intended consumer supports it, and consumers MUST fail closed for +recognized unsupported versions. diff --git a/examples/README.md b/examples/README.md index c17750835..7077fe82c 100644 --- a/examples/README.md +++ b/examples/README.md @@ -77,6 +77,22 @@ mvn test -Dtest=CloudBasedIntegrationTest \ -Dtest.aws.region=us-east-1 ``` +The filesystem payload-offloader cloud test is disabled by default because it requires VPC and EFS infrastructure. +Create the persistent infrastructure stack once: + +```bash +python3 generate-template.py \ + --file-system-infrastructure-only \ + --output filesystem-infrastructure-template.yaml +aws cloudformation deploy \ + --template-file filesystem-infrastructure-template.yaml \ + --stack-name JavaSDKFileSystemPayloadE2EInfrastructureStack +``` + +Then generate, build, and deploy the filesystem Lambda stack with +`FileSystemInfrastructureStackName=JavaSDKFileSystemPayloadE2EInfrastructureStack`, and include +`-Dtest.filesystem.enabled=true` when running `CloudBasedIntegrationTest`. + ## Examples | Example | Description | diff --git a/examples/generate-template.py b/examples/generate-template.py index 2ffcd5d70..3309608db 100755 --- a/examples/generate-template.py +++ b/examples/generate-template.py @@ -21,6 +21,7 @@ class ExampleFunction: package_name: str suffix: str condition: str | None + file_system: bool @property def logical_id(self) -> str: @@ -56,21 +57,23 @@ def is_top_level_durable_handler(source: str, class_name: str) -> bool: return bool(match and "extends DurableHandler" in match.group("header")) -def read_template_condition(source: str, class_name: str) -> str | None: +def read_template_metadata(source: str, class_name: str) -> tuple[str | None, bool]: class_match = re.search(rf"public\s+(?:final\s+)?class\s+{class_name}\b", source) if not class_match: - return None + return None, False prefix = source[: class_match.start()] matches = list( re.finditer(rf"@(?:[A-Za-z_][\w.]*\.)?{TEMPLATE_ANNOTATION}\s*(?:\((?P.*?)\))?", prefix, re.DOTALL) ) if not matches: - return None + return None, False body = matches[-1].group("body") or "" condition_match = re.search(r'condition\s*=\s*"([^"]+)"', body) - return condition_match.group(1) if condition_match else None + condition = condition_match.group(1) if condition_match else None + file_system = bool(re.search(r"\bfileSystem\s*=\s*true\b", body)) + return condition, file_system def discover_examples() -> list[ExampleFunction]: @@ -81,7 +84,7 @@ def discover_examples() -> list[ExampleFunction]: if not is_top_level_durable_handler(source, class_name): continue - condition = read_template_condition(source, class_name) + condition, file_system = read_template_metadata(source, class_name) package_name = read_package(source, path) examples.append( ExampleFunction( @@ -89,6 +92,7 @@ def discover_examples() -> list[ExampleFunction]: package_name=package_name, suffix=kebab_case(class_name), condition=condition, + file_system=file_system, ) ) return examples @@ -103,7 +107,10 @@ def emit_function(lines: list[str], example: ExampleFunction) -> None: ) if example.condition: lines.append(f" Condition: {example.condition}") - lines.append(f" DependsOn: {example.log_group_logical_id}") + if example.file_system: + lines.extend([" DependsOn:", f" - {example.log_group_logical_id}"]) + else: + lines.append(f" DependsOn: {example.log_group_logical_id}") lines.extend( [ " Properties:", @@ -112,6 +119,26 @@ def emit_function(lines: list[str], example: ExampleFunction) -> None: " Role: !Ref RoleArn", ] ) + if example.file_system: + lines.extend( + [ + " VpcConfig:", + " SecurityGroupIds:", + " - Fn::ImportValue:", + ' Fn::Sub: "${FileSystemInfrastructureStackName}-LambdaSecurityGroupId"', + " SubnetIds:", + " - Fn::ImportValue:", + ' Fn::Sub: "${FileSystemInfrastructureStackName}-SubnetId"', + " FileSystemConfigs:", + " - Arn:", + " Fn::ImportValue:", + ' Fn::Sub: "${FileSystemInfrastructureStackName}-AccessPointArn"', + " LocalMountPath: /mnt/efs", + " Environment:", + " Variables:", + " FILESYSTEM_PAYLOAD_PATH: /mnt/efs", + ] + ) lines.append("") @@ -134,6 +161,130 @@ def emit_log_group(lines: list[str], example: ExampleFunction) -> None: ) +def emit_file_system_resources(lines: list[str]) -> None: + lines.extend( + [ + " FileSystemVpc:", + " Type: AWS::EC2::VPC", + " Properties:", + " CidrBlock: 10.0.0.0/24", + " EnableDnsHostnames: true", + " EnableDnsSupport: true", + "", + " FileSystemSubnet:", + " Type: AWS::EC2::Subnet", + " Properties:", + " CidrBlock: 10.0.0.0/26", + " VpcId: !Ref FileSystemVpc", + "", + " FileSystemLambdaSecurityGroup:", + " Type: AWS::EC2::SecurityGroup", + " Properties:", + " GroupDescription: Lambda access to EFS and the Lambda API endpoint", + " VpcId: !Ref FileSystemVpc", + "", + " FileSystemMountSecurityGroup:", + " Type: AWS::EC2::SecurityGroup", + " Properties:", + " GroupDescription: EFS mount access from Lambda", + " VpcId: !Ref FileSystemVpc", + " SecurityGroupIngress:", + " - IpProtocol: tcp", + " FromPort: 2049", + " ToPort: 2049", + " SourceSecurityGroupId: !Ref FileSystemLambdaSecurityGroup", + "", + " FileSystemEndpointSecurityGroup:", + " Type: AWS::EC2::SecurityGroup", + " Properties:", + " GroupDescription: Lambda API endpoint access from Lambda", + " VpcId: !Ref FileSystemVpc", + " SecurityGroupIngress:", + " - IpProtocol: tcp", + " FromPort: 443", + " ToPort: 443", + " SourceSecurityGroupId: !Ref FileSystemLambdaSecurityGroup", + "", + " FileSystemLambdaEndpoint:", + " Type: AWS::EC2::VPCEndpoint", + " Properties:", + " PrivateDnsEnabled: true", + " SecurityGroupIds:", + " - !Ref FileSystemEndpointSecurityGroup", + ' ServiceName: !Sub "com.amazonaws.${AWS::Region}.lambda"', + " SubnetIds:", + " - !Ref FileSystemSubnet", + " VpcEndpointType: Interface", + " VpcId: !Ref FileSystemVpc", + "", + " FileSystem:", + " Type: AWS::EFS::FileSystem", + " Properties:", + " Encrypted: true", + " PerformanceMode: generalPurpose", + " ThroughputMode: bursting", + "", + " FileSystemMountTarget:", + " Type: AWS::EFS::MountTarget", + " Properties:", + " FileSystemId: !Ref FileSystem", + " SecurityGroups:", + " - !Ref FileSystemMountSecurityGroup", + " SubnetId: !Ref FileSystemSubnet", + "", + " FileSystemAccessPoint:", + " Type: AWS::EFS::AccessPoint", + " Properties:", + " FileSystemId: !Ref FileSystem", + " PosixUser:", + ' Gid: "1000"', + ' Uid: "1000"', + " RootDirectory:", + " CreationInfo:", + ' OwnerGid: "1000"', + ' OwnerUid: "1000"', + ' Permissions: "0777"', + " Path: /durable-payloads", + "", + ] + ) + + +def emit_file_system_outputs(lines: list[str]) -> None: + lines.extend( + [ + "Outputs:", + " SubnetId:", + " Value: !Ref FileSystemSubnet", + " Export:", + ' Name: !Sub "${AWS::StackName}-SubnetId"', + "", + " LambdaSecurityGroupId:", + " Value: !Ref FileSystemLambdaSecurityGroup", + " Export:", + ' Name: !Sub "${AWS::StackName}-LambdaSecurityGroupId"', + "", + " AccessPointArn:", + " Value: !GetAtt FileSystemAccessPoint.Arn", + " Export:", + ' Name: !Sub "${AWS::StackName}-AccessPointArn"', + ] + ) + + +def render_file_system_infrastructure_template() -> str: + lines = [ + "# This file is generated by examples/generate-template.py. Do not edit it by hand.", + 'AWSTemplateFormatVersion: "2010-09-09"', + "Description: Persistent shared EFS infrastructure for filesystem payload offloader E2E tests", + "", + "Resources:", + ] + emit_file_system_resources(lines) + emit_file_system_outputs(lines) + return "\n".join(lines) + "\n" + + def render_template(examples: list[ExampleFunction]) -> str: lines = [ "# This file is generated by examples/generate-template.py. Do not edit it by hand.", @@ -160,6 +311,17 @@ def render_template(examples: list[ExampleFunction]) -> str: " RoleArn:", " Type: String", " Description: IAM Role ARN for Lambda function execution", + ] + if any(example.file_system for example in examples): + lines.extend( + [ + " FileSystemInfrastructureStackName:", + " Type: String", + " Description: Name of the shared persistent filesystem infrastructure stack", + ] + ) + lines.extend( + [ "", "Conditions:", " IsJava21OrLater:", @@ -182,7 +344,8 @@ def render_template(examples: list[ExampleFunction]) -> str: " FUNCTION_NAME_PREFIX: !Ref FunctionNamePrefix", "", "Resources:", - ] + ] + ) for example in examples: emit_log_group(lines, example) @@ -208,9 +371,18 @@ def render_template(examples: list[ExampleFunction]) -> str: def main() -> None: parser = argparse.ArgumentParser(description="Generate the examples SAM template from Java example handlers.") parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT, help="Path to write the generated template.") + template_selection = parser.add_mutually_exclusive_group() + template_selection.add_argument("--file-system-only", action="store_true") + template_selection.add_argument("--file-system-infrastructure-only", action="store_true") args = parser.parse_args() + if args.file_system_infrastructure_only: + args.output.write_text(render_file_system_infrastructure_template(), encoding="utf-8") + print(f"Generated persistent filesystem infrastructure template at {args.output}.") + return + examples = discover_examples() + examples = [example for example in examples if example.file_system == args.file_system_only] if not examples: raise RuntimeError("No DurableHandler examples found") diff --git a/examples/src/main/java/software/amazon/lambda/durable/examples/ExampleTemplate.java b/examples/src/main/java/software/amazon/lambda/durable/examples/ExampleTemplate.java index de93e95ca..55eab7934 100644 --- a/examples/src/main/java/software/amazon/lambda/durable/examples/ExampleTemplate.java +++ b/examples/src/main/java/software/amazon/lambda/durable/examples/ExampleTemplate.java @@ -12,4 +12,6 @@ @Target(ElementType.TYPE) public @interface ExampleTemplate { String condition() default ""; + + boolean fileSystem() default false; } diff --git a/examples/src/main/java/software/amazon/lambda/durable/examples/general/FileSystemPayloadOffloaderExample.java b/examples/src/main/java/software/amazon/lambda/durable/examples/general/FileSystemPayloadOffloaderExample.java new file mode 100644 index 000000000..701b92d07 --- /dev/null +++ b/examples/src/main/java/software/amazon/lambda/durable/examples/general/FileSystemPayloadOffloaderExample.java @@ -0,0 +1,67 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.examples.general; + +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.time.Duration; +import java.util.HexFormat; +import software.amazon.lambda.durable.DurableConfig; +import software.amazon.lambda.durable.DurableContext; +import software.amazon.lambda.durable.DurableHandler; +import software.amazon.lambda.durable.examples.ExampleTemplate; +import software.amazon.lambda.durable.offload.filesystem.FileSystemPayloadOffloader; +import software.amazon.lambda.durable.offload.filesystem.PreviewConfig; +import software.amazon.lambda.durable.offload.filesystem.PreviewField; +import software.amazon.lambda.durable.offload.filesystem.PreviewMode; + +/** E2E fixture that offloads durable payloads to an EFS mount and reads them after reinvocation. */ +@ExampleTemplate(fileSystem = true) +public class FileSystemPayloadOffloaderExample + extends DurableHandler { + private static final String FILE_SYSTEM_PATH_ENV = "FILESYSTEM_PAYLOAD_PATH"; + + @Override + protected DurableConfig createConfiguration() { + var path = System.getenv(FILE_SYSTEM_PATH_ENV); + if (path == null || path.isBlank()) { + throw new IllegalStateException(FILE_SYSTEM_PATH_ENV + " must identify the mounted durable filesystem"); + } + var preview = PreviewConfig.builder(PreviewMode.EXCLUDE_ALL) + .include(PreviewField.anywhere("id"), PreviewField.anywhere("length")) + .mask(PreviewField.anywhere("value")) + .build(); + var offloader = FileSystemPayloadOffloader.builder(Path.of(path)) + .previewConfig(preview) + .build(); + return DurableConfig.builder().withPayloadOffloader(offloader).build(); + } + + @Override + public Output handleRequest(Input input, DurableContext context) { + var stored = + context.step("store-payload", Payload.class, stepContext -> new Payload(input.id(), input.value())); + context.wait("force-filesystem-replay", Duration.ofSeconds(1)); + return context.step( + "verify-payload", + Output.class, + stepContext -> new Output(stored.id(), stored.value().length(), sha256(stored.value()))); + } + + private static String sha256(String value) { + try { + return HexFormat.of() + .formatHex(MessageDigest.getInstance("SHA-256").digest(value.getBytes(StandardCharsets.UTF_8))); + } catch (NoSuchAlgorithmException e) { + throw new IllegalStateException("SHA-256 is unavailable", e); + } + } + + public record Input(String id, String value) {} + + public record Payload(String id, String value) {} + + public record Output(String id, int length, String checksum) {} +} diff --git a/examples/src/test/java/software/amazon/lambda/durable/examples/CloudBasedIntegrationTest.java b/examples/src/test/java/software/amazon/lambda/durable/examples/CloudBasedIntegrationTest.java index 7490961a5..bed5f0741 100644 --- a/examples/src/test/java/software/amazon/lambda/durable/examples/CloudBasedIntegrationTest.java +++ b/examples/src/test/java/software/amazon/lambda/durable/examples/CloudBasedIntegrationTest.java @@ -5,6 +5,7 @@ import static org.junit.jupiter.api.Assertions.*; import static software.amazon.lambda.durable.TypeToken.get; +import com.fasterxml.jackson.databind.ObjectMapper; import java.time.Duration; import java.util.HashMap; import java.util.List; @@ -14,6 +15,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.condition.EnabledForJreRange; import org.junit.jupiter.api.condition.EnabledIf; +import org.junit.jupiter.api.condition.EnabledIfSystemProperty; import org.junit.jupiter.api.condition.JRE; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; @@ -24,6 +26,7 @@ import software.amazon.awssdk.services.lambda.model.OperationStatus; import software.amazon.awssdk.services.sts.StsClient; import software.amazon.lambda.durable.TypeToken; +import software.amazon.lambda.durable.examples.general.FileSystemPayloadOffloaderExample; import software.amazon.lambda.durable.examples.general.GenericTypesExample; import software.amazon.lambda.durable.examples.types.ApprovalRequest; import software.amazon.lambda.durable.examples.types.GreetingRequest; @@ -38,6 +41,7 @@ @EnabledIf("isEnabled") class CloudBasedIntegrationTest { private static final int PERFORMANCE_TEST_REPEAT = 3; + private static final ObjectMapper MAPPER = new ObjectMapper(); private static String account; private static String region; @@ -85,6 +89,33 @@ private static String arn(String functionName) { + ":$LATEST"; } + @Test + @EnabledIfSystemProperty(named = "test.filesystem.enabled", matches = "true") + void testFileSystemPayloadOffloaderExample() throws Exception { + var value = "filesystem-e2e-".repeat(24 * 1024); + var input = new FileSystemPayloadOffloaderExample.Input("payload-1", value); + var runner = CloudDurableTestRunner.create( + arn("file-system-payload-offloader-example"), + FileSystemPayloadOffloaderExample.Input.class, + FileSystemPayloadOffloaderExample.Output.class, + lambdaClient); + + var result = runner.run(input); + + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + assertTrue(result.getHistoryEvents().stream() + .filter(event -> event.eventType() + == software.amazon.awssdk.services.lambda.model.EventType.INVOCATION_COMPLETED) + .count() + >= 2); + var stored = result.getOperation("store-payload").getStepDetails().result(); + var envelope = MAPPER.readTree(stored.substring("@aws-durable-payload:v1:".length())); + assertEquals("REFERENCE", envelope.get("mode").textValue()); + assertEquals("payload-1", envelope.get("preview").get("id").textValue()); + assertEquals("***", envelope.get("preview").get("value").textValue()); + assertTrue(envelope.get("payloadDigest").textValue().matches("[0-9a-f]{64}")); + } + /** Custom SerDes that tracks serialization calls. */ static class TrackingSerDes implements SerDes { private final JacksonSerDes delegate = new JacksonSerDes(); diff --git a/examples/test_generate_template.py b/examples/test_generate_template.py new file mode 100644 index 000000000..e25ae72b5 --- /dev/null +++ b/examples/test_generate_template.py @@ -0,0 +1,52 @@ +#!/usr/bin/env python3 + +from __future__ import annotations + +import importlib.util +import sys +import unittest +from pathlib import Path + + +GENERATOR_PATH = Path(__file__).with_name("generate-template.py") +SPEC = importlib.util.spec_from_file_location("generate_template", GENERATOR_PATH) +if SPEC is None or SPEC.loader is None: + raise RuntimeError(f"Unable to load {GENERATOR_PATH}") +generate_template = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = generate_template +SPEC.loader.exec_module(generate_template) + + +class GenerateTemplateTest(unittest.TestCase): + def test_default_template_does_not_include_file_system_infrastructure(self) -> None: + examples = [example for example in generate_template.discover_examples() if not example.file_system] + + template = generate_template.render_template(examples) + + self.assertNotIn("FileSystemInfrastructureStackName", template) + self.assertNotIn("AWS::EFS::FileSystem", template) + + def test_file_system_lambda_template_imports_persistent_infrastructure(self) -> None: + examples = [example for example in generate_template.discover_examples() if example.file_system] + + template = generate_template.render_template(examples) + + self.assertIn("FileSystemInfrastructureStackName:", template) + self.assertIn("${FileSystemInfrastructureStackName}-SubnetId", template) + self.assertIn("${FileSystemInfrastructureStackName}-LambdaSecurityGroupId", template) + self.assertIn("${FileSystemInfrastructureStackName}-AccessPointArn", template) + self.assertNotIn("AWS::EFS::FileSystem", template) + + def test_file_system_infrastructure_template_exports_shared_resources(self) -> None: + template = generate_template.render_file_system_infrastructure_template() + + self.assertIn("AWS::EFS::FileSystem", template) + self.assertIn("AWS::EFS::MountTarget", template) + self.assertIn("${AWS::StackName}-SubnetId", template) + self.assertIn("${AWS::StackName}-LambdaSecurityGroupId", template) + self.assertIn("${AWS::StackName}-AccessPointArn", template) + self.assertNotIn("AWS::Serverless::Function", template) + + +if __name__ == "__main__": + unittest.main()