test: add LMI observability acceptance

.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') != ''

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.

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": [

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: ("<redacted>" if k in {"controlUrl", "CheckpointToken"} else scrub(v))
+        return {k: ("<redacted>" 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"

lmi-tests/otel_cases.py

@@ -0,0 +1,97 @@
+# 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

lmi-tests/otel_evidence.py

@@ -0,0 +1,82 @@
+# 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")

lmi-tests/pom.xml

@@ -11,6 +11,22 @@
     <name>LMI cloud regression fixtures</name>
     <properties><maven.deploy.skip>true</maven.deploy.skip></properties>
     <dependencies>
+        <dependency>
+            <groupId>software.amazon.lambda.durable</groupId>
+            <artifactId>aws-durable-execution-sdk-java-plugin-otel</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+            <version>1.6.5</version>
+        </dependency>
+        <dependency>
+            <groupId>software.amazon.lambda.durable</groupId>
+            <artifactId>aws-durable-execution-sdk-java-testing</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>software.amazon.lambda.durable</groupId>
             <artifactId>aws-durable-execution-sdk-java</artifactId>