diff --git a/.github/workflows/otel-conformance-tests.yml b/.github/workflows/otel-conformance-tests.yml index 80002fe51..3c7e3f4cb 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: "c32b40f5548f4776b7d533a76838e02358bc66d3" + default: "main" type: string permissions: {} @@ -69,15 +69,15 @@ jobs: actions: write contents: read id-token: write - uses: aws/aws-durable-execution-conformance-tests/.github/workflows/opentelemetry-orchestrator.yml@04f194f44a5136074dfb17c8add1d75b24055cad + uses: aws/aws-durable-execution-conformance-tests/.github/workflows/opentelemetry-orchestrator.yml@main with: - case_count: 26 + case_count: 27 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) }} language: java 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 || 'c32b40f5548f4776b7d533a76838e02358bc66d3' }} + conformance_test_ref: ${{ inputs.conformance_test_ref || 'main' }} 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 285e1e892..092d9b732 100644 --- a/conformance-tests-otel/README.md +++ b/conformance-tests-otel/README.md @@ -106,3 +106,9 @@ For case 25, Java reports no plugin error details for an entirely empty error container. The persisted container, callback failure, and caller error stay unchanged. Any present type, message, data, or stack-trace field remains an error detail, including an empty string or an explicitly supplied empty stack list. + +Case 27 (`chained-invoke-suspend`) invokes a child that performs a one-second +durable wait and returns the original payload after resuming. Both views +check two caller invocations, two child invocations and the child wait's +parentage and continuation links. It replaces the suspended-child scenario +formerly exercised by the short mode of `otel-long-running-4`. diff --git a/conformance-tests-otel/src/main/java/software/amazon/lambda/durable/conformance/otel/Otel27ChainedInvokeSuspend.java b/conformance-tests-otel/src/main/java/software/amazon/lambda/durable/conformance/otel/Otel27ChainedInvokeSuspend.java new file mode 100644 index 000000000..fd2d26841 --- /dev/null +++ b/conformance-tests-otel/src/main/java/software/amazon/lambda/durable/conformance/otel/Otel27ChainedInvokeSuspend.java @@ -0,0 +1,21 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.conformance.otel; + +import java.util.Map; +import software.amazon.lambda.durable.DurableContext; +import software.amazon.lambda.durable.TypeToken; + +/** Chained invoke with child suspension for OTel view case 27. */ +public final class Otel27ChainedInvokeSuspend extends OtelConformanceHandler> { + + @Override + public Map handleRequest(Map event, DurableContext context) { + requireScenario(event, "chained-invoke-suspend"); + return context.invoke( + "otel-suspended-invoke", + System.getenv("OTEL_INVOKE_TARGET_FUNCTION_NAME"), + event, + new TypeToken>() {}); + } +} diff --git a/conformance-tests-otel/src/main/java/software/amazon/lambda/durable/conformance/otel/Otel27InvokeTarget.java b/conformance-tests-otel/src/main/java/software/amazon/lambda/durable/conformance/otel/Otel27InvokeTarget.java new file mode 100644 index 000000000..3b9cf6ce4 --- /dev/null +++ b/conformance-tests-otel/src/main/java/software/amazon/lambda/durable/conformance/otel/Otel27InvokeTarget.java @@ -0,0 +1,18 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.conformance.otel; + +import java.time.Duration; +import java.util.Map; +import software.amazon.lambda.durable.DurableContext; + +/** Suspending child for OTel view case 27. */ +public final class Otel27InvokeTarget extends OtelConformanceHandler> { + + @Override + public Map handleRequest(Map event, DurableContext context) { + requireScenario(event, "chained-invoke-suspend"); + context.wait("otel-suspended-target-wait", Duration.ofSeconds(1)); + return event; + } +} diff --git a/conformance-tests-otel/template.yaml b/conformance-tests-otel/template.yaml index 2a183257b..3cc47b559 100644 --- a/conformance-tests-otel/template.yaml +++ b/conformance-tests-otel/template.yaml @@ -738,3 +738,51 @@ Resources: Environment: Variables: OTEL_PLUGIN_MODE: execution + + Otel27ChainedInvokeSuspend: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: + - otel-invocation-27 + Properties: + CodeUri: . + FunctionName: !Sub "${AWS::StackName}-otel-invocation-27" + Handler: software.amazon.lambda.durable.conformance.otel.Otel27ChainedInvokeSuspend + Role: !Ref LambdaExecutionRoleArn + Environment: + Variables: + OTEL_INVOKE_TARGET_FUNCTION_NAME: !Sub "${Otel27InvokeTarget.Arn}:$LATEST" + + Otel27InvokeTarget: + Type: AWS::Serverless::Function + Properties: + CodeUri: . + FunctionName: !Sub "${AWS::StackName}-otel-invocation-27-target" + Handler: software.amazon.lambda.durable.conformance.otel.Otel27InvokeTarget + Role: !Ref LambdaExecutionRoleArn + + OtelExecution27ChainedInvokeSuspend: + Type: AWS::Serverless::Function + TestingMetadata: + TestDescription: + - otel-execution-27 + Properties: + CodeUri: . + FunctionName: !Sub "${AWS::StackName}-otel-execution-27" + Handler: software.amazon.lambda.durable.conformance.otel.Otel27ChainedInvokeSuspend + Role: !Ref LambdaExecutionRoleArn + Environment: + Variables: + OTEL_PLUGIN_MODE: execution + OTEL_INVOKE_TARGET_FUNCTION_NAME: !Sub "${OtelExecution27InvokeTarget.Arn}:$LATEST" + + OtelExecution27InvokeTarget: + Type: AWS::Serverless::Function + Properties: + CodeUri: . + FunctionName: !Sub "${AWS::StackName}-otel-execution-27-target" + Handler: software.amazon.lambda.durable.conformance.otel.Otel27InvokeTarget + Role: !Ref LambdaExecutionRoleArn + Environment: + Variables: + OTEL_PLUGIN_MODE: execution diff --git a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/SuspendedChildConformanceHandlerTest.java b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/SuspendedChildConformanceHandlerTest.java new file mode 100644 index 000000000..5cebf6310 --- /dev/null +++ b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/SuspendedChildConformanceHandlerTest.java @@ -0,0 +1,218 @@ +// 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 SuspendedChildConformanceHandlerTest { + private static final String INVOKE = "otel-suspended-invoke"; + private static final String WAIT = "otel-suspended-target-wait"; + + @TempDir + Path directory; + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void actualChildSuspendsAndResumesBeforeCallerCompletes(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())) { + verifyHandlers(executionView, loader, exporter, workers); + } finally { + workers.shutdownNow(); + assertTrue(workers.awaitTermination(5, TimeUnit.SECONDS)); + exporter.close(); + DeterministicIdGenerator.clearSharedStateForTest(); + DurableSamplingDecision.clearSharedStateForTest(); + OtelPluginAutoConfigurationState.resetInstalledForTest(); + } + } + + private void verifyHandlers( + boolean executionView, URLClassLoader loader, InMemorySpanExporter exporter, ExecutorService workers) + throws Exception { + var ends = new CopyOnWriteArrayList(); + var childEnds = new CopyOnWriteArrayList(); + var handler = loadHandler(loader, "Otel27ChainedInvokeSuspend"); + var childHandler = loadHandler(loader, "Otel27InvokeTarget"); + var runner = LocalDurableTestRunner.create( + new TypeToken>() {}, + handler::handleRequest, + configuration(executionView, exporter, workers, ends)); + var child = LocalDurableTestRunner.create( + new TypeToken>() {}, + childHandler::handleRequest, + configuration(executionView, exporter, workers, childEnds)); + verifyExecution(runner, child); + for (var invocations : List.of(ends, childEnds)) { + assertEquals( + List.of(InvocationStatus.PENDING, InvocationStatus.SUCCEEDED), + invocations.stream() + .map(InvocationEndInfo::invocationStatus) + .toList()); + } + verifyWaitTelemetry(executionView, exporter); + assertEquals(1, terminalInvokes(exporter).size()); + } + + @SuppressWarnings("unchecked") + private DurableHandler, Map> loadHandler(URLClassLoader loader, String name) + throws Exception { + return (DurableHandler, Map>) + loader.loadClass("software.amazon.lambda.durable.conformance.otel." + name) + .getConstructor() + .newInstance(); + } + + private void verifyExecution( + LocalDurableTestRunner, Map> runner, + LocalDurableTestRunner, Map> child) + throws Exception { + Map input = Map.of("scenario", "chained-invoke-suspend", "nested", Map.of("value", 42)); + var first = runner.run(input); + assertEquals(ExecutionStatus.PENDING, first.getStatus()); + var invokeId = first.getOperation(INVOKE).getId(); + var childFirst = child.run(input); + assertEquals(ExecutionStatus.PENDING, childFirst.getStatus()); + var waitId = childFirst.getOperation(WAIT).getId(); + child.advanceTime(); + var childLast = child.run(input); + assertEquals(ExecutionStatus.SUCCEEDED, childLast.getStatus()); + assertEquals(waitId, childLast.getOperation(WAIT).getId()); + var childResult = childLast.getResult(new TypeToken>() {}); + assertEquals(input, childResult); + runner.completeChainedInvoke(INVOKE, new ObjectMapper().writeValueAsString(childResult)); + var last = runner.run(input); + assertEquals(ExecutionStatus.SUCCEEDED, last.getStatus()); + assertEquals(input, last.getResult(new TypeToken>() {})); + assertEquals(invokeId, last.getOperation(INVOKE).getId()); + } + + private void verifyWaitTelemetry(boolean executionView, InMemorySpanExporter exporter) { + var waits = exporter.getFinishedSpanItems().stream() + .filter(span -> WAIT.equals(span.getName())) + .toList(); + assertEquals(executionView ? 1 : 2, waits.size()); + assertEquals( + 1, + waits.stream() + .map(span -> span.getAttributes().get(DURABLE_OPERATION_ID)) + .distinct() + .count()); + var completed = waits.get(waits.size() - 1); + assertEquals(StatusCode.OK, completed.getStatus().getStatusCode()); + if (executionView) { + var childArn = completed.getAttributes().get(DURABLE_EXECUTION_ARN); + var resumed = exporter.getFinishedSpanItems().stream() + .filter(span -> "Invocation".equals(span.getName())) + .filter(span -> childArn.equals(span.getAttributes().get(DURABLE_EXECUTION_ARN))) + .filter(span -> Boolean.FALSE.equals(span.getAttributes().get(DURABLE_FIRST_INVOCATION))) + .findFirst() + .orElseThrow(); + assertEquals( + resumed.getSpanId(), + completed.getLinks().get(0).getSpanContext().getSpanId()); + } else { + assertEquals( + waits.get(0).getSpanId(), + completed.getLinks().get(0).getSpanContext().getSpanId()); + } + } + + 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("Otel27ChainedInvokeSuspend.java").toString(), + source.resolve("Otel27InvokeTarget.java").toString())); + return classes; + } +}