diff --git a/.github/scripts/maven_publish.sh b/.github/scripts/maven_publish.sh index 655ad5c69..75c194830 100644 --- a/.github/scripts/maven_publish.sh +++ b/.github/scripts/maven_publish.sh @@ -45,5 +45,6 @@ echo "=== Step 3: Upload to Sonatype Central Portal ===" mvn clean deploy -s "${SETTINGS_FILE}" -pl sdk -P publishing -DskipTests --no-transfer-progress mvn clean deploy -s "${SETTINGS_FILE}" -pl sdk-testing -P publishing -DskipTests --no-transfer-progress mvn clean deploy -s "${SETTINGS_FILE}" -pl otel-plugin -P publishing -DskipTests --no-transfer-progress +mvn clean deploy -s "${SETTINGS_FILE}" -pl insight-plugin -P publishing -DskipTests --no-transfer-progress echo "=== Release ${RELEASE_VERSION} uploaded successfully; review and publish it in Sonatype Central Portal. ===" diff --git a/.github/workflows/publish_maven.yml b/.github/workflows/publish_maven.yml index a664ca282..bb170a29f 100644 --- a/.github/workflows/publish_maven.yml +++ b/.github/workflows/publish_maven.yml @@ -92,6 +92,7 @@ jobs: "sdk/target/aws-durable-execution-sdk-java-${RELEASE_VERSION}.jar" \ "sdk-testing/target/aws-durable-execution-sdk-java-testing-${RELEASE_VERSION}.jar" \ "otel-plugin/target/aws-durable-execution-sdk-java-plugin-otel-${RELEASE_VERSION}.jar" \ + "insight-plugin/target/aws-durable-execution-sdk-java-plugin-insight-${RELEASE_VERSION}.jar" \ --clobber - name: Checkout default branch diff --git a/RELEASE.md b/RELEASE.md index 90c347ff1..1dda68efd 100644 --- a/RELEASE.md +++ b/RELEASE.md @@ -43,9 +43,9 @@ The publication workflow: 1. Verifies that the tag is a semantic version, points to a commit on the default branch, and matches the Maven version in the tagged POM. -2. Builds, signs, and uploads the SDK, testing library, and OpenTelemetry plugin - to Sonatype Central Portal. -3. Uploads the three JARs to the existing GitHub release. +2. Builds, signs, and uploads the SDK, testing library, OpenTelemetry plugin, + and Workflow Insight plugin to Sonatype Central Portal. +3. Uploads the four JARs to the existing GitHub release. 4. Opens a pull request for the next development version. A final release increments the patch version, so `2.1.1` produces `2.1.2-SNAPSHOT`. A prerelease keeps the same base version, so `2.1.1-rc1` produces @@ -56,7 +56,8 @@ After **Publish Maven Release** succeeds: 1. Open [Publishing Deployments](https://central.sonatype.com/publishing/deployments) in Sonatype Central Portal. 2. Find the deployments for the release version and verify that they contain - the expected SDK, testing library, and OpenTelemetry plugin artifacts. + the expected SDK, testing library, OpenTelemetry plugin, and Workflow Insight + plugin artifacts. 3. Click **Publish** for each deployment and wait for publication to complete. The workflow uses `autoPublish=false`, so this manual action is required. 4. Confirm that the GitHub release contains the expected JARs and that the diff --git a/insight-plugin/README.md b/insight-plugin/README.md new file mode 100644 index 000000000..75896b7cc --- /dev/null +++ b/insight-plugin/README.md @@ -0,0 +1,65 @@ +# Workflow Insight Plugin (preview) + +Instrumentation plugin for the AWS Lambda Durable Execution Java SDK that emits a curated, +per-execution **Workflow Insight record** to one or more pluggable exporters. It ports the +JavaScript `workflowInsight()` contract (canonical record schema `1.0`) to the Java plugin hook +surface. + +> **Preview API.** Every public type is annotated `@Deprecated` to signal it is experimental and +> may change or be removed in a future release. + +## Usage + +```java +DurableConfig config = DurableConfig.builder() + .withPlugins(WorkflowInsight.workflowInsight(WorkflowInsightConfig.builder() + .samplingRate(1.0) + .emitMode(WorkflowInsightConfig.EmitMode.ON_COMPLETE) // ON_COMPLETE | ON_CHANGE | ON_FAILURE + .operationDetail(WorkflowInsightConfig.OperationDetail.TOP_LEVEL) // TOP_LEVEL | FULL_TREE + .content(ContentConfig.builder() + .input(true).output(true).includeErrors(true) + .addOverride(OperationOverride.withResult("compute", r -> r)) + .build()) + .addExporter(S3Exporter.builder().bucket("my-bucket").build()) + .build())) + .build(); +``` + +Exporters: `LambdaLogExporter` (default; writes the `operationsByName` map to stdout → +CloudWatch), `S3Exporter` (canonical `operations` array, one object per execution), +`CloudWatchLogsExporter` (PutLogEvents to a specific log group, `operationsByName` map). Implement +`InsightExporter` for custom sinks. + +## Design + +- **Snapshot-based, not accumulated.** Each record is built directly from the current-invocation + operation snapshot the SDK provides — `InvocationInfo.operations()` at start / operation change + and `InvocationEndInfo.operations()` at end. Execution input/output come from + `InvocationInfo.executionInput()` / `InvocationEndInfo.executionResult()`, and per-operation + results from `OperationChangeItemInfo.result()` (all surfaced by SDK PR #618). There is no global + "current ARN" or cross-hook operation accumulation. +- **Per-execution state keyed by execution ARN** holds only the stable start time, parsed ARN, + cached input, and the one-time deterministic sampling decision. State is **preserved across + non-terminal (PENDING/RETRYING) invocations** so suspend/resume keeps a single stable start time + and correct duration, and is removed **only** once the execution is terminal. +- **Deterministic sampling.** FNV-1a-32 over the execution ARN mapped into `[0,1)`, identical to + the JS implementation, so a resumed execution always reaches the same in/out decision. +- **Emission modes.** `ON_COMPLETE` emits one terminal record; `ON_FAILURE` emits only on terminal + failure; `ON_CHANGE` emits at invocation start, on every operation change, and at invocation end + (matching JS). Non-terminal statuses map to `RUNNING`. +- **Operation filtering** mirrors JS: the `EXECUTION` pseudo-operation and unnamed operations are + dropped; `TOP_LEVEL` detail drops any operation with a `parentId`; an `OperationOverride.exclude` + drops by name. Operation `result` is included only when an `OperationOverride.withResult` + transform opts in — the checkpointed JSON is parsed before the transform, falling back to the raw + string, and a throwing transform omits the field. +- **Per-exporter size truncation** (`Truncation`) drops, in order: operation results oldest-first, + then whole operations oldest-first, then execution input, then output — setting `truncated`, + `droppedOperations`, `droppedInput`, `droppedOutput` as applicable. The size is measured against + the exact shape each exporter emits (its `render`). +- **Exporter isolation.** Every exporter is truncated, exported, and flushed independently; a + failing exporter is logged and never blocks the others or the execution. + +## Conformance + +Validated against the Workflow Insight conformance suite behaviors `insight-1 … insight-18` +(PR #73 Java examples). See the module tests for the behavior mapping. diff --git a/insight-plugin/REVIEW_FINDINGS.md b/insight-plugin/REVIEW_FINDINGS.md new file mode 100644 index 000000000..9a9bebd6b --- /dev/null +++ b/insight-plugin/REVIEW_FINDINGS.md @@ -0,0 +1,185 @@ +# PR #661 — Codex Review Findings & Resolution + +Workflow Insight plugin (`insight-plugin`). Review run by Codex AI on commit +`9cb8201` of branch `workflow-insight-plugin`. + +Legend — **Status**: `FIXED` (addressed in this change), `OPEN` (intentionally +left unaddressed for now), `DEFERRED` (blocked on coordinated cross-SDK work). + +| # | Finding ID | Pri | Title | Status | +|---|------------|-----|-------|--------| +| 1 | `arf_v1_v6wv3rjugx56mymsuekqwwjamc` | P1 | Publish the new plugin artifact | FIXED | +| 2 | `arf_v1_blkdgaotf7rzua2ga3uyouqm5e` | P1 | Support the SDK's default payload types (Java-time) | FIXED | +| 3 | `arf_v1_xryomwjxrzqt3z55wozggpsrsl` | P2 | Keep remote exports off the checkpoint callback | OPEN | +| 4 | `arf_v1_ig2yofnonmfn7zknekoxwd6xmg` | P2 | Deterministic chronological operation order | FIXED | +| 5 | `arf_v1_lndcqjp4vnrm6vjmmfzrqrfdin` | P2 | Preserve the checkpointed operation error type | FIXED | +| 6 | `arf_v1_oirdg3rnpdnwvpldafyiqrnncz` | P2 | Distinguish included JSON null from an omitted field | OPEN | +| 7 | `arf_v1_rzwdoyquezpr2qgby63imtmxcn` | P2 | Prevent one exporter mutating later exporters' records | FIXED | +| 8 | `arf_v1_47kz274xvx4sow4mmbm7gzor7a` | P3 | Normalize the hash into the half-open interval | DEFERRED | + +--- + +## 1 — Publish the new plugin artifact — `FIXED` (P1) + +- Discussion: https://github.com/aws/aws-durable-execution-sdk-java/pull/661#discussion_r3883200348 +- Classification: release/packaging gap. +- Problem: the reactor builds `insight-plugin`, but the release pipeline deploys + and uploads only `sdk`, `sdk-testing`, and `otel-plugin`, so + `aws-durable-execution-sdk-java-plugin-insight` would be absent from Maven + Central and the GitHub release. +- Resolution: + - `.github/scripts/maven_publish.sh`: added a fourth + `mvn clean deploy ... -pl insight-plugin -P publishing` step, matching the + existing per-module deploy style. + - `.github/workflows/publish_maven.yml`: added + `insight-plugin/target/aws-durable-execution-sdk-java-plugin-insight-${RELEASE_VERSION}.jar` + to the `gh release upload` list. + - `RELEASE.md`: updated the artifact list ("… and Workflow Insight plugin") + and the JAR count ("four JARs") in the publication and verification sections. +- Caveat: the finding also asked for a "release-workflow check". No existing + automated guard/parser for the release workflow exists in the repo, and a + bespoke script that greps the workflow for module names would be brittle + (exactly the anti-pattern to avoid). No guard was added; the `publishing` + profile is inherited from the parent POM, so the module produces the same + signed/sources/javadoc artifacts as `otel-plugin`. + +## 2 — Support the SDK's default payload types — `FIXED` (P1) + +- Discussion: https://github.com/aws/aws-durable-execution-sdk-java/pull/661#discussion_r3883200520 +- Classification: correctness / silent data loss. +- Problem: the bare `ObjectMapper` cannot serialize Java-time values (e.g. + `Instant`), which `JacksonSerDes` supports. An included input/output/result + containing one made size calculation return `null` and final serialization + throw; `emit` catches the exception and silently drops every record. +- Resolution: + - `insight-plugin/pom.xml`: added a direct + `com.fasterxml.jackson.datatype:jackson-datatype-jsr310` dependency (version + managed by the imported `jackson-bom`). + - `Json.MAPPER`: registered `JavaTimeModule` and disabled + `SerializationFeature.WRITE_DATES_AS_TIMESTAMPS` so Java-time values emit as + ISO-8601 strings. + - Tests (`JsonJavaTimeTest`): `Instant` stringifies to + `"2026-08-05T12:34:56Z"`; `byteSize` of a payload containing an `Instant` is + non-null; end-to-end plugin output with an `Instant` in the execution input + serializes and carries the ISO-8601 value. + +## 3 — Keep remote exports off the checkpoint callback — `OPEN` (P2) + +- Discussion: https://github.com/aws/aws-durable-execution-sdk-java/pull/661#discussion_r3883200693 +- Classification: performance / durability risk in `ON_CHANGE` mode. +- Problem: `emit` runs synchronously from `onOperationChange`, which the SDK + invokes before checkpoint futures/pollers complete. Synchronous S3/CloudWatch + calls plus immediate flush on every change can stall checkpoint coordination + and risk invocation timeouts. +- Disposition: left OPEN. The suggested fix (queue records on non-terminal + hooks; perform serial export/flush from `onInvocationEnd`; add a + delayed-exporter test proving change hooks return promptly) is a non-trivial + behavioral redesign of the emit lifecycle and is out of scope for this change. + Not addressed here. + +## 4 — Deterministic chronological operation order — `FIXED` (P2) + +- Discussion: https://github.com/aws/aws-durable-execution-sdk-java/pull/661#discussion_r3883200875 +- Classification: correctness / non-determinism. +- Problem: the hook supplies a map with no iteration-order guarantee (the core + snapshot comes from a concurrent map). Iterating `values()` yields unstable + operation arrays, and `OperationsIndex` treats the last-encountered repeated + name as the latest — so an older occurrence could overwrite the real latest + status/type/subType. +- Resolution: + - `WorkflowInsight.buildOperationRecords`: sorts `operations.values()` by + `startTimestamp` ascending (nulls last) with a stable operation-`id` + tie-breaker (nulls last) *before* filtering/building records. Chronological + order then flows into `OperationsIndex`, so `operationsByName` latest scalar + fields (`status`, `type`, `subType`) reflect the chronologically last + occurrence. + - Tests (`OperationOrderingTest`): shuffled start timestamps emitted + chronologically; equal timestamps broken by ascending id; null timestamps + last; repeated-name latest scalars reflect chronological (not map) order with + differing status/subType and a preserved `failedCount`. + +## 5 — Preserve the checkpointed operation error type — `FIXED` (P2) + +- Discussion: https://github.com/aws/aws-durable-execution-sdk-java/pull/661#discussion_r3883201083 +- Classification: correctness / observability fidelity. +- Problem: operation and execution snapshot errors arrive wrapped as + `DurableOperationException` (via `PluginInfoConverter`), so `toErrorInfo` + emitted the generic wrapper class name, losing the original failure identity. +- Resolution: + - `WorkflowInsight.toErrorInfo`: when the throwable is a + `DurableOperationException` with a non-null `ErrorObject`, derives + `name`/`message` from `errorType()`/`errorMessage()`, falling back to the + throwable's simple class name / message for any null field. Applies to both + the operation-level path (`item.error()`) and the execution-level path + (`executionError`) since both flow through `toErrorInfo`. + - Tests (`OperationErrorIdentityTest`): asserts exact original type and + message for both operation-level and execution-level errors, plus the + fallback path when `errorType` is null. + +## 6 — Distinguish included JSON null from an omitted field — `OPEN` (P2) + +- Discussion: https://github.com/aws/aws-durable-execution-sdk-java/pull/661#discussion_r3883201219 +- Classification: wire-shape fidelity. +- Problem: an enabled-but-`null` value is conflated with content excluded by + configuration; the wire map omits both. A successful handler returning `null` + therefore has no `output` field even though output inclusion defaults to true. +- Disposition: left OPEN. The fix (track field presence separately or use an + explicit JSON-null sentinel, emitting `null` for included terminal values + while omitting only unavailable/disabled content, plus null input/output + tests) touches the record's presence/omission model and its cross-SDK wire + contract; deferred to a dedicated change. Not addressed here. + +## 7 — Prevent one exporter mutating later exporters' records — `FIXED` (P2) + +- Discussion: https://github.com/aws/aws-durable-execution-sdk-java/pull/661#discussion_r3883201369 +- Classification: robustness / isolation. +- Problem: the accessor exposed the live operations list (with mutable + elements), and truncation returns the original record when it already fits, so + all exporters shared one mutable object — a custom exporter that clears or + redacts operations/nested content could corrupt every subsequent export. +- Resolution: + - `WorkflowInsightRecord.operations()` now returns an immutable snapshot + (`Collections.unmodifiableList`). A package-private `addOperation(...)` + mutator is used for record construction (allowed here — preview API). + - Added `WorkflowInsightRecord.deepCopy()` (deep-copies each operation and the + execution input/output content) and `OperationRecord.deepCopy()` + (deep-copies the result payload). `Json.deepCopyContent` rebuilds mutable + `Map`/`List` container structure recursively and converts other serializable + payload objects into detached JSON-compatible object graphs, preserving the + emitted JSON while preventing mutable POJO state from being shared between + exporters (`ErrorInfo` is immutable and shared safely). + - `WorkflowInsight.emit`: takes a per-exporter `deepCopy()` of the record + before truncation/shaping, so each exporter operates on an isolated graph. + - Test (`ExporterIsolationTest`): a hostile first exporter mutates operation + fields and clears the nested input map; the second exporter still sees the + pristine operation name/status, nested result content, and input content. +- Caveat: nested `Map`/`List` structure is deep-copied (the mutation vector for + JSON payloads); arbitrary user POJO leaves are shared as immutable values. If + a payload is a mutable custom POJO, a hostile exporter could still mutate its + internal fields — considered out of scope and unlikely for + serialized/deserialized payloads. + +## 8 — Normalize the hash into the half-open interval — `DEFERRED` (P3) + +- Discussion: https://github.com/aws/aws-durable-execution-sdk-java/pull/661#discussion_r3883201548 +- Classification: cross-SDK sampling-parity nuance. +- Problem: dividing by max unsigned 32-bit (`0xffffffff`) maps `0xffffffff` to + exactly `1.0`, yielding `[0,1]` rather than the documented `[0,1)` and + potentially diverging from other SDKs' deterministic sampling decisions. The + suggested change divides by `2^32` (`0x1_0000_0000L`). +- Disposition: DEFERRED. Changing the sampling denominator alters the + deterministic decision boundary and must be coordinated with the JavaScript + and Java reference implementations so all SDKs agree on the same + fixed-vector decisions; a Java-only change here would risk cross-SDK drift. + Per direction, the sampling denominator is **left unchanged** pending that + coordinated change. + +--- + +## Verification + +- `mvn spotless:apply` then `mvn -B -q spotless:check --file pom.xml`: clean. +- `mvn -pl insight-plugin -am clean verify`: **BUILD SUCCESS** — + Tests run: 45, Failures: 0, Errors: 0, Skipped: 0. +- `mvn -q clean verify`: full reactor **BUILD SUCCESS**. +- `bash -n .github/scripts/maven_publish.sh`: clean. diff --git a/insight-plugin/pom.xml b/insight-plugin/pom.xml new file mode 100644 index 000000000..be0fbcbbc --- /dev/null +++ b/insight-plugin/pom.xml @@ -0,0 +1,112 @@ + + + 4.0.0 + + + software.amazon.lambda.durable + aws-durable-execution-sdk-java-parent + 2.2.1-SNAPSHOT + + + aws-durable-execution-sdk-java-plugin-insight + AWS Lambda Durable Execution SDK Workflow Insight Plugin + Workflow Insight instrumentation plugin for AWS Lambda Durable Execution SDK + + + + + software.amazon.lambda.durable + aws-durable-execution-sdk-java + ${project.version} + + + + + com.fasterxml.jackson.core + jackson-databind + + + + + com.fasterxml.jackson.datatype + jackson-datatype-jsr310 + + + + + software.amazon.awssdk + s3 + + + software.amazon.awssdk + cloudwatchlogs + + + + + org.slf4j + slf4j-api + + + + + org.junit.jupiter + junit-jupiter + test + + + software.amazon.lambda.durable + aws-durable-execution-sdk-java-testing + ${project.version} + test + + + org.mockito + mockito-core + test + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + + + org.apache.maven.plugins + maven-surefire-plugin + + + com.diffplug.spotless + spotless-maven-plugin + + + org.apache.maven.plugins + maven-source-plugin + + + attach-sources + + jar-no-fork + + + + + + org.apache.maven.plugins + maven-javadoc-plugin + + + attach-javadocs + + jar + + + + + + + diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/ArnParser.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/ArnParser.java new file mode 100644 index 000000000..544044d7e --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/ArnParser.java @@ -0,0 +1,60 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +/** + * Parses the durable execution ARN into its component fields, mirroring the JS {@code parseExecutionArn}. + * + *

Format: + * {@code arn::lambda:::function::/durable-execution//}. + * + * @deprecated This is a preview API that is experimental and may be changed or removed in future releases. + */ +@Deprecated +public final class ArnParser { + private final String functionName; + private final String qualifier; + private final String region; + private final String accountId; + private final String executionName; + + private ArnParser(String functionName, String qualifier, String region, String accountId, String executionName) { + this.functionName = functionName; + this.qualifier = qualifier; + this.region = region; + this.accountId = accountId; + this.executionName = executionName; + } + + public static ArnParser parse(String executionArn) { + String[] parts = executionArn.split(":", -1); + String lastPart = parts.length > 7 ? parts[7] : ""; + String[] segments = lastPart.split("/", -1); + return new ArnParser( + parts.length > 6 ? parts[6] : "", + segments.length > 0 ? segments[0] : "", + parts.length > 3 ? parts[3] : "", + parts.length > 4 ? parts[4] : "", + segments.length > 2 ? segments[2] : ""); + } + + public String functionName() { + return functionName; + } + + public String qualifier() { + return qualifier; + } + + public String region() { + return region; + } + + public String accountId() { + return accountId; + } + + public String executionName() { + return executionName; + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/ContentConfig.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/ContentConfig.java new file mode 100644 index 000000000..64d9d50d7 --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/ContentConfig.java @@ -0,0 +1,111 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import java.util.ArrayList; +import java.util.List; +import java.util.function.Function; + +/** + * Controls what data is included in emitted records (input/output transforms, operation overrides, error inclusion). + * + *

Mirrors the JS {@code ContentConfig}. {@code input}/{@code output} may be included as-is, excluded, or + * transformed; all three are honored — the plugin reads execution input from {@code InvocationInfo.executionInput()} + * and output from {@code InvocationEndInfo.executionResult()}. {@code includeErrors} is honored via the operation error + * carried on each operation snapshot item. Per-operation result opt-in is honored via + * {@link OperationOverride#withResult} (see {@code OperationChangeItemInfo.result()}). + * + * @deprecated This is a preview API that is experimental and may be changed or removed in future releases. + */ +@Deprecated +public final class ContentConfig { + private final boolean includeInput; + private final Function inputTransform; + private final boolean includeOutput; + private final Function outputTransform; + private final boolean includeErrors; + private final List overrides; + + private ContentConfig(Builder b) { + this.includeInput = b.includeInput; + this.inputTransform = b.inputTransform; + this.includeOutput = b.includeOutput; + this.outputTransform = b.outputTransform; + this.includeErrors = b.includeErrors; + this.overrides = List.copyOf(b.overrides); + } + + public static Builder builder() { + return new Builder(); + } + + public boolean includeInput() { + return includeInput; + } + + public Function inputTransform() { + return inputTransform; + } + + public boolean includeOutput() { + return includeOutput; + } + + public Function outputTransform() { + return outputTransform; + } + + public boolean includeErrors() { + return includeErrors; + } + + public List overrides() { + return overrides; + } + + /** Builder for {@link ContentConfig}. */ + public static final class Builder { + private boolean includeInput = true; + private Function inputTransform; + private boolean includeOutput = true; + private Function outputTransform; + private boolean includeErrors = true; + private final List overrides = new ArrayList<>(); + + public Builder input(boolean include) { + this.includeInput = include; + return this; + } + + public Builder inputTransform(Function transform) { + this.inputTransform = transform; + this.includeInput = true; + return this; + } + + public Builder output(boolean include) { + this.includeOutput = include; + return this; + } + + public Builder outputTransform(Function transform) { + this.outputTransform = transform; + this.includeOutput = true; + return this; + } + + public Builder includeErrors(boolean include) { + this.includeErrors = include; + return this; + } + + public Builder addOverride(OperationOverride override) { + this.overrides.add(override); + return this; + } + + public ContentConfig build() { + return new ContentConfig(this); + } + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/ErrorInfo.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/ErrorInfo.java new file mode 100644 index 000000000..27407c145 --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/ErrorInfo.java @@ -0,0 +1,38 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import java.util.LinkedHashMap; +import java.util.Map; + +/** + * Error detail carried on a record or operation, mirroring the JS {@code {name, message}} shape. + * + * @deprecated This is a preview API that is experimental and may be changed or removed in future releases. + */ +@Deprecated +public final class ErrorInfo { + private final String name; + private final String message; + + public ErrorInfo(String name, String message) { + this.name = name; + this.message = message; + } + + public String name() { + return name; + } + + public String message() { + return message; + } + + /** Serializes to the camelCase wire shape {@code {"name": ..., "message": ...}}. */ + public Map toWireMap() { + Map data = new LinkedHashMap<>(); + data.put("name", name); + data.put("message", message); + return data; + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/InsightExporter.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/InsightExporter.java new file mode 100644 index 000000000..28283ece9 --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/InsightExporter.java @@ -0,0 +1,34 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +/** + * Exports workflow insight records to a destination. + * + * @deprecated This is a preview API that is experimental and may be changed or removed in future releases. + */ +@Deprecated +public interface InsightExporter { + + /** Emits one record to the destination. */ + void export(WorkflowInsightRecord record); + + /** Flushes any buffered records; no-op by default. */ + default void flush() {} + + /** + * Maximum serialized record size, in bytes, this exporter will emit; {@code null} disables truncation. Measured + * against {@link #render(WorkflowInsightRecord)}. + */ + default Integer maxRecordSizeBytes() { + return null; + } + + /** + * Maps a record to the exact value this exporter serializes/sends (defaults to the canonical {@code operations} + * array wire map). Overriding exporters (CloudWatch/Lambda log) return the {@code operationsByName} rendering. + */ + default Object render(WorkflowInsightRecord record) { + return record.toWireMap(); + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/Json.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/Json.java new file mode 100644 index 000000000..d400f8e9d --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/Json.java @@ -0,0 +1,83 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializationFeature; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** + * Minimal JSON helper for emitting insight records and measuring their serialized size. + * + * @deprecated This is a preview API that is experimental and may be changed or removed in future releases. + */ +@Deprecated +public final class Json { + // Register JavaTimeModule and emit ISO-8601 (not numeric timestamps) so SDK-default payload types such as + // java.time.Instant/Duration in an included input/output/result serialize instead of throwing and silently + // dropping the record. + static final ObjectMapper MAPPER = new ObjectMapper() + .registerModule(new JavaTimeModule()) + .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS); + + private Json() {} + + public static String stringify(Object value) { + try { + return MAPPER.writeValueAsString(value); + } catch (JsonProcessingException e) { + throw new IllegalStateException("failed to serialize insight record", e); + } + } + + /** UTF-8 byte length of the value's JSON, or {@code null} if it can't be serialized. */ + public static Integer byteSize(Object value) { + try { + return MAPPER.writeValueAsString(value).getBytes(StandardCharsets.UTF_8).length; + } catch (JsonProcessingException e) { + return null; + } + } + + /** + * Deep-copies JSON content (execution input/output and operation results) so a mutation by one exporter cannot leak + * into another exporter's record. Maps and lists are rebuilt recursively. Immutable scalar values are shared. Other + * serializable objects are converted into detached JSON-compatible object graphs, preserving their emitted JSON + * while preventing mutable POJO state from being shared between exporters. + */ + static Object deepCopyContent(Object value) { + if (value == null + || value instanceof String + || value instanceof Number + || value instanceof Boolean + || value instanceof Character + || value instanceof Enum) { + return value; + } + if (value instanceof Map map) { + Map copy = new LinkedHashMap<>(); + for (Map.Entry e : map.entrySet()) { + copy.put(e.getKey(), deepCopyContent(e.getValue())); + } + return copy; + } + if (value instanceof List list) { + List copy = new ArrayList<>(list.size()); + for (Object o : list) { + copy.add(deepCopyContent(o)); + } + return copy; + } + try { + return MAPPER.convertValue(value, Object.class); + } catch (IllegalArgumentException e) { + throw new IllegalStateException("failed to copy insight record content", e); + } + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/OperationOverride.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/OperationOverride.java new file mode 100644 index 000000000..745ef03ae --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/OperationOverride.java @@ -0,0 +1,48 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import java.util.function.Function; + +/** + * Per-operation override controlling inclusion and result transformation, matched by {@code operationName}. + * + *

Mirrors the JS {@code OperationOverride}. The {@code result} transform receives the operation's checkpointed, + * JSON-parsed result (from {@code OperationChangeItemInfo.result()}); the raw string is passed through when it is not + * valid JSON, and a throwing transform omits the field rather than leaking the raw value. An {@code exclude} override + * drops the operation entirely. + * + * @deprecated This is a preview API that is experimental and may be changed or removed in future releases. + */ +@Deprecated +public final class OperationOverride { + private final String operationName; + private final boolean exclude; + private final Function result; + + private OperationOverride(String operationName, boolean exclude, Function result) { + this.operationName = operationName; + this.exclude = exclude; + this.result = result; + } + + public static OperationOverride exclude(String operationName) { + return new OperationOverride(operationName, true, null); + } + + public static OperationOverride withResult(String operationName, Function result) { + return new OperationOverride(operationName, false, result); + } + + public String operationName() { + return operationName; + } + + public boolean isExclude() { + return exclude; + } + + public Function result() { + return result; + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/OperationRecord.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/OperationRecord.java new file mode 100644 index 000000000..c72096a68 --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/OperationRecord.java @@ -0,0 +1,223 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import java.util.LinkedHashMap; +import java.util.Map; + +/** + * A single operation within an execution (step, wait, invoke, callback, or context). + * + *

Mirrors the JS {@code OperationRecord} interface field-for-field so the emitted wire JSON is identical. A + * {@code null} field is treated as absent and omitted from the wire map (distinct from an explicit JSON null). + * + * @deprecated This is a preview API that is experimental and may be changed or removed in future releases. + */ +@Deprecated +public final class OperationRecord { + private String id; + private String name; + private String type; + private String subType; + private String parentId; + private String status; + private String startTime; + private String endTime; + private Long durationMs; + private Integer attempt; + private ErrorInfo error; + private Object result; + private Boolean truncated; + + public String id() { + return id; + } + + public OperationRecord id(String v) { + this.id = v; + return this; + } + + public String name() { + return name; + } + + public OperationRecord name(String v) { + this.name = v; + return this; + } + + public String type() { + return type; + } + + public OperationRecord type(String v) { + this.type = v; + return this; + } + + public String subType() { + return subType; + } + + public OperationRecord subType(String v) { + this.subType = v; + return this; + } + + public String parentId() { + return parentId; + } + + public OperationRecord parentId(String v) { + this.parentId = v; + return this; + } + + public String status() { + return status; + } + + public OperationRecord status(String v) { + this.status = v; + return this; + } + + public String startTime() { + return startTime; + } + + public OperationRecord startTime(String v) { + this.startTime = v; + return this; + } + + public String endTime() { + return endTime; + } + + public OperationRecord endTime(String v) { + this.endTime = v; + return this; + } + + public Long durationMs() { + return durationMs; + } + + public OperationRecord durationMs(Long v) { + this.durationMs = v; + return this; + } + + public Integer attempt() { + return attempt; + } + + public OperationRecord attempt(Integer v) { + this.attempt = v; + return this; + } + + public ErrorInfo error() { + return error; + } + + public OperationRecord error(ErrorInfo v) { + this.error = v; + return this; + } + + public Object result() { + return result; + } + + public OperationRecord result(Object v) { + this.result = v; + return this; + } + + public Boolean truncated() { + return truncated; + } + + public OperationRecord truncated(Boolean v) { + this.truncated = v; + return this; + } + + /** Shallow copy — used by the size limiter, which must not mutate the shared record. */ + public OperationRecord copy() { + OperationRecord c = new OperationRecord(); + c.id = id; + c.name = name; + c.type = type; + c.subType = subType; + c.parentId = parentId; + c.status = status; + c.startTime = startTime; + c.endTime = endTime; + c.durationMs = durationMs; + c.attempt = attempt; + c.error = error; + c.result = result; + c.truncated = truncated; + return c; + } + + /** + * Deep copy for per-exporter isolation: like {@link #copy()} but the {@code result} payload's mutable container + * structure is rebuilt so one exporter cannot mutate a later exporter's copy. {@code error} is an immutable + * {@link ErrorInfo} and is shared safely. + */ + public OperationRecord deepCopy() { + OperationRecord c = copy(); + c.result = Json.deepCopyContent(result); + return c; + } + + /** Serializes to the camelCase wire map, omitting absent (null) fields, in the JS field order. */ + public Map toWireMap() { + Map data = new LinkedHashMap<>(); + if (id != null) { + data.put("id", id); + } + if (name != null) { + data.put("name", name); + } + if (type != null) { + data.put("type", type); + } + if (subType != null) { + data.put("subType", subType); + } + if (parentId != null) { + data.put("parentId", parentId); + } + if (status != null) { + data.put("status", status); + } + if (startTime != null) { + data.put("startTime", startTime); + } + if (endTime != null) { + data.put("endTime", endTime); + } + if (durationMs != null) { + data.put("durationMs", durationMs); + } + if (attempt != null) { + data.put("attempt", attempt); + } + if (error != null) { + data.put("error", error.toWireMap()); + } + if (result != null) { + data.put("result", result); + } + if (truncated != null) { + data.put("truncated", truncated); + } + return data; + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/OperationSummary.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/OperationSummary.java new file mode 100644 index 000000000..cf1dd1186 --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/OperationSummary.java @@ -0,0 +1,66 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import java.util.LinkedHashMap; +import java.util.Map; + +/** + * A per-operation-name summary emitted (as an {@code operationsByName} map) by point-access exporters (CloudWatch + * Logs). + * + *

Mirrors the JS {@code OperationSummary} interface. Metric fields aggregate across all occurrences of the name; + * {@code type}, {@code subType}, {@code status} reflect the most recently seen occurrence; {@code result}/{@code error} + * are included only when the name occurs exactly once. + * + * @deprecated This is a preview API that is experimental and may be changed or removed in future releases. + */ +@Deprecated +public final class OperationSummary { + String type; + String subType; + int count; + Long minDurationMs; + Long maxDurationMs; + Long totalDurationMs; + int failedCount; + Integer maxAttempt; + String status; + Object result; + ErrorInfo error; + + /** Serializes to the camelCase wire map, omitting absent (null) fields, in the JS field order. */ + public Map toWireMap() { + Map data = new LinkedHashMap<>(); + if (type != null) { + data.put("type", type); + } + if (subType != null) { + data.put("subType", subType); + } + data.put("count", count); + if (minDurationMs != null) { + data.put("minDurationMs", minDurationMs); + } + if (maxDurationMs != null) { + data.put("maxDurationMs", maxDurationMs); + } + if (totalDurationMs != null) { + data.put("totalDurationMs", totalDurationMs); + } + data.put("failedCount", failedCount); + if (maxAttempt != null) { + data.put("maxAttempt", maxAttempt); + } + if (status != null) { + data.put("status", status); + } + if (result != null) { + data.put("result", result); + } + if (error != null) { + data.put("error", error.toWireMap()); + } + return data; + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/OperationsIndex.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/OperationsIndex.java new file mode 100644 index 000000000..6f0840ff4 --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/OperationsIndex.java @@ -0,0 +1,90 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** + * Builds a name-keyed index of operation summaries from the canonical operations array. + * + *

Ports the JS {@code buildOperationsByName} logic exactly: on the first occurrence of a name insert a summary + * (including its {@code result}/{@code error}); on a repeated name aggregate the metrics and drop {@code result} and + * {@code error} (no single representative value). Scalar fields ({@code type}, {@code subType}, {@code status}) reflect + * the most recently seen occurrence. Unnamed operations are skipped. + * + * @deprecated This is a preview API that is experimental and may be changed or removed in future releases. + */ +@Deprecated +public final class OperationsIndex { + + private OperationsIndex() {} + + public static Map buildOperationsByName(List operations) { + // Insertion-ordered so the emitted map mirrors first-seen order (matches the JS Map iteration order). + Map groups = new LinkedHashMap<>(); + + for (OperationRecord op : operations) { + if (op.name() == null) { + continue; + } + Long duration = op.durationMs(); + Integer attempt = op.attempt(); + int failed = "FAILED".equals(op.status()) ? 1 : 0; + + OperationSummary existing = groups.get(op.name()); + if (existing == null) { + OperationSummary s = new OperationSummary(); + s.type = op.type(); + s.count = 1; + s.failedCount = failed; + s.status = op.status(); + if (op.subType() != null) { + s.subType = op.subType(); + } + if (duration != null) { + s.minDurationMs = duration; + s.maxDurationMs = duration; + s.totalDurationMs = duration; + } + if (attempt != null) { + s.maxAttempt = attempt; + } + if (op.result() != null) { + s.result = op.result(); + } + if (op.error() != null) { + s.error = op.error(); + } + groups.put(op.name(), s); + continue; + } + + // Repeated name: aggregate and drop the per-occurrence result/error. + existing.count += 1; + existing.failedCount += failed; + existing.type = op.type(); + existing.status = op.status(); + if (op.subType() != null) { + existing.subType = op.subType(); + } else { + existing.subType = null; + } + if (duration != null) { + existing.minDurationMs = + existing.minDurationMs == null ? duration : Math.min(existing.minDurationMs, duration); + existing.maxDurationMs = + existing.maxDurationMs == null ? duration : Math.max(existing.maxDurationMs, duration); + existing.totalDurationMs = + (existing.totalDurationMs == null ? 0L : existing.totalDurationMs) + duration; + } + if (attempt != null) { + existing.maxAttempt = existing.maxAttempt == null ? attempt : Math.max(existing.maxAttempt, attempt); + } + existing.result = null; + existing.error = null; + } + return groups; + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/Truncation.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/Truncation.java new file mode 100644 index 000000000..4c2278eb9 --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/Truncation.java @@ -0,0 +1,162 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.function.Function; + +/** + * Best-effort size limiter, porting the JS {@code truncateRecord} drop order: + * + *

    + *
  1. operation {@code result} fields, oldest operation first; + *
  2. whole operations, oldest first; + *
  3. last resort — execution {@code input}, then {@code output}. + *
+ * + *

Identity/timeline fields are never dropped. When anything is dropped the returned record has {@code truncated: + * true}; each operation whose result was dropped is itself marked {@code truncated: true}, and + * {@code droppedOperations} / {@code droppedInput} / {@code droppedOutput} markers are set as applicable. The input + * record is never mutated. {@code render} maps the record to the exact shape the exporter serializes so the size check + * measures what is actually emitted. + * + * @deprecated This is a preview API that is experimental and may be changed or removed in future releases. + */ +@Deprecated +public final class Truncation { + + private Truncation() {} + + public static WorkflowInsightRecord truncateRecord( + WorkflowInsightRecord record, Integer maxBytes, Function render) { + if (maxBytes == null || maxBytes <= 0) { + return record; + } + Integer initialSize = Json.byteSize(render.apply(record)); + if (initialSize == null || initialSize <= maxBytes) { + return record; + } + + List ops = new ArrayList<>(record.operations().size()); + for (OperationRecord op : record.operations()) { + ops.add(op.copy()); + } + boolean[] kept = new boolean[ops.size()]; + java.util.Arrays.fill(kept, true); + int[] order = oldestFirstOrder(ops); + + boolean[] anyResultDropped = {false}; + int[] droppedOperations = {0}; + boolean[] droppedInput = {false}; + boolean[] droppedOutput = {false}; + + // Phase 1: drop operation results, oldest first (kept + marked truncated). + for (int idx : order) { + if (fits(record, ops, kept, droppedOperations[0], droppedInput[0], droppedOutput[0], render, maxBytes)) { + break; + } + if (kept[idx] && ops.get(idx).result() != null) { + ops.set(idx, ops.get(idx).copy().result(null).truncated(true)); + anyResultDropped[0] = true; + } + } + + // Phase 2: drop whole operations, oldest first. + for (int idx : order) { + if (fits(record, ops, kept, droppedOperations[0], droppedInput[0], droppedOutput[0], render, maxBytes)) { + break; + } + if (kept[idx]) { + kept[idx] = false; + droppedOperations[0]++; + } + } + + // Phase 3 (last resort): drop execution input, then output. + if (!fits(record, ops, kept, droppedOperations[0], droppedInput[0], droppedOutput[0], render, maxBytes) + && record.input != null) { + droppedInput[0] = true; + } + if (!fits(record, ops, kept, droppedOperations[0], droppedInput[0], droppedOutput[0], render, maxBytes) + && record.output != null) { + droppedOutput[0] = true; + } + + if (!anyResultDropped[0] && droppedOperations[0] == 0 && !droppedInput[0] && !droppedOutput[0]) { + return record; + } + return candidate(record, ops, kept, droppedOperations[0], droppedInput[0], droppedOutput[0]); + } + + private static boolean fits( + WorkflowInsightRecord record, + List ops, + boolean[] kept, + int droppedOperations, + boolean droppedInput, + boolean droppedOutput, + Function render, + int maxBytes) { + Integer size = Json.byteSize( + render.apply(candidate(record, ops, kept, droppedOperations, droppedInput, droppedOutput))); + return size != null && size <= maxBytes; + } + + private static WorkflowInsightRecord candidate( + WorkflowInsightRecord record, + List ops, + boolean[] kept, + int droppedOperations, + boolean droppedInput, + boolean droppedOutput) { + WorkflowInsightRecord out = record.copy(); + List retained = new ArrayList<>(); + for (int i = 0; i < ops.size(); i++) { + if (kept[i]) { + retained.add(ops.get(i)); + } + } + out.operations = retained; + out.truncated = Boolean.TRUE; + if (droppedOperations > 0) { + out.droppedOperations = droppedOperations; + } + if (droppedInput) { + out.input = null; + out.droppedInput = Boolean.TRUE; + } + if (droppedOutput) { + out.output = null; + out.droppedOutput = Boolean.TRUE; + } + return out; + } + + /** Oldest-first by startTime ascending; operations without a parseable startTime are treated as newest. */ + private static int[] oldestFirstOrder(List operations) { + List indices = new ArrayList<>(); + for (int i = 0; i < operations.size(); i++) { + indices.add(i); + } + indices.sort(Comparator.comparingDouble((Integer i) -> startKey(operations.get(i))) + .thenComparingInt(i -> i)); + int[] order = new int[indices.size()]; + for (int i = 0; i < order.length; i++) { + order[i] = indices.get(i); + } + return order; + } + + private static double startKey(OperationRecord op) { + if (op.startTime() == null) { + return Double.POSITIVE_INFINITY; + } + try { + return java.time.Instant.parse(op.startTime()).toEpochMilli(); + } catch (RuntimeException e) { + return Double.POSITIVE_INFINITY; + } + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/WorkflowInsight.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/WorkflowInsight.java new file mode 100644 index 000000000..cef6d6e58 --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/WorkflowInsight.java @@ -0,0 +1,406 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Function; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import software.amazon.awssdk.services.lambda.model.ErrorObject; +import software.amazon.lambda.durable.exception.DurableOperationException; +import software.amazon.lambda.durable.insight.exporters.LambdaLogExporter; +import software.amazon.lambda.durable.plugin.DurableExecutionPlugin; +import software.amazon.lambda.durable.plugin.InvocationEndInfo; +import software.amazon.lambda.durable.plugin.InvocationInfo; +import software.amazon.lambda.durable.plugin.InvocationStatus; +import software.amazon.lambda.durable.plugin.OperationChangeInfo; +import software.amazon.lambda.durable.plugin.OperationChangeItemInfo; + +/** + * Workflow Insight instrumentation plugin for the Durable Execution Java SDK. + * + *

Ports the JS {@code workflowInsight()} emission model to the Java plugin hook surface. Each record is built from + * the current invocation operation snapshot the SDK hands the plugin — {@link InvocationInfo#operations()} at + * invocation start / operation change, and {@link InvocationEndInfo#operations()} at invocation end — rather than from + * per-hook accumulation. Execution input/output flow through {@link InvocationInfo#executionInput()} and + * {@link InvocationEndInfo#executionResult()}, and per-operation results through + * {@link OperationChangeItemInfo#result()}; these are the fields PR #618 surfaced on the hook records, so + * {@code input}, {@code output}, and operation {@code result} are now populated exactly as in the JS plugin. + * + *

Per-execution state (keyed by execution ARN) holds only the stable start time, the parsed ARN, the cached input, + * and the one-time sampling decision. State is preserved across non-terminal (PENDING/RETRYING) invocations so + * suspend/resume keeps a single stable start time, and is removed only once the execution reaches a terminal state. + * + * @deprecated This is a preview API that is experimental and may be changed or removed in future releases. + */ +@Deprecated +public final class WorkflowInsight { + + private static final Logger logger = LoggerFactory.getLogger(WorkflowInsight.class); + + private WorkflowInsight() {} + + /** Creates a Workflow Insight plugin from the given config. Mirrors the JS {@code workflowInsight(config)}. */ + public static DurableExecutionPlugin workflowInsight(WorkflowInsightConfig config) { + return new InsightPlugin(config); + } + + /** Per-execution state, keyed by execution ARN, to prevent warm-container bleed and handle resume. */ + private static final class ExecutionState { + final Instant startTime; + final ArnParser arn; + final boolean sampledIn; + volatile Object cachedInput; + + ExecutionState(Instant startTime, ArnParser arn, boolean sampledIn) { + this.startTime = startTime; + this.arn = arn; + this.sampledIn = sampledIn; + } + } + + private static final class InsightPlugin implements DurableExecutionPlugin { + private final double samplingRate; + private final WorkflowInsightConfig.EmitMode emitMode; + private final boolean topLevelOnly; + private final boolean includeErrors; + private final ContentConfig content; + private final Map overridesByName = new LinkedHashMap<>(); + private final List exporters; + + private final Map byArn = new ConcurrentHashMap<>(); + + InsightPlugin(WorkflowInsightConfig config) { + this.samplingRate = resolveSamplingRate(config.samplingRate()); + this.emitMode = config.emitMode() != null ? config.emitMode() : WorkflowInsightConfig.EmitMode.ON_COMPLETE; + this.topLevelOnly = config.operationDetail() != WorkflowInsightConfig.OperationDetail.FULL_TREE; + this.content = config.content(); + this.includeErrors = content == null || content.includeErrors(); + if (content != null) { + for (OperationOverride o : content.overrides()) { + overridesByName.put(o.operationName(), o); + } + } + this.exporters = + config.exporters().isEmpty() ? List.of(new LambdaLogExporter()) : List.copyOf(config.exporters()); + } + + private ExecutionState getState(String arn, Instant startTime) { + return byArn.computeIfAbsent( + arn, a -> new ExecutionState(startTime, ArnParser.parse(a), shouldSample(a, samplingRate))); + } + + @Override + public void onInvocationStart(InvocationInfo info) { + ExecutionState state = getState(info.durableExecutionArn(), info.executionStartTime()); + if (!state.sampledIn) { + return; + } + state.cachedInput = info.executionInput(); + if (emitMode == WorkflowInsightConfig.EmitMode.ON_CHANGE) { + emit(buildRecord( + state, + info.durableExecutionArn(), + "RUNNING", + info.operations(), + null, + info.executionInput(), + null, + null)); + } + } + + @Override + public void onOperationChange(OperationChangeInfo info) { + if (emitMode != WorkflowInsightConfig.EmitMode.ON_CHANGE) { + return; + } + ExecutionState state = byArn.get(info.durableExecutionArn()); + if (state == null || !state.sampledIn) { + return; + } + emit(buildRecord( + state, + info.durableExecutionArn(), + "RUNNING", + info.operations(), + null, + state.cachedInput, + null, + null)); + } + + @Override + public void onInvocationEnd(InvocationEndInfo info) { + ExecutionState state = getState(info.durableExecutionArn(), info.executionStartTime()); + String status = mapStatus(info.invocationStatus()); + boolean isTerminal = "SUCCEEDED".equals(status) || "FAILED".equals(status); + boolean isFailure = "FAILED".equals(status); + boolean shouldEmit; + switch (emitMode) { + case ON_CHANGE: + shouldEmit = true; + break; + case ON_FAILURE: + shouldEmit = isFailure; + break; + case ON_COMPLETE: + default: + shouldEmit = isTerminal; + break; + } + + if (state.sampledIn && shouldEmit) { + emit(buildRecord( + state, + info.durableExecutionArn(), + status, + info.operations(), + Instant.now(), + info.executionInput(), + info.executionResult(), + info.executionError())); + } + + // Only clear state once the execution is truly finished. onInvocationEnd also fires on non-terminal + // suspends (PENDING/RETRYING); clearing there would lose the original startTime/cachedInput across resumes. + if (isTerminal) { + byArn.remove(info.durableExecutionArn()); + } + } + + /** Serializes each record to every exporter, isolating failures so one exporter never blocks the others. */ + private void emit(WorkflowInsightRecord record) { + for (InsightExporter exporter : exporters) { + try { + // Give each exporter its own deep copy: truncation returns the original record when it already + // fits, so without this a custom exporter that mutates operations or nested content would corrupt + // every exporter that runs after it. + WorkflowInsightRecord isolated = record.deepCopy(); + WorkflowInsightRecord shaped = + Truncation.truncateRecord(isolated, exporter.maxRecordSizeBytes(), exporter::render); + exporter.export(shaped); + exporter.flush(); + } catch (RuntimeException e) { + logger.warn("[workflow-insight] exporter failed", e); + } + } + } + + private WorkflowInsightRecord buildRecord( + ExecutionState state, + String arn, + String status, + Map operations, + Instant endTime, + Object input, + Object output, + Throwable error) { + WorkflowInsightRecord record = new WorkflowInsightRecord(); + ArnParser a = state.arn; + record.emittedAt = Instant.now().toString(); + record.executionArn = arn; + record.executionName = emptyToNull(a.executionName()); + record.functionName = a.functionName(); + record.functionQualifier = a.qualifier(); + record.region = a.region(); + record.accountId = a.accountId(); + record.status = status; + record.startTime = state.startTime != null ? state.startTime.toString() : null; + if (endTime != null) { + record.endTime = endTime.toString(); + if (state.startTime != null) { + record.durationMs = endTime.toEpochMilli() - state.startTime.toEpochMilli(); + } + } + record.input = applyDataContent( + input, + content == null || content.includeInput(), + content == null ? null : content.inputTransform()); + record.output = applyDataContent( + output, + content == null || content.includeOutput(), + content == null ? null : content.outputTransform()); + if (error != null) { + record.error = toErrorInfo(error); + } + record.operations = buildOperationRecords(operations); + return record; + } + + private List buildOperationRecords(Map operations) { + List out = new ArrayList<>(); + if (operations == null) { + return out; + } + // The hook contract supplies a map with no iteration-order guarantee (the core snapshot originates from a + // concurrent map). Sort by startTimestamp ascending (null timestamps last), then by a stable operation id + // tie-breaker, so the emitted operations array is deterministic and OperationsIndex's "latest occurrence" + // scalar fields reflect true chronological order rather than arbitrary map iteration order. + List items = new ArrayList<>(operations.values()); + items.sort(Comparator.comparing( + OperationChangeItemInfo::startTimestamp, Comparator.nullsLast(Comparator.naturalOrder())) + .thenComparing(OperationChangeItemInfo::id, Comparator.nullsLast(Comparator.naturalOrder()))); + for (OperationChangeItemInfo item : items) { + // The SDK core tracks the invocation/execution itself as a pseudo-entry of type EXECUTION; it is not a + // customer operation and the record already carries the execution status/timing at top level. + if ("EXECUTION".equals(item.type())) { + continue; + } + // Unnamed operations can't be targeted or keyed — excluded by default (matches JS `if (!op.name)`). + if (item.name() == null) { + continue; + } + // top-level detail drops anything nested under a context (parallel branches, map items, nested steps). + if (topLevelOnly && item.parentId() != null) { + continue; + } + OperationOverride override = overridesByName.get(item.name()); + if (override != null && override.isExclude()) { + continue; + } + OperationRecord rec = new OperationRecord() + .id(item.id()) + .name(item.name()) + .type(item.type()) + .subType(item.subType()) + .parentId(item.parentId()) + .status(item.status() != null ? item.status().toString() : "UNKNOWN") + .startTime( + item.startTimestamp() != null + ? item.startTimestamp().toString() + : null) + .endTime( + item.endTimestamp() != null + ? item.endTimestamp().toString() + : null) + .attempt(item.attempt()); + if (item.startTimestamp() != null && item.endTimestamp() != null) { + rec.durationMs(item.endTimestamp().toEpochMilli() + - item.startTimestamp().toEpochMilli()); + } + if (includeErrors && item.error() != null) { + rec.error(toErrorInfo(item.error())); + } + // Results are omitted unless an override explicitly opts in via a transform (matches JS). + if (override != null && override.result() != null) { + rec.result(applyResultOverride(override.result(), item.result())); + } + out.add(rec); + } + return out; + } + } + + // --- helpers --- + + /** + * Applies a user-supplied result transform to an operation's checkpointed (serialized JSON) result. Parses the JSON + * before handing it to the transform, falling back to the raw string when it isn't valid JSON. User transforms are + * untrusted: a throwing transform omits the field rather than leaking the raw value or failing the execution. + */ + static Object applyResultOverride(Function transform, String rawResult) { + if (rawResult == null) { + return null; + } + Object parsed; + try { + parsed = Json.MAPPER.readValue(rawResult, Object.class); + } catch (RuntimeException | com.fasterxml.jackson.core.JsonProcessingException e) { + parsed = rawResult; + } + try { + return transform.apply(parsed); + } catch (RuntimeException e) { + return null; + } + } + + /** + * Resolves a {@code content.input}/{@code content.output} setting against a value: excluded means omit; a transform + * is applied (omit on throw so a failing redactor never leaks the raw value); otherwise include as-is. + */ + static Object applyDataContent(Object value, boolean include, Function transform) { + if (!include || value == null) { + return null; + } + if (transform != null) { + try { + return transform.apply(value); + } catch (RuntimeException e) { + return null; + } + } + return value; + } + + private static ErrorInfo toErrorInfo(Throwable t) { + // Operation and execution snapshot errors are exposed as DurableOperationException wrappers, so the wrapper's + // own class/message would lose the original checkpointed failure identity. When the checkpointed ErrorObject is + // present, derive name/message from its errorType/errorMessage, falling back to the throwable's own fields for + // any value the ErrorObject leaves null. + if (t instanceof DurableOperationException doe && doe.getErrorObject() != null) { + ErrorObject error = doe.getErrorObject(); + String name = + error.errorType() != null ? error.errorType() : t.getClass().getSimpleName(); + String message = error.errorMessage() != null ? error.errorMessage() : t.getMessage(); + return new ErrorInfo(name, message); + } + return new ErrorInfo(t.getClass().getSimpleName(), t.getMessage()); + } + + private static String emptyToNull(String s) { + return s == null || s.isEmpty() ? null : s; + } + + private static String mapStatus(InvocationStatus status) { + if (status == InvocationStatus.SUCCEEDED) { + return "SUCCEEDED"; + } + if (status == InvocationStatus.FAILED) { + return "FAILED"; + } + // PENDING / RETRYING are still in flight from the execution's point of view. + return "RUNNING"; + } + + /** FNV-1a 32-bit hash, identical to the JS implementation (Java int multiply wraps mod 2^32 like Math.imul). */ + static int fnv1a32(String input) { + int hash = 0x811c9dc5; + for (int i = 0; i < input.length(); i++) { + hash ^= input.charAt(i); + hash *= 0x01000193; + } + return hash; + } + + static boolean shouldSample(String executionArn, double rate) { + if (rate >= 1) { + return true; + } + if (rate <= 0) { + return false; + } + long unsigned = fnv1a32(executionArn) & 0xffffffffL; + return (double) unsigned / 0xffffffffL < rate; + } + + static double resolveSamplingRate(Double rate) { + if (rate == null) { + return 1; + } + if (Double.isNaN(rate)) { + return 1; + } + if (rate < 0 || rate > 1) { + return Math.max(0, Math.min(1, rate)); + } + return rate; + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/WorkflowInsightConfig.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/WorkflowInsightConfig.java new file mode 100644 index 000000000..080912c44 --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/WorkflowInsightConfig.java @@ -0,0 +1,104 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import java.util.ArrayList; +import java.util.List; + +/** + * Configuration for the Workflow Insight plugin. Mirrors the JS {@code WorkflowInsightConfig}. + * + * @deprecated This is a preview API that is experimental and may be changed or removed in future releases. + */ +@Deprecated +public final class WorkflowInsightConfig { + + /** When records are emitted. */ + public enum EmitMode { + ON_COMPLETE, + ON_CHANGE, + ON_FAILURE + } + + /** Which operations to include in each record's operations array. */ + public enum OperationDetail { + TOP_LEVEL, + FULL_TREE + } + + private final List exporters; + private final Double samplingRate; + private final EmitMode emitMode; + private final OperationDetail operationDetail; + private final ContentConfig content; + + private WorkflowInsightConfig(Builder b) { + this.exporters = List.copyOf(b.exporters); + this.samplingRate = b.samplingRate; + this.emitMode = b.emitMode; + this.operationDetail = b.operationDetail; + this.content = b.content; + } + + public static Builder builder() { + return new Builder(); + } + + public List exporters() { + return exporters; + } + + public Double samplingRate() { + return samplingRate; + } + + public EmitMode emitMode() { + return emitMode; + } + + public OperationDetail operationDetail() { + return operationDetail; + } + + public ContentConfig content() { + return content; + } + + /** Builder for {@link WorkflowInsightConfig}. */ + public static final class Builder { + private final List exporters = new ArrayList<>(); + private Double samplingRate; + private EmitMode emitMode; + private OperationDetail operationDetail; + private ContentConfig content; + + public Builder addExporter(InsightExporter exporter) { + this.exporters.add(exporter); + return this; + } + + public Builder samplingRate(double rate) { + this.samplingRate = rate; + return this; + } + + public Builder emitMode(EmitMode mode) { + this.emitMode = mode; + return this; + } + + public Builder operationDetail(OperationDetail detail) { + this.operationDetail = detail; + return this; + } + + public Builder content(ContentConfig content) { + this.content = content; + return this; + } + + public WorkflowInsightConfig build() { + return new WorkflowInsightConfig(this); + } + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/WorkflowInsightRecord.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/WorkflowInsightRecord.java new file mode 100644 index 000000000..cbf6accc3 --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/WorkflowInsightRecord.java @@ -0,0 +1,208 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** + * The curated execution record emitted to destinations. + * + *

Mirrors the JS {@code WorkflowInsightRecord} interface field-for-field so the emitted wire JSON is identical + * (camelCase names, absent fields omitted). Exactly one of {@code operations} (array) or the {@code operationsByName} + * rendering is emitted, depending on the exporter's {@code render}. + * + * @deprecated This is a preview API that is experimental and may be changed or removed in future releases. + */ +@Deprecated +public final class WorkflowInsightRecord { + String recordType = "WorkflowInsight"; + String schemaVersion = "1.0"; + String emittedAt; + String executionArn; + String executionName; + String functionName; + String functionQualifier; + String region; + String accountId; + String status; + String startTime; + String endTime; + Long durationMs; + Object input; + Object output; + ErrorInfo error; + List operations = new ArrayList<>(); + Boolean truncated; + Integer droppedOperations; + Boolean droppedInput; + Boolean droppedOutput; + + public List operations() { + return Collections.unmodifiableList(operations); + } + + /** Package-private mutator to append an operation during record construction (public accessor stays immutable). */ + void addOperation(OperationRecord operation) { + operations.add(operation); + } + + public String executionArn() { + return executionArn; + } + + public String executionName() { + return executionName; + } + + public String functionName() { + return functionName; + } + + public String startTime() { + return startTime; + } + + public String status() { + return status; + } + + /** Shallow copy with a fresh operations list — used by the size limiter so the shared record is never mutated. */ + public WorkflowInsightRecord copy() { + WorkflowInsightRecord c = new WorkflowInsightRecord(); + c.recordType = recordType; + c.schemaVersion = schemaVersion; + c.emittedAt = emittedAt; + c.executionArn = executionArn; + c.executionName = executionName; + c.functionName = functionName; + c.functionQualifier = functionQualifier; + c.region = region; + c.accountId = accountId; + c.status = status; + c.startTime = startTime; + c.endTime = endTime; + c.durationMs = durationMs; + c.input = input; + c.output = output; + c.error = error; + c.operations = new ArrayList<>(operations); + c.truncated = truncated; + c.droppedOperations = droppedOperations; + c.droppedInput = droppedInput; + c.droppedOutput = droppedOutput; + return c; + } + + /** + * Deep copy for per-exporter isolation. Each operation record is deep-copied and the execution + * {@code input}/{@code output} payloads have their mutable container structure rebuilt, so a custom exporter that + * clears, redacts, or mutates operations or nested content cannot corrupt any exporter that runs after it. Because + * truncation returns the original record when it already fits, this copy is taken per exporter before shaping. + * {@code error} is an immutable {@link ErrorInfo} and is shared safely. + */ + public WorkflowInsightRecord deepCopy() { + WorkflowInsightRecord c = copy(); + List deepOps = new ArrayList<>(operations.size()); + for (OperationRecord op : operations) { + deepOps.add(op.deepCopy()); + } + c.operations = deepOps; + c.input = Json.deepCopyContent(input); + c.output = Json.deepCopyContent(output); + return c; + } + + /** Common scalar fields shared by the array and by-name renderings, in JS field order (absent fields omitted). */ + private Map baseWireMap() { + Map data = new LinkedHashMap<>(); + data.put("recordType", recordType); + data.put("schemaVersion", schemaVersion); + if (emittedAt != null) { + data.put("emittedAt", emittedAt); + } + if (executionArn != null) { + data.put("executionArn", executionArn); + } + if (executionName != null) { + data.put("executionName", executionName); + } + if (functionName != null) { + data.put("functionName", functionName); + } + if (functionQualifier != null) { + data.put("functionQualifier", functionQualifier); + } + if (region != null) { + data.put("region", region); + } + if (accountId != null) { + data.put("accountId", accountId); + } + if (status != null) { + data.put("status", status); + } + if (startTime != null) { + data.put("startTime", startTime); + } + if (endTime != null) { + data.put("endTime", endTime); + } + if (durationMs != null) { + data.put("durationMs", durationMs); + } + if (input != null) { + data.put("input", input); + } + if (output != null) { + data.put("output", output); + } + if (error != null) { + data.put("error", error.toWireMap()); + } + return data; + } + + private void putTruncationMarkers(Map data) { + if (truncated != null) { + data.put("truncated", truncated); + } + if (droppedOperations != null) { + data.put("droppedOperations", droppedOperations); + } + if (droppedInput != null) { + data.put("droppedInput", droppedInput); + } + if (droppedOutput != null) { + data.put("droppedOutput", droppedOutput); + } + } + + /** Canonical {@code operations}-array wire map (the shape S3Exporter serializes). */ + public Map toWireMap() { + Map data = baseWireMap(); + List> ops = new ArrayList<>(operations.size()); + for (OperationRecord op : operations) { + ops.add(op.toWireMap()); + } + data.put("operations", ops); + putTruncationMarkers(data); + return data; + } + + /** {@code operationsByName} wire map (the shape LambdaLogExporter / CloudWatchLogsExporter serialize). */ + public Map toByNameWireMap() { + Map data = baseWireMap(); + Map byName = new LinkedHashMap<>(); + for (Map.Entry e : + OperationsIndex.buildOperationsByName(operations).entrySet()) { + byName.put(e.getKey(), e.getValue().toWireMap()); + } + data.put("operationsByName", byName); + putTruncationMarkers(data); + return data; + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/CloudWatchLogsExporter.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/CloudWatchLogsExporter.java new file mode 100644 index 000000000..8d659be23 --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/CloudWatchLogsExporter.java @@ -0,0 +1,132 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight.exporters; + +import java.time.Instant; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.cloudwatchlogs.CloudWatchLogsClient; +import software.amazon.awssdk.services.cloudwatchlogs.CloudWatchLogsClientBuilder; +import software.amazon.awssdk.services.cloudwatchlogs.model.CreateLogStreamRequest; +import software.amazon.awssdk.services.cloudwatchlogs.model.InputLogEvent; +import software.amazon.awssdk.services.cloudwatchlogs.model.PutLogEventsRequest; +import software.amazon.awssdk.services.cloudwatchlogs.model.ResourceAlreadyExistsException; +import software.amazon.lambda.durable.insight.InsightExporter; +import software.amazon.lambda.durable.insight.Json; +import software.amazon.lambda.durable.insight.WorkflowInsightRecord; + +/** + * Exports workflow insight records to a specific CloudWatch Logs group via PutLogEvents, emitting the + * {@code operationsByName} map. Mirrors the JS {@code CloudWatchLogsExporter}. Requires {@code logs:CreateLogStream} + * and {@code logs:PutLogEvents} on the target log group. + * + * @deprecated This is a preview API that is experimental and may be changed or removed in future releases. + */ +@Deprecated +public final class CloudWatchLogsExporter implements InsightExporter { + private final String logGroupName; + private final String logStreamPrefix; + private final Integer maxRecordSizeBytes; + private final CloudWatchLogsClient client; + private final Set createdStreams = new HashSet<>(); + + private CloudWatchLogsExporter(Builder b) { + this.logGroupName = b.logGroupName; + this.logStreamPrefix = b.logStreamPrefix != null ? b.logStreamPrefix : "workflow-insight/"; + this.maxRecordSizeBytes = b.maxRecordSizeBytes != null ? b.maxRecordSizeBytes : 256_000; + CloudWatchLogsClientBuilder cb = CloudWatchLogsClient.builder(); + if (b.region != null) { + cb = cb.region(Region.of(b.region)); + } + this.client = b.client != null ? b.client : cb.build(); + } + + public static Builder builder() { + return new Builder(); + } + + @Override + public Integer maxRecordSizeBytes() { + return maxRecordSizeBytes; + } + + @Override + public Object render(WorkflowInsightRecord record) { + return record.toByNameWireMap(); + } + + @Override + public void export(WorkflowInsightRecord record) { + String streamName = buildStreamName(); + ensureStream(streamName); + client.putLogEvents(PutLogEventsRequest.builder() + .logGroupName(logGroupName) + .logStreamName(streamName) + .logEvents(List.of(InputLogEvent.builder() + .timestamp(System.currentTimeMillis()) + .message(Json.stringify(render(record))) + .build())) + .build()); + } + + private String buildStreamName() { + java.time.ZonedDateTime d = Instant.now().atZone(java.time.ZoneOffset.UTC); + return String.format("%s%d/%02d/%02d", logStreamPrefix, d.getYear(), d.getMonthValue(), d.getDayOfMonth()); + } + + private void ensureStream(String streamName) { + if (createdStreams.contains(streamName)) { + return; + } + try { + client.createLogStream(CreateLogStreamRequest.builder() + .logGroupName(logGroupName) + .logStreamName(streamName) + .build()); + } catch (ResourceAlreadyExistsException ignored) { + // stream already exists — fine + } + createdStreams.add(streamName); + } + + /** Builder for {@link CloudWatchLogsExporter}. */ + public static final class Builder { + private String logGroupName; + private String logStreamPrefix; + private String region; + private Integer maxRecordSizeBytes; + private CloudWatchLogsClient client; + + public Builder logGroupName(String logGroupName) { + this.logGroupName = logGroupName; + return this; + } + + public Builder logStreamPrefix(String logStreamPrefix) { + this.logStreamPrefix = logStreamPrefix; + return this; + } + + public Builder region(String region) { + this.region = region; + return this; + } + + public Builder maxRecordSizeBytes(Integer maxRecordSizeBytes) { + this.maxRecordSizeBytes = maxRecordSizeBytes; + return this; + } + + /** Test seam: inject a client. */ + public Builder client(CloudWatchLogsClient client) { + this.client = client; + return this; + } + + public CloudWatchLogsExporter build() { + return new CloudWatchLogsExporter(this); + } + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/LambdaLogExporter.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/LambdaLogExporter.java new file mode 100644 index 000000000..d83291dfb --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/LambdaLogExporter.java @@ -0,0 +1,44 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight.exporters; + +import software.amazon.lambda.durable.insight.InsightExporter; +import software.amazon.lambda.durable.insight.Json; +import software.amazon.lambda.durable.insight.WorkflowInsightRecord; + +/** + * Exports workflow insight records to CloudWatch Logs via {@code System.out.println} (Lambda captures stdout to the + * function's own log group, so this needs no extra IAM). Emits the {@code operationsByName} map as ONE single-line JSON + * record, mirroring the JS {@code LambdaLogExporter} (which uses {@code console.log}). The conformance CloudWatch sink + * decodes both raw top-level JSON lines and the Lambda structured-logging envelope. + * + * @deprecated This is a preview API that is experimental and may be changed or removed in future releases. + */ +@Deprecated +public final class LambdaLogExporter implements InsightExporter { + /** CloudWatch Logs caps a single log event at 256 KB. */ + private final Integer maxRecordSizeBytes; + + public LambdaLogExporter() { + this(256_000); + } + + public LambdaLogExporter(Integer maxRecordSizeBytes) { + this.maxRecordSizeBytes = maxRecordSizeBytes; + } + + @Override + public Integer maxRecordSizeBytes() { + return maxRecordSizeBytes; + } + + @Override + public Object render(WorkflowInsightRecord record) { + return record.toByNameWireMap(); + } + + @Override + public void export(WorkflowInsightRecord record) { + System.out.println(Json.stringify(render(record))); + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/S3Exporter.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/S3Exporter.java new file mode 100644 index 000000000..d767fc6db --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/S3Exporter.java @@ -0,0 +1,139 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight.exporters; + +import software.amazon.awssdk.core.sync.RequestBody; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.s3.S3Client; +import software.amazon.awssdk.services.s3.S3ClientBuilder; +import software.amazon.awssdk.services.s3.model.PutObjectRequest; +import software.amazon.lambda.durable.insight.InsightExporter; +import software.amazon.lambda.durable.insight.Json; +import software.amazon.lambda.durable.insight.WorkflowInsightRecord; + +/** + * Exports workflow insight records to Amazon S3, one JSON object per execution (keyed by execution name so updates + * overwrite the same object). Emits the canonical {@code operations}-array wire shape. Mirrors the JS + * {@code S3Exporter}. + * + * @deprecated This is a preview API that is experimental and may be changed or removed in future releases. + */ +@Deprecated +public final class S3Exporter implements InsightExporter { + + /** How to partition objects in S3. */ + public enum Partitioning { + DATE, + FUNCTION_NAME, + NONE + } + + private final String bucket; + private final String prefix; + private final Partitioning partitioning; + private final Integer maxRecordSizeBytes; + private final S3Client client; + + private S3Exporter(Builder b) { + this.bucket = b.bucket; + this.prefix = b.prefix != null ? b.prefix : "workflow-insight/"; + this.partitioning = b.partitioning != null ? b.partitioning : Partitioning.DATE; + this.maxRecordSizeBytes = b.maxRecordSizeBytes != null ? b.maxRecordSizeBytes : 5_000_000; + S3ClientBuilder cb = S3Client.builder(); + if (b.region != null) { + cb = cb.region(Region.of(b.region)); + } + this.client = b.client != null ? b.client : cb.build(); + } + + public static Builder builder() { + return new Builder(); + } + + @Override + public Integer maxRecordSizeBytes() { + return maxRecordSizeBytes; + } + + @Override + public void export(WorkflowInsightRecord record) { + String key = buildKey(record); + String body = Json.stringify(record.toWireMap()); + client.putObject( + PutObjectRequest.builder() + .bucket(bucket) + .key(key) + .contentType("application/json") + .build(), + RequestBody.fromString(body)); + } + + private String buildKey(WorkflowInsightRecord record) { + String fileName = + sanitize(record.executionName() != null ? record.executionName() : record.executionArn()) + ".json"; + return prefix + buildPartition(record) + fileName; + } + + private String buildPartition(WorkflowInsightRecord record) { + switch (partitioning) { + case DATE: + java.time.ZonedDateTime d = + java.time.Instant.parse(record.startTime()).atZone(java.time.ZoneOffset.UTC); + return String.format("year=%d/month=%02d/day=%02d/", d.getYear(), d.getMonthValue(), d.getDayOfMonth()); + case FUNCTION_NAME: + return "function=" + sanitize(record.functionName()) + "/"; + case NONE: + default: + return ""; + } + } + + private static String sanitize(String value) { + return value.replaceAll("[^a-zA-Z0-9._-]", "_"); + } + + /** Builder for {@link S3Exporter}. */ + public static final class Builder { + private String bucket; + private String prefix; + private Partitioning partitioning; + private String region; + private Integer maxRecordSizeBytes; + private S3Client client; + + public Builder bucket(String bucket) { + this.bucket = bucket; + return this; + } + + public Builder prefix(String prefix) { + this.prefix = prefix; + return this; + } + + public Builder partitioning(Partitioning partitioning) { + this.partitioning = partitioning; + return this; + } + + public Builder region(String region) { + this.region = region; + return this; + } + + public Builder maxRecordSizeBytes(Integer maxRecordSizeBytes) { + this.maxRecordSizeBytes = maxRecordSizeBytes; + return this; + } + + /** Test seam: inject a client. */ + public Builder client(S3Client client) { + this.client = client; + return this; + } + + public S3Exporter build() { + return new S3Exporter(this); + } + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/ArnParserTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/ArnParserTest.java new file mode 100644 index 000000000..7bcebca9f --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/ArnParserTest.java @@ -0,0 +1,22 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import org.junit.jupiter.api.Test; + +class ArnParserTest { + + @Test + void parsesAllComponentsFromDurableExecutionArn() { + String arn = "arn:aws:lambda:us-west-2:590183769840:function:my-fn:$LATEST" + + "/durable-execution/exec-abc/invocation-1"; + ArnParser p = ArnParser.parse(arn); + assertEquals("us-west-2", p.region()); + assertEquals("590183769840", p.accountId()); + assertEquals("my-fn", p.functionName()); + assertEquals("$LATEST", p.qualifier()); + assertEquals("exec-abc", p.executionName()); + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/CloudWatchLogsExporterTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/CloudWatchLogsExporterTest.java new file mode 100644 index 000000000..7fc4db2dc --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/CloudWatchLogsExporterTest.java @@ -0,0 +1,65 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import software.amazon.awssdk.services.cloudwatchlogs.CloudWatchLogsClient; +import software.amazon.awssdk.services.cloudwatchlogs.model.CreateLogStreamRequest; +import software.amazon.awssdk.services.cloudwatchlogs.model.CreateLogStreamResponse; +import software.amazon.awssdk.services.cloudwatchlogs.model.PutLogEventsRequest; +import software.amazon.awssdk.services.cloudwatchlogs.model.PutLogEventsResponse; +import software.amazon.lambda.durable.insight.exporters.CloudWatchLogsExporter; + +class CloudWatchLogsExporterTest { + + private WorkflowInsightRecord sampleRecord() { + WorkflowInsightRecord r = new WorkflowInsightRecord(); + r.executionArn = "arn:aws:lambda:us-west-2:1:function:f:$LATEST/durable-execution/exec-1/invocation-1"; + r.status = "SUCCEEDED"; + r.startTime = "2026-08-05T00:00:00Z"; + r.addOperation(new OperationRecord() + .id("op-1") + .name("greet") + .type("STEP") + .subType("Step") + .status("SUCCEEDED")); + return r; + } + + @Test + void createsStreamOnceAndPutsOperationsByNameEvent() { + CloudWatchLogsClient client = mock(CloudWatchLogsClient.class); + when(client.createLogStream(any(CreateLogStreamRequest.class))) + .thenReturn(CreateLogStreamResponse.builder().build()); + when(client.putLogEvents(any(PutLogEventsRequest.class))) + .thenReturn(PutLogEventsResponse.builder().build()); + + CloudWatchLogsExporter exporter = CloudWatchLogsExporter.builder() + .logGroupName("/my/group") + .client(client) + .build(); + + exporter.export(sampleRecord()); + exporter.export(sampleRecord()); + + // stream created once (cached), events put twice + verify(client, times(1)).createLogStream(any(CreateLogStreamRequest.class)); + ArgumentCaptor put = ArgumentCaptor.forClass(PutLogEventsRequest.class); + verify(client, times(2)).putLogEvents(put.capture()); + + PutLogEventsRequest req = put.getValue(); + assertEquals("/my/group", req.logGroupName()); + String message = req.logEvents().get(0).message(); + assertTrue(message.contains("operationsByName"), "CloudWatch emits the by-name map"); + assertTrue(!message.contains("\"operations\""), "CloudWatch must not emit the canonical array"); + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/ExporterIsolationTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/ExporterIsolationTest.java new file mode 100644 index 000000000..b9590d9a1 --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/ExporterIsolationTest.java @@ -0,0 +1,116 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.services.lambda.model.OperationStatus; +import software.amazon.lambda.durable.plugin.DurableExecutionPlugin; +import software.amazon.lambda.durable.plugin.InvocationInfo; +import software.amazon.lambda.durable.plugin.OperationChangeItemInfo; + +/** + * Finding {@code arf_v1_rzwdoyquezpr2qgby63imtmxcn} ([P2] prevent one exporter from mutating later exporters' records): + * because truncation returns the original record when it already fits, every exporter would otherwise share one mutable + * object. Each exporter must receive a deep copy so a hostile first exporter cannot corrupt records seen by exporters + * that run after it. + */ +class ExporterIsolationTest { + + private static final String ARN = + "arn:aws:lambda:us-west-2:1:function:f:$LATEST/durable-execution/exec-1/invocation-1"; + private static final Instant START = Instant.parse("2026-08-05T00:00:00Z"); + + private static final class MutablePayload { + public String value; + + MutablePayload(String value) { + this.value = value; + } + } + + /** A malicious exporter that mutates operation fields and nested input content. */ + private static final class MutatingExporter implements InsightExporter { + @Override + @SuppressWarnings("unchecked") + public void export(WorkflowInsightRecord record) { + for (OperationRecord op : record.operations()) { + op.name("HACKED").status("HACKED").result("HACKED"); + } + if (record.input instanceof Map input) { + ((Map) input).put("k", "HACKED"); + Object nested = input.get("pojo"); + if (nested instanceof Map nestedMap) { + ((Map) nestedMap).put("value", "HACKED"); + } else if (nested instanceof MutablePayload payload) { + payload.value = "HACKED"; + } + } + } + } + + private static final class CapturingExporter implements InsightExporter { + final List records = new ArrayList<>(); + + @Override + public void export(WorkflowInsightRecord record) { + records.add(record); + } + } + + @Test + void firstExporterMutationsDoNotLeakIntoLaterExporter() { + var mutating = new MutatingExporter(); + var good = new CapturingExporter(); + DurableExecutionPlugin plugin = WorkflowInsight.workflowInsight(WorkflowInsightConfig.builder() + .emitMode(WorkflowInsightConfig.EmitMode.ON_CHANGE) + .content(ContentConfig.builder() + .addOverride(OperationOverride.withResult("compute", r -> r)) + .build()) + .addExporter(mutating) + .addExporter(good) + .build()); + + Map input = new LinkedHashMap<>(); + input.put("k", "v"); + input.put("pojo", new MutablePayload("original")); + Map ops = new LinkedHashMap<>(); + ops.put( + "op-1", + new OperationChangeItemInfo( + "op-1", + "compute", + "STEP", + "Step", + null, + START, + START.plusMillis(5), + OperationStatus.SUCCEEDED, + 1, + false, + null, + "{\"x\":1}")); + plugin.onInvocationStart(new InvocationInfo("req", ARN, true, START, input, ops, Map.of())); + + assertEquals(1, good.records.size()); + var rec = good.records.get(0); + OperationRecord op = rec.operations().get(0); + assertEquals("compute", op.name(), "operation name not corrupted by the earlier exporter"); + assertEquals("SUCCEEDED", op.status(), "operation status not corrupted by the earlier exporter"); + assertInstanceOf(Map.class, op.result(), "operation result payload preserved"); + assertEquals(1, ((Map) op.result()).get("x"), "nested result content not corrupted"); + assertNotNull(rec.input); + var copiedInput = assertInstanceOf(Map.class, rec.input); + assertEquals("v", copiedInput.get("k"), "nested input content not corrupted by the earlier exporter"); + var copiedPojo = assertInstanceOf(Map.class, copiedInput.get("pojo")); + assertEquals("original", copiedPojo.get("value"), "mutable POJO content not corrupted by the earlier exporter"); + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/JsonJavaTimeTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/JsonJavaTimeTest.java new file mode 100644 index 000000000..331a26a14 --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/JsonJavaTimeTest.java @@ -0,0 +1,90 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.services.lambda.model.OperationStatus; +import software.amazon.lambda.durable.plugin.DurableExecutionPlugin; +import software.amazon.lambda.durable.plugin.InvocationInfo; +import software.amazon.lambda.durable.plugin.OperationChangeItemInfo; + +/** + * Finding {@code arf_v1_blkdgaotf7rzua2ga3uyouqm5e} ([P1] support the SDK's default payload types when exporting): the + * mapper must serialize Java-time values (as {@code JacksonSerDes} does) so an included {@code Instant} in an + * input/output/result serializes to ISO-8601 instead of throwing and silently dropping the record. + */ +class JsonJavaTimeTest { + + private static final Instant TS = Instant.parse("2026-08-05T12:34:56Z"); + private static final String ARN = + "arn:aws:lambda:us-west-2:1:function:f:$LATEST/durable-execution/exec-1/invocation-1"; + private static final Instant START = Instant.parse("2026-08-05T00:00:00Z"); + + @Test + void instantSerializesAsIso8601String() { + assertEquals("\"2026-08-05T12:34:56Z\"", Json.stringify(TS)); + } + + @Test + void byteSizeOfPayloadContainingInstantIsNotNull() { + Map payload = new LinkedHashMap<>(); + payload.put("when", TS); + payload.put("label", "created"); + Integer size = Json.byteSize(payload); + assertNotNull(size, "a payload containing an Instant must be measurable, not silently unserializable"); + assertTrue(size > 0); + } + + private static final class CapturingExporter implements InsightExporter { + final List records = new ArrayList<>(); + + @Override + public void export(WorkflowInsightRecord record) { + records.add(record); + } + } + + @Test + void pluginOutputWithInstantInInputSerializesInsteadOfDropping() { + var exporter = new CapturingExporter(); + DurableExecutionPlugin plugin = WorkflowInsight.workflowInsight(WorkflowInsightConfig.builder() + .emitMode(WorkflowInsightConfig.EmitMode.ON_CHANGE) + .addExporter(exporter) + .build()); + + Map input = new LinkedHashMap<>(); + input.put("startedAt", TS); + Map ops = new LinkedHashMap<>(); + ops.put( + "op-1", + new OperationChangeItemInfo( + "op-1", + "greet", + "STEP", + "Step", + null, + START, + START.plusMillis(5), + OperationStatus.STARTED, + 1, + false, + null, + null)); + plugin.onInvocationStart(new InvocationInfo("req", ARN, true, START, input, ops, Map.of())); + + assertEquals(1, exporter.records.size()); + var rec = exporter.records.get(0); + // The emitted wire JSON must serialize (no exception, non-null size) and carry the ISO-8601 Instant. + assertNotNull(Json.byteSize(rec.toWireMap()), "record with an Instant input must serialize"); + assertTrue(Json.stringify(rec.toWireMap()).contains("2026-08-05T12:34:56Z")); + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/OperationErrorIdentityTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/OperationErrorIdentityTest.java new file mode 100644 index 000000000..5e67cd554 --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/OperationErrorIdentityTest.java @@ -0,0 +1,116 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.services.lambda.model.ErrorObject; +import software.amazon.awssdk.services.lambda.model.Operation; +import software.amazon.awssdk.services.lambda.model.OperationStatus; +import software.amazon.lambda.durable.exception.DurableOperationException; +import software.amazon.lambda.durable.plugin.DurableExecutionPlugin; +import software.amazon.lambda.durable.plugin.InvocationEndInfo; +import software.amazon.lambda.durable.plugin.InvocationStatus; +import software.amazon.lambda.durable.plugin.OperationChangeItemInfo; + +/** + * Finding {@code arf_v1_lndcqjp4vnrm6vjmmfzrqrfdin} ([P2] preserve the checkpointed operation error type): operation + * and execution snapshot errors arrive wrapped as {@code DurableOperationException}. The record must derive the error + * name/message from the wrapper's {@code ErrorObject} (the original checkpointed identity), falling back to the + * throwable's own fields only for values the {@code ErrorObject} leaves null. + */ +class OperationErrorIdentityTest { + + private static final String ARN = + "arn:aws:lambda:us-west-2:1:function:f:$LATEST/durable-execution/exec-1/invocation-1"; + private static final Instant START = Instant.parse("2026-08-05T00:00:00Z"); + + private static DurableOperationException wrapped(String errorType, String errorMessage) { + ErrorObject error = ErrorObject.builder() + .errorType(errorType) + .errorMessage(errorMessage) + .build(); + return new DurableOperationException(Operation.builder().id("op-1").build(), error); + } + + private static final class CapturingExporter implements InsightExporter { + final List records = new ArrayList<>(); + + @Override + public void export(WorkflowInsightRecord record) { + records.add(record); + } + } + + private Map failedOps(Throwable opError) { + Map m = new LinkedHashMap<>(); + m.put( + "op-1", + new OperationChangeItemInfo( + "op-1", + "failing-step", + "STEP", + "Step", + null, + START, + START.plusMillis(5), + OperationStatus.FAILED, + 1, + false, + opError, + null)); + return m; + } + + @Test + void operationAndExecutionErrorUseCheckpointedErrorObjectIdentity() { + var exporter = new CapturingExporter(); + DurableExecutionPlugin plugin = WorkflowInsight.workflowInsight( + WorkflowInsightConfig.builder().addExporter(exporter).build()); + + Throwable opError = wrapped("CustomerValidationError", "invalid postal code"); + Throwable execError = wrapped("OrchestrationFailure", "workflow aborted"); + plugin.onInvocationEnd(new InvocationEndInfo( + "req", ARN, true, START, failedOps(opError), InvocationStatus.FAILED, execError, "in", null)); + + assertEquals(1, exporter.records.size()); + var rec = exporter.records.get(0); + + // Execution-level error identity. + assertNotNull(rec.error); + assertEquals("OrchestrationFailure", rec.error.name()); + assertEquals("workflow aborted", rec.error.message()); + + // Operation-level error identity — exact original type and message, not the DurableOperationException wrapper. + OperationRecord op = rec.operations().get(0); + assertNotNull(op.error()); + assertEquals("CustomerValidationError", op.error().name()); + assertEquals("invalid postal code", op.error().message()); + } + + @Test + void fallsBackToThrowableFieldsWhenErrorObjectFieldsMissing() { + var exporter = new CapturingExporter(); + DurableExecutionPlugin plugin = WorkflowInsight.workflowInsight( + WorkflowInsightConfig.builder().addExporter(exporter).build()); + + // ErrorObject present but errorType null: name falls back to the throwable's simple class name. + ErrorObject partial = + ErrorObject.builder().errorMessage("only a message").build(); + Throwable opError = + new DurableOperationException(Operation.builder().id("op-1").build(), partial); + plugin.onInvocationEnd(new InvocationEndInfo( + "req", ARN, true, START, failedOps(opError), InvocationStatus.FAILED, opError, "in", null)); + + var rec = exporter.records.get(0); + assertEquals("DurableOperationException", rec.error.name(), "null errorType falls back to throwable class"); + assertEquals("only a message", rec.error.message()); + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/OperationOrderingTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/OperationOrderingTest.java new file mode 100644 index 000000000..6c9d121fb --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/OperationOrderingTest.java @@ -0,0 +1,111 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.services.lambda.model.OperationStatus; +import software.amazon.lambda.durable.plugin.DurableExecutionPlugin; +import software.amazon.lambda.durable.plugin.InvocationInfo; +import software.amazon.lambda.durable.plugin.OperationChangeItemInfo; + +/** + * Finding {@code arf_v1_ig2yofnonmfn7zknekoxwd6xmg} ([P2] establish a deterministic chronological operation order): the + * hook contract supplies a map with no iteration-order guarantee. Operations must be sorted by {@code startTimestamp} + * (null last) with a stable id tie-breaker before records are built, and the {@code operationsByName} "latest" scalar + * fields must reflect that chronological order. + */ +class OperationOrderingTest { + + private static final String ARN = + "arn:aws:lambda:us-west-2:1:function:f:$LATEST/durable-execution/exec-1/invocation-1"; + private static final Instant START = Instant.parse("2026-08-05T00:00:00Z"); + + private static final class CapturingExporter implements InsightExporter { + final List records = new ArrayList<>(); + + @Override + public void export(WorkflowInsightRecord record) { + records.add(record); + } + } + + private static OperationChangeItemInfo item( + String id, String name, String type, String subType, Instant start, OperationStatus status) { + Instant end = start == null ? null : start.plusMillis(1); + return new OperationChangeItemInfo(id, name, type, subType, null, start, end, status, 1, false, null, null); + } + + private WorkflowInsightRecord emitStart(Map ops) { + var exporter = new CapturingExporter(); + DurableExecutionPlugin plugin = WorkflowInsight.workflowInsight(WorkflowInsightConfig.builder() + .emitMode(WorkflowInsightConfig.EmitMode.ON_CHANGE) + .addExporter(exporter) + .build()); + plugin.onInvocationStart(new InvocationInfo("req", ARN, true, START, "in", ops, Map.of())); + return exporter.records.get(0); + } + + @Test + void shuffledStartTimestampsAreEmittedChronologically() { + // Inserted deliberately out of order: c (T+30), a (T+10), b (T+20). + Map ops = new LinkedHashMap<>(); + ops.put("c", item("c", "c-op", "STEP", "Step", START.plusMillis(30), OperationStatus.SUCCEEDED)); + ops.put("a", item("a", "a-op", "STEP", "Step", START.plusMillis(10), OperationStatus.SUCCEEDED)); + ops.put("b", item("b", "b-op", "STEP", "Step", START.plusMillis(20), OperationStatus.SUCCEEDED)); + + var rec = emitStart(ops); + List ids = rec.operations().stream().map(OperationRecord::id).toList(); + assertEquals(List.of("a", "b", "c"), ids, "operations sorted by startTimestamp ascending"); + } + + @Test + void equalTimestampsBreakTiesByStableId() { + Instant t = START.plusMillis(10); + Map ops = new LinkedHashMap<>(); + ops.put("z", item("z", "z-op", "STEP", "Step", t, OperationStatus.SUCCEEDED)); + ops.put("m", item("m", "m-op", "STEP", "Step", t, OperationStatus.SUCCEEDED)); + ops.put("a", item("a", "a-op", "STEP", "Step", t, OperationStatus.SUCCEEDED)); + + var rec = emitStart(ops); + List ids = rec.operations().stream().map(OperationRecord::id).toList(); + assertEquals(List.of("a", "m", "z"), ids, "equal timestamps break ties by ascending id"); + } + + @Test + void nullTimestampsSortLast() { + Map ops = new LinkedHashMap<>(); + ops.put("n", item("n", "n-op", "STEP", "Step", null, OperationStatus.STARTED)); + ops.put("b", item("b", "b-op", "STEP", "Step", START.plusMillis(20), OperationStatus.SUCCEEDED)); + ops.put("a", item("a", "a-op", "STEP", "Step", START.plusMillis(10), OperationStatus.SUCCEEDED)); + + var rec = emitStart(ops); + List ids = rec.operations().stream().map(OperationRecord::id).toList(); + assertEquals(List.of("a", "b", "n"), ids, "null startTimestamp sorts last"); + } + + @Test + void repeatedNameLatestScalarsReflectChronologicalOrderNotMapOrder() { + // Same name "task": later occurrence inserted FIRST so a naive values() walk would report the older one. + Map ops = new LinkedHashMap<>(); + ops.put("late", item("late", "task", "STEP", "StepV2", START.plusMillis(20), OperationStatus.SUCCEEDED)); + ops.put("early", item("early", "task", "STEP", "StepV1", START.plusMillis(10), OperationStatus.FAILED)); + + var rec = emitStart(ops); + Map byName = OperationsIndex.buildOperationsByName(rec.operations()); + OperationSummary task = byName.get("task"); + assertEquals(2, task.count); + // Chronologically last occurrence is the T+20 SUCCEEDED/StepV2 one. + assertEquals("SUCCEEDED", task.status, "latest status reflects the chronologically last occurrence"); + assertEquals("StepV2", task.subType, "latest subType reflects the chronologically last occurrence"); + assertEquals(1, task.failedCount, "the earlier FAILED occurrence is still counted"); + assertNull(task.result, "repeated name drops the representative result"); + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/OperationsIndexTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/OperationsIndexTest.java new file mode 100644 index 000000000..dfd150219 --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/OperationsIndexTest.java @@ -0,0 +1,98 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; + +class OperationsIndexTest { + + @Test + void singleOccurrenceKeepsResultAndError() { + OperationRecord op = new OperationRecord() + .id("a") + .name("compute") + .type("STEP") + .subType("Step") + .status("SUCCEEDED") + .attempt(1) + .result(42); + Map byName = OperationsIndex.buildOperationsByName(List.of(op)); + OperationSummary s = byName.get("compute"); + assertEquals(1, s.count); + assertEquals(0, s.failedCount); + assertEquals("STEP", s.type); + assertEquals(42, s.result); + } + + @Test + void repeatedNameAggregatesAndDropsResult() { + OperationRecord a = new OperationRecord() + .id("1") + .name("task") + .type("STEP") + .subType("Step") + .status("SUCCEEDED"); + OperationRecord b = new OperationRecord() + .id("2") + .name("task") + .type("STEP") + .subType("Step") + .status("SUCCEEDED"); + OperationRecord c = new OperationRecord() + .id("3") + .name("task") + .type("STEP") + .subType("Step") + .status("SUCCEEDED"); + Map byName = OperationsIndex.buildOperationsByName(List.of(a, b, c)); + OperationSummary s = byName.get("task"); + assertEquals(3, s.count); + assertEquals(0, s.failedCount); + assertNull(s.result); + assertNull(s.error); + } + + @Test + void maxAttemptReflectsHighestAttempt() { + OperationRecord retried = new OperationRecord() + .id("1") + .name("retried-step") + .type("STEP") + .subType("Step") + .status("SUCCEEDED") + .attempt(2); + Map byName = OperationsIndex.buildOperationsByName(List.of(retried)); + assertEquals(Integer.valueOf(2), byName.get("retried-step").maxAttempt); + assertEquals(1, byName.get("retried-step").count); + } + + @Test + void unnamedOperationsAreSkipped() { + OperationRecord named = + new OperationRecord().id("1").name("named").type("STEP").status("SUCCEEDED"); + OperationRecord unnamed = new OperationRecord().id("2").type("STEP").status("SUCCEEDED"); + Map byName = OperationsIndex.buildOperationsByName(List.of(named, unnamed)); + assertTrue(byName.containsKey("named")); + assertEquals(1, byName.size()); + } + + @Test + void failedOccurrenceCountsAsFailed() { + OperationRecord failed = new OperationRecord() + .id("1") + .name("failing") + .type("STEP") + .subType("Step") + .status("FAILED"); + Map byName = OperationsIndex.buildOperationsByName(List.of(failed)); + assertEquals(1, byName.get("failing").failedCount); + assertFalse(byName.get("failing").count == 0); + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/S3ExporterTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/S3ExporterTest.java new file mode 100644 index 000000000..37a7e93bf --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/S3ExporterTest.java @@ -0,0 +1,72 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import software.amazon.awssdk.core.sync.RequestBody; +import software.amazon.awssdk.services.s3.S3Client; +import software.amazon.awssdk.services.s3.model.PutObjectRequest; +import software.amazon.awssdk.services.s3.model.PutObjectResponse; +import software.amazon.lambda.durable.insight.exporters.S3Exporter; + +class S3ExporterTest { + + private WorkflowInsightRecord sampleRecord() { + WorkflowInsightRecord r = new WorkflowInsightRecord(); + r.executionArn = "arn:aws:lambda:us-west-2:1:function:f:$LATEST/durable-execution/exec-1/invocation-1"; + r.executionName = "exec-1"; + r.functionName = "f"; + r.status = "SUCCEEDED"; + r.startTime = "2026-08-05T00:00:00Z"; + r.addOperation(new OperationRecord() + .id("op-1") + .name("greet") + .type("STEP") + .subType("Step") + .status("SUCCEEDED")); + return r; + } + + @Test + void putsCanonicalOperationsArrayObjectKeyedByExecutionName() throws Exception { + S3Client client = mock(S3Client.class); + when(client.putObject(any(PutObjectRequest.class), any(RequestBody.class))) + .thenReturn(PutObjectResponse.builder().build()); + + S3Exporter exporter = S3Exporter.builder() + .bucket("my-bucket") + .partitioning(S3Exporter.Partitioning.NONE) + .client(client) + .build(); + exporter.export(sampleRecord()); + + ArgumentCaptor req = ArgumentCaptor.forClass(PutObjectRequest.class); + ArgumentCaptor body = ArgumentCaptor.forClass(RequestBody.class); + verify(client).putObject(req.capture(), body.capture()); + + assertEquals("my-bucket", req.getValue().bucket()); + assertEquals("workflow-insight/exec-1.json", req.getValue().key()); + assertEquals("application/json", req.getValue().contentType()); + + String json = readBody(body.getValue()); + assertTrue(json.contains("\"operations\""), "S3 emits the canonical operations array"); + assertTrue(!json.contains("operationsByName"), "S3 must not emit the by-name map"); + assertTrue(json.contains("\"greet\"")); + } + + private static String readBody(RequestBody body) throws Exception { + try (InputStream in = body.contentStreamProvider().newStream()) { + return new String(in.readAllBytes(), StandardCharsets.UTF_8); + } + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/TruncationTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/TruncationTest.java new file mode 100644 index 000000000..d843bee85 --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/TruncationTest.java @@ -0,0 +1,88 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.junit.jupiter.api.Test; + +class TruncationTest { + + private WorkflowInsightRecord recordWithResults(int bytesEach) { + WorkflowInsightRecord r = new WorkflowInsightRecord(); + r.executionArn = "arn:aws:lambda:us-west-2:1:function:f:$LATEST/durable-execution/e/i"; + r.status = "SUCCEEDED"; + r.startTime = "2026-08-05T00:00:00Z"; + String big = "x".repeat(bytesEach); + r.operations.add(new OperationRecord() + .id("1") + .name("bulk-1") + .type("STEP") + .status("SUCCEEDED") + .startTime("2026-08-05T00:00:01Z") + .result(big)); + r.operations.add(new OperationRecord() + .id("2") + .name("bulk-2") + .type("STEP") + .status("SUCCEEDED") + .startTime("2026-08-05T00:00:02Z") + .result(big)); + r.operations.add(new OperationRecord() + .id("3") + .name("bulk-3") + .type("STEP") + .status("SUCCEEDED") + .startTime("2026-08-05T00:00:03Z") + .result(big)); + return r; + } + + @Test + void phase1DropsResultsOldestFirstAndKeepsNewest() { + WorkflowInsightRecord r = recordWithResults(2000); + // Big enough for all three result-stripped ops, too small for the ~2 KB results. + WorkflowInsightRecord out = Truncation.truncateRecord(r, 4096, WorkflowInsightRecord::toWireMap); + assertEquals(Boolean.TRUE, out.truncated); + assertEquals(3, out.operations().size()); + assertNull(out.operations().get(0).result(), "oldest result dropped"); + assertEquals(Boolean.TRUE, out.operations().get(0).truncated()); + assertTrue(out.operations().get(2).result() != null, "newest result retained"); + assertNull(out.droppedOperations, "no whole operation dropped in phase 1"); + } + + @Test + void phase2DropsWholeOperationsOldestFirst() { + // No results to drop, tiny limit → whole operations dropped oldest-first. + WorkflowInsightRecord r = new WorkflowInsightRecord(); + r.executionArn = "arn:aws:lambda:us-west-2:1:function:f:$LATEST/durable-execution/e/i"; + r.status = "SUCCEEDED"; + r.startTime = "2026-08-05T00:00:00Z"; + for (int i = 1; i <= 3; i++) { + r.operations.add(new OperationRecord() + .id(String.valueOf(i)) + .name("bulk-" + i) + .type("STEP") + .status("SUCCEEDED") + .startTime("2026-08-05T00:00:0" + i + "Z")); + } + int base = Json.byteSize(r.toWireMap()); + // Force at least one whole-operation drop. + WorkflowInsightRecord out = Truncation.truncateRecord(r, base - 60, WorkflowInsightRecord::toWireMap); + assertEquals(Boolean.TRUE, out.truncated); + assertTrue(out.droppedOperations != null && out.droppedOperations >= 1); + boolean bulk1Present = out.operations().stream().anyMatch(o -> "bulk-1".equals(o.name())); + boolean bulk3Present = out.operations().stream().anyMatch(o -> "bulk-3".equals(o.name())); + assertTrue(!bulk1Present, "oldest operation dropped first"); + assertTrue(bulk3Present, "newest operation retained"); + } + + @Test + void noTruncationWhenUnderLimit() { + WorkflowInsightRecord r = recordWithResults(10); + WorkflowInsightRecord out = Truncation.truncateRecord(r, 5_000_000, WorkflowInsightRecord::toWireMap); + assertNull(out.truncated); + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/WorkflowInsightHookTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/WorkflowInsightHookTest.java new file mode 100644 index 000000000..a24291db2 --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/WorkflowInsightHookTest.java @@ -0,0 +1,140 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.services.lambda.model.OperationStatus; +import software.amazon.lambda.durable.plugin.DurableExecutionPlugin; +import software.amazon.lambda.durable.plugin.InvocationEndInfo; +import software.amazon.lambda.durable.plugin.InvocationInfo; +import software.amazon.lambda.durable.plugin.InvocationStatus; +import software.amazon.lambda.durable.plugin.OperationChangeInfo; +import software.amazon.lambda.durable.plugin.OperationChangeItemInfo; + +/** + * Hand-driven hook tests for behaviors the local runner cannot easily produce deterministically: on-change emission + * (invocation start / operation change / invocation end), cross-invocation state preservation on PENDING/RETRYING, and + * exporter isolation + flush. + */ +class WorkflowInsightHookTest { + + private static final String ARN = + "arn:aws:lambda:us-west-2:1:function:f:$LATEST/durable-execution/exec-1/invocation-1"; + private static final Instant START = Instant.parse("2026-08-05T00:00:00Z"); + + private static final class CapturingExporter implements InsightExporter { + final List records = new ArrayList<>(); + int flushes; + + @Override + public void export(WorkflowInsightRecord record) { + records.add(record); + } + + @Override + public void flush() { + flushes++; + } + } + + private Map ops(String name, OperationStatus status) { + Map m = new LinkedHashMap<>(); + m.put( + "op-1", + new OperationChangeItemInfo( + "op-1", name, "STEP", "Step", null, START, START.plusMillis(5), status, 1, false, null, null)); + return m; + } + + private InvocationInfo start(boolean first) { + return new InvocationInfo("req", ARN, first, START, "in", ops("greet", OperationStatus.STARTED), Map.of()); + } + + private InvocationEndInfo end(InvocationStatus status, Object result, Throwable error) { + return new InvocationEndInfo( + "req", ARN, true, START, ops("greet", OperationStatus.SUCCEEDED), status, error, "in", result); + } + + @Test + void onChangeEmitsAtStartChangeAndEnd() { + var exporter = new CapturingExporter(); + DurableExecutionPlugin plugin = WorkflowInsight.workflowInsight(WorkflowInsightConfig.builder() + .emitMode(WorkflowInsightConfig.EmitMode.ON_CHANGE) + .addExporter(exporter) + .build()); + + plugin.onInvocationStart(start(true)); + plugin.onOperationChange(new OperationChangeInfo( + "req", ARN, ops("greet", OperationStatus.SUCCEEDED), ops("greet", OperationStatus.SUCCEEDED))); + plugin.onInvocationEnd(end(InvocationStatus.SUCCEEDED, "out", null)); + + assertEquals(3, exporter.records.size()); + assertEquals("RUNNING", exporter.records.get(0).status()); + assertEquals("RUNNING", exporter.records.get(1).status()); + assertEquals("SUCCEEDED", exporter.records.get(2).status()); + } + + @Test + void onCompleteSkipsNonTerminalAndEmitsTerminalOnly() { + var exporter = new CapturingExporter(); + DurableExecutionPlugin plugin = WorkflowInsight.workflowInsight( + WorkflowInsightConfig.builder().addExporter(exporter).build()); + + plugin.onInvocationStart(start(true)); + plugin.onOperationChange(new OperationChangeInfo( + "req", ARN, ops("greet", OperationStatus.SUCCEEDED), ops("greet", OperationStatus.SUCCEEDED))); + assertTrue(exporter.records.isEmpty(), "no record before terminal in on-complete mode"); + plugin.onInvocationEnd(end(InvocationStatus.SUCCEEDED, "out", null)); + assertEquals(1, exporter.records.size()); + assertEquals("SUCCEEDED", exporter.records.get(0).status()); + } + + @Test + void statePreservedAcrossPendingKeepsStableStartTime() { + var exporter = new CapturingExporter(); + DurableExecutionPlugin plugin = WorkflowInsight.workflowInsight(WorkflowInsightConfig.builder() + .emitMode(WorkflowInsightConfig.EmitMode.ON_CHANGE) + .addExporter(exporter) + .build()); + + plugin.onInvocationStart(start(true)); + plugin.onInvocationEnd(end(InvocationStatus.PENDING, null, null)); // suspend, state retained + plugin.onInvocationEnd(end(InvocationStatus.SUCCEEDED, "out", null)); // resume + terminal + + // start(RUNNING) + pending(RUNNING) + terminal(SUCCEEDED); all share the stable startTime. + assertEquals(3, exporter.records.size()); + String startTime = exporter.records.get(0).startTime(); + assertTrue(exporter.records.stream().allMatch(r -> startTime.equals(r.startTime()))); + assertEquals(START.toString(), startTime); + } + + @Test + void exporterFailureIsIsolatedAndOthersStillReceiveAndFlush() { + var throwing = new InsightExporter() { + @Override + public void export(WorkflowInsightRecord record) { + throw new RuntimeException("exporter down"); + } + }; + var good = new CapturingExporter(); + DurableExecutionPlugin plugin = WorkflowInsight.workflowInsight(WorkflowInsightConfig.builder() + .addExporter(throwing) + .addExporter(good) + .build()); + + plugin.onInvocationStart(start(true)); + plugin.onInvocationEnd(end(InvocationStatus.SUCCEEDED, "out", null)); + + assertEquals(1, good.records.size(), "failing exporter never blocks the others"); + assertFalse(good.flushes == 0, "surviving exporter is flushed"); + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/WorkflowInsightPluginTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/WorkflowInsightPluginTest.java new file mode 100644 index 000000000..8d5cf0fff --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/WorkflowInsightPluginTest.java @@ -0,0 +1,377 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; +import software.amazon.lambda.durable.DurableConfig; +import software.amazon.lambda.durable.ParallelDurableFuture; +import software.amazon.lambda.durable.config.ParallelConfig; +import software.amazon.lambda.durable.config.StepConfig; +import software.amazon.lambda.durable.model.ExecutionStatus; +import software.amazon.lambda.durable.retry.JitterStrategy; +import software.amazon.lambda.durable.retry.RetryStrategies; +import software.amazon.lambda.durable.retry.RetryStrategy; +import software.amazon.lambda.durable.testing.LocalDurableTestRunner; + +/** + * End-to-end behavior tests exercising the plugin via the local durable test runner, which drives the real SDK + * coordinator so the plugin consumes the same hook snapshots (operations, execution input/output, per-op result) it + * receives in production. These map to the Workflow Insight conformance behaviors (insight-1 … insight-18). + */ +class WorkflowInsightPluginTest { + + private static final class CapturingExporter implements InsightExporter { + final List records = Collections.synchronizedList(new ArrayList<>()); + + @Override + public void export(WorkflowInsightRecord record) { + records.add(record); + } + } + + private static final RetryStrategy RETRY_ONCE = RetryStrategies.exponentialBackoff( + 2, Duration.ofSeconds(1), Duration.ofSeconds(1), 2.0, JitterStrategy.NONE); + + private DurableConfig configWith(CapturingExporter exporter, WorkflowInsightConfig.Builder cfg) { + var plugin = WorkflowInsight.workflowInsight(cfg.addExporter(exporter).build()); + return DurableConfig.builder().withPlugins(plugin).build(); + } + + private OperationRecord op(WorkflowInsightRecord rec, String name) { + return rec.operations().stream() + .filter(o -> name.equals(o.name())) + .findFirst() + .orElseThrow(); + } + + // insight-1 + @Test + void basicSuccessEmitsOneRecordWithNamedStepAndEchoedIo() { + var exporter = new CapturingExporter(); + var runner = LocalDurableTestRunner.create( + String.class, + (input, context) -> context.step("greet", String.class, sc -> "Hello, " + input + "!"), + configWith(exporter, WorkflowInsightConfig.builder())); + var result = runner.runUntilComplete("World"); + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + assertEquals(1, exporter.records.size()); + var rec = exporter.records.get(0); + assertEquals("WorkflowInsight", rec.recordType); + assertEquals("1.0", rec.schemaVersion); + assertEquals("SUCCEEDED", rec.status()); + assertEquals("World", rec.input); + assertEquals("Hello, World!", rec.output); + assertNull(rec.error); + assertNull(rec.truncated); + var greet = op(rec, "greet"); + assertEquals("STEP", greet.type()); + assertEquals("SUCCEEDED", greet.status()); + assertEquals(Integer.valueOf(1), greet.attempt()); + assertNull(greet.result(), "result omitted without an override"); + } + + // insight-2 + @Test + void executionFailureRecordCarriesErrorAndFailedOp() { + var exporter = new CapturingExporter(); + var runner = LocalDurableTestRunner.create( + String.class, + (input, context) -> context.step( + "failing-step", + String.class, + sc -> { + throw new RuntimeException("boom"); + }, + StepConfig.builder() + .retryStrategy(RetryStrategies.Presets.NO_RETRY) + .build()), + configWith(exporter, WorkflowInsightConfig.builder())); + var result = runner.run("World"); + assertEquals(ExecutionStatus.FAILED, result.getStatus()); + assertEquals(1, exporter.records.size()); + var rec = exporter.records.get(0); + assertEquals("FAILED", rec.status()); + assertNotNull(rec.error); + assertNotNull(rec.error.name()); + assertEquals("FAILED", op(rec, "failing-step").status()); + } + + // insight-3 + @Test + void onFailureModeEmitsNothingForSuccess() { + var exporter = new CapturingExporter(); + var runner = LocalDurableTestRunner.create( + String.class, + (input, context) -> context.step("greet", String.class, sc -> "hi"), + configWith( + exporter, WorkflowInsightConfig.builder().emitMode(WorkflowInsightConfig.EmitMode.ON_FAILURE))); + runner.runUntilComplete("World"); + assertTrue(exporter.records.isEmpty()); + } + + // insight-4 + @Test + void onFailureModeEmitsExactlyOneFailedRecord() { + var exporter = new CapturingExporter(); + var runner = LocalDurableTestRunner.create( + String.class, + (input, context) -> context.step( + "failing-step", + String.class, + sc -> { + throw new RuntimeException("boom"); + }, + StepConfig.builder() + .retryStrategy(RetryStrategies.Presets.NO_RETRY) + .build()), + configWith( + exporter, WorkflowInsightConfig.builder().emitMode(WorkflowInsightConfig.EmitMode.ON_FAILURE))); + var result = runner.run("World"); + assertEquals(ExecutionStatus.FAILED, result.getStatus()); + assertEquals(1, exporter.records.size()); + assertEquals("FAILED", exporter.records.get(0).status()); + } + + // insight-5 + @Test + void waitYieldsExactlyOneTerminalRecordNoRunningRecord() { + var exporter = new CapturingExporter(); + var runner = LocalDurableTestRunner.create( + String.class, + (input, context) -> { + context.wait("pause", Duration.ofSeconds(1)); + return context.step("after-wait", String.class, sc -> "done"); + }, + configWith(exporter, WorkflowInsightConfig.builder())); + var result = runner.runUntilComplete("World"); + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + assertEquals(1, exporter.records.size(), "on-complete emits exactly one terminal record across suspend/resume"); + assertEquals("SUCCEEDED", exporter.records.get(0).status()); + } + + // insight-6 + @Test + void stepRetryReflectsAttemptNumber() { + var exporter = new CapturingExporter(); + var calls = new AtomicInteger(); + var runner = LocalDurableTestRunner.create( + String.class, + (input, context) -> context.step( + "retried-step", + String.class, + sc -> { + if (calls.incrementAndGet() < 2) { + throw new RuntimeException("transient"); + } + return "ok"; + }, + StepConfig.builder().retryStrategy(RETRY_ONCE).build()), + configWith(exporter, WorkflowInsightConfig.builder())); + var result = runner.runUntilComplete("World"); + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + var retried = op(exporter.records.get(0), "retried-step"); + assertEquals("SUCCEEDED", retried.status()); + assertEquals(Integer.valueOf(2), retried.attempt(), "attempt reflects the retry"); + } + + // insight-7 + @Test + void repeatedNameAggregatesToCountThreeInByName() { + var exporter = new CapturingExporter(); + var runner = LocalDurableTestRunner.create( + String.class, + (input, context) -> { + for (int i = 0; i < 3; i++) { + context.step("task", Integer.class, sc -> 1); + } + return "done"; + }, + configWith(exporter, WorkflowInsightConfig.builder())); + runner.runUntilComplete("World"); + assertEquals(1, exporter.records.size()); + Map byName = + OperationsIndex.buildOperationsByName(exporter.records.get(0).operations()); + assertEquals(3, byName.get("task").count); + assertNull(byName.get("task").result, "repeated name drops representative result"); + } + + // insight-8 + @Test + void samplingRateZeroEmitsNothing() { + var exporter = new CapturingExporter(); + var runner = LocalDurableTestRunner.create( + String.class, + (input, context) -> context.step("greet", String.class, sc -> "hi"), + configWith(exporter, WorkflowInsightConfig.builder().samplingRate(0))); + runner.runUntilComplete("World"); + assertTrue(exporter.records.isEmpty()); + } + + // insight-9 + @Test + void contentOmittedDropsInputAndOutputWithoutTruncationMarkers() { + var exporter = new CapturingExporter(); + var runner = LocalDurableTestRunner.create( + String.class, + (input, context) -> context.step("greet", String.class, sc -> "Hello, " + input + "!"), + configWith( + exporter, + WorkflowInsightConfig.builder() + .content(ContentConfig.builder() + .input(false) + .output(false) + .build()))); + runner.runUntilComplete("World"); + var rec = exporter.records.get(0); + assertNull(rec.input); + assertNull(rec.output); + assertNull(rec.droppedInput, "config-omit is distinct from a size-drop"); + assertNull(rec.droppedOutput); + } + + // insight-10 + @Test + void includeErrorsFalseDropsOperationError() { + var exporter = new CapturingExporter(); + var runner = LocalDurableTestRunner.create( + String.class, + (input, context) -> context.step( + "failing-step", + String.class, + sc -> { + throw new RuntimeException("boom"); + }, + StepConfig.builder() + .retryStrategy(RetryStrategies.Presets.NO_RETRY) + .build()), + configWith( + exporter, + WorkflowInsightConfig.builder() + .content(ContentConfig.builder() + .includeErrors(false) + .build()))); + var result = runner.run("World"); + assertEquals(ExecutionStatus.FAILED, result.getStatus()); + var rec = exporter.records.get(0); + assertEquals("FAILED", rec.status()); + assertNull(op(rec, "failing-step").error(), "operation error suppressed by includeErrors:false"); + } + + // insight-11 + @Test + void operationResultOptInSurfacesCheckpointedValue() { + var exporter = new CapturingExporter(); + var runner = LocalDurableTestRunner.create( + String.class, + (input, context) -> context.step("compute", Integer.class, sc -> 42), + configWith( + exporter, + WorkflowInsightConfig.builder() + .content(ContentConfig.builder() + .addOverride(OperationOverride.withResult("compute", r -> r)) + .build()))); + runner.runUntilComplete("World"); + var compute = op(exporter.records.get(0), "compute"); + assertEquals(42, compute.result(), "identity override surfaces the checkpointed JSON value"); + } + + // insight-13 + @Test + void topLevelOnlyDropsNestedChildren() { + var exporter = new CapturingExporter(); + var runner = LocalDurableTestRunner.create( + String.class, + (input, context) -> { + ParallelDurableFuture parallel = context.parallel( + "parallel-work", ParallelConfig.builder().build()); + parallel.branch( + "branch-a", String.class, ctx -> ctx.step("branch-a-step", String.class, sc -> "a")); + parallel.branch( + "branch-b", String.class, ctx -> ctx.step("branch-b-step", String.class, sc -> "b")); + parallel.get(); + return "done"; + }, + configWith(exporter, WorkflowInsightConfig.builder())); + runner.runUntilComplete("World"); + var rec = exporter.records.get(0); + assertNotNull(op(rec, "parallel-work")); + assertTrue( + rec.operations().stream().allMatch(o -> o.parentId() == null), + "top-level mode keeps only parentId-less operations"); + assertTrue(rec.operations().stream().noneMatch(o -> "branch-a-step".equals(o.name()))); + } + + // insight-14 + @Test + void fullTreeIncludesChildrenLinkedToParent() { + var exporter = new CapturingExporter(); + var runner = LocalDurableTestRunner.create( + String.class, + (input, context) -> context.runInChildContext( + "parent-context", String.class, child -> child.step("child-step", String.class, sc -> "c")), + configWith( + exporter, + WorkflowInsightConfig.builder() + .operationDetail(WorkflowInsightConfig.OperationDetail.FULL_TREE))); + runner.runUntilComplete("World"); + var rec = exporter.records.get(0); + var parent = op(rec, "parent-context"); + var child = op(rec, "child-step"); + assertEquals("CONTEXT", parent.type()); + assertEquals(parent.id(), child.parentId(), "child parentId links to parent id in full-tree"); + } + + // insight-17 + @Test + void unnamedOperationsAreDropped() { + var exporter = new CapturingExporter(); + var runner = LocalDurableTestRunner.create( + String.class, + (input, context) -> { + context.wait(null, Duration.ofSeconds(1)); // unnamed WAIT -> dropped + return context.step("named-step", String.class, sc -> "ok"); + }, + configWith(exporter, WorkflowInsightConfig.builder())); + runner.runUntilComplete("World"); + var rec = exporter.records.get(0); + assertTrue(rec.operations().stream().allMatch(o -> o.name() != null), "no unnamed operation appears"); + assertNotNull(op(rec, "named-step")); + } + + // insight-18 + @Test + void summaryMaxAttemptReflectsRetry() { + var exporter = new CapturingExporter(); + var calls = new AtomicInteger(); + var runner = LocalDurableTestRunner.create( + String.class, + (input, context) -> context.step( + "retried-step", + String.class, + sc -> { + if (calls.incrementAndGet() < 2) { + throw new RuntimeException("transient"); + } + return "ok"; + }, + StepConfig.builder().retryStrategy(RETRY_ONCE).build()), + configWith(exporter, WorkflowInsightConfig.builder())); + runner.runUntilComplete("World"); + Map byName = + OperationsIndex.buildOperationsByName(exporter.records.get(0).operations()); + assertEquals(Integer.valueOf(2), byName.get("retried-step").maxAttempt); + assertEquals(1, byName.get("retried-step").count); + assertEquals(0, byName.get("retried-step").failedCount); + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/WorkflowInsightSamplingTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/WorkflowInsightSamplingTest.java new file mode 100644 index 000000000..763f7755a --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/WorkflowInsightSamplingTest.java @@ -0,0 +1,41 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.junit.jupiter.api.Test; + +class WorkflowInsightSamplingTest { + + @Test + void rateOneAlwaysSamplesIn() { + assertTrue(WorkflowInsight.shouldSample("any-arn", 1.0)); + } + + @Test + void rateZeroNeverSamplesIn() { + assertFalse(WorkflowInsight.shouldSample("any-arn", 0.0)); + } + + @Test + void fnv1a32IsDeterministic() { + assertEquals(WorkflowInsight.fnv1a32("hello"), WorkflowInsight.fnv1a32("hello")); + } + + @Test + void resolveSamplingRateClampsOutOfRange() { + assertEquals(1.0, WorkflowInsight.resolveSamplingRate(5.0)); + assertEquals(0.0, WorkflowInsight.resolveSamplingRate(-1.0)); + assertEquals(1.0, WorkflowInsight.resolveSamplingRate(null)); + assertEquals(0.25, WorkflowInsight.resolveSamplingRate(0.25)); + } + + @Test + void samplingDecisionIsStableAcrossCalls() { + String arn = "arn:aws:lambda:us-west-2:1:function:f:$LATEST/durable-execution/e/i"; + assertEquals(WorkflowInsight.shouldSample(arn, 0.5), WorkflowInsight.shouldSample(arn, 0.5)); + } +} diff --git a/pom.xml b/pom.xml index 1ed663b25..c77026784 100644 --- a/pom.xml +++ b/pom.xml @@ -43,6 +43,7 @@ sdk-testing sdk-integration-tests otel-plugin + insight-plugin examples conformance-tests coverage-report