diff --git a/telemetry-exporter/README.md b/telemetry-exporter/README.md index 8f162f522..9f92527d2 100644 --- a/telemetry-exporter/README.md +++ b/telemetry-exporter/README.md @@ -29,7 +29,7 @@ Implement `TelemetryExporter` for custom sinks. Every exporter has a builder wit | Exporter | Destination | Operations rendering | Default size limit | Artifact (optional) | |---|---|---|---|---| -| `LambdaLogExporter` | Function log group (stdout) | `operationsByName` | 256 KB | none | +| `LambdaLogExporter` | Function log group (stdout) | `operationsByName` | 256,000 bytes (the printed line) | 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` | @@ -57,6 +57,51 @@ at first export with a message naming the artifact; the plugin logs it and conti ``` +### LambdaLogExporter + +Writes one line per record to `System.out`, in the Lambda JSON log envelope: + +```json +{"timestamp":"2026-10-21T08:00:00.000Z","level":"INFO","message":""} +``` + +`timestamp` is UTC ISO-8601 with exactly three fraction digits. The keys appear in that order. The +[Lambda Java logging docs](https://docs.aws.amazon.com/lambda/latest/dg/java-logging.html) and the +[Lambda log format docs](https://docs.aws.amazon.com/lambda/latest/dg/monitoring-cloudwatchlogs-logformat.html) +list `LambdaLogger` and Log4j2 as the Java methods that Lambda captures as structured JSON, so `System.out` is not +wrapped and the exporter writes the envelope itself. Lambda does not double-encode logs that are already JSON, so the +line reaches CloudWatch Logs as written. + +The size limit bounds the printed line, envelope included: the exporter's `render` returns the envelope, so the one +truncation pass measures exactly the line that is printed. The default limit is 256,000 bytes; CloudWatch Logs accepts +log events of up to 1 MB. `message` is the record JSON escaped again, so a record with many `"` or `\` characters can +grow to about twice its size. The exporter computes the line's size without building it (the envelope's fixed 68 bytes +plus the record JSON's size once escaped again), so the size checks stay cheap on large records. Truncation drops +error causes first, then operation results, whole operations, input and output (see +[Error names and causes](#error-names-and-causes) and [Design](#design)). + +The execution `error` and the identity fields are never dropped, so they set the smallest line truncation can reach. +Every error `name` and `message` is capped at 8,192 code points. On the line, one code point takes at most 7 bytes (a +control character is written as `\u00XX` in the record and its backslash is escaped again), so a capped name or message +takes at most about 57 KB, and about 33 KB when it is all quotes, backslashes or emoji. With identity fields of the +lengths Lambda allows (each a few hundred characters at most): + +- the line fits any limit of 128 KB or more, including the default, whatever the error's name and message; +- the line fits 64 KB when the execution error's name is under 500 characters, whatever its message. + +An execution error whose name and message are both 8,192 escape-heavy characters makes a line of about 66 KB to 115 KB. +With a limit smaller than the minimal line (for example a limit below the 68-byte envelope) the line is still printed, +over the limit, with `truncated` and the `dropped*` markers set. + +Lines have an explicit `"level":"INFO"`. Lambda already treats a line without a level as INFO, so log level filtering +is unchanged: with the JSON log format and an application log level of `WARN` or above, Lambda drops these lines, as +it dropped the raw record lines before. What changed for readers is the line itself: the record is the JSON string in +`message`, so queries and metric filters must parse `message` instead of matching top-level record fields. + +`LambdaLogExporter.render` now returns the envelope map (`timestamp`, `level`, `message`) instead of the record map; +`message` is the record JSON string. Code that called `render` to get the `operationsByName` record should call +`DurableExecutionTelemetryRecord.toByNameWireMap()` instead. + ### DynamoDBExporter Table keyed by `pk` (string), optionally with sort key `sk`. IAM: `dynamodb:PutItem`. @@ -157,6 +202,60 @@ FileExporter.builder().directory("/mnt/efs/durable-execution-telemetry").mode(Fi `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`). +## Error names and causes + +Every error in a record (the execution `error`, each `operations[].error`, and each +`operationsByName..error`) has this shape, with keys in this order: + +```json +"error": { + "name": "", + "message": "", + "cause": { "name": "...", "message": "...", "cause": { ... } } +} +``` + +`cause` is present only when the error has one. It comes from `Throwable.getCause()`, has the same shape, is cut +after 16 levels, and stops at a cause that was already visited, so a cycle cannot loop. Levels are written exactly as +`getCause()` returns them; a level that repeats the one above is not skipped. When an SDK wrapper carries a +checkpointed `ErrorObject`, `name` and `message` come from the `ErrorObject` and the chain still follows +`getCause()`, so a live cause behind the wrapper (for example the real failure behind a checkpoint continuation +failure) is kept. Reading a cause never throws: if a cause's `getCause()`, name or message fails, the chain ends there +and the record is kept. If the top-level error's `getMessage()` fails, its `message` is written as `null` and its +`name` is the class's simple name; the record is kept. Java names and messages are always strings (`message` is `null` when `getMessage()` is), and +`getCause()` always returns a `Throwable`, so the shared rule for causes that are not error objects does not arise in +this SDK. + +Every `name` and `message`, on errors and on causes, is capped at 8,192 characters (Unicode code points), so a cut +never splits a surrogate pair. A longer value keeps its first 8,192 code points followed by `…`. When a record is over an exporter's size limit, truncation first removes +every `cause` from every error and sets `droppedCauses: true`, then runs the other truncation steps. + +Which errors carry a cause in this SDK: + +- **Only the execution-level `error`.** It is the live `Throwable` the handler ended with. +- **Never `operations[].error` or `operationsByName..error`.** The SDK builds each operation error from the + operation's checkpoint, which stores only the error type, message and stack trace, so there is no cause to write. + This holds for the invocation where the operation failed and for every replay. +- **A rebuilt cause can be named `Throwable`.** When a step throws an exception that has a cause, the SDK checkpoints + the exception and later rebuilds it; the rebuilt cause is a `java.lang.Throwable` carrying the original message, + not an instance of the original class. For example, a step that throws + `new RuntimeException("boom", new IOException("io-detail"))` and fails the execution gives + `{"name":"RuntimeException","message":"boom","cause":{"name":"Throwable","message":"io-detail"}}`. Do not rely on + `error.cause.name` to find the original cause type. + +`includeErrors(false)` removes errors together with their causes. With `includeErrors(true)`, cause messages are +written unredacted, up to 16 of them per error, and there is no transform for errors. Library exceptions often put +hostnames, file paths, SQL or request identifiers in their messages; use `includeErrors(false)` if that is a concern. + +Error names are each language's own type names, so they differ across SDKs. For example, a step that throws +`PaymentError` can be reported as `StepError` in JS, `aws_durable_execution_sdk_python.exceptions.StepError` or +`PaymentError` in Python, and `com.example.PaymentError` or `PaymentError` in Java. In Java, an operation error's +`name` is the fully qualified class name stored in the checkpoint (`com.example.PaymentError`), while the +execution-level `error` uses the simple class name of the exception the handler ended with (`PaymentError`). When the +SDK cannot rebuild the step's exception and throws its `StepFailedException` fallback instead, the execution-level +name is the checkpointed fully qualified name. To compare failures across SDKs, query per SDK, or query on `message` +and the `cause` chain. + ## Design - **Snapshot-based, not accumulated.** Each record is built directly from the current-invocation @@ -187,7 +286,8 @@ one event per record to `{logStreamPrefix}YYYY/MM/DD` (IAM: `logs:CreateLogStrea the cached input snapshot or any later emission), and a transform that throws omits the field and logs the failure rather than silently dropping it or failing the execution. - **`includeErrors`** gates **both** the execution-level error and each operation-level error; with - `includeErrors(false)` neither is emitted, so a sensitive failure message never reaches a record. + `includeErrors(false)` neither is emitted, nor its `cause` chain, so a sensitive failure message never reaches a + record. - **Non-fatal plugin failures are isolated.** Record construction, snapshotting, transforms, truncation and exporter render/export/flush contain and log ordinary failures, including `AssertionError` and optional-dependency `NoClassDefFoundError`. `VirtualMachineError` and @@ -197,10 +297,14 @@ one event per record to `{logStreamPrefix}YYYY/MM/DD` (IAM: `logs:CreateLogStrea fail, including when another exporter is blocked. Subsequent scheduler calls rethrow that fatal; a finished fatal pump cannot later be mistaken for a successful drain or flush. Already-running customer exporter code cannot be forcibly stopped, but it does not delay fatal delivery. -- **Per-exporter size truncation** (`Truncation`) drops, in order: operation results oldest-first, - then whole operations oldest-first, then execution input, then output — setting `truncated`, - `droppedOperations`, `droppedInput`, `droppedOutput` as applicable. The size is measured against - the exact shape each exporter emits (its `render`). +- **Per-exporter size truncation** (`Truncation`) is one pass that drops, in order: every error + `cause`, then operation results oldest-first, then whole operations oldest-first, then execution + input, then output — setting `truncated`, `droppedCauses`, `droppedOperations`, `droppedInput`, + `droppedOutput` as applicable. Markers already on a record are kept. The size is measured against + the exact value each exporter emits (its `render`). `LambdaLogExporter` computes that size without + building the line, and the pass skips checks that cannot succeed yet (each operation's largest + possible saving is sized once), so the cost grows with the record rather than with the number of + dropped operations times the record. The execution `error` and identity fields are never dropped. - **Exporter isolation.** Every exporter receives its own copy of each record, truncated to its own limit; ordinary exporter failures are logged without preventing the other exporters from running. - **Export scheduling.** Exporter I/O never runs on the SDK threads that deliver plugin hooks. diff --git a/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/DurableExecutionTelemetry.java b/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/DurableExecutionTelemetry.java index 591b69e7b..f9cd9871e 100644 --- a/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/DurableExecutionTelemetry.java +++ b/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/DurableExecutionTelemetry.java @@ -3,6 +3,9 @@ package software.amazon.lambda.durable.telemetry; import com.fasterxml.jackson.core.JsonProcessingException; +import java.util.Collections; +import java.util.IdentityHashMap; +import java.util.Set; import java.util.function.Function; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -14,6 +17,7 @@ import software.amazon.lambda.durable.plugin.InvocationInfo; import software.amazon.lambda.durable.plugin.InvocationStatus; import software.amazon.lambda.durable.plugin.OperationChangeItemInfo; +import software.amazon.lambda.durable.telemetry.exporters.LambdaLogExporter; import software.amazon.lambda.durable.telemetry.internal.FatalErrors; /** @@ -76,8 +80,9 @@ static void exportRecord(DurableExecutionTelemetryRecord record, TelemetryExport // so without this a custom exporter that mutates operations or nested content would corrupt every // other exporter's view of the same record. DurableExecutionTelemetryRecord isolated = record.deepCopy(); - DurableExecutionTelemetryRecord shaped = - Truncation.truncateRecord(isolated, exporter.maxRecordSizeBytes(), exporter::render); + DurableExecutionTelemetryRecord shaped = exporter instanceof LambdaLogExporter lambdaLog + ? Truncation.truncateMeasured(isolated, exporter.maxRecordSizeBytes(), lambdaLog::renderedSize) + : Truncation.truncateRecord(isolated, exporter.maxRecordSizeBytes(), exporter::render); exporter.export(shaped); } catch (Throwable t) { // Ordinary Errors (such as missing optional dependencies) are isolated too. @@ -154,7 +159,86 @@ static void logSafely(String message, Throwable t) { } } + /** The cause chain written under an error is cut after this many levels. */ + static final int MAX_CAUSE_DEPTH = 16; + static ErrorInfo toErrorInfo(Throwable t) { + Set visited = Collections.newSetFromMap(new IdentityHashMap<>()); + visited.add(t); + return toErrorInfo(t, visited, 0); + } + + /** + * Converts {@code t} and, recursively, its {@link Throwable#getCause()} chain. Name and message come from the + * checkpointed {@link ErrorObject} when the SDK wrapper carries one; the chain still follows {@code getCause()}, so + * a live cause behind such a wrapper (for example a checkpoint continuation failure) is kept. The chain is cut + * after {@link #MAX_CAUSE_DEPTH} levels and stops at a cause already visited (by identity), so a cycle cannot loop. + * A non-fatal failure while reading a cause's name or message ends the chain at that cause. For the root error + * ({@code depth == 0}) such a failure does not drop the error: the message is written as {@code null} and the name + * falls back to the class's simple name. + */ + private static ErrorInfo toErrorInfo(Throwable t, Set visited, int depth) { + ErrorInfo cause = null; + Throwable next = readCause(t); + if (next != null && depth < MAX_CAUSE_DEPTH && visited.add(next)) { + try { + cause = toErrorInfo(next, visited, depth + 1); + } catch (Throwable failure) { + FatalErrors.rethrow(failure); + cause = null; // A cause whose name or message cannot be read ends the chain. + } + } + return depth == 0 ? rootNameAndMessage(t, cause) : nameAndMessage(t, cause); + } + + /** {@link #nameAndMessage} for the root error: an accessor that throws yields the class name and a null message. */ + private static ErrorInfo rootNameAndMessage(Throwable t, ErrorInfo cause) { + try { + return nameAndMessage(t, cause); + } catch (Throwable failure) { + FatalErrors.rethrow(failure); + } + ErrorObject error = null; + try { + error = extractErrorObject(t); + } catch (Throwable failure) { + FatalErrors.rethrow(failure); + } + String name = null; + String message = null; + if (error != null) { + try { + name = error.errorType(); + } catch (Throwable failure) { + FatalErrors.rethrow(failure); + } + try { + message = error.errorMessage(); + } catch (Throwable failure) { + FatalErrors.rethrow(failure); + } + } + if (message == null) { + try { + message = t.getMessage(); + } catch (Throwable failure) { + FatalErrors.rethrow(failure); + } + } + return new ErrorInfo(name != null ? name : t.getClass().getSimpleName(), message, cause); + } + + /** {@code t.getCause()}, or {@code null} when the accessor fails with a non-fatal error. */ + private static Throwable readCause(Throwable t) { + try { + return t.getCause(); + } catch (Throwable failure) { + FatalErrors.rethrow(failure); + return null; + } + } + + private static ErrorInfo nameAndMessage(Throwable t, ErrorInfo cause) { // Operation and execution snapshot errors are exposed wrapped: operation failures as DurableOperationException // and unrecoverable execution failures as UnrecoverableDurableExecutionException. The wrapper's own // class/message would lose the original checkpointed failure identity. When the checkpointed ErrorObject is @@ -165,9 +249,9 @@ static ErrorInfo toErrorInfo(Throwable t) { String name = error.errorType() != null ? error.errorType() : t.getClass().getSimpleName(); String message = error.errorMessage() != null ? error.errorMessage() : t.getMessage(); - return new ErrorInfo(name, message); + return new ErrorInfo(name, message, cause); } - return new ErrorInfo(t.getClass().getSimpleName(), t.getMessage()); + return new ErrorInfo(t.getClass().getSimpleName(), t.getMessage(), cause); } /** diff --git a/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/DurableExecutionTelemetryRecord.java b/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/DurableExecutionTelemetryRecord.java index 97468f825..fd5ba65b9 100644 --- a/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/DurableExecutionTelemetryRecord.java +++ b/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/DurableExecutionTelemetryRecord.java @@ -34,6 +34,7 @@ public final class DurableExecutionTelemetryRecord { ErrorInfo error; List operations = new ArrayList<>(); Boolean truncated; + Boolean droppedCauses; Integer droppedOperations; Boolean droppedInput; Boolean droppedOutput; @@ -88,6 +89,7 @@ public DurableExecutionTelemetryRecord copy() { c.error = error; c.operations = new ArrayList<>(operations); c.truncated = truncated; + c.droppedCauses = droppedCauses; c.droppedOperations = droppedOperations; c.droppedInput = droppedInput; c.droppedOutput = droppedOutput; @@ -167,6 +169,9 @@ private void putTruncationMarkers(Map data) { if (truncated != null) { data.put("truncated", truncated); } + if (droppedCauses != null) { + data.put("droppedCauses", droppedCauses); + } if (droppedOperations != null) { data.put("droppedOperations", droppedOperations); } diff --git a/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/ErrorInfo.java b/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/ErrorInfo.java index 08889819f..6deb64862 100644 --- a/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/ErrorInfo.java +++ b/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/ErrorInfo.java @@ -5,14 +5,42 @@ import java.util.LinkedHashMap; import java.util.Map; -/** Error detail carried on a record or operation, mirroring the JS {@code {name, message}} shape. */ +/** + * Error detail carried on a record or operation: {@code {name, message, cause?}}, the shape shared by the Telemetry + * exporters of every SDK. {@code cause} has the same shape and is {@code null} when the error has no cause. + * + *

{@code name} and {@code message} are each capped at 8,192 characters, counted as Unicode code points: a longer + * value keeps its first 8,192 code points followed by {@code …}, and a cut never splits a surrogate pair. The cap is + * applied here, when the error is created, so it holds for the error and every cause. Applying it again leaves the + * value unchanged. + */ public final class ErrorInfo { + /** Longest {@code name} or {@code message} written, in Unicode code points, before the {@code …} marker. */ + static final int MAX_CHARS = 8_192; + private final String name; private final String message; + private final ErrorInfo cause; public ErrorInfo(String name, String message) { - this.name = name; - this.message = message; + this(name, message, null); + } + + public ErrorInfo(String name, String message, ErrorInfo cause) { + this.name = cap(name); + this.message = cap(message); + this.cause = cause; + } + + /** Caps {@code value} at {@link #MAX_CHARS} code points plus {@code …}; never splits a surrogate pair. */ + static String cap(String value) { + if (value == null || value.length() <= MAX_CHARS) { + return value; + } + if (value.codePointCount(0, value.length()) <= MAX_CHARS) { + return value; + } + return value.substring(0, value.offsetByCodePoints(0, MAX_CHARS)) + "\u2026"; } public String name() { @@ -23,11 +51,22 @@ public String message() { return message; } - /** Serializes to the camelCase wire shape {@code {"name": ..., "message": ...}}. */ + /** The error's cause, or {@code null} when it has none. */ + public ErrorInfo cause() { + return cause; + } + + /** + * Serializes to the wire shape {@code {"name": ..., "message": ..., "cause": ...}}, keys in that order; + * {@code cause} is omitted when absent. + */ public Map toWireMap() { Map data = new LinkedHashMap<>(); data.put("name", name); data.put("message", message); + if (cause != null) { + data.put("cause", cause.toWireMap()); + } return data; } } diff --git a/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/Json.java b/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/Json.java index 3459391a0..2b64b958e 100644 --- a/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/Json.java +++ b/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/Json.java @@ -74,6 +74,39 @@ public static Integer byteSize(Object value) { } } + /** + * The UTF-8 size {@code text} takes when it is written as the content of a JSON string (quotes excluded), using + * {@link #stringify}'s escaping: {@code "} and {@code \} and the control characters {@code \b \t \n \f \r} take 2 + * bytes, other control characters below U+0020 take 6 ({@code \}{@code u00XX}), and everything else is written as + * UTF-8 (a surrogate pair is 4 bytes; a lone surrogate is replaced by {@code ?}, 1 byte). + */ + public static long escapedUtf8Size(CharSequence text) { + long size = 0; + int length = text.length(); + for (int i = 0; i < length; i++) { + char c = text.charAt(i); + if (c < 0x80) { + if (c == '"' || c == '\\' || c == '\b' || c == '\t' || c == '\n' || c == '\f' || c == '\r') { + size += 2; + } else if (c < 0x20) { + size += 6; + } else { + size += 1; + } + } else if (c < 0x800) { + size += 2; + } else if (Character.isHighSurrogate(c) && i + 1 < length && Character.isLowSurrogate(text.charAt(i + 1))) { + size += 4; + i++; + } else if (Character.isSurrogate(c)) { + size += 1; + } else { + size += 3; + } + } + return size; + } + /** * Deep-copies JSON content (execution input/output and operation results) so a mutation by one exporter cannot leak * into another exporter's record. Maps and lists are rebuilt recursively. Known-immutable scalar values diff --git a/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/TelemetryExporter.java b/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/TelemetryExporter.java index 92d6b399b..42aa6f79e 100644 --- a/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/TelemetryExporter.java +++ b/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/TelemetryExporter.java @@ -48,7 +48,8 @@ default Integer maxRecordSizeBytes() { /** * Maps a record to the exact value this exporter serializes/sends (defaults to the canonical {@code operations} - * array wire map). Overriding exporters (CloudWatch/Lambda log) return the {@code operationsByName} rendering. + * array wire map). {@code CloudWatchLogsExporter} returns the {@code operationsByName} rendering, and + * {@code LambdaLogExporter} returns the log envelope it prints, whose {@code message} is that rendering. */ default Object render(DurableExecutionTelemetryRecord record) { return record.toWireMap(); diff --git a/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/Truncation.java b/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/Truncation.java index bca6285ae..8b0e0e224 100644 --- a/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/Truncation.java +++ b/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/Truncation.java @@ -6,21 +6,27 @@ import java.util.Comparator; import java.util.List; import java.util.function.Function; +import java.util.function.Supplier; /** - * Best-effort size limiter, porting the JS {@code truncateRecord} drop order: + * Best-effort size limiter. When the record is over the limit it drops, in order: * *

    + *
  1. every error {@code cause} (on the execution {@code error} and every operation {@code error}, and so on every + * {@code operationsByName} error), setting {@code droppedCauses: true}; *
  2. operation {@code result} fields, oldest operation first; *
  3. whole operations, oldest first; *
  4. last resort — execution {@code input}, then {@code output}. *
* *

Identity/timeline fields are never dropped. When anything is dropped the returned record has {@code truncated: - * true}; each operation whose result was dropped is itself marked {@code truncated: true}, and - * {@code droppedOperations} / {@code droppedInput} / {@code droppedOutput} markers are set as applicable. The input - * record is never mutated. {@code render} maps the record to the exact shape the exporter serializes so the size check - * measures what is actually emitted. + * true}; each operation whose result was dropped is itself marked {@code truncated: true}, and {@code droppedCauses} / + * {@code droppedOperations} / {@code droppedInput} / {@code droppedOutput} markers are set as applicable. Markers + * already on the incoming record are kept, and {@code droppedOperations} adds to an incoming count, so truncating an + * already-truncated record never under-reports. The input record is never mutated. {@code render} maps the record to + * the exact value the exporter serializes so the size check measures what is actually emitted. This is one pass; the + * execution {@code error} and identity fields are never dropped, so a record can still be over a limit smaller than + * those fields. */ public final class Truncation { @@ -30,77 +36,186 @@ public static DurableExecutionTelemetryRecord truncateRecord( DurableExecutionTelemetryRecord record, Integer maxBytes, Function render) { + return truncate(record, maxBytes, r -> Json.byteSize(render.apply(r)), false); + } + + /** + * {@link #truncateRecord} for a rendering whose size {@code size} measures directly, such as + * {@link software.amazon.lambda.durable.telemetry.exporters.LambdaLogExporter#renderedSize}. The rendering must be + * the record's JSON, or that JSON embedded as one JSON string value, so that dropping one result or operation + * shrinks it by at most that operation's escaped JSON size plus a small constant. Size checks that cannot succeed + * under that bound are skipped; the outcome is the same as checking after every step. + */ + static DurableExecutionTelemetryRecord truncateMeasured( + DurableExecutionTelemetryRecord record, + Integer maxBytes, + Function size) { + return truncate(record, maxBytes, size, true); + } + + /** + * Slack added to each operation's escaped JSON size when bounding how much dropping its result, or the whole + * operation, can shrink the rendering. It covers the {@code operationsByName} summary fields that are not in the + * operation's own JSON ({@code count}, {@code failedCount} and the duration aggregates) and separators. + */ + private static final long SHRINK_SLACK_BYTES = 1_024; + + private static DurableExecutionTelemetryRecord truncate( + DurableExecutionTelemetryRecord record, + Integer maxBytes, + Function size, + boolean bounded) { if (maxBytes == null || maxBytes <= 0) { return record; } - Integer initialSize = Json.byteSize(render.apply(record)); + Integer initialSize = size.apply(record); if (initialSize == null || initialSize <= maxBytes) { return record; } + // Step 1: drop every error cause. operationsByName errors are derived from the operation errors. + boolean causesDropped = record.error != null && record.error.cause() != null; List ops = new ArrayList<>(record.operations().size()); for (OperationRecord op : record.operations()) { - ops.add(op.copy()); + OperationRecord copy = op.copy(); + if (copy.error() != null && copy.error().cause() != null) { + copy.error(withoutCause(copy.error())); + causesDropped = true; + } + ops.add(copy); } + if (causesDropped) { + DurableExecutionTelemetryRecord base = record.copy(); + base.error = withoutCause(record.error); + base.droppedCauses = Boolean.TRUE; + record = base; + } + DurableExecutionTelemetryRecord record0 = record; boolean[] kept = new boolean[ops.size()]; java.util.Arrays.fill(kept, true); int[] order = oldestFirstOrder(ops); + long[] shrinkBound = bounded ? shrinkBounds(ops) : null; + Meter meter = new Meter(size, maxBytes, bounded); boolean[] anyResultDropped = {false}; - int[] droppedOperations = {0}; - boolean[] droppedInput = {false}; - boolean[] droppedOutput = {false}; + int[] droppedOperations = {record.droppedOperations != null ? record.droppedOperations : 0}; + boolean[] droppedInput = {Boolean.TRUE.equals(record.droppedInput)}; + boolean[] droppedOutput = {Boolean.TRUE.equals(record.droppedOutput)}; + int incomingDroppedOperations = droppedOperations[0]; - // Phase 1: drop operation results, oldest first (kept + marked truncated). + // Step 2: drop operation results, oldest first (kept + marked truncated). for (int idx : order) { - if (fits(record, ops, kept, droppedOperations[0], droppedInput[0], droppedOutput[0], render, maxBytes)) { + if (meter.fits( + () -> candidate(record0, ops, kept, droppedOperations[0], droppedInput[0], droppedOutput[0]))) { break; } if (kept[idx] && ops.get(idx).result() != null) { ops.set(idx, ops.get(idx).copy().result(null).truncated(true)); anyResultDropped[0] = true; + meter.shrunkBy(bounded ? shrinkBound[idx] : 0); } } - // Phase 2: drop whole operations, oldest first. + // Step 3: drop whole operations, oldest first. for (int idx : order) { - if (fits(record, ops, kept, droppedOperations[0], droppedInput[0], droppedOutput[0], render, maxBytes)) { + if (meter.fits( + () -> candidate(record0, ops, kept, droppedOperations[0], droppedInput[0], droppedOutput[0]))) { break; } if (kept[idx]) { kept[idx] = false; droppedOperations[0]++; + meter.shrunkBy(bounded ? shrinkBound[idx] : 0); } } - // Phase 3 (last resort): drop execution input, then output. - if (!fits(record, ops, kept, droppedOperations[0], droppedInput[0], droppedOutput[0], render, maxBytes) + // Step 4 (last resort): drop execution input, then output. + if (!meter.fits(() -> candidate(record0, ops, kept, droppedOperations[0], droppedInput[0], droppedOutput[0])) && record.input != null) { droppedInput[0] = true; + meter.forget(); // Input can be any size: measure again. } - if (!fits(record, ops, kept, droppedOperations[0], droppedInput[0], droppedOutput[0], render, maxBytes) + if (!meter.fits(() -> candidate(record0, ops, kept, droppedOperations[0], droppedInput[0], droppedOutput[0])) && record.output != null) { droppedOutput[0] = true; } - if (!anyResultDropped[0] && droppedOperations[0] == 0 && !droppedInput[0] && !droppedOutput[0]) { + if (!causesDropped + && !anyResultDropped[0] + && droppedOperations[0] == incomingDroppedOperations + && Boolean.TRUE.equals(record.droppedInput) == droppedInput[0] + && Boolean.TRUE.equals(record.droppedOutput) == droppedOutput[0]) { return record; } return candidate(record, ops, kept, droppedOperations[0], droppedInput[0], droppedOutput[0]); } - private static boolean fits( - DurableExecutionTelemetryRecord record, - List ops, - boolean[] kept, - int droppedOperations, - boolean droppedInput, - boolean droppedOutput, - Function render, - int maxBytes) { - Integer size = Json.byteSize( - render.apply(candidate(record, ops, kept, droppedOperations, droppedInput, droppedOutput))); - return size != null && size <= maxBytes; + private static ErrorInfo withoutCause(ErrorInfo error) { + if (error == null || error.cause() == null) { + return error; + } + return new ErrorInfo(error.name(), error.message()); + } + + /** + * Upper bound, per operation, on how much dropping its result or the whole operation can shrink the rendering: the + * operation's JSON and its name, sized as if escaped again inside a JSON string, plus {@link #SHRINK_SLACK_BYTES}. + * Each operation is serialized once. + */ + private static long[] shrinkBounds(List ops) { + long[] bounds = new long[ops.size()]; + for (int i = 0; i < bounds.length; i++) { + OperationRecord op = ops.get(i); + long bound = SHRINK_SLACK_BYTES; + try { + bound += Json.escapedUtf8Size(Json.stringify(op.toWireMap())); + if (op.name() != null) { + bound += Json.escapedUtf8Size(Json.stringify(op.name())); + } + } catch (IllegalStateException e) { + bound = Long.MAX_VALUE / 4; // Unknown: never skip a check after dropping this operation. + } + bounds[i] = bound; + } + return bounds; + } + + /** + * Runs the size checks. With a bound, after an exact measurement it skips checks while the measured size minus the + * most the drops since then could have removed is still over the limit; such a check could not succeed, so the + * outcome is the same as measuring every time. + */ + static final class Meter { + private final Function size; + private final int maxBytes; + private final boolean bounded; + private long lastSize = -1; // -1: no usable measurement + private long shrinkSinceLast; + + Meter(Function size, int maxBytes, boolean bounded) { + this.size = size; + this.maxBytes = maxBytes; + this.bounded = bounded; + } + + boolean fits(Supplier candidate) { + if (bounded && lastSize >= 0 && lastSize - shrinkSinceLast > maxBytes) { + return false; + } + Integer measured = size.apply(candidate.get()); + lastSize = measured != null ? measured : -1; + shrinkSinceLast = 0; + return measured != null && measured <= maxBytes; + } + + void shrunkBy(long bound) { + shrinkSinceLast += bound; + } + + /** The next check measures, whatever was dropped since the last measurement. */ + void forget() { + lastSize = -1; + } } private static DurableExecutionTelemetryRecord candidate( diff --git a/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/exporters/LambdaLogExporter.java b/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/exporters/LambdaLogExporter.java index bf92ff627..a201a1c83 100644 --- a/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/exporters/LambdaLogExporter.java +++ b/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/exporters/LambdaLogExporter.java @@ -2,18 +2,51 @@ // SPDX-License-Identifier: Apache-2.0 package software.amazon.lambda.durable.telemetry.exporters; +import java.time.Instant; +import java.time.ZoneOffset; +import java.time.format.DateTimeFormatter; +import java.util.LinkedHashMap; +import java.util.Map; import software.amazon.lambda.durable.telemetry.DurableExecutionTelemetryRecord; import software.amazon.lambda.durable.telemetry.Json; import software.amazon.lambda.durable.telemetry.TelemetryExporter; /** * Exports workflow telemetry records to CloudWatch Logs via {@code System.out.println} (Lambda captures stdout to the - * function's own log group, so this needs no extra IAM). Emits the {@code operationsByName} map as ONE single-line JSON - * record, mirroring the JS {@code LambdaLogExporter} (which uses {@code console.log}). The conformance CloudWatch sink - * decodes both raw top-level JSON lines and the Lambda structured-logging envelope. + * function's own log group, so this needs no extra IAM). Emits the {@code operationsByName} map as ONE line. + * + *

The line is the Lambda JSON log envelope {@code {"timestamp", "level": "INFO", "message"}} whose {@code message} + * is the compact record JSON. The Java runtime does not wrap {@code System.out}, so this exporter writes the envelope + * itself; Lambda does not encode a line that is already JSON a second time. + * + *

Line size. {@link #render} returns the envelope, so the single truncation pass run before {@link #export} + * measures the printed line, envelope and second escaping included (every {@code "} and {@code \} in the record is + * escaped again inside {@code message}, so the line can be about twice the record). The pass measures it with + * {@link #renderedSize}, which computes the line's size without building it. The pass drops error causes first, then + * operation results, whole operations, input and output. The execution {@code error} and the identity fields are never + * dropped. Every error name and message is capped at 8,192 code points, at most 7 bytes each on the line, so once + * everything droppable is dropped the line fits any limit of 128 KB or more (including the default 256,000 bytes), and + * fits 64 KB when the execution error's name is under 500 characters; both assume identity fields of the lengths Lambda + * allows. With a limit smaller than the minimal line (for example a limit below the 68-byte envelope), the line is + * still printed, over the limit, with everything droppable dropped and the markers set. + * + *

Lines carry an explicit {@code "level":"INFO"}. Lambda already treats a line without a level as INFO, so + * application log level filtering is unchanged: with the JSON log format and an application log level of {@code WARN} + * or above, these lines are dropped, as raw record lines were before. */ public final class LambdaLogExporter implements TelemetryExporter { - /** CloudWatch Logs caps a single log event at 256 KB. */ + /** UTC ISO-8601 with exactly three fraction digits. */ + private static final DateTimeFormatter TIMESTAMP = + DateTimeFormatter.ofPattern("uuuu-MM-dd'T'HH:mm:ss.SSS'Z'").withZone(ZoneOffset.UTC); + + /** UTF-8 size of the envelope around an empty {@code message}; the timestamp always has the same width. */ + private static final int ENVELOPE_BYTES = + logLine("", Instant.EPOCH).getBytes(java.nio.charset.StandardCharsets.UTF_8).length; + + /** + * Budget for the printed line, envelope included. The default is 256,000 bytes; CloudWatch Logs accepts log events + * of up to 1 MB. + */ private final Integer maxRecordSizeBytes; public LambdaLogExporter() { @@ -29,13 +62,48 @@ public Integer maxRecordSizeBytes() { return maxRecordSizeBytes; } + /** + * Returns the envelope {@code {"timestamp", "level", "message"}} whose {@code message} is the compact + * {@code operationsByName} record JSON. This is exactly what {@link #export} prints, so the size limit measures the + * printed line. The timestamp always has the same width, so the size does not depend on when it is taken. + */ @Override public Object render(DurableExecutionTelemetryRecord record) { - return record.toByNameWireMap(); + return envelope(Json.stringify(record.toByNameWireMap()), Instant.now()); + } + + /** + * The UTF-8 size of the line {@link #export} prints for {@code record}, computed without building the envelope: the + * envelope's fixed size plus the size of the record JSON once it is escaped again as the {@code message} string. + * Equal to the size of {@code Json.stringify(render(record))}; {@code null} if the record cannot be serialized. The + * truncation pass calls this instead of serializing {@link #render} on every size check. + */ + public Integer renderedSize(DurableExecutionTelemetryRecord record) { + String recordJson; + try { + recordJson = Json.stringify(record.toByNameWireMap()); + } catch (IllegalStateException e) { + return null; // Json.stringify reports a serialization failure this way, as Json.byteSize reports null. + } + long size = ENVELOPE_BYTES + Json.escapedUtf8Size(recordJson); + return size > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) size; } @Override public void export(DurableExecutionTelemetryRecord record) { System.out.println(Json.stringify(render(record))); } + + /** Builds the one-line Lambda JSON log envelope, keys in the order timestamp, level, message. */ + static String logLine(String recordJson, Instant now) { + return Json.stringify(envelope(recordJson, now)); + } + + private static Map envelope(String recordJson, Instant now) { + Map line = new LinkedHashMap<>(); + line.put("timestamp", TIMESTAMP.format(now)); + line.put("level", "INFO"); + line.put("message", recordJson); + return line; + } } diff --git a/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/ErrorCauseFormatTest.java b/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/ErrorCauseFormatTest.java new file mode 100644 index 000000000..c77274058 --- /dev/null +++ b/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/ErrorCauseFormatTest.java @@ -0,0 +1,456 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.telemetry; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +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 com.fasterxml.jackson.databind.ObjectMapper; +import java.io.IOException; +import java.time.Instant; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.services.lambda.model.ErrorObject; +import software.amazon.awssdk.services.lambda.model.Operation; +import software.amazon.awssdk.services.lambda.model.OperationStatus; +import software.amazon.lambda.durable.exception.DurableOperationException; +import software.amazon.lambda.durable.exception.UnrecoverableDurableExecutionException; +import software.amazon.lambda.durable.plugin.InvocationEndInfo; +import software.amazon.lambda.durable.plugin.InvocationStatus; +import software.amazon.lambda.durable.plugin.OperationChangeItemInfo; + +/** + * The shared error shape {@code {name, message, cause?}}: keys in that order, {@code cause} from + * {@link Throwable#getCause()} cut after 16 levels and stopped at an already-visited cause, followed behind an SDK + * wrapper that also carries a checkpointed {@link ErrorObject}, never throwing while a cause is read, and every message + * capped at 8,192 characters. Java's {@code getCause()} always returns a {@link Throwable}, so the non-error cause rule + * of the shared format does not arise in this SDK. + */ +class ErrorCauseFormatTest { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + private static final String ARN = + "arn:aws:lambda:us-west-2:1:function:f:$LATEST/durable-execution/exec-1/invocation-1"; + private static final Instant START = Instant.parse("2026-08-05T00:00:00Z"); + + static final class PaymentError extends RuntimeException { + PaymentError(String message, Throwable cause) { + super(message, cause); + } + } + + /** A throwable whose cause is set after construction, so tests can build cycles {@code getCause()} returns. */ + static final class LoopingError extends RuntimeException { + Throwable next; + + LoopingError(String message) { + super(message); + } + + @Override + public synchronized Throwable getCause() { + return next; + } + } + + private static final class CapturingExporter implements TelemetryExporter { + final List records = new ArrayList<>(); + + @Override + public void export(DurableExecutionTelemetryRecord record) { + records.add(record); + } + } + + private static Map failedOp(Throwable opError) { + Map m = new LinkedHashMap<>(); + m.put( + "op-1", + new OperationChangeItemInfo( + "op-1", + "charge", + "STEP", + "Step", + null, + START, + START.plusMillis(5), + OperationStatus.FAILED, + 1, + false, + opError, + null)); + return m; + } + + private static DurableExecutionTelemetryRecord run(boolean includeErrors, Throwable opError, Throwable execError) { + var exporter = new CapturingExporter(); + var plugin = Executions.plugin( + DurableExecutionTelemetry.durableExecutionTelemetry(DurableExecutionTelemetryConfig.builder() + .content(ContentConfig.builder() + .includeErrors(includeErrors) + .build()) + .addExporter(exporter) + .build()), + ARN, + START); + plugin.onInvocationEnd(new InvocationEndInfo( + "req", ARN, true, START, failedOp(opError), InvocationStatus.FAILED, execError, "in", null)); + assertEquals(1, exporter.records.size()); + return exporter.records.get(0); + } + + /** Builds a chain of {@code causes} live causes below a top-level error. */ + private static Throwable chain(int causes) { + Throwable t = null; + for (int i = causes; i >= 1; i--) { + t = new IllegalStateException("level " + i, t); + } + return new PaymentError("card declined", t); + } + + private static int depth(ErrorInfo e) { + int d = 0; + for (ErrorInfo c = e.cause(); c != null; c = c.cause()) { + d++; + } + return d; + } + + @SuppressWarnings("unchecked") + private static Map parse(String json) throws Exception { + return MAPPER.readValue(json, Map.class); + } + + @Test + @SuppressWarnings("unchecked") + void shapeAndKeyOrderInEveryErrorLocation() throws Exception { + var rec = run(true, chain(1), chain(1)); + Map expected = new LinkedHashMap<>(); + expected.put("name", "PaymentError"); + expected.put("message", "card declined"); + Map cause = new LinkedHashMap<>(); + cause.put("name", "IllegalStateException"); + cause.put("message", "level 1"); + expected.put("cause", cause); + + Map array = parse(Json.stringify(rec.toWireMap())); + Map execError = (Map) array.get("error"); + Map opError = (Map) + ((List>) array.get("operations")).get(0).get("error"); + Map byName = parse(Json.stringify(rec.toByNameWireMap())); + Map byNameError = + (Map) ((Map>) byName.get("operationsByName")) + .get("charge") + .get("error"); + + for (Map error : List.of(execError, opError, byNameError)) { + assertEquals(expected, error); + assertEquals(List.of("name", "message", "cause"), List.copyOf(error.keySet())); + assertEquals(List.of("name", "message"), List.copyOf(((Map) error.get("cause")).keySet())); + } + // The serialized text keeps the order too (Jackson parses into a LinkedHashMap, but check the bytes). + assertTrue(Json.stringify(rec.toByNameWireMap()) + .contains("\"error\":{\"name\":\"PaymentError\",\"message\":\"card declined\"," + + "\"cause\":{\"name\":\"IllegalStateException\",\"message\":\"level 1\"}}")); + } + + @Test + void errorWithoutCauseOmitsTheKey() { + var rec = run(true, new PaymentError("card declined", null), new PaymentError("boom", null)); + assertEquals( + List.of("name", "message"), List.copyOf(rec.error.toWireMap().keySet())); + assertNull(rec.operations().get(0).error().cause()); + } + + @Test + void causeChainDepthOneSixteenAndSeventeen() { + assertEquals(1, depth(DurableExecutionTelemetry.toErrorInfo(chain(1)))); + ErrorInfo sixteen = DurableExecutionTelemetry.toErrorInfo(chain(16)); + assertEquals(16, depth(sixteen)); + ErrorInfo seventeen = DurableExecutionTelemetry.toErrorInfo(chain(17)); + assertEquals(16, depth(seventeen), "chain cut after 16 levels"); + ErrorInfo last = seventeen; + while (last.cause() != null) { + last = last.cause(); + } + assertEquals("level 16", last.message()); + } + + @Test + void cycleStopsAtAnAlreadyVisitedCause() { + LoopingError a = new LoopingError("a"); + LoopingError b = new LoopingError("b"); + a.next = b; + b.next = a; + ErrorInfo info = DurableExecutionTelemetry.toErrorInfo(a); + assertEquals("a", info.message()); + assertEquals("b", info.cause().message()); + assertNull(info.cause().cause(), "a was already visited"); + + LoopingError self = new LoopingError("self"); + self.next = self; + assertNull(DurableExecutionTelemetry.toErrorInfo(self).cause()); + } + + @Test + void namesAreUnchanged() { + ErrorInfo info = DurableExecutionTelemetry.toErrorInfo(chain(1)); + assertEquals("PaymentError", info.name()); + assertEquals("IllegalStateException", info.cause().name()); + } + + @Test + void checkpointedOperationErrorHasNoCause() { + ErrorObject error = ErrorObject.builder() + .errorType("PaymentError") + .errorMessage("card declined") + .build(); + var opError = + new DurableOperationException(Operation.builder().id("op-1").build(), error); + var rec = run(true, opError, opError); + ErrorInfo op = rec.operations().get(0).error(); + assertEquals("PaymentError", op.name()); + assertEquals("card declined", op.message()); + assertNull(op.cause()); + assertNull(rec.error.cause()); + assertFalse(Json.stringify(rec.toByNameWireMap()).contains("\"cause\"")); + } + + @Test + void includeErrorsFalseRemovesErrorsAndTheirCauses() { + var rec = run(false, chain(3), chain(3)); + assertNull(rec.error); + assertNull(rec.operations().get(0).error()); + String array = Json.stringify(rec.toWireMap()); + String byName = Json.stringify(rec.toByNameWireMap()); + for (String json : List.of(array, byName)) { + assertFalse(json.contains("card declined"), json); + assertFalse(json.contains("level 1"), json); + assertFalse(json.contains("\"error\""), json); + } + } + + /** A throwable whose accessors fail, to check that reading a cause never throws. */ + static final class BrokenError extends RuntimeException { + final boolean brokenCause; + final RuntimeException failure; + + BrokenError(boolean brokenCause, RuntimeException failure) { + super("broken"); + this.brokenCause = brokenCause; + this.failure = failure; + } + + @Override + public synchronized Throwable getCause() { + if (brokenCause) { + throw failure; + } + return null; + } + + @Override + public String getMessage() { + if (!brokenCause) { + throw failure; + } + return "broken"; + } + } + + /** A checkpoint continuation failure keeps the ErrorObject identity and the live cause. */ + @Test + void wrapperWithErrorObjectAndLiveCauseFollowsTheCause() { + var io = new IOException("disk full", new IllegalStateException("inner")); + var control = new UnrecoverableDurableExecutionException( + ErrorObject.builder() + .errorType("java.io.IOException") + .errorMessage("Error in SDK checkpoint continuation") + .build(), + true, + io); + ErrorInfo info = DurableExecutionTelemetry.toErrorInfo(control); + assertEquals("java.io.IOException", info.name()); + assertEquals("Error in SDK checkpoint continuation", info.message()); + assertEquals("IOException", info.cause().name()); + assertEquals("disk full", info.cause().message()); + assertEquals("IllegalStateException", info.cause().cause().name()); + assertNull(info.cause().cause().cause()); + + // A user exception wrapping such a wrapper continues through it. + ErrorInfo outer = DurableExecutionTelemetry.toErrorInfo(new RuntimeException("outer", control)); + assertEquals("Error in SDK checkpoint continuation", outer.cause().message()); + assertEquals("disk full", outer.cause().cause().message()); + + // An operation wrapper with an ErrorObject and a live cause follows it too. + var opError = new DurableOperationException( + Operation.builder().id("op-1").build(), + ErrorObject.builder() + .errorType("CallbackError") + .errorMessage("cb") + .build(), + "cb", + io); + ErrorInfo op = DurableExecutionTelemetry.toErrorInfo(opError); + assertEquals("CallbackError", op.name()); + assertEquals("disk full", op.cause().message()); + } + + @Test + void depthAndIdentityGuardsStillApplyBehindAWrapper() { + LoopingError loop = new LoopingError("loop"); + var control = new UnrecoverableDurableExecutionException( + ErrorObject.builder().errorType("X").errorMessage("m").build(), true, loop); + loop.next = control; + ErrorInfo info = DurableExecutionTelemetry.toErrorInfo(control); + assertEquals("loop", info.cause().message()); + assertNull(info.cause().cause(), "the wrapper was already visited"); + + var deep = new UnrecoverableDurableExecutionException( + ErrorObject.builder().errorType("X").errorMessage("m").build(), true, chain(20)); + assertEquals(16, depth(DurableExecutionTelemetry.toErrorInfo(deep))); + } + + /** Rule 8: levels are written exactly as getCause() returns them, even when one repeats the level above. */ + @Test + void repeatedLevelsAreNotSkipped() { + var inner = new IllegalStateException("same"); + var outer = new IllegalStateException("same", inner); + ErrorInfo info = DurableExecutionTelemetry.toErrorInfo(outer); + assertEquals("same", info.message()); + assertEquals("IllegalStateException", info.cause().name()); + assertEquals("same", info.cause().message()); + } + + /** Rule 6: a failure while reading a cause ends the chain there, and the error is still written. */ + @Test + void readingACauseNeverThrows() { + // getCause() throws on the top-level error: written without a cause. + ErrorInfo top = DurableExecutionTelemetry.toErrorInfo(new BrokenError(true, new IllegalStateException("x"))); + assertEquals("broken", top.message()); + assertNull(top.cause()); + + // getCause() throws one level down: the chain ends at that level. + ErrorInfo mid = DurableExecutionTelemetry.toErrorInfo( + new RuntimeException("outer", new BrokenError(true, new IllegalStateException("x")))); + assertEquals("broken", mid.cause().message()); + assertNull(mid.cause().cause()); + + // A cause whose message cannot be read ends the chain above it. + ErrorInfo unreadable = DurableExecutionTelemetry.toErrorInfo( + new RuntimeException("outer", new BrokenError(false, new IllegalStateException("x")))); + assertEquals("outer", unreadable.message()); + assertNull(unreadable.cause()); + + // Through the plugin, the record is kept. + var rec = run(true, chain(1), new RuntimeException("outer", new BrokenError(false, new RuntimeException()))); + assertEquals("outer", rec.error.message()); + assertNull(rec.error.cause()); + } + + /** A root error whose message cannot be read is written with a null message and its class name; it is kept. */ + @Test + void rootErrorWithUnreadableMessageIsWrittenWithNullMessage() { + ErrorInfo root = DurableExecutionTelemetry.toErrorInfo(new BrokenError(false, new IllegalStateException("x"))); + assertEquals("BrokenError", root.name()); + assertNull(root.message()); + + // A readable cause under it is still written. + var withCause = new RuntimeException("unused", new IllegalStateException("inner")) { + @Override + public String getMessage() { + throw new IllegalStateException("unreadable"); + } + }; + ErrorInfo info = DurableExecutionTelemetry.toErrorInfo(withCause); + assertNull(info.message()); + assertEquals("inner", info.cause().message()); + + // Through the plugin, the record and its error are kept. + var rec = run(true, chain(1), new BrokenError(false, new RuntimeException())); + assertEquals("BrokenError", rec.error.name()); + assertNull(rec.error.message()); + } + + @Test + void fatalErrorWhileReadingACausePropagates() { + var oom = new OutOfMemoryError("simulated"); + var t = new RuntimeException("outer") { + @Override + public synchronized Throwable getCause() { + throw oom; + } + }; + assertSame(oom, assertThrows(OutOfMemoryError.class, () -> DurableExecutionTelemetry.toErrorInfo(t))); + } + + /** Rule 7: every message, on the error and on each cause, is capped at 8,192 characters plus an ellipsis. */ + @Test + void everyMessageIsCappedAt8192Characters() { + String exact = "a".repeat(ErrorInfo.MAX_CHARS); + String over = "b".repeat(ErrorInfo.MAX_CHARS + 1); + String huge = "\\".repeat(100_000); + ErrorInfo info = DurableExecutionTelemetry.toErrorInfo( + new RuntimeException(huge, new IllegalStateException(over, new IllegalStateException(exact)))); + assertEquals("\\".repeat(8_192) + "\u2026", info.message()); + assertEquals("b".repeat(8_192) + "\u2026", info.cause().message()); + assertEquals(exact, info.cause().cause().message(), "exactly 8,192 characters is not cut"); + + // The cap counts code points and never splits a surrogate pair; applying it again changes nothing. + String emoji = "\uD83D\uDE00".repeat(8_193); + ErrorInfo e = new ErrorInfo("E", emoji); + assertEquals("\uD83D\uDE00".repeat(8_192) + "\u2026", e.message()); + assertEquals(e.message(), new ErrorInfo("E", e.message()).message()); + + // Operation errors and the ErrorObject message path are capped too. + var opError = new DurableOperationException( + Operation.builder().id("op-1").build(), + ErrorObject.builder().errorType("T").errorMessage(over).build()); + var rec = run(true, opError, opError); + assertEquals(8_193, rec.operations().get(0).error().message().length()); + assertEquals(8_193, rec.error.message().length()); + } + + /** Size limit step 1: every cause is removed from every error before anything else is dropped. */ + @Test + @SuppressWarnings("unchecked") + void truncationDropsEveryCauseFirst() throws Exception { + var rec = run(true, chain(2), chain(2)); + rec.input = "small"; + int full = Json.byteSize(rec.toByNameWireMap()); + var out = Truncation.truncateRecord(rec, full - 1, DurableExecutionTelemetryRecord::toByNameWireMap); + assertEquals(Boolean.TRUE, out.truncated); + assertEquals(Boolean.TRUE, out.droppedCauses); + assertNull(out.error.cause()); + assertEquals("card declined", out.error.message()); + assertNull(out.operations().get(0).error().cause()); + assertEquals("small", out.input, "dropping causes was enough"); + assertNull(out.droppedOperations); + assertNotNull(rec.error.cause(), "the input record is not mutated"); + + for (String json : List.of(Json.stringify(out.toByNameWireMap()), Json.stringify(out.toWireMap()))) { + assertFalse(json.contains("\"cause\""), json); + Map parsed = parse(json); + List keys = List.copyOf(parsed.keySet()); + assertEquals(Boolean.TRUE, parsed.get("droppedCauses")); + assertEquals(keys.indexOf("truncated") + 1, keys.indexOf("droppedCauses"), json); + } + } + + @Test + void recordWithoutCausesHasNoDroppedCausesMarker() { + var rec = run(true, new PaymentError("a", null), new PaymentError("b", null)); + rec.input = "x".repeat(10_000); + var out = Truncation.truncateRecord(rec, 2_000, DurableExecutionTelemetryRecord::toByNameWireMap); + assertEquals(Boolean.TRUE, out.droppedInput); + assertNull(out.droppedCauses); + } +} diff --git a/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/LambdaLogLineSizeTest.java b/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/LambdaLogLineSizeTest.java new file mode 100644 index 000000000..208e764c4 --- /dev/null +++ b/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/LambdaLogLineSizeTest.java @@ -0,0 +1,219 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.telemetry; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.fasterxml.jackson.databind.ObjectMapper; +import java.io.ByteArrayOutputStream; +import java.io.PrintStream; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; +import software.amazon.lambda.durable.telemetry.exporters.LambdaLogExporter; + +/** Size checks for the {@code LambdaLogExporter} line through the plugin's export path. */ +class LambdaLogLineSizeTest { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + /** Exports through the plugin's path (truncation, then export) and returns the printed line. */ + private static String exportThroughPlugin(LambdaLogExporter exporter, DurableExecutionTelemetryRecord record) { + PrintStream original = System.out; + ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + try (PrintStream capture = new PrintStream(buffer, true, StandardCharsets.UTF_8)) { + System.setOut(capture); + DurableExecutionTelemetry.exportRecord(record, exporter); + } finally { + System.setOut(original); + } + String out = buffer.toString(StandardCharsets.UTF_8); + assertEquals(1, out.lines().count(), "exactly one line"); + return out.strip(); + } + + @SuppressWarnings("unchecked") + private static Map messageOf(String line) throws Exception { + Map envelope = MAPPER.readValue(line, Map.class); + return MAPPER.readValue((String) envelope.get("message"), Map.class); + } + + private static int bytes(String line) { + return line.getBytes(StandardCharsets.UTF_8).length; + } + + /** {@code count} succeeded operations named {@code op-}, one second apart, each with {@code result}. */ + private static DurableExecutionTelemetryRecord withResults(int count, Object result) { + DurableExecutionTelemetryRecord r = RecordFactory.sample(); + r.operations.clear(); + Instant start = Instant.parse("2026-07-15T11:00:00Z"); + for (int i = 0; i < count; i++) { + r.addOperation(new OperationRecord() + .id("id-" + i) + .name("op-" + i) + .type("STEP") + .status("SUCCEEDED") + .startTime(start.plusSeconds(i).toString()) + .result(result)); + } + return r; + } + + /** + * Coarse cost guard, independent of timing: 3,000 operations, each with a 400-quote result, at the default limit. + * The pass measures the line only a few dozen times rather than once per dropped result and operation, and the + * printed line is within the limit and counts every dropped operation. + */ + @Test + @SuppressWarnings("unchecked") + void manyEscapeHeavyResultsNeedFewMeasurements() throws Exception { + var exporter = new LambdaLogExporter(); + var record = withResults(3_000, "\"".repeat(400)); + AtomicInteger measurements = new AtomicInteger(); + var shaped = Truncation.truncateMeasured(record, exporter.maxRecordSizeBytes(), r -> { + measurements.incrementAndGet(); + return exporter.renderedSize(r); + }); + assertTrue(measurements.get() <= 100, measurements + " measurements; one per step would be thousands"); + assertTrue(exporter.renderedSize(shaped) <= 256_000); + + String line = exportThroughPlugin(exporter, record); + assertTrue(bytes(line) <= 256_000, "line within the default limit: " + bytes(line)); + Map parsed = messageOf(line); + int kept = ((Map) parsed.get("operationsByName")).size(); + assertEquals(3_000, kept + (Integer) parsed.get("droppedOperations")); + assertEquals(Json.stringify(shaped.toByNameWireMap()), (String) + MAPPER.readValue(line, Map.class).get("message")); + } + + /** + * Skipping checks does not change the outcome: on records with repeated names (whose summaries change when one + * occurrence is dropped), small and large results, failed operations with causes, and input and output, the + * measured pass returns the same record as measuring {@code render} after every step. + */ + @Test + void measuredPassMatchesMeasuringEveryStep() { + var exporter = new LambdaLogExporter(); + Instant start = Instant.parse("2026-07-15T11:00:00Z"); + for (int limit : new int[] {600, 2_000, 9_000, 30_000, 120_000}) { + DurableExecutionTelemetryRecord r = RecordFactory.sampleWithError("boom \"x\"", "io"); + r.operations.clear(); + r.input = "\\".repeat(3_000); + r.output = Map.of("k", "v".repeat(2_000)); + for (int i = 0; i < 240; i++) { + var op = new OperationRecord() + .id("id-" + i) + .name(i % 5 == 0 ? "op-" + i : "group-" + (i % 7)) + .type(i % 3 == 0 ? "STEP" : "WAIT") + .status(i % 4 == 0 ? "FAILED" : "SUCCEEDED") + .startTime(i % 11 == 0 ? null : start.plusSeconds(i).toString()) + .durationMs((long) i * 997) + .attempt(i % 3 + 1) + .result(i % 6 == 0 ? Integer.valueOf(i) : "\"\\é😀".repeat(i % 50)); + if (i % 4 == 0) { + op.error(new ErrorInfo("E" + i, "m\u0001".repeat(i), new ErrorInfo("C", "c"))); + } + r.addOperation(op); + } + var measured = Truncation.truncateMeasured(r.deepCopy(), limit, exporter::renderedSize); + var everyStep = Truncation.truncateRecord(r.deepCopy(), limit, exporter::render); + assertEquals(Json.stringify(everyStep.toWireMap()), Json.stringify(measured.toWireMap()), "limit " + limit); + } + } + + /** Rule 7: a 70,000-character name is capped at 8,192 code points plus an ellipsis, on the line too. */ + @Test + @SuppressWarnings("unchecked") + void longNameIsCapped() throws Exception { + String name = "N".repeat(70_000); + DurableExecutionTelemetryRecord r = RecordFactory.sample(); + r.status = "FAILED"; + r.error = new ErrorInfo(name, "m", new ErrorInfo(name, "c")); + assertEquals("N".repeat(8_192) + "\u2026", r.error.name()); + assertEquals(r.error.name(), r.error.cause().name()); + + String line = exportThroughPlugin(new LambdaLogExporter(), r); + Map error = (Map) messageOf(line).get("error"); + assertEquals("N".repeat(8_192) + "\u2026", error.get("name")); + } + + /** Long identity fields, at the lengths Lambda allows and beyond, for the size bound tests. */ + private static DurableExecutionTelemetryRecord worstCaseRecord(String name, String message) { + DurableExecutionTelemetryRecord r = RecordFactory.sample(); + r.functionName = "f".repeat(64); + r.functionQualifier = "q".repeat(128); + r.executionName = "e".repeat(256); + r.executionArn = "arn:aws:lambda:us-east-1:123456789012:function:" + "a".repeat(190); + r.region = "ap-southeast-5"; + r.accountId = "123456789012"; + r.status = "FAILED"; + r.endTime = "2026-07-15T12:00:00.000Z"; + r.durationMs = 3_600_000L; + ErrorInfo chain = null; + for (int i = 0; i < 16; i++) { + chain = new ErrorInfo(name, message, chain); + } + r.error = new ErrorInfo(name, message, chain); + r.operations.clear(); + for (int i = 0; i < 50; i++) { + r.addOperation(new OperationRecord() + .id("id-" + i) + .name("op-" + i) + .type("STEP") + .status("FAILED") + .startTime( + Instant.parse("2026-07-15T11:00:00Z").plusSeconds(i).toString()) + .result(message) + .error(new ErrorInfo(name, message, new ErrorInfo(name, message)))); + } + r.input = message.repeat(20); + r.output = message.repeat(20); + return r; + } + + /** + * The 64 KB bound: with an 8,192-character message of the costliest characters (each takes up to 7 bytes of line) + * and a name under 500 characters, once causes, operations, input and output are dropped the line fits 64,000 + * bytes, even with long identity fields. + */ + @Test + void worstCaseMessageFits64KB() throws Exception { + for (String ch : List.of("\u0001", "\\", "\"", "\uD83D\uDE00", "\u2028", "\u00E9")) { + String message = ch.repeat(9_000); + String name = "\u0001".repeat(499); // the costliest name under 500 characters + for (int limit : new int[] {64_000, 65_536}) { + String line = exportThroughPlugin(new LambdaLogExporter(limit), worstCaseRecord(name, message)); + assertTrue(bytes(line) <= limit, "U+" + Integer.toHexString(ch.codePointAt(0)) + ": " + bytes(line)); + Map parsed = messageOf(line); + assertEquals(Boolean.TRUE, parsed.get("droppedCauses")); + assertEquals(50, parsed.get("droppedOperations")); + assertEquals(Boolean.TRUE, parsed.get("droppedInput")); + assertEquals(Boolean.TRUE, parsed.get("droppedOutput")); + } + } + } + + /** + * When the name and the message are both 8,192 of the costliest characters, the error alone can take about 115 KB + * of line, so 64 KB cannot be met; the line still fits any limit of 128,000 bytes or more, including the default. + */ + @Test + void worstCaseNameAndMessageFit128KB() throws Exception { + for (String ch : List.of("\u0001", "\\", "\"", "\uD83D\uDE00")) { + String text = ch.repeat(9_000); + for (int limit : new int[] {128_000, 256_000}) { + String line = exportThroughPlugin(new LambdaLogExporter(limit), worstCaseRecord(text, text)); + assertTrue(bytes(line) <= limit, "U+" + Integer.toHexString(ch.codePointAt(0)) + ": " + bytes(line)); + assertNull(messageOf(line).get("input")); + } + } + String control = "\u0001".repeat(9_000); + String line = exportThroughPlugin(new LambdaLogExporter(64_000), worstCaseRecord(control, control)); + assertTrue(bytes(line) > 64_000, "over 64 KB, as documented: " + bytes(line)); + } +} diff --git a/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/RecordFactory.java b/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/RecordFactory.java index fc116c2c6..8d270ed46 100644 --- a/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/RecordFactory.java +++ b/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/RecordFactory.java @@ -18,4 +18,52 @@ public static DurableExecutionTelemetryRecord sample() { new OperationRecord().id("op-1").name("fetch-user").type("STEP").status("SUCCEEDED")); return r; } + + /** {@link #sample()} with the given execution {@code input}. */ + public static DurableExecutionTelemetryRecord sampleWithInput(Object input) { + DurableExecutionTelemetryRecord r = sample(); + r.input = input; + return r; + } + + /** {@code record} with the given execution {@code input}. */ + public static DurableExecutionTelemetryRecord withInput(DurableExecutionTelemetryRecord record, Object input) { + record.input = input; + return record; + } + + /** {@link #sample()} with a failed status and an execution {@code error} carrying one cause. */ + public static DurableExecutionTelemetryRecord sampleWithError(String message, String causeMessage) { + DurableExecutionTelemetryRecord r = sample(); + r.status = "FAILED"; + r.error = new ErrorInfo("PaymentError", message, new ErrorInfo("IOException", causeMessage)); + return r; + } + + /** + * {@link #sample()} with {@code count} failed operations named {@code op-}, started one second apart, each with + * an error whose message is {@code message} and that has one cause. + */ + public static DurableExecutionTelemetryRecord sampleWithFailedOperations(int count, String message) { + DurableExecutionTelemetryRecord r = sample(); + r.operations.clear(); + java.time.Instant start = java.time.Instant.parse("2026-07-15T11:00:00Z"); + for (int i = 0; i < count; i++) { + r.addOperation(new OperationRecord() + .id("id-" + i) + .name("op-" + i) + .type("STEP") + .status("FAILED") + .startTime(start.plusSeconds(i).toString()) + .error(new ErrorInfo("java.lang.RuntimeException", message, new ErrorInfo("Throwable", "c")))); + } + return r; + } + + /** {@code record} with {@code operation} appended. */ + public static DurableExecutionTelemetryRecord withOperation( + DurableExecutionTelemetryRecord record, OperationRecord operation) { + record.addOperation(operation); + return record; + } } diff --git a/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/TruncationTest.java b/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/TruncationTest.java index 384561cee..e94ff2d40 100644 --- a/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/TruncationTest.java +++ b/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/TruncationTest.java @@ -4,6 +4,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; import org.junit.jupiter.api.Test; @@ -88,4 +89,41 @@ void noTruncationWhenUnderLimit() { Truncation.truncateRecord(r, 5_000_000, DurableExecutionTelemetryRecord::toWireMap); assertNull(out.truncated); } + + /** + * Truncating an already-truncated record adds to its {@code droppedOperations} and keeps its other markers, so a + * second pass can never under-report what was dropped. + */ + @Test + void truncatingATruncatedRecordKeepsItsMarkers() { + DurableExecutionTelemetryRecord r = RecordFactory.sampleWithFailedOperations(40, "m".repeat(200)); + r.input = "i".repeat(5_000); + int full = Json.byteSize(r.toWireMap()); + DurableExecutionTelemetryRecord first = + Truncation.truncateRecord(r, full / 2, DurableExecutionTelemetryRecord::toWireMap); + int firstDropped = first.droppedOperations; + assertTrue(firstDropped > 0, "first pass drops operations"); + assertEquals(Boolean.TRUE, first.droppedCauses); + + DurableExecutionTelemetryRecord second = + Truncation.truncateRecord(first, full / 4, DurableExecutionTelemetryRecord::toWireMap); + assertEquals(40, second.operations().size() + second.droppedOperations, "every dropped operation is counted"); + assertTrue(second.droppedOperations > firstDropped); + assertEquals(Boolean.TRUE, second.droppedCauses, "kept from the first pass"); + assertEquals(Boolean.TRUE, second.truncated); + + // A pass that has nothing more to drop returns the record unchanged. + assertSame(second, Truncation.truncateRecord(second, full, DurableExecutionTelemetryRecord::toWireMap)); + } + + /** One pass reports every operation it dropped. */ + @Test + void droppedOperationsCountsEveryDroppedOperation() { + DurableExecutionTelemetryRecord r = RecordFactory.sampleWithFailedOperations(400, "\"".repeat(200)); + DurableExecutionTelemetryRecord out = + Truncation.truncateRecord(r, 100_000, DurableExecutionTelemetryRecord::toWireMap); + assertTrue(out.droppedOperations > 0); + assertEquals(400, out.operations().size() + out.droppedOperations); + assertTrue(Json.byteSize(out.toWireMap()) <= 100_000); + } } diff --git a/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/exporters/LambdaLogExporterTest.java b/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/exporters/LambdaLogExporterTest.java new file mode 100644 index 000000000..493ac5907 --- /dev/null +++ b/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/exporters/LambdaLogExporterTest.java @@ -0,0 +1,295 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.telemetry.exporters; + +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.ObjectMapper; +import java.io.ByteArrayOutputStream; +import java.io.PrintStream; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.List; +import java.util.Map; +import java.util.regex.Pattern; +import org.junit.jupiter.api.Test; +import software.amazon.lambda.durable.telemetry.DurableExecutionTelemetryRecord; +import software.amazon.lambda.durable.telemetry.Json; +import software.amazon.lambda.durable.telemetry.RecordFactory; +import software.amazon.lambda.durable.telemetry.Truncation; + +/** + * The {@code LambdaLogExporter} line is the Lambda JSON log envelope. Readers query it with {@code filter level = + * "INFO" and message like /DurableExecutionTelemetry/} and then parse {@code message}. + */ +class LambdaLogExporterTest { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + private static final Pattern TIMESTAMP = Pattern.compile("\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{3}Z"); + /** The reader's Logs Insights {@code message like /DurableExecutionTelemetry/} filter. */ + private static final Pattern READER_MESSAGE_FILTER = Pattern.compile("DurableExecutionTelemetry"); + + private static String exportAndCapture(DurableExecutionTelemetryRecord record) { + return exportAndCapture(new LambdaLogExporter(), record); + } + + private static String exportAndCapture(LambdaLogExporter exporter, DurableExecutionTelemetryRecord record) { + PrintStream original = System.out; + ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + try (PrintStream capture = new PrintStream(buffer, true, StandardCharsets.UTF_8)) { + System.setOut(capture); + exporter.export(record); + } finally { + System.setOut(original); + } + return buffer.toString(StandardCharsets.UTF_8); + } + + @Test + @SuppressWarnings("unchecked") + void exportPrintsOneEnvelopeLineWhoseMessageIsTheRecord() throws Exception { + var record = RecordFactory.sample(); + String out = exportAndCapture(record); + + List lines = out.lines().toList(); + assertEquals(1, lines.size(), "exactly one line: " + out); + String line = lines.get(0); + + // The line parses, with keys in the shared cross-SDK order and no requestId (not available to an exporter). + Map envelope = MAPPER.readValue(line, Map.class); + assertEquals(List.of("timestamp", "level", "message"), List.copyOf(envelope.keySet())); + assertTrue(TIMESTAMP.matcher((String) envelope.get("timestamp")).matches(), line); + assertEquals("INFO", envelope.get("level")); + + // message is the compact record JSON string, and parses back to the record's render shape. + String message = (String) envelope.get("message"); + assertEquals(Json.stringify(record.toByNameWireMap()), message); + Map parsed = MAPPER.readValue(message, Map.class); + assertEquals("DurableExecutionTelemetry", parsed.get("recordType")); + assertEquals(record.executionArn(), parsed.get("executionArn")); + assertTrue(parsed.containsKey("operationsByName")); + assertFalse(parsed.containsKey("operations")); + + // The reader's filter: level = "INFO" and message like /DurableExecutionTelemetry/. + assertTrue("INFO".equals(envelope.get("level")) + && READER_MESSAGE_FILTER.matcher(message).find()); + } + + @Test + void timestampAlwaysHasThreeFractionDigitsInUtc() { + assertEquals( + "{\"timestamp\":\"2026-10-21T08:00:00.000Z\",\"level\":\"INFO\",\"message\":\"{}\"}", + LambdaLogExporter.logLine("{}", Instant.parse("2026-10-21T08:00:00Z"))); + assertTrue(LambdaLogExporter.logLine("{}", Instant.parse("2026-10-21T08:00:00.123456789Z")) + .startsWith("{\"timestamp\":\"2026-10-21T08:00:00.123Z\"")); + } + + @Test + @SuppressWarnings("unchecked") + void renderIsThePrintedEnvelope() throws Exception { + var record = RecordFactory.sample(); + var exporter = new LambdaLogExporter(); + Map rendered = (Map) exporter.render(record); + assertEquals(List.of("timestamp", "level", "message"), List.copyOf(rendered.keySet())); + String line = exportAndCapture(record).strip(); + Map printed = MAPPER.readValue(line, Map.class); + assertEquals(rendered.get("message"), printed.get("message")); + assertEquals(Json.byteSize(rendered), line.getBytes(StandardCharsets.UTF_8).length, "same size as measured"); + } + + @Test + void lineThatFitsIsPrintedWhole() { + var record = RecordFactory.sample(); + String line = exportAndCapture(record).strip(); + String recordJson = Json.stringify(record.toByNameWireMap()); + assertEquals( + Json.byteSize(record.toByNameWireMap()) + ENVELOPE_BYTES + escapedChars(recordJson), + line.getBytes(StandardCharsets.UTF_8).length, + "line = record + 68-byte envelope + one byte per quote or backslash in the record"); + assertFalse(line.contains("\\\"truncated\\\""), "not truncated: " + line); + } + + /** Bytes the envelope adds around an empty {@code message}. */ + private static final int ENVELOPE_BYTES = 68; + + private static long escapedChars(String json) { + return json.chars().filter(c -> c == '"' || c == '\\').count(); + } + + @Test + void envelopeAddsSixtyEightBytes() { + String empty = LambdaLogExporter.logLine("", Instant.parse("2026-10-21T08:00:00Z")); + assertEquals(ENVELOPE_BYTES, empty.getBytes(StandardCharsets.UTF_8).length, empty); + } + + /** Runs the one truncation pass the plugin runs before export, then exports, and returns the printed line. */ + private static String shapeAndExport(LambdaLogExporter exporter, DurableExecutionTelemetryRecord record) { + var shaped = Truncation.truncateRecord(record, exporter.maxRecordSizeBytes(), exporter::render); + String out = exportAndCapture(exporter, shaped); + assertEquals(1, out.lines().count(), "exactly one line"); + return out.strip(); + } + + @SuppressWarnings("unchecked") + private static Map messageOf(String line) throws Exception { + Map envelope = MAPPER.readValue(line, Map.class); + return MAPPER.readValue((String) envelope.get("message"), Map.class); + } + + /** + * With the default 256,000-byte limit, an execution error whose message is 100,000 backslashes (about 400 KB of + * line uncapped) prints a line within the limit. The message cap does it; the cause is dropped first. + */ + @Test + @SuppressWarnings("unchecked") + void defaultLimitHoldsForAnEscapeHeavyExecutionError() throws Exception { + var exporter = new LambdaLogExporter(); + String backslashes = "\\".repeat(100_000); + for (String input : new String[] {"small", "\\".repeat(100_000)}) { + var record = RecordFactory.sampleWithError(backslashes, backslashes); + record = RecordFactory.withInput(record, input); + String line = shapeAndExport(exporter, record); + int lineBytes = line.getBytes(StandardCharsets.UTF_8).length; + assertTrue(lineBytes <= 256_000, "printed line within the default limit: " + lineBytes); + + Map parsed = messageOf(line); + Map error = (Map) parsed.get("error"); + assertEquals("\\".repeat(8_192) + "\u2026", error.get("message"), "message capped"); + assertEquals(record.executionArn(), parsed.get("executionArn")); + if (input.length() > 5) { + // The cause is dropped first, then input. + assertEquals(Boolean.TRUE, parsed.get("truncated")); + assertEquals(Boolean.TRUE, parsed.get("droppedCauses")); + assertFalse(error.containsKey("cause")); + assertEquals(Boolean.TRUE, parsed.get("droppedInput")); + } else { + // Capped messages fit without truncation, so the cause is kept. + assertNull(parsed.get("truncated")); + assertEquals("\\".repeat(8_192) + "\u2026", ((Map) error.get("cause")).get("message")); + } + } + } + + /** + * One pass, measured on the printed line, reports every dropped operation. 400 failed operations whose messages are + * full of quotes, limit 100,000. + */ + @Test + @SuppressWarnings("unchecked") + void droppedOperationsIsAccurateWhenTheLineIsMeasured() throws Exception { + var exporter = new LambdaLogExporter(100_000); + var record = RecordFactory.sampleWithFailedOperations(400, "\"".repeat(200)); + String line = shapeAndExport(exporter, record); + assertTrue(line.getBytes(StandardCharsets.UTF_8).length <= 100_000); + + Map parsed = messageOf(line); + int kept = ((Map) parsed.get("operationsByName")).size(); + int dropped = (Integer) parsed.get("droppedOperations"); + assertTrue(kept > 0 && dropped > 0, kept + " kept, " + dropped + " dropped"); + assertEquals(400, kept + dropped, "every dropped operation is counted"); + assertEquals(Boolean.TRUE, parsed.get("droppedCauses")); + } + + /** + * The limit caps the printed line, escaping included: an escape-heavy input that fits the record budget but whose + * line does not is dropped in the one pass, and the message still parses. + */ + @Test + void escapeHeavyInputIsDroppedSoThePrintedLineFits() throws Exception { + for (int limit : new int[] {256_000, 1_000_000}) { + var exporter = new LambdaLogExporter(limit); + for (char ch : new char[] {'"', '\\'}) { + int count = limit == 1_000_000 ? 300_000 : 127_000; + var record = RecordFactory.sampleWithInput(String.valueOf(ch).repeat(count)); + assertTrue(Json.byteSize(record.toByNameWireMap()) <= limit, "record alone fits"); + + String line = shapeAndExport(exporter, record); + int lineBytes = line.getBytes(StandardCharsets.UTF_8).length; + assertTrue(lineBytes <= limit, limit + ": printed line within the limit: " + lineBytes); + Map parsed = messageOf(line); + assertEquals(Boolean.TRUE, parsed.get("truncated"), "record says it was truncated"); + assertEquals(Boolean.TRUE, parsed.get("droppedInput")); + assertEquals(record.executionArn(), parsed.get("executionArn")); + } + } + } + + /** + * A limit smaller than the envelope overhead cannot be met: the execution error and identity fields are never + * dropped. The line is still printed whole and parses, with everything droppable dropped and the markers set. + */ + @Test + @SuppressWarnings("unchecked") + void limitBelowTheEnvelopeOverheadPrintsTheMinimalLine() throws Exception { + var exporter = new LambdaLogExporter(50); + var record = RecordFactory.withInput(RecordFactory.sampleWithError("boom", "io"), "in"); + String line = shapeAndExport(exporter, record); + assertTrue(line.getBytes(StandardCharsets.UTF_8).length > 50, "over the limit, by design"); + + Map parsed = messageOf(line); + assertEquals(Boolean.TRUE, parsed.get("truncated")); + assertEquals(Boolean.TRUE, parsed.get("droppedCauses")); + assertEquals(1, parsed.get("droppedOperations")); + assertEquals(Boolean.TRUE, parsed.get("droppedInput")); + assertEquals(Map.of(), parsed.get("operationsByName")); + assertEquals("boom", ((Map) parsed.get("error")).get("message"), "error kept"); + assertEquals(record.executionArn(), parsed.get("executionArn")); + } + + /** Control characters and non-ASCII text round-trip through the two levels of escaping. */ + @Test + @SuppressWarnings("unchecked") + void controlAndNonAsciiCharactersRoundTrip() throws Exception { + String text = "tab\t nl\n ctl\u0001\u001F é 日本 \uD83D\uDE00 \u2028"; + var record = RecordFactory.sampleWithInput(text); + String line = exportAndCapture(record).strip(); + assertEquals(1, exportAndCapture(record).lines().count()); + assertFalse(line.contains("\u0001"), "control characters are escaped"); + assertEquals(text, messageOf(line).get("input")); + } + + /** + * {@code renderedSize} is computed without building the envelope, and equals the UTF-8 size of the printed line for + * content that is escaped differently at each level: quotes, backslashes, short and long control-character escapes, + * DEL, two- and three-byte UTF-8, line and paragraph separators, emoji (surrogate pairs) and a lone surrogate. + */ + @Test + void renderedSizeEqualsThePrintedLineSize() { + String tricky = + "q\" b\\ \b\t\n\f\r \u0000\u0001\u001F\u007F é ß 日本 \u2028\u2029 \uD83D\uDE00\uD83E\uDD16 lone\uD800x"; + var exporter = new LambdaLogExporter(); + var record = + RecordFactory.withInput(RecordFactory.sampleWithError(tricky, tricky), Map.of(tricky, List.of(tricky))); + RecordFactory.withOperation( + record, + new software.amazon.lambda.durable.telemetry.OperationRecord() + .id("op-2") + .name(tricky) + .type("STEP") + .status("FAILED") + .result(tricky) + .error(new software.amazon.lambda.durable.telemetry.ErrorInfo(tricky, tricky))); + for (var r : List.of(RecordFactory.sample(), record)) { + String line = exportAndCapture(exporter, r).strip(); + assertEquals(line.getBytes(StandardCharsets.UTF_8).length, exporter.renderedSize(r), line); + assertEquals(Json.byteSize(exporter.render(r)), exporter.renderedSize(r)); + } + } + + /** The escaped size of every single character, and of mixed text, matches the serializer. */ + @Test + void escapedSizeMatchesTheSerializerForEveryChar() { + for (int c = 0; c <= 0xFFFF; c++) { + String s = String.valueOf((char) c); + assertEquals( + Json.stringify(s).getBytes(StandardCharsets.UTF_8).length - 2, + Json.escapedUtf8Size(s), + "U+" + Integer.toHexString(c)); + } + String pairs = "\uD83D\uDE00a\uDE00\uD83Db\uD83D"; + assertEquals(Json.stringify(pairs).getBytes(StandardCharsets.UTF_8).length - 2, Json.escapedUtf8Size(pairs)); + } +}