diff --git a/.github/workflows/otel-conformance-tests.yml b/.github/workflows/otel-conformance-tests.yml index 96afa4848..2cac91fed 100644 --- a/.github/workflows/otel-conformance-tests.yml +++ b/.github/workflows/otel-conformance-tests.yml @@ -49,7 +49,7 @@ on: conformance_test_ref: description: Optional conformance test commit SHA or branch name required: false - default: "05960beb6c81d7b24b53114f2581b3b6f9fef6d4" + default: "c32b40f5548f4776b7d533a76838e02358bc66d3" type: string permissions: {} @@ -69,7 +69,7 @@ jobs: actions: write contents: read id-token: write - uses: aws/aws-durable-execution-conformance-tests/.github/workflows/opentelemetry-orchestrator.yml@d8b8399264628dae57b7a96bf4cbf3e39a99fd8d + uses: aws/aws-durable-execution-conformance-tests/.github/workflows/opentelemetry-orchestrator.yml@c32b40f5548f4776b7d533a76838e02358bc66d3 with: case_count: 26 runs_on: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository && 'ubuntu-latest' || format('codebuild-github-actions-runner-{0}-{1}', github.run_id, github.run_attempt) }} @@ -77,7 +77,7 @@ jobs: resource_prefix: j sdk_repository: aws/aws-durable-execution-sdk-java sdk_ref: ${{ github.event.pull_request.head.sha || github.sha }} - conformance_test_ref: ${{ inputs.conformance_test_ref || '05960beb6c81d7b24b53114f2581b3b6f9fef6d4' }} + conformance_test_ref: ${{ inputs.conformance_test_ref || 'c32b40f5548f4776b7d533a76838e02358bc66d3' }} checkout_sdk: true # Build the handlers from this repo's checked-out module instead of the conformance repo's # bundled examples/java. Path is relative to the conformance workspace where the SDK is diff --git a/conformance-tests-otel/README.md b/conformance-tests-otel/README.md index 52f67dd3b..285e1e892 100644 --- a/conformance-tests-otel/README.md +++ b/conformance-tests-otel/README.md @@ -17,6 +17,14 @@ The [SAM template](template.yaml) maps each case to its handler in both views. When investigating a run, read the shared requirements at the `conformance_test_ref` selected by the [OTel workflow](../.github/workflows/otel-conformance-tests.yml). +Case 11 pauses at `otel-invoke-replay-barrier` after the child invocation returns. +The shared driver completes that callback after the second caller invocation +ends. The third invocation replays the saved invoke result and returns it, +without starting another child or exporting another terminal invoke span. +The workflow pins the matching requirements from shared conformance PR #131. +The shared cloud run checks the service trace carrier and full X-Ray ancestry; +local replay tests do not establish those service-side properties. + The following notes cover cases 21–26. | Case | Scenario | Behavior | diff --git a/conformance-tests-otel/src/main/java/software/amazon/lambda/durable/conformance/otel/Otel11ChainedInvoke.java b/conformance-tests-otel/src/main/java/software/amazon/lambda/durable/conformance/otel/Otel11ChainedInvoke.java index b2d50e822..ad642da36 100644 --- a/conformance-tests-otel/src/main/java/software/amazon/lambda/durable/conformance/otel/Otel11ChainedInvoke.java +++ b/conformance-tests-otel/src/main/java/software/amazon/lambda/durable/conformance/otel/Otel11ChainedInvoke.java @@ -12,10 +12,12 @@ public final class Otel11ChainedInvoke extends OtelConformanceHandler handleRequest(Map event, DurableContext context) { requireScenario(event, "chained-invoke"); - return context.invoke( + var result = context.invoke( "otel-invoke", System.getenv("OTEL_INVOKE_TARGET_FUNCTION_NAME"), event, new TypeToken>() {}); + context.waitForCallback("otel-invoke-replay-barrier", String.class, (callbackId, step) -> {}); + return result; } } diff --git a/docs/advanced/propagation-metadata.md b/docs/advanced/propagation-metadata.md index 33e142990..4aac1d32e 100644 --- a/docs/advanced/propagation-metadata.md +++ b/docs/advanced/propagation-metadata.md @@ -38,10 +38,18 @@ invoke path collects after that hook and therefore uses the operation context it The invocation view can read the sampling flags from an operation span that has already started. The execution view defers that span until completion, so its placeholder does not establish a per-span sampling decision. It contributes -metadata only when the visible provider uses `DurableSampler`, including the application-owned builder setup. It -abstains for replacement samplers and opaque providers whose effective sampler cannot be verified, including opaque -agent facades. In those cases the backend's omitted-header fallback applies; no new invoke-parent linkage is promised. -The invocation view applies the same restriction if asked before an operation span is available. +metadata when the visible provider uses `DurableSampler`, including the application-owned builder setup. With an +opaque Java-agent provider, the existing autoconfiguration SPI's post-build callback checks the actual provider's final +sampler. Only an unwrapped `DurableSampler` is accepted, and only that exact instance can confirm consumption of the +actual Invocation span's sampling intent through the thread-scoped bridge. A filter around `DurableSampler` cannot +authorize propagation merely by delegating Invocation sampling. A new configuration clears the previous verification; +an absent callback or an unrecognized final sampler leaves propagation disabled for an opaque provider. +The confirmation matches the trace ID, span name and execution ARN, restores an enclosing +observation when nested, and is cleared when the span-creation scope closes. This lets the producer use the deterministic +invoke parent ID and actual invocation sampling flags without accessing the agent's provider. +Replacement samplers and opaque providers without confirmation still abstain. In those cases the backend's omitted-header +fallback applies; no new invoke-parent linkage is promised. The invocation view still requires a visible durable sampler +if asked before an operation span is available. This does not change the configured sampler, create an early operation span, or query the sampler again. ## Replay and retry diff --git a/otel-plugin/README.md b/otel-plugin/README.md index c032e0f7a..149bd2cbb 100644 --- a/otel-plugin/README.md +++ b/otel-plugin/README.md @@ -104,12 +104,17 @@ checkpoints carry that context in the flat optional `ChainedInvokeOptions.XAmznT AWS SDK for Java 2.56.0 provides the generated typed setter. Deployed parent/child trace behavior still requires cloud validation. See [the implemented path and validation status](../docs/advanced/propagation-metadata.md). -Execution-view operation spans are deferred until completion. Their placeholders can provide sampling metadata only -with a visible provider using the SDK's `DurableSampler`, as configured by the application-owned builder constructor. -Replacement samplers and opaque provider/agent facades therefore cause this view to abstain; the backend's existing -omitted-header fallback applies. Invocation view uses the actual started operation span's flags; before that span is -available, it applies the same sampling-policy restriction. Neither path changes -the customer's sampler or creates a span just to obtain a sampling decision. +Execution-view operation spans are deferred until completion. Their placeholders can provide sampling metadata with +a visible provider using the SDK's `DurableSampler`, or an agent provider verified after SDK autoconfiguration. +The post-build callback checks the immutable provider's final sampler, after all customizers have run. Only that exact +`DurableSampler` instance can confirm consumption of this invocation's sampling intent; a later filtering wrapper is +rejected even if it delegates Invocation sampling to an inner `DurableSampler`. Confirmation crosses the application/agent class-loader boundary through +the existing thread-scoped bridge and is cleared after span creation. The propagated parent is the deterministic +invoke span ID, and sampling flags come from the actual Invocation span. +Replacement samplers and opaque providers without post-build verification and confirmation still abstain; the backend's existing +omitted-header fallback applies. Invocation view uses the actual started operation span's flags. Before that span is +available, it requires a visible durable sampler. Neither path changes the customer's sampler, samples again, or +creates a span just to obtain propagation metadata. ## Fallback execution roots diff --git a/otel-plugin/pom.xml b/otel-plugin/pom.xml index 46b6ff399..a6ffb8d0f 100644 --- a/otel-plugin/pom.xml +++ b/otel-plugin/pom.xml @@ -114,6 +114,7 @@ --> <_X_AMZN_TRACE_ID/> + case11-target:1 diff --git a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/DurableSampler.java b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/DurableSampler.java index 2c9ae7f06..88dc8fa2f 100644 --- a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/DurableSampler.java +++ b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/DurableSampler.java @@ -128,6 +128,10 @@ public SamplingResult shouldSample( key -> delegate.shouldSample( Context.root(), key.traceId(), name, spanKind, attributes, Collections.emptyList())) : intent.resolved(); + if (OtelPluginAutoConfigurationState.isConfiguredSampler(this)) { + DurableSamplingDecision.confirmConsumption( + traceId, name, attributes.get(SpanAttributes.DURABLE_EXECUTION_ARN)); + } if (name.equals("DurableExecutionRoot") && Boolean.TRUE.equals(attributes.get(SpanAttributes.DURABLE_EXECUTION_SYNTHETIC_ROOT))) { // Re-exported anchors carry only SDK identity and the sampled decision. Per-invocation sampler diff --git a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/DurableSamplingDecision.java b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/DurableSamplingDecision.java index 28bbcfd24..570c61109 100644 --- a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/DurableSamplingDecision.java +++ b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/DurableSamplingDecision.java @@ -91,13 +91,53 @@ static Scope openScope(Intent intent) { var key = scopedProperty(); var previous = System.getProperty(key); System.setProperty(key, encode(intent)); - return () -> { - if (previous == null) { - System.clearProperty(key); - } else { - System.setProperty(key, previous); - } - }; + return () -> restoreProperty(key, previous); + } + + /** Observes whether the actual sampler consumes the intent for this span, including across agent class loaders. */ + static Observation observe(Intent intent, String traceId, String spanName, String executionArn) { + var key = observationProperty(); + var previous = System.getProperty(key); + var expected = spanIdentity(traceId, spanName, executionArn); + var scope = openScope(intent); + System.setProperty(key, expected); + return new Observation(scope, key, previous, expected); + } + + /** Called only by the verified final sampler after it resolves an intent for the span being sampled. */ + static void confirmConsumption(String traceId, String spanName, String executionArn) { + var key = observationProperty(); + var expected = System.getProperty(key); + if (expected != null && expected.equals(spanIdentity(traceId, spanName, executionArn))) { + System.setProperty(key, "CONSUMED:" + expected); + } + } + + private static String observationProperty() { + // An older agent copy that only observes sampler participation must not satisfy final-sampler verification. + return scopedProperty() + ".verifiedSamplerObservation"; + } + + private static String spanIdentity(String traceId, String spanName, String executionArn) { + return traceId + "\n" + spanName + "\n" + executionArn; + } + + private static void restoreProperty(String key, String previous) { + if (previous == null) System.clearProperty(key); + else System.setProperty(key, previous); + } + + /** A scoped confirmation for one span; it never infers the sampler from the provider's wrapper type. */ + record Observation(Scope scope, String key, String previous, String expected) implements AutoCloseable { + boolean wasConsumed() { + return ("CONSUMED:" + expected).equals(System.getProperty(key)); + } + + @Override + public void close() { + restoreProperty(key, previous); + scope.close(); + } } /** Consumes both carriers before callbacks can reuse the parent context or cross-loader bridge. */ diff --git a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/ExecutionOtelPlugin.java b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/ExecutionOtelPlugin.java index 617bab5bb..0ada70103 100644 --- a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/ExecutionOtelPlugin.java +++ b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/ExecutionOtelPlugin.java @@ -136,6 +136,7 @@ public final class ExecutionOtelPlugin implements DurableExecutionPlugin { private final DurableSamplingDecision.Intent samplingIntent; private final Span invocationSpan; + private final boolean invocationUsesDurableSampler; private final Span executionRootSpan; /** @@ -280,6 +281,7 @@ private ExecutionOtelPlugin(OtelPluginEnvironment environment, InvocationInfo in this.executionTrace = null; this.executionAncestor = null; this.invocationSpan = null; + this.invocationUsesDurableSampler = false; this.executionRootSpan = null; this.workflowSpanContext = null; return; @@ -343,7 +345,16 @@ private ExecutionOtelPlugin(OtelPluginEnvironment environment, InvocationInfo in spanBuilder.setAttribute(AttributeKey.stringKey("faas.invocation_id"), info.requestId()); } - this.invocationSpan = startDurableSpan(spanBuilder); + if (OtelPluginSupport.usesDurableSamplingBridge(sdkTracerProvider)) { + try (var observation = DurableSamplingDecision.observe( + samplingIntent, canonicalTraceId, "Invocation", durableExecutionArn)) { + this.invocationSpan = spanBuilder.startSpan(); + this.invocationUsesDurableSampler = observation.wasConsumed(); + } + } else { + this.invocationSpan = spanBuilder.startSpan(); + this.invocationUsesDurableSampler = false; + } // Defer the recording Workflow span until terminal completion. The placeholder uses the Invocation span's // resolved sampling metadata so operation parents/links match the span that is eventually exported. @@ -456,9 +467,11 @@ public String providePropagationMetadata( if (disabled() || !executionArn.equals(durableExecutionArn)) return null; var trace = executionTrace; if (trace == null) return null; - // These contexts are deferred placeholders, not sampled operation spans. A replacement sampler may decide - // differently for Invocation and invoke; an opaque provider does not expose its effective policy. - if (sdkTracerProvider == null || !OtelPluginSupport.usesDurableSamplingBridge(sdkTracerProvider)) return null; + // Deferred operation contexts may reuse the invocation's decision only when DurableSampler owns it. + // An agent hides its provider, so require a bridge confirmation from its verified final sampler instance. + // A replacement sampler can decide differently per span and must continue to abstain here. + if (!OtelPluginSupport.usesDurableSamplingBridge(sdkTracerProvider) + || (sdkTracerProvider == null && !invocationUsesDurableSampler)) return null; var context = operationContexts.get(operationId); if (context == null) { // Before a new operation starts, use the same deterministic ID its initial span will receive. diff --git a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/OtelPluginAutoConfigurationCustomizerProvider.java b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/OtelPluginAutoConfigurationCustomizerProvider.java index d9cd38ec0..946806c76 100644 --- a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/OtelPluginAutoConfigurationCustomizerProvider.java +++ b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/OtelPluginAutoConfigurationCustomizerProvider.java @@ -2,14 +2,17 @@ // SPDX-License-Identifier: Apache-2.0 package software.amazon.lambda.durable.otel; +import io.opentelemetry.sdk.OpenTelemetrySdk; import io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizer; import io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizerProvider; +import io.opentelemetry.sdk.autoconfigure.spi.internal.AutoConfigureListener; /** * Wraps the Java agent's configured ID generator with scoped durable-execution overrides so durable spans get * deterministic IDs. */ -public final class OtelPluginAutoConfigurationCustomizerProvider implements AutoConfigurationCustomizerProvider { +public final class OtelPluginAutoConfigurationCustomizerProvider + implements AutoConfigurationCustomizerProvider, AutoConfigureListener { @Override public void customize(AutoConfigurationCustomizer autoConfiguration) { @@ -23,6 +26,14 @@ public void customize(AutoConfigurationCustomizer autoConfiguration) { autoConfiguration.addSamplerCustomizer((sampler, config) -> DurableSampler.wrap(sampler)); } + @Override + public void afterAutoConfigure(OpenTelemetrySdk sdk) { + // This callback runs after every customizer and provider build. Checking the immutable provider here catches + // even a later customizer wrapping DurableSampler, without inspecting or modifying another extension's filter. + OtelPluginAutoConfigurationState.recordConfiguredSampler( + sdk.getSdkTracerProvider().getSampler()); + } + @Override public int order() { return Integer.MAX_VALUE; diff --git a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/OtelPluginAutoConfigurationState.java b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/OtelPluginAutoConfigurationState.java index 765489b69..12963fe56 100644 --- a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/OtelPluginAutoConfigurationState.java +++ b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/OtelPluginAutoConfigurationState.java @@ -2,10 +2,14 @@ // SPDX-License-Identifier: Apache-2.0 package software.amazon.lambda.durable.otel; +import io.opentelemetry.sdk.trace.samplers.Sampler; + final class OtelPluginAutoConfigurationState { private static final String INSTALLED_PROPERTY = "software.amazon.lambda.durable.otel.autoConfigurationCustomizerProviderInstalled"; + // Kept in the agent class loader: only the exact final sampler may confirm an application's propagation request. + private static volatile DurableSampler configuredSampler; private OtelPluginAutoConfigurationState() {} @@ -14,10 +18,20 @@ static boolean isInstalled() { } static void markInstalled() { + configuredSampler = null; System.setProperty(INSTALLED_PROPERTY, Boolean.TRUE.toString()); } + static void recordConfiguredSampler(Sampler sampler) { + configuredSampler = sampler instanceof DurableSampler durableSampler ? durableSampler : null; + } + + static boolean isConfiguredSampler(DurableSampler sampler) { + return sampler == configuredSampler; + } + static void resetInstalledForTest() { + configuredSampler = null; System.clearProperty(INSTALLED_PROPERTY); } } diff --git a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/ChainedInvokeConformanceHandlerTest.java b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/ChainedInvokeConformanceHandlerTest.java new file mode 100644 index 000000000..5b269ac93 --- /dev/null +++ b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/ChainedInvokeConformanceHandlerTest.java @@ -0,0 +1,173 @@ +// 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 com.fasterxml.jackson.databind.ObjectMapper; +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.net.URL; +import java.net.URLClassLoader; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import javax.tools.ToolProvider; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import software.amazon.lambda.durable.DurableConfig; +import software.amazon.lambda.durable.DurableHandler; +import software.amazon.lambda.durable.TypeToken; +import software.amazon.lambda.durable.model.ExecutionStatus; +import software.amazon.lambda.durable.plugin.DurableExecutionPlugin; +import software.amazon.lambda.durable.plugin.DurableExecutionPluginFactory; +import software.amazon.lambda.durable.plugin.InvocationEndInfo; +import software.amazon.lambda.durable.plugin.InvocationStatus; +import software.amazon.lambda.durable.testing.LocalDurableTestRunner; + +@Timeout(20) +class ChainedInvokeConformanceHandlerTest { + private static final String INVOKE = "otel-invoke"; + private static final String BARRIER = "otel-invoke-replay-barrier"; + + @TempDir + Path directory; + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + @SuppressWarnings("unchecked") + void actualHandlerReplaysInvokeResultWithoutAnotherTerminalExport(boolean executionView) throws Exception { + DeterministicIdGenerator.clearSharedStateForTest(); + DurableSamplingDecision.clearSharedStateForTest(); + OtelPluginAutoConfigurationState.resetInstalledForTest(); + var classes = compileHandler(); + var workers = Executors.newFixedThreadPool(4); + var exporter = InMemorySpanExporter.create(); + try (var loader = new URLClassLoader( + new URL[] {classes.toUri().toURL()}, getClass().getClassLoader())) { + var handler = (DurableHandler, Map>) + loader.loadClass("software.amazon.lambda.durable.conformance.otel.Otel11ChainedInvoke") + .getConstructor() + .newInstance(); + var ends = new CopyOnWriteArrayList(); + var runner = LocalDurableTestRunner.create( + new TypeToken>() {}, + handler::handleRequest, + configuration(executionView, exporter, workers, ends)); + Map input = Map.of("scenario", "chained-invoke", "nested", Map.of("value", 42)); + var first = runner.run(input); + assertEquals(ExecutionStatus.PENDING, first.getStatus()); + var invokeId = first.getOperation(INVOKE).getId(); + assertTrue(terminalInvokes(exporter).isEmpty()); + runner.completeChainedInvoke(INVOKE, new ObjectMapper().writeValueAsString(input)); + + var second = runner.run(input); + assertEquals(ExecutionStatus.PENDING, second.getStatus()); + assertEquals(invokeId, second.getOperation(INVOKE).getId()); + assertEquals(1, terminalInvokes(exporter).size()); + var completed = terminalInvokes(exporter).get(0); + assertEquals(StatusCode.OK, completed.getStatus().getStatusCode()); + runner.completeCallback(runner.getCallbackId(BARRIER + "-callback"), "\"resume\""); + + var third = runner.run(input); + assertEquals(ExecutionStatus.SUCCEEDED, third.getStatus()); + assertEquals(input, third.getResult(new TypeToken>() {})); + assertEquals(invokeId, third.getOperation(INVOKE).getId()); + assertEquals(List.of(completed), terminalInvokes(exporter)); + assertEquals( + List.of(InvocationStatus.PENDING, InvocationStatus.PENDING, InvocationStatus.SUCCEEDED), + ends.stream().map(InvocationEndInfo::invocationStatus).toList()); + assertEquals( + 1, + third.getHistoryEvents().stream() + .filter(event -> "ChainedInvokeStarted".equals(event.eventTypeAsString())) + .count()); + assertEquals( + 1, + third.getHistoryEvents().stream() + .filter(event -> "ChainedInvokeSucceeded".equals(event.eventTypeAsString())) + .count()); + assertEquals( + 1, + third.getHistoryEvents().stream() + .filter(event -> "CallbackSucceeded".equals(event.eventTypeAsString())) + .count()); + } finally { + workers.shutdownNow(); + assertTrue(workers.awaitTermination(5, TimeUnit.SECONDS)); + exporter.close(); + DeterministicIdGenerator.clearSharedStateForTest(); + DurableSamplingDecision.clearSharedStateForTest(); + OtelPluginAutoConfigurationState.resetInstalledForTest(); + } + } + + private DurableConfig configuration( + boolean executionView, + InMemorySpanExporter exporter, + ExecutorService workers, + List ends) { + var provider = SdkTracerProvider.builder().addSpanProcessor(SimpleSpanProcessor.create(exporter)); + var options = OtelPluginConfig.builder() + .contextExtractor(() -> null) + .enableMdc(false) + .build(); + DurableExecutionPluginFactory otel = executionView + ? ExecutionOtelPlugin.factory(provider, options) + : InvocationOtelPlugin.factory(provider, options); + var observer = new DurableExecutionPlugin() { + @Override + public void onInvocationEnd(InvocationEndInfo info) { + ends.add(info); + } + }; + return DurableConfig.builder() + .withExecutorService(workers) + .withPlugins(otel, ignored -> observer) + .build(); + } + + private List terminalInvokes(InMemorySpanExporter exporter) { + return exporter.getFinishedSpanItems().stream() + .filter(span -> INVOKE.equals(span.getName())) + .filter(span -> "SUCCEEDED".equals(span.getAttributes().get(DURABLE_OPERATION_STATUS))) + .toList(); + } + + private Path compileHandler() throws Exception { + var root = Path.of(System.getProperty("user.dir")).toAbsolutePath(); + while (root != null && !Files.isDirectory(root.resolve("conformance-tests-otel"))) root = root.getParent(); + assertNotNull(root, "repository root containing deployment handler sources"); + var source = + root.resolve("conformance-tests-otel/src/main/java/software/amazon/lambda/durable/conformance/otel"); + var classes = Files.createDirectories(directory.resolve("classes")); + var compiler = ToolProvider.getSystemJavaCompiler(); + assertNotNull(compiler); + assertEquals( + 0, + compiler.run( + null, + null, + null, + "--release", + "17", + "-cp", + System.getProperty("java.class.path"), + "-d", + classes.toString(), + source.resolve("OtelConformanceHandler.java").toString(), + source.resolve("Otel11ChainedInvoke.java").toString())); + return classes; + } +} diff --git a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/DurableSamplerTest.java b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/DurableSamplerTest.java index f84338d2e..f558950bb 100644 --- a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/DurableSamplerTest.java +++ b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/DurableSamplerTest.java @@ -4,6 +4,7 @@ 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.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; import static software.amazon.lambda.durable.otel.Invocations.started; @@ -28,6 +29,8 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import software.amazon.lambda.durable.plugin.InvocationEndInfo; import software.amazon.lambda.durable.plugin.InvocationInfo; import software.amazon.lambda.durable.plugin.InvocationStatus; @@ -63,6 +66,48 @@ void tearDown() { OtelPluginAutoConfigurationState.resetInstalledForTest(); } + @ParameterizedTest + @ValueSource(strings = {"trace", "span", "execution"}) + void propagationObservationRejectsAnotherSpan(String mismatch) { + var sampler = DurableSampler.wrap(Sampler.alwaysOff()); + OtelPluginAutoConfigurationState.recordConfiguredSampler(sampler); + var intent = DurableSamplingDecision.Intent.resolved(SamplingResult.recordOnly()); + try (var observation = DurableSamplingDecision.observe(intent, TRACE_ID, "Invocation", ARN)) { + var result = sampler.shouldSample( + Context.root(), + mismatch.equals("trace") ? "b".repeat(32) : TRACE_ID, + mismatch.equals("span") ? "another-span" : "Invocation", + SpanKind.INTERNAL, + Attributes.of(SpanAttributes.DURABLE_EXECUTION_ARN, mismatch.equals("execution") ? "other" : ARN), + List.of()); + assertEquals(SamplingDecision.RECORD_ONLY, result.getDecision()); + assertFalse(observation.wasConsumed()); + } + assertNull(DurableSamplingDecision.consume(Context.root())); + } + + @Test + void propagationObservationRestoresNestedScopesAndDoesNotLeak() { + var sampler = DurableSampler.wrap(Sampler.alwaysOff()); + OtelPluginAutoConfigurationState.recordConfiguredSampler(sampler); + var intent = DurableSamplingDecision.Intent.resolved(SamplingResult.recordAndSample()); + var attributes = Attributes.of(SpanAttributes.DURABLE_EXECUTION_ARN, ARN); + try (var outer = DurableSamplingDecision.observe(intent, TRACE_ID, "Invocation", ARN)) { + try (var inner = DurableSamplingDecision.observe(intent, TRACE_ID, "Workflow", ARN)) { + sampler.shouldSample(Context.root(), TRACE_ID, "Workflow", SpanKind.INTERNAL, attributes, List.of()); + assertTrue(inner.wasConsumed()); + assertFalse(outer.wasConsumed()); + } + assertFalse(outer.wasConsumed()); + sampler.shouldSample(Context.root(), TRACE_ID, "Invocation", SpanKind.INTERNAL, attributes, List.of()); + assertTrue(outer.wasConsumed()); + } + try (var next = DurableSamplingDecision.observe(intent, TRACE_ID, "Invocation", ARN)) { + assertFalse(next.wasConsumed()); + } + assertNull(DurableSamplingDecision.consume(Context.root())); + } + @Test void consumingContextIntentAlsoClearsCrossLoaderFallback() { var delegate = new CountingSampler(Sampler.alwaysOff()); diff --git a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/DurableSamplingDecisionClassLoaderTest.java b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/DurableSamplingDecisionClassLoaderTest.java index 68f29f4c0..e430c0e12 100644 --- a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/DurableSamplingDecisionClassLoaderTest.java +++ b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/DurableSamplingDecisionClassLoaderTest.java @@ -21,6 +21,7 @@ import io.opentelemetry.api.trace.TracerProvider; import io.opentelemetry.context.Context; import io.opentelemetry.context.propagation.ContextPropagators; +import io.opentelemetry.sdk.OpenTelemetrySdk; import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter; import io.opentelemetry.sdk.trace.IdGenerator; import io.opentelemetry.sdk.trace.ReadWriteSpan; @@ -124,6 +125,15 @@ public String getDescription() { .setExportUnsampledSpans(true) .build()) .build()) { + var customizerType = + Class.forName(OtelPluginAutoConfigurationCustomizerProvider.class.getName(), true, loader); + customizerType + .getMethod("afterAutoConfigure", OpenTelemetrySdk.class) + .invoke( + customizerType.getConstructor().newInstance(), + OpenTelemetrySdk.builder() + .setTracerProvider(provider) + .build()); var hidden = new TracerProvider() { public Tracer get(String name) { return provider.get(name); @@ -159,6 +169,11 @@ public ContextPropagators getPropagators() { System.getProperty("com.amazonaws.xray.traceHeader")); var plugin = factory.createPlugin(info); plugin.onInvocationStart(info); + if (pluginName.equals("ExecutionOtelPlugin")) { + var metadata = plugin.providePropagationMetadata(arn, "invoke-op", null, "target:live"); + assertNotNull(metadata, "The sampler's decision must be available across the agent boundary"); + assertTrue(metadata.endsWith(";Sampled=" + (recordOnly ? "0" : "1"))); + } plugin.onInvocationEnd(new InvocationEndInfo( "request", arn, diff --git a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/InvokePropagationIntegrationTest.java b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/InvokePropagationIntegrationTest.java index dadd9f2c7..380ac0624 100644 --- a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/InvokePropagationIntegrationTest.java +++ b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/InvokePropagationIntegrationTest.java @@ -4,7 +4,13 @@ import static org.junit.jupiter.api.Assertions.*; +import io.opentelemetry.api.GlobalOpenTelemetry; +import io.opentelemetry.api.OpenTelemetry; import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.api.trace.TracerProvider; +import io.opentelemetry.context.propagation.ContextPropagators; +import io.opentelemetry.sdk.OpenTelemetrySdk; import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter; import io.opentelemetry.sdk.trace.SdkTracerProvider; import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor; @@ -70,6 +76,94 @@ class InvokePropagationIntegrationTest { private static final String PAYLOAD = "{\"traceparent\":\"customer-owned\",\"value\":42}"; private static final BiFunction INVOKE = (input, context) -> context.invoke("invoke", "target:live", Map.of("value", 42), String.class); + private static final BiFunction INVOKE_WITH_BARRIER = (input, context) -> { + var result = INVOKE.apply(input, context); + context.waitForCallback("resume", String.class, (callbackId, step) -> {}); + return result; + }; + + @ParameterizedTest + @CsvSource({"true,true", "true,false", "false,true", "false,false"}) + void executionViewPropagatesThroughOpaqueAgentProvider(boolean sampled, boolean explicitDecision) { + GlobalOpenTelemetry.resetForTest(); + OtelPluginAutoConfigurationState.markInstalled(); + var delegateSamples = explicitDecision ? !sampled : sampled; + try (var exporter = InMemorySpanExporter.create(); + var provider = SdkTracerProvider.builder() + .setSampler(DurableSampler.wrap(delegateSamples ? Sampler.alwaysOn() : Sampler.alwaysOff())) + .setIdGenerator(new DeterministicIdGenerator()) + .addSpanProcessor(SimpleSpanProcessor.create(exporter)) + .build()) { + new OtelPluginAutoConfigurationCustomizerProvider() + .afterAutoConfigure(OpenTelemetrySdk.builder() + .setTracerProvider(provider) + .build()); + installOpaqueProvider(provider); + assertOpaqueInvokePropagation(exporter, sampled, explicitDecision); + } finally { + GlobalOpenTelemetry.resetForTest(); + OtelPluginAutoConfigurationState.resetInstalledForTest(); + DurableSamplingDecision.clearSharedStateForTest(); + DeterministicIdGenerator.clearSharedStateForTest(); + } + } + + private static void assertOpaqueInvokePropagation( + InMemorySpanExporter exporter, boolean sampled, boolean explicitDecision) { + var config = OtelPluginConfig.builder() + .enableMdc(false) + .contextExtractor(() -> new ExtractedContext( + TRACE, + "1234567890123456", + explicitDecision + ? (sampled ? ExtractedContext.Sampling.SAMPLED : ExtractedContext.Sampling.NOT_SAMPLED) + : ExtractedContext.Sampling.UNDECIDED)) + .build(); + try (var fixture = new Fixture(ExecutionOtelPlugin.factory(config))) { + assertEquals( + ExecutionStatus.PENDING, fixture.run(INVOKE_WITH_BARRIER).status()); + var update = fixture.client.starts().get(0); + var header = update.chainedInvokeOptions().xAmznTraceId(); + var spanId = new DeterministicIdGenerator().generateSpanIdForOperation(ARN, update.id()); + assertEquals( + "Root=1-6955b900-123456789012345678901234;Parent=" + spanId + ";Sampled=" + (sampled ? "1" : "0"), + header); + fixture.client.completeChainedInvoke("invoke", OperationResult.succeeded("\"result\"")); + assertEquals( + ExecutionStatus.PENDING, fixture.run(INVOKE_WITH_BARRIER).status()); + fixture.client.completeCallback( + fixture.client.getCallbackId("resume-callback"), OperationResult.succeeded("\"resume\"")); + assertEquals( + ExecutionStatus.SUCCEEDED, fixture.run(INVOKE_WITH_BARRIER).status()); + assertEquals(1, fixture.client.starts().size(), "Completed invoke replay must not send another START"); + var invokes = exporter.getFinishedSpanItems().stream() + .filter(span -> span.getName().equals("invoke")) + .toList(); + assertEquals(sampled ? 1 : 0, invokes.size()); + if (sampled) assertEquals(spanId, invokes.get(0).getSpanId()); + } + } + + private static void installOpaqueProvider(SdkTracerProvider provider) { + var facade = new TracerProvider() { + public Tracer get(String name) { + return provider.get(name); + } + + public Tracer get(String name, String version) { + return provider.get(name, version); + } + }; + GlobalOpenTelemetry.set(new OpenTelemetry() { + public TracerProvider getTracerProvider() { + return facade; + } + + public ContextPropagators getPropagators() { + return ContextPropagators.noop(); + } + }); + } @Test void committedStartAndTerminalReplayNeverRecollectOrReserialize() { diff --git a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/OtelAgentSamplerVerificationTest.java b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/OtelAgentSamplerVerificationTest.java new file mode 100644 index 000000000..e5da58897 --- /dev/null +++ b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/OtelAgentSamplerVerificationTest.java @@ -0,0 +1,120 @@ +// 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.assertFalse; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.context.Context; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.autoconfigure.spi.internal.AutoConfigureListener; +import io.opentelemetry.sdk.trace.SdkTracerProvider; +import io.opentelemetry.sdk.trace.data.LinkData; +import io.opentelemetry.sdk.trace.samplers.Sampler; +import io.opentelemetry.sdk.trace.samplers.SamplingResult; +import java.util.List; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +class OtelAgentSamplerVerificationTest { + private static final String TRACE = "a".repeat(32); + private static final String ARN = "arn:test:agent-sampler"; + + @AfterEach + void clearState() { + OtelPluginAutoConfigurationState.resetInstalledForTest(); + DurableSamplingDecision.clearSharedStateForTest(); + } + + @Test + void onlyThePostBuildCallbackAuthorizesTheInstalledInstance() { + var sampler = DurableSampler.wrap(Sampler.alwaysOn()); + var other = DurableSampler.wrap(Sampler.alwaysOn()); + OtelPluginAutoConfigurationState.markInstalled(); + try (var sdk = OpenTelemetrySdk.builder() + .setTracerProvider( + SdkTracerProvider.builder().setSampler(sampler).build()) + .build()) { + assertFalse(confirm(sampler)); + AutoConfigureListener listener = new OtelPluginAutoConfigurationCustomizerProvider(); + listener.afterAutoConfigure(sdk); + assertTrue(confirm(sampler)); + assertFalse(confirm(other)); + assertSame(sampler, sdk.getSdkTracerProvider().getSampler()); + OtelPluginAutoConfigurationState.markInstalled(); + assertFalse(confirm(sampler), "A new configuration cannot reuse the previous verification"); + } + } + + @Test + void legacyConsumptionConfirmationCannotAuthorizePropagation() { + var key = "software.amazon.lambda.durable.otel.scopedSamplingDecision." + + Thread.currentThread().getId() + ".observation"; + var intent = DurableSamplingDecision.Intent.resolved(SamplingResult.recordAndSample()); + try (var observation = DurableSamplingDecision.observe(intent, TRACE, "Invocation", ARN)) { + System.setProperty(key, "CONSUMED:" + TRACE + "\nInvocation\n" + ARN); + assertFalse(observation.wasConsumed()); + } finally { + System.clearProperty(key); + } + } + + @Test + void finalBuiltProviderWinsOverAnEarlierBuilderOrSamplerObservation() { + var sampler = DurableSampler.wrap(Sampler.alwaysOn()); + var builder = SdkTracerProvider.builder().setSampler(sampler); + AutoConfigureListener listener = new OtelPluginAutoConfigurationCustomizerProvider(); + try (var original = + OpenTelemetrySdk.builder().setTracerProvider(builder.build()).build()) { + listener.afterAutoConfigure(original); + assertTrue(confirm(sampler)); + builder.setSampler(filter(sampler)); + listener.afterAutoConfigure(original); + assertTrue(confirm(sampler), "Mutating the builder cannot change the already-built provider"); + try (var filtered = OpenTelemetrySdk.builder() + .setTracerProvider(builder.build()) + .build()) { + listener.afterAutoConfigure(filtered); + assertFalse(confirm(sampler), "The inner DurableSampler is not the final sampler"); + assertFalse(confirm(filtered.getSdkTracerProvider().getSampler())); + } + } + } + + private static boolean confirm(Sampler sampler) { + var intent = DurableSamplingDecision.Intent.resolved(SamplingResult.recordAndSample()); + try (var observation = DurableSamplingDecision.observe(intent, TRACE, "Invocation", ARN)) { + sampler.shouldSample( + Context.root(), + TRACE, + "Invocation", + SpanKind.INTERNAL, + Attributes.of(SpanAttributes.DURABLE_EXECUTION_ARN, ARN), + List.of()); + return observation.wasConsumed(); + } + } + + private static Sampler filter(Sampler delegate) { + return new Sampler() { + public SamplingResult shouldSample( + Context parent, + String traceId, + String name, + SpanKind kind, + Attributes attributes, + List links) { + return name.equals("invoke") + ? SamplingResult.drop() + : delegate.shouldSample(parent, traceId, name, kind, attributes, links); + } + + public String getDescription() { + return "filter-around-durable-sampler"; + } + }; + } +} diff --git a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/PropagationSamplingAuthorityTest.java b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/PropagationSamplingAuthorityTest.java index 862542d35..156827ba4 100644 --- a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/PropagationSamplingAuthorityTest.java +++ b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/PropagationSamplingAuthorityTest.java @@ -12,6 +12,7 @@ import io.opentelemetry.api.trace.TracerProvider; import io.opentelemetry.context.Context; import io.opentelemetry.context.propagation.ContextPropagators; +import io.opentelemetry.sdk.OpenTelemetrySdk; import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter; import io.opentelemetry.sdk.trace.SdkTracerProvider; import io.opentelemetry.sdk.trace.data.LinkData; @@ -35,12 +36,15 @@ class PropagationSamplingAuthorityTest { @ParameterizedTest @CsvSource({ - "true,false,true", "true,false,false", "true,true,true", "true,true,false", - "false,false,true", "false,false,false", "false,true,true", "false,true,false" + "true,false,true,false", "true,false,false,false", "true,true,true,false", "true,true,false,false", + "false,false,true,false", "false,false,false,false", "false,true,true,false", "false,true,false,false", + "true,false,true,true", "true,false,false,true", "true,true,true,true", "true,true,false,true", + "false,false,true,true", "false,false,false,true", "false,true,true,true", "false,true,false,true" }) void replacementSamplerMayChooseDifferentInvocationAndInvokeFlags( - boolean executionView, boolean opaque, boolean sampleInvoke) { + boolean executionView, boolean opaque, boolean sampleInvoke, boolean delegateInvocation) { var decisions = new ArrayList(); + var innerSampler = DurableSampler.wrap(sampleInvoke ? Sampler.alwaysOff() : Sampler.alwaysOn()); var replacement = new Sampler() { public SamplingResult shouldSample( Context parent, @@ -51,6 +55,9 @@ public SamplingResult shouldSample( List links) { boolean sampled = name.equals("Invocation") ? !sampleInvoke : sampleInvoke; decisions.add(name + ":" + sampled); + if (delegateInvocation && name.equals("Invocation")) { + return innerSampler.shouldSample(parent, traceId, name, kind, attributes, links); + } return sampled ? SamplingResult.recordAndSample() : SamplingResult.drop(); } @@ -66,6 +73,10 @@ public String getDescription() { .setIdGenerator(new DeterministicIdGenerator()) .addSpanProcessor(SimpleSpanProcessor.create(exporter)) .build()) { + new OtelPluginAutoConfigurationCustomizerProvider() + .afterAutoConfigure(OpenTelemetrySdk.builder() + .setTracerProvider(provider) + .build()); var facade = new TracerProvider() { public Tracer get(String name) { return provider.get(name);