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..8999a0a47
--- /dev/null
+++ b/examples/src/main/java/software/amazon/lambda/durable/examples/insight/WorkflowInsightExample.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 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.WorkflowInsight;
+import software.amazon.lambda.durable.insight.WorkflowInsightConfig;
+
+/**
+ * Example demonstrating the Workflow Insight plugin with zero extra infrastructure and its default configuration.
+ *
+ * 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.
+ */
+public class WorkflowInsightExample extends DurableHandler {
+
+ @Override
+ protected DurableConfig createConfiguration() {
+ var insight =
+ WorkflowInsight.workflowInsight(WorkflowInsightConfig.builder().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..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
@@ -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 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(
+ 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..78b78a7e9
--- /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 returnsGreetingAndRecordsOperations() {
+ 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"));
+ }
+}