faust.contrib.opentelemetry

OpenTelemetry tracing for Faust, and for a co-hosted ASGI application.

Why this exists

https://pypi.org/project/opentelemetry-instrumentation-aiokafka/ already wraps AIOKafkaProducer.send and AIOKafkaConsumer.getmany, which is exactly what Faust’s aiokafka driver calls. So with that package installed you already get most of a distributed trace for free:

FastAPI server span
  -> aiokafka "{topic} send"     (PRODUCER, injects ``traceparent``)
  -> [Kafka]
  -> aiokafka "{topic} receive"  (CONSUMER, extracts ``traceparent``)
  -> ???

The last hop is the one nobody outside Faust can supply. Faust’s consumer runs in its own thread (ConsumerThread), and contextvars never cross threads – so the receive span is opened and closed inside getmany, on a thread the agent never runs on. Without help you get an orphaned receive span and an unparented agent, which reads worse in a trace viewer than no instrumentation at all.

OpenTelemetrySensor closes that gap. It extracts the trace context from the Kafka message headers and opens a {topic} process span that stays current for exactly as long as the stream is processing the event, so anything the agent does – an HTTP call, a database query, another topic.send() – nests underneath it.

Usage

from faust.contrib.opentelemetry import setup_opentelemetry

app = faust.App("myapp", broker="kafka://localhost:9092")
setup_opentelemetry(app)

Configure an SDK the usual way (opentelemetry-sdk plus an exporter, or the opentelemetry-instrument CLI). Until you do, the OpenTelemetry API is a no-op and this module costs effectively nothing.

Install with pip install faust-streaming[opentelemetry].

Notes

  • This module depends only on opentelemetry-api. A library must never configure the SDK, so nothing here calls set_tracer_provider().

  • Trace context is read from message headers, never written. On the produce side opentelemetry-instrumentation-aiokafka already injects, and its setter appends unconditionally – a second injector would put two traceparent headers on the wire.

  • Faust’s older https://pypi.org/project/opentracing/ support (faust.sensors.distributed_tracing.TracingSensor) does inject into Kafka headers. Do not run both; setup_opentelemetry() warns if it sees one already registered.

class faust.contrib.opentelemetry.OpenTelemetrySensor(*, tracer: Optional[Any] = None, **kwargs: Any)[source]

Open an OpenTelemetry span around each event a stream processes.

The span is parented to the producer’s span via the traceparent header on the Kafka message, and stays current for the duration of processing.

property tracer: Any

Tracer used to create spans (resolved lazily). :rtype: Any

on_stream_event_in(tp: TP, offset: int, stream: StreamT, event: EventT) → Optional[Dict][source]

Start a process span and make it current.

Return type:

_UnionGenericAlias[_SpecialGenericAlias, None]

on_stream_event_out(tp: TP, offset: int, stream: StreamT, event: EventT, state: Optional[Dict[Any, Any]] = None) → None[source]

Detach the context and end the span.

Return type:

None

logger: logging.Logger = <Logger faust.contrib.opentelemetry (WARNING)>
faust.contrib.opentelemetry.instrument_asgi_app(asgi_app: Any, *, tracer_provider: Any = None, force: bool = False) → bool[source]

Attach OpenTelemetry instrumentation to a FastAPI/Starlette app.

Returns True if instrumentation was attached.

Does nothing – and returns False – when the instrumentation package is missing, when no SDK has been configured (unless force), or when the application is already instrumented (for example because it was started under opentelemetry-instrument).

Return type:

bool

faust.contrib.opentelemetry.opentelemetry_available() → bool[source]

Return True if the OpenTelemetry API is importable.

Return type:

bool

faust.contrib.opentelemetry.sdk_is_configured() → bool[source]

Return True if a real TracerProvider is installed.

The OpenTelemetry API ships a proxy/no-op provider until an application calls set_tracer_provider(). Treating that as “tracing is off” is what lets this integration be enabled by default without doing anything behind an operator’s back.

Return type:

bool

faust.contrib.opentelemetry.setup_opentelemetry(app: AppT, *, tracer: Optional[Any] = None) → Optional[OpenTelemetrySensor][source]

Register the OpenTelemetry sensor on app.

Returns the sensor, or None if OpenTelemetry is not installed.

Return type:

_UnionGenericAlias[OpenTelemetrySensor, None]