From e1045eadbfec58b1f6f18e3f75d351281b1abb6f Mon Sep 17 00:00:00 2001 From: Eric Law <39393654+acn-ericlaw@users.noreply.github.com> Date: Sat, 14 Nov 2026 19:53:10 +0800 Subject: [PATCH] feat(tracing): the host leaves the records a Java or Rust callee leaves, so a span lineage is continuous into the pack A live drive of the engines' agent loop through both packs found the pack legs absent from the trace: the engine's HTTP client leg is in skip.rpc.tracing and the host suppressed its record for every RPC-served execution. Now a traced call at /api/event leaves http.request (the edge's round trip, under the caller's span), event.api.service and the function's round-trip record with the responder's span - the records REST automation, EventApiService and InboxBase leave in the Java engine; a reply carries the responder's span (WorkerHandler parity) and the PostOffice records a local RPC round trip as an engine inbox does. Re-driven: 20 records per run, every parent owned by a record. Co-Authored-By: Claude Fable 5.1 --- CHANGELOG.md | 14 +++ docs/guides/config-logging-actuators.md | 26 +++- extensions/mcp-binding/README.md | 10 ++ memory/continuity.md | 20 +++ memory/sessions/2026-11-14-115310.md | 58 +++++++++ src/mercury_composable/bus.py | 41 +++++++ src/mercury_composable/client.py | 36 ++++-- src/mercury_composable/server.py | 155 +++++++++++++++++++++--- tests/test_event_stream.py | 139 ++++++++++++++++----- 9 files changed, 436 insertions(+), 63 deletions(-) create mode 100644 memory/sessions/2026-11-14-115310.md diff --git a/CHANGELOG.md b/CHANGELOG.md index 2b016b3..c4792ce 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -91,6 +91,20 @@ the engines' agent loop (RFC-0011) live in the language packs and not in the engines. Nothing starts one yet: the stdio MCP binding is a later step. The README's Scope also says the LLM helper is an application on the package, whose `llm.chat` is a stateless chat or a loop turn with a configured transcript store. **Upgrade note:** none. +4. **Tracing: the host leaves the records a Java or Rust callee leaves, so a span lineage is continuous from an engine + into the pack.** A live drive of the engines' agent loop through both packs (2026-11-14) found the pack legs absent + from the trace: the calling engine's HTTP client leg is in `skip.rpc.tracing` by default and the host suppressed its + own record for every RPC-served execution, so `llm.chat` and an MCP binding's route appeared only as time inside the + engine caller's span. Now a traced call arriving at `/api/event` leaves `http.request` (`POST /api/event`, the edge's + round trip, parented onto the caller's span - the envelope's span, else the `traceparent` header's; the backend's + SERVER span), `event.api.service` (the ingress, parented onto the edge) and, for an RPC, the function's round-trip + record with the responder's span and `round_trip`, parented onto the service - the records REST automation, + `EventApiService` and `InboxBase` leave in the Java engine; a drop-n-forget or streaming function's own record + parents onto the service; a refusal or a timeout leaves the edge record with its status. A reply now carries the + responder's span (`WorkerHandler` parity), and the `PostOffice` records a local RPC round trip the way an engine + inbox does. An untraced call records nothing. **Upgrade note:** a backend shows three more spans per traced call into + the pack and one per traced local RPC; a dashboard that read the pack's response time off a function's span moves to + the `http.request` edge record, as for the engines. ## Version 4.12.21, 10/7/2026 diff --git a/docs/guides/config-logging-actuators.md b/docs/guides/config-logging-actuators.md index 97dd646..7a9b093 100644 --- a/docs/guides/config-logging-actuators.md +++ b/docs/guides/config-logging-actuators.md @@ -49,11 +49,27 @@ comes from the `LOG_LEVEL` environment variable when set, else `log.level`. ## Distributed tracing — the OpenTelemetry forwarder (opt-in) -Every traced, non-RPC execution emits the engines' distributed-trace dataset on the -`distributed.tracing` log stream — `{"trace": {...}, "annotations": {...}}`, the record the -Java engine logs — so a log aggregation already stitches the span tree across all four -runtimes. The forwarder is the second sink, for backends that take spans directly: -Dynatrace, Splunk, an OpenTelemetry Collector. +Every traced execution leaves one distributed-trace dataset on the `distributed.tracing` +log stream — `{"trace": {...}, "annotations": {...}}`, the record the Java engine logs — so +a log aggregation already stitches the span tree across all four runtimes. Who emits it +follows the engines' rules, so a span lineage stays continuous from an engine into this host: + +- a function execution that is not an RPC emits its own record when it finishes, parented + onto the caller's span; +- an RPC round trip is recorded by the caller's side, never by the responder (the engines' + `InboxBase`): the `PostOffice` records a local call with the responder's span, which rides + the reply, the caller's span as parent, the responder's `exec_time` and the `round_trip` + it measured; +- a traced call arriving at `/api/event` leaves what a Java or Rust callee leaves for it: + `http.request` (`POST /api/event`, the edge's round trip, parented onto the caller's span — + the envelope's span, else the `traceparent` header's; the backend's SERVER span), + `event.api.service` (the ingress, parented onto the edge) and, for an RPC, the function's + round-trip record parented onto the service; a drop-n-forget or streaming function emits + its own record under the service. A refused call (403, 404) and a timeout (408) leave the + edge record with that status. An untraced call records nothing. + +The forwarder is the second sink, for backends that take spans directly: Dynatrace, Splunk, +an OpenTelemetry Collector. ```yaml otel.forwarding: true diff --git a/extensions/mcp-binding/README.md b/extensions/mcp-binding/README.md index 30560f3..2280786 100644 --- a/extensions/mcp-binding/README.md +++ b/extensions/mcp-binding/README.md @@ -133,6 +133,16 @@ is a service of its own, with its own mailbox and workers. A call learns the rou context (`get_trace().route`, the pack's counterpart of the engines' read-only `my_route` header), which names the server and the remote tool; a call to `mcp.tool` by its own name is a 404. +**Tracing.** A trace names the bound route, never `mcp.tool`: every trace field comes from the service entry the event +was addressed to, so the route's own record, the execution context's route, the reply's sender and the host's `Handled` +log line all read `diagram.render.mcp`. A traced call from an engine leaves the records a Java or Rust callee leaves (the +pack's guide `docs/guides/config-logging-actuators.md`): the edge's round trip `http.request` (`POST /api/event`), +parented onto the engine function that called; `event.api.service` under it; and the route's round-trip record under +that, carrying the execution's span, its `exec_time` and the `round_trip` - the MCP server's `tools/call` is inside that +`exec_time`, not a span of its own. Driven live through the Java Playground's `diagram-assistant` graph on 2026-11-14, +the lineage is continuous from the engine's `diagram.render` into this route; the record is the engines' +`docs/test-reports/agent-loop-trace-lineage.md`. + ## Security - **No engine runs a child process.** The MCP client, the subprocess and the credentials live here, in the pack, behind diff --git a/memory/continuity.md b/memory/continuity.md index 6cebbdb..4d0e69c 100644 --- a/memory/continuity.md +++ b/memory/continuity.md @@ -173,6 +173,26 @@ credential is a reference in the server file, resolved from this process's environment. +- **A traced call into the pack leaves the records a Java or Rust callee leaves, so a span lineage is continuous from an + engine into the pack (2026-11-14; Eric's criterion: for OpenTelemetry to be certified, the span lineage ought to be + continuous end to end).** The live drive of the engines' agent loop through both packs found the pack legs absent from the + trace: the calling engine's `async.http.request` leg is in `skip.rpc.tracing` by default, and the host suppressed the + worker-side record of every RPC-served execution with no callee-side record to replace it, so `llm.chat` and + `diagram.render.mcp` showed only as time inside the engine's `llm.run` and `diagram.render`. Now `server.py` records, for a + traced call at `/api/event`, `http.request` (`POST /api/event`, the edge's round trip; parent the caller's span from the + envelope, else the `traceparent` header; the backend's SERVER span), `event.api.service` (the ingress; parent the edge) + and, for an RPC, the function's round-trip record (parent the service; the responder's span from the reply, `exec_time` + and `round_trip`); the relayed envelope carries the service's span, so a drop-n-forget or streaming function parents onto + it; a refusal or a timeout leaves the edge record with its status; an untraced call records nothing. `bus.py` puts the + execution's span on the reply (`WorkerHandler` parity) and `client.py` records a local RPC round trip as `InboxBase` does; + the worker-side record of an RPC-served execution stays suppressed. Re-driven the same day on the `diagram-assistant` + graph: 20 records per run, every parent owned by a record, in both packs (the engines' + `docs/test-reports/agent-loop-trace-lineage.md`; the Node twin in lockstep). **How to apply:** the rule is who emits a + record, never whether one exists - every traced execution leaves exactly one; before changing a host or client path, ask + which side records it in the Java engine and mirror that side; a lineage check parses every record of the trace from every + process and verifies each parent is a span some record owns. + + ## Conventions - **Quality gates (adopted 2026-08-23, Eric's IDE review round): ruff + basedpyright + diff --git a/memory/sessions/2026-11-14-115310.md b/memory/sessions/2026-11-14-115310.md new file mode 100644 index 0000000..a922ce8 --- /dev/null +++ b/memory/sessions/2026-11-14-115310.md @@ -0,0 +1,58 @@ +# Session (2026-11-14T11:53:10.000Z) + +**Agent:** Claude Code +**Title:** Trace records mirror the engines' callee and caller - a span lineage continuous from an engine into the pack + +## Context + +Follows 2026-11-14-104712 (the `@preload` shape of the MCP binding, #52). Eric asked how the alias shape appears in a +trace, then for a live end-to-end drive from the Java engine to both language packs with the telemetry recorded, and set +the criterion: for OpenTelemetry to be certified, the span lineage ought to be continuous end to end. + +## What the drive found (the baseline, negative control first) + +The Playground's `diagram-assistant` graph on Claude Haiku 4.5, the engine calling this pack's LLM helper (`llm.chat`, +port 8086) and MCP binding (`diagram.render.mcp`, port 8088, the public Mermaid Chart server). Every record the engine +logged chained to a parent that exists, but the pack legs left no record anywhere: 11 engine records, none from the +pack. Two rules met: the calling engine's HTTP client leg `async.http.request` is in `skip.rpc.tracing` by default, so +the caller records nothing for a wire RPC; and this host suppressed the worker-side record of every RPC-served execution +("engine parity") without the callee-side records a Java engine leaves for the same call - the `http.request` edge +(`POST /api/event`), `event.api.service` and the function's round-trip record from the local inbox. The forwarder drops a +record without a valid span id, so even an engine inbox record for a pack route would not have reached a backend: the +reply carried no span. + +## The design Eric chose + +Three shapes were offered: mirror the Java callee exactly, one record per wire RPC, or keep the rule and document it. +Eric chose the first. + +## What changed (branch `feat/trace-records-engine-parity`) + +- `src/mercury_composable/bus.py`: the reply carries the responder's span (`WorkerHandler` parity); `round_trip_record` + builds the engines' `InboxBase` record; `EventBus.emit` logs a dataset and forwards it. +- `src/mercury_composable/client.py`: a traced local RPC (plain or interceptor) is recorded by the caller's side with the + responder's span, the caller's span as parent, the responder's `exec_time` and the measured `round_trip`. +- `src/mercury_composable/server.py`: `_Edge` records a traced call at `/api/event` as the Java callee does - + `http.request` (the edge's round trip, parent the caller's span: the envelope's, else the `traceparent` header's), + `event.api.service` (parent the edge), the function's round-trip record for an RPC (parent the service); the relayed + envelope carries the service's span, so a drop-n-forget or streaming function parents onto it; a refusal (403, 404), a + timeout (408) and an in-band stream failure leave the edge record with that status; an untraced call records nothing. +- `tests/test_event_stream.py`: the two tests that pinned "RPC legs emit no dataset" now pin the engines' rule, and a + new test pins the wire RPC's three records, the responder's span on the reply, the refusal's edge record and the silence + of an untraced call. +- `docs/guides/config-logging-actuators.md` states who emits which record; CHANGELOG Unreleased Changed 4; the MCP + binding's README gains a Tracing paragraph (the alias never appears; the route is the identity). + +## Verification + +- `ruff` clean; `basedpyright` 0 errors; `pytest` 398 passed, 64 skipped. +- Re-driven live the same way: 20 records per run (11 engine, 9 pack: three per wire call, two `llm.chat` turns and one + `diagram.render.mcp`), every parent owned by a record. The helper's edge sits under the engine's `llm.run`, the + binding's under `diagram.render`, each function under its `event.api.service`, and the timings nest (edge >= the + function's round trip >= its `exec_time`). The engines' repository carries the record: + `docs/test-reports/agent-loop-trace-lineage.md`. + +## Memory References + +- Relied on: mcp-binding-route-per-tool (the alias shape, whose trace identity the README now states; unchanged) +- Created: trace-records-engine-parity (tier: working) diff --git a/src/mercury_composable/bus.py b/src/mercury_composable/bus.py index b9d63a4..9cc104a 100644 --- a/src/mercury_composable/bus.py +++ b/src/mercury_composable/bus.py @@ -161,6 +161,32 @@ def _emit_trace(delivery: _Delivery, info: TraceInfo, start: str, return dataset +def round_trip_record(*, service: str, sender: str | None, trace_id: str, trace_path: str | None, + reply: EventEnvelope, parent_span_id: str | None, start: str, + round_trip_ms: float) -> dict[str, Any]: + """The caller-side record of an RPC round trip - the engines' InboxBase record. Its + span is the responder's own (the reply carries it), its parent the caller's span, its + exec_time the responder's and its round_trip the wait measured by the caller; the + worker-side record of an RPC-served execution is suppressed, so this is the one record + for that span (the wire-format contract's rpc tag).""" + from .actuator import app_origin # late: actuator imports the registry chain + trace: dict[str, Any] = { + "origin": app_origin(), "id": trace_id, "path": trace_path, "service": service, + "start": start, "exec_time": reply.exec_time, "round_trip": round(round_trip_ms, 3), + "status": reply.get_status(), "success": not reply.has_error(), + } + if sender: + trace["from"] = sender + if reply.span_id: + trace["span_id"] = reply.span_id + if parent_span_id: + trace["parent_span_id"] = parent_span_id + if reply.has_error(): + # for data privacy, only a recognized error message is shown (the engines' InboxBase) + trace["exception"] = reply.body if isinstance(reply.body, str) else "***" + return trace + + class EventBus: """Per-registry bus: one FIFO mailbox and N workers per registered route.""" @@ -320,6 +346,10 @@ async def _execute(self, delivery: _Delivery) -> EventEnvelope: _reset_trace(token) reply.sender = reply.sender or service.route reply.exec_time = round((time.perf_counter() - start) * 1000, 3) + if info.span_id: + # the responder's span rides the reply, so the caller's round-trip record chains + # onto it - the engines' WorkerHandler contract + reply.span_id = info.span_id if info.annotations: reply.annotations.update(info.annotations) self._forward(_emit_trace(delivery, info, start_iso, reply.exec_time, @@ -371,6 +401,17 @@ async def _execute_interceptor(self, delivery: _Delivery) -> EventEnvelope: # an interceptor's own outcome is never auto-replied return EventEnvelope() + def emit(self, trace: dict[str, Any], annotations: dict[str, Any] | None = None) -> None: + """Emit one distributed-trace dataset the engines' way: the record on the telemetry + log stream, then the extension route. The caller-side records - an RPC round trip + recorded by the PostOffice, the edge records of a call through /api/event - come + through here; a worker-side record is emitted by the execution itself.""" + dataset: dict[str, Any] = {"trace": trace} + if annotations: + dataset["annotations"] = dict(annotations) + telemetry_log.info(dataset) + self._forward(dataset) + def _forward(self, dataset: dict[str, Any] | None) -> None: """Hand an emitted dataset to the engines' extension route when a function is registered there (the OpenTelemetry forwarder, or an diff --git a/src/mercury_composable/client.py b/src/mercury_composable/client.py index cc15ce5..3c02fef 100644 --- a/src/mercury_composable/client.py +++ b/src/mercury_composable/client.py @@ -37,13 +37,14 @@ import concurrent.futures import json import re +import time from collections.abc import AsyncIterator, Callable, Coroutine from typing import Any import aiohttp -from .bus import DeliveryTimeout, get_host_loop -from .envelope import EventEnvelope +from .bus import DeliveryTimeout, get_host_loop, round_trip_record +from .envelope import EventEnvelope, iso_utc from .event_stream import ( DATA, ENVELOPE, @@ -166,27 +167,44 @@ async def _call_local(self, route: str, body: Any, headers: dict[str, str] | Non # unmarked = the reply; marked = a streaming target refusing a # single-shot caller (the pinned 406) sink_route, queue = bus.open_sink() + start, started = iso_utc(), time.perf_counter() try: bus.publish_envelope(service, event.set_reply_to(sink_route)) try: first = await asyncio.wait_for(queue.get(), max(100, timeout_ms) / 1000) except asyncio.TimeoutError: - return EventEnvelope().set_status(408).set_body( - f"Timeout for {timeout_ms} ms") + first = EventEnvelope().set_status(408).set_body(f"Timeout for {timeout_ms} ms") if stream_signal(first) is not None: - return EventEnvelope().set_status(406).set_body(STREAM_CALLER_REQUIRED) + first = EventEnvelope().set_status(406).set_body(STREAM_CALLER_REQUIRED) + self._record_round_trip(route, event, first, start, started) return first finally: bus.close_sink(sink_route) if is_async: return bus.publish(service, event.headers, event.body, trace_id=event.trace_id, trace_path=event.trace_path, cid=event.cid, envelope=event) + start, started = iso_utc(), time.perf_counter() try: - return await bus.deliver(service, event.headers, event.body, timeout_ms, - trace_id=event.trace_id, trace_path=event.trace_path, - cid=event.cid, envelope=event) + reply = await bus.deliver(service, event.headers, event.body, timeout_ms, + trace_id=event.trace_id, trace_path=event.trace_path, + cid=event.cid, envelope=event) except DeliveryTimeout: - return EventEnvelope().set_status(408).set_body(f"Timeout for {timeout_ms} ms") + reply = EventEnvelope().set_status(408).set_body(f"Timeout for {timeout_ms} ms") + self._record_round_trip(route, event, reply, start, started) + return reply + + def _record_round_trip(self, route: str, event: EventEnvelope, reply: EventEnvelope, + start: str, started: float) -> None: + """A traced caller's record of a local RPC round trip - the engines' InboxBase: + the responder's span from the reply, the caller's span (on the outbound envelope) + as the parent, and the round trip measured here. Untraced calls record nothing.""" + if not event.trace_id: + return + self._registry.bus.emit( + round_trip_record(service=route, sender=event.sender, trace_id=event.trace_id, + trace_path=event.trace_path, reply=reply, parent_span_id=event.span_id, + start=start, round_trip_ms=(time.perf_counter() - started) * 1000), + reply.annotations) async def _call(self, route: str, body: Any, headers: dict[str, str] | None, timeout_ms: int, endpoint: str | None, is_async: bool, diff --git a/src/mercury_composable/server.py b/src/mercury_composable/server.py index 6d1d1ac..6d751b5 100644 --- a/src/mercury_composable/server.py +++ b/src/mercury_composable/server.py @@ -19,21 +19,33 @@ - The host also serves the engines' actuator endpoints (``/info``, ``/info/routes``, ``/env``, ``/health``, ``/livenessprobe``) for operations and Kubernetes probes - see :mod:`mercury_composable.actuator`. +- A traced call leaves the records a Java or Rust callee leaves for the same + call, so a span lineage stays continuous into this host: ``http.request`` + (``POST /api/event``, the edge's round trip, parented onto the caller's span + - the envelope's span, else the ``traceparent`` header's), ``event.api.service`` + (the ingress, parented onto the edge) and, for an RPC, the function's + round-trip record with the responder's span (the caller side's record in + the engines); a drop-n-forget or streaming function emits its own record, + parented onto the service. An untraced call records nothing. """ from __future__ import annotations import asyncio import contextlib +import re +import secrets +import time from collections.abc import Awaitable, Callable +from typing import Any from aiohttp import web from . import otel -from .actuator import Actuator -from .bus import DeliveryTimeout +from .actuator import Actuator, app_origin +from .bus import DeliveryTimeout, EventBus, round_trip_record from .config import app_config -from .envelope import EventEnvelope +from .envelope import EventEnvelope, iso_utc from .event_stream import ( DATA, STREAM_CALLER_REQUIRED, @@ -54,6 +66,10 @@ X_EVENT_API = "x-event-api" # the engines' reserved route name for the Event-over-HTTP ingress EVENT_API_SERVICE = "event.api.service" +# the edge's own service name and path in its round-trip record (the engines' REST edge) +HTTP_REQUEST = "http.request" +EDGE_PATH = "POST /api/event" +TRACEPARENT = re.compile(r"^00-([0-9a-f]{32})-([0-9a-f]{16})-[0-9a-f]{2}$") log = get_logger("mercury.server") @@ -63,6 +79,71 @@ def _transport_error(status: int, message: str) -> web.Response: return web.Response(status=status, body=reply.to_bytes(), content_type=OCTET_STREAM) +def _caller_span(request: web.Request, event: EventEnvelope) -> str | None: + """The remote caller's span: the one on the envelope (the engines' PostOffice stamps + the calling function's span), else the W3C traceparent header's.""" + if event.span_id: + return event.span_id + match = TRACEPARENT.match((request.headers.get("traceparent") or "").strip().lower()) + return match.group(2) if match else None + + +class _Edge: + """The trace records of one traced call through the /api/event edge - what a Java or + Rust callee leaves for the same call: ``http.request`` (the edge's round trip, the + backend's SERVER span, parented onto the caller's span) and ``event.api.service`` (the + ingress, parented onto the edge); the relayed event carries the service's span, so the + function parents onto it. An RPC adds the function's round-trip record - the caller + side's record in the engines (InboxBase) - with the responder's span from the reply.""" + + def __init__(self, bus: EventBus, event: EventEnvelope, caller_span: str | None) -> None: + self.bus = bus + self.trace_id = event.trace_id or "" + self.trace_path = event.trace_path + self.route = event.to or "" + self.caller_span = caller_span + self.edge_span = secrets.token_hex(8) + self.service_span = secrets.token_hex(8) + self.start = iso_utc() + self.started = time.perf_counter() + + def _elapsed(self) -> float: + return round((time.perf_counter() - self.started) * 1000, 3) + + def record_service(self) -> None: + """The ingress record, when the event is handed on - the engines' EventApiService + execution, parented onto the edge.""" + self.bus.emit({ + "origin": app_origin(), "id": self.trace_id, "path": EDGE_PATH, "service": EVENT_API_SERVICE, + "start": self.start, "from": HTTP_REQUEST, "exec_time": self._elapsed(), "status": 200, + "success": True, "span_id": self.service_span, "parent_span_id": self.edge_span, + }) + + def record_function(self, sender: str | None, reply: EventEnvelope, dispatched: str, + started: float) -> None: + """The RPC's round-trip record, parented onto the service - the engines' InboxBase + record for the relayed call.""" + self.bus.emit(round_trip_record(service=self.route, sender=sender, trace_id=self.trace_id, + trace_path=self.trace_path, reply=reply, + parent_span_id=self.service_span, start=dispatched, + round_trip_ms=(time.perf_counter() - started) * 1000), + reply.annotations) + + def record_edge(self, status: int, error: str | None = None) -> None: + """The edge's round-trip record when the response is complete - the engines' REST + edge: the whole call under the caller's span, the HTTP status sent.""" + trace: dict[str, Any] = { + "origin": app_origin(), "id": self.trace_id, "path": EDGE_PATH, "service": HTTP_REQUEST, + "start": self.start, "exec_time": self._elapsed(), "status": status, "success": status < 400, + "span_id": self.edge_span, + } + if self.caller_span: + trace["parent_span_id"] = self.caller_span + if status >= 400 and error: + trace["exception"] = error + self.bus.emit(trace) + + def _handler_headers(event: EventEnvelope) -> dict[str, str]: headers = {k: v for k, v in event.headers.items() if k.lower() != X_EVENT_API and not k.lower().startswith("my_")} @@ -94,20 +175,28 @@ async def handle_event(self, request: web.Request) -> web.StreamResponse: return _transport_error(400, str(e)) if not event.to: return _transport_error(400, "Missing routing path") + bus = self.registry.bus + edge = _Edge(bus, event, _caller_span(request, event)) if event.trace_id else None service = self.registry.get(event.to) if service is None: - return _transport_error(404, f"Route {event.to} not found") + return self._refuse(edge, 404, f"Route {event.to} not found") if service.private: - return _transport_error(403, f"{event.to} is private") + return self._refuse(edge, 403, f"{event.to} is private") if not event.sender: # the engines' EventApiService parity: its PostOffice fills the # sender with its own route when the wire envelope carries none event.set_from(EVENT_API_SERVICE) + if edge: + # the relayed event carries the service's span, so the function + # parents onto it - the engines' relay + event.set_span_id(edge.service_span) + edge.record_service() headers = _handler_headers(event) - bus = self.registry.bus if is_async: ack = bus.publish(service, headers, event.body, trace_id=event.trace_id, trace_path=event.trace_path, cid=event.cid, envelope=event) + if edge: + edge.record_edge(202) return web.Response(status=202, body=ack.to_bytes(), content_type=OCTET_STREAM) if service.interceptor: # interceptor dispatch (the reply_to mechanism): the handler @@ -115,21 +204,38 @@ async def handle_event(self, request: web.Request) -> web.StreamResponse: # reply address and answers manually - single-shot or streaming capable = TEXT_EVENT_STREAM in (request.headers.get("accept") or "") return await self._dispatch_interceptor(request, service, event, headers, - ttl, capable) + ttl, capable, edge) + dispatched, started = iso_utc(), time.perf_counter() try: reply = await bus.deliver(service, headers, event.body, ttl, trace_id=event.trace_id, trace_path=event.trace_path, cid=event.cid, envelope=event) except DeliveryTimeout: log.warning("Event %s timeout for %d ms (trace_id=%s)", event.to, ttl, event.trace_id) + if edge: + timeout = EventEnvelope().set_status(408).set_body(f"Timeout for {ttl} ms") + edge.record_function(event.sender, timeout, dispatched, started) + edge.record_edge(408, f"Timeout for {ttl} ms") return _transport_error(408, f"Timeout for {ttl} ms") log.info("Handled %s status=%d exec_time=%sms trace_id=%s", event.to, reply.get_status(), reply.exec_time, event.trace_id) + if edge: + edge.record_function(event.sender, reply, dispatched, started) + edge.record_edge(200) return web.Response(status=200, body=reply.to_bytes(), content_type=OCTET_STREAM) + @staticmethod + def _refuse(edge: _Edge | None, status: int, message: str) -> web.Response: + """A transport refusal, recorded at the edge when the call was traced.""" + if edge: + edge.record_service() + edge.record_edge(status, message) + return _transport_error(status, message) + async def _dispatch_interceptor(self, request: web.Request, service: ServiceDef, event: EventEnvelope, headers: dict[str, str], - ttl: int, capable: bool) -> web.StreamResponse: + ttl: int, capable: bool, + edge: _Edge | None = None) -> web.StreamResponse: """Dispatch to an interceptor and classify its first reply exactly like the engines: unmarked = the classic single-shot response, byte identical; marked = the envelope-mode SSE dialect for a caller that @@ -149,7 +255,8 @@ async def _dispatch_interceptor(self, request: web.Request, service: ServiceDef, if event.trace_id: handler_event.set_trace(event.trace_id, event.trace_path or service.route) if event.span_id: - # the caller's span - the handler's span parents onto it + # the service's span (the caller's when the call is untraced at the + # edge) - the handler's span parents onto it handler_event.set_span_id(event.span_id) if event.sender: handler_event.set_from(event.sender) @@ -159,6 +266,8 @@ async def _dispatch_interceptor(self, request: web.Request, service: ServiceDef, except asyncio.TimeoutError: log.warning("Event %s timeout for %d ms (trace_id=%s)", event.to, ttl, event.trace_id) + if edge: + edge.record_edge(408, f"Timeout for {ttl} ms") return _transport_error(408, f"Timeout for {ttl} ms") marker = stream_signal(first) if marker is None: @@ -166,19 +275,28 @@ async def _dispatch_interceptor(self, request: web.Request, service: ServiceDef, # error contract for an uncaught interceptor exception) log.info("Handled %s status=%d exec_time=%sms trace_id=%s", event.to, first.get_status(), first.exec_time, event.trace_id) + if edge: + edge.record_edge(200) return web.Response(status=200, body=first.to_bytes(), content_type=OCTET_STREAM) if not capable: # a streaming reply cannot ride a single-shot response + if edge: + edge.record_edge(406, STREAM_CALLER_REQUIRED) return _transport_error(406, STREAM_CALLER_REQUIRED) - return await self._stream_response(request, queue, first, marker, ttl) + response, status = await self._stream_response(request, queue, first, marker, ttl) + if edge: + # the engines' edge: a stream failing in-band after its head reports + # the failure's own status + edge.record_edge(status) + return response finally: bus.close_sink(sink_route) async def _stream_response(self, request: web.Request, queue: asyncio.Queue[EventEnvelope], first: EventEnvelope, first_marker: str, - ttl: int) -> web.StreamResponse: + ttl: int) -> tuple[web.StreamResponse, int]: """Render the envelope-mode SSE dialect: envelope frames for the head, the terminals and non-text segments; raw frames for plain text. The x-ttl allowance (overridable by the producer's head control, in @@ -196,19 +314,22 @@ async def _stream_response(self, request: web.Request, response.headers["content-type"] = TEXT_EVENT_STREAM response.headers.setdefault("cache-control", "no-cache") await response.prepare(request) + status = first.get_status() try: await response.write(envelope_frame(first)) if first_marker == DATA: - await self._stream_segments(response, queue, idle_ms) + status = await self._stream_segments(response, queue, idle_ms) except ConnectionError: # a disconnected client ends the stream; late segments are no-op drops log.debug("Client disconnected from event stream") with contextlib.suppress(ConnectionError): await response.write_eof() - return response + return response, status async def _stream_segments(self, response: web.StreamResponse, - queue: asyncio.Queue[EventEnvelope], idle_ms: int) -> None: + queue: asyncio.Queue[EventEnvelope], idle_ms: int) -> int: + """Render the segments until the terminal; the status returned is the stream's + outcome - 200, or an in-band failure's own status - for the edge's record.""" ping_ms = keep_alive_ms() while True: event = await self._next_segment(response, queue, idle_ms, ping_ms) @@ -217,7 +338,7 @@ async def _stream_segments(self, response: web.StreamResponse, seconds = idle_ms // 1000 frame = envelope_frame(exception_envelope(408, f"Timeout for {seconds} seconds")) await response.write(frame) - return + return 408 marker = stream_signal(event) if marker == DATA: frame = data_frame(event, first_frame=False) @@ -227,14 +348,14 @@ async def _stream_segments(self, response: web.StreamResponse, # eof or exception: the terminal envelope frame ends the # response cleanly - no cosmetic frames on this wire await response.write(envelope_frame(event)) - return + return event.get_status() if event.has_error() else 200 elif event.has_error(): # the bus's error contract for an uncaught interceptor # exception mid-stream - fail in-band with the exact status message = str(event.body) if event.body is not None else "Stream failed" frame = envelope_frame(exception_envelope(event.get_status(), message)) await response.write(frame) - return + return event.get_status() else: log.warning("Dropping event - invalid %s signal", "x-event-stream") diff --git a/tests/test_event_stream.py b/tests/test_event_stream.py index 539722d..13b5fa8 100644 --- a/tests/test_event_stream.py +++ b/tests/test_event_stream.py @@ -238,26 +238,60 @@ async def test_business_cid_rides_the_streaming_hop(stream_host: tuple[str, Func assert events[-1].body["my_correlation_id"] == "biz-42" -async def test_span_lineage_continues_across_the_hop(stream_host: tuple[str, FunctionRegistry]): - # the engines' span model: the caller's span rides the outbound envelope; - # the receiving execution mints its own span with the caller's as parent +def _traces(caplog: pytest.LogCaptureFixture) -> list[dict[str, Any]]: + """The trace sections of the distributed.tracing records a test captured.""" + out: list[dict[str, Any]] = [] + for r in caplog.records: + if r.name == "distributed.tracing" and isinstance(r.msg, dict): + section = r.msg.get("trace") + if isinstance(section, dict): + out.append(section) + return out + + +def _one(traces: list[dict[str, Any]], service: str) -> dict[str, Any]: + found = [t for t in traces if t["service"] == service] + assert len(found) == 1, f"{service}: {[t['service'] for t in traces]}" + return found[0] + + +async def test_span_lineage_continues_across_the_hop(stream_host: tuple[str, FunctionRegistry], + caplog: pytest.LogCaptureFixture): + # the engines' callee shape: the caller's span rides the outbound envelope; the host + # records the /api/event edge's round trip under it (http.request, the backend's SERVER + # span), the ingress under the edge (event.api.service), and the producer mints its own + # span under the ingress - one connected tree, as a Java or Rust callee leaves it url, _ = stream_host caller_span = "ab" * 8 - async with PostOffice() as po: - with trace_context("4bf92f3577b34da6a3ce929d0e0e4746", "TEST /lineage", - span_id=caller_span): - events = await collect(po, "unit.tokens", url, mode="biz") + with caplog.at_level("INFO", logger="distributed.tracing"): + async with PostOffice() as po: + with trace_context("4bf92f3577b34da6a3ce929d0e0e4746", "TEST /lineage", + span_id=caller_span): + events = await collect(po, "unit.tokens", url, mode="biz") + await asyncio.sleep(0.05) # the edge's record follows the response body = events[-1].body - assert body["parent_span_id"] == caller_span - assert body["span_id"] != caller_span - assert re.fullmatch(r"[0-9a-f]{16}", body["span_id"]), "16-hex W3C-shaped span" + traces = _traces(caplog) + edge = _one(traces, "http.request") + ingress = _one(traces, "event.api.service") + producer = _one(traces, "unit.tokens") + assert edge["parent_span_id"] == caller_span + assert edge["path"] == "POST /api/event" + assert ingress["parent_span_id"] == edge["span_id"] + assert ingress["from"] == "http.request" + assert producer["parent_span_id"] == ingress["span_id"] + assert body["span_id"] == producer["span_id"] + assert body["parent_span_id"] == ingress["span_id"] + for record in (edge, ingress, producer): + assert re.fullmatch(r"[0-9a-f]{16}", record["span_id"]), "16-hex W3C-shaped span" + assert edge["exec_time"] >= producer["exec_time"], "the round trip covers the execution" async def test_trace_dataset_emitted_with_engine_shape( stream_host: tuple[str, FunctionRegistry], caplog: pytest.LogCaptureFixture): - # non-RPC executions emit the engines' distributed-trace dataset record; - # RPC round-trips are suppressed (their metrics fold into the caller) + # every traced execution leaves one record, the engines' way: a non-RPC execution emits + # its own; an RPC round trip is recorded by the caller's side (the PostOffice here, with + # the responder's span from the reply and the round trip), never by the responder url, registry = stream_host caller_span = "cd" * 8 with caplog.at_level("INFO", logger="distributed.tracing"): @@ -268,29 +302,70 @@ async def test_trace_dataset_emitted_with_engine_shape( rpc = await PostOffice(registry=registry).request( "unit.biz", None, timeout_ms=5000) assert rpc.get_status() == 200 - def trace_of(message: object) -> dict[str, Any] | None: - # a dataset record's message is {"trace": {...}[, "annotations": ...]} - section = message.get("trace") if isinstance(message, dict) else None - return section if isinstance(section, dict) else None - - sections = (trace_of(r.msg) for r in caplog.records - if r.name == "distributed.tracing") - traces = [t for t in sections if t is not None] - services = [t["service"] for t in traces] - assert "unit.biz" not in services, "RPC legs emit no dataset (engine parity)" - tokens = [t for t in traces if t["service"] == "unit.tokens"] - assert len(tokens) == 1 - trace = tokens[0] - assert trace["id"] == "4bf92f3577b34da6a3ce929d0e0e4747" - assert trace["path"] == "TEST /telemetry" - assert trace["parent_span_id"] == caller_span - assert trace["success"] is True - assert trace["status"] == 200 + await asyncio.sleep(0.05) # the edge's record follows the response + traces = _traces(caplog) + biz = _one(traces, "unit.biz") + assert biz["parent_span_id"] == caller_span + assert biz["span_id"] == rpc.span_id, "the responder's span rides the reply and names the record" + assert biz["round_trip"] >= biz["exec_time"] + assert biz["path"] == "TEST /telemetry" + assert biz["success"] is True + assert biz["status"] == 200 + tokens = _one(traces, "unit.tokens") + assert tokens["id"] == "4bf92f3577b34da6a3ce929d0e0e4747" + assert tokens["path"] == "TEST /telemetry" + assert tokens["parent_span_id"] == _one(traces, "event.api.service")["span_id"] + assert tokens["success"] is True + assert tokens["status"] == 200 # an anonymous /api/event caller: the host fills the sender with its own # identity, the engines' EventApiService parity - assert trace["from"] == "event.api.service" + assert tokens["from"] == "event.api.service" for key in ("origin", "start", "exec_time", "span_id"): - assert key in trace, f"engine dataset key {key}" + assert key in tokens, f"engine dataset key {key}" + + +async def test_a_wire_rpc_leaves_the_callee_records(stream_host: tuple[str, FunctionRegistry], + caplog: pytest.LogCaptureFixture): + # a call through /api/event leaves what a Java or Rust callee leaves: the edge's round + # trip (http.request, parented onto the caller's span), the ingress (event.api.service) + # and the function's round-trip record with the responder's span - the one the reply + # carries, so an engine's inbox chains onto it too + url, _ = stream_host + caller_span = "ef" * 8 + with caplog.at_level("INFO", logger="distributed.tracing"): + async with PostOffice() as po: + with trace_context("4bf92f3577b34da6a3ce929d0e0e4748", "TEST /rpc", span_id=caller_span): + reply = await po.request("unit.echo", {"n": 1}, endpoint=url, timeout_ms=5000) + refused = await po.request("unit.nowhere", {}, endpoint=url, timeout_ms=5000) + await asyncio.sleep(0.05) # the edge's record follows the response + assert reply.get_status() == 200 + assert reply.body == {"echo": {"n": 1}} + assert refused.get_status() == 404 + traces = _traces(caplog) + edges = [t for t in traces if t["service"] == "http.request"] + assert sorted(t["status"] for t in edges) == [200, 404], "every traced call leaves its edge record" + edge = next(t for t in edges if t["status"] == 200) + ingress = next(t for t in traces if t["service"] == "event.api.service" + and t["parent_span_id"] == edge["span_id"]) + function = _one(traces, "unit.echo") + assert edge["parent_span_id"] == caller_span + assert ingress["parent_span_id"] == edge["span_id"] + assert function["parent_span_id"] == ingress["span_id"] + assert function["span_id"] == reply.span_id, "the responder's span rides the reply" + assert function["round_trip"] >= function["exec_time"] + assert function["from"] == "event.api.service", "an anonymous wire caller" + assert function["path"] == "TEST /rpc" + assert edge["exec_time"] >= function["round_trip"] + refusal = next(t for t in edges if t["status"] == 404) + assert refusal["success"] is False + assert "unit.nowhere" in refusal["exception"] + # an untraced call records nothing + caplog.clear() + with caplog.at_level("INFO", logger="distributed.tracing"): + async with PostOffice() as po: + await po.request("unit.echo", {"n": 2}, endpoint=url, timeout_ms=5000) + await asyncio.sleep(0.05) + assert _traces(caplog) == [] async def test_business_cid_injected_on_local_delivery(stream_host: tuple[str, FunctionRegistry]):