feat: add Faust opentelemetry instrumentation#4829
Conversation
Instrument faust-streaming applications using faust's sensor API: - A producer span is recorded for every message sent to a Kafka topic and the span context is injected into the message headers. - A consumer span is recorded around every event processed by a stream (agent), continuing the trace extracted from the incoming message headers, and made current while the event is processed. Apps created after FaustInstrumentor().instrument() are instrumented automatically by wrapping faust.App.__init__; existing apps can be instrumented with instrument_app()/uninstrument_app(). Assisted-by: Claude Fable 5 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NoKAa2xLqorEdDQKh77E8Q
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NoKAa2xLqorEdDQKh77E8Q
One or more co-authors of this pull request were not found. You must specify co-authors in commit message trailer via: Supported
Alternatively, if the co-author should not be included, remove the Please update your commit message(s) by doing |
|
I’ll be back soon to test this further! |
|
@wbarnha since you seem to be the upstream as well, why not adding native opentelemetry support to faust itself? |
|
My motivation for this PR stems from dependency management. The otel packages this needs are 0.x releases that are version-locked to each other and to specific api/sdk versions, and they break between releases (the messaging attributes are still incubating and keep getting renamed). If faust depended on them directly, every otel release train would couple to a faust release, and anyone pinning a different otel version than faust would hit resolver conflicts. Faust's dependency graph is heavy enough as it is. I get plenty of install-conflict issues without adding that. The most faust could safely carry natively is an extra depending only on opentelemetry-api, but then I'd be hardcoding semconv attribute names with nobody checking them against the spec. Keeping it in contrib means the version coupling lives in the repo whose release process is actually built to manage it, and faust ships zero new dependencies. I'm going to tinker with faust to see if I'm wrong about this. |
Description
Adds a new instrumentation package,
opentelemetry-instrumentation-faust, forfaust-streaming (Kafka stream processing).
The instrumentation is built on faust's Sensor API rather than patching library
internals (both the pure-Python and compiled Cython stream paths invoke sensors):
on_send_initiated/completed/errorrecord a{topic} sendPRODUCER span for everymessage published to Kafka and inject the trace context into the message headers.
on_stream_event_in/outrecord a{topic} processCONSUMER span around each eventprocessed by a stream/agent, parented from the context extracted from the incoming
message headers. The hooks run in the agent's task context, so the process span is
current inside the agent body and sends made while processing nest under it.
faust.App.__init__, used to auto-register the sensor on newapps;
FaustInstrumentor.instrument_app()/uninstrument_app()handle pre-existingapps. Optional
produce_hook/process_hookcallbacks are supported.Known gaps/concerns:
includelist: faust/mode are effectively untypedand cascade
Unknowntypes that can't be fixed without pervasive casts.send_soon()(buffered, flushed from the producer service task)may not parent under the caller's span since publishing happens in another task.
Type of change
How Has This Been Tested?
tox -e py311-test-instrumentation-faust— 11 tests covering: instrument/uninstrumentAPI, sensor registration, producer span attributes + header injection (validated
against semconv attribute names and value types), partition attribute, send error
status + exception event, end-to-end stream processing through real faust
channels/streams (exercises the compiled Cython iterator), trace continuation from
producer to process to nested send, hooks, and no-spans-after-uninstrument.
tox -e lint-instrumentation-faust(pylint 10/10), ruff,tox -e typecheck, docsbuild for the new page, license header/README/codespell checks.
Does This PR Require a Core Repo Change?
Checklist: