diff --git a/telemetry-exporter/README.md b/telemetry-exporter/README.md index 9f92527d2..6b17d7b23 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,23 @@ 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. + +`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 db791a654..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,12 +9,14 @@ 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; 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 @@ -42,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 @@ -108,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. --- /** @@ -167,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); @@ -347,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, @@ -356,7 +402,7 @@ private DurableExecutionTelemetryRecord buildRecord( Throwable error) { ContentConfig content = settings.content; DurableExecutionTelemetryRecord record = new DurableExecutionTelemetryRecord(); - record.emittedAt = Instant.now().toString(); + record.emittedAt = nextEmittedAt(); record.executionArn = executionArn; record.executionName = DurableExecutionTelemetry.emptyToNull(arn.executionName()); record.functionName = arn.functionName(); @@ -364,9 +410,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 +475,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/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(); + } + } +} 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");