From 29b35f4451630954d397036db9252db725ae28be Mon Sep 17 00:00:00 2001 From: Pooya Paridel Date: Fri, 23 Oct 2026 12:10:20 -0700 Subject: [PATCH 1/2] fix(telemetry): write fixed-width UTC timestamps Record timestamps were written with Instant.toString(). That method writes 0, 3, 6 or 9 fraction digits, depending on the value. So one record could hold startTime with 3 digits and endTime with 9. Destinations such as DynamoDB and CloudWatch Logs Insights compare these values as text. Text order equals time order only when every value has the same width. So mixed widths sort and filter incorrectly. emittedAt, startTime and endTime, on the record and on each operation, now use YYYY-MM-DDTHH:MM:SS.sssZ. This is the format of JavaScript's Date.prototype.toISOString(). Digits below the millisecond are dropped. The new internal Timestamps.format writes it. LambdaLogExporter uses the same method for the envelope timestamp, so its output is unchanged. --- telemetry-exporter/README.md | 17 ++- .../durable/telemetry/TelemetryPlugin.java | 14 +-- .../exporters/LambdaLogExporter.java | 9 +- .../telemetry/internal/Timestamps.java | 31 ++++++ .../telemetry/CloudWatchLogsExporterTest.java | 2 +- .../DurableExecutionTelemetryHookTest.java | 2 +- .../telemetry/LambdaLogLineSizeTest.java | 9 +- .../durable/telemetry/RecordFactory.java | 4 +- .../telemetry/RecordTimestampFormatTest.java | 104 ++++++++++++++++++ .../durable/telemetry/S3ExporterTest.java | 2 +- .../telemetry/StateCleanupLifecycleTest.java | 5 +- 11 files changed, 173 insertions(+), 26 deletions(-) create mode 100644 telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/internal/Timestamps.java create mode 100644 telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/RecordTimestampFormatTest.java diff --git a/telemetry-exporter/README.md b/telemetry-exporter/README.md index 9f92527d2..b6bf263e6 100644 --- a/telemetry-exporter/README.md +++ b/telemetry-exporter/README.md @@ -65,8 +65,8 @@ Writes one line per record to `System.out`, in the Lambda JSON log envelope: {"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 +`timestamp` has the same format as the record timestamps (see [Timestamps](#timestamps)). 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 @@ -202,6 +202,19 @@ 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`). +## Timestamps + +Every timestamp in a record has one format: `YYYY-MM-DDTHH:MM:SS.sssZ`, for example `2026-10-23T18:42:44.078Z`. +This covers `emittedAt`, `startTime` and `endTime` on the record, and `startTime` and `endTime` on each operation. + +- The time is in UTC, and the value always ends with the literal `Z`. +- The fraction always has exactly three digits. A whole second is written as `.000`. +- Digits below the millisecond are dropped, not rounded. So `.078757096` is written as `.078`. + +This is the format of JavaScript's `Date.prototype.toISOString()`. Every value has the same width. So comparing two +values as text gives the same order as comparing them as times. This matters for destinations that compare strings, +such as a DynamoDB sort key or a CloudWatch Logs Insights filter. + ## Error names and causes Every error in a record (the execution `error`, each `operations[].error`, and each diff --git a/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/TelemetryPlugin.java b/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/TelemetryPlugin.java index db791a654..b0f0dbfb6 100644 --- a/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/TelemetryPlugin.java +++ b/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/TelemetryPlugin.java @@ -15,6 +15,7 @@ import software.amazon.lambda.durable.plugin.OperationChangeInfo; import software.amazon.lambda.durable.plugin.OperationChangeItemInfo; import software.amazon.lambda.durable.telemetry.internal.FatalErrors; +import software.amazon.lambda.durable.telemetry.internal.Timestamps; /** * The Durable Execution Telemetry plugin instance for one Lambda invocation: both the state that invocation's records @@ -356,7 +357,7 @@ private DurableExecutionTelemetryRecord buildRecord( Throwable error) { ContentConfig content = settings.content; DurableExecutionTelemetryRecord record = new DurableExecutionTelemetryRecord(); - record.emittedAt = Instant.now().toString(); + record.emittedAt = Timestamps.format(Instant.now()); record.executionArn = executionArn; record.executionName = DurableExecutionTelemetry.emptyToNull(arn.executionName()); record.functionName = arn.functionName(); @@ -364,9 +365,9 @@ private DurableExecutionTelemetryRecord buildRecord( record.region = arn.region(); record.accountId = arn.accountId(); record.status = status; - record.startTime = startTime != null ? startTime.toString() : null; + record.startTime = startTime != null ? Timestamps.format(startTime) : null; if (endTime != null) { - record.endTime = endTime.toString(); + record.endTime = Timestamps.format(endTime); if (startTime != null) { record.durationMs = endTime.toEpochMilli() - startTime.toEpochMilli(); } @@ -429,11 +430,8 @@ private List buildOperationRecords(Map envelope(String recordJson, Instant now) { Map line = new LinkedHashMap<>(); - line.put("timestamp", TIMESTAMP.format(now)); + line.put("timestamp", Timestamps.format(now)); line.put("level", "INFO"); line.put("message", recordJson); return line; diff --git a/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/internal/Timestamps.java b/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/internal/Timestamps.java new file mode 100644 index 000000000..19031af77 --- /dev/null +++ b/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/internal/Timestamps.java @@ -0,0 +1,31 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.telemetry.internal; + +import java.time.Instant; +import java.time.ZoneOffset; +import java.time.format.DateTimeFormatter; + +/** Internal timestamp format for Telemetry records and the Lambda log envelope. */ +public final class Timestamps { + /** UTC ISO-8601 with exactly three fraction digits and a literal {@code Z}. */ + private static final DateTimeFormatter FORMAT = + DateTimeFormatter.ofPattern("uuuu-MM-dd'T'HH:mm:ss.SSS'Z'").withZone(ZoneOffset.UTC); + + private Timestamps() {} + + /** + * Formats {@code instant} as {@code YYYY-MM-DDTHH:MM:SS.sssZ}, the format of JavaScript's + * {@code Date.prototype.toISOString()}. + * + *

The fraction always has three digits. A zero fraction is written as {@code .000}. Digits below the millisecond + * are dropped, not rounded, so {@code .078757096} becomes {@code .078} and {@code .999999999} becomes {@code .999}. + * + *

Destinations such as DynamoDB and CloudWatch Logs Insights compare these values as text. Text order equals + * time order only when every value has the same width. {@link Instant#toString()} writes 0, 3, 6 or 9 fraction + * digits. So every record timestamp goes through this method instead. + */ + public static String format(Instant instant) { + return FORMAT.format(instant); + } +} diff --git a/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/CloudWatchLogsExporterTest.java b/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/CloudWatchLogsExporterTest.java index 7f09af0a5..cad3ca0ef 100644 --- a/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/CloudWatchLogsExporterTest.java +++ b/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/CloudWatchLogsExporterTest.java @@ -39,7 +39,7 @@ private DurableExecutionTelemetryRecord sampleRecord() { DurableExecutionTelemetryRecord r = new DurableExecutionTelemetryRecord(); r.executionArn = "arn:aws:lambda:us-west-2:1:function:f:$LATEST/durable-execution/exec-1/invocation-1"; r.status = "SUCCEEDED"; - r.startTime = "2026-08-05T00:00:00Z"; + r.startTime = "2026-08-05T00:00:00.000Z"; r.addOperation(new OperationRecord() .id("op-1") .name("greet") diff --git a/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/DurableExecutionTelemetryHookTest.java b/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/DurableExecutionTelemetryHookTest.java index dc2195592..aea43013b 100644 --- a/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/DurableExecutionTelemetryHookTest.java +++ b/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/DurableExecutionTelemetryHookTest.java @@ -265,7 +265,7 @@ void suspendResumeKeepsStableStartTimeAndLeavesNoRetainedState() { assertEquals(4, exporter.records.size()); String startTime = exporter.records.get(0).startTime(); assertTrue(exporter.records.stream().allMatch(r -> startTime.equals(r.startTime()))); - assertEquals(START.toString(), startTime); + assertEquals("2026-08-05T00:00:00.000Z", startTime); assertFalse(Executions.outstanding(first), "the suspended invocation left the scheduler owing nothing"); assertFalse(Executions.outstanding(resumed), "nor did the resumed one"); } 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 index 208e764c4..54dfa8267 100644 --- 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 @@ -16,6 +16,7 @@ import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; import software.amazon.lambda.durable.telemetry.exporters.LambdaLogExporter; +import software.amazon.lambda.durable.telemetry.internal.Timestamps; /** Size checks for the {@code LambdaLogExporter} line through the plugin's export path. */ class LambdaLogLineSizeTest { @@ -58,7 +59,7 @@ private static DurableExecutionTelemetryRecord withResults(int count, Object res .name("op-" + i) .type("STEP") .status("SUCCEEDED") - .startTime(start.plusSeconds(i).toString()) + .startTime(Timestamps.format(start.plusSeconds(i))) .result(result)); } return r; @@ -111,7 +112,7 @@ void measuredPassMatchesMeasuringEveryStep() { .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()) + .startTime(i % 11 == 0 ? null : Timestamps.format(start.plusSeconds(i))) .durationMs((long) i * 997) .attempt(i % 3 + 1) .result(i % 6 == 0 ? Integer.valueOf(i) : "\"\\é😀".repeat(i % 50)); @@ -166,8 +167,8 @@ private static DurableExecutionTelemetryRecord worstCaseRecord(String name, Stri .name("op-" + i) .type("STEP") .status("FAILED") - .startTime( - Instant.parse("2026-07-15T11:00:00Z").plusSeconds(i).toString()) + .startTime(Timestamps.format( + Instant.parse("2026-07-15T11:00:00Z").plusSeconds(i))) .result(message) .error(new ErrorInfo(name, message, new ErrorInfo(name, message)))); } 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 8d270ed46..1a20e2de6 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 @@ -2,6 +2,8 @@ // SPDX-License-Identifier: Apache-2.0 package software.amazon.lambda.durable.telemetry; +import software.amazon.lambda.durable.telemetry.internal.Timestamps; + /** Builds a minimal, fully populated record for exporter tests outside this package. */ public final class RecordFactory { private RecordFactory() {} @@ -54,7 +56,7 @@ public static DurableExecutionTelemetryRecord sampleWithFailedOperations(int cou .name("op-" + i) .type("STEP") .status("FAILED") - .startTime(start.plusSeconds(i).toString()) + .startTime(Timestamps.format(start.plusSeconds(i))) .error(new ErrorInfo("java.lang.RuntimeException", message, new ErrorInfo("Throwable", "c")))); } return r; diff --git a/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/RecordTimestampFormatTest.java b/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/RecordTimestampFormatTest.java new file mode 100644 index 000000000..f87b0b5e4 --- /dev/null +++ b/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/RecordTimestampFormatTest.java @@ -0,0 +1,104 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.telemetry; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.time.Instant; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.regex.Pattern; +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.services.lambda.model.OperationStatus; +import software.amazon.lambda.durable.plugin.InvocationEndInfo; +import software.amazon.lambda.durable.plugin.InvocationInfo; +import software.amazon.lambda.durable.plugin.InvocationStatus; +import software.amazon.lambda.durable.plugin.OperationChangeItemInfo; +import software.amazon.lambda.durable.telemetry.internal.Timestamps; + +/** + * Every record timestamp has one fixed-width UTC format, {@code YYYY-MM-DDTHH:MM:SS.sssZ}. Destinations compare these + * values as text. Text order equals time order only when every value has the same width. + */ +class RecordTimestampFormatTest { + + private static final Pattern FORMAT = Pattern.compile("^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{3}Z$"); + + private static final String ARN = + "arn:aws:lambda:us-west-2:1:function:f:$LATEST/durable-execution/exec-1/invocation-1"; + + /** Zero nanos: {@code Instant.toString()} writes no fraction for this value. */ + private static final Instant ZERO_NANOS = Instant.parse("2026-10-23T18:42:40Z"); + + /** Nine fraction digits: {@code Instant.toString()} writes all nine. */ + private static final Instant NINE_DIGITS = Instant.parse("2026-10-23T18:42:44.078757096Z"); + + /** The last nanosecond of a second: truncation must give {@code .999}, not round up to the next second. */ + private static final Instant LAST_NANO = Instant.parse("2026-10-23T18:42:46.999999999Z"); + + @Test + void formatWritesThreeFractionDigitsAndTruncates() { + assertEquals("2026-10-23T18:42:40.000Z", Timestamps.format(ZERO_NANOS)); + assertEquals("2026-10-23T18:42:44.078Z", Timestamps.format(NINE_DIGITS)); + assertEquals("2026-10-23T18:42:46.999Z", Timestamps.format(LAST_NANO)); + assertEquals("2026-10-23T18:42:44.500Z", Timestamps.format(Instant.parse("2026-10-23T18:42:44.5Z"))); + assertEquals("1970-01-01T00:00:00.000Z", Timestamps.format(Instant.EPOCH)); + } + + @Test + void fixedWidthValuesSortAsTextInTimeOrder() { + // With Instant.toString(), "...40Z" sorts after "...40.500Z" as text, because 'Z' > '.'. + String whole = Timestamps.format(ZERO_NANOS); + String half = Timestamps.format(ZERO_NANOS.plusMillis(500)); + assertTrue(whole.compareTo(half) < 0, whole + " must sort before " + half); + } + + @Test + void everyRecordAndOperationTimestampHasTheFixedFormat() { + var records = new CopyOnWriteArrayList(); + var factory = DurableExecutionTelemetry.durableExecutionTelemetry(DurableExecutionTelemetryConfig.builder() + .addExporter(records::add) + .build()); + Map ops = new LinkedHashMap<>(); + ops.put("op-1", op("op-1", "first", ZERO_NANOS, NINE_DIGITS)); + ops.put("op-2", op("op-2", "second", NINE_DIGITS, LAST_NANO)); + ops.put("op-3", op("op-3", "running", LAST_NANO, null)); + + var plugin = Executions.started(factory, new InvocationInfo("req", ARN, true, ZERO_NANOS, "in", ops, Map.of())); + plugin.onInvocationEnd(new InvocationEndInfo( + "req", ARN, true, ZERO_NANOS, ops, InvocationStatus.SUCCEEDED, null, "in", "out")); + + assertEquals(1, records.size()); + var record = records.get(0); + assertEquals("2026-10-23T18:42:40.000Z", record.startTime); + assertFixed("emittedAt", record.emittedAt); + assertFixed("endTime", record.endTime); + + List operations = record.operations(); + assertEquals(3, operations.size()); + assertEquals("2026-10-23T18:42:40.000Z", operations.get(0).startTime()); + assertEquals("2026-10-23T18:42:44.078Z", operations.get(0).endTime()); + assertEquals("2026-10-23T18:42:44.078Z", operations.get(1).startTime()); + assertEquals("2026-10-23T18:42:46.999Z", operations.get(1).endTime()); + assertEquals("2026-10-23T18:42:46.999Z", operations.get(2).startTime()); + assertNull(operations.get(2).endTime(), "a missing timestamp stays absent"); + for (var op : operations) { + assertFixed(op.name() + ".startTime", op.startTime()); + } + } + + private static OperationChangeItemInfo op(String id, String name, Instant start, Instant end) { + return new OperationChangeItemInfo( + id, name, "STEP", "Step", null, start, end, OperationStatus.SUCCEEDED, 1, false, null, null); + } + + private static void assertFixed(String field, String value) { + assertNotNull(value, field); + assertTrue(FORMAT.matcher(value).matches(), field + " = " + value); + } +} diff --git a/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/S3ExporterTest.java b/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/S3ExporterTest.java index e3fd8405e..c35421a8d 100644 --- a/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/S3ExporterTest.java +++ b/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/S3ExporterTest.java @@ -27,7 +27,7 @@ private DurableExecutionTelemetryRecord sampleRecord() { r.executionName = "exec-1"; r.functionName = "f"; r.status = "SUCCEEDED"; - r.startTime = "2026-08-05T00:00:00Z"; + r.startTime = "2026-08-05T00:00:00.000Z"; r.addOperation(new OperationRecord() .id("op-1") .name("greet") diff --git a/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/StateCleanupLifecycleTest.java b/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/StateCleanupLifecycleTest.java index 55693dade..fd935c126 100644 --- a/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/StateCleanupLifecycleTest.java +++ b/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/StateCleanupLifecycleTest.java @@ -270,7 +270,10 @@ void resumeReSeedsStableStartTimeAndInput() { var terminal = exporter.records.get(exporter.records.size() - 1); assertEquals("SUCCEEDED", terminal.status()); - assertEquals(START.toString(), terminal.startTime(), "stable start time recreated across the suspend boundary"); + assertEquals( + "2026-08-05T00:00:00.000Z", + terminal.startTime(), + "stable start time recreated across the suspend boundary"); assertEquals("alpha", terminal.input, "input re-seeded from resume onInvocationStart"); assertFalse(Executions.outstanding(resumed), "the terminal end leaves the scheduler owing nothing"); assertFalse(resumed.scheduler.retains(resumed), "the environment holds no reference to the resumed invocation"); From 28fa366b2286415947396ad277e13b3962239f7d Mon Sep 17 00:00:00 2001 From: Pooya Paridel Date: Fri, 23 Oct 2026 13:14:43 -0700 Subject: [PATCH 2/2] fix(telemetry): keep emittedAt unique per invocation emittedAt has millisecond precision. With ON_CHANGE, one invocation can build two records in the same millisecond. One example is the last RUNNING record and the terminal record. So both records got the same emittedAt. SQSExporter builds the FIFO deduplication id from executionArn and emittedAt. DynamoDBExporter uses emittedAt as its default sort key. So SQS dropped the second message, and DynamoDB overwrote the first item. TelemetryPlugin now keeps the last emittedAt it assigned, in epoch milliseconds. If the clock is past that value, the clock is used. Otherwise the last value plus 1 ms is used. So emittedAt strictly increases within one plugin instance. One instance serves one invocation. This is the rule of aws/aws-durable-execution-sdk-js#1071. Hook threads can build records at the same time. No lock serializes the builds. So the last value is an AtomicLong, and the read and the update are one updateAndGet(prev -> Math.max(prev + 1, now)). --- telemetry-exporter/README.md | 4 + .../durable/telemetry/TelemetryPlugin.java | 47 +++++- .../telemetry/EmittedAtOrderingTest.java | 152 ++++++++++++++++++ 3 files changed, 202 insertions(+), 1 deletion(-) create mode 100644 telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/EmittedAtOrderingTest.java diff --git a/telemetry-exporter/README.md b/telemetry-exporter/README.md index b6bf263e6..6b17d7b23 100644 --- a/telemetry-exporter/README.md +++ b/telemetry-exporter/README.md @@ -215,6 +215,10 @@ This is the format of JavaScript's `Date.prototype.toISOString()`. Every value h values as text gives the same order as comparing them as times. This matters for destinations that compare strings, such as a DynamoDB sort key or a CloudWatch Logs Insights filter. +`emittedAt` strictly increases within one invocation: two records built in the same millisecond get values 1 ms +apart, because the SQS FIFO deduplication id and the default DynamoDB sort key use `executionArn` and `emittedAt`, so +equal values would drop or overwrite a record. + ## Error names and causes Every error in a record (the execution `error`, each `operations[].error`, and each diff --git a/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/TelemetryPlugin.java b/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/TelemetryPlugin.java index b0f0dbfb6..7d2b40434 100644 --- a/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/TelemetryPlugin.java +++ b/telemetry-exporter/src/main/java/software/amazon/lambda/durable/telemetry/TelemetryPlugin.java @@ -9,6 +9,7 @@ import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicLong; +import java.util.function.LongSupplier; import software.amazon.lambda.durable.plugin.DurableExecutionPlugin; import software.amazon.lambda.durable.plugin.InvocationEndInfo; import software.amazon.lambda.durable.plugin.InvocationInfo; @@ -43,6 +44,8 @@ *

  • The build revision — {@link #buildRevision} — counts the record builds this invocation has started, so * that a build which was overtaken can be recognized at hand-off time and its record dropped. Atomic rather than * {@code volatile}, because the case it exists for is two builds running at once. See the field. + *
  • The last emittedAt — {@link #lastEmittedMs} — is atomic for the same reason: two builds can run at + * once. See {@link #nextEmittedAt()}. *
  • Scheduling state — {@link #record}, {@link #settled}, {@link #exporting}, {@link #drainWaiters} and * {@link #closed} — is shared with the export pump and guarded by the monitor of {@link #scheduler}. One monitor * for the whole environment, not one per invocation, so the {@code closed} check and the hand-off of a record are @@ -109,6 +112,17 @@ final class TelemetryPlugin implements DurableExecutionPlugin { */ private final AtomicLong buildRevision = new AtomicLong(); + // --- emittedAt ordering. --- + + /** Current time in epoch milliseconds. {@code System::currentTimeMillis} outside tests. */ + private final LongSupplier clockMillis; + + /** + * The last {@code emittedAt} this instance assigned, in epoch milliseconds. {@link Long#MIN_VALUE} before the first + * record. See {@link #nextEmittedAt()}. + */ + private final AtomicLong lastEmittedMs = new AtomicLong(Long.MIN_VALUE); + // --- Scheduling state: guarded by the scheduler's monitor. --- /** @@ -168,7 +182,14 @@ final class TelemetryPlugin implements DurableExecutionPlugin { * contains a hook failure, by skipping this plugin for the invocation */ TelemetryPlugin(TelemetrySettings settings, ExportScheduler scheduler, InvocationInfo info) { + this(settings, scheduler, info, System::currentTimeMillis); + } + + /** Test seam: the same instance with a supplied clock for {@code emittedAt}. */ + TelemetryPlugin( + TelemetrySettings settings, ExportScheduler scheduler, InvocationInfo info, LongSupplier clockMillis) { this.settings = settings; + this.clockMillis = clockMillis; this.scheduler = scheduler; this.executionArn = info.durableExecutionArn(); this.arn = ArnParser.parse(executionArn); @@ -348,6 +369,30 @@ boolean isNewestBuild(long revision) { return buildRevision.get() == revision; } + /** + * Returns the {@code emittedAt} value for the next record this instance builds. + * + *
      + *
    1. {@code emittedAt} has millisecond precision. + *
    2. With {@code ON_CHANGE}, this instance can build two records in the same millisecond. One example is a + * RUNNING record and the terminal record. + *
    3. The SQS FIFO deduplication id and the default DynamoDB sort key use {@code executionArn} and + * {@code emittedAt}. + *
    4. So two records with equal values collide. SQS drops the second message. DynamoDB overwrites the first item. + *
    5. So {@code emittedAt} strictly increases within this instance. This instance serves one invocation. + *
    + * + *

    The rule: if the clock is past the last value, the clock is used. Otherwise the last value plus 1 ms is used. + * + *

    Hook threads can build records at the same time, and no lock serializes the builds. So the read and the update + * of the last value are one atomic {@code updateAndGet}. {@code Math.max(prev + 1, now)} is the rule above. + */ + String nextEmittedAt() { + long nowMs = clockMillis.getAsLong(); + long ms = lastEmittedMs.updateAndGet(prev -> Math.max(prev + 1, nowMs)); + return Timestamps.format(Instant.ofEpochMilli(ms)); + } + private DurableExecutionTelemetryRecord buildRecord( String status, Map operations, @@ -357,7 +402,7 @@ private DurableExecutionTelemetryRecord buildRecord( Throwable error) { ContentConfig content = settings.content; DurableExecutionTelemetryRecord record = new DurableExecutionTelemetryRecord(); - record.emittedAt = Timestamps.format(Instant.now()); + record.emittedAt = nextEmittedAt(); record.executionArn = executionArn; record.executionName = DurableExecutionTelemetry.emptyToNull(arn.executionName()); record.functionName = arn.functionName(); diff --git a/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/EmittedAtOrderingTest.java b/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/EmittedAtOrderingTest.java new file mode 100644 index 000000000..49799a05d --- /dev/null +++ b/telemetry-exporter/src/test/java/software/amazon/lambda/durable/telemetry/EmittedAtOrderingTest.java @@ -0,0 +1,152 @@ +// 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 java.time.Instant; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import org.junit.jupiter.api.Test; +import software.amazon.lambda.durable.plugin.InvocationEndInfo; +import software.amazon.lambda.durable.plugin.InvocationInfo; +import software.amazon.lambda.durable.plugin.InvocationStatus; +import software.amazon.lambda.durable.plugin.OperationChangeInfo; + +/** + * {@code emittedAt} strictly increases within one plugin instance. + * + *

    {@code emittedAt} has millisecond precision. With {@code ON_CHANGE}, two records can be built in the same + * millisecond. The SQS FIFO deduplication id and the default DynamoDB sort key use {@code executionArn} and + * {@code emittedAt}. So two equal values drop or overwrite a record. + */ +class EmittedAtOrderingTest { + + private static final String ARN = + "arn:aws:lambda:us-west-2:111122223333:function:f:$LATEST/durable-execution/exec-1/invocation-1"; + private static final Instant START = Instant.parse("2026-10-23T18:42:40Z"); + private static final long T123 = Instant.parse("2026-10-23T18:42:44.123Z").toEpochMilli(); + + private static final class Fixture { + final List exported = Collections.synchronizedList(new ArrayList<>()); + final AtomicLong clock = new AtomicLong(T123); + final TelemetryPlugin plugin; + + Fixture() { + var config = DurableExecutionTelemetryConfig.builder() + .emitMode(DurableExecutionTelemetryConfig.EmitMode.ON_CHANGE) + .addExporter(exported::add) + .build(); + // Runnable::run runs the pump inside each hand-off. So every record reaches the exporter before the next + // record is built, and the scheduler coalesces none of them. + var scheduler = new ExportScheduler( + config.exporters(), (record, target) -> target.export(record), t -> {}, Runnable::run); + plugin = new TelemetryPlugin( + new TelemetrySettings(config), + scheduler, + new InvocationInfo("req", ARN, true, START, null, Map.of(), Map.of()), + clock::get); + } + + void start() { + plugin.onInvocationStart(new InvocationInfo("req", ARN, true, START, "in", Map.of(), Map.of())); + } + + void change() { + plugin.onOperationChange(new OperationChangeInfo("req", ARN, Map.of(), Map.of())); + } + + void end() { + plugin.onInvocationEnd(new InvocationEndInfo( + "req", ARN, true, START, Map.of(), InvocationStatus.SUCCEEDED, null, "in", "out")); + } + + List statusAndEmittedAt() { + List out = new ArrayList<>(); + synchronized (exported) { + for (var record : exported) { + out.add(record.status() + " " + record.emittedAt); + } + } + return out; + } + } + + @Test + void recordsBuiltInOneMillisecondGetIncreasingValues() { + var fixture = new Fixture(); + + fixture.start(); + fixture.change(); + fixture.end(); + + assertEquals( + List.of( + "RUNNING 2026-10-23T18:42:44.123Z", + "RUNNING 2026-10-23T18:42:44.124Z", + "SUCCEEDED 2026-10-23T18:42:44.125Z"), + fixture.statusAndEmittedAt()); + } + + @Test + void theClockIsUsedOnceItPassesTheLastValue() { + var fixture = new Fixture(); + + fixture.start(); + fixture.clock.set(T123 + 377); + fixture.change(); + // The clock moves backwards. The last value plus 1 ms is used. + fixture.clock.set(T123 + 100); + fixture.end(); + + assertEquals( + List.of( + "RUNNING 2026-10-23T18:42:44.123Z", + "RUNNING 2026-10-23T18:42:44.500Z", + "SUCCEEDED 2026-10-23T18:42:44.501Z"), + fixture.statusAndEmittedAt()); + } + + @Test + void concurrentCallsGetDistinctValues() throws Exception { + var fixture = new Fixture(); + int threads = 8; + int callsPerThread = 2_000; + var ready = new CountDownLatch(threads); + var go = new CountDownLatch(1); + ExecutorService pool = Executors.newFixedThreadPool(threads); + try { + List>> futures = new ArrayList<>(); + for (int t = 0; t < threads; t++) { + futures.add(pool.submit(() -> { + List values = new ArrayList<>(callsPerThread); + ready.countDown(); + go.await(); + for (int i = 0; i < callsPerThread; i++) { + values.add(fixture.plugin.nextEmittedAt()); + } + return values; + })); + } + ready.await(); + go.countDown(); + Set all = new HashSet<>(); + for (var future : futures) { + all.addAll(future.get(30, TimeUnit.SECONDS)); + } + assertEquals(threads * callsPerThread, all.size(), "every emittedAt must be distinct"); + } finally { + pool.shutdownNow(); + } + } +}