feat(filter_runtime): MQ hop OTel spans + per-frame W3C trace propagation (PLAT-866)

[stwilt]

📋 What does this PR do?

Adds OpenTelemetry hop spans around the MQ transport and propagates W3C trace context per-frame through every ZMQ envelope. With tracing enabled (TELEMETRY_EXPORTER_TYPE=otlp_grpc or otlp_http), every filter-to-filter hop now emits:

• mq.send / mq.recv outer hop spans, attributes: topic, payload_bytes, frame.id, frame.format, mq.id.
• frame.serialize / frame.deserialize CPU-work sub-spans.
• frame.encode_jpg / frame.decode_jpg codec sub-spans (jpg transport only).
• zmq.send_multipart / zmq.recv_multipart kernel-syscall sub-spans. Duration minus (zero) sibling spans inside is pure kernel-copy cost.

opentelemetry.propagate.inject(env) runs on the send side so each envelope carries traceparent / tracestate; propagate.extract runs on the receive side and the consumer-side {FilterClass}.process span (added in PLAT-848) now parents on that per-frame context. The TRACEPARENT env-var fallback (PLAT-848 behavior) is retained for source filters that never call mq.recv (e.g. VideoIn).

When tracing is disabled (the default), MQ.tracer is None, the codec / kernel sub-span helper short-circuits on a single Span.is_recording() check, and no Span objects are allocated on the hot path.

🔍 Why is this needed?

PLAT-848 added {FilterClass}.process spans — one span per filter per frame. Everything between filters was invisible: hop transport, frame serialize, ZMQ kernel copy, codec cost. An operator investigating a slow frame in Cloud Trace saw a flat sequence of process spans with no explanation of the gaps between them.

PLAT-848 also read TRACEPARENT from an env var once at filter startup, so every frame a filter processed inherited the same parent span. Per-frame distributed traces across filters did not actually work — filter A's frame-17 spans were not linked to filter B's frame-17 spans.

Landing the hop spans and the per-frame propagation together is a single piece of work because the sub-spans nest structurally inside the hop spans and context propagation has to ride the same ZMQ envelope that the hop spans describe — splitting would force an interdependency graph between PRs for no gain.

🧪 How was it tested?

• New unit tests (tests/test_tracing.py, 9 new): assert raw vs jpg hop span structure (codec sub-spans fire only on jpg), the no-Span-allocation contract when there is no recording outer span, W3C envelope round-trip (inject then extract yields matching trace_id), and ZMQReceiver.recv_once populating last_extracted_ctx when traceparent is present.
• New integration test (tests/integration/test_tracing_export.py, @pytest.mark.slow): two MQ endpoints across an IPC hop, both exporting to a jaeger-all-in-one container, assert that mq.send / mq.recv / filterA.process / filterB.process all land in the same trace ID via per-frame envelope propagation. Requires docker; exercised by make test-integration.
• Local suite: 135 / 135 pass for the touched unit-test surface (tests/test_tracing.py tests/test_mq.py tests/test_zeromq.py tests/test_frame.py tests/test_timing_metrics.py tests/test_telemetry.py tests/test_lazy_imports.py — 56s). tests/test_filter.py passes 30 / 30 alone (38s). Combined-file runs hit a pre-existing fork-after-thread flake that reproduces identically on the baseline without these changes (verified via git stash + re-run).
• PR description follow-up: measured cost of enabling tracing on make waterfall's 4k_raw scenario will be reported in the PR description for informational purposes — not a gating target per the PLAT-866 acceptance criteria.

🔗 Related Issues

• PLAT-866 — this story; runtime-side instrumentation for MQ hop transport and per-frame context propagation.
• PLAT-848 — sub-task that added the {FilterClass}.process spans this work builds on. The propagation fix here closes the per-frame distributed trace gap PLAT-848 left open.
• PLAT-827 — parent-story sibling scoped to the api→agent→controller→openfilter context-propagation chain. This story extends that work with per-frame propagation and per-hop visibility inside openfilter.
• FILTER-419 — SHM transport investigation; the zmq.send_multipart / zmq.recv_multipart sub-span duration at 4K raw is the diagnostic signal that motivated that ticket.

🖼️ Screenshots or Logs (if applicable)

N/A — instrumentation work; no UI surface. Trace waterfalls land in the operator's existing Cloud Trace / jaeger tooling; sample trace hierarchy is documented in RELEASE.md and asserted in the unit and integration tests.

✅ Checklist

• I have read and agreed to the terms of the LICENSE
• I have read the CONTRIBUTING guide
• I have followed the coding style
• I have signed all commits in compliance with the DCO (git commit -s)
• I have added or updated tests as needed (9 new unit tests in tests/test_tracing.py; new jaeger-backed integration test in tests/integration/test_tracing_export.py)
• I have added or updated documentation as needed (RELEASE.md documents the new spans, the per-frame propagation contract, and the no-allocation-when-disabled guarantee)

[shingonoide]

Summary

Adds OTel mq.send / mq.recv hop spans around ZMQSender.send / ZMQReceiver.recv and W3C traceparent / tracestate injection+extraction on every envelope, so {FilterClass}.process spans link across filters into a single distributed trace. Codec sub-spans (frame.encode_jpg, frame.decode_jpg, zmq.send_multipart, zmq.recv_multipart, frame.serialize, frame.deserialize) are gated by maybe_start_span, keeping the hot path allocation-free when tracing is off.

Findings

Non-blocking

1. recv_span leaks if topicmsgs2frames / metrics_.incoming raises — openfilter/filter_runtime/mq.py:329-374

recv_span = tracer.start_span("mq.recv", ...) at line 329 is created without a context manager, and the only try/finally between creation and recv_span.end() at line 374 wraps otel_context.detach(token) (lines 357-361). If MQ.topicmsgs2frames or metrics_.incoming raises inside the frame.deserialize block, recv_span.end() is unreachable and the span is dropped without an end time. Either wrap the body in a try/finally or use start_as_current_span for the outer hop span as well.

2. Fan-in: "last extraction wins" parents the consumer's {Filter}.process span under one arbitrary upstream — openfilter/filter_runtime/zeromq.py:836-843

The comment already calls this out as intentional ("closest we can do without joining traces"). Worth surfacing in the PR description / module docstring so future maintainers wondering about incomplete fan-in traces find the explanation quickly.

3. _HOP_TRACER is mutated without a lock — openfilter/observability/tracing.py:32-43

register_hop_tracer is a bare global assignment. Atomic in CPython, but if two MQ instances with different tracers are constructed concurrently the last writer silently wins. Not a hazard in the single-filter-per-process model in use today; raising as a future-proofing nit.

Nits

4. send_span is not None guard couples to contextlib.nullcontext()'s yielded None — openfilter/filter_runtime/mq.py:265-283

Works today (zero-arg nullcontext() yields None); changing to contextlib.nullcontext(some_obj) in a refactor would silently break the guard. if tracer: would express the intent more directly.

5. zmq_t_start = ... or t_recv_start produces a zero-duration zmq.recv_multipart span when last_recv_t_start_ns is 0 — openfilter/filter_runtime/mq.py:341-342

Edge case (path is reached only when timing fields are unset). Cosmetic.

What looks good

• W3C propagation goes through standard opentelemetry.propagate.{inject,extract} rather than hand-rolled parsing.
• maybe_start_span exits before any tracer lookup or span allocation when no recording parent is in scope; tests cover this directly (tests/test_tracing.py TestCodecSubspansAreHotPathFree).
• Integration test uses polling with rcv.recv(timeout=200) over for _ in range(50), no sleep+assert pattern.
• Filter._process_frames_single wiring of mq.recv_parent_ctx into the process span's context= parameter is correct (filter.py:1296-1302).

[lucasmundim]

Approving — the architecture is sound, test coverage is strong (especially the TestMQRecvSpanEndsOnException regression guard), and the trade-offs are either acknowledged in code or genuinely minor. Two items I'd like addressed before merge, plus a few small gaps that are fine to defer; details in inline comments.

Before merge:

1. Record exceptions on the manual mq.recv / zmq.recv_multipart spans — see inline on mq.py.
2. Either gate the unconditional time_ns() / byte bookkeeping in ZMQReceiver.recv_once on a _trace_enabled flag, or soften RELEASE.md so it claims "no Span allocation" rather than implying zero overhead — see inline on zeromq.py.

Smaller gaps, fine to defer:

• recv_parent_ctx is cleared when self.receiver is None but not on the timeout (res is None) path — minor asymmetry, currently benign because callers don't read it after recv() returns None. Inline comment on mq.py.
• No unit test for the fan-in caveat (multiple upstreams in one recv batch, last-extracted-wins). Out of scope per the PR description and well-documented in code; a future-maintainer-friendly @pytest.mark.skip placeholder would help.
• TestMQRecvSpanEndsOnException asserts the span ends on exception, but not that an error status / recorded exception lands on it. Worth adding once item #1 is in place.

Nice work on the manual-span lifetime regression guard and the W3C round-trip test — those are exactly the right places to invest test effort on this kind of instrumentation.

[stwilt]

Thanks for the review. Pushed f0c51d4 addressing all five findings:

#1 — recv_span leak on exception

Fixed. Wrapped the span lifetime in try/finally and moved end_time capture to after deserialize completes — the old t_recv_end snapshot was pre-deserialize, which meant the span window ended before its own frame.deserialize child, which is malformed in the trace UI even on the happy path. Added TestMQRecvSpanEndsOnException as a regression guard: builds a stand-in receiver returning a malformed envelope that makes topicmsgs2frames raise, asserts the span still reaches the exporter.

#2 — fan-in last-extraction-wins

Surfaced. Added a docstring block at the recv_parent_ctx stash site in MQ.recv explaining that frames produced by other upstreams in the same recv batch end up parented under one arbitrary upstream's mq.send context, that joining all of them would require span links across producer traces, and that this is intentionally out of scope for PLAT-866. A maintainer grepping for "incomplete fan-in traces" will land directly on the explanation.

#3 — _HOP_TRACER mutated without a lock

Documented rather than locked. Added a comment block on the global explaining the concurrency model: single Filter per process, single MQ construction from the main thread before workers start, CPython-GIL-atomic reads on the hot path. Adding a lock would defeat the helper's short-circuit behavior and, as you noted, wouldn't actually fix last-writer-wins — it'd just paper over it. Flagged the comment as the place to revisit if the multi-filter-per-process design ever lands.

#4 — send_span is not None couples to nullcontext()'s yielded None

Fixed — replaced with if tracer:. Agreed the original guard was load-bearing on an implementation detail of contextlib.nullcontext() with no arg.

#5 — zero-duration zmq.recv_multipart span when last_recv_t_start_ns == 0

Fixed — the span is now skipped entirely when last_recv_t_start_ns and last_recv_t_end_ns are both 0, instead of emitting a zero-duration span pinned to t_recv_start. Reachable only on a degenerate path (receiver returns data without any tracked recv_multipart), but cosmetic spans on degenerate paths are still noise in a trace UI.

136/136 unit tests pass locally on the touched surface.

[lucasmundim]

Re-review pass at f0c51d4 — prior pre-merge items still apply, no new blockers. Two cosmetic nits below, fine to defer / fold in with the existing review fixes.

[shingonoide]

The force-push (ecec177) addresses all three items from @lucasmundim's review:

1. Span error status + exception recording on mq.recv — recv_span.record_exception(e) and set_status(StatusCode.ERROR) are now in the except branch (mq.py:392-393), with TestMQRecvSpanEndsOnException as regression guard validating both the span-end guarantee and the ERROR status/event.
2. Hot-path gating — time_ns(), byte-tally accumulation, and otel_extract in ZMQReceiver.recv are behind trace_on = _otel_tracing._HOP_TRACER is not None, snapshotted once per recv(). Symmetric gating on the send side. Matches the RELEASE.md "no overhead when tracing is disabled" contract.
3. send_span guard comment — present and accurate (mq.py:269-272).

Full-file pass notes:

• _register_hop_tracer(tracer) is called unconditionally in MQ.__init__, including the None case, preventing a stale tracer from being reused. Intentional and correct (see comment at mq.py:87-91).
• zmq.recv_multipart sibling span is guarded by if zmq_t_start and zmq_t_end: so zero-duration phantom spans are avoided on OOB-only paths (mq.py:359).
• otel_inject(env) fires unconditionally in ZMQSender.send_maybe, but the W3C propagator only writes traceparent when a recording mq.send span is active — when _HOP_TRACER is None no span is pushed, so no key is written. Behavior is correct.
• The fan-in caveat (last-extracted context wins for multi-upstream receivers) is documented and acknowledged via the skipped placeholder test. Reasonable for PLAT-866 scope.

LGTM.

[lucasmundim]

Re-review pass at ecec177 — LGTM. All three pre-merge items from the prior review are addressed:

1. mq.recv span carries ERROR status + recorded exception via the try/except/finally at mq.py:386-402. TestMQRecvSpanEndsOnException asserts both span lifetime and the ERROR status + exception event — strong guard.
2. Hot-path gating in ZMQSender.send_maybe / ZMQReceiver.recv_once snapshots _HOP_TRACER once per call and skips time_ns(), byte-tally, and otel_extract when tracing is off. RELEASE.md updated to match.
3. The if tracer: guard at mq.py:270 now has the explanatory comment.

Deferred items also picked up: recv_parent_ctx is cleared on the timeout path (mq.py:307-314), and TestFanInLastExtractedWins is in place as a @unittest.skip placeholder with future-maintainer intent.

One non-blocking nit inline.