Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions otel-plugin/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -411,12 +411,17 @@ This telemetry decision does not remove the SDK's failure notification: a presen
remains a `DurableOperationException` in operation hooks and snapshots. An absent container retains its existing
null metadata behavior. Persisted errors, caller failures, replay, and user-function error handling are unchanged.

Operation exception events use the original `ErrorObject.errorType` when supplied, including types from another
language such as `RejectedError`. If the stored type is absent, the exception's Java type is used.
Workflow and Invocation spans report failure through their `ERROR` status and status description without
recording exception events. Operation and attempt spans carry the exception events.

Execution-view operation spans with a known, nonzero backend end timestamp record the exception event at that
same timestamp. This path uses the public `exception` event with standard `exception.type`, `exception.message`
(when present), and `exception.stacktrace` attributes. Java's standard stack-trace text and the provider's ordinary
event attribute count/value limits apply; native `recordException` resolver behavior is not preserved on this path.
Invocation-view events and operations without a backend end (including the SDK's epoch-zero clock sentinel) keep
native `recordException` behavior.
Invocation-view events and operations without a backend end (including the SDK's epoch-zero clock sentinel) use
native `recordException` with the stored exception type supplied as an attribute override.

When an Execution-view completion has no observed or backend start timestamp, its known end also supplies the start.
Existing earliest-known-start selection is unchanged and is not clamped to the backend end, so clock skew can still
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -450,9 +450,6 @@ private void finishInvocationEnd(InvocationEndInfo info) {
? info.executionError().getMessage()
: null;
workflowSpan.setStatus(StatusCode.ERROR, message);
if (info.executionError() != null) {
workflowSpan.recordException(info.executionError());
}
}
default -> workflowSpan.setStatus(StatusCode.OK); // SUCCEEDED
}
Expand Down Expand Up @@ -666,7 +663,7 @@ private boolean disabled() {
private void applyInvocationStatus(Span span, InvocationEndInfo info) {
// Invocation span status mapping:
// SUCCEEDED, PENDING -> OK
// FAILED -> ERROR (records the execution error when present)
// FAILED -> ERROR (exception events belong to operations and attempts)
// RETRYING -> UNSET (left unset)
// RETRYING is left UNSET on purpose: the plugin interface does not expose whether the invocation/workflow
// was STOPPED or TIMED_OUT versus retried for a transient error, so an ERROR status cannot be asserted
Expand All @@ -677,9 +674,6 @@ private void applyInvocationStatus(Span span, InvocationEndInfo info) {
var message =
info.executionError() != null ? info.executionError().getMessage() : null;
span.setStatus(StatusCode.ERROR, message);
if (info.executionError() != null) {
span.recordException(info.executionError());
}
}
case RETRYING -> {
// UNSET — see note above.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -407,7 +407,7 @@ private void finishInvocationEnd(InvocationEndInfo info) {

// Invocation span status mapping:
// SUCCEEDED, PENDING -> OK
// FAILED -> ERROR (records the execution error when present)
// FAILED -> ERROR (exception events belong to operations and attempts)
// RETRYING -> UNSET
// RETRYING is left UNSET on purpose: the plugin interface does not expose whether the invocation/workflow
// was STOPPED or TIMED_OUT versus retried for a transient error, so an ERROR status cannot be asserted
Expand All @@ -418,9 +418,6 @@ private void finishInvocationEnd(InvocationEndInfo info) {
var message =
info.executionError() != null ? info.executionError().getMessage() : null;
invocationSpan.setStatus(StatusCode.ERROR, message);
if (info.executionError() != null) {
invocationSpan.recordException(info.executionError());
}
}
case RETRYING -> {
// UNSET — see note above.
Expand All @@ -446,9 +443,6 @@ private void finishInvocationEnd(InvocationEndInfo info) {
? info.executionError().getMessage()
: null;
workflowSpan.setStatus(StatusCode.ERROR, message);
if (info.executionError() != null) {
workflowSpan.recordException(info.executionError());
}
}
default -> workflowSpan.setStatus(StatusCode.OK); // SUCCEEDED
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,7 @@ static void applyOperationStatus(Span span, OperationEndInfo info, Instant endTi
span.setStatus(StatusCode.ERROR, info.error().getMessage());
if (endTimestamp == null || endTimestamp.equals(Instant.EPOCH)) {
// The SDK treats epoch zero as its current-clock sentinel when ending a span.
span.recordException(info.error());
span.recordException(info.error(), operationExceptionAttributes(info.error()));
} else {
recordExceptionAt(span, info.error(), endTimestamp);
}
Expand All @@ -113,6 +113,16 @@ static void applyOperationStatus(Span span, OperationEndInfo info, Instant endTi
}
}

/** Checkpoint metadata carries the original type, which may belong to another language. */
private static Attributes operationExceptionAttributes(Throwable error) {
var attributes = Attributes.builder();
if (error instanceof DurableOperationException failure && failure.getErrorObject() != null) {
var type = failure.getErrorObject().errorType();
if (type != null) attributes.put("exception.type", type);
}
return attributes.build();
}

/** Timestamped public exception event; the provider applies its event attribute limits. */
private static void recordExceptionAt(Span span, Throwable error, Instant timestamp) {
if (!span.isRecording()) return;
Expand All @@ -126,6 +136,7 @@ private static void recordExceptionAt(Span span, Throwable error, Instant timest
error.printStackTrace(writer);
}
attributes.put("exception.stacktrace", stacktrace.toString());
attributes.putAll(operationExceptionAttributes(error));
span.addEvent("exception", attributes.build(), timestamp);
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -155,9 +155,7 @@ public void onOperationEnd(OperationEndInfo info) {
assertEquals(1, span.getEvents().size());
var diagnostic = span.getEvents().get(0);
var failure = ends.get(0).error();
assertEquals(
failure.getClass().getCanonicalName(),
diagnostic.getAttributes().get(AttributeKey.stringKey("exception.type")));
assertEquals(details.errorType(), diagnostic.getAttributes().get(AttributeKey.stringKey("exception.type")));
assertEquals(
failure.getMessage(), diagnostic.getAttributes().get(AttributeKey.stringKey("exception.message")));
assertTrue(diagnostic
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,141 @@
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package software.amazon.lambda.durable.otel;

import static org.junit.jupiter.api.Assertions.*;
import static software.amazon.lambda.durable.otel.SpanAttributes.*;

import io.opentelemetry.api.common.AttributeKey;
import io.opentelemetry.api.trace.StatusCode;
import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter;
import io.opentelemetry.sdk.trace.SdkTracerProvider;
import io.opentelemetry.sdk.trace.data.SpanData;
import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import software.amazon.awssdk.services.lambda.model.ErrorObject;
import software.amazon.lambda.durable.DurableConfig;
import software.amazon.lambda.durable.DurableContext;
import software.amazon.lambda.durable.model.ExecutionStatus;
import software.amazon.lambda.durable.plugin.DurableExecutionPluginFactory;
import software.amazon.lambda.durable.testing.LocalDurableTestRunner;

@Timeout(20)
class ExceptionAttributionIntegrationTest {
private static final String NAME = "failed-operation";
private static final String MESSAGE = "Intentional failure";
private static final AttributeKey<String> TYPE = AttributeKey.stringKey("exception.type");

@AfterEach
void clearTracingState() {
DeterministicIdGenerator.clearSharedStateForTest();
DurableSamplingDecision.clearSharedStateForTest();
OtelPluginAutoConfigurationState.resetInstalledForTest();
}

static Stream<Arguments> scenarios() {
return Stream.of(false, true)
.flatMap(execution -> Stream.of("child", "condition", "callback", "invoke", "handler")
.map(scenario -> Arguments.of(execution, scenario)));
}

@ParameterizedTest
@MethodSource("scenarios")
void failuresRecordOriginalTypeOnlyOnOperationsAndAttempts(boolean executionView, String scenario) {
try (var exporter = InMemorySpanExporter.create()) {
var calls = new AtomicInteger();
var runner = LocalDurableTestRunner.create(
String.class, (input, context) -> fail(context, scenario, calls), config(executionView, exporter));
completeBackendFailure(runner, scenario);
var result = runner.run("input");
assertEquals(ExecutionStatus.FAILED, result.getStatus());
assertAttribution(exporter.getFinishedSpanItems(), scenario);
var replay = runner.run("input");
assertEquals(ExecutionStatus.FAILED, replay.getStatus());
assertEquals(result.getError(), replay.getError());
if (!scenario.equals("handler")) assertEquals(scenario.equals("invoke") ? 0 : 1, calls.get());
}
}

private static void completeBackendFailure(LocalDurableTestRunner<String, String> runner, String scenario) {
if (!scenario.equals("callback") && !scenario.equals("invoke")) return;
assertEquals(ExecutionStatus.PENDING, runner.run("input").getStatus());
var details = ErrorObject.builder()
.errorType(scenario.equals("callback") ? "RejectedError" : RuntimeException.class.getName())
.errorMessage(MESSAGE)
.build();
if (scenario.equals("callback")) runner.failCallback(runner.getCallbackId(NAME + "-callback"), details);
else runner.failChainedInvoke(NAME, details);
}

private static String fail(DurableContext context, String scenario, AtomicInteger calls) {
return switch (scenario) {
case "child" -> context.runInChildContext(NAME, String.class, child -> throwFailure(calls));
case "condition" -> context.waitForCondition(NAME, String.class, (state, step) -> throwFailure(calls));
case "callback" -> context.waitForCallback(NAME, String.class, (id, step) -> calls.incrementAndGet());
case "invoke" -> context.invoke(NAME, "target-function", "input", String.class);
case "handler" -> throwFailure(calls);
default -> throw new IllegalArgumentException(scenario);
};
}

private static <T> T throwFailure(AtomicInteger calls) {
calls.incrementAndGet();
throw new RuntimeException(MESSAGE);
}

private static DurableConfig config(boolean execution, InMemorySpanExporter exporter) {
var provider = SdkTracerProvider.builder().addSpanProcessor(SimpleSpanProcessor.create(exporter));
var options = OtelPluginConfig.builder()
.enableMdc(false)
.contextExtractor(() -> null)
.build();
DurableExecutionPluginFactory factory = execution
? ExecutionOtelPlugin.factory(provider, options)
: InvocationOtelPlugin.factory(provider, options);
return DurableConfig.builder().withPlugins(factory).build();
}

private static void assertAttribution(List<SpanData> spans, String scenario) {
var failedOperations = spans.stream()
.filter(span -> span.getAttributes().get(DURABLE_OPERATION_ID) != null)
.filter(span -> span.getStatus().getStatusCode() == StatusCode.ERROR)
.toList();
int expected = switch (scenario) {
case "handler" -> 0;
case "condition", "callback" -> 2;
default -> 1;
};
assertEquals(expected, failedOperations.size());
for (var operation : failedOperations) {
assertEquals(1, operation.getEvents().size(), operation.getName());
var event = operation.getEvents().get(0);
assertEquals("exception", event.getName());
assertEquals(
scenario.equals("callback") ? "RejectedError" : RuntimeException.class.getName(),
event.getAttributes().get(TYPE),
operation.getName());
assertEquals(MESSAGE, event.getAttributes().get(AttributeKey.stringKey("exception.message")));
}
assertRootStatus(spans);
}

private static void assertRootStatus(List<SpanData> spans) {
var failedRoots = spans.stream()
.filter(span ->
span.getName().equals("Workflow") || span.getName().equals("Invocation"))
.filter(span -> span.getStatus().getStatusCode() == StatusCode.ERROR)
.toList();
assertEquals(2, failedRoots.size());
for (var span : failedRoots) {
assertTrue(span.getStatus().getDescription().contains(MESSAGE), span.getName());
assertTrue(span.getEvents().isEmpty(), span.getName());
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -2125,9 +2125,8 @@ void failedInvocation_setsErrorOnBothWorkflowAndInvocationSpans() {
StatusCode.ERROR,
invocationSpan.getStatus().getStatusCode(),
"Invocation span should be ERROR on a FAILED invocation");
assertTrue(
workflowSpan.getEvents().stream().anyMatch(e -> e.getName().equals("exception")),
"Workflow span should record the execution exception on FAILED");
assertTrue(workflowSpan.getEvents().isEmpty(), "Workflow reports failure through its status");
assertTrue(invocationSpan.getEvents().isEmpty(), "Invocation reports failure through its status");
}

// ─── Helpers ─────────────────────────────────────────────────────────
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,15 @@
import java.util.List;
import java.util.concurrent.CompletionException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.ValueSource;
import software.amazon.awssdk.services.lambda.model.ErrorObject;
import software.amazon.awssdk.services.lambda.model.Operation;
import software.amazon.lambda.durable.exception.DurableOperationException;
import software.amazon.lambda.durable.plugin.InvocationEndInfo;
import software.amazon.lambda.durable.plugin.InvocationInfo;
import software.amazon.lambda.durable.plugin.InvocationStatus;
Expand All @@ -34,6 +40,43 @@ class OperationExceptionEventTest {
private static final AttributeKey<String> MESSAGE = AttributeKey.stringKey("exception.message");
private static final AttributeKey<String> STACK = AttributeKey.stringKey("exception.stacktrace");

static Stream<Arguments> storedErrors() {
return Stream.of("java.lang.RuntimeException", "RejectedError", "", null)
.flatMap(type -> Stream.of(null, Instant.EPOCH, Instant.ofEpochSecond(20))
.map(end -> Arguments.of(type, end)));
}

@ParameterizedTest
@MethodSource("storedErrors")
void operationExceptionUsesStoredTypeWithoutChangingTheFailure(String type, Instant end) {
var details = ErrorObject.builder()
.errorType(type)
.errorMessage("original failure")
.build();
var error = new DurableOperationException(Operation.builder().build(), details);
try (var exporter = InMemorySpanExporter.create();
var provider = SdkTracerProvider.builder()
.addSpanProcessor(SimpleSpanProcessor.create(exporter))
.build()) {
var span = provider.get("test").spanBuilder(NAME).startSpan();
var info = new OperationEndInfo(
"op", NAME, "CALLBACK", "Callback", null, null, end, "FAILED", null, true, error, null);
OtelPluginSupport.applyOperationStatus(span, info, end);
span.end();
var events = exporter.getFinishedSpanItems().get(0).getEvents();
assertEquals(1, events.size());
assertEquals(
type != null ? type : DurableOperationException.class.getCanonicalName(),
events.get(0).getAttributes().get(TYPE));
assertEquals("original failure", events.get(0).getAttributes().get(MESSAGE));
assertSame(error, info.error());
assertSame(details, error.getErrorObject());
if (end != null && !end.equals(Instant.EPOCH)) {
assertEquals(20_000_000_000L, events.get(0).getEpochNanos());
}
}
}

@ParameterizedTest
@ValueSource(booleans = {false, true})
void knownEndRetainsRealExceptionAndWrapperDiagnostics(boolean wrapped) {
Expand Down
Loading