From d09a4422299bbe1eef46e6025db74fae04e9553e Mon Sep 17 00:00:00 2001 From: Alex Wang Date: Wed, 9 Sep 2026 00:06:02 +0000 Subject: [PATCH 1/3] feat(examples): add workflow insight example --- examples/README.md | 1 + examples/pom.xml | 15 ++ .../insight/WorkflowInsightExample.java | 68 ++++++++ .../examples/CloudBasedIntegrationTest.java | 159 ++++++++++++++++++ .../insight/WorkflowInsightExampleTest.java | 45 +++++ 5 files changed, 288 insertions(+) create mode 100644 examples/src/main/java/software/amazon/lambda/durable/examples/insight/WorkflowInsightExample.java create mode 100644 examples/src/test/java/software/amazon/lambda/durable/examples/insight/WorkflowInsightExampleTest.java diff --git a/examples/README.md b/examples/README.md index c17750835..61e82522d 100644 --- a/examples/README.md +++ b/examples/README.md @@ -95,6 +95,7 @@ mvn test -Dtest=CloudBasedIntegrationTest \ | [SimpleMapExample](src/main/java/software/amazon/lambda/durable/examples/map/SimpleMapExample.java) | Concurrent map over a collection with durable steps | | [CustomShouldCompleteMapExample](src/main/java/software/amazon/lambda/durable/examples/map/CustomShouldCompleteMapExample.java) | Custom map completion with `shouldComplete` decisions | | [WaitForConditionExample](src/main/java/software/amazon/lambda/durable/examples/wait/WaitForConditionExample.java) | Poll a condition until met with `waitForCondition()` | +| [WorkflowInsightExample](src/main/java/software/amazon/lambda/durable/examples/insight/WorkflowInsightExample.java) | Emit a curated per-execution insight record to CloudWatch Logs via the Workflow Insight plugin (zero extra infrastructure) | ## Cleanup diff --git a/examples/pom.xml b/examples/pom.xml index 18654cf8e..f17bb957c 100644 --- a/examples/pom.xml +++ b/examples/pom.xml @@ -31,6 +31,13 @@ ${project.version} + + + software.amazon.lambda.durable + aws-durable-execution-sdk-java-plugin-insight + ${project.version} + + com.amazonaws @@ -72,6 +79,14 @@ sts test + + + software.amazon.awssdk + cloudwatchlogs + test + com.fasterxml.jackson.core jackson-databind diff --git a/examples/src/main/java/software/amazon/lambda/durable/examples/insight/WorkflowInsightExample.java b/examples/src/main/java/software/amazon/lambda/durable/examples/insight/WorkflowInsightExample.java new file mode 100644 index 000000000..f2befdf76 --- /dev/null +++ b/examples/src/main/java/software/amazon/lambda/durable/examples/insight/WorkflowInsightExample.java @@ -0,0 +1,68 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.examples.insight; + +import software.amazon.lambda.durable.DurableConfig; +import software.amazon.lambda.durable.DurableContext; +import software.amazon.lambda.durable.DurableHandler; +import software.amazon.lambda.durable.examples.types.GreetingRequest; +import software.amazon.lambda.durable.insight.ContentConfig; +import software.amazon.lambda.durable.insight.WorkflowInsight; +import software.amazon.lambda.durable.insight.WorkflowInsightConfig; +import software.amazon.lambda.durable.insight.exporters.LambdaLogExporter; + +/** + * Example demonstrating the Workflow Insight plugin with zero extra infrastructure. + * + *

The plugin is registered in {@link #createConfiguration()} with a {@link LambdaLogExporter}, which writes one + * curated {@code WorkflowInsight} JSON record to {@code stdout} at the end of each execution. On Lambda, {@code stdout} + * is captured to the function's own CloudWatch Logs group, so no bucket, extra log group, or additional IAM permission + * is required — the managed function log group is the destination. + * + *

Configuration used here: + * + *

+ * + *

The handler itself runs two named steps ({@code create-greeting} and {@code transform}) and returns a + * {@code HELLO, !} greeting. + */ +public class WorkflowInsightExample extends DurableHandler { + + @Override + protected DurableConfig createConfiguration() { + var insight = WorkflowInsight.workflowInsight(WorkflowInsightConfig.builder() + // Write the record to stdout -> the function's own CloudWatch Logs group (no extra infrastructure). + .addExporter(new LambdaLogExporter()) + // Emit a single record when the execution completes. + .emitMode(WorkflowInsightConfig.EmitMode.ON_COMPLETE) + // Summarize only top-level operations. + .operationDetail(WorkflowInsightConfig.OperationDetail.TOP_LEVEL) + // Include the execution input, output, and any errors in the record. + .content(ContentConfig.builder() + .input(true) + .output(true) + .includeErrors(true) + .build()) + .build()); + + return DurableConfig.builder().withPlugins(insight).build(); + } + + @Override + public String handleRequest(GreetingRequest input, DurableContext context) { + context.getLogger().info("Building greeting for {}", input.getName()); + + // Step 1: create the greeting. + var greeting = context.step("create-greeting", String.class, stepCtx -> "Hello, " + input.getName()); + + // Step 2: transform it into the final HELLO, ! form. + var result = context.step("transform", String.class, stepCtx -> greeting.toUpperCase() + "!"); + + context.getLogger().info("Workflow Insight example complete: {}", result); + return result; + } +} diff --git a/examples/src/test/java/software/amazon/lambda/durable/examples/CloudBasedIntegrationTest.java b/examples/src/test/java/software/amazon/lambda/durable/examples/CloudBasedIntegrationTest.java index 7490961a5..fa44f0697 100644 --- a/examples/src/test/java/software/amazon/lambda/durable/examples/CloudBasedIntegrationTest.java +++ b/examples/src/test/java/software/amazon/lambda/durable/examples/CloudBasedIntegrationTest.java @@ -5,11 +5,16 @@ import static org.junit.jupiter.api.Assertions.*; import static software.amazon.lambda.durable.TypeToken.get; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; import java.time.Duration; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Optional; +import java.util.UUID; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.LockSupport; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.condition.EnabledForJreRange; @@ -19,6 +24,9 @@ import org.junit.jupiter.params.provider.CsvSource; import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider; import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.cloudwatchlogs.CloudWatchLogsClient; +import software.amazon.awssdk.services.cloudwatchlogs.model.FilterLogEventsRequest; +import software.amazon.awssdk.services.cloudwatchlogs.model.ResourceNotFoundException; import software.amazon.awssdk.services.lambda.LambdaClient; import software.amazon.awssdk.services.lambda.model.ErrorObject; import software.amazon.awssdk.services.lambda.model.OperationStatus; @@ -43,6 +51,10 @@ class CloudBasedIntegrationTest { private static String region; private static String functionNamePrefix; private static LambdaClient lambdaClient; + private static CloudWatchLogsClient logsClient; + + /** Reused for parsing insight records out of CloudWatch log messages; ObjectMapper is thread-safe for reads. */ + private static final ObjectMapper INSIGHT_MAPPER = new ObjectMapper(); static boolean isEnabled() { var enabled = "true".equals(System.getProperty("test.cloud.enabled")); @@ -77,6 +89,11 @@ static void setup() { .region(Region.of(region)) .build(); + logsClient = CloudWatchLogsClient.builder() + .credentialsProvider(DefaultCredentialsProvider.builder().build()) + .region(Region.of(region)) + .build(); + System.out.println("☁️ Running cloud integration tests against account " + account + " in " + region); } @@ -853,4 +870,146 @@ void testPluginExample() { assertNotNull(runner.getOperation("create-greeting")); assertNotNull(runner.getOperation("transform")); } + + @Test + void testWorkflowInsightExample() { + // Unique alphanumeric token so we match THIS execution's insight record by input.name, never by a broad + // time-only window that would race with concurrent CI runs (Java 17/21/25) writing to the same log group. + var uniqueName = "insight" + UUID.randomUUID().toString().replace("-", ""); + + var runner = CloudDurableTestRunner.create( + arn("workflow-insight-example"), GreetingRequest.class, String.class, lambdaClient); + + // Bound the log query window to just before this invocation (minus a small skew for clock/ingestion). + var queryStartMillis = + System.currentTimeMillis() - Duration.ofMinutes(1).toMillis(); + + var result = runner.run(new GreetingRequest(uniqueName)); + + // 1) Execution-level assertions from the durable history. + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + var expectedOutput = "HELLO, " + uniqueName.toUpperCase() + "!"; + assertEquals(expectedOutput, result.getResult()); + assertNotNull(runner.getOperation("create-greeting")); + assertNotNull(runner.getOperation("transform")); + + // 2) Insight-record assertions from the managed function log group (LambdaLogExporter -> stdout -> CW Logs). + var logGroup = "/aws/lambda/" + functionNamePrefix + "workflow-insight-example"; + var record = pollForInsightRecord(logGroup, uniqueName, queryStartMillis); + assertTrue( + record.isPresent(), "No WorkflowInsight record found in " + logGroup + " for input.name=" + uniqueName); + + var node = record.get(); + assertEquals("WorkflowInsight", node.path("recordType").asText()); + assertEquals("1.0", node.path("schemaVersion").asText()); + assertEquals("SUCCEEDED", node.path("status").asText()); + assertEquals(uniqueName, node.path("input").path("name").asText()); + assertEquals(expectedOutput, node.path("output").asText()); + + var byName = node.path("operationsByName"); + assertTrue(byName.has("create-greeting"), "operationsByName missing create-greeting"); + assertTrue(byName.has("transform"), "operationsByName missing transform"); + assertEquals("SUCCEEDED", byName.path("create-greeting").path("status").asText()); + assertEquals("SUCCEEDED", byName.path("transform").path("status").asText()); + assertEquals(1, byName.path("create-greeting").path("count").asInt()); + assertEquals(1, byName.path("transform").path("count").asInt()); + } + + /** + * Polls the managed function log group for the WorkflowInsight record whose {@code input.name} equals + * {@code uniqueName}. Bounded by a wall-clock deadline and uses {@link LockSupport#parkNanos(long)} (never + * {@code Thread.sleep}) to absorb CloudWatch Logs ingestion lag. Deterministic across JREs: the loop terminates on + * either a match or the deadline, and a final attempt runs after the last park so the deadline edge is not a lost + * poll. + */ + private static Optional pollForInsightRecord(String logGroup, String uniqueName, long queryStartMillis) { + var deadline = System.nanoTime() + Duration.ofSeconds(120).toNanos(); + while (System.nanoTime() < deadline) { + var found = queryInsightRecordOnce(logGroup, uniqueName, queryStartMillis); + if (found.isPresent()) { + return found; + } + LockSupport.parkNanos(Duration.ofSeconds(3).toNanos()); + } + return queryInsightRecordOnce(logGroup, uniqueName, queryStartMillis); + } + + /** + * One bounded pass over the log group: a server-side {@code filterPattern} narrows to events containing the unique + * token, and pagination is followed up to a fixed page cap so a single pass cannot run unbounded. A missing log + * group (not yet created) is treated as "not found yet" so the caller retries. + */ + private static Optional queryInsightRecordOnce( + String logGroup, String uniqueName, long queryStartMillis) { + String nextToken = null; + var pages = 0; + try { + do { + var req = FilterLogEventsRequest.builder() + .logGroupName(logGroup) + .startTime(queryStartMillis) + // Quoted term match narrows server-side to events containing the unique token, keeping the + // scanned set small regardless of other executions writing to the same group. + .filterPattern("\"" + uniqueName + "\"") + .limit(100) + .nextToken(nextToken) + .build(); + var resp = logsClient.filterLogEvents(req); + for (var event : resp.events()) { + var parsed = parseInsightMessage(event.message(), uniqueName); + if (parsed.isPresent()) { + return parsed; + } + } + nextToken = resp.nextToken(); + pages++; + } while (nextToken != null && pages < 20); + } catch (ResourceNotFoundException e) { + // Managed log group / stream not created yet — let the caller retry within the deadline. + return Optional.empty(); + } + return Optional.empty(); + } + + /** + * Parses one CloudWatch log message into a matching WorkflowInsight record node. Handles both a raw top-level JSON + * record (the {@code System.out} line LambdaLogExporter writes) and a Lambda structured-logging envelope whose + * {@code message} field carries the insight JSON as a string. Only a record whose {@code input.name} equals + * {@code uniqueName} is returned. + */ + private static Optional parseInsightMessage(String message, String uniqueName) { + if (message == null || message.isEmpty()) { + return Optional.empty(); + } + var node = tryParseJson(message); + if (node == null) { + return Optional.empty(); + } + // Case 1: raw top-level insight record. + if (isMatchingInsight(node, uniqueName)) { + return Optional.of(node); + } + // Case 2: Lambda structured-logging envelope: {"timestamp":...,"message":"",...}. + var inner = node.get("message"); + if (inner != null && inner.isTextual()) { + var innerNode = tryParseJson(inner.asText()); + if (innerNode != null && isMatchingInsight(innerNode, uniqueName)) { + return Optional.of(innerNode); + } + } + return Optional.empty(); + } + + private static boolean isMatchingInsight(JsonNode node, String uniqueName) { + return "WorkflowInsight".equals(node.path("recordType").asText(null)) + && uniqueName.equals(node.path("input").path("name").asText(null)); + } + + private static JsonNode tryParseJson(String text) { + try { + return INSIGHT_MAPPER.readTree(text); + } catch (Exception e) { + return null; + } + } } diff --git a/examples/src/test/java/software/amazon/lambda/durable/examples/insight/WorkflowInsightExampleTest.java b/examples/src/test/java/software/amazon/lambda/durable/examples/insight/WorkflowInsightExampleTest.java new file mode 100644 index 000000000..3f69c7b4a --- /dev/null +++ b/examples/src/test/java/software/amazon/lambda/durable/examples/insight/WorkflowInsightExampleTest.java @@ -0,0 +1,45 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.examples.insight; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +import org.junit.jupiter.api.Test; +import software.amazon.lambda.durable.examples.types.GreetingRequest; +import software.amazon.lambda.durable.model.ExecutionStatus; +import software.amazon.lambda.durable.testing.LocalDurableTestRunner; + +class WorkflowInsightExampleTest { + + @Test + void emitsInsightAndReturnsGreeting() { + var handler = new WorkflowInsightExample(); + var runner = LocalDurableTestRunner.create(GreetingRequest.class, handler); + + var result = runner.runUntilComplete(new GreetingRequest("Alice")); + + // The Workflow Insight plugin writes its record to stdout via LambdaLogExporter; we do not capture + // global stdout here. The local assertions verify the execution succeeds and both named operations run, + // which is what feeds the emitted operationsByName summaries. + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + assertEquals("HELLO, ALICE!", result.getResult(String.class)); + + assertNotNull(result.getOperation("create-greeting")); + assertNotNull(result.getOperation("transform")); + } + + @Test + void usesDefaultNameWhenAbsent() { + var handler = new WorkflowInsightExample(); + var runner = LocalDurableTestRunner.create(GreetingRequest.class, handler); + + var result = runner.runUntilComplete(new GreetingRequest()); + + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + assertEquals("HELLO, WORLD!", result.getResult(String.class)); + + assertNotNull(result.getOperation("create-greeting")); + assertNotNull(result.getOperation("transform")); + } +} From 4b30d5e0cb7c338d47a07d603c996e43e0578eae Mon Sep 17 00:00:00 2001 From: Alex Wang Date: Wed, 9 Sep 2026 00:10:23 +0000 Subject: [PATCH 2/3] test(examples): clarify insight coverage --- .../lambda/durable/examples/CloudBasedIntegrationTest.java | 2 +- .../durable/examples/insight/WorkflowInsightExampleTest.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/examples/src/test/java/software/amazon/lambda/durable/examples/CloudBasedIntegrationTest.java b/examples/src/test/java/software/amazon/lambda/durable/examples/CloudBasedIntegrationTest.java index fa44f0697..9f7ee43f2 100644 --- a/examples/src/test/java/software/amazon/lambda/durable/examples/CloudBasedIntegrationTest.java +++ b/examples/src/test/java/software/amazon/lambda/durable/examples/CloudBasedIntegrationTest.java @@ -874,7 +874,7 @@ void testPluginExample() { @Test void testWorkflowInsightExample() { // Unique alphanumeric token so we match THIS execution's insight record by input.name, never by a broad - // time-only window that would race with concurrent CI runs (Java 17/21/25) writing to the same log group. + // time-only window that could select a stale record from an earlier run against the same deployed function. var uniqueName = "insight" + UUID.randomUUID().toString().replace("-", ""); var runner = CloudDurableTestRunner.create( diff --git a/examples/src/test/java/software/amazon/lambda/durable/examples/insight/WorkflowInsightExampleTest.java b/examples/src/test/java/software/amazon/lambda/durable/examples/insight/WorkflowInsightExampleTest.java index 3f69c7b4a..78b78a7e9 100644 --- a/examples/src/test/java/software/amazon/lambda/durable/examples/insight/WorkflowInsightExampleTest.java +++ b/examples/src/test/java/software/amazon/lambda/durable/examples/insight/WorkflowInsightExampleTest.java @@ -13,7 +13,7 @@ class WorkflowInsightExampleTest { @Test - void emitsInsightAndReturnsGreeting() { + void returnsGreetingAndRecordsOperations() { var handler = new WorkflowInsightExample(); var runner = LocalDurableTestRunner.create(GreetingRequest.class, handler); From 817e80ad1f209e3f80275df94f81dbc6f2a1f52b Mon Sep 17 00:00:00 2001 From: Alex Wang Date: Wed, 9 Sep 2026 20:44:50 +0000 Subject: [PATCH 3/3] refactor(examples): use default insight config --- .../insight/WorkflowInsightExample.java | 37 ++++--------------- 1 file changed, 7 insertions(+), 30 deletions(-) diff --git a/examples/src/main/java/software/amazon/lambda/durable/examples/insight/WorkflowInsightExample.java b/examples/src/main/java/software/amazon/lambda/durable/examples/insight/WorkflowInsightExample.java index f2befdf76..8999a0a47 100644 --- a/examples/src/main/java/software/amazon/lambda/durable/examples/insight/WorkflowInsightExample.java +++ b/examples/src/main/java/software/amazon/lambda/durable/examples/insight/WorkflowInsightExample.java @@ -6,26 +6,16 @@ import software.amazon.lambda.durable.DurableContext; import software.amazon.lambda.durable.DurableHandler; import software.amazon.lambda.durable.examples.types.GreetingRequest; -import software.amazon.lambda.durable.insight.ContentConfig; import software.amazon.lambda.durable.insight.WorkflowInsight; import software.amazon.lambda.durable.insight.WorkflowInsightConfig; -import software.amazon.lambda.durable.insight.exporters.LambdaLogExporter; /** - * Example demonstrating the Workflow Insight plugin with zero extra infrastructure. + * Example demonstrating the Workflow Insight plugin with zero extra infrastructure and its default configuration. * - *

The plugin is registered in {@link #createConfiguration()} with a {@link LambdaLogExporter}, which writes one - * curated {@code WorkflowInsight} JSON record to {@code stdout} at the end of each execution. On Lambda, {@code stdout} - * is captured to the function's own CloudWatch Logs group, so no bucket, extra log group, or additional IAM permission - * is required — the managed function log group is the destination. - * - *

Configuration used here: - * - *

+ *

The empty {@link WorkflowInsightConfig} uses the default Lambda log exporter, which writes one curated + * {@code WorkflowInsight} JSON record to {@code stdout} when the execution completes. It includes top-level operations, + * input, output, and errors. On Lambda, {@code stdout} is captured by the function's own CloudWatch Logs group, so no + * bucket, extra log group, or additional IAM permission is required. * *

The handler itself runs two named steps ({@code create-greeting} and {@code transform}) and returns a * {@code HELLO, !} greeting. @@ -34,21 +24,8 @@ public class WorkflowInsightExample extends DurableHandler the function's own CloudWatch Logs group (no extra infrastructure). - .addExporter(new LambdaLogExporter()) - // Emit a single record when the execution completes. - .emitMode(WorkflowInsightConfig.EmitMode.ON_COMPLETE) - // Summarize only top-level operations. - .operationDetail(WorkflowInsightConfig.OperationDetail.TOP_LEVEL) - // Include the execution input, output, and any errors in the record. - .content(ContentConfig.builder() - .input(true) - .output(true) - .includeErrors(true) - .build()) - .build()); - + var insight = + WorkflowInsight.workflowInsight(WorkflowInsightConfig.builder().build()); return DurableConfig.builder().withPlugins(insight).build(); }