From c0f7a047fa3a1434c17f6d110df011dca9315026 Mon Sep 17 00:00:00 2001 From: Frank Chen <65260095+zhongkechen@users.noreply.github.com> Date: Fri, 23 Oct 2026 14:51:22 -0700 Subject: [PATCH 1/2] test: add LMI observability acceptance --- .github/workflows/lmi-e2e-tests.yml | 5 +- lmi-tests/README.md | 41 +++ lmi-tests/cloud_suite.py | 22 +- lmi-tests/cloud_support.py | 25 +- lmi-tests/otel_cases.py | 98 +++++ lmi-tests/otel_evidence.py | 83 +++++ lmi-tests/pom.xml | 16 + .../lambda/durable/lmi/FixtureInput.java | 17 +- .../lambda/durable/lmi/InvocationTrace.java | 16 + .../lambda/durable/lmi/LifecycleHandler.java | 12 +- .../durable/lmi/ObservedOtelPlugin.java | 160 +++++++++ .../amazon/lambda/durable/lmi/OtelInvoke.java | 57 +++ .../durable/lmi/OtelLedgerProcessor.java | 118 +++++++ .../lambda/durable/lmi/OtelRuntime.java | 133 +++++++ lmi-tests/src/main/resources/logback.xml | 7 + .../durable/lmi/OtelInvocationTest.java | 334 ++++++++++++++++++ .../lambda/durable/lmi/OtelInvokeTest.java | 58 +++ lmi-tests/tests/test_evidence.py | 6 +- lmi-tests/tests/test_otel_evidence.py | 118 +++++++ 19 files changed, 1315 insertions(+), 11 deletions(-) create mode 100644 lmi-tests/otel_cases.py create mode 100644 lmi-tests/otel_evidence.py create mode 100644 lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/ObservedOtelPlugin.java create mode 100644 lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/OtelInvoke.java create mode 100644 lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/OtelLedgerProcessor.java create mode 100644 lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/OtelRuntime.java create mode 100644 lmi-tests/src/main/resources/logback.xml create mode 100644 lmi-tests/src/test/java/software/amazon/lambda/durable/lmi/OtelInvocationTest.java create mode 100644 lmi-tests/src/test/java/software/amazon/lambda/durable/lmi/OtelInvokeTest.java create mode 100644 lmi-tests/tests/test_otel_evidence.py diff --git a/.github/workflows/lmi-e2e-tests.yml b/.github/workflows/lmi-e2e-tests.yml index 1e0c110d94..00fb767a61 100644 --- a/.github/workflows/lmi-e2e-tests.yml +++ b/.github/workflows/lmi-e2e-tests.yml @@ -7,6 +7,7 @@ on: paths: - 'lmi-tests/**' - 'sdk/**' + - 'otel-plugin/**' - 'pom.xml' - '.github/workflows/lmi-e2e-tests.yml' @@ -35,7 +36,7 @@ jobs: if: github.event_name == 'workflow_dispatch' needs: local runs-on: ubuntu-latest - timeout-minutes: 95 + timeout-minutes: 130 permissions: contents: read id-token: write @@ -83,7 +84,7 @@ jobs: aws-region: ${{ env.AWS_REGION }} allowed-account-ids: ${{ secrets.TEST_ACCOUNT_ID }} - name: LMI cloud regression assertions (expected red until issue 726 is fixed) - timeout-minutes: 30 + timeout-minutes: 60 run: python3 -u lmi-tests/cloud_suite.py test --cloud-enabled - name: Refresh AWS credentials for evidence collection if: always() && hashFiles('lmi-tests/artifacts/manifest.json') != '' diff --git a/lmi-tests/README.md b/lmi-tests/README.md index a88c5b73f9..0e1e083d27 100644 --- a/lmi-tests/README.md +++ b/lmi-tests/README.md @@ -260,3 +260,44 @@ sets the limits in the resource declaration. LMI provisioning and version behavi are documented in the AWS guides for [scaling](https://docs.aws.amazon.com/lambda/latest/dg/lambda-managed-instances-scaling.html) and [$LATEST.PUBLISHED](https://docs.aws.amazon.com/lambda/latest/dg/lambda-managed-instances-version-publishing.html). + +## OpenTelemetry acceptance + +The suite adds 32 observability cases to the original 13 lifecycle cases. Both +OTel views run on the existing concurrency-1 and concurrency-2 fixtures with +native sampled/unsampled headers, warm reuse and replay, export failures and +flush-budget controls. Concurrency-2 also repeats overlap, suspension cleanup, +deadline, non-cooperative-child and in-flight-return contracts with plugins enabled. +The existing default8/fixed/nested core fixtures and assertions remain in the suite. + +The fixture uses the real Java 3.x OTel factories and a process-owned provider. +Its observing adapter forwards every hook and the actual `@HandlerScoped` opener; +it only supplies an independent diagnostic owner to the span exporter. Raw span +IDs, attributes, status and timestamps are retained with runtime request/JVM IDs. +The Logback binding exercises MDC restoration. Export completion has its own +counter, so post-return export work must settle before another case or deployment. + +`OtelInvoke`, packaged in the same shaded fixture JAR, uses the AWS SDK's signed +Invoke request with an explicit X-Ray header. The header is never injected into +the handler payload. Runtime `Context.getXrayTraceId()` observations are compared +with the immutable snapshot captured before worker dispatch and with actual +exported trace/parent/sampling values. + +The controllable exporter implements the OpenTelemetry export/flush interfaces. +A held export can outlive the plugin's five-second flush wait; the test records +that boundary before releasing the exporter. Shared-provider ownership, End on +the handler owner, context/MDC restoration, and late export drainage remain +separate assertions. These fixtures do not repair non-cooperative code or claim +lossless delivery after timeout. + +Local public-executor tests cover sampling, missing/invalid/unreadable carriers, +concurrent owners, real local suspend/resume, exported fallback roots, MDC and +exporter/flush behavior. A loopback AWS SDK test checks the actual Invoke header +and payload. Python negative controls reject wrong attribution, missing evidence, +changed owner threads and unexpected sampled output. Final collection revalidates +the observability assertions as well as the original lifecycle assertions. + +Fresh deployed LMI evidence is still required before closing #727. The #726 +lifecycle limitations and the remaining #733/#735 fixes remain independently +tracked. The expanded assertion phase has a 60-minute bound within a 130-minute +cloud job; the five persistent functions and provider capacity are unchanged. diff --git a/lmi-tests/cloud_suite.py b/lmi-tests/cloud_suite.py index 95d51e25bc..f8787fc0c1 100644 --- a/lmi-tests/cloud_suite.py +++ b/lmi-tests/cloud_suite.py @@ -10,6 +10,7 @@ from pathlib import Path import re import subprocess +import sys import time import traceback import uuid @@ -534,6 +535,13 @@ def runtime_requests_drained(events, markers, allow_no_entry=False): task_exits = Counter(event.get("name") for event in selected(trace, "TASK_EXIT")) if task_enters != task_exits: return False + if len(selected(trace, "OTEL_EXPORT_ENTER")) != len(selected(trace, "OTEL_EXPORT_EXIT")): + return False + if len(selected(trace, "OTEL_EXPORT_ENTER")) != ( + len(selected(trace, "OTEL_EXPORT")) + len(selected(trace, "OTEL_EXPORT_ERROR"))): + return False + if trace and max(trace, key=lambda event: event["sequence"]).get("exports", 0): + return False returned_at = wrapper_returns[-1]["sequence"] if any(event["sequence"] > returned_at and event["kind"] in { "BODY", "CHECKPOINT_CALL", "POLL_CALL", "CHECKPOINT_EXIT", "POLL_EXIT", @@ -890,7 +898,7 @@ def collect_warm_snapshots(cloud, items, seconds=30): time.sleep(1) -def cases_for_fixture(cloud, fixture): +def core_cases_for_fixture(cloud, fixture): if fixture == "default1": return [("baseline-concurrency1", lambda: replay_case(cloud, fixture, "baseline"))] if fixture in {"fixed2", "nested2"}: @@ -908,6 +916,13 @@ def cases_for_fixture(cloud, fixture): return cases +def cases_for_fixture(cloud, fixture): + from otel_cases import additional_cases + observed = [(name, case) for target, name, case in additional_cases(cloud, sys.modules[__name__]) + if target == fixture] + return core_cases_for_fixture(cloud, fixture) + observed + + def assert_original_healthy_admission(events, admission): """Reject changed admission evidence instead of repinning earlier timing assertions.""" marker, environment, request_id = admission @@ -961,6 +976,9 @@ def assert_case_request_lifecycles(cloud, invocation_start, invocation_end=None) assert_original_timeout_admission(events, item["timeoutAdmission"]) if "replayHistory" in item: assert_replay(events, item["replayHistory"], item["marker"]) + if "otel" in item: + from otel_evidence import validate_invocations + validate_invocations(events, item["otel"]) if "warmEnvironment" in item: warm_environment(events, item["warmEnvironment"]) warm_snapshots.extend(selected(events, "SNAPSHOT")) @@ -1001,7 +1019,7 @@ def record_failure(node, failure, kind=None): def save_case_assertions(cloud, records): fields = {"marker", "fixture", "scenario", "healthyAdmission", "timeoutAdmission", "warmEnvironment", - "warmRequests", "fixedAdmission", "replayHistory"} + "warmRequests", "fixedAdmission", "replayHistory", "otel"} save(ARTIFACTS / "case-assertions.json", { "runId": cloud.manifest["runId"], "commit": cloud.manifest["commit"], "cases": [{"name": node.get("name"), "items": [ diff --git a/lmi-tests/cloud_support.py b/lmi-tests/cloud_support.py index e7a18b9aac..86a4bf9af3 100644 --- a/lmi-tests/cloud_support.py +++ b/lmi-tests/cloud_support.py @@ -9,6 +9,7 @@ import subprocess import tempfile import time +import uuid TERMINAL_DURABLE_STATUSES = {"SUCCEEDED", "FAILED", "TIMED_OUT", "STOPPED"} @@ -29,7 +30,7 @@ def require(condition, message): def scrub(value): if isinstance(value, dict): - return {k: ("" if k in {"controlUrl", "CheckpointToken"} else scrub(v)) + return {k: ("" if k in {"controlUrl", "exportControlUrl", "CheckpointToken"} else scrub(v)) for k, v in value.items()} if isinstance(value, list): return [scrub(v) for v in value] @@ -289,6 +290,16 @@ def launch(self, fixture, scenario, marker, cohort=None, target=None, peers=1, h payload = {"runId": self.manifest["runId"], "cohort": cohort or marker, "scenario": scenario, "marker": marker, "controlUrl": gate, "targetEnvironment": target, "peers": peers, "holdMillis": hold_ms} + observed = getattr(self, "observability", None) + if isinstance(observed, dict): + header = (f"Root=1-{int(time.time()):08x}-{uuid.uuid4().hex[:24]};" + f"Parent={uuid.uuid4().hex[:16]};Sampled={observed.get('sampled', '1')}") + mode = observed.get("export", "success") + payload.update(otelView=observed["view"], otelExport=mode, traceHeader=header) + item["otel"] = {"view": observed["view"], "mode": mode, "traceHeader": header} + if mode == "held": + item["exportGate"] = marker + "-export" + payload["exportControlUrl"] = self.gate(item["exportGate"]) item["future"] = self.pool.submit(self._invoke, fixture, payload) self.invocations.append(item) return item @@ -311,6 +322,18 @@ def _invoke(self, fixture, payload): raise def _invoke_request(self, fixture, payload): + if payload.get("otelView"): + jar = Path(__file__).resolve().parent / "target/lmi-fixtures.jar" + request = {"function": self.manifest["functions"][fixture]["arn"], + "executionName": invocation_name(payload["runId"], payload["marker"]), + "traceHeader": payload["traceHeader"], + "payload": json.dumps({key: value for key, value in payload.items() if key != "traceHeader"})} + result = subprocess.run( + ["java", "-cp", str(jar), "software.amazon.lambda.durable.lmi.OtelInvoke"], + input=json.dumps(request), capture_output=True, text=True, timeout=40) + if result.returncode: + raise CollectionError(f"Instrumented Invoke failed: {scrub(result.stderr[-4000:])}") + return json.loads(result.stdout) with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "response.json" source = Path(directory) / "payload.json" diff --git a/lmi-tests/otel_cases.py b/lmi-tests/otel_cases.py new file mode 100644 index 0000000000..aa934c20c2 --- /dev/null +++ b/lmi-tests/otel_cases.py @@ -0,0 +1,98 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Additional native-runtime observability cases; original lifecycle assertions remain.""" + +from contextlib import contextmanager +import uuid + +from cloud_support import CollectionError, PreconditionError, require, selected +from otel_evidence import validate_invocations + + +@contextmanager +def observing(cloud, view, sampled="1", export="success"): + previous = getattr(cloud, "observability", None) + cloud.observability = {"view": view, "sampled": sampled, "export": export} + start = len(cloud.invocations) + try: + yield + for item in cloud.invocations[start:]: + wait_evidence(cloud, item) + finally: + cloud.observability = previous + + +def wait_evidence(cloud, item): + def complete(events): + local = [event for event in events if event["marker"] == item["marker"]] + try: + validate_invocations(local, item["otel"]) + except CollectionError: + return False + return True + cloud.poll(item["fixture"], complete, seconds=30, category=CollectionError, items=[item]) + + +def single(cloud, fixture, scenario="success"): + marker = "otel-" + uuid.uuid4().hex[:10] + item = cloud.launch(fixture, scenario, marker) + cloud.finish(item, "FAILED" if scenario == "failure" else "SUCCEEDED") + wait_evidence(cloud, item) + return item + + +def warm(cloud, fixture): + environments, providers = set(), set() + for scenario in ["success", "failure", "replay"] * 2: + item = single(cloud, fixture, scenario) + for event in selected(cloud.events_for(item), "OTEL_ENTER"): + environments.add(event["environment"]) + providers.add(event["provider"]) + if scenario == "replay": + item["replayHistory"] = cloud.history(item) + if len(environments) != 1: + raise PreconditionError("Warm observability cases did not reuse one JVM") + require(len(providers) == 1, "Provider was rebuilt for every invocation") + + +def held_export(cloud, fixture): + item = cloud.launch(fixture, "success", "otel-held-" + uuid.uuid4().hex[:10]) + def returned(events): + local = [event for event in events if event["marker"] == item["marker"]] + starts = selected(local, "OTEL_EXPORT_HELD") + ends = selected(local, "WRAPPER_RETURN") + if starts and ends: + require(ends[0]["epochMillis"] - starts[0]["epochMillis"] <= 10000, + "The five-second flush wait did not bound finalization") + require(not selected(local, "OTEL_EXPORT"), "Held exporter released before the assertion") + return True + return False + cloud.poll(fixture, returned, seconds=40, category=CollectionError, items=[item]) + cloud.gate(item["exportGate"], release=True) + cloud.finish(item) + wait_evidence(cloud, item) + + +def additional_cases(cloud, lifecycle): + cases = [] + for fixture in ("default1", "default2"): + for view in ("invocation", "execution"): + def add(label, work, sampled="1", export="success", fixture=fixture, view=view): + def run(): + with observing(cloud, view, sampled, export): + work() + cases.append((fixture, f"otel-{fixture}-{view}-{label}", run)) + for sampled in ("0", "1"): + add("sampling-" + sampled, lambda fixture=fixture: single(cloud, fixture), sampled) + add("warm-replay", lambda fixture=fixture: warm(cloud, fixture)) + add("export-failure", lambda fixture=fixture: single(cloud, fixture), export="failure") + add("flush-budget", lambda fixture=fixture: held_export(cloud, fixture), export="held") + if fixture == "default2": + add("overlap", lambda: lifecycle.overlap_case(cloud, "default2")) + add("suspend-cleanup", lambda: lifecycle.replay_case(cloud, "default2", "suspend")) + add("deadline", lambda: lifecycle.timeout_case(cloud, "default2")) + add("non-cooperative", lambda: lifecycle.timeout_case(cloud, "default2", True)) + for scenario in ("return-inflight", "failure-inflight"): + add(scenario, lambda scenario=scenario: lifecycle.inflight_case(cloud, "default2", scenario)) + return cases + diff --git a/lmi-tests/otel_evidence.py b/lmi-tests/otel_evidence.py new file mode 100644 index 0000000000..ffc5e685e2 --- /dev/null +++ b/lmi-tests/otel_evidence.py @@ -0,0 +1,83 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Native Java OTel acceptance over independent runtime and raw-span evidence.""" + +import re + +from cloud_support import CollectionError, require, selected + + +def header(value): + if not isinstance(value, str): + return None + fields = dict(part.strip().split("=", 1) for part in value.split(";") if "=" in part) + root, parent, sampling = (fields.get(key) for key in ("Root", "Parent", "Sampled")) + if not root or not re.fullmatch(r"1-[0-9a-f]{8}-[0-9a-f]{24}", root): + return None + if not parent or not re.fullmatch(r"[0-9a-f]{16}", parent) or sampling not in {"0", "1"}: + return None + return {"trace": root[2:].replace("-", ""), "parent": parent, "sampled": sampling == "1"} + + +def validate_invocations(events, settings): + entries = selected(events, "OTEL_ENTER") + if not entries: + raise CollectionError("Missing OTel runtime entry/carrier evidence") + instances = [] + for entry in entries: + local = [event for event in events + if event["requestId"] == entry["requestId"] and event["environment"] == entry["environment"]] + ends = selected(local, "OTEL_END") + factories = selected(local, "OTEL_FACTORY") + if len(ends) != 1 or len(factories) != 1: + raise CollectionError("Missing OTel factory or owner-End evidence") + end, factory = ends[0], factories[0] + require(end["contextRestored"] and end["mdcRestored"], "OTel/MDC scope leaked after End") + require(end["ownerThread"] == end["endThread"], "End changed handler-owner thread") + require(end["exporterShutdowns"] == 0, "Invocation shut down a shared exporter") + require(factory["pluginExecution"] == entry["executionArn"], "Factory received another execution") + require(entry["view"] == settings["view"], "Wrong OTel view was deployed") + observed = header(entry.get("carrier")) + if not observed: + raise CollectionError("Runtime did not provide the required valid X-Ray header") + require(factory["capturedHeader"] == entry["carrier"], "Runtime caller snapshot changed before dispatch") + instances.append(factory["pluginInstance"]) + exports = selected(local, "OTEL_EXPORT") + require(not selected(local, "OTEL_EXPORT_ERROR"), "Evidence exporter failed unexpectedly") + if not observed["sampled"]: + require(not exports, "Sampled=0 exported spans") + continue + roots = [event["span"] for event in exports if event["span"]["name"] == "Invocation"] + if not roots: + raise CollectionError("Sampled invocation span has not been collected") + require(len(roots) == 1, "Duplicate Invocation spans") + require(roots[0]["parentId"] == observed["parent"], "Runtime X-Ray Parent was lost") + for event in exports: + span = event["span"] + expected = "failure" if settings["mode"] == "failure" else "success" + require(event["outcome"] == expected, "Unexpected exporter outcome") + require(span["attributes"].get("durable.execution.arn") == entry["executionArn"], + "Span belongs to another execution") + require(span["traceId"] == observed["trace"] and span["sampled"], "Runtime trace/sampling was lost") + require(span["startNanos"] <= span["endNanos"], "Negative span duration") + require(len(instances) == len(set(instances)), "Plugin instance reused across invocations") + initial = {event["requestId"] for event in selected(events, "OTEL_FACTORY") + if event.get("firstInvocation")} + candidates = [entry for entry in entries if entry["requestId"] in initial] or entries + first = min(candidates, key=lambda event: event.get("epochMillis", event["sequence"])) + sent, observed = header(settings["traceHeader"]), header(first["carrier"]) + require(sent and observed["trace"] == sent["trace"] and observed["sampled"] == sent["sampled"], + "Instrumented Invoke did not preserve the caller's Root/Sampled") + terminals = [event for event in selected(events, "OTEL_END") if event.get("status") in {"SUCCEEDED", "FAILED"}] + if terminals: + terminal = max(terminals, key=lambda event: event["sequence"]) + terminal_entry = next(event for event in entries if event["requestId"] == terminal["requestId"]) + if header(terminal_entry["carrier"])["sampled"]: + workflows = [event["span"] for event in selected(events, "OTEL_EXPORT") + if event["span"]["name"] == "Workflow"] + if not workflows: + raise CollectionError("Terminal Workflow span has not been collected") + require(len(workflows) == 1, "Workflow repeated across resume/retry") + expected = "ERROR" if terminal["status"] == "FAILED" else "OK" + require(workflows[0]["status"] == expected, "Workflow status changed") + diff --git a/lmi-tests/pom.xml b/lmi-tests/pom.xml index bbc928b7a9..5909fcd92b 100644 --- a/lmi-tests/pom.xml +++ b/lmi-tests/pom.xml @@ -11,6 +11,22 @@ LMI cloud regression fixtures true + + software.amazon.lambda.durable + aws-durable-execution-sdk-java-plugin-otel + ${project.version} + + + ch.qos.logback + logback-classic + 1.6.5 + + + software.amazon.lambda.durable + aws-durable-execution-sdk-java-testing + ${project.version} + test + software.amazon.lambda.durable aws-durable-execution-sdk-java diff --git a/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/FixtureInput.java b/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/FixtureInput.java index a51cea5569..42962e173f 100644 --- a/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/FixtureInput.java +++ b/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/FixtureInput.java @@ -11,4 +11,19 @@ public record FixtureInput( String controlUrl, String targetEnvironment, int peers, - int holdMillis) {} + int holdMillis, + String otelView, + String otelExport, + String exportControlUrl) { + public FixtureInput( + String runId, + String cohort, + String scenario, + String marker, + String controlUrl, + String targetEnvironment, + int peers, + int holdMillis) { + this(runId, cohort, scenario, marker, controlUrl, targetEnvironment, peers, holdMillis, null, null, null); + } +} diff --git a/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/InvocationTrace.java b/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/InvocationTrace.java index da14c8b65e..ce25996a85 100644 --- a/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/InvocationTrace.java +++ b/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/InvocationTrace.java @@ -27,6 +27,7 @@ final class InvocationTrace { final String executionArn; final long deadlineNanos; final AtomicInteger tasks = new AtomicInteger(); + private final AtomicInteger exports = new AtomicInteger(); volatile boolean rootExited; volatile boolean wrapperReturned; @@ -44,6 +45,20 @@ void wrapperEnter() { } } + void exportEnter() { + synchronized (EVENT_LOCK) { + exports.incrementAndGet(); + event("OTEL_EXPORT_ENTER"); + } + } + + void exportExit() { + synchronized (EVENT_LOCK) { + exports.decrementAndGet(); + event("OTEL_EXPORT_EXIT"); + } + } + void rootEnter() { synchronized (EVENT_LOCK) { LIVE_ROOTS.incrementAndGet(); @@ -131,6 +146,7 @@ private void writeEvent(String kind, Map details) { data.put("executionArn", executionArn); data.put("thread", Thread.currentThread().getName()); data.put("tasks", tasks.get()); + data.put("exports", exports.get()); data.put("rootExited", rootExited); data.put("wrapperReturned", wrapperReturned); data.put("liveRoots", LIVE_ROOTS.get()); diff --git a/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/LifecycleHandler.java b/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/LifecycleHandler.java index 4db3041e24..2689647247 100644 --- a/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/LifecycleHandler.java +++ b/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/LifecycleHandler.java @@ -52,6 +52,7 @@ public final class LifecycleHandler implements RequestStreamHandler { private static final StepConfig NO_RETRY = StepConfig.builder().retryStrategy(RetryStrategies.Presets.NO_RETRY).build(); private final DurableConfig shared = createSharedConfiguration(); + private final ConcurrentHashMap telemetry = new ConcurrentHashMap<>(); private static DurableConfig createSharedConfiguration() { if ("fixed".equals(System.getenv("LMI_EXECUTOR"))) { @@ -78,10 +79,15 @@ public void handleRequest(InputStream source, OutputStream destination, Context trace.wrapperEnter(); var status = "THREW"; try { - var config = DurableConfig.builder() + var builder = DurableConfig.builder() .withExecutorService(shared.getExecutorService()) - .withDurableExecutionClient(new ObservedClient(shared.getDurableExecutionClient(), trace)) - .build(); + .withDurableExecutionClient(new ObservedClient(shared.getDurableExecutionClient(), trace)); + if (trace.input.otelView() != null) { + var observed = telemetry.computeIfAbsent(trace.input.otelView(), OtelRuntime::new); + observed.enter(trace, runtime); + builder.withPlugins(observed.forInvocation(trace)); + } + var config = builder.build(); var response = DurableExecutor.execute( input, runtime, diff --git a/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/ObservedOtelPlugin.java b/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/ObservedOtelPlugin.java new file mode 100644 index 0000000000..5de163ec43 --- /dev/null +++ b/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/ObservedOtelPlugin.java @@ -0,0 +1,160 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.lmi; + +import io.opentelemetry.context.Context; +import java.util.Map; +import java.util.Objects; +import java.util.function.Function; +import java.util.function.IntSupplier; +import java.util.function.Supplier; +import org.slf4j.MDC; +import software.amazon.lambda.durable.otel.ExecutionOtelPlugin; +import software.amazon.lambda.durable.otel.InvocationOtelPlugin; +import software.amazon.lambda.durable.plugin.DurableExecutionPlugin; +import software.amazon.lambda.durable.plugin.HandlerScoped; +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.OperationEndInfo; +import software.amazon.lambda.durable.plugin.OperationInfo; +import software.amazon.lambda.durable.plugin.PropagationInput; +import software.amazon.lambda.durable.plugin.PropagationMetadata; +import software.amazon.lambda.durable.plugin.UserFunctionEndInfo; +import software.amazon.lambda.durable.plugin.UserFunctionStartInfo; + +/** Adds only diagnostic ownership; every hook and handler scope reaches the real plugin. */ +@HandlerScoped(ObservedOtelPlugin.ScopeOpener.class) +final class ObservedOtelPlugin implements DurableExecutionPlugin { + static final ThreadLocal EXPORT_OWNER = new ThreadLocal<>(); + private final DurableExecutionPlugin delegate; + private final InvocationTrace trace; + private final IntSupplier shutdowns; + private Context before; + private Map beforeMdc; + private long ownerThread; + + ObservedOtelPlugin(DurableExecutionPlugin delegate, InvocationTrace trace, IntSupplier shutdowns) { + this.delegate = delegate; + this.trace = trace; + this.shutdowns = shutdowns; + } + + private T observe(Supplier action) { + return withOwner(trace, action); + } + + static T withOwner(InvocationTrace trace, Supplier action) { + var previous = EXPORT_OWNER.get(); + EXPORT_OWNER.set(trace); + try { + return action.get(); + } finally { + restore(previous); + } + } + + private void observe(Runnable action) { + observe(() -> { + action.run(); + return null; + }); + } + + private static void restore(InvocationTrace previous) { + if (previous == null) EXPORT_OWNER.remove(); + else EXPORT_OWNER.set(previous); + } + + @Override + public void onInvocationStart(InvocationInfo info) { + before = Context.current(); + beforeMdc = MDC.getCopyOfContextMap(); + ownerThread = Thread.currentThread().getId(); + trace.event("OTEL_START", Map.of("ownerThread", ownerThread)); + observe(() -> delegate.onInvocationStart(info)); + } + + @Override + public void onInvocationEnd(InvocationEndInfo info) { + try { + observe(() -> delegate.onInvocationEnd(info)); + } finally { + trace.event( + "OTEL_END", + Map.of( + "contextRestored", + Context.current() == before, + "mdcRestored", + Objects.equals(beforeMdc, MDC.getCopyOfContextMap()), + "ownerThread", + ownerThread, + "endThread", + Thread.currentThread().getId(), + "status", + info.invocationStatus().name(), + "exporterShutdowns", + shutdowns.getAsInt())); + } + } + + @Override + public void onOperationStart(OperationInfo info) { + observe(() -> delegate.onOperationStart(info)); + } + + @Override + public void onOperationEnd(OperationEndInfo info) { + observe(() -> delegate.onOperationEnd(info)); + } + + @Override + public void onOperationChange(OperationChangeInfo info) { + observe(() -> delegate.onOperationChange(info)); + } + + @Override + public void onUserFunctionStart(UserFunctionStartInfo info) { + observe(() -> delegate.onUserFunctionStart(info)); + } + + @Override + public void onUserFunctionEnd(UserFunctionEndInfo info) { + observe(() -> delegate.onUserFunctionEnd(info)); + } + + @Override + public PropagationMetadata providePropagationMetadata(PropagationInput info) { + return observe(() -> delegate.providePropagationMetadata(info)); + } + + @Override + public String providePropagationMetadata(String arn, String operation, String parent, String target) { + return observe(() -> delegate.providePropagationMetadata(arn, operation, parent, target)); + } + + private AutoCloseable openScope() { + var scope = observe(() -> delegate instanceof ExecutionOtelPlugin execution + ? new ExecutionOtelPlugin.HandlerScopeOpener().apply(execution) + : new InvocationOtelPlugin.HandlerScopeOpener().apply((InvocationOtelPlugin) delegate)); + if (scope == null) return null; + return () -> { + var previous = EXPORT_OWNER.get(); + EXPORT_OWNER.set(trace); + try { + scope.close(); + } finally { + restore(previous); + } + }; + } + + public static final class ScopeOpener implements Function { + public ScopeOpener() {} + + @Override + public AutoCloseable apply(ObservedOtelPlugin plugin) { + return plugin.openScope(); + } + } +} diff --git a/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/OtelInvoke.java b/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/OtelInvoke.java new file mode 100644 index 0000000000..fb887d8e88 --- /dev/null +++ b/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/OtelInvoke.java @@ -0,0 +1,57 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.lmi; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Map; +import software.amazon.awssdk.core.SdkBytes; +import software.amazon.awssdk.core.retry.RetryPolicy; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.lambda.LambdaClient; +import software.amazon.awssdk.services.lambda.model.InvocationType; +import software.amazon.awssdk.services.lambda.model.InvokeRequest; +import software.amazon.lambda.durable.TypeToken; +import software.amazon.lambda.durable.serde.JacksonSerDes; + +/** Instrumented test caller: the trace header traverses the real signed Invoke request. */ +public final class OtelInvoke { + private static final JacksonSerDes JSON = new JacksonSerDes(); + + private OtelInvoke() {} + + public record Input(String function, String executionName, String payload, String traceHeader) {} + + public static void main(String[] args) throws IOException { + var input = JSON.deserialize( + new String(System.in.readAllBytes(), StandardCharsets.UTF_8), TypeToken.get(Input.class)); + try (var client = LambdaClient.builder() + .region(Region.of(System.getenv("AWS_REGION"))) + .overrideConfiguration(config -> config.apiCallTimeout(Duration.ofSeconds(30)) + .apiCallAttemptTimeout(Duration.ofSeconds(25)) + .retryPolicy(RetryPolicy.none())) + .build()) { + var result = client.invoke(request(input)); + System.out.println(JSON.serialize(Map.of( + "headers", + Map.of("DurableExecutionArn", result.durableExecutionArn(), "StatusCode", result.statusCode()), + "body", + ""))); + } + } + + static InvokeRequest request(Input input) { + if (input.traceHeader() == null + || !input.traceHeader().matches("Root=1-[0-9a-f]{8}-[0-9a-f]{24};Parent=[0-9a-f]{16};Sampled=[01]")) { + throw new IllegalArgumentException("A valid test caller trace header is required"); + } + return InvokeRequest.builder() + .functionName(input.function()) + .durableExecutionName(input.executionName()) + .invocationType(InvocationType.EVENT) + .payload(SdkBytes.fromUtf8String(input.payload())) + .overrideConfiguration(config -> config.putHeader("X-Amzn-Trace-Id", input.traceHeader())) + .build(); + } +} diff --git a/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/OtelLedgerProcessor.java b/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/OtelLedgerProcessor.java new file mode 100644 index 0000000000..130ff0be72 --- /dev/null +++ b/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/OtelLedgerProcessor.java @@ -0,0 +1,118 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.lmi; + +import io.opentelemetry.context.Context; +import io.opentelemetry.sdk.common.CompletableResultCode; +import io.opentelemetry.sdk.trace.ReadWriteSpan; +import io.opentelemetry.sdk.trace.ReadableSpan; +import io.opentelemetry.sdk.trace.SpanProcessor; +import io.opentelemetry.sdk.trace.data.SpanData; +import io.opentelemetry.sdk.trace.export.SpanExporter; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Consumer; + +/** Tracks real export completions so forceFlush can exercise an outstanding exporter. */ +final class OtelLedgerProcessor implements SpanProcessor, SpanExporter { + private final Set pending = ConcurrentHashMap.newKeySet(); + private final Consumer hold; + final AtomicInteger shutdowns = new AtomicInteger(); + private final ExecutorService exporter = Executors.newSingleThreadExecutor(task -> { + var thread = new Thread(task, "lmi-otel-export"); + thread.setDaemon(true); + return thread; + }); + + OtelLedgerProcessor(Consumer hold) { + this.hold = hold; + } + + @Override + public void onStart(Context parent, ReadWriteSpan span) {} + + @Override + public boolean isStartRequired() { + return false; + } + + @Override + public boolean isEndRequired() { + return true; + } + + @Override + public void onEnd(ReadableSpan ended) { + export(List.of(ended.toSpanData())); + } + + @Override + public CompletableResultCode export(Collection spans) { + var trace = ObservedOtelPlugin.EXPORT_OWNER.get(); + if (trace == null) throw new IllegalStateException("Export lost independent invocation ownership"); + var results = new ArrayList(); + for (var data : spans) results.add(enqueue(trace, data)); + return CompletableResultCode.ofAll(results); + } + + private CompletableResultCode enqueue(InvocationTrace trace, SpanData data) { + var result = new CompletableResultCode(); + pending.add(result); + result.whenComplete(() -> pending.remove(result)); + trace.exportEnter(); + if ("held".equals(trace.input.otelExport())) { + trace.event("OTEL_EXPORT_HELD"); + exporter.execute(() -> export(trace, data, result)); + } else { + export(trace, data, result); + } + return result; + } + + private void export(InvocationTrace trace, SpanData data, CompletableResultCode result) { + var success = false; + try { + if ("held".equals(trace.input.otelExport())) hold.accept(trace); + var failed = "failure".equals(trace.input.otelExport()); + trace.event( + "OTEL_EXPORT", Map.of("span", OtelRuntime.span(data), "outcome", failed ? "failure" : "success")); + success = !failed; + } catch (RuntimeException failure) { + trace.event("OTEL_EXPORT_ERROR", Map.of("error", failure.getClass().getSimpleName())); + } finally { + trace.exportExit(); + if (success) result.succeed(); + else result.fail(); + } + } + + @Override + public CompletableResultCode forceFlush() { + return CompletableResultCode.ofAll(new ArrayList<>(pending)); + } + + @Override + public CompletableResultCode flush() { + return forceFlush(); + } + + @Override + public CompletableResultCode shutdown() { + shutdowns.incrementAndGet(); + exporter.shutdown(); + return forceFlush(); + } + + @Override + public void close() { + shutdown().join(5, TimeUnit.SECONDS); + } +} diff --git a/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/OtelRuntime.java b/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/OtelRuntime.java new file mode 100644 index 0000000000..09dc472c1f --- /dev/null +++ b/lmi-tests/src/main/java/software/amazon/lambda/durable/lmi/OtelRuntime.java @@ -0,0 +1,133 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.lmi; + +import com.amazonaws.services.lambda.runtime.Context; +import io.opentelemetry.sdk.trace.SdkTracerProvider; +import io.opentelemetry.sdk.trace.data.SpanData; +import io.opentelemetry.sdk.trace.samplers.Sampler; +import java.io.IOException; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.time.Duration; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.UUID; +import java.util.WeakHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; +import software.amazon.lambda.durable.otel.ExecutionOtelPlugin; +import software.amazon.lambda.durable.otel.InvocationOtelPlugin; +import software.amazon.lambda.durable.otel.OtelPluginConfig; +import software.amazon.lambda.durable.plugin.DurableExecutionPlugin; +import software.amazon.lambda.durable.plugin.DurableExecutionPluginFactory; + +/** Process-owned real OTel provider with a controllable evidence exporter. */ +final class OtelRuntime implements AutoCloseable { + private static final HttpClient HTTP = + HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(2)).build(); + private final String identity = UUID.randomUUID().toString(); + private final String view; + private final DurableExecutionPluginFactory factory; + private final Map instances = Collections.synchronizedMap(new WeakHashMap<>()); + private final OtelLedgerProcessor processor; + + OtelRuntime(String view) { + this(view, OtelRuntime::waitForExportRelease); + } + + OtelRuntime(String view, Consumer hold) { + this.view = view; + processor = new OtelLedgerProcessor(hold); + var builder = SdkTracerProvider.builder().setSampler(Sampler.alwaysOn()).addSpanProcessor(processor); + var config = OtelPluginConfig.builder().enableMdc(true).build(); + factory = switch (view) { + case "invocation" -> InvocationOtelPlugin.factory(builder, config); + case "execution" -> ExecutionOtelPlugin.factory(builder, config); + default -> throw new IllegalArgumentException("Unknown OTel view"); + }; + } + + void enter(InvocationTrace trace, Context context) { + var fields = new LinkedHashMap(); + fields.put("view", view); + fields.put("provider", identity); + fields.put("exporterShutdowns", processor.shutdowns.get()); + fields.put("carrierSource", "runtime_context"); + try { + fields.put("carrier", context.getXrayTraceId()); + } catch (RuntimeException error) { + fields.put("carrier", null); + fields.put("carrierError", error.getClass().getSimpleName()); + } + trace.event("OTEL_ENTER", fields); + } + + DurableExecutionPluginFactory forInvocation(InvocationTrace trace) { + return info -> { + var plugin = ObservedOtelPlugin.withOwner(trace, () -> factory.createPlugin(info)); + var instance = + instances.computeIfAbsent(plugin, key -> UUID.randomUUID().toString()); + trace.event( + "OTEL_FACTORY", + Map.of( + "pluginInstance", + instance, + "pluginExecution", + info.durableExecutionArn(), + "firstInvocation", + info.isFirstInvocation(), + "capturedHeader", + info.xRayTraceId() == null ? "" : info.xRayTraceId())); + return new ObservedOtelPlugin(plugin, trace, processor.shutdowns::get); + }; + } + + @Override + public void close() { + processor.shutdown().join(5, TimeUnit.SECONDS); + } + + private static void waitForExportRelease(InvocationTrace trace) { + var until = System.nanoTime() + TimeUnit.SECONDS.toNanos(120); + while (System.nanoTime() < until) { + try { + var request = HttpRequest.newBuilder(URI.create(trace.input.exportControlUrl())) + .timeout(Duration.ofSeconds(2)) + .GET() + .build(); + var response = HTTP.send(request, HttpResponse.BodyHandlers.ofString()); + if (response.statusCode() != 200) throw new IllegalStateException("Export gate read failed"); + if ("release".equals(response.body().trim())) return; + if (!"hold".equals(response.body().trim())) throw new IllegalStateException("Invalid export gate"); + new CountDownLatch(1).await(250, TimeUnit.MILLISECONDS); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("Export gate interrupted", interrupted); + } catch (IOException failure) { + throw new IllegalStateException("Export gate unavailable", failure); + } + } + throw new IllegalStateException("Export fixture escape; never acceptance"); + } + + static Map span(SpanData data) { + var result = new LinkedHashMap(); + result.put("name", data.getName()); + result.put("traceId", data.getTraceId()); + result.put("spanId", data.getSpanId()); + result.put("parentId", data.getParentSpanId()); + result.put("sampled", data.getSpanContext().isSampled()); + result.put("startNanos", data.getStartEpochNanos()); + result.put("endNanos", data.getEndEpochNanos()); + result.put("status", data.getStatus().getStatusCode().name()); + var attributes = new LinkedHashMap(); + data.getAttributes().forEach((key, value) -> attributes.put(key.getKey(), value)); + result.put("attributes", attributes); + return result; + } +} diff --git a/lmi-tests/src/main/resources/logback.xml b/lmi-tests/src/main/resources/logback.xml new file mode 100644 index 0000000000..7b4fdfdc72 --- /dev/null +++ b/lmi-tests/src/main/resources/logback.xml @@ -0,0 +1,7 @@ + + + System.err + %level %logger - %msg%n + + + diff --git a/lmi-tests/src/test/java/software/amazon/lambda/durable/lmi/OtelInvocationTest.java b/lmi-tests/src/test/java/software/amazon/lambda/durable/lmi/OtelInvocationTest.java new file mode 100644 index 0000000000..2eaa4939a8 --- /dev/null +++ b/lmi-tests/src/test/java/software/amazon/lambda/durable/lmi/OtelInvocationTest.java @@ -0,0 +1,334 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.lmi; + +import static org.junit.jupiter.api.Assertions.*; + +import com.amazonaws.services.lambda.runtime.Context; +import com.amazonaws.services.lambda.runtime.LambdaLogger; +import io.opentelemetry.api.trace.Span; +import java.lang.reflect.Proxy; +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BiFunction; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.ValueSource; +import org.slf4j.MDC; +import software.amazon.awssdk.services.lambda.model.CheckpointUpdatedExecutionState; +import software.amazon.awssdk.services.lambda.model.ExecutionDetails; +import software.amazon.awssdk.services.lambda.model.Operation; +import software.amazon.awssdk.services.lambda.model.OperationStatus; +import software.amazon.awssdk.services.lambda.model.OperationType; +import software.amazon.lambda.durable.DurableConfig; +import software.amazon.lambda.durable.DurableContext; +import software.amazon.lambda.durable.TypeToken; +import software.amazon.lambda.durable.execution.DurableExecutor; +import software.amazon.lambda.durable.model.DurableExecutionInput; +import software.amazon.lambda.durable.model.DurableExecutionOutput; +import software.amazon.lambda.durable.model.ExecutionStatus; +import software.amazon.lambda.durable.serde.JacksonSerDes; +import software.amazon.lambda.durable.testing.local.LocalMemoryExecutionClient; + +class OtelInvocationTest { + private static final String HEADER = "Root=1-6955b900-123456789012345678901234;Parent=1234567890123456;Sampled=1"; + private final ExecutorService workers = Executors.newCachedThreadPool(); + + @AfterEach + void stopWorkers() throws InterruptedException { + workers.shutdownNow(); + assertTrue(workers.awaitTermination(5, TimeUnit.SECONDS)); + } + + @ParameterizedTest + @CsvSource({"invocation,1", "invocation,0", "execution,1", "execution,0"}) + void runtimeCarrierSamplingAndOwnerCleanup(String view, String sampled) { + try (var capture = new TraceCapture(); + var runtime = new OtelRuntime(view)) { + var request = new Request("sampled", view, "success", HEADER.substring(0, HEADER.length() - 1) + sampled); + var output = request.run(runtime, (value, context) -> { + assertTrue(Span.current().getSpanContext().isValid(), "Handler scope was not forwarded"); + return context.step("saved", String.class, step -> value.marker()); + }); + assertEquals(ExecutionStatus.SUCCEEDED, output.status()); + assertEquals(Set.of(Thread.currentThread().getName()), request.headerThreads); + var exported = spans(capture.events()); + assertEquals(sampled.equals("0"), exported.isEmpty()); + if (sampled.equals("1")) assertSpans(exported, request.arn); + assertCleanup(capture.events()); + } + } + + @ParameterizedTest + @ValueSource(strings = {"invocation", "execution"}) + void realSuspendResumeDoesNotRepeatCompletedWork(String view) { + var bodies = new AtomicInteger(); + try (var capture = new TraceCapture(); + var runtime = new OtelRuntime(view)) { + var request = new Request("replay", view, "success", HEADER); + BiFunction handler = (input, context) -> { + var value = context.step("saved", String.class, step -> { + bodies.incrementAndGet(); + return input.marker(); + }); + context.wait("pause", Duration.ofSeconds(1)); + return value; + }; + assertEquals(ExecutionStatus.PENDING, request.run(runtime, handler).status()); + assertTrue(request.storage.advanceTime()); + assertEquals( + ExecutionStatus.SUCCEEDED, request.run(runtime, handler).status()); + assertEquals(1, bodies.get()); + assertEquals( + 1, + spans(capture.events()).stream() + .filter(s -> s.get("name").equals("Workflow")) + .count()); + assertCleanup(capture.events()); + } + } + + @ParameterizedTest + @CsvSource({ + "invocation,null", "invocation,empty", "invocation,invalid", "invocation,throw", + "execution,null", "execution,empty", "execution,invalid", "execution,throw" + }) + void authoritativeMissingOrInvalidCarrierCannotBorrowGlobalState(String view, String value) { + var key = "com.amazonaws.xray.traceHeader"; + var previous = System.getProperty(key); + System.setProperty(key, HEADER); + try (var capture = new TraceCapture(); + var runtime = new OtelRuntime(view)) { + var header = value.equals("null") ? null : value.equals("empty") ? "" : value; + var request = new Request("missing-" + value, view, "success", header); + assertEquals( + ExecutionStatus.SUCCEEDED, + request.run( + runtime, + (input, context) -> context.step("saved", String.class, step -> input.marker())) + .status()); + var spans = spans(capture.events()); + assertFalse(spans.isEmpty()); + assertTrue(spans.stream().anyMatch(span -> "DurableExecutionRoot".equals(span.get("name")))); + assertTrue( + spans.stream().noneMatch(span -> "6955b900123456789012345678901234".equals(span.get("traceId")))); + assertEquals(HEADER, System.getProperty(key)); + assertCleanup(capture.events()); + } finally { + if (previous == null) System.clearProperty(key); + else System.setProperty(key, previous); + } + } + + @ParameterizedTest + @ValueSource(strings = {"invocation", "execution"}) + void overlappingOwnersKeepContextAndProviderIndependent(String view) throws Exception { + var entered = new CountDownLatch(2); + var release = new CountDownLatch(1); + try (var capture = new TraceCapture(); + var runtime = new OtelRuntime(view)) { + var first = new Request("first", view, "success", HEADER); + var second = new Request( + "second", view, "success", HEADER.replace("123456789012345678901234", "abcdefabcdefabcdefabcdef")); + BiFunction handler = + (input, context) -> context.step("same-name", String.class, step -> { + entered.countDown(); + await(release); + return input.marker(); + }); + var one = workers.submit(() -> first.run(runtime, handler)); + var two = workers.submit(() -> second.run(runtime, handler)); + try { + assertTrue(entered.await(5, TimeUnit.SECONDS)); + } finally { + release.countDown(); + } + assertEquals( + ExecutionStatus.SUCCEEDED, one.get(10, TimeUnit.SECONDS).status()); + assertEquals( + ExecutionStatus.SUCCEEDED, two.get(10, TimeUnit.SECONDS).status()); + assertOwners(capture.events()); + assertCleanup(capture.events()); + } + } + + @ParameterizedTest + @ValueSource(strings = {"invocation", "execution"}) + void failedExporterDoesNotPoisonTheNextInvocation(String view) { + try (var capture = new TraceCapture(); + var runtime = new OtelRuntime(view)) { + var failed = new Request("failed-export", view, "failure", HEADER); + var healthy = new Request("healthy", view, "success", HEADER); + BiFunction handler = + (input, context) -> context.step("saved", String.class, step -> input.marker()); + assertEquals(ExecutionStatus.SUCCEEDED, failed.run(runtime, handler).status()); + assertEquals( + ExecutionStatus.SUCCEEDED, healthy.run(runtime, handler).status()); + assertTrue(capture.events().stream().anyMatch(event -> "failure".equals(event.get("outcome")))); + assertCleanup(capture.events()); + } + } + + @ParameterizedTest + @ValueSource(strings = {"invocation", "execution"}) + void flushBudgetCanReturnBeforeControlledExportCompletes(String view) throws Exception { + var entered = new CountDownLatch(1); + var release = new CountDownLatch(1); + try (var capture = new TraceCapture(); + var runtime = new OtelRuntime(view, trace -> { + entered.countDown(); + await(release); + })) { + var request = new Request("held", view, "held", HEADER); + var start = System.nanoTime(); + var result = workers.submit(() -> request.run( + runtime, (input, context) -> context.step("saved", String.class, step -> input.marker()))); + try { + assertTrue(entered.await(5, TimeUnit.SECONDS)); + assertEquals( + ExecutionStatus.SUCCEEDED, + result.get(8, TimeUnit.SECONDS).status()); + assertTrue(System.nanoTime() - start < TimeUnit.SECONDS.toNanos(8)); + assertTrue(spans(capture.events()).isEmpty()); + } finally { + release.countDown(); + } + } + } + + private static void await(CountDownLatch latch) { + try { + assertTrue(latch.await(15, TimeUnit.SECONDS), "Fixture was not released"); + } catch (InterruptedException error) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(error); + } + } + + private static List> spans(List> events) { + return events.stream() + .filter(e -> "OTEL_EXPORT".equals(e.get("kind"))) + .>map(e -> (Map) e.get("span")) + .toList(); + } + + private static void assertSpans(List> spans, String arn) { + assertFalse(spans.isEmpty()); + for (var span : spans) { + assertEquals("6955b900123456789012345678901234", span.get("traceId")); + assertEquals(arn, ((Map) span.get("attributes")).get("durable.execution.arn")); + if ("Invocation".equals(span.get("name"))) assertEquals("1234567890123456", span.get("parentId")); + } + } + + private static void assertCleanup(List> events) { + var ends = events.stream().filter(e -> "OTEL_END".equals(e.get("kind"))).toList(); + assertFalse(ends.isEmpty()); + for (var end : ends) { + assertEquals(true, end.get("contextRestored")); + assertEquals(true, end.get("mdcRestored")); + assertEquals(end.get("ownerThread"), end.get("endThread")); + assertEquals(0, ((Number) end.get("exporterShutdowns")).intValue()); + } + } + + private static void assertOwners(List> events) { + for (var event : events) { + if ("OTEL_EXPORT".equals(event.get("kind"))) { + var span = (Map) event.get("span"); + var attributes = (Map) span.get("attributes"); + assertEquals(event.get("executionArn"), attributes.get("durable.execution.arn")); + } + } + } + + private final class Request { + private final LocalMemoryExecutionClient storage = new LocalMemoryExecutionClient(); + private final Set headerThreads = ConcurrentHashMap.newKeySet(); + private final Instant start = Instant.now(); + private final FixtureInput value; + private final String arn; + private final String header; + private int attempt; + + Request(String marker, String view, String mode, String header) { + value = new FixtureInput("unit", marker, "success", marker, null, null, 1, 1000, view, mode, null); + arn = "arn:aws:lambda:us-west-2:123456789012:function:lmi:1/durable-execution/" + marker + "/execution"; + this.header = header; + } + + DurableExecutionOutput run(OtelRuntime runtime, BiFunction handler) { + var context = context(); + var trace = new InvocationTrace(value, context, arn); + runtime.enter(trace, context); + var config = DurableConfig.builder() + .withExecutorService(workers) + .withDurableExecutionClient(storage) + .withPlugins(runtime.forInvocation(trace)) + .build(); + MDC.put("lmi-existing", value.marker()); + try { + return DurableExecutor.execute(input(), context, TypeToken.get(FixtureInput.class), handler, config); + } finally { + assertEquals(value.marker(), MDC.get("lmi-existing")); + MDC.remove("lmi-existing"); + } + } + + private DurableExecutionInput input() { + var root = Operation.builder() + .id("execution") + .type(OperationType.EXECUTION) + .status(OperationStatus.STARTED) + .startTimestamp(start) + .executionDetails(ExecutionDetails.builder() + .inputPayload(new JacksonSerDes().serialize(value)) + .build()) + .build(); + var operations = new ArrayList<>(List.of(root)); + operations.addAll(storage.getAllOperations()); + return new DurableExecutionInput( + arn, + "token-" + attempt, + CheckpointUpdatedExecutionState.builder() + .operations(operations) + .build(), + storage.getUpdatedOperationIdsSinceLastInvocation()); + } + + private Context context() { + var request = value.marker() + "-" + (++attempt); + var logger = new LambdaLogger() { + public void log(String message) {} + + public void log(byte[] message) {} + }; + return (Context) Proxy.newProxyInstance( + Context.class.getClassLoader(), + new Class[] {Context.class}, + (proxy, method, args) -> switch (method.getName()) { + case "getAwsRequestId" -> request; + case "getRemainingTimeInMillis" -> 60000; + case "getXrayTraceId" -> { + headerThreads.add(Thread.currentThread().getName()); + if ("throw".equals(header)) throw new IllegalStateException("controlled accessor failure"); + yield header; + } + case "getLogger" -> logger; + case "getMemoryLimitInMB" -> 2048; + default -> null; + }); + } + } +} diff --git a/lmi-tests/src/test/java/software/amazon/lambda/durable/lmi/OtelInvokeTest.java b/lmi-tests/src/test/java/software/amazon/lambda/durable/lmi/OtelInvokeTest.java new file mode 100644 index 0000000000..dacbdaae4f --- /dev/null +++ b/lmi-tests/src/test/java/software/amazon/lambda/durable/lmi/OtelInvokeTest.java @@ -0,0 +1,58 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.lmi; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import com.sun.net.httpserver.HttpServer; +import java.net.InetSocketAddress; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.lambda.LambdaClient; + +class OtelInvokeTest { + @Test + void realAwsSdkSendsTheTraceHeaderWithoutChangingPayload() throws Exception { + var header = "Root=1-6955b900-123456789012345678901234;Parent=1234567890123456;Sampled=0"; + var observed = new AtomicReference(); + var payload = new AtomicReference(); + var server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/", exchange -> { + observed.set(exchange.getRequestHeaders().getFirst("X-Amzn-Trace-Id")); + payload.set(new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8)); + exchange.getResponseHeaders().add("X-Amz-Durable-Execution-Arn", "execution"); + exchange.sendResponseHeaders(202, -1); + exchange.close(); + }); + server.start(); + try (var client = LambdaClient.builder() + .endpointOverride( + URI.create("http://127.0.0.1:" + server.getAddress().getPort())) + .region(Region.US_WEST_2) + .credentialsProvider(StaticCredentialsProvider.create(AwsBasicCredentials.create("test", "test"))) + .overrideConfiguration(config -> config.apiCallTimeout(Duration.ofSeconds(5))) + .build()) { + var input = new OtelInvoke.Input("function:qualified", "stable-name", "{\"marker\":\"input\"}", header); + var response = client.invoke(OtelInvoke.request(input)); + assertEquals(202, response.statusCode()); + assertEquals(header, observed.get()); + assertEquals(input.payload(), payload.get()); + } finally { + server.stop(0); + } + } + + @Test + void invalidHttpCarrierIsRejectedBeforeInvoke() { + assertThrows( + IllegalArgumentException.class, + () -> OtelInvoke.request(new OtelInvoke.Input("function:1", "name", "{}", "invalid\r\nheader"))); + } +} diff --git a/lmi-tests/tests/test_evidence.py b/lmi-tests/tests/test_evidence.py index d7a814e01a..c77c6e5c46 100644 --- a/lmi-tests/tests/test_evidence.py +++ b/lmi-tests/tests/test_evidence.py @@ -219,9 +219,11 @@ def test_one_stack_declares_all_five_functions_and_their_scaling_limits(self): def test_full_suite_preserves_all_cases_on_their_own_fixture(self): cloud = Mock() + cloud.invocations = [] names = [] patches = [patch("cloud_suite." + name) for name in ["replay_case", "overlap_case", "fixed_case", "timeout_case", "inflight_case", "warm_case"]] + patches += [patch("otel_cases." + name) for name in ["single", "warm", "held_export"]] mocks = [p.start() for p in patches] try: for fixture in FIXTURES: @@ -237,8 +239,8 @@ def test_full_suite_preserves_all_cases_on_their_own_fixture(self): finally: for p in patches: p.stop() - self.assertEqual(13, len(names)) - self.assertEqual(13, len(set(names))) + self.assertEqual(45, len(names)) + self.assertEqual(45, len(set(names))) @patch("cloud_suite.Cloud") def test_cloud_suite_requires_all_five_functions(self, cloud): diff --git a/lmi-tests/tests/test_otel_evidence.py b/lmi-tests/tests/test_otel_evidence.py new file mode 100644 index 0000000000..56937e9e7c --- /dev/null +++ b/lmi-tests/tests/test_otel_evidence.py @@ -0,0 +1,118 @@ +"""Negative controls for observability evidence and the instrumented Invoke boundary.""" + +import copy +import json +from pathlib import Path +import sys +from types import SimpleNamespace +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +import cloud_suite +from cloud_support import Cloud, CollectionError +from otel_cases import additional_cases +from otel_evidence import validate_invocations + + +HEADER = "Root=1-6955b900-123456789012345678901234;Parent=1234567890123456;Sampled=1" +SETTINGS = {"view": "execution", "mode": "success", "traceHeader": HEADER} + + +def events(): + base = {"requestId": "request", "executionArn": "execution", "environment": "jvm", + "marker": "a", "sequence": 1, "epochMillis": 1} + span = {"name": "Invocation", "traceId": "6955b900123456789012345678901234", + "parentId": "1234567890123456", "sampled": True, "startNanos": 1, + "endNanos": 2, "status": "OK", "attributes": {"durable.execution.arn": "execution"}} + return [ + {**base, "kind": "OTEL_ENTER", "carrier": HEADER, "view": "execution"}, + {**base, "kind": "OTEL_FACTORY", "pluginInstance": "unique", "pluginExecution": "execution", + "capturedHeader": HEADER, "firstInvocation": True}, + {**base, "kind": "OTEL_EXPORT", "outcome": "success", "span": span}, + {**base, "kind": "OTEL_EXPORT", "outcome": "success", "span": {**span, "name": "Workflow"}}, + {**base, "kind": "OTEL_END", "contextRestored": True, "mdcRestored": True, + "ownerThread": 1, "endThread": 1, "exporterShutdowns": 0, "status": "SUCCEEDED"}, + ] + + +class OtelEvidenceTest(unittest.TestCase): + def test_export_completion_and_outcome_are_required_for_quiescence(self): + from test_evidence import event + trace = [ + event("WRAPPER_ENTER", 1), event("ROOT_ENTER", 2), + event("OTEL_EXPORT_ENTER", 3, exports=1), + event("ROOT_EXIT", 4, exports=1), + event("WRAPPER_RETURN", 5, exports=1, status="SUCCEEDED"), + ] + self.assertFalse(cloud_suite.runtime_requests_drained(trace, "a")) + trace.append(event("OTEL_EXPORT", 6, exports=1)) + self.assertFalse(cloud_suite.runtime_requests_drained(trace, "a")) + trace.append(event("OTEL_EXPORT_EXIT", 7, exports=0)) + self.assertTrue(cloud_suite.runtime_requests_drained(trace, "a")) + del trace[-2] + self.assertFalse(cloud_suite.runtime_requests_drained(trace, "a")) + + def test_valid_evidence(self): + validate_invocations(events(), SETTINGS) + + def test_corrupted_owner_parent_trace_and_context_fail(self): + for mutation in ("owner", "trace", "parent", "context", "thread", "duplicate", "carrier"): + with self.subTest(mutation=mutation): + records = events() + if mutation == "owner": + records[2]["span"]["attributes"]["durable.execution.arn"] = "other" + elif mutation == "trace": + records[2]["span"]["traceId"] = "0" * 32 + elif mutation == "parent": + records[2]["span"]["parentId"] = "0" * 16 + elif mutation == "context": + records[-1]["mdcRestored"] = False + elif mutation == "thread": + records[-1]["endThread"] = 2 + elif mutation == "duplicate": + records.append(copy.deepcopy(records[2])) + else: + records[1]["capturedHeader"] = "stale" + with self.assertRaises(AssertionError): + validate_invocations(records, SETTINGS) + + def test_missing_span_or_runtime_evidence_is_not_success(self): + for missing in (0, 1, 2, 4): + records = events() + del records[missing] + with self.assertRaises(CollectionError): + validate_invocations(records, SETTINGS) + + def test_unsampled_requires_zero_exports(self): + records = [event for event in events() if event["kind"] != "OTEL_EXPORT"] + records[0]["carrier"] = records[1]["capturedHeader"] = HEADER[:-1] + "0" + settings = {**SETTINGS, "traceHeader": HEADER[:-1] + "0"} + validate_invocations(records, settings) + records.insert(2, events()[2]) + with self.assertRaisesRegex(AssertionError, "Sampled=0"): + validate_invocations(records, settings) + + def test_extended_catalog_has_32_distinct_observability_cases(self): + catalog = additional_cases(SimpleNamespace(), cloud_suite) + self.assertEqual(32, len(catalog)) + self.assertEqual(32, len({name for _, name, _ in catalog})) + self.assertEqual({"default1", "default2"}, {fixture for fixture, _, _ in catalog}) + + def test_instrumented_invoker_keeps_carrier_out_of_handler_payload(self): + cloud = Cloud.__new__(Cloud) + cloud.manifest = {"functions": {"default2": {"arn": "function:qualified"}}} + payload = {"runId": "run", "marker": "marker", "scenario": "success", + "otelView": "execution", "traceHeader": HEADER} + reply = {"headers": {"DurableExecutionArn": "execution"}, "body": ""} + with patch("cloud_support.subprocess.run", return_value=SimpleNamespace( + returncode=0, stdout=json.dumps(reply), stderr="")) as invoke: + self.assertEqual(reply, cloud._invoke_request("default2", payload)) + args, kwargs = invoke.call_args + self.assertEqual("java", args[0][0]) + self.assertEqual("software.amazon.lambda.durable.lmi.OtelInvoke", args[0][-1]) + request = json.loads(kwargs["input"]) + self.assertEqual(HEADER, request["traceHeader"]) + self.assertNotIn("traceHeader", json.loads(request["payload"])) + self.assertEqual("run-marker", request["executionName"]) + From 969c822049c1f5cf6b9f0ba608c37a2ad0ee2d5d Mon Sep 17 00:00:00 2001 From: Frank Chen <65260095+zhongkechen@users.noreply.github.com> Date: Fri, 23 Oct 2026 14:51:56 -0700 Subject: [PATCH 2/2] style: normalize LMI harness file endings --- lmi-tests/otel_cases.py | 1 - lmi-tests/otel_evidence.py | 1 - lmi-tests/tests/test_otel_evidence.py | 1 - 3 files changed, 3 deletions(-) diff --git a/lmi-tests/otel_cases.py b/lmi-tests/otel_cases.py index aa934c20c2..0a163d5314 100644 --- a/lmi-tests/otel_cases.py +++ b/lmi-tests/otel_cases.py @@ -95,4 +95,3 @@ def run(): for scenario in ("return-inflight", "failure-inflight"): add(scenario, lambda scenario=scenario: lifecycle.inflight_case(cloud, "default2", scenario)) return cases - diff --git a/lmi-tests/otel_evidence.py b/lmi-tests/otel_evidence.py index ffc5e685e2..a04cb30cfd 100644 --- a/lmi-tests/otel_evidence.py +++ b/lmi-tests/otel_evidence.py @@ -80,4 +80,3 @@ def validate_invocations(events, settings): require(len(workflows) == 1, "Workflow repeated across resume/retry") expected = "ERROR" if terminal["status"] == "FAILED" else "OK" require(workflows[0]["status"] == expected, "Workflow status changed") - diff --git a/lmi-tests/tests/test_otel_evidence.py b/lmi-tests/tests/test_otel_evidence.py index 56937e9e7c..a77a76c3c9 100644 --- a/lmi-tests/tests/test_otel_evidence.py +++ b/lmi-tests/tests/test_otel_evidence.py @@ -115,4 +115,3 @@ def test_instrumented_invoker_keeps_carrier_out_of_handler_payload(self): self.assertEqual(HEADER, request["traceHeader"]) self.assertNotIn("traceHeader", json.loads(request["payload"])) self.assertEqual("run-marker", request["executionName"]) -