Add native OpenTelemetry tracing foundation (Phase 1a) - #782
Conversation
Groundwork for migrating Faust's built-in distributed tracing off the archived OpenTracing API to native OpenTelemetry, keeping the full span tree. This first increment adds the decision record and a self-contained, unit-tested tracing core; no existing call sites are rewired yet, so the running tracing layer is unaffected. - docs/adr/0001-native-opentelemetry-tracing.md: architecture decision record (options, why no opentelemetry-instrumentation-faust exists, semconv coexistence, re-expression of the two non-portable hacks, and the phased migration plan). - faust/utils/otel_tracing.py: api-only tracing primitives that degrade to no-ops when OpenTelemetry is absent -- tracer resolution against the global provider, contextvar current-span, deterministic rebalance parent context (native replacement for the immutable-context trace_id mutation), a deferred PendingSpan (native replacement for the finish monkeypatch), semconv-coexistence attribute helpers (legacy tags + messaging.*), and W3C header inject/extract. - tests/unit/utils/test_otel_tracing.py: 15 tests covering every primitive against an in-memory SDK exporter. - requirements/extras/opentelemetry.txt + setup.py: faust[opentelemetry] extra (opentelemetry-api/sdk). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL
The Phase 1a tests guard on `pytest.importorskip("opentelemetry.sdk.trace")`,
so without OpenTelemetry installed in CI they were all skipped, leaving
faust/utils/otel_tracing.py at 29% patch coverage and failing the codecov
gate.
- requirements/test.txt: pull in extras/opentelemetry.txt so the test job
installs opentelemetry-api/sdk and the 15 tests actually execute
(matches the existing `-r extras/*.txt` pattern in test.txt).
- otel_tracing.py: mark the dependency-absent `if not HAS_OTEL:` fallbacks
with `# pragma: no cover` -- they only run when OpenTelemetry is not
installed, a config CI no longer exercises.
- test_otel_tracing.py: add tests for the real runtime guards (None span,
None tracer, non-bytes key, default carrier) instead of pragma-ing them.
Module coverage is now 96%.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL
Brings the OTel tracing foundation up to date with master, which has since landed the sensor-level OpenTelemetry path (#748), the OTel metrics monitor (#746), and moved `opentracing` out of core requirements into an extra (#686). Conflict resolutions: - `requirements/extras/opentelemetry.txt` (add/add): keep master's API floor and the fastapi/aiokafka instrumentation packages, and add `opentelemetry-sdk` on top so apps can configure an exporter out of the box. The tracing core itself still needs only the API. - `setup.py`: take master's bundle list, which already contains the `opentelemetry` entry this branch added. - `requirements/test.txt`: drop this branch's `-r extras/opentelemetry.txt` line; master already pulls the extra in, plus explicit api/sdk pins. The ADR is refreshed against the new tree: `opentracing` is described as an extra rather than a hard core dependency, a section explains how this work relates to `faust.contrib.opentelemetry` (sensor covers the message boundary, this covers the built-in span tree), and the phase list records what has already shipped. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TBE5RZ5mb1h5MjYyWGwmHZ
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 7308235bde
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| _MSG_PARTITION = "messaging.destination.partition.id" | ||
| _MSG_OPERATION_TYPE = "messaging.operation.type" | ||
| _MSG_KAFKA_KEY = "messaging.kafka.message.key" | ||
| _MSG_KAFKA_OFFSET = "messaging.kafka.offset" |
There was a problem hiding this comment.
When an offset is provided, this helper records it as messaging.kafka.offset, but the Kafka semantic-convention key used elsewhere in this repository (faust.contrib.opentelemetry) is messaging.kafka.message.offset. Backends and dashboards querying the standard key will therefore treat every span produced through this new core as lacking an offset; the test currently enshrines the same incorrect key.
Useful? React with 👍 / 👎.
| def extract_trace_context(carrier: Mapping[str, Any]) -> Optional["Context"]: | ||
| """Extract a parent context from *carrier* (W3C trace headers).""" | ||
| if not HAS_OTEL: # pragma: no cover | ||
| return None | ||
| return propagate.extract(carrier) |
There was a problem hiding this comment.
When this is called with Faust's normal Kafka header representation—a list of (str, bytes) pairs—the default OpenTelemetry getter expects a mapping and attempts mapping operations on the list; byte-valued mapping headers also are not converted to text for the W3C parser. The existing faust.contrib.opentelemetry sensor needs a custom Kafka header getter for exactly these cases, so using this foundation on real consumed messages will fail to recover the producer context and create disconnected traces.
Useful? React with 👍 / 👎.
| self._span = self.tracer.start_span( | ||
| self.name, | ||
| context=self.context, | ||
| kind=kind, | ||
| attributes=self.attributes, | ||
| ) |
There was a problem hiding this comment.
When a PendingSpan is created at operation start but materialized only after the Kafka generation becomes known, start_span receives no buffered start_time, so the exported span starts at the later callback rather than when the traced operation began. If it is then ended as the deferred-finish replacement describes, a potentially long rebalance collapses to a near-zero-duration span; capture the creation timestamp and pass it to start_span so the migration preserves timing as well as the span name and attributes.
Useful? React with 👍 / 👎.
| def cancel(self, note: str = "CANCELLED") -> None: | ||
| """Drop the pending span; if already materialized, mark and end it.""" | ||
| if self._span is not None and HAS_OTEL: |
There was a problem hiding this comment.
When cancel() is called while the span is still pending, _span is None, so this method records no cancelled state and a later materialize() still emits the span. That contradicts the documented behavior of dropping a pending span and can reintroduce spans after a flush if any callback retains the descriptor; persist cancellation and have materialize() remain a no-op afterward.
Useful? React with 👍 / 👎.
| def set_current_span(span: Optional["Span"]) -> None: | ||
| """Set the current Faust span for the current context.""" | ||
| _current_span.set(span) |
There was a problem hiding this comment.
When callers establish a span through this module's set_current_span, only the private Faust ContextVar changes; the OpenTelemetry current context read by propagate.inject() and by Tracer.start_span() without an explicit parent remains unchanged. Produced headers will therefore omit that span and implicit descendants will become unrelated roots unless every call site separately calls context.attach(); the round-trip test masks this by manually attaching its parent instead of using set_current_span.
Useful? React with 👍 / 👎.
| digest = hashlib.blake2b(seed.encode(), digest_size=16).digest() | ||
| trace_id = int.from_bytes(digest, "big") |
There was a problem hiding this comment.
During a rolling migration where old and native-tracing members participate in the same Kafka generation, the legacy driver derives the shared trace ID with murmur2, while this replacement derives a different ID with BLAKE2. Members computing the same reb-{app_id}-{generation} seed will consequently split the rebalance across two traces precisely during the advertised compatibility period; use the existing murmur2-derived value so cross-version members preserve the deterministic grouping.
Useful? React with 👍 / 👎.
| self.attributes = attributes | ||
| self.context = context | ||
| self._span: Optional["Span"] = None |
There was a problem hiding this comment.
When a PendingSpan is constructed while a parent is current but materialized after that context has been detached or replaced, retaining None here makes start_span consult the context current at materialization time. The deferred span can therefore become a root or a child of an unrelated operation instead of preserving the original span tree; when no explicit context is supplied, snapshot OpenTelemetry's current context in the constructor.
Useful? React with 👍 / 👎.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #782 +/- ##
==========================================
+ Coverage 96.20% 96.24% +0.03%
==========================================
Files 110 111 +1
Lines 11790 11893 +103
Branches 1281 1294 +13
==========================================
+ Hits 11343 11446 +103
Misses 350 350
Partials 97 97 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Master's project coverage has risen to 96.20%, which raises codecov's auto-target for this PR's patch above the 94.33% it has always measured, so `codecov/patch` and `codecov/project` went red on the sync commit rather than on any new code. Two causes, both in the new module: - `.coveragerc` excludes the `if typing.TYPE_CHECKING:` spelling, which is what the rest of faust uses. `otel_tracing.py` imports the name directly and writes bare `if TYPE_CHECKING:`, so the regex misses it and the two import lines under it count as unexecuted. Marked `# pragma: no cover`, matching the one other bare-form site (`faust/models/typing.py`). - The optional-argument arms of `add_consume_attributes` / `add_produce_attributes` were only ever called with every optional supplied, leaving four partial branches. Added two tests covering the topic-only path, which also pin the intended behavior: an unknown partition/key/offset is omitted from the span rather than recorded as a null. `faust/utils/otel_tracing.py` now reports 100% (103 statements, 26 branches). Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TBE5RZ5mb1h5MjYyWGwmHZ
9643140 to
3ec48dc
Compare
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 3ec48dcd7e
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| ) -> None: | ||
| span.set_attribute(_MSG_SYSTEM, "kafka") | ||
| span.set_attribute(_MSG_DESTINATION, topic) | ||
| span.set_attribute(_MSG_OPERATION_TYPE, operation) |
There was a problem hiding this comment.
When Phase 1b routes consume and produce spans through these helpers, they will emit messaging.operation.type but omit messaging.operation.name; Faust's existing native sensor emits both attributes in faust/contrib/opentelemetry.py:182-183. Backends and dashboards that classify messaging spans by the operation-name semantic-convention attribute will therefore fail to recognize spans from the new tracing core, so set the corresponding operation name here as well.
Useful? React with 👍 / 👎.
| trace_id=trace_id, | ||
| span_id=span_id, | ||
| is_remote=False, | ||
| trace_flags=TraceFlags(TraceFlags.SAMPLED), |
There was a problem hiding this comment.
When an application configures a parent-based sampler such as ParentBased(TraceIdRatioBased(0.01)), marking this fabricated parent as sampled makes the sampler take its sampled-parent branch, so every rebalance child is recorded regardless of the configured root sampling rate. Because this synthetic context carries no real upstream sampling decision, derive or accept the trace flags according to the application's sampling policy rather than unconditionally forcing SAMPLED.
Useful? React with 👍 / 👎.
| def add_consume_attributes( | ||
| span: Optional["Span"], | ||
| *, | ||
| topic: str, | ||
| partition: Optional[int] = None, |
There was a problem hiding this comment.
When multiple Faust consumer groups process the same topic, spans produced by this helper cannot be distinguished by the standard consumer-group dimension because the API neither accepts nor emits messaging.consumer.group.name. The existing native sensor obtains the known group from app.conf.id and emits it in faust/contrib/opentelemetry.py:189-192; Phase 1b should preserve that behavior by accepting the group here and setting the corresponding attribute.
Useful? React with 👍 / 👎.
| carrier = {} | ||
| if not HAS_OTEL: # pragma: no cover | ||
| return carrier | ||
| propagate.inject(carrier) |
There was a problem hiding this comment.
When an application enables opentelemetry-instrumentation-aiokafka from the same extra, that instrumentor already injects a traceparent during producer.send and appends its header unconditionally, as documented in faust/contrib/opentelemetry.py:49-55. If Phase 1b first uses this helper for the native producer span, each Kafka record receives two traceparent headers; consumers that take the first value will skip the transport send span, while other propagators may reject the ambiguous context. Suppress one injector or explicitly replace the existing propagation headers.
Useful? React with 👍 / 👎.
2 participants
Groundwork for migrating Faust's built-in distributed tracing off the
archived OpenTracing API to native OpenTelemetry, keeping the full span
tree. This first increment adds the decision record and a self-contained,
unit-tested tracing core; no existing call sites are rewired yet, so the
running tracing layer is unaffected.
record (options, why no opentelemetry-instrumentation-faust exists,
semconv coexistence, re-expression of the two non-portable hacks, and
the phased migration plan).
no-ops when OpenTelemetry is absent -- tracer resolution against the
global provider, contextvar current-span, deterministic rebalance parent
context (native replacement for the immutable-context trace_id mutation),
a deferred PendingSpan (native replacement for the finish monkeypatch),
semconv-coexistence attribute helpers (legacy tags + messaging.*), and
W3C header inject/extract.
against an in-memory SDK exporter.
extra (opentelemetry-api/sdk).
Co-Authored-By: Claude Opus 4.8 noreply@anthropic.com
Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL