diff --git a/insight-plugin/README.md b/insight-plugin/README.md index 3a13b97b6..195fca6fa 100644 --- a/insight-plugin/README.md +++ b/insight-plugin/README.md @@ -26,32 +26,141 @@ DurableConfig config = DurableConfig.builder() .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. +## Exporters -`LambdaLogExporter` needs no extra dependency. The AWS SDK service modules used by the remote -exporters are optional so applications that use only Lambda logs do not package them. Add the module -for each remote exporter you configure, using the AWS SDK for Java 2.x version managed by your -application: +Implement `InsightExporter` for custom sinks. Every exporter has a builder with one setter per option and a +`maxRecordSizeBytes` override; records over the limit are truncated before export. + +| Exporter | Destination | Operations rendering | Default size limit | Artifact (optional) | +|---|---|---|---|---| +| `LambdaLogExporter` | Function log group (stdout) | `operationsByName` | 256 KB | none | +| `CloudWatchLogsExporter` | Any log group, PutLogEvents | `operationsByName` | 256 KB | `cloudwatchlogs` | +| `S3Exporter` | One object per execution | `operations` array | 5 MB | `s3` | +| `DynamoDBExporter` | One item per record (or per execution) | `operationsByName` | 400 KB | `dynamodb` | +| `AuroraExporter` | One row per execution, RDS Data API | `operations` array | 1 MB | `rdsdata` | +| `RedshiftExporter` | One row per execution, Redshift Data API | `operations` array | 1 MB | `redshiftdata` | +| `OpenSearchExporter` | One document per execution | `operations` array | 10 MB | `http-auth-aws`, `auth` | +| `FirehoseExporter` | One NDJSON record, PutRecord | `operationsFormat` | 1 MB | `firehose` | +| `EventBridgeExporter` | One event, PutEvents | `operationsFormat` | 256 KB | `eventbridge` | +| `SQSExporter` | One message, SendMessage | `operationsFormat` | 256 KB | `sqs` | +| `OTelExporter` | OTLP/HTTP JSON log record | `operationsFormat` (body) | 1 MB | none | +| `HttpExporter` | POST or PUT JSON to a URL | `operationsFormat` | none | none | +| `FileExporter` | NDJSON or JSON files in a directory | `operationsFormat` | none | none | + +`operationsFormat` is `ARRAY` (default), `BY_NAME`, or `BOTH`. + +Artifacts are `software.amazon.awssdk` modules and are optional: add only the ones for the exporters you configure, +at the AWS SDK for Java 2.x version your application manages. A configured exporter whose artifact is missing fails +at first export with a message naming the artifact; the plugin logs it and continues with the other exporters. ```xml - software.amazon.awssdk - s3 + dynamodb AWS_SDK_VERSION +``` - - - software.amazon.awssdk - cloudwatchlogs - AWS_SDK_VERSION - +### DynamoDBExporter + +Table keyed by `pk` (string), optionally with sort key `sk`. IAM: `dynamodb:PutItem`. + +```java +DynamoDBExporter.builder().tableName("workflow-insight").build() // history: pk = ARN, sk = emittedAt +DynamoDBExporter.builder().tableName("workflow-insight").sortKey("").build() // upsert: pk only +``` + +### AuroraExporter + +Cluster with the Data API enabled. Time values are bound as ISO-8601 strings. Table columns: `execution_arn +VARCHAR(512) PRIMARY KEY`, `execution_name VARCHAR(256)`, `function_name VARCHAR(128)`, `status VARCHAR(20)`, +`start_time VARCHAR(30)`, `end_time VARCHAR(30)`, `duration_ms BIGINT`, `record_json` (`JSONB` on PostgreSQL, +`LONGTEXT` on MySQL), `emitted_at VARCHAR(30)`. On PostgreSQL the time columns may instead be `TIMESTAMPTZ`; the +statement casts the values. IAM: `rds-data:ExecuteStatement`, `secretsmanager:GetSecretValue`. + +```java +AuroraExporter.builder() + .resourceArn(clusterArn).secretArn(secretArn).database("insight") + .engine(AuroraExporter.Engine.POSTGRESQL) // or MYSQL + .build() +``` + +### RedshiftExporter + +Serverless workgroup or provisioned cluster. Same columns as Aurora with `record_json SUPER` and `TIMESTAMPTZ` time +columns; rows are upserted with `MERGE`. IAM: `redshift-data:ExecuteStatement` plus `redshift-serverless:GetCredentials` +(Serverless) or `secretsmanager:GetSecretValue` / `redshift:GetClusterCredentialsWithIAM` (provisioned). + +```java +RedshiftExporter.builder().workgroupName("insight").database("dev").build() +RedshiftExporter.builder().clusterIdentifier("my-cluster").database("dev").secretArn(secretArn).build() +``` + +### OpenSearchExporter + +Domain endpoint; the index is created on first write. The document id is the execution ARN. IAM (SigV4): +`es:ESHttpPut` on `domain//workflow-insight/*`. + +```java +OpenSearchExporter.builder().endpoint("https://my-domain.us-east-1.es.amazonaws.com").region("us-east-1").build() +OpenSearchExporter.builder().endpoint(url).auth(OpenSearchExporter.Auth.BASIC).username(u).password(p).build() +``` + +### FirehoseExporter + +Delivery stream with any destination. IAM: `firehose:PutRecord`. + +```java +FirehoseExporter.builder().deliveryStreamName("workflow-insight").build() +``` + +### EventBridgeExporter + +Default bus or a custom bus. `DetailType` is the record status, so rules can match `FAILED`. IAM: `events:PutEvents`. + +```java +EventBridgeExporter.builder().build() // default bus, source software.amazon.lambda.durable.insight +EventBridgeExporter.builder().eventBusName("insight-bus").build() ``` +### SQSExporter + +Standard or FIFO queue; FIFO queues receive a group id (execution ARN) and a deduplication id. IAM: `sqs:SendMessage`. + +```java +SQSExporter.builder().queueUrl("https://sqs.us-east-1.amazonaws.com/123456789012/insight.fifo").build() +``` + +### OTelExporter + +Any OTLP/HTTP logs endpoint; authenticate with headers. `http/protobuf` is not supported. No IAM. + +```java +OTelExporter.builder().endpoint("https://otlp.example.com/v1/logs").headers(Map.of("x-api-key", key)).build() +``` + +### HttpExporter + +Any endpoint accepting JSON. `timeoutMs` defaults to 10000. No IAM. + +```java +HttpExporter.builder().url("https://hooks.example.com/insight").method(HttpExporter.Method.PUT).build() +``` + +### FileExporter + +A writable directory such as an EFS mount or `/tmp`. `NDJSON` appends `{date}.ndjson`; `JSON` writes +`{executionName}.json`. + +```java +FileExporter.builder().directory("/mnt/efs/workflow-insight").mode(FileExporter.Mode.JSON).build() +``` + +### S3Exporter and CloudWatchLogsExporter + +`S3Exporter` writes `{prefix}{partition}{executionName}.json` (IAM: `s3:PutObject`). `CloudWatchLogsExporter` writes +one event per record to `{logStreamPrefix}YYYY/MM/DD` (IAM: `logs:CreateLogStream`, `logs:PutLogEvents`). + ## Design - **Snapshot-based, not accumulated.** Each record is built directly from the current-invocation diff --git a/insight-plugin/pom.xml b/insight-plugin/pom.xml index bf17469fc..f04cfb22e 100644 --- a/insight-plugin/pom.xml +++ b/insight-plugin/pom.xml @@ -34,7 +34,7 @@ jackson-datatype-jsr310 - + software.amazon.awssdk s3 @@ -45,6 +45,47 @@ cloudwatchlogs true + + software.amazon.awssdk + dynamodb + true + + + software.amazon.awssdk + rdsdata + true + + + software.amazon.awssdk + firehose + true + + + software.amazon.awssdk + eventbridge + true + + + software.amazon.awssdk + redshiftdata + true + + + software.amazon.awssdk + sqs + true + + + + software.amazon.awssdk + http-auth-aws + true + + + software.amazon.awssdk + auth + true + 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 index 352bb132d..dc72a7092 100644 --- 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 @@ -3,6 +3,9 @@ package software.amazon.lambda.durable.insight; import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.core.util.DefaultIndenter; +import com.fasterxml.jackson.core.util.DefaultPrettyPrinter; +import com.fasterxml.jackson.core.util.Separators; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.SerializationFeature; import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; @@ -23,6 +26,14 @@ public final class Json { .registerModule(new JavaTimeModule()) .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS); + private static final DefaultPrettyPrinter PRETTY_PRINTER = new DefaultPrettyPrinter() + .withObjectIndenter(new DefaultIndenter(" ", "\n")) + .withArrayIndenter(new DefaultIndenter(" ", "\n")) + .withSeparators(Separators.createDefaultInstance() + .withObjectFieldValueSpacing(Separators.Spacing.AFTER) + .withObjectEmptySeparator("") + .withArrayEmptySeparator("")); + private Json() {} public static String stringify(Object value) { @@ -33,6 +44,25 @@ public static String stringify(Object value) { } } + /** + * Serializes with two-space indentation, {@code ": "} between name and value, and {@code []}/{@code {}} when empty. + */ + public static String prettyStringify(Object value) { + try { + return MAPPER.writer(PRETTY_PRINTER).writeValueAsString(value); + } catch (JsonProcessingException e) { + throw new IllegalStateException("failed to serialize insight record", e); + } + } + + /** + * Converts a value to its JSON-compatible form: maps, lists, strings, numbers, booleans, or {@code null}. Used by + * exporters that marshal the record into a destination's native document type instead of a JSON string. + */ + public static Object toJsonValue(Object value) { + return deepCopyContent(value); + } + /** 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 { diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/AuroraExporter.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/AuroraExporter.java new file mode 100644 index 000000000..a84008a56 --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/AuroraExporter.java @@ -0,0 +1,203 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight.exporters; + +import static java.util.Objects.requireNonNull; + +import java.util.List; +import java.util.Map; +import software.amazon.awssdk.services.rdsdata.RdsDataClient; +import software.amazon.awssdk.services.rdsdata.model.ExecuteStatementRequest; +import software.amazon.awssdk.services.rdsdata.model.Field; +import software.amazon.awssdk.services.rdsdata.model.SqlParameter; +import software.amazon.lambda.durable.annotations.Experimental; +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 Aurora (PostgreSQL or MySQL) through the RDS Data API, upserting one row + * per execution keyed by execution ARN. Requires the Data API enabled on the cluster, {@code rds-data:ExecuteStatement} + * and {@code secretsmanager:GetSecretValue}. + */ +@Experimental +public final class AuroraExporter implements InsightExporter { + + /** Database engine; selects the upsert dialect. */ + @Experimental + public enum Engine { + POSTGRESQL("postgresql"), + MYSQL("mysql"); + + private final String value; + + Engine(String value) { + this.value = value; + } + + /** The configuration string for this engine. */ + public String value() { + return value; + } + + /** Parses a configuration string; unknown values are rejected. */ + public static Engine fromValue(String value) { + for (Engine e : values()) { + if (e.value.equals(value)) { + return e; + } + } + throw new IllegalArgumentException("Unknown engine: \"" + value + "\". Expected postgresql or mysql."); + } + } + + private final String resourceArn; + private final String secretArn; + private final String database; + private final String table; + private final Engine engine; + private final Integer maxRecordSizeBytes; + private final LazyClient client; + + private AuroraExporter(Builder b) { + this.resourceArn = requireNonNull(b.resourceArn, "resourceArn"); + this.secretArn = requireNonNull(b.secretArn, "secretArn"); + this.database = requireNonNull(b.database, "database"); + this.table = SqlIdentifiers.validate(b.table != null ? b.table : "workflow_insight"); + this.engine = requireNonNull(b.engine, "engine"); + this.maxRecordSizeBytes = b.maxRecordSizeBytes != null ? b.maxRecordSizeBytes : 1_000_000; + this.client = LazyClient.forSdkClient( + b.client, "rdsdata", "software.amazon.awssdk.services.rdsdata.RdsDataClient", b.region); + } + + public static Builder builder() { + return new Builder(); + } + + @Override + public Integer maxRecordSizeBytes() { + return maxRecordSizeBytes; + } + + @Override + public void export(WorkflowInsightRecord record) { + RdsDataClient rdsData = client.get(); + Map wire = record.toWireMap(); + String sql = engine == Engine.POSTGRESQL ? buildPostgresUpsert() : buildMysqlUpsert(); + List parameters = List.of( + string("execution_arn", record.executionArn()), + string("execution_name", record.executionName()), + string("function_name", record.functionName()), + string("status", record.status()), + string("start_time", record.startTime()), + string("end_time", (String) wire.get("endTime")), + longValue("duration_ms", (Long) wire.get("durationMs")), + string("record_json", Json.stringify(wire)), + string("emitted_at", (String) wire.get("emittedAt"))); + rdsData.executeStatement(ExecuteStatementRequest.builder() + .resourceArn(resourceArn) + .secretArn(secretArn) + .database(database) + .sql(sql) + .parameters(parameters) + .build()); + } + + private static SqlParameter string(String name, String value) { + Field field = value != null + ? Field.builder().stringValue(value).build() + : Field.builder().isNull(true).build(); + return SqlParameter.builder().name(name).value(field).build(); + } + + private static SqlParameter longValue(String name, Long value) { + Field field = value != null + ? Field.builder().longValue(value).build() + : Field.builder().isNull(true).build(); + return SqlParameter.builder().name(name).value(field).build(); + } + + private String buildPostgresUpsert() { + return "INSERT INTO " + table + "\n" + + " (execution_arn, execution_name, function_name, status, start_time, end_time, duration_ms, record_json, emitted_at)\n" + + " VALUES\n" + + " (:execution_arn, :execution_name, :function_name, :status, :start_time::timestamptz, :end_time::timestamptz, :duration_ms, :record_json::jsonb, :emitted_at::timestamptz)\n" + + " ON CONFLICT (execution_arn) DO UPDATE SET\n" + + " status = EXCLUDED.status,\n" + + " end_time = EXCLUDED.end_time,\n" + + " duration_ms = EXCLUDED.duration_ms,\n" + + " record_json = EXCLUDED.record_json,\n" + + " emitted_at = EXCLUDED.emitted_at"; + } + + private String buildMysqlUpsert() { + return "INSERT INTO " + table + "\n" + + " (execution_arn, execution_name, function_name, status, start_time, end_time, duration_ms, record_json, emitted_at)\n" + + " VALUES\n" + + " (:execution_arn, :execution_name, :function_name, :status, :start_time, :end_time, :duration_ms, :record_json, :emitted_at)\n" + + " ON DUPLICATE KEY UPDATE\n" + + " status = VALUES(status),\n" + + " end_time = VALUES(end_time),\n" + + " duration_ms = VALUES(duration_ms),\n" + + " record_json = VALUES(record_json),\n" + + " emitted_at = VALUES(emitted_at)"; + } + + /** Builder for {@link AuroraExporter}. */ + public static final class Builder { + private String resourceArn; + private String secretArn; + private String database; + private String table; + private Engine engine; + private String region; + private Integer maxRecordSizeBytes; + private RdsDataClient client; + + public Builder resourceArn(String resourceArn) { + this.resourceArn = resourceArn; + return this; + } + + public Builder secretArn(String secretArn) { + this.secretArn = secretArn; + return this; + } + + public Builder database(String database) { + this.database = database; + return this; + } + + /** Table name; letters, digits, and underscores only. Default {@code workflow_insight}. */ + public Builder table(String table) { + this.table = table; + return this; + } + + public Builder engine(Engine engine) { + this.engine = engine; + 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(RdsDataClient client) { + this.client = client; + return this; + } + + public AuroraExporter build() { + return new AuroraExporter(this); + } + } +} 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 index 3074cab65..75ef232b4 100644 --- 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 @@ -6,13 +6,11 @@ import java.util.List; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; -import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.awscore.exception.AwsServiceException; 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.annotations.Experimental; import software.amazon.lambda.durable.insight.InsightExporter; import software.amazon.lambda.durable.insight.Json; @@ -28,7 +26,7 @@ public final class CloudWatchLogsExporter implements InsightExporter { private final String logGroupName; private final String logStreamPrefix; private final Integer maxRecordSizeBytes; - private final CloudWatchLogsClient client; + private final LazyClient client; // Concurrent set: the plugin can emit from multiple threads (e.g. concurrent child-context branches), so the // create-once cache must be thread-safe. An unsynchronized HashSet could corrupt its internal table or spin under // concurrent structural modification. @@ -38,11 +36,11 @@ 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(); + this.client = LazyClient.forSdkClient( + b.client, + "cloudwatchlogs", + "software.amazon.awssdk.services.cloudwatchlogs.CloudWatchLogsClient", + b.region); } public static Builder builder() { @@ -61,9 +59,10 @@ public Object render(WorkflowInsightRecord record) { @Override public void export(WorkflowInsightRecord record) { + CloudWatchLogsClient logs = client.get(); String streamName = buildStreamName(); - ensureStream(streamName); - client.putLogEvents(PutLogEventsRequest.builder() + ensureStream(logs, streamName); + logs.putLogEvents(PutLogEventsRequest.builder() .logGroupName(logGroupName) .logStreamName(streamName) .logEvents(List.of(InputLogEvent.builder() @@ -78,16 +77,20 @@ private String buildStreamName() { return String.format("%s%d/%02d/%02d", logStreamPrefix, d.getYear(), d.getMonthValue(), d.getDayOfMonth()); } - private void ensureStream(String streamName) { + private void ensureStream(CloudWatchLogsClient logs, String streamName) { if (createdStreams.contains(streamName)) { return; } try { - client.createLogStream(CreateLogStreamRequest.builder() + logs.createLogStream(CreateLogStreamRequest.builder() .logGroupName(logGroupName) .logStreamName(streamName) .build()); - } catch (ResourceAlreadyExistsException ignored) { + } catch (AwsServiceException e) { + // The catch names a core type so linking this class never needs the service artifact. + if (!"ResourceAlreadyExistsException".equals(e.awsErrorDetails().errorCode())) { + throw e; + } // stream already exists — fine } createdStreams.add(streamName); diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/DynamoDBExporter.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/DynamoDBExporter.java new file mode 100644 index 000000000..f84e0ee2f --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/DynamoDBExporter.java @@ -0,0 +1,151 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight.exporters; + +import static java.util.Objects.requireNonNull; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import software.amazon.awssdk.services.dynamodb.DynamoDbClient; +import software.amazon.awssdk.services.dynamodb.model.AttributeValue; +import software.amazon.awssdk.services.dynamodb.model.PutItemRequest; +import software.amazon.lambda.durable.annotations.Experimental; +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 DynamoDB, one PutItem per record keyed by execution ARN. With the default + * sort key ({@code emittedAt}) every export adds a history item; with the sort key disabled each export overwrites the + * execution's item. Emits the {@code operationsByName} map. Requires {@code dynamodb:PutItem}. + */ +@Experimental +public final class DynamoDBExporter implements InsightExporter { + private final String tableName; + private final String partitionKey; + private final String sortKey; + private final Integer maxRecordSizeBytes; + private final LazyClient client; + + private DynamoDBExporter(Builder b) { + this.tableName = requireNonNull(b.tableName, "tableName"); + this.partitionKey = b.partitionKey != null ? b.partitionKey : "pk"; + String sk = b.sortKey != null ? b.sortKey : "sk"; + this.sortKey = sk.isEmpty() ? null : sk; + this.maxRecordSizeBytes = b.maxRecordSizeBytes != null ? b.maxRecordSizeBytes : 400_000; + this.client = LazyClient.forSdkClient( + b.client, "dynamodb", "software.amazon.awssdk.services.dynamodb.DynamoDbClient", b.region); + } + + 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) { + Map item = record.toByNameWireMap(); + item.put(partitionKey, record.executionArn()); + if (sortKey != null) { + item.put(sortKey, item.get("emittedAt")); + } + client.get() + .putItem(PutItemRequest.builder() + .tableName(tableName) + .item(marshalMap(item)) + .build()); + } + + private static Map marshalMap(Map map) { + Map out = new LinkedHashMap<>(); + for (Map.Entry e : map.entrySet()) { + out.put(String.valueOf(e.getKey()), marshal(e.getValue())); + } + return out; + } + + private static AttributeValue marshal(Object value) { + Object v = Json.toJsonValue(value); + if (v == null) { + return AttributeValue.builder().nul(true).build(); + } + if (v instanceof String s) { + return AttributeValue.builder().s(s).build(); + } + if (v instanceof Boolean bool) { + return AttributeValue.builder().bool(bool).build(); + } + if (v instanceof Number n) { + return AttributeValue.builder().n(n.toString()).build(); + } + if (v instanceof Map m) { + return AttributeValue.builder().m(marshalMap(m)).build(); + } + if (v instanceof List list) { + List items = new ArrayList<>(list.size()); + for (Object o : list) { + items.add(marshal(o)); + } + return AttributeValue.builder().l(items).build(); + } + return AttributeValue.builder().s(v.toString()).build(); + } + + /** Builder for {@link DynamoDBExporter}. */ + public static final class Builder { + private String tableName; + private String partitionKey; + private String sortKey; + private String region; + private Integer maxRecordSizeBytes; + private DynamoDbClient client; + + public Builder tableName(String tableName) { + this.tableName = tableName; + return this; + } + + /** Partition key attribute name; its value is the execution ARN. Default {@code pk}. */ + public Builder partitionKey(String partitionKey) { + this.partitionKey = partitionKey; + return this; + } + + /** Sort key attribute name; its value is {@code emittedAt}. Default {@code sk}; an empty string disables it. */ + public Builder sortKey(String sortKey) { + this.sortKey = sortKey; + 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(DynamoDbClient client) { + this.client = client; + return this; + } + + public DynamoDBExporter build() { + return new DynamoDBExporter(this); + } + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/EventBridgeExporter.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/EventBridgeExporter.java new file mode 100644 index 000000000..e11f2877d --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/EventBridgeExporter.java @@ -0,0 +1,126 @@ +// 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.Map; +import software.amazon.awssdk.services.eventbridge.EventBridgeClient; +import software.amazon.awssdk.services.eventbridge.model.PutEventsRequest; +import software.amazon.awssdk.services.eventbridge.model.PutEventsRequestEntry; +import software.amazon.awssdk.services.eventbridge.model.PutEventsResponse; +import software.amazon.awssdk.services.eventbridge.model.PutEventsResultEntry; +import software.amazon.lambda.durable.annotations.Experimental; +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 EventBridge, one PutEvents entry per record with the record status as + * {@code DetailType} and the record JSON as {@code Detail}. Requires {@code events:PutEvents}. + */ +@Experimental +public final class EventBridgeExporter implements InsightExporter { + /** + * Default event source. The {@code aws.} prefix is reserved for AWS service events and is rejected by PutEvents. + */ + public static final String DEFAULT_SOURCE = "software.amazon.lambda.durable.insight"; + + private final String eventBusName; + private final String source; + private final OperationsFormat operationsFormat; + private final Integer maxRecordSizeBytes; + private final LazyClient client; + + private EventBridgeExporter(Builder b) { + this.eventBusName = b.eventBusName != null ? b.eventBusName : "default"; + this.source = b.source != null ? b.source : DEFAULT_SOURCE; + this.operationsFormat = b.operationsFormat != null ? b.operationsFormat : OperationsFormat.ARRAY; + this.maxRecordSizeBytes = b.maxRecordSizeBytes != null ? b.maxRecordSizeBytes : 256_000; + this.client = LazyClient.forSdkClient( + b.client, "eventbridge", "software.amazon.awssdk.services.eventbridge.EventBridgeClient", b.region); + } + + public static Builder builder() { + return new Builder(); + } + + @Override + public Integer maxRecordSizeBytes() { + return maxRecordSizeBytes; + } + + @Override + public Object render(WorkflowInsightRecord record) { + return operationsFormat.apply(record); + } + + @Override + public void export(WorkflowInsightRecord record) { + EventBridgeClient eventBridge = client.get(); + Map detail = operationsFormat.apply(record); + PutEventsRequestEntry entry = PutEventsRequestEntry.builder() + .eventBusName(eventBusName) + .source(source) + .detailType(record.status()) + .detail(Json.stringify(detail)) + .time(Instant.parse((String) detail.get("emittedAt"))) + .build(); + PutEventsResponse response = + eventBridge.putEvents(PutEventsRequest.builder().entries(entry).build()); + Integer failed = response.failedEntryCount(); + if (failed != null && failed > 0) { + PutEventsResultEntry result = + response.entries().isEmpty() ? null : response.entries().get(0); + throw new IllegalStateException("EventBridge PutEvents failed: " + + (result != null ? result.errorCode() : null) + " — " + + (result != null ? result.errorMessage() : null)); + } + } + + /** Builder for {@link EventBridgeExporter}. */ + public static final class Builder { + private String eventBusName; + private String source; + private String region; + private OperationsFormat operationsFormat; + private Integer maxRecordSizeBytes; + private EventBridgeClient client; + + /** Event bus name or ARN. Default {@code default}. */ + public Builder eventBusName(String eventBusName) { + this.eventBusName = eventBusName; + return this; + } + + /** Event source. Default {@link EventBridgeExporter#DEFAULT_SOURCE}. */ + public Builder source(String source) { + this.source = source; + return this; + } + + public Builder region(String region) { + this.region = region; + return this; + } + + public Builder operationsFormat(OperationsFormat operationsFormat) { + this.operationsFormat = operationsFormat; + return this; + } + + public Builder maxRecordSizeBytes(Integer maxRecordSizeBytes) { + this.maxRecordSizeBytes = maxRecordSizeBytes; + return this; + } + + /** Test seam: inject a client. */ + public Builder client(EventBridgeClient client) { + this.client = client; + return this; + } + + public EventBridgeExporter build() { + return new EventBridgeExporter(this); + } + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/FileExporter.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/FileExporter.java new file mode 100644 index 000000000..6d14f6c06 --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/FileExporter.java @@ -0,0 +1,158 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight.exporters; + +import static java.util.Objects.requireNonNull; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.Map; +import software.amazon.lambda.durable.annotations.Experimental; +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 directory: appended as one JSON line per record to a date-named NDJSON file + * (default), or written as one pretty-printed JSON file per execution that later exports overwrite. Suitable for an EFS + * mount or {@code /tmp}. No truncation limit by default. + */ +@Experimental +public final class FileExporter implements InsightExporter { + + /** File layout. */ + @Experimental + public enum Mode { + /** Append every record to {@code {directory}/{YYYY-MM-DD}.ndjson}. */ + NDJSON("ndjson"), + /** Write {@code {directory}/{executionName}.json}, overwriting on each export. */ + JSON("json"); + + private final String value; + + Mode(String value) { + this.value = value; + } + + /** The configuration string for this mode. */ + public String value() { + return value; + } + + /** Parses a configuration string; unknown values are rejected. */ + public static Mode fromValue(String value) { + for (Mode m : values()) { + if (m.value.equals(value)) { + return m; + } + } + throw new IllegalArgumentException("Unknown mode: \"" + value + "\". Expected ndjson or json."); + } + } + + private final Path directory; + private final Mode mode; + private final OperationsFormat operationsFormat; + private final Integer maxRecordSizeBytes; + private final Object appendLock = new Object(); + private volatile boolean directoryCreated; + + private FileExporter(Builder b) { + this.directory = Path.of(requireNonNull(b.directory, "directory")); + this.mode = b.mode != null ? b.mode : Mode.NDJSON; + this.operationsFormat = b.operationsFormat != null ? b.operationsFormat : OperationsFormat.ARRAY; + this.maxRecordSizeBytes = b.maxRecordSizeBytes; + } + + public static Builder builder() { + return new Builder(); + } + + @Override + public Integer maxRecordSizeBytes() { + return maxRecordSizeBytes; + } + + @Override + public Object render(WorkflowInsightRecord record) { + return operationsFormat.apply(record); + } + + @Override + public void export(WorkflowInsightRecord record) { + Map formatted = operationsFormat.apply(record); + try { + ensureDirectory(); + if (mode == Mode.NDJSON) { + String date = ((String) formatted.get("emittedAt")).substring(0, 10); + byte[] line = (Json.stringify(formatted) + "\n").getBytes(StandardCharsets.UTF_8); + // Appends are written in chunks, so concurrent exports (child-context branches) could interleave lines + // longer than one chunk. Hold a lock for the whole record so each line lands intact. + synchronized (appendLock) { + Files.write( + directory.resolve(date + ".ndjson"), + line, + StandardOpenOption.CREATE, + StandardOpenOption.APPEND); + } + } else { + String name = record.executionName() != null ? record.executionName() : record.executionArn(); + Files.write( + directory.resolve(sanitize(name) + ".json"), + Json.prettyStringify(formatted).getBytes(StandardCharsets.UTF_8)); + } + } catch (IOException e) { + throw new UncheckedIOException("FileExporter: write to " + directory + " failed", e); + } + } + + private void ensureDirectory() throws IOException { + if (!directoryCreated) { + Files.createDirectories(directory); + directoryCreated = true; + } + } + + private static String sanitize(String value) { + return value.replaceAll("[^a-zA-Z0-9._-]", "_"); + } + + /** Builder for {@link FileExporter}. */ + public static final class Builder { + private String directory; + private Mode mode; + private OperationsFormat operationsFormat; + private Integer maxRecordSizeBytes; + + /** Base directory, for example {@code /mnt/efs/workflow-insight} or {@code /tmp/insight}. */ + public Builder directory(String directory) { + this.directory = directory; + return this; + } + + /** Default {@code NDJSON}. */ + public Builder mode(Mode mode) { + this.mode = mode; + return this; + } + + public Builder operationsFormat(OperationsFormat operationsFormat) { + this.operationsFormat = operationsFormat; + return this; + } + + /** No default: the filesystem has no practical per-record limit. */ + public Builder maxRecordSizeBytes(Integer maxRecordSizeBytes) { + this.maxRecordSizeBytes = maxRecordSizeBytes; + return this; + } + + public FileExporter build() { + return new FileExporter(this); + } + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/FirehoseExporter.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/FirehoseExporter.java new file mode 100644 index 000000000..ecd7c8c94 --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/FirehoseExporter.java @@ -0,0 +1,99 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight.exporters; + +import static java.util.Objects.requireNonNull; + +import software.amazon.awssdk.core.SdkBytes; +import software.amazon.awssdk.services.firehose.FirehoseClient; +import software.amazon.awssdk.services.firehose.model.PutRecordRequest; +import software.amazon.awssdk.services.firehose.model.Record; +import software.amazon.lambda.durable.annotations.Experimental; +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 Data Firehose, one PutRecord per record as a newline-terminated JSON line + * so concatenated deliveries stay parseable as NDJSON. Requires {@code firehose:PutRecord}. + */ +@Experimental +public final class FirehoseExporter implements InsightExporter { + private final String deliveryStreamName; + private final OperationsFormat operationsFormat; + private final Integer maxRecordSizeBytes; + private final LazyClient client; + + private FirehoseExporter(Builder b) { + this.deliveryStreamName = requireNonNull(b.deliveryStreamName, "deliveryStreamName"); + this.operationsFormat = b.operationsFormat != null ? b.operationsFormat : OperationsFormat.ARRAY; + this.maxRecordSizeBytes = b.maxRecordSizeBytes != null ? b.maxRecordSizeBytes : 1_000_000; + this.client = LazyClient.forSdkClient( + b.client, "firehose", "software.amazon.awssdk.services.firehose.FirehoseClient", b.region); + } + + public static Builder builder() { + return new Builder(); + } + + @Override + public Integer maxRecordSizeBytes() { + return maxRecordSizeBytes; + } + + @Override + public Object render(WorkflowInsightRecord record) { + return operationsFormat.apply(record); + } + + @Override + public void export(WorkflowInsightRecord record) { + String data = Json.stringify(render(record)) + "\n"; + client.get() + .putRecord(PutRecordRequest.builder() + .deliveryStreamName(deliveryStreamName) + .record(Record.builder() + .data(SdkBytes.fromUtf8String(data)) + .build()) + .build()); + } + + /** Builder for {@link FirehoseExporter}. */ + public static final class Builder { + private String deliveryStreamName; + private String region; + private OperationsFormat operationsFormat; + private Integer maxRecordSizeBytes; + private FirehoseClient client; + + public Builder deliveryStreamName(String deliveryStreamName) { + this.deliveryStreamName = deliveryStreamName; + return this; + } + + public Builder region(String region) { + this.region = region; + return this; + } + + public Builder operationsFormat(OperationsFormat operationsFormat) { + this.operationsFormat = operationsFormat; + return this; + } + + public Builder maxRecordSizeBytes(Integer maxRecordSizeBytes) { + this.maxRecordSizeBytes = maxRecordSizeBytes; + return this; + } + + /** Test seam: inject a client. */ + public Builder client(FirehoseClient client) { + this.client = client; + return this; + } + + public FirehoseExporter build() { + return new FirehoseExporter(this); + } + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/HttpExporter.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/HttpExporter.java new file mode 100644 index 000000000..a629313e6 --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/HttpExporter.java @@ -0,0 +1,155 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight.exporters; + +import static java.util.Objects.requireNonNull; + +import java.io.IOException; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Map; +import software.amazon.lambda.durable.annotations.Experimental; +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 an HTTP endpoint as a JSON body, one request per record. A non-2xx response fails + * the export. No truncation limit by default; set one if the endpoint caps request size. + */ +@Experimental +public final class HttpExporter implements InsightExporter { + + /** Request method. */ + @Experimental + public enum Method { + POST, + PUT; + + /** Parses a configuration string; unknown values are rejected. */ + public static Method fromValue(String value) { + for (Method m : values()) { + if (m.name().equals(value)) { + return m; + } + } + throw new IllegalArgumentException("Unknown method: \"" + value + "\". Expected POST or PUT."); + } + } + + private final URI url; + private final Method method; + private final Map headers; + private final Duration timeout; + private final OperationsFormat operationsFormat; + private final Integer maxRecordSizeBytes; + private final HttpClient httpClient; + + private HttpExporter(Builder b) { + this.url = URI.create(requireNonNull(b.url, "url")); + this.method = b.method != null ? b.method : Method.POST; + this.headers = b.headers != null ? Map.copyOf(b.headers) : Map.of(); + this.timeout = Duration.ofMillis(b.timeoutMs != null ? b.timeoutMs : 10_000L); + this.operationsFormat = b.operationsFormat != null ? b.operationsFormat : OperationsFormat.ARRAY; + this.maxRecordSizeBytes = b.maxRecordSizeBytes; + this.httpClient = b.httpClient != null + ? b.httpClient + : HttpClient.newBuilder().connectTimeout(timeout).build(); + } + + public static Builder builder() { + return new Builder(); + } + + @Override + public Integer maxRecordSizeBytes() { + return maxRecordSizeBytes; + } + + @Override + public Object render(WorkflowInsightRecord record) { + return operationsFormat.apply(record); + } + + @Override + public void export(WorkflowInsightRecord record) { + String body = Json.stringify(render(record)); + HttpRequest.Builder rb = HttpRequest.newBuilder(url) + .method(method.name(), HttpRequest.BodyPublishers.ofString(body, StandardCharsets.UTF_8)) + .timeout(timeout) + .setHeader("Content-Type", "application/json"); + headers.forEach(rb::setHeader); + HttpResponse response; + try { + response = httpClient.send(rb.build(), HttpResponse.BodyHandlers.discarding()); + } catch (IOException e) { + throw new IllegalStateException("HttpExporter: request to " + url + " failed", e); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("HttpExporter: interrupted while sending to " + url, e); + } + int status = response.statusCode(); + if (status < 200 || status >= 300) { + throw new IllegalStateException("HttpExporter: endpoint returned " + status); + } + } + + /** Builder for {@link HttpExporter}. */ + public static final class Builder { + private String url; + private Map headers; + private Method method; + private Long timeoutMs; + private OperationsFormat operationsFormat; + private Integer maxRecordSizeBytes; + private HttpClient httpClient; + + public Builder url(String url) { + this.url = url; + return this; + } + + /** Extra request headers, such as authorization tokens; may override {@code Content-Type}. */ + public Builder headers(Map headers) { + this.headers = headers; + return this; + } + + /** Default {@code POST}. */ + public Builder method(Method method) { + this.method = method; + return this; + } + + /** Request timeout in milliseconds. Default 10000. */ + public Builder timeoutMs(Long timeoutMs) { + this.timeoutMs = timeoutMs; + return this; + } + + public Builder operationsFormat(OperationsFormat operationsFormat) { + this.operationsFormat = operationsFormat; + return this; + } + + /** No default: a generic endpoint has no known limit. */ + public Builder maxRecordSizeBytes(Integer maxRecordSizeBytes) { + this.maxRecordSizeBytes = maxRecordSizeBytes; + return this; + } + + /** Test seam: inject an HTTP client. */ + public Builder httpClient(HttpClient httpClient) { + this.httpClient = httpClient; + return this; + } + + public HttpExporter build() { + return new HttpExporter(this); + } + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/LazyClient.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/LazyClient.java new file mode 100644 index 000000000..bfd427288 --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/LazyClient.java @@ -0,0 +1,80 @@ +// 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.lang.reflect.InvocationTargetException; +import java.util.function.Supplier; +import software.amazon.awssdk.awscore.client.builder.AwsClientBuilder; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.utils.builder.SdkBuilder; + +/** + * Holds an exporter's transport client: an injected instance, or one created on first use from an optional AWS SDK + * artifact. A missing artifact fails at first use with a message naming it instead of a bare linkage error. + */ +final class LazyClient { + private final String artifact; + private final Supplier factory; + private volatile T client; + + LazyClient(T injected, String artifact, Supplier factory) { + this.client = injected; + this.artifact = artifact; + this.factory = factory; + } + + /** + * A holder for an AWS SDK sync client named by class, built with {@code builder()} and the optional region. The + * class is looked up by name on first use, so an exporter can be built while its artifact is absent and the failure + * surfaces at export, where the plugin isolates it. + */ + static LazyClient forSdkClient(T injected, String artifact, String clientClassName, String region) { + return new LazyClient<>(injected, artifact, () -> buildSdkClient(clientClassName, region)); + } + + T get() { + T c = client; + if (c == null) { + synchronized (this) { + c = client; + if (c == null) { + c = create(); + client = c; + } + } + } + return c; + } + + private T create() { + try { + return factory.get(); + } catch (NoClassDefFoundError | MissingArtifactException e) { + throw new IllegalStateException( + "Missing dependency software.amazon.awssdk:" + artifact + " required by this exporter", e); + } + } + + @SuppressWarnings("unchecked") + private static T buildSdkClient(String clientClassName, String region) { + try { + Object builder = Class.forName(clientClassName).getMethod("builder").invoke(null); + if (region != null) { + ((AwsClientBuilder) builder).region(Region.of(region)); + } + return (T) ((SdkBuilder) builder).build(); + } catch (ClassNotFoundException e) { + throw new MissingArtifactException(e); + } catch (ReflectiveOperationException e) { + Throwable cause = e instanceof InvocationTargetException ? e.getCause() : e; + throw new IllegalStateException("Failed to create " + clientClassName, cause); + } + } + + /** Signals that the client class could not be found. */ + private static final class MissingArtifactException extends RuntimeException { + MissingArtifactException(Throwable cause) { + super(cause); + } + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/OTelExporter.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/OTelExporter.java new file mode 100644 index 000000000..6c01feb89 --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/OTelExporter.java @@ -0,0 +1,232 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight.exporters; + +import static java.util.Objects.requireNonNull; + +import java.io.IOException; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import software.amazon.lambda.durable.annotations.Experimental; +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 as OpenTelemetry log records over OTLP/HTTP JSON. Each record becomes one log record + * with identity fields as resource and log attributes and the full record JSON as the body. Compatible with any + * OTLP-capable backend; authentication is via headers. + */ +@Experimental +public final class OTelExporter implements InsightExporter { + + /** OTLP transport encoding. */ + @Experimental + public enum Protocol { + HTTP_JSON("http/json"), + HTTP_PROTOBUF("http/protobuf"); + + private final String value; + + Protocol(String value) { + this.value = value; + } + + /** The configuration string for this protocol. */ + public String value() { + return value; + } + + /** Parses a configuration string; unknown values are rejected. */ + public static Protocol fromValue(String value) { + for (Protocol p : values()) { + if (p.value.equals(value)) { + return p; + } + } + throw new IllegalArgumentException( + "Unknown protocol: \"" + value + "\". Expected http/json or http/protobuf."); + } + } + + private static final String SCOPE_NAME = "software.amazon.lambda.durable.insight"; + + private final URI endpoint; + private final Map headers; + private final OperationsFormat operationsFormat; + private final Integer maxRecordSizeBytes; + private final HttpClient httpClient; + + private OTelExporter(Builder b) { + if (b.protocol == Protocol.HTTP_PROTOBUF) { + throw new IllegalArgumentException("OTelExporter: http/protobuf is not yet supported. Use http/json."); + } + this.endpoint = URI.create(requireNonNull(b.endpoint, "endpoint")); + this.headers = b.headers != null ? Map.copyOf(b.headers) : Map.of(); + this.operationsFormat = b.operationsFormat != null ? b.operationsFormat : OperationsFormat.ARRAY; + this.maxRecordSizeBytes = b.maxRecordSizeBytes != null ? b.maxRecordSizeBytes : 1_000_000; + this.httpClient = b.httpClient != null ? b.httpClient : HttpClient.newHttpClient(); + } + + public static Builder builder() { + return new Builder(); + } + + @Override + public Integer maxRecordSizeBytes() { + return maxRecordSizeBytes; + } + + /** The full OTLP request, so size limits cover the record content and the OTLP envelope together. */ + @Override + public Object render(WorkflowInsightRecord record) { + return buildPayload(record); + } + + @Override + public void export(WorkflowInsightRecord record) { + String body = Json.stringify(buildPayload(record)); + HttpRequest.Builder rb = HttpRequest.newBuilder(endpoint) + .POST(HttpRequest.BodyPublishers.ofString(body, StandardCharsets.UTF_8)) + .setHeader("Content-Type", "application/json"); + headers.forEach(rb::setHeader); + HttpResponse response; + try { + response = httpClient.send(rb.build(), HttpResponse.BodyHandlers.discarding()); + } catch (IOException e) { + throw new IllegalStateException("OTelExporter: request to OTLP endpoint failed", e); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("OTelExporter: interrupted while sending to OTLP endpoint", e); + } + int status = response.statusCode(); + if (status < 200 || status >= 300) { + throw new IllegalStateException("OTelExporter: OTLP endpoint returned " + status); + } + } + + private Map buildPayload(WorkflowInsightRecord record) { + Map wire = record.toWireMap(); + String functionName = record.functionName(); + Map resource = Map.of( + "attributes", + List.of( + kv("service.name", functionName), + kv("cloud.region", (String) wire.get("region")), + kv("cloud.account.id", (String) wire.get("accountId")), + kv("faas.name", functionName), + kv("faas.version", (String) wire.get("functionQualifier")))); + Map scope = new LinkedHashMap<>(); + scope.put("name", SCOPE_NAME); + scope.put("version", wire.get("schemaVersion")); + + Long durationMs = (Long) wire.get("durationMs"); + String executionName = record.executionName(); + Map logRecord = new LinkedHashMap<>(); + logRecord.put("timeUnixNano", toNano((String) wire.get("emittedAt"))); + logRecord.put("severityNumber", severityFor(record.status())); + logRecord.put("severityText", record.status()); + logRecord.put("body", Map.of("stringValue", Json.stringify(operationsFormat.apply(record)))); + logRecord.put( + "attributes", + List.of( + kv("workflow.execution_arn", record.executionArn()), + kv("workflow.execution_name", executionName != null ? executionName : ""), + kv("workflow.status", record.status()), + kv("workflow.duration_ms", durationMs != null ? durationMs : 0L))); + + Map scopeLogs = new LinkedHashMap<>(); + scopeLogs.put("scope", scope); + scopeLogs.put("logRecords", List.of(logRecord)); + Map resourceLogs = new LinkedHashMap<>(); + resourceLogs.put("resource", resource); + resourceLogs.put("scopeLogs", List.of(scopeLogs)); + Map payload = new LinkedHashMap<>(); + payload.put("resourceLogs", List.of(resourceLogs)); + return payload; + } + + /** String attribute; an absent value renders as an empty {@code value} object. */ + private static Map kv(String key, String value) { + Map attr = new LinkedHashMap<>(); + attr.put("key", key); + attr.put("value", value != null ? Map.of("stringValue", value) : Map.of()); + return attr; + } + + /** Integer attribute; OTLP JSON carries 64-bit integers as strings. */ + private static Map kv(String key, long value) { + Map attr = new LinkedHashMap<>(); + attr.put("key", key); + attr.put("value", Map.of("intValue", Long.toString(value))); + return attr; + } + + private static String toNano(String isoTimestamp) { + return Long.toString(Instant.parse(isoTimestamp).toEpochMilli() * 1_000_000L); + } + + private static int severityFor(String status) { + if ("FAILED".equals(status)) { + return 17; // ERROR + } + if ("RUNNING".equals(status) || "SUCCEEDED".equals(status)) { + return 9; // INFO + } + return 0; // UNSPECIFIED + } + + /** Builder for {@link OTelExporter}. */ + public static final class Builder { + private String endpoint; + private Map headers; + private Protocol protocol; + private OperationsFormat operationsFormat; + private Integer maxRecordSizeBytes; + private HttpClient httpClient; + + /** OTLP logs endpoint, for example {@code http://localhost:4318/v1/logs}. */ + public Builder endpoint(String endpoint) { + this.endpoint = endpoint; + return this; + } + + public Builder headers(Map headers) { + this.headers = headers; + return this; + } + + /** Default {@code HTTP_JSON}; {@code HTTP_PROTOBUF} is rejected at build time. */ + public Builder protocol(Protocol protocol) { + this.protocol = protocol; + return this; + } + + public Builder operationsFormat(OperationsFormat operationsFormat) { + this.operationsFormat = operationsFormat; + return this; + } + + public Builder maxRecordSizeBytes(Integer maxRecordSizeBytes) { + this.maxRecordSizeBytes = maxRecordSizeBytes; + return this; + } + + /** Test seam: inject an HTTP client. */ + public Builder httpClient(HttpClient httpClient) { + this.httpClient = httpClient; + return this; + } + + public OTelExporter build() { + return new OTelExporter(this); + } + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/OpenSearchExporter.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/OpenSearchExporter.java new file mode 100644 index 000000000..6725a2cfe --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/OpenSearchExporter.java @@ -0,0 +1,252 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight.exporters; + +import static java.util.Objects.requireNonNull; + +import java.io.IOException; +import java.net.URI; +import java.net.URLEncoder; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.util.Base64; +import java.util.List; +import java.util.Map; +import java.util.Set; +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; +import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider; +import software.amazon.awssdk.http.ContentStreamProvider; +import software.amazon.awssdk.http.SdkHttpMethod; +import software.amazon.awssdk.http.SdkHttpRequest; +import software.amazon.awssdk.http.auth.aws.signer.AwsV4HttpSigner; +import software.amazon.awssdk.http.auth.spi.signer.SignedRequest; +import software.amazon.lambda.durable.annotations.Experimental; +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 OpenSearch Service with the index API, using the execution ARN as the + * document id so later exports overwrite the same document. Authenticates with SigV4 (default) or HTTP basic auth. + * Requires {@code es:ESHttpPut} on the index for SigV4. + */ +@Experimental +public final class OpenSearchExporter implements InsightExporter { + + /** Authentication method. */ + @Experimental + public enum Auth { + /** IAM request signing for Amazon OpenSearch Service (default). */ + SIGV4("sigv4"), + /** Username and password. */ + BASIC("basic"); + + private final String value; + + Auth(String value) { + this.value = value; + } + + /** The configuration string for this method. */ + public String value() { + return value; + } + + /** Parses a configuration string; unknown values are rejected. */ + public static Auth fromValue(String value) { + for (Auth a : values()) { + if (a.value.equals(value)) { + return a; + } + } + throw new IllegalArgumentException("Unknown auth: \"" + value + "\". Expected sigv4 or basic."); + } + } + + /** Headers the JDK HTTP client manages itself and refuses to accept from callers. */ + private static final Set CLIENT_MANAGED_HEADERS = Set.of("host", "content-length"); + + private final String endpoint; + private final String indexName; + private final String region; + private final Auth auth; + private final String username; + private final String password; + private final Integer maxRecordSizeBytes; + private final HttpClient httpClient; + private final LazyClient signer; + + private OpenSearchExporter(Builder b) { + this.endpoint = requireNonNull(b.endpoint, "endpoint").replaceAll("/$", ""); + this.indexName = b.indexName != null ? b.indexName : "workflow-insight"; + this.region = b.region; + this.auth = b.auth != null ? b.auth : Auth.SIGV4; + this.username = b.username; + this.password = b.password; + this.maxRecordSizeBytes = b.maxRecordSizeBytes != null ? b.maxRecordSizeBytes : 10_000_000; + this.httpClient = b.httpClient != null ? b.httpClient : HttpClient.newHttpClient(); + if (this.auth == Auth.SIGV4) { + requireNonNull(region, "region is required for sigv4 auth"); + } else { + requireNonNull(username, "username is required for basic auth"); + requireNonNull(password, "password is required for basic auth"); + } + AwsCredentialsProvider credentials = b.credentialsProvider; + this.signer = new LazyClient<>(null, "http-auth-aws", () -> new Signer(credentials)); + } + + public static Builder builder() { + return new Builder(); + } + + @Override + public Integer maxRecordSizeBytes() { + return maxRecordSizeBytes; + } + + @Override + public void export(WorkflowInsightRecord record) { + URI url = URI.create(endpoint + "/" + indexName + "/_doc/" + encodeComponent(record.executionArn())); + String body = Json.stringify(record.toWireMap()); + HttpRequest.Builder rb = HttpRequest.newBuilder(url) + .PUT(HttpRequest.BodyPublishers.ofString(body, StandardCharsets.UTF_8)) + .setHeader("Content-Type", "application/json"); + if (auth == Auth.BASIC) { + String token = + Base64.getEncoder().encodeToString((username + ":" + password).getBytes(StandardCharsets.UTF_8)); + rb.setHeader("Authorization", "Basic " + token); + } else { + // Use the signed header set as-is: it already carries content-type, the date, and the authorization. + signer.get().sign(url, body).forEach((name, values) -> { + if (!CLIENT_MANAGED_HEADERS.contains(name.toLowerCase())) { + rb.setHeader(name, String.join(",", values)); + } + }); + } + HttpResponse response = send(rb.build()); + int status = response.statusCode(); + if (status < 200 || status >= 300) { + String detail = response.body() != null ? response.body() : ""; + throw new IllegalStateException("OpenSearch index failed: " + status + + (detail.isEmpty() ? "" : " — " + detail.substring(0, Math.min(500, detail.length())))); + } + } + + private HttpResponse send(HttpRequest request) { + try { + return httpClient.send(request, HttpResponse.BodyHandlers.ofString()); + } catch (IOException e) { + throw new IllegalStateException("OpenSearch index request failed", e); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("OpenSearch index request interrupted", e); + } + } + + /** Percent-encodes a path segment, keeping the unreserved characters {@code A-Z a-z 0-9 - _ . ! ~ * ' ( )}. */ + static String encodeComponent(String value) { + return URLEncoder.encode(value, StandardCharsets.UTF_8) + .replace("+", "%20") + .replace("%21", "!") + .replace("%27", "'") + .replace("%28", "(") + .replace("%29", ")") + .replace("%7E", "~"); + } + + /** SigV4 signing for the {@code es} service; created on first use so basic-auth users never load the signer. */ + private final class Signer { + private final AwsV4HttpSigner v4 = AwsV4HttpSigner.create(); + private final AwsCredentialsProvider credentials; + + Signer(AwsCredentialsProvider credentials) { + this.credentials = credentials != null ? credentials : DefaultCredentialsProvider.create(); + } + + Map> sign(URI url, String body) { + String host = url.getPort() == -1 ? url.getHost() : url.getHost() + ":" + url.getPort(); + SdkHttpRequest request = SdkHttpRequest.builder() + .method(SdkHttpMethod.PUT) + .uri(url) + .putHeader("host", host) + .putHeader("content-type", "application/json") + .build(); + SignedRequest signed = v4.sign(r -> r.identity(credentials.resolveCredentials()) + .request(request) + .payload(ContentStreamProvider.fromUtf8String(body)) + .putProperty(AwsV4HttpSigner.SERVICE_SIGNING_NAME, "es") + .putProperty(AwsV4HttpSigner.REGION_NAME, region)); + return signed.request().headers(); + } + } + + /** Builder for {@link OpenSearchExporter}. */ + public static final class Builder { + private String endpoint; + private String indexName; + private String region; + private Auth auth; + private String username; + private String password; + private Integer maxRecordSizeBytes; + private HttpClient httpClient; + private AwsCredentialsProvider credentialsProvider; + + /** Domain endpoint, for example {@code https://my-domain.us-east-1.es.amazonaws.com}. */ + public Builder endpoint(String endpoint) { + this.endpoint = endpoint; + return this; + } + + /** Index name. Default {@code workflow-insight}. */ + public Builder indexName(String indexName) { + this.indexName = indexName; + return this; + } + + /** Signing region; required for {@code SIGV4}. */ + public Builder region(String region) { + this.region = region; + return this; + } + + /** Default {@code SIGV4}. */ + public Builder auth(Auth auth) { + this.auth = auth; + return this; + } + + public Builder username(String username) { + this.username = username; + return this; + } + + public Builder password(String password) { + this.password = password; + return this; + } + + public Builder maxRecordSizeBytes(Integer maxRecordSizeBytes) { + this.maxRecordSizeBytes = maxRecordSizeBytes; + return this; + } + + /** Test seam: inject an HTTP client. */ + public Builder httpClient(HttpClient httpClient) { + this.httpClient = httpClient; + return this; + } + + /** Credentials for SigV4 signing; defaults to the SDK default provider chain. */ + public Builder credentialsProvider(AwsCredentialsProvider credentialsProvider) { + this.credentialsProvider = credentialsProvider; + return this; + } + + public OpenSearchExporter build() { + return new OpenSearchExporter(this); + } + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/OperationsFormat.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/OperationsFormat.java new file mode 100644 index 000000000..56e63f4c8 --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/OperationsFormat.java @@ -0,0 +1,56 @@ +// 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.util.Map; +import software.amazon.lambda.durable.annotations.Experimental; +import software.amazon.lambda.durable.insight.WorkflowInsightRecord; + +/** How an exporter renders operations in the emitted record. */ +@Experimental +public enum OperationsFormat { + /** The canonical {@code operations} array (lossless; default). */ + ARRAY("array"), + /** The {@code operationsByName} map in place of the array. */ + BY_NAME("by-name"), + /** Both the {@code operations} array and the {@code operationsByName} map. */ + BOTH("both"); + + private final String value; + + OperationsFormat(String value) { + this.value = value; + } + + /** The configuration string for this format ({@code array}, {@code by-name}, {@code both}). */ + public String value() { + return value; + } + + /** Parses a configuration string; unknown values are rejected. */ + public static OperationsFormat fromValue(String value) { + for (OperationsFormat f : values()) { + if (f.value.equals(value)) { + return f; + } + } + throw new IllegalArgumentException( + "Unknown operationsFormat: \"" + value + "\". Expected array, by-name, or both."); + } + + /** Renders the record's wire map in this format. */ + public Map apply(WorkflowInsightRecord record) { + switch (this) { + case BY_NAME: + return record.toByNameWireMap(); + case BOTH: { + Map data = record.toWireMap(); + data.put("operationsByName", record.toByNameWireMap().get("operationsByName")); + return data; + } + case ARRAY: + default: + return record.toWireMap(); + } + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/RedshiftExporter.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/RedshiftExporter.java new file mode 100644 index 000000000..3d3a921c7 --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/RedshiftExporter.java @@ -0,0 +1,225 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight.exporters; + +import static java.util.Objects.requireNonNull; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import software.amazon.awssdk.services.redshiftdata.RedshiftDataClient; +import software.amazon.awssdk.services.redshiftdata.model.ExecuteStatementRequest; +import software.amazon.awssdk.services.redshiftdata.model.SqlParameter; +import software.amazon.lambda.durable.annotations.Experimental; +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 Redshift (provisioned or Serverless) through the Redshift Data API, + * upserting one row per execution with a MERGE keyed by execution ARN. The statement is submitted without waiting for + * completion. Requires {@code redshift-data:ExecuteStatement} plus the credential action for the target + * ({@code redshift-serverless:GetCredentials}, {@code redshift:GetClusterCredentialsWithIAM}, or + * {@code secretsmanager:GetSecretValue}). + */ +@Experimental +public final class RedshiftExporter implements InsightExporter { + private final String database; + private final String fqTable; + private final String workgroupName; + private final String clusterIdentifier; + private final String dbUser; + private final String secretArn; + private final Integer maxRecordSizeBytes; + private final LazyClient client; + + private RedshiftExporter(Builder b) { + if (b.workgroupName == null && b.clusterIdentifier == null) { + throw new IllegalArgumentException("RedshiftExporter: provide either workgroupName or clusterIdentifier."); + } + if (b.workgroupName != null && b.clusterIdentifier != null) { + throw new IllegalArgumentException("RedshiftExporter: workgroupName and clusterIdentifier are exclusive."); + } + if (b.dbUser != null && b.secretArn != null) { + throw new IllegalArgumentException("RedshiftExporter: dbUser and secretArn are exclusive."); + } + this.database = requireNonNull(b.database, "database"); + String table = SqlIdentifiers.validate(b.table != null ? b.table : "workflow_insight"); + String schema = SqlIdentifiers.validate(b.schema != null ? b.schema : "public"); + this.fqTable = schema + "." + table; + this.workgroupName = b.workgroupName; + this.clusterIdentifier = b.clusterIdentifier; + this.dbUser = b.dbUser; + this.secretArn = b.secretArn; + this.maxRecordSizeBytes = b.maxRecordSizeBytes != null ? b.maxRecordSizeBytes : 1_000_000; + this.client = LazyClient.forSdkClient( + b.client, "redshiftdata", "software.amazon.awssdk.services.redshiftdata.RedshiftDataClient", b.region); + } + + public static Builder builder() { + return new Builder(); + } + + @Override + public Integer maxRecordSizeBytes() { + return maxRecordSizeBytes; + } + + @Override + public void export(WorkflowInsightRecord record) { + RedshiftDataClient redshift = client.get(); + Map wire = record.toWireMap(); + List parameters = new ArrayList<>(); + parameters.add(param("execution_arn", record.executionArn())); + parameters.add(param("function_name", record.functionName())); + parameters.add(param("status", record.status())); + parameters.add(param("record_json", Json.stringify(wire))); + parameters.add(param("emitted_at", (String) wire.get("emittedAt"))); + + // The Data API rejects NULL and empty-string parameter values, so an absent value becomes a typed NULL + // literal in the source projection instead of a bound parameter. + String startTime = record.startTime(); + String startTimeSel = "NULL::timestamptz"; + if (startTime != null) { + parameters.add(param("start_time", startTime)); + startTimeSel = ":start_time::timestamptz"; + } + String endTime = (String) wire.get("endTime"); + String endTimeSel = "NULL::timestamptz"; + if (endTime != null) { + parameters.add(param("end_time", endTime)); + endTimeSel = ":end_time::timestamptz"; + } + Long durationMs = (Long) wire.get("durationMs"); + String durationSel = "NULL::bigint"; + if (durationMs != null) { + parameters.add(param("duration_ms", Long.toString(durationMs))); + durationSel = ":duration_ms::bigint"; + } + String executionName = record.executionName(); + String execNameSel = "NULL::varchar"; + if (executionName != null) { + parameters.add(param("execution_name", executionName)); + execNameSel = ":execution_name::varchar"; + } + + redshift.executeStatement(ExecuteStatementRequest.builder() + .workgroupName(workgroupName) + .clusterIdentifier(clusterIdentifier) + .database(database) + .dbUser(dbUser) + .secretArn(secretArn) + .sql(buildMerge(execNameSel, startTimeSel, endTimeSel, durationSel)) + .parameters(parameters) + .build()); + } + + private static SqlParameter param(String name, String value) { + return SqlParameter.builder().name(name).value(value).build(); + } + + /** + * MERGE joins target and source on a source column: Redshift rejects a MERGE whose join key is a parameter or + * constant. {@code JSON_PARSE} lands the record in a SUPER column and the time fields are cast to TIMESTAMPTZ. + */ + private String buildMerge(String execNameSel, String startTimeSel, String endTimeSel, String durationSel) { + return "MERGE INTO " + fqTable + " USING (\n" + + " SELECT\n" + + " :execution_arn::varchar AS execution_arn,\n" + + " " + execNameSel + " AS execution_name,\n" + + " :function_name::varchar AS function_name,\n" + + " :status::varchar AS status,\n" + + " " + startTimeSel + " AS start_time,\n" + + " " + endTimeSel + " AS end_time,\n" + + " " + durationSel + " AS duration_ms,\n" + + " JSON_PARSE(:record_json) AS record_json,\n" + + " :emitted_at::timestamptz AS emitted_at\n" + + " ) AS src\n" + + " ON " + fqTable + ".execution_arn = src.execution_arn\n" + + " WHEN MATCHED THEN UPDATE SET\n" + + " status = src.status,\n" + + " end_time = src.end_time,\n" + + " duration_ms = src.duration_ms,\n" + + " record_json = src.record_json,\n" + + " emitted_at = src.emitted_at\n" + + " WHEN NOT MATCHED THEN INSERT\n" + + " (execution_arn, execution_name, function_name, status, start_time, end_time, duration_ms, record_json, emitted_at)\n" + + " VALUES\n" + + " (src.execution_arn, src.execution_name, src.function_name, src.status, src.start_time, src.end_time, src.duration_ms, src.record_json, src.emitted_at)"; + } + + /** Builder for {@link RedshiftExporter}. */ + public static final class Builder { + private String workgroupName; + private String clusterIdentifier; + private String database; + private String dbUser; + private String secretArn; + private String table; + private String schema; + private String region; + private Integer maxRecordSizeBytes; + private RedshiftDataClient client; + + /** Redshift Serverless workgroup name. One of workgroupName or clusterIdentifier is required. */ + public Builder workgroupName(String workgroupName) { + this.workgroupName = workgroupName; + return this; + } + + /** Provisioned cluster identifier. One of workgroupName or clusterIdentifier is required. */ + public Builder clusterIdentifier(String clusterIdentifier) { + this.clusterIdentifier = clusterIdentifier; + return this; + } + + public Builder database(String database) { + this.database = database; + return this; + } + + /** Database user for provisioned clusters using temporary credentials. */ + public Builder dbUser(String dbUser) { + this.dbUser = dbUser; + return this; + } + + /** Secrets Manager secret ARN; an alternative to dbUser for provisioned clusters. */ + public Builder secretArn(String secretArn) { + this.secretArn = secretArn; + return this; + } + + /** Table name; letters, digits, and underscores only. Default {@code workflow_insight}. */ + public Builder table(String table) { + this.table = table; + return this; + } + + /** Schema name; letters, digits, and underscores only. Default {@code public}. */ + public Builder schema(String schema) { + this.schema = schema; + 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(RedshiftDataClient client) { + this.client = client; + return this; + } + + public RedshiftExporter build() { + return new RedshiftExporter(this); + } + } +} 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 index c0fb6c037..47bda0eb3 100644 --- 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 @@ -3,9 +3,7 @@ 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.annotations.Experimental; import software.amazon.lambda.durable.insight.InsightExporter; @@ -32,18 +30,14 @@ public enum Partitioning { private final String prefix; private final Partitioning partitioning; private final Integer maxRecordSizeBytes; - private final S3Client client; + private final LazyClient 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(); + this.client = LazyClient.forSdkClient(b.client, "s3", "software.amazon.awssdk.services.s3.S3Client", b.region); } public static Builder builder() { @@ -59,13 +53,14 @@ public Integer maxRecordSizeBytes() { 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)); + client.get() + .putObject( + PutObjectRequest.builder() + .bucket(bucket) + .key(key) + .contentType("application/json") + .build(), + RequestBody.fromString(body)); } private String buildKey(WorkflowInsightRecord record) { diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/SQSExporter.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/SQSExporter.java new file mode 100644 index 000000000..e7f9ff5cc --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/SQSExporter.java @@ -0,0 +1,142 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight.exporters; + +import static java.util.Objects.requireNonNull; + +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.HexFormat; +import java.util.LinkedHashMap; +import java.util.Map; +import software.amazon.awssdk.services.sqs.SqsClient; +import software.amazon.awssdk.services.sqs.model.MessageAttributeValue; +import software.amazon.awssdk.services.sqs.model.SendMessageRequest; +import software.amazon.lambda.durable.annotations.Experimental; +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 SQS, one SendMessage per record with the record JSON as the body and + * {@code status} / {@code functionName} message attributes. On a FIFO queue the group id defaults to the execution ARN + * and the deduplication id is the execution ARN plus {@code emittedAt}. Requires {@code sqs:SendMessage}. + */ +@Experimental +public final class SQSExporter implements InsightExporter { + private final String queueUrl; + private final String messageGroupId; + private final boolean fifo; + private final OperationsFormat operationsFormat; + private final Integer maxRecordSizeBytes; + private final LazyClient client; + + private SQSExporter(Builder b) { + this.queueUrl = requireNonNull(b.queueUrl, "queueUrl"); + this.messageGroupId = b.messageGroupId; + this.fifo = queueUrl.endsWith(".fifo"); + this.operationsFormat = b.operationsFormat != null ? b.operationsFormat : OperationsFormat.ARRAY; + this.maxRecordSizeBytes = b.maxRecordSizeBytes != null ? b.maxRecordSizeBytes : 256_000; + this.client = + LazyClient.forSdkClient(b.client, "sqs", "software.amazon.awssdk.services.sqs.SqsClient", b.region); + } + + public static Builder builder() { + return new Builder(); + } + + @Override + public Integer maxRecordSizeBytes() { + return maxRecordSizeBytes; + } + + @Override + public Object render(WorkflowInsightRecord record) { + return operationsFormat.apply(record); + } + + @Override + public void export(WorkflowInsightRecord record) { + SqsClient sqs = client.get(); + Map body = operationsFormat.apply(record); + Map attributes = new LinkedHashMap<>(); + attributes.put("status", stringAttribute(record.status())); + attributes.put("functionName", stringAttribute(record.functionName())); + SendMessageRequest.Builder rb = SendMessageRequest.builder() + .queueUrl(queueUrl) + .messageBody(Json.stringify(body)) + .messageAttributes(attributes); + if (fifo) { + rb.messageGroupId(fifoId(messageGroupId != null ? messageGroupId : record.executionArn())) + .messageDeduplicationId(fifoId(record.executionArn() + ":" + body.get("emittedAt"))); + } + sqs.sendMessage(rb.build()); + } + + /** SQS caps FIFO group and deduplication ids at 128 characters; longer values are replaced by a SHA-256 digest. */ + static String fifoId(String value) { + if (value.length() <= 128) { + return value; + } + try { + byte[] digest = MessageDigest.getInstance("SHA-256").digest(value.getBytes(StandardCharsets.UTF_8)); + return HexFormat.of().formatHex(digest); + } catch (NoSuchAlgorithmException e) { + throw new IllegalStateException("SHA-256 unavailable", e); + } + } + + private static MessageAttributeValue stringAttribute(String value) { + return MessageAttributeValue.builder() + .dataType("String") + .stringValue(value) + .build(); + } + + /** Builder for {@link SQSExporter}. */ + public static final class Builder { + private String queueUrl; + private String messageGroupId; + private String region; + private OperationsFormat operationsFormat; + private Integer maxRecordSizeBytes; + private SqsClient client; + + public Builder queueUrl(String queueUrl) { + this.queueUrl = queueUrl; + return this; + } + + /** FIFO message group id. Default: the execution ARN. Ignored for standard queues. */ + public Builder messageGroupId(String messageGroupId) { + this.messageGroupId = messageGroupId; + return this; + } + + public Builder region(String region) { + this.region = region; + return this; + } + + public Builder operationsFormat(OperationsFormat operationsFormat) { + this.operationsFormat = operationsFormat; + return this; + } + + public Builder maxRecordSizeBytes(Integer maxRecordSizeBytes) { + this.maxRecordSizeBytes = maxRecordSizeBytes; + return this; + } + + /** Test seam: inject a client. */ + public Builder client(SqsClient client) { + this.client = client; + return this; + } + + public SQSExporter build() { + return new SQSExporter(this); + } + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/SqlIdentifiers.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/SqlIdentifiers.java new file mode 100644 index 000000000..515137c2a --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/SqlIdentifiers.java @@ -0,0 +1,21 @@ +// 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.util.regex.Pattern; + +/** Validates SQL identifiers (table and schema names) before they are spliced into statement text. */ +final class SqlIdentifiers { + private static final Pattern IDENTIFIER = Pattern.compile("^[a-zA-Z_][a-zA-Z0-9_]*$"); + + private SqlIdentifiers() {} + + /** Returns the name unchanged, or throws when it contains anything but letters, digits, and underscores. */ + static String validate(String name) { + if (name == null || !IDENTIFIER.matcher(name).matches()) { + throw new IllegalArgumentException( + "Invalid SQL identifier: \"" + name + "\". Only letters, digits, and underscores are allowed."); + } + return name; + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/AuroraExporterTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/AuroraExporterTest.java new file mode 100644 index 000000000..af1f5901e --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/AuroraExporterTest.java @@ -0,0 +1,128 @@ +// 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.assertThrows; +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.util.LinkedHashMap; +import java.util.Map; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import software.amazon.awssdk.services.rdsdata.RdsDataClient; +import software.amazon.awssdk.services.rdsdata.model.ExecuteStatementRequest; +import software.amazon.awssdk.services.rdsdata.model.ExecuteStatementResponse; +import software.amazon.awssdk.services.rdsdata.model.Field; +import software.amazon.awssdk.services.rdsdata.model.SqlParameter; +import software.amazon.lambda.durable.insight.exporters.AuroraExporter; + +class AuroraExporterTest { + + private static final String ARN = "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST"; + + private WorkflowInsightRecord sampleRecord() { + WorkflowInsightRecord r = new WorkflowInsightRecord(); + r.emittedAt = "2026-07-15T12:00:00.000Z"; + r.executionArn = ARN; + r.functionName = "fn"; + r.status = "RUNNING"; + r.startTime = "2026-07-15T11:59:00.000Z"; + r.addOperation( + new OperationRecord().id("op-1").name("fetch-user").type("STEP").status("SUCCEEDED")); + return r; + } + + private static ExecuteStatementRequest export(AuroraExporter.Builder builder, WorkflowInsightRecord record) { + RdsDataClient client = mock(RdsDataClient.class); + when(client.executeStatement(any(ExecuteStatementRequest.class))) + .thenReturn(ExecuteStatementResponse.builder().build()); + builder.client(client).build().export(record); + ArgumentCaptor req = ArgumentCaptor.forClass(ExecuteStatementRequest.class); + verify(client).executeStatement(req.capture()); + return req.getValue(); + } + + private static Map params(ExecuteStatementRequest req) { + Map out = new LinkedHashMap<>(); + for (SqlParameter p : req.parameters()) { + out.put(p.name(), p.value()); + } + return out; + } + + private static AuroraExporter.Builder base() { + return AuroraExporter.builder() + .resourceArn("arn:aws:rds:us-east-1:123456789012:cluster:c") + .secretArn("arn:aws:secretsmanager:us-east-1:123456789012:secret:s") + .database("insight"); + } + + @Test + void postgresUpsertBindsEveryColumnWithNullsForAbsentFields() { + WorkflowInsightRecord record = sampleRecord(); + ExecuteStatementRequest req = export(base().engine(AuroraExporter.Engine.POSTGRESQL), record); + + assertEquals("arn:aws:rds:us-east-1:123456789012:cluster:c", req.resourceArn()); + assertEquals("arn:aws:secretsmanager:us-east-1:123456789012:secret:s", req.secretArn()); + assertEquals("insight", req.database()); + assertTrue(req.sql().startsWith("INSERT INTO workflow_insight")); + assertTrue(req.sql().contains("ON CONFLICT (execution_arn) DO UPDATE SET")); + assertTrue(req.sql().contains(":record_json::jsonb")); + + Map p = params(req); + assertEquals(9, p.size()); + assertEquals(ARN, p.get("execution_arn").stringValue()); + assertTrue(p.get("execution_name").isNull()); + assertEquals("fn", p.get("function_name").stringValue()); + assertEquals("RUNNING", p.get("status").stringValue()); + assertEquals("2026-07-15T11:59:00.000Z", p.get("start_time").stringValue()); + assertTrue(p.get("end_time").isNull()); + assertTrue(p.get("duration_ms").isNull()); + assertEquals("2026-07-15T12:00:00.000Z", p.get("emitted_at").stringValue()); + assertEquals(Json.stringify(record.toWireMap()), p.get("record_json").stringValue()); + assertTrue(p.get("record_json").stringValue().contains("\"operations\":[")); + } + + @Test + void mysqlUpsertBindsCompletedFields() { + WorkflowInsightRecord record = sampleRecord(); + record.executionName = "exec-1"; + record.status = "SUCCEEDED"; + record.endTime = "2026-07-15T12:00:00.000Z"; + record.durationMs = 60_000L; + ExecuteStatementRequest req = + export(base().engine(AuroraExporter.Engine.MYSQL).table("insight_rows"), record); + + assertTrue(req.sql().startsWith("INSERT INTO insight_rows")); + assertTrue(req.sql().contains("ON DUPLICATE KEY UPDATE")); + assertFalse(req.sql().contains("::timestamptz")); + + Map p = params(req); + assertEquals("exec-1", p.get("execution_name").stringValue()); + assertEquals("2026-07-15T12:00:00.000Z", p.get("end_time").stringValue()); + assertEquals(60_000L, p.get("duration_ms").longValue()); + } + + @Test + void rejectsUnsafeTableNameAtBuildTime() { + assertThrows(IllegalArgumentException.class, () -> base().engine(AuroraExporter.Engine.MYSQL) + .table("t; DROP TABLE x") + .build()); + assertThrows(NullPointerException.class, () -> base().build(), "engine is required"); + assertEquals( + 1_000_000, base().engine(AuroraExporter.Engine.MYSQL).build().maxRecordSizeBytes()); + } + + @Test + void engineParsesConfigurationStrings() { + assertEquals(AuroraExporter.Engine.POSTGRESQL, AuroraExporter.Engine.fromValue("postgresql")); + assertEquals(AuroraExporter.Engine.MYSQL, AuroraExporter.Engine.fromValue("mysql")); + assertThrows(IllegalArgumentException.class, () -> AuroraExporter.Engine.fromValue("oracle")); + } +} 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 index b789230a5..2a3f2ab24 100644 --- 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 @@ -3,6 +3,7 @@ package software.amazon.lambda.durable.insight; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.atLeast; @@ -22,11 +23,14 @@ import java.util.concurrent.Future; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; +import software.amazon.awssdk.awscore.exception.AwsErrorDetails; 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.awssdk.services.cloudwatchlogs.model.ResourceAlreadyExistsException; +import software.amazon.awssdk.services.cloudwatchlogs.model.ResourceNotFoundException; import software.amazon.lambda.durable.insight.exporters.CloudWatchLogsExporter; class CloudWatchLogsExporterTest { @@ -73,6 +77,38 @@ void createsStreamOnceAndPutsOperationsByNameEvent() { assertTrue(!message.contains("\"operations\""), "CloudWatch must not emit the canonical array"); } + @Test + void toleratesAnExistingStreamButPropagatesOtherErrors() { + CloudWatchLogsClient client = mock(CloudWatchLogsClient.class); + when(client.createLogStream(any(CreateLogStreamRequest.class))) + .thenThrow(ResourceAlreadyExistsException.builder() + .awsErrorDetails(AwsErrorDetails.builder() + .errorCode("ResourceAlreadyExistsException") + .build()) + .build()) + .thenThrow(ResourceNotFoundException.builder() + .awsErrorDetails(AwsErrorDetails.builder() + .errorCode("ResourceNotFoundException") + .build()) + .build()); + when(client.putLogEvents(any(PutLogEventsRequest.class))) + .thenReturn(PutLogEventsResponse.builder().build()); + + CloudWatchLogsExporter existing = CloudWatchLogsExporter.builder() + .logGroupName("/my/group") + .client(client) + .build(); + existing.export(sampleRecord()); + verify(client, times(1)).putLogEvents(any(PutLogEventsRequest.class)); + + CloudWatchLogsExporter missingGroup = CloudWatchLogsExporter.builder() + .logGroupName("/missing/group") + .client(client) + .build(); + assertThrows(ResourceNotFoundException.class, () -> missingGroup.export(sampleRecord())); + verify(client, times(1)).putLogEvents(any(PutLogEventsRequest.class)); + } + @Test void concurrentExportsShareStreamCacheWithoutCollectionRace() throws Exception { CloudWatchLogsClient client = mock(CloudWatchLogsClient.class); diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/DynamoDBExporterTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/DynamoDBExporterTest.java new file mode 100644 index 000000000..54e3e47b6 --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/DynamoDBExporterTest.java @@ -0,0 +1,108 @@ +// 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.assertThrows; +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.util.Map; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import software.amazon.awssdk.services.dynamodb.DynamoDbClient; +import software.amazon.awssdk.services.dynamodb.model.AttributeValue; +import software.amazon.awssdk.services.dynamodb.model.PutItemRequest; +import software.amazon.awssdk.services.dynamodb.model.PutItemResponse; +import software.amazon.lambda.durable.insight.exporters.DynamoDBExporter; + +class DynamoDBExporterTest { + + private static final String ARN = "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST"; + + private WorkflowInsightRecord sampleRecord() { + WorkflowInsightRecord r = new WorkflowInsightRecord(); + r.emittedAt = "2026-07-15T12:00:00.000Z"; + r.executionArn = ARN; + r.executionName = "exec-1"; + r.functionName = "fn"; + r.status = "SUCCEEDED"; + r.startTime = "2026-07-15T11:59:00.000Z"; + r.durationMs = 60_000L; + r.input = Map.of("k", "v", "n", 2, "flag", true); + r.addOperation(new OperationRecord() + .id("op-1") + .name("fetch-user") + .type("STEP") + .subType("Step") + .status("SUCCEEDED") + .durationMs(5L) + .attempt(1)); + return r; + } + + private static PutItemRequest export(DynamoDBExporter.Builder builder) { + DynamoDbClient client = mock(DynamoDbClient.class); + when(client.putItem(any(PutItemRequest.class))) + .thenReturn(PutItemResponse.builder().build()); + builder.client(client).build().export(new DynamoDBExporterTest().sampleRecord()); + ArgumentCaptor req = ArgumentCaptor.forClass(PutItemRequest.class); + verify(client).putItem(req.capture()); + return req.getValue(); + } + + @Test + void writesHistoryItemKeyedByArnAndEmittedAt() { + PutItemRequest req = export(DynamoDBExporter.builder().tableName("insight")); + Map item = req.item(); + + assertEquals("insight", req.tableName()); + assertEquals(ARN, item.get("pk").s()); + assertEquals("2026-07-15T12:00:00.000Z", item.get("sk").s()); + assertEquals("SUCCEEDED", item.get("status").s()); + assertEquals("60000", item.get("durationMs").n()); + assertEquals("v", item.get("input").m().get("k").s()); + assertEquals("2", item.get("input").m().get("n").n()); + assertTrue(item.get("input").m().get("flag").bool()); + assertNull(item.get("operations"), "the by-name rendering replaces the operations array"); + Map byName = item.get("operationsByName").m(); + assertEquals("1", byName.get("fetch-user").m().get("count").n()); + assertEquals("STEP", byName.get("fetch-user").m().get("type").s()); + } + + @Test + void upsertsWithoutSortKeyAndHonorsCustomPartitionKey() { + PutItemRequest req = export(DynamoDBExporter.builder() + .tableName("insight") + .partitionKey("executionArnKey") + .sortKey("")); + Map item = req.item(); + + assertEquals(ARN, item.get("executionArnKey").s()); + assertFalse(item.containsKey("pk")); + assertFalse(item.containsKey("sk")); + } + + @Test + void renderIsTheByNameMapAndDefaultLimitIsFourHundredKilobytes() { + DynamoDBExporter exporter = DynamoDBExporter.builder() + .tableName("insight") + .client(mock(DynamoDbClient.class)) + .build(); + assertEquals(400_000, exporter.maxRecordSizeBytes()); + Map rendered = (Map) exporter.render(sampleRecord()); + assertTrue(rendered.containsKey("operationsByName")); + assertFalse(rendered.containsKey("operations")); + } + + @Test + void requiresTableName() { + assertThrows( + NullPointerException.class, () -> DynamoDBExporter.builder().build()); + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/EventBridgeExporterTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/EventBridgeExporterTest.java new file mode 100644 index 000000000..75d2efa48 --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/EventBridgeExporterTest.java @@ -0,0 +1,106 @@ +// 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.assertThrows; +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.time.Instant; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import software.amazon.awssdk.services.eventbridge.EventBridgeClient; +import software.amazon.awssdk.services.eventbridge.model.PutEventsRequest; +import software.amazon.awssdk.services.eventbridge.model.PutEventsRequestEntry; +import software.amazon.awssdk.services.eventbridge.model.PutEventsResponse; +import software.amazon.awssdk.services.eventbridge.model.PutEventsResultEntry; +import software.amazon.lambda.durable.insight.exporters.EventBridgeExporter; +import software.amazon.lambda.durable.insight.exporters.OperationsFormat; + +class EventBridgeExporterTest { + + private static final String ARN = "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST"; + + private WorkflowInsightRecord sampleRecord() { + WorkflowInsightRecord r = new WorkflowInsightRecord(); + r.emittedAt = "2026-07-15T12:00:00.000Z"; + r.executionArn = ARN; + r.functionName = "fn"; + r.status = "SUCCEEDED"; + r.startTime = "2026-07-15T11:59:00.000Z"; + r.addOperation( + new OperationRecord().id("op-1").name("fetch-user").type("STEP").status("SUCCEEDED")); + return r; + } + + private static EventBridgeClient clientReturning(PutEventsResponse response) { + EventBridgeClient client = mock(EventBridgeClient.class); + when(client.putEvents(any(PutEventsRequest.class))).thenReturn(response); + return client; + } + + private static PutEventsRequestEntry capture(EventBridgeClient client) { + ArgumentCaptor req = ArgumentCaptor.forClass(PutEventsRequest.class); + verify(client).putEvents(req.capture()); + assertEquals(1, req.getValue().entries().size()); + return req.getValue().entries().get(0); + } + + @Test + void publishesOneEventWithStatusDetailTypeAndRecordDetail() { + EventBridgeClient client = + clientReturning(PutEventsResponse.builder().failedEntryCount(0).build()); + WorkflowInsightRecord record = sampleRecord(); + EventBridgeExporter.builder().client(client).build().export(record); + + PutEventsRequestEntry entry = capture(client); + assertEquals("default", entry.eventBusName()); + assertEquals("software.amazon.lambda.durable.insight", entry.source()); + assertEquals(EventBridgeExporter.DEFAULT_SOURCE, entry.source()); + assertFalse(entry.source().startsWith("aws."), "the aws. source namespace is reserved for AWS services"); + assertEquals("SUCCEEDED", entry.detailType()); + assertEquals(Instant.parse("2026-07-15T12:00:00.000Z"), entry.time()); + assertEquals(Json.stringify(record.toWireMap()), entry.detail()); + assertTrue(entry.detail().contains("\"operations\":[")); + } + + @Test + void honorsCustomBusSourceAndByNameFormat() { + EventBridgeClient client = clientReturning(PutEventsResponse.builder().build()); + EventBridgeExporter.builder() + .eventBusName("insight-bus") + .source("my.source") + .operationsFormat(OperationsFormat.BY_NAME) + .client(client) + .build() + .export(sampleRecord()); + + PutEventsRequestEntry entry = capture(client); + assertEquals("insight-bus", entry.eventBusName()); + assertEquals("my.source", entry.source()); + assertTrue(entry.detail().contains("\"operationsByName\"")); + assertFalse(entry.detail().contains("\"operations\"")); + } + + @Test + void throwsWhenPutEventsReportsAFailedEntry() { + EventBridgeClient client = clientReturning(PutEventsResponse.builder() + .failedEntryCount(1) + .entries(PutEventsResultEntry.builder() + .errorCode("ThrottlingException") + .errorMessage("slow down") + .build()) + .build()); + EventBridgeExporter exporter = + EventBridgeExporter.builder().client(client).build(); + IllegalStateException e = assertThrows(IllegalStateException.class, () -> exporter.export(sampleRecord())); + assertTrue(e.getMessage().contains("ThrottlingException")); + assertTrue(e.getMessage().contains("slow down")); + assertEquals(256_000, exporter.maxRecordSizeBytes()); + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/FileExporterTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/FileExporterTest.java new file mode 100644 index 000000000..e1580b78d --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/FileExporterTest.java @@ -0,0 +1,141 @@ +// 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 com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import software.amazon.lambda.durable.insight.exporters.FileExporter; +import software.amazon.lambda.durable.insight.exporters.OperationsFormat; + +class FileExporterTest { + + private static final String ARN = "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST"; + private static final ObjectMapper MAPPER = new ObjectMapper(); + + @TempDir + Path tempDir; + + private WorkflowInsightRecord sampleRecord() { + WorkflowInsightRecord r = new WorkflowInsightRecord(); + r.emittedAt = "2026-07-15T12:00:00.000Z"; + r.executionArn = ARN; + r.functionName = "fn"; + r.status = "SUCCEEDED"; + r.startTime = "2026-07-15T11:59:00.000Z"; + r.addOperation( + new OperationRecord().id("op-1").name("fetch-user").type("STEP").status("SUCCEEDED")); + return r; + } + + @Test + void appendsDatePartitionedNdjsonLinesAndCreatesTheDirectory() throws Exception { + Path dir = tempDir.resolve("nested/insight"); + FileExporter exporter = FileExporter.builder().directory(dir.toString()).build(); + WorkflowInsightRecord first = sampleRecord(); + WorkflowInsightRecord second = sampleRecord(); + second.status = "FAILED"; + exporter.export(first); + exporter.export(second); + + Path file = dir.resolve("2026-07-15.ndjson"); + assertTrue(Files.exists(file)); + List lines = Files.readAllLines(file); + assertEquals(2, lines.size(), "append, not overwrite"); + assertEquals(Json.stringify(first.toWireMap()), lines.get(0)); + assertEquals(Json.stringify(second.toWireMap()), lines.get(1)); + assertTrue(Files.readString(file).endsWith("\n")); + assertTrue(MAPPER.readTree(lines.get(0)).get("operations").isArray()); + assertNull(exporter.maxRecordSizeBytes(), "no default size limit"); + } + + @Test + void jsonModeWritesOnePrettyFilePerExecutionAndOverwrites() throws Exception { + FileExporter exporter = FileExporter.builder() + .directory(tempDir.toString()) + .mode(FileExporter.Mode.JSON) + .operationsFormat(OperationsFormat.BY_NAME) + .build(); + WorkflowInsightRecord record = sampleRecord(); + record.executionName = "my exec/1"; + record.input = Map.of("empty", List.of()); + exporter.export(record); + record.status = "FAILED"; + exporter.export(record); + + Path file = tempDir.resolve("my_exec_1.json"); + String content = Files.readString(file); + assertTrue(content.startsWith("{\n \"recordType\": \"WorkflowInsight\",\n"), content); + assertTrue(content.contains("\"empty\": []"), content); + assertFalse(content.contains(" : "), "no space before the colon"); + JsonNode parsed = MAPPER.readTree(content); + assertEquals("FAILED", parsed.get("status").asText(), "second export overwrote the first"); + assertTrue(parsed.has("operationsByName")); + assertFalse(parsed.has("operations")); + assertEquals(1, Files.list(tempDir).count()); + } + + @Test + void concurrentLargeRecordsAppendWholeLines() throws Exception { + FileExporter exporter = + FileExporter.builder().directory(tempDir.toString()).build(); + int threads = 8; + int perThread = 5; + String payload = "x".repeat(40_000); // several append chunks per line + ExecutorService pool = Executors.newFixedThreadPool(threads); + CountDownLatch go = new CountDownLatch(1); + List> futures = new ArrayList<>(); + try { + for (int t = 0; t < threads; t++) { + futures.add(pool.submit(() -> { + go.await(); + for (int i = 0; i < perThread; i++) { + WorkflowInsightRecord r = sampleRecord(); + r.input = Map.of("payload", payload); + exporter.export(r); + } + return null; + })); + } + go.countDown(); + for (Future f : futures) { + f.get(); + } + } finally { + pool.shutdownNow(); + } + + List lines = Files.readAllLines(tempDir.resolve("2026-07-15.ndjson")); + assertEquals(threads * perThread, lines.size()); + for (String line : lines) { + assertEquals( + payload, MAPPER.readTree(line).get("input").get("payload").asText()); + } + } + + @Test + void jsonModeFallsBackToTheArnWhenThereIsNoExecutionName() { + FileExporter.builder() + .directory(tempDir.toString()) + .mode(FileExporter.Mode.JSON) + .build() + .export(sampleRecord()); + assertTrue(Files.exists(tempDir.resolve("arn_aws_lambda_us-east-1_123456789012_function_fn__LATEST.json"))); + assertEquals(FileExporter.Mode.JSON, FileExporter.Mode.fromValue("json")); + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/FirehoseExporterTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/FirehoseExporterTest.java new file mode 100644 index 000000000..48604fc0b --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/FirehoseExporterTest.java @@ -0,0 +1,73 @@ +// 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 static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +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.firehose.FirehoseClient; +import software.amazon.awssdk.services.firehose.model.PutRecordRequest; +import software.amazon.awssdk.services.firehose.model.PutRecordResponse; +import software.amazon.lambda.durable.insight.exporters.FirehoseExporter; +import software.amazon.lambda.durable.insight.exporters.OperationsFormat; + +class FirehoseExporterTest { + + private WorkflowInsightRecord sampleRecord() { + WorkflowInsightRecord r = new WorkflowInsightRecord(); + r.emittedAt = "2026-07-15T12:00:00.000Z"; + r.executionArn = "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST"; + r.functionName = "fn"; + r.status = "SUCCEEDED"; + r.startTime = "2026-07-15T11:59:00.000Z"; + r.addOperation( + new OperationRecord().id("op-1").name("fetch-user").type("STEP").status("SUCCEEDED")); + return r; + } + + private static PutRecordRequest export(FirehoseExporter.Builder builder, WorkflowInsightRecord record) { + FirehoseClient client = mock(FirehoseClient.class); + when(client.putRecord(any(PutRecordRequest.class))) + .thenReturn(PutRecordResponse.builder().build()); + builder.client(client).build().export(record); + ArgumentCaptor req = ArgumentCaptor.forClass(PutRecordRequest.class); + verify(client).putRecord(req.capture()); + return req.getValue(); + } + + @Test + void putsOneNewlineTerminatedJsonRecord() { + WorkflowInsightRecord record = sampleRecord(); + PutRecordRequest req = export(FirehoseExporter.builder().deliveryStreamName("insight-stream"), record); + + assertEquals("insight-stream", req.deliveryStreamName()); + String data = req.record().data().asUtf8String(); + assertEquals(Json.stringify(record.toWireMap()) + "\n", data); + assertTrue(data.contains("\"operations\":[")); + } + + @Test + void honorsByNameFormatAndDefaultLimit() { + FirehoseExporter.Builder builder = FirehoseExporter.builder() + .deliveryStreamName("insight-stream") + .operationsFormat(OperationsFormat.BY_NAME); + PutRecordRequest req = export(builder, sampleRecord()); + String data = req.record().data().asUtf8String(); + assertTrue(data.contains("\"operationsByName\"")); + assertFalse(data.contains("\"operations\"")); + assertEquals( + 1_000_000, + FirehoseExporter.builder() + .deliveryStreamName("s") + .client(mock(FirehoseClient.class)) + .build() + .maxRecordSizeBytes()); + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/HttpExporterTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/HttpExporterTest.java new file mode 100644 index 000000000..493d91e44 --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/HttpExporterTest.java @@ -0,0 +1,99 @@ +// 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.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.net.http.HttpTimeoutException; +import java.util.Map; +import org.junit.jupiter.api.Test; +import software.amazon.lambda.durable.insight.exporters.HttpExporter; +import software.amazon.lambda.durable.insight.exporters.OperationsFormat; + +class HttpExporterTest { + + private WorkflowInsightRecord sampleRecord() { + WorkflowInsightRecord r = new WorkflowInsightRecord(); + r.emittedAt = "2026-07-15T12:00:00.000Z"; + r.executionArn = "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST"; + r.functionName = "fn"; + r.status = "SUCCEEDED"; + r.startTime = "2026-07-15T11:59:00.000Z"; + r.addOperation( + new OperationRecord().id("op-1").name("fetch-user").type("STEP").status("SUCCEEDED")); + return r; + } + + @Test + void postsJsonWithContentTypeAndCustomHeaders() throws Exception { + try (LocalHttpServer server = new LocalHttpServer()) { + WorkflowInsightRecord record = sampleRecord(); + HttpExporter exporter = HttpExporter.builder() + .url(server.url("/hook")) + .headers(Map.of("Authorization", "Bearer t0k")) + .build(); + exporter.export(record); + + LocalHttpServer.Captured req = server.only(); + assertEquals("POST", req.method); + assertEquals("/hook", req.path); + assertEquals("application/json", req.headers.getFirst("Content-Type")); + assertEquals("Bearer t0k", req.headers.getFirst("Authorization")); + assertEquals(Json.stringify(record.toWireMap()), req.body); + assertNull(exporter.maxRecordSizeBytes(), "no default size limit"); + } + } + + @Test + void putsByNameBodyWhenConfigured() throws Exception { + try (LocalHttpServer server = new LocalHttpServer()) { + HttpExporter.builder() + .url(server.url("/records/1")) + .method(HttpExporter.Method.PUT) + .operationsFormat(OperationsFormat.BY_NAME) + .build() + .export(sampleRecord()); + + LocalHttpServer.Captured req = server.only(); + assertEquals("PUT", req.method); + assertTrue(req.body.contains("\"operationsByName\"")); + assertFalse(req.body.contains("\"operations\"")); + } + } + + @Test + void nonSuccessStatusThrows() throws Exception { + try (LocalHttpServer server = new LocalHttpServer()) { + server.status = 500; + HttpExporter exporter = + HttpExporter.builder().url(server.url("/hook")).build(); + IllegalStateException e = assertThrows(IllegalStateException.class, () -> exporter.export(sampleRecord())); + assertTrue(e.getMessage().contains("500")); + } + } + + @Test + void timesOutWhenTheEndpointIsSlow() throws Exception { + try (LocalHttpServer server = new LocalHttpServer()) { + server.delayMillis = 2_000; + HttpExporter exporter = HttpExporter.builder() + .url(server.url("/hook")) + .timeoutMs(200L) + .build(); + IllegalStateException e = assertThrows(IllegalStateException.class, () -> exporter.export(sampleRecord())); + assertInstanceOf(HttpTimeoutException.class, e.getCause()); + } + } + + @Test + void methodParsesConfigurationStrings() { + assertEquals(HttpExporter.Method.PUT, HttpExporter.Method.fromValue("PUT")); + assertThrows(IllegalArgumentException.class, () -> HttpExporter.Method.fromValue("PATCH")); + assertThrows(NullPointerException.class, () -> HttpExporter.builder().build()); + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/LocalHttpServer.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/LocalHttpServer.java new file mode 100644 index 000000000..dc554af72 --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/LocalHttpServer.java @@ -0,0 +1,75 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import com.sun.net.httpserver.Headers; +import com.sun.net.httpserver.HttpServer; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +/** Loopback HTTP server that records every request and answers with a configurable status. */ +final class LocalHttpServer implements AutoCloseable { + + /** One captured request. */ + static final class Captured { + final String method; + final String path; + final Headers headers; + final String body; + + Captured(String method, String path, Headers headers, String body) { + this.method = method; + this.path = path; + this.headers = headers; + this.body = body; + } + } + + private final HttpServer server; + final List requests = new CopyOnWriteArrayList<>(); + volatile int status = 200; + volatile long delayMillis = 0; + + LocalHttpServer() throws IOException { + server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/", exchange -> { + String body = new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8); + requests.add(new Captured( + exchange.getRequestMethod(), + exchange.getRequestURI().getRawPath(), + exchange.getRequestHeaders(), + body)); + if (delayMillis > 0) { + try { + Thread.sleep(delayMillis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + byte[] reply = "denied".getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(status, reply.length); + exchange.getResponseBody().write(reply); + exchange.close(); + }); + server.start(); + } + + String url(String path) { + return "http://127.0.0.1:" + server.getAddress().getPort() + path; + } + + Captured only() { + if (requests.size() != 1) { + throw new AssertionError("expected exactly one request, got " + requests.size()); + } + return requests.get(0); + } + + @Override + public void close() { + server.stop(0); + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/OTelExporterTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/OTelExporterTest.java new file mode 100644 index 000000000..25888d68e --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/OTelExporterTest.java @@ -0,0 +1,165 @@ +// 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.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.util.Map; +import org.junit.jupiter.api.Test; +import software.amazon.lambda.durable.insight.exporters.OTelExporter; +import software.amazon.lambda.durable.insight.exporters.OperationsFormat; + +class OTelExporterTest { + + private static final String ARN = "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST"; + private static final ObjectMapper MAPPER = new ObjectMapper(); + + private WorkflowInsightRecord sampleRecord() { + WorkflowInsightRecord r = new WorkflowInsightRecord(); + r.emittedAt = "2026-07-15T12:00:00.000Z"; + r.executionArn = ARN; + r.executionName = "exec-1"; + r.functionName = "fn"; + r.functionQualifier = "$LATEST"; + r.region = "us-east-1"; + r.accountId = "123456789012"; + r.status = "SUCCEEDED"; + r.startTime = "2026-07-15T11:59:00.000Z"; + r.durationMs = 60_000L; + r.addOperation( + new OperationRecord().id("op-1").name("fetch-user").type("STEP").status("SUCCEEDED")); + return r; + } + + private static JsonNode attr(JsonNode attributes, String key) { + for (JsonNode a : attributes) { + if (key.equals(a.get("key").asText())) { + return a.get("value"); + } + } + throw new AssertionError("missing attribute " + key); + } + + @Test + void postsOneOtlpLogRecordWithResourceAndLogAttributes() throws Exception { + try (LocalHttpServer server = new LocalHttpServer()) { + OTelExporter exporter = OTelExporter.builder() + .endpoint(server.url("/v1/logs")) + .headers(Map.of("x-api-key", "k1")) + .build(); + exporter.export(sampleRecord()); + + LocalHttpServer.Captured req = server.only(); + assertEquals("POST", req.method); + assertEquals("/v1/logs", req.path); + assertEquals("application/json", req.headers.getFirst("Content-Type")); + assertEquals("k1", req.headers.getFirst("x-api-key")); + + JsonNode payload = MAPPER.readTree(req.body); + JsonNode resourceLogs = payload.get("resourceLogs").get(0); + JsonNode resourceAttrs = resourceLogs.get("resource").get("attributes"); + assertEquals( + "fn", attr(resourceAttrs, "service.name").get("stringValue").asText()); + assertEquals( + "us-east-1", + attr(resourceAttrs, "cloud.region").get("stringValue").asText()); + assertEquals( + "123456789012", + attr(resourceAttrs, "cloud.account.id").get("stringValue").asText()); + assertEquals( + "$LATEST", + attr(resourceAttrs, "faas.version").get("stringValue").asText()); + + JsonNode scopeLogs = resourceLogs.get("scopeLogs").get(0); + assertEquals("1.0", scopeLogs.get("scope").get("version").asText()); + JsonNode log = scopeLogs.get("logRecords").get(0); + assertEquals("1784116800000000000", log.get("timeUnixNano").asText()); + assertEquals(9, log.get("severityNumber").asInt()); + assertEquals("SUCCEEDED", log.get("severityText").asText()); + JsonNode logAttrs = log.get("attributes"); + assertEquals( + ARN, + attr(logAttrs, "workflow.execution_arn").get("stringValue").asText()); + assertEquals( + "60000", + attr(logAttrs, "workflow.duration_ms").get("intValue").asText()); + + JsonNode body = MAPPER.readTree(log.get("body").get("stringValue").asText()); + assertEquals(ARN, body.get("executionArn").asText()); + assertTrue(body.get("operations").isArray()); + } + } + + @Test + void absentFieldsRenderEmptyValuesAndByNameBody() throws Exception { + try (LocalHttpServer server = new LocalHttpServer()) { + WorkflowInsightRecord record = sampleRecord(); + record.region = null; + record.executionName = null; + record.durationMs = null; + record.status = "FAILED"; + OTelExporter exporter = OTelExporter.builder() + .endpoint(server.url("/v1/logs")) + .operationsFormat(OperationsFormat.BY_NAME) + .build(); + exporter.export(record); + + JsonNode payload = MAPPER.readTree(server.only().body); + JsonNode resourceLogs = payload.get("resourceLogs").get(0); + assertEquals( + 0, + attr(resourceLogs.get("resource").get("attributes"), "cloud.region") + .size()); + JsonNode log = + resourceLogs.get("scopeLogs").get(0).get("logRecords").get(0); + assertEquals(17, log.get("severityNumber").asInt()); + assertEquals( + "", + attr(log.get("attributes"), "workflow.execution_name") + .get("stringValue") + .asText()); + assertEquals( + "0", + attr(log.get("attributes"), "workflow.duration_ms") + .get("intValue") + .asText()); + JsonNode body = MAPPER.readTree(log.get("body").get("stringValue").asText()); + assertTrue(body.has("operationsByName")); + assertTrue(!body.has("operations")); + } + } + + @Test + void renderMeasuresTheWholeOtlpEnvelope() { + OTelExporter exporter = + OTelExporter.builder().endpoint("http://127.0.0.1:1/v1/logs").build(); + assertEquals(1_000_000, exporter.maxRecordSizeBytes()); + Map rendered = (Map) exporter.render(sampleRecord()); + assertTrue(rendered.containsKey("resourceLogs")); + } + + @Test + void nonSuccessStatusThrows() throws Exception { + try (LocalHttpServer server = new LocalHttpServer()) { + server.status = 503; + OTelExporter exporter = + OTelExporter.builder().endpoint(server.url("/v1/logs")).build(); + IllegalStateException e = assertThrows(IllegalStateException.class, () -> exporter.export(sampleRecord())); + assertTrue(e.getMessage().contains("503")); + } + } + + @Test + void protobufIsRejectedAtBuildTime() { + assertThrows(IllegalArgumentException.class, () -> OTelExporter.builder() + .endpoint("http://127.0.0.1:1/v1/logs") + .protocol(OTelExporter.Protocol.HTTP_PROTOBUF) + .build()); + assertEquals(OTelExporter.Protocol.HTTP_JSON, OTelExporter.Protocol.fromValue("http/json")); + assertThrows(IllegalArgumentException.class, () -> OTelExporter.Protocol.fromValue("grpc")); + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/OpenSearchExporterTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/OpenSearchExporterTest.java new file mode 100644 index 000000000..59c08939c --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/OpenSearchExporterTest.java @@ -0,0 +1,115 @@ +// 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.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.charset.StandardCharsets; +import java.util.Base64; +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; +import software.amazon.lambda.durable.insight.exporters.OpenSearchExporter; + +class OpenSearchExporterTest { + + private static final String ARN = "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST"; + private static final String ENCODED_ARN = + "arn%3Aaws%3Alambda%3Aus-east-1%3A123456789012%3Afunction%3Afn%3A%24LATEST"; + + private WorkflowInsightRecord sampleRecord() { + WorkflowInsightRecord r = new WorkflowInsightRecord(); + r.emittedAt = "2026-07-15T12:00:00.000Z"; + r.executionArn = ARN; + r.functionName = "fn"; + r.status = "SUCCEEDED"; + r.startTime = "2026-07-15T11:59:00.000Z"; + r.addOperation( + new OperationRecord().id("op-1").name("fetch-user").type("STEP").status("SUCCEEDED")); + return r; + } + + @Test + void putsSignedDocumentKeyedByEncodedArn() throws Exception { + try (LocalHttpServer server = new LocalHttpServer()) { + WorkflowInsightRecord record = sampleRecord(); + OpenSearchExporter.builder() + .endpoint(server.url("/")) + .region("us-east-1") + .credentialsProvider(StaticCredentialsProvider.create(AwsBasicCredentials.create("AKID", "secret"))) + .build() + .export(record); + + LocalHttpServer.Captured req = server.only(); + assertEquals("PUT", req.method); + assertEquals("/workflow-insight/_doc/" + ENCODED_ARN, req.path); + assertEquals("application/json", req.headers.getFirst("Content-Type")); + String authorization = req.headers.getFirst("Authorization"); + assertNotNull(authorization); + assertTrue(authorization.startsWith("AWS4-HMAC-SHA256 Credential=AKID/"), authorization); + assertTrue(authorization.contains("/us-east-1/es/aws4_request"), authorization); + assertNotNull(req.headers.getFirst("X-Amz-Date")); + assertEquals(Json.stringify(record.toWireMap()), req.body); + } + } + + @Test + void usesBasicAuthWithoutSigning() throws Exception { + try (LocalHttpServer server = new LocalHttpServer()) { + OpenSearchExporter.builder() + .endpoint(server.url("")) + .auth(OpenSearchExporter.Auth.BASIC) + .username("admin") + .password("secret") + .indexName("custom-index") + .build() + .export(sampleRecord()); + + LocalHttpServer.Captured req = server.only(); + assertEquals("/custom-index/_doc/" + ENCODED_ARN, req.path); + String expected = + "Basic " + Base64.getEncoder().encodeToString("admin:secret".getBytes(StandardCharsets.UTF_8)); + assertEquals(expected, req.headers.getFirst("Authorization")); + assertEquals(null, req.headers.getFirst("X-Amz-Date")); + } + } + + @Test + void throwsWithStatusAndDetailOnFailure() throws Exception { + try (LocalHttpServer server = new LocalHttpServer()) { + server.status = 403; + OpenSearchExporter exporter = OpenSearchExporter.builder() + .endpoint(server.url("")) + .auth(OpenSearchExporter.Auth.BASIC) + .username("u") + .password("p") + .build(); + IllegalStateException e = assertThrows(IllegalStateException.class, () -> exporter.export(sampleRecord())); + assertTrue(e.getMessage().contains("403"), e.getMessage()); + assertTrue(e.getMessage().contains("denied"), e.getMessage()); + assertEquals(10_000_000, exporter.maxRecordSizeBytes()); + } + } + + @Test + void sigv4RequiresRegionAtBuildTime() { + assertThrows(NullPointerException.class, () -> OpenSearchExporter.builder() + .endpoint("https://d.us-east-1.es.amazonaws.com") + .build()); + assertThrows(NullPointerException.class, () -> OpenSearchExporter.builder() + .endpoint("https://d.us-east-1.es.amazonaws.com") + .auth(OpenSearchExporter.Auth.BASIC) + .password("p") + .build()); + assertThrows(NullPointerException.class, () -> OpenSearchExporter.builder() + .endpoint("https://d.us-east-1.es.amazonaws.com") + .auth(OpenSearchExporter.Auth.BASIC) + .username("u") + .build()); + assertEquals(OpenSearchExporter.Auth.BASIC, OpenSearchExporter.Auth.fromValue("basic")); + assertThrows(IllegalArgumentException.class, () -> OpenSearchExporter.Auth.fromValue("oauth")); + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/OperationsFormatTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/OperationsFormatTest.java new file mode 100644 index 000000000..58ea53631 --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/OperationsFormatTest.java @@ -0,0 +1,63 @@ +// 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.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; +import software.amazon.lambda.durable.insight.exporters.OperationsFormat; + +class OperationsFormatTest { + + private WorkflowInsightRecord sampleRecord() { + WorkflowInsightRecord r = new WorkflowInsightRecord(); + r.executionArn = "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST"; + r.status = "SUCCEEDED"; + r.truncated = true; + r.addOperation( + new OperationRecord().id("op-1").name("fetch-user").type("STEP").status("SUCCEEDED")); + r.addOperation( + new OperationRecord().id("op-2").name("fetch-user").type("STEP").status("FAILED")); + return r; + } + + @Test + void arrayKeepsTheCanonicalOperations() { + Map data = OperationsFormat.ARRAY.apply(sampleRecord()); + assertEquals(2, ((List) data.get("operations")).size()); + assertFalse(data.containsKey("operationsByName")); + assertEquals(true, data.get("truncated")); + } + + @Test + void byNameReplacesTheArrayWithTheSummaryMap() { + Map data = OperationsFormat.BY_NAME.apply(sampleRecord()); + assertFalse(data.containsKey("operations")); + Map summary = (Map) ((Map) data.get("operationsByName")).get("fetch-user"); + assertEquals(2, summary.get("count")); + assertEquals(1, summary.get("failedCount")); + } + + @Test + void bothCarriesTheArrayAndTheMap() { + Map data = OperationsFormat.BOTH.apply(sampleRecord()); + assertEquals(2, ((List) data.get("operations")).size()); + assertTrue(data.containsKey("operationsByName")); + List keys = List.copyOf(data.keySet()); + assertEquals("operationsByName", keys.get(keys.size() - 1), "appended after the canonical fields"); + } + + @Test + void parsesConfigurationStrings() { + assertEquals(OperationsFormat.ARRAY, OperationsFormat.fromValue("array")); + assertEquals(OperationsFormat.BY_NAME, OperationsFormat.fromValue("by-name")); + assertEquals(OperationsFormat.BOTH, OperationsFormat.fromValue("both")); + assertEquals("by-name", OperationsFormat.BY_NAME.value()); + assertThrows(IllegalArgumentException.class, () -> OperationsFormat.fromValue("map")); + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/RecordFactory.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/RecordFactory.java new file mode 100644 index 000000000..4ffb31d31 --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/RecordFactory.java @@ -0,0 +1,21 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +/** Builds a minimal, fully populated record for exporter tests outside this package. */ +public final class RecordFactory { + private RecordFactory() {} + + public static WorkflowInsightRecord sample() { + WorkflowInsightRecord r = new WorkflowInsightRecord(); + r.emittedAt = "2026-07-15T12:00:00.000Z"; + r.executionArn = "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST"; + r.executionName = "exec-1"; + r.functionName = "fn"; + r.status = "SUCCEEDED"; + r.startTime = "2026-07-15T11:59:00.000Z"; + r.addOperation( + new OperationRecord().id("op-1").name("fetch-user").type("STEP").status("SUCCEEDED")); + return r; + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/RedshiftExporterTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/RedshiftExporterTest.java new file mode 100644 index 000000000..91d6a8fc5 --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/RedshiftExporterTest.java @@ -0,0 +1,177 @@ +// 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.assertThrows; +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.util.LinkedHashMap; +import java.util.Map; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import software.amazon.awssdk.services.redshiftdata.RedshiftDataClient; +import software.amazon.awssdk.services.redshiftdata.model.ExecuteStatementRequest; +import software.amazon.awssdk.services.redshiftdata.model.ExecuteStatementResponse; +import software.amazon.awssdk.services.redshiftdata.model.SqlParameter; +import software.amazon.lambda.durable.insight.exporters.RedshiftExporter; + +class RedshiftExporterTest { + + private static final String ARN = "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST"; + + private WorkflowInsightRecord sampleRecord() { + WorkflowInsightRecord r = new WorkflowInsightRecord(); + r.emittedAt = "2026-07-15T12:00:00.000Z"; + r.executionArn = ARN; + r.functionName = "fn"; + r.status = "RUNNING"; + r.startTime = "2026-07-15T11:59:00.000Z"; + r.addOperation( + new OperationRecord().id("op-1").name("fetch-user").type("STEP").status("SUCCEEDED")); + return r; + } + + private static ExecuteStatementRequest export(RedshiftExporter.Builder builder, WorkflowInsightRecord record) { + RedshiftDataClient client = mock(RedshiftDataClient.class); + when(client.executeStatement(any(ExecuteStatementRequest.class))) + .thenReturn(ExecuteStatementResponse.builder().build()); + builder.client(client).build().export(record); + ArgumentCaptor req = ArgumentCaptor.forClass(ExecuteStatementRequest.class); + verify(client).executeStatement(req.capture()); + return req.getValue(); + } + + private static Map params(ExecuteStatementRequest req) { + Map out = new LinkedHashMap<>(); + for (SqlParameter p : req.parameters()) { + out.put(p.name(), p.value()); + } + return out; + } + + @Test + void serverlessMergeUsesTypedNullLiteralsForAbsentFields() { + WorkflowInsightRecord record = sampleRecord(); + ExecuteStatementRequest req = + export(RedshiftExporter.builder().workgroupName("wg").database("insight"), record); + + assertEquals("wg", req.workgroupName()); + assertNull(req.clusterIdentifier()); + assertEquals("insight", req.database()); + assertNull(req.dbUser()); + assertNull(req.secretArn()); + assertTrue(req.sql().startsWith("MERGE INTO public.workflow_insight USING (")); + assertTrue(req.sql().contains("NULL::varchar AS execution_name")); + assertTrue(req.sql().contains("NULL::timestamptz AS end_time")); + assertTrue(req.sql().contains("NULL::bigint AS duration_ms")); + assertTrue(req.sql().contains("JSON_PARSE(:record_json) AS record_json")); + assertTrue(req.sql().contains("ON public.workflow_insight.execution_arn = src.execution_arn")); + + Map p = params(req); + assertEquals(6, p.size()); + assertEquals(ARN, p.get("execution_arn")); + assertEquals("fn", p.get("function_name")); + assertEquals("RUNNING", p.get("status")); + assertEquals("2026-07-15T11:59:00.000Z", p.get("start_time")); + assertEquals("2026-07-15T12:00:00.000Z", p.get("emitted_at")); + assertEquals(Json.stringify(record.toWireMap()), p.get("record_json")); + } + + @Test + void provisionedMergeBindsCompletedFieldsAndCustomSchemaTable() { + WorkflowInsightRecord record = sampleRecord(); + record.executionName = "exec-1"; + record.status = "SUCCEEDED"; + record.endTime = "2026-07-15T12:00:00.000Z"; + record.durationMs = 60_000L; + ExecuteStatementRequest req = export( + RedshiftExporter.builder() + .clusterIdentifier("cluster-1") + .database("insight") + .secretArn("arn:aws:secretsmanager:us-east-1:123456789012:secret:s") + .schema("analytics") + .table("wf"), + record); + + assertEquals("cluster-1", req.clusterIdentifier()); + assertNull(req.dbUser()); + assertEquals("arn:aws:secretsmanager:us-east-1:123456789012:secret:s", req.secretArn()); + assertTrue(req.sql().startsWith("MERGE INTO analytics.wf USING (")); + assertTrue(req.sql().contains(":execution_name::varchar AS execution_name")); + assertTrue(req.sql().contains(":end_time::timestamptz AS end_time")); + assertTrue(req.sql().contains(":duration_ms::bigint AS duration_ms")); + assertFalse(req.sql().contains("NULL::")); + + Map p = params(req); + assertEquals(9, p.size()); + assertEquals("2026-07-15T12:00:00.000Z", p.get("end_time")); + assertEquals("60000", p.get("duration_ms")); + assertEquals("exec-1", p.get("execution_name")); + } + + @Test + void omitsStartTimeParameterWhenTheRecordHasNone() { + WorkflowInsightRecord record = sampleRecord(); + record.startTime = null; + ExecuteStatementRequest req = + export(RedshiftExporter.builder().workgroupName("wg").database("insight"), record); + + assertTrue(req.sql().contains("NULL::timestamptz AS start_time")); + assertFalse(req.sql().contains(":start_time")); + assertFalse(params(req).containsKey("start_time")); + assertEquals(5, req.parameters().size()); + } + + @Test + void validatesTargetAndIdentifiersAtBuildTime() { + assertThrows( + IllegalArgumentException.class, + () -> RedshiftExporter.builder().database("insight").build(), + "workgroupName or clusterIdentifier is required"); + assertThrows( + IllegalArgumentException.class, + () -> RedshiftExporter.builder() + .workgroupName("wg") + .clusterIdentifier("cluster-1") + .database("insight") + .build(), + "workgroupName and clusterIdentifier are exclusive"); + assertThrows( + IllegalArgumentException.class, + () -> RedshiftExporter.builder() + .clusterIdentifier("cluster-1") + .database("insight") + .dbUser("admin") + .secretArn("arn:aws:secretsmanager:us-east-1:123456789012:secret:s") + .build(), + "dbUser and secretArn are exclusive"); + assertEquals( + "admin", + export( + RedshiftExporter.builder() + .clusterIdentifier("cluster-1") + .database("insight") + .dbUser("admin"), + sampleRecord()) + .dbUser()); + assertThrows(IllegalArgumentException.class, () -> RedshiftExporter.builder() + .workgroupName("wg") + .database("insight") + .schema("public; DROP") + .build()); + assertEquals( + 1_000_000, + RedshiftExporter.builder() + .workgroupName("wg") + .database("insight") + .build() + .maxRecordSizeBytes()); + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/SQSExporterTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/SQSExporterTest.java new file mode 100644 index 000000000..cc0579319 --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/SQSExporterTest.java @@ -0,0 +1,114 @@ +// 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 static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +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.sqs.SqsClient; +import software.amazon.awssdk.services.sqs.model.SendMessageRequest; +import software.amazon.awssdk.services.sqs.model.SendMessageResponse; +import software.amazon.lambda.durable.insight.exporters.OperationsFormat; +import software.amazon.lambda.durable.insight.exporters.SQSExporter; + +class SQSExporterTest { + + private static final String ARN = "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST"; + private static final String STANDARD = "https://sqs.us-east-1.amazonaws.com/123456789012/insight"; + private static final String FIFO = STANDARD + ".fifo"; + + private WorkflowInsightRecord sampleRecord() { + WorkflowInsightRecord r = new WorkflowInsightRecord(); + r.emittedAt = "2026-07-15T12:00:00.000Z"; + r.executionArn = ARN; + r.functionName = "fn"; + r.status = "SUCCEEDED"; + r.startTime = "2026-07-15T11:59:00.000Z"; + r.addOperation( + new OperationRecord().id("op-1").name("fetch-user").type("STEP").status("SUCCEEDED")); + return r; + } + + private static SendMessageRequest export(SQSExporter.Builder builder, WorkflowInsightRecord record) { + SqsClient client = mock(SqsClient.class); + when(client.sendMessage(any(SendMessageRequest.class))) + .thenReturn(SendMessageResponse.builder().build()); + builder.client(client).build().export(record); + ArgumentCaptor req = ArgumentCaptor.forClass(SendMessageRequest.class); + verify(client).sendMessage(req.capture()); + return req.getValue(); + } + + @Test + void sendsStandardQueueMessageWithAttributesAndNoFifoFields() { + WorkflowInsightRecord record = sampleRecord(); + SendMessageRequest req = export(SQSExporter.builder().queueUrl(STANDARD), record); + + assertEquals(STANDARD, req.queueUrl()); + assertNull(req.messageGroupId()); + assertNull(req.messageDeduplicationId()); + assertEquals("String", req.messageAttributes().get("status").dataType()); + assertEquals("SUCCEEDED", req.messageAttributes().get("status").stringValue()); + assertEquals("fn", req.messageAttributes().get("functionName").stringValue()); + assertEquals(Json.stringify(record.toWireMap()), req.messageBody()); + } + + @Test + void setsFifoGroupAndDeduplicationIds() { + SendMessageRequest req = export(SQSExporter.builder().queueUrl(FIFO), sampleRecord()); + assertEquals(ARN, req.messageGroupId()); + assertEquals(ARN + ":2026-07-15T12:00:00.000Z", req.messageDeduplicationId()); + } + + @Test + void boundsFifoIdsToOneHundredTwentyEightCharacters() { + WorkflowInsightRecord record = sampleRecord(); + record.executionArn = "arn:aws:lambda:us-east-1:123456789012:function:" + "f".repeat(80) + + ":$LATEST/durable-execution/" + "e".repeat(40) + "/1"; + SendMessageRequest req = export(SQSExporter.builder().queueUrl(FIFO), record); + + assertEquals(64, req.messageGroupId().length(), "digest replaces an over-long default group id"); + assertEquals(64, req.messageDeduplicationId().length()); + assertTrue(req.messageDeduplicationId().matches("[0-9a-f]{64}")); + assertEquals( + req.messageDeduplicationId(), + export(SQSExporter.builder().queueUrl(FIFO), record).messageDeduplicationId(), + "same record yields the same deduplication id"); + + String longGroup = "g".repeat(129); + SendMessageRequest custom = export(SQSExporter.builder().queueUrl(FIFO).messageGroupId(longGroup), record); + assertEquals(64, custom.messageGroupId().length(), "explicit group ids are bounded too"); + assertEquals( + "g".repeat(128), + export(SQSExporter.builder().queueUrl(FIFO).messageGroupId("g".repeat(128)), record) + .messageGroupId()); + } + + @Test + void honorsExplicitGroupIdAndByNameFormat() { + SendMessageRequest req = export( + SQSExporter.builder() + .queueUrl(FIFO) + .messageGroupId("custom-group") + .operationsFormat(OperationsFormat.BY_NAME), + sampleRecord()); + assertEquals("custom-group", req.messageGroupId()); + assertTrue(req.messageBody().contains("\"operationsByName\"")); + assertFalse(req.messageBody().contains("\"operations\"")); + assertEquals( + 256_000, + SQSExporter.builder() + .queueUrl(STANDARD) + .client(mock(SqsClient.class)) + .build() + .maxRecordSizeBytes()); + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/exporters/LazyClientTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/exporters/LazyClientTest.java new file mode 100644 index 000000000..bf79cadb7 --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/exporters/LazyClientTest.java @@ -0,0 +1,46 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight.exporters; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; + +class LazyClientTest { + + @Test + void returnsTheInjectedInstanceWithoutCallingTheFactory() { + Object injected = new Object(); + LazyClient client = new LazyClient<>(injected, "sqs", () -> { + throw new AssertionError("factory must not run"); + }); + assertSame(injected, client.get()); + } + + @Test + void createsOnceOnFirstUse() { + AtomicInteger calls = new AtomicInteger(); + LazyClient client = new LazyClient<>(null, "sqs", () -> { + calls.incrementAndGet(); + return new Object(); + }); + assertEquals(0, calls.get(), "nothing created at build time"); + assertSame(client.get(), client.get()); + assertEquals(1, calls.get()); + } + + @Test + void missingArtifactFailsWithAMessageNamingIt() { + LazyClient client = new LazyClient<>(null, "redshiftdata", () -> { + throw new NoClassDefFoundError("software/amazon/awssdk/services/redshiftdata/RedshiftDataClient"); + }); + IllegalStateException e = assertThrows(IllegalStateException.class, client::get); + assertTrue(e.getMessage().contains("software.amazon.awssdk:redshiftdata"), e.getMessage()); + assertInstanceOf(NoClassDefFoundError.class, e.getCause()); + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/exporters/OptionalArtifactTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/exporters/OptionalArtifactTest.java new file mode 100644 index 000000000..a5b518c42 --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/exporters/OptionalArtifactTest.java @@ -0,0 +1,150 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight.exporters; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.File; +import java.net.URL; +import java.net.URLClassLoader; +import java.nio.file.Path; +import java.util.function.Supplier; +import java.util.stream.Stream; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import software.amazon.lambda.durable.insight.InsightExporter; +import software.amazon.lambda.durable.insight.RecordFactory; + +/** + * Optional AWS SDK artifacts must not be needed to build an exporter: a configured exporter whose artifact is absent + * has to fail at export, inside the plugin's per-exporter isolation, with a message naming the artifact. + */ +class OptionalArtifactTest { + + /** Builds one exporter and exports one record; loaded into the isolated class loader so it links there. */ + public static final class Scenario implements Supplier { + private final String exporter; + + public Scenario(String exporter) { + this.exporter = exporter; + } + + /** Returns the export failure message, or throws if build or export did not behave as required. */ + @Override + public String get() { + InsightExporter built = build(); + try { + built.export(RecordFactory.sample()); + } catch (IllegalStateException e) { + return e.getMessage(); + } + throw new AssertionError(exporter + " exported without its artifact"); + } + + private InsightExporter build() { + switch (exporter) { + case "DynamoDBExporter": + return DynamoDBExporter.builder().tableName("t").build(); + case "FirehoseExporter": + return FirehoseExporter.builder().deliveryStreamName("s").build(); + case "EventBridgeExporter": + return EventBridgeExporter.builder().build(); + case "SQSExporter": + return SQSExporter.builder() + .queueUrl("https://sqs.us-east-1.amazonaws.com/1/q") + .build(); + case "RedshiftExporter": + return RedshiftExporter.builder() + .workgroupName("wg") + .database("d") + .build(); + case "AuroraExporter": + return AuroraExporter.builder() + .resourceArn("r") + .secretArn("s") + .database("d") + .engine(AuroraExporter.Engine.MYSQL) + .build(); + case "S3Exporter": + return S3Exporter.builder().bucket("b").build(); + case "CloudWatchLogsExporter": + return CloudWatchLogsExporter.builder().logGroupName("g").build(); + default: + throw new IllegalArgumentException(exporter); + } + } + } + + static Stream exporters() { + return Stream.of( + Arguments.of("DynamoDBExporter", "dynamodb", "DynamoDbClient"), + Arguments.of("FirehoseExporter", "firehose", "FirehoseClient"), + Arguments.of("EventBridgeExporter", "eventbridge", "EventBridgeClient"), + Arguments.of("SQSExporter", "sqs", "SqsClient"), + Arguments.of("RedshiftExporter", "redshiftdata", "RedshiftDataClient"), + Arguments.of("AuroraExporter", "rdsdata", "RdsDataClient"), + Arguments.of("S3Exporter", "s3", "S3Client"), + Arguments.of("CloudWatchLogsExporter", "cloudwatchlogs", "CloudWatchLogsClient")); + } + + @ParameterizedTest(name = "{0} builds without {1} and fails at export naming it") + @MethodSource("exporters") + @SuppressWarnings("unchecked") + void buildsWithoutTheArtifactAndFailsAtExport(String exporter, String artifact, String clientSimpleName) + throws Exception { + try (URLClassLoader loader = loaderWithout(artifact)) { + String client = "software.amazon.awssdk.services." + artifact + "." + clientSimpleName; + assertThrows( + ClassNotFoundException.class, + () -> Class.forName(client, false, loader), + "the " + artifact + " client must be hidden from this loader"); + + Supplier scenario = (Supplier) loader.loadClass(Scenario.class.getName()) + .getConstructor(String.class) + .newInstance(exporter); + String message = scenario.get(); + assertTrue(message.contains("software.amazon.awssdk:" + artifact), message); + } + } + + /** + * A child-first loader over the test classpath with one service artifact removed, so the exporter classes are + * defined by this loader (not the parent) and cannot see that service's client. + */ + private static URLClassLoader loaderWithout(String artifact) throws Exception { + String prefix = File.separator + artifact + "-"; + String classPath = System.getProperty("surefire.test.class.path", System.getProperty("java.class.path")); + URL[] urls = Stream.of(classPath.split(File.pathSeparator)) + .filter(p -> !p.contains(prefix)) + .map(p -> { + try { + return Path.of(p).toUri().toURL(); + } catch (Exception e) { + throw new IllegalStateException(e); + } + }) + .toArray(URL[]::new); + assertEquals(classPath.split(File.pathSeparator).length - 1, urls.length, "exactly one artifact jar removed"); + return new URLClassLoader(urls, null) { + @Override + protected Class loadClass(String name, boolean resolve) throws ClassNotFoundException { + if (name.startsWith("java.") || name.startsWith("javax.") || name.startsWith("jdk.")) { + return super.loadClass(name, resolve); + } + synchronized (getClassLoadingLock(name)) { + Class c = findLoadedClass(name); + if (c == null) { + c = findClass(name); + } + if (resolve) { + resolveClass(c); + } + return c; + } + } + }; + } +}