diff --git a/docs/adr/0001-native-opentelemetry-tracing.md b/docs/adr/0001-native-opentelemetry-tracing.md new file mode 100644 index 000000000..ba49f582b --- /dev/null +++ b/docs/adr/0001-native-opentelemetry-tracing.md @@ -0,0 +1,144 @@ +# ADR 0001: Migrate built-in distributed tracing to native OpenTelemetry + +- Status: **Proposed** +- Date: 2026-07-18 +- Supersedes: the OpenTracing shim bridge (`faust[opentelemetry]` extra, PR #688) — that + remains a valid *interim* for existing deployments, not the destination. + +## Context + +Faust ships a built-in distributed-tracing layer that emits spans for the +processing pipeline — agents, streams, table recovery, the assignor, timers, +Crontabs, and the Kafka rebalance sequence. It is wired through the public +`app.tracer` extension point (`faust.types.app.TracerT`) and is currently +coupled to the **OpenTracing** API: + +- `opentracing>=1.3.0,<=2.4.0` is a **hard core dependency** + (`requirements/requirements.txt`). +- Spans are OpenTracing `Span` objects threaded through ~134 call sites across + 12 modules. +- Current-span propagation uses Python `contextvars` + (`faust/utils/tracing.py`). +- Cross-service context crosses the Kafka boundary via OpenTracing + `inject`/`extract` (`Format.TEXT_MAP`) into message headers + (`faust/sensors/distributed_tracing.py`). + +OpenTracing is **archived / EOL**, and the OpenTelemetry project formally +**deprecated OpenTracing compatibility on 2026-03-19** (spec PR #4938). The +`opentelemetry-opentracing-shim` package still ships (0.65b0, 2026-07-16) but +its earliest removal is no sooner than **March 2027**. Building the long-term +story on the shim is building on a sunset path. + +### Why not delegate to an instrumentor + +There is **no `opentelemetry-instrumentation-faust`** package — not on PyPI, +not in `opentelemetry-python-contrib`, and nothing community-maintained. +(`opentelemetry-instrumentation-faststream` targets the unrelated *FastStream* +library.) Transport-level instrumentors do exist — +`opentelemetry-instrumentation-aiokafka` and +`opentelemetry-instrumentation-confluent-kafka` (both 0.65b0, beta) — but they +patch the *raw* Kafka clients and produce only produce/consume spans. They know +nothing about Faust's agent / stream / recovery / rebalance boundaries, which is +precisely the value Faust's built-in tracing adds. They **complement** Faust's +spans; they cannot replace them. + +## Decision + +Rewrite the built-in tracing layer to emit **native OpenTelemetry spans**, +depending only on `opentelemetry-api`, while **keeping the full span tree**. + +Key design points: + +1. **API-only dependency.** As a library, Faust depends on `opentelemetry-api` + (an optional extra), never the SDK. Calls are cheap no-ops until the *app* + registers a `TracerProvider`. This replaces the hard `opentracing` core + dependency. + +2. **`app.tracer` becomes optional.** Native OTel apps configure a global + `TracerProvider`; Faust resolves its tracer via + `opentelemetry.trace.get_tracer("faust", __version__)`. Users no longer need + to implement `TracerT` at all. `app.tracer` is retained as an explicit + override/legacy seam. + +3. **Contextvars map 1:1.** `opentelemetry.context` is contextvars-based by + default, so Faust's existing current-span propagation — including across + `await` boundaries in agents/streams — carries over without new plumbing. + Extracted remote context is made current with `context.attach`/`detach`. + +4. **W3C propagation across Kafka headers.** Header inject/extract moves from + OpenTracing to `opentelemetry.propagate` (W3C `traceparent`). This is a + wire-format change, and an improvement: it is the standard format and lets + Faust spans nest correctly under the transport instrumentors when a user + enables them. + +5. **Semantic-convention coexistence** (see below). + +6. **Breaking change, phased with a compatibility adapter** (see below). + +### Semantic-convention coexistence + +Faust adopts the OpenTelemetry `messaging.*` semantic conventions **additively**, +without breaking existing dashboards: + +- **Span names stay legacy by default** (`consume-from-{topic}`, `job-{topic}`, + `produce-to-{topic}`, …). A one-time deprecation notice is logged, stating the + legacy names will become opt-in in a future major and semconv names + (`{topic} process`, `{topic} publish`, …) will become the default. +- **Attributes carry both.** Every span emits its legacy tags + (`kafka-topic`, `kafka-partition`, …) *and* the semconv attributes + (`messaging.system=kafka`, `messaging.destination.name`, + `messaging.operation.type`, `messaging.destination.partition.id`, …), plus the + appropriate `SpanKind` (`CONSUMER`/`PRODUCER`). +- **Opt-in name flip.** A setting (`app.conf`, e.g. `tracing_span_names`, + default `"legacy"`) lets early adopters switch span *names* to semconv now. + Attributes coexist in both modes. + +This satisfies "preserve current names but indicate they're deprecated, and +adopt semconv in a coexisting way." + +### Re-expressing the two non-portable hacks + +Both were prototyped against the real SDK before this ADR. + +1. **Deterministic rebalance trace_id.** Today + `span.context.trace_id = murmur2(f"reb-{app_id}-{generation}")` mutates the + span context — impossible in OTel (contexts are immutable). Native + re-expression: mint a `NonRecordingSpan` parent whose `SpanContext` carries + the murmur2-derived trace_id, put it in context via `set_span_in_context`, + and start the real rebalance span as its child; the child inherits the + deterministic trace_id. Cross-generation relationships that previously relied + on trace_id rewriting use **span links**. + +2. **Lazy/monkeypatched `span.finish`.** Today `_transform_span_lazy` builds a + dynamic `LazySpan` subclass overriding `.finish`. OTel spans are not + subclassable that way. Native re-expression: **defer span creation** — buffer + a lightweight pending descriptor and start+end the real span when + `on_generation_id_known` fires (or cancel it in `flush_spans`). + +## Migration phases + +- **Phase 0 — interim (done):** `opentracing` optional (#686) + shim bridge + (#688). Non-breaking. +- **Phase 1a — foundation (this change):** native OTel tracing core + (`faust/utils/otel_tracing.py`) with tracer resolution, deterministic + rebalance context, semconv coexistence helpers, W3C header propagation, and + the deferred-finish helper — fully unit-tested, **no call sites rewired yet**. + Adds the `faust[opentelemetry]` extra (`opentelemetry-api`). +- **Phase 1b — rewire:** convert the 12 modules to the native core; redefine + `TracerT` in OTel terms; ship a legacy adapter that wraps a user's existing + OpenTracing `app.tracer` through the shim, kept behind `faust[opentracing]` + for one major cycle. Target a **major version bump**. +- **Phase 2 — remove:** drop `opentracing` from core requirements and delete the + legacy adapter in the following major. + +## Consequences + +- **Positive:** no archived/EOL dependency in core; standard W3C wire format; + interoperates with the transport instrumentors; cleaner code (no context + mutation, no monkeypatch); `app.tracer` becomes optional; SDK choice deferred + to the app as OTel intends. +- **Negative / risk:** `app.tracer`/`TracerT` signature change is breaking — + mitigated by the phased adapter and a major-version bump. Span *names* remain + legacy by default to protect existing dashboards; the eventual default flip to + semconv is a separately announced deprecation. The rebalance link/`NonRecordingSpan` + approach must be validated against real backends (Jaeger/Tempo) during 1b. diff --git a/faust/utils/otel_tracing.py b/faust/utils/otel_tracing.py new file mode 100644 index 000000000..46ae4f38b --- /dev/null +++ b/faust/utils/otel_tracing.py @@ -0,0 +1,323 @@ +"""Native OpenTelemetry tracing core for Faust. + +Phase 1a foundation for the native-OpenTelemetry rewrite of Faust's built-in +distributed-tracing layer. See +``docs/adr/0001-native-opentelemetry-tracing.md`` for the full decision record. + +This module provides the reusable primitives the call-site conversion (Phase 1b) +will build on: + +* tracer resolution against the globally-registered ``TracerProvider`` +* current-span propagation via ``contextvars`` (parity with the legacy layer) +* a deterministic parent context for rebalance spans -- the native + re-expression of the old ``span.context.trace_id = murmur2(...)`` mutation, + which is impossible in OpenTelemetry because span contexts are immutable +* a deferred ("lazy") span helper -- the native re-expression of the old + ``span.finish`` monkeypatch +* semantic-convention *coexistence* helpers that set both the legacy tags and + the OpenTelemetry ``messaging.*`` attributes on every span +* W3C header inject/extract for cross-service context propagation + +It depends only on ``opentelemetry-api`` (the optional ``faust[opentelemetry]`` +extra) and degrades to no-ops when OpenTelemetry is not installed, so importing +it never hard-fails Faust core. +""" + +import hashlib +from contextvars import ContextVar +from typing import TYPE_CHECKING, Any, Dict, Mapping, Optional + +try: + from opentelemetry import propagate, trace + from opentelemetry.trace import ( + NonRecordingSpan, + SpanContext, + SpanKind, + TraceFlags, + set_span_in_context, + ) + + HAS_OTEL = True +except ImportError: # pragma: no cover + HAS_OTEL = False + +if TYPE_CHECKING: + from opentelemetry.context import Context + from opentelemetry.trace import Span, Tracer + +__all__ = [ + "HAS_OTEL", + "SPAN_NAMES_LEGACY", + "SPAN_NAMES_SEMCONV", + "PendingSpan", + "otel_available", + "resolve_tracer", + "current_span", + "set_current_span", + "noop_span", + "trace_id_from_seed", + "deterministic_parent_context", + "add_default_tags", + "add_consume_attributes", + "add_produce_attributes", + "inject_trace_headers", + "extract_trace_context", +] + +#: Span-name mode: keep the historical Faust names (deprecated, default). +SPAN_NAMES_LEGACY = "legacy" +#: Span-name mode: use OpenTelemetry messaging semantic-convention names. +SPAN_NAMES_SEMCONV = "semconv" + +# -- OpenTelemetry messaging semantic-convention attribute keys -*- +_MSG_SYSTEM = "messaging.system" +_MSG_DESTINATION = "messaging.destination.name" +_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" + +_current_span: "ContextVar[Optional[Any]]" = ContextVar( + "faust_otel_current_span", default=None +) + + +def otel_available() -> bool: + """Return :const:`True` if the OpenTelemetry API is importable.""" + return HAS_OTEL + + +def resolve_tracer( + service_name: str = "faust", version: Optional[str] = None +) -> Optional["Tracer"]: + """Return a tracer from the globally-registered ``TracerProvider``. + + Returns :const:`None` when OpenTelemetry is not installed. When no + provider has been configured by the application, the OpenTelemetry API + returns a no-op tracer whose spans are cheap and non-exporting, so callers + never need to special-case the unconfigured state. + """ + if not HAS_OTEL: # pragma: no cover + return None + return trace.get_tracer(service_name, version) + + +def current_span() -> Optional["Span"]: + """Return the current Faust span for this context, if any.""" + return _current_span.get() + + +def set_current_span(span: Optional["Span"]) -> None: + """Set the current Faust span for the current context.""" + _current_span.set(span) + + +def noop_span() -> Optional["Span"]: + """Return a non-recording span (or :const:`None` without OpenTelemetry).""" + if not HAS_OTEL: # pragma: no cover + return None + return trace.INVALID_SPAN + + +def trace_id_from_seed(seed: str) -> int: + """Derive a stable, non-zero 128-bit trace id from a seed string. + + Used for rebalance spans so that every member computing the same + ``reb-{app_id}-{generation}`` seed lands on the same trace, without + mutating an (immutable) OpenTelemetry span context. + """ + digest = hashlib.blake2b(seed.encode(), digest_size=16).digest() + trace_id = int.from_bytes(digest, "big") + # A trace id of 0 is invalid in OpenTelemetry; blake2b collisions to zero + # are astronomically unlikely, but guard anyway. + return trace_id or 1 + + +def deterministic_parent_context(trace_id: int) -> Optional["Context"]: + """Build a context carrying a deterministic ``trace_id`` for a child span. + + Wraps a :class:`~opentelemetry.trace.NonRecordingSpan` whose + :class:`~opentelemetry.trace.SpanContext` carries *trace_id*; starting a + span with the returned context makes it a child that inherits *trace_id*. + This is the native replacement for the old + ``span.context.trace_id = murmur2(...)`` mutation. + """ + if not HAS_OTEL: # pragma: no cover + return None + # span_id must be a non-zero 64-bit int; derive it from the trace id so it + # is stable per (app, generation) too. + span_id = (trace_id & 0xFFFFFFFFFFFFFFFF) or 1 + parent_context = SpanContext( + trace_id=trace_id, + span_id=span_id, + is_remote=False, + trace_flags=TraceFlags(TraceFlags.SAMPLED), + ) + return set_span_in_context(NonRecordingSpan(parent_context)) + + +def add_default_tags(span: Optional["Span"], app_name: str, app_id: str) -> None: + """Set Faust's default identifying tags on *span*.""" + if span is None or not HAS_OTEL: + return + span.set_attribute("faust_app", app_name) + span.set_attribute("faust_id", app_id) + + +def _set_messaging_common( + span: "Span", + *, + topic: str, + partition: Optional[int], + key: Any, + offset: Optional[int], + operation: str, +) -> None: + span.set_attribute(_MSG_SYSTEM, "kafka") + span.set_attribute(_MSG_DESTINATION, topic) + span.set_attribute(_MSG_OPERATION_TYPE, operation) + if partition is not None: + span.set_attribute(_MSG_PARTITION, str(partition)) + if offset is not None: + span.set_attribute(_MSG_KAFKA_OFFSET, offset) + if key is not None: + span.set_attribute(_MSG_KAFKA_KEY, _stringify_key(key)) + + +def _stringify_key(key: Any) -> str: + if isinstance(key, bytes): + try: + return key.decode("utf-8") + except UnicodeDecodeError: + return key.hex() + return str(key) + + +def add_consume_attributes( + span: Optional["Span"], + *, + topic: str, + partition: Optional[int] = None, + key: Any = None, + offset: Optional[int] = None, +) -> None: + """Set consume-side attributes, coexisting legacy tags and semconv. + + Emits the historical ``kafka-*`` tags *and* the OpenTelemetry + ``messaging.*`` attributes on the same span (see the ADR: names stay + legacy, attributes carry both). + """ + if span is None or not HAS_OTEL: + return + # Legacy tags (deprecated -- retained for existing dashboards). + span.set_attribute("kafka-topic", topic) + if partition is not None: + span.set_attribute("kafka-partition", partition) + if key is not None: + span.set_attribute("kafka-key", _stringify_key(key)) + # Semantic conventions. + _set_messaging_common( + span, + topic=topic, + partition=partition, + key=key, + offset=offset, + operation="process", + ) + + +def add_produce_attributes( + span: Optional["Span"], + *, + topic: str, + partition: Optional[int] = None, + key: Any = None, + offset: Optional[int] = None, +) -> None: + """Set produce-side attributes, coexisting legacy tags and semconv.""" + if span is None or not HAS_OTEL: + return + span.set_attribute("kafka-topic", topic) + if partition is not None: + span.set_attribute("kafka-partition", partition) + if offset is not None: + span.set_attribute("kafka-offset", offset) + _set_messaging_common( + span, + topic=topic, + partition=partition, + key=key, + offset=offset, + operation="send", + ) + + +def inject_trace_headers(carrier: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: + """Inject the current context into *carrier* as W3C trace headers.""" + if carrier is None: + carrier = {} + if not HAS_OTEL: # pragma: no cover + return carrier + propagate.inject(carrier) + return carrier + + +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) + + +class PendingSpan: + """A span whose creation is deferred until enough state is known. + + Native replacement for the old lazy-finish monkeypatch: instead of starting + a span and rewriting its ``.finish`` to run later, buffer the *intent* to + trace and materialize the real span once the deferred state (e.g. the Kafka + generation id) is available -- or drop it entirely. + """ + + def __init__( + self, + tracer: Optional["Tracer"], + name: str, + *, + kind: Optional[Any] = None, + attributes: Optional[Dict[str, Any]] = None, + context: Optional["Context"] = None, + ) -> None: + self.tracer = tracer + self.name = name + self.kind = kind + self.attributes = attributes + self.context = context + self._span: Optional["Span"] = None + + def materialize(self) -> Optional["Span"]: + """Start and return the real span, or :const:`None` without OTel.""" + if self.tracer is None or not HAS_OTEL: + return None + kind = self.kind if self.kind is not None else SpanKind.INTERNAL + self._span = self.tracer.start_span( + self.name, + context=self.context, + kind=kind, + attributes=self.attributes, + ) + return self._span + + 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: + self._span.update_name(f"{self.name} ({note})") + self._span.end() + self._span = None + + +#: Mapping of legacy operation -> semconv span-name builder, for the opt-in +#: ``SPAN_NAMES_SEMCONV`` mode. Consumed in Phase 1b when call sites are +#: rewired; declared here so the naming contract lives with the core. +def semconv_span_name(operation: str, destination: str) -> str: + """Return the semantic-convention span name ``{destination} {operation}``.""" + return f"{destination} {operation}" diff --git a/requirements/extras/opentelemetry.txt b/requirements/extras/opentelemetry.txt new file mode 100644 index 000000000..721d17b5d --- /dev/null +++ b/requirements/extras/opentelemetry.txt @@ -0,0 +1,5 @@ +# Native OpenTelemetry tracing (see docs/adr/0001-native-opentelemetry-tracing.md). +# Faust's tracing core depends only on the API; the SDK is included so apps can +# configure an exporter out of the box. +opentelemetry-api>=1.20.0 +opentelemetry-sdk>=1.20.0 diff --git a/requirements/test.txt b/requirements/test.txt index 832762190..0b7ecfedc 100644 --- a/requirements/test.txt +++ b/requirements/test.txt @@ -36,3 +36,4 @@ intervaltree -r extras/statsd.txt -r extras/yaml.txt -r extras/prometheus.txt +-r extras/opentelemetry.txt diff --git a/setup.py b/setup.py index 390cad0a9..27bd37fe4 100644 --- a/setup.py +++ b/setup.py @@ -30,6 +30,7 @@ "datadog", "debug", "fast", + "opentelemetry", "orjson", "prometheus", "redis", diff --git a/tests/unit/utils/test_otel_tracing.py b/tests/unit/utils/test_otel_tracing.py new file mode 100644 index 000000000..6ca9a5715 --- /dev/null +++ b/tests/unit/utils/test_otel_tracing.py @@ -0,0 +1,229 @@ +"""Unit tests for the native OpenTelemetry tracing core (Phase 1a foundation).""" + +import pytest + +from faust.utils import otel_tracing + +# The whole module is meaningless without the OpenTelemetry API, and its SDK is +# what lets us assert on emitted spans -- skip cleanly when absent. +pytest.importorskip("opentelemetry.sdk.trace") + +from opentelemetry import trace # noqa: E402 +from opentelemetry.sdk.trace import TracerProvider # noqa: E402 +from opentelemetry.sdk.trace.export import SimpleSpanProcessor # noqa: E402 +from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( # noqa: E402 + InMemorySpanExporter, +) +from opentelemetry.trace import SpanKind # noqa: E402 + + +@pytest.fixture +def exporter(): + """Register a global provider backed by an in-memory exporter.""" + exp = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exp)) + # ``set_tracer_provider`` only takes effect once per process; override the + # internal proxy so every test gets a fresh exporter. + trace._TRACER_PROVIDER = provider + yield exp + exp.clear() + + +@pytest.fixture +def tracer(exporter): + return otel_tracing.resolve_tracer("faust-test") + + +def _finished(exporter): + return {s.name: s for s in exporter.get_finished_spans()} + + +def test_otel_available(): + assert otel_tracing.otel_available() is True + + +def test_resolve_tracer_returns_tracer(tracer): + assert tracer is not None + assert hasattr(tracer, "start_span") + + +def test_current_span_roundtrip(): + assert otel_tracing.current_span() is None + sentinel = object() + otel_tracing.set_current_span(sentinel) + assert otel_tracing.current_span() is sentinel + otel_tracing.set_current_span(None) + assert otel_tracing.current_span() is None + + +def test_noop_span_is_non_recording(): + span = otel_tracing.noop_span() + assert span is trace.INVALID_SPAN + assert span.is_recording() is False + + +def test_trace_id_from_seed_is_deterministic_and_nonzero(): + a = otel_tracing.trace_id_from_seed("reb-myapp-7") + b = otel_tracing.trace_id_from_seed("reb-myapp-7") + c = otel_tracing.trace_id_from_seed("reb-myapp-8") + assert a == b + assert a != c + assert 0 < a < (1 << 128) + + +def test_deterministic_parent_context_yields_child_with_seed_trace_id(tracer, exporter): + trace_id = otel_tracing.trace_id_from_seed("reb-myapp-7") + ctx = otel_tracing.deterministic_parent_context(trace_id) + span = tracer.start_span("rebalancing", context=ctx, kind=SpanKind.CONSUMER) + span.end() + + reb = _finished(exporter)["rebalancing"] + assert reb.context.trace_id == trace_id + assert reb.kind is SpanKind.CONSUMER + + +def test_two_members_same_generation_share_trace(tracer, exporter): + for _ in range(2): + ctx = otel_tracing.deterministic_parent_context( + otel_tracing.trace_id_from_seed("reb-myapp-42") + ) + tracer.start_span("rebalancing", context=ctx).end() + + trace_ids = {s.context.trace_id for s in exporter.get_finished_spans()} + assert len(trace_ids) == 1 + + +def test_default_tags(tracer, exporter): + span = tracer.start_span("op") + otel_tracing.add_default_tags(span, "myapp", "myapp-id") + span.end() + + attrs = _finished(exporter)["op"].attributes + assert attrs["faust_app"] == "myapp" + assert attrs["faust_id"] == "myapp-id" + + +def test_consume_attributes_coexist_legacy_and_semconv(tracer, exporter): + span = tracer.start_span("consume-from-orders", kind=SpanKind.CONSUMER) + otel_tracing.add_consume_attributes( + span, topic="orders", partition=3, key=b"user-1", offset=99 + ) + span.end() + + got = _finished(exporter) + # Legacy name preserved. + assert "consume-from-orders" in got + attrs = got["consume-from-orders"].attributes + # Legacy tags (deprecated) present. + assert attrs["kafka-topic"] == "orders" + assert attrs["kafka-partition"] == 3 + assert attrs["kafka-key"] == "user-1" + # Semantic conventions present on the same span. + assert attrs["messaging.system"] == "kafka" + assert attrs["messaging.destination.name"] == "orders" + assert attrs["messaging.operation.type"] == "process" + assert attrs["messaging.destination.partition.id"] == "3" + assert attrs["messaging.kafka.offset"] == 99 + + +def test_produce_attributes_coexist_legacy_and_semconv(tracer, exporter): + span = tracer.start_span("produce-to-orders", kind=SpanKind.PRODUCER) + otel_tracing.add_produce_attributes(span, topic="orders", partition=1, offset=5) + span.end() + + attrs = _finished(exporter)["produce-to-orders"].attributes + assert attrs["kafka-topic"] == "orders" + assert attrs["kafka-offset"] == 5 + assert attrs["messaging.system"] == "kafka" + assert attrs["messaging.operation.type"] == "send" + + +def test_stringify_key_handles_non_utf8(tracer, exporter): + span = tracer.start_span("op") + otel_tracing.add_consume_attributes(span, topic="t", key=b"\xff\xfe") + span.end() + # Non-decodable bytes fall back to hex rather than raising. + assert _finished(exporter)["op"].attributes["kafka-key"] == "fffe" + + +def test_inject_extract_roundtrip_reparents(tracer, exporter): + # Produce under a parent span, inject headers. + parent = tracer.start_span("produce-to-t", kind=SpanKind.PRODUCER) + token = trace.set_span_in_context(parent) + from opentelemetry import context as otel_ctx + + ctx_token = otel_ctx.attach(token) + carrier = otel_tracing.inject_trace_headers({}) + otel_ctx.detach(ctx_token) + parent.end() + + assert "traceparent" in carrier + + # Consume: extract and start a child under the extracted context. + extracted = otel_tracing.extract_trace_context(carrier) + child = tracer.start_span("consume-from-t", context=extracted) + child.end() + + got = _finished(exporter) + assert ( + got["consume-from-t"].context.trace_id == got["produce-to-t"].context.trace_id + ) + + +def test_pending_span_materializes_late(tracer, exporter): + pending = otel_tracing.PendingSpan( + tracer, + "rebalancing", + kind=SpanKind.CONSUMER, + attributes={"kafka_member_id": "m1"}, + ) + # Nothing emitted until materialized. + assert exporter.get_finished_spans() == () + span = pending.materialize() + assert span is not None + span.end() + + reb = _finished(exporter)["rebalancing"] + assert reb.attributes["kafka_member_id"] == "m1" + assert reb.kind is SpanKind.CONSUMER + + +def test_pending_span_cancel_marks_and_ends(tracer, exporter): + pending = otel_tracing.PendingSpan(tracer, "rebalancing") + pending.materialize() + pending.cancel() + + names = [s.name for s in exporter.get_finished_spans()] + assert names == ["rebalancing (CANCELLED)"] + + +def test_semconv_span_name_builder(): + assert otel_tracing.semconv_span_name("process", "orders") == "orders process" + assert otel_tracing.semconv_span_name("publish", "orders") == "orders publish" + + +def test_attribute_helpers_noop_on_none_span(): + # None spans are common (noop tracing); helpers must be silent no-ops. + otel_tracing.add_default_tags(None, "myapp", "myapp-id") + otel_tracing.add_consume_attributes(None, topic="t", partition=1) + otel_tracing.add_produce_attributes(None, topic="t", offset=2) + + +def test_stringify_key_str_and_int(tracer, exporter): + span = tracer.start_span("op") + otel_tracing.add_consume_attributes(span, topic="t", key="plain-str") + span.end() + assert _finished(exporter)["op"].attributes["kafka-key"] == "plain-str" + + +def test_inject_trace_headers_defaults_to_new_carrier(): + carrier = otel_tracing.inject_trace_headers() + assert isinstance(carrier, dict) + + +def test_pending_span_without_tracer_is_noop(): + pending = otel_tracing.PendingSpan(None, "x") + assert pending.materialize() is None + # cancel is a no-op when nothing was materialized. + pending.cancel()