FastAPI co-hosting, OpenTelemetry tracing, and per-endpoint web feature flags

[wbarnha]

Description

Three related pieces of work on the web and observability layers:

1. Run a FastAPI (or any ASGI) application in the same process and event loop as a Faust worker.
2. Get a single distributed trace across the HTTP request, Kafka, and the agent that consumes the message.
3. Control each built-in web endpoint individually, and expose worker performance metrics as JSON.

Fixes #448
Fixes #435
Fixes #322

Reviewable commit by commit, in order.

---

1. The event-loop fix

This is the bug behind all three issues, and it is not really a FastAPI bug at all.

Faust apps are declared at module scope, but under uvicorn the app is started from a loop created later by asyncio.run(). Several call sites resolved app.loop at declaration time, pinning the App to whichever loop mode's get_event_loop() found at import — a loop that is never run. Everything built during startup then inherited that dead loop:

AssertionError: Please create objects with the same loop as running with
RuntimeError: Task ... got Future ... attached to a different loop

faust worker survived by accident: faust/cli/base.py reused the same policy loop mode had already grabbed, so the pinned loop happened to be the right one. asyncio.run() creates a different one, so the pin became fatal.

The fix is to stop resolving the loop eagerly and let mode.Service late-bind it inside _default_start() — i.e. in the loop that actually runs the app:

• Agent, Collection, Web and livecheck Case no longer pass loop=app.loop to Service.__init__.
• App._new_transport() / _new_producer_transport() no longer pass loop=, and Transport.loop resolves lazily (still settable; an explicit loop= still wins).
• Transport.create_conductor() no longer passes loop=.
• App.tables no longer passes loop= to the table manager.

Worth a reviewer's eye: the transport chain (@app.agent → app.topics → _new_conductor() → app.transport) is the load-bearing part. Fixing only the four Service.__init__ sites leaves app._loop pinned — found by measuring, not reading.

A consequence caught by CI: nothing calls set_event_loop() before the CLI runs any more, and Python 3.14 removed the get-or-create fallback from get_event_loop_policy().get_event_loop(). faust/cli/base.py now routes through mode.utils.loops.get_event_loop(), which keeps get-or-create across versions. Only the 3.14 legs caught this; 3.10–3.13 still auto-create.

2. faust.contrib.fastapi

The ASGI server drives:

api = FastAPI(lifespan=faust_lifespan(faust_app))

@api.post("/greet")
async def greet(text: str):
    await greetings.send(value=text)
    return {"ok": True}

or the worker drives, serving your ASGI app on its own loop:

serve_asgi(faust_app, api, port=8000)

Details that matter:

• bind_to_running_loop() reads app._loop, not app.loop — the public property resolves and caches a loop as a side effect of being read, which is exactly what it is trying to detect. Raises LoopMismatch naming the usual causes.
• faust_app_running() uses maybe_start(), so it composes with an already-running app and will not stop one it did not start.
• faust_lifespan() yields None: Starlette merges a non-None lifespan value into the ASGI scope as state.
• serve_asgi() registers via the public App.service() decorator, so the server starts after table recovery — when it is actually safe to serve.
• Imports nothing from fastapi/starlette, so it works with Starlette, Quart and Litestar too. uvicorn is imported lazily.

Also fixes faust[aerospike], which shipped requirements/extras/aerospike.txt without the matching BUNDLES entry and so installed nothing despite being advertised in the README. tests/unit/test_packaging.py guards both directions of that mapping.

3. OpenTelemetry

opentelemetry-instrumentation-aiokafka already wraps AIOKafkaProducer.send and AIOKafkaConsumer.getmany, so most of a trace already works. The hop nobody outside Faust can supply is receive → process: the consumer runs in its own thread (ConsumerThread) and contextvars never cross threads, so the receive span is opened and closed on a thread the agent never runs on. The result is an orphaned span and an unparented agent, which reads worse than no instrumentation.

OpenTelemetrySensor closes it — extracting trace context from the message headers and opening a {topic} process span that stays current while the stream processes the event:

from faust.contrib.opentelemetry import setup_opentelemetry
setup_opentelemetry(app)

FastAPI apps are instrumented automatically, but only when OpenTelemetry is installed and a real TracerProvider is configured — until an SDK is configured the API is a no-op, so this never enables telemetry nobody asked for.

• Depends on opentelemetry-api only, and never calls set_tracer_provider() — configuring the SDK is the application's job.
• Reads trace context from headers, never writes it. The aiokafka instrumentation already injects on produce and its setter appends unconditionally, so a second injector would put two traceparent headers on the wire.
• Warns when the opentracing TracingSensor is registered, since that one does inject.
• Every OpenTelemetry call is wrapped — telemetry must never break message processing.

4. Per-endpoint web feature flags

Which built-in endpoints get served was decided entirely by debug: it enabled the statistics and graph endpoints together or neither, and /router and /table were mounted unconditionally with no way to turn them off — even though /table serves table data over HTTP and /router publishes the URL of every other worker.

web_stats_enabled     default_alias="debug"   /  and  /assignment/
web_graph_enabled     default_alias="debug"   /graph
web_router_enabled    default=True            /router
web_tables_enabled    default=True            /table
web_metrics_enabled   default=False           /performance/

default_alias="debug" resolves to conf.debug at read time when unset, so an app setting none of these serves exactly what it served before. /router and /table keep defaulting to on because disabling /router breaks @table_route across nodes — they are simply switchable now.

Selection moved into Web._enabled_blueprints(), driven by a blueprint_flags mapping. Blueprints with no entry are always enabled, so anything a subclass or user adds is unaffected.

5. /performance/ metrics endpoint

Off by default, no new dependency, independent of debug. Not a flat Monitor.asdict() dump — two things are computed that monitor does not provide:

• Consumer lag, from log end offsets minus offsets actually read. Monitor tracks both but never subtracts them. Partitions with an unknown read offset are skipped rather than reported as fully lagged, which would spike alerts on every restart.
• Latency percentiles. Monitor keeps raw deques of thousands of samples. Nearest-rank rather than statistics.quantiles, which raises on fewer than two samples — a metrics endpoint should not fail on a worker that just started.

New extras

pip install "faust-streaming[fastapi]"          # fastapi + uvicorn
pip install "faust-streaming[opentelemetry]"    # api + fastapi/aiokafka instrumentation

requirements/test.txt gains only opentelemetry-api, -sdk and uvicorn — all pure Python — so the new code is covered in CI without pulling fastapi/pydantic-core into all 15 matrix legs. FastAPI-dependent tests use importorskip; the OpenTelemetry instrumentation paths are covered by stubbing the module through sys.modules.

Backwards compatibility

Audited by snapshotting seven production-shaped app configs (plain, debug, web_enabled=False, web_in_thread, custom port/host, prod-like, debug+in_thread) on master and on this branch, then diffing settings, route tables, reverse route names, agents, tables, topics and loop identity.

The only difference is loop identity — False on master, True here — which is the bug being fixed. Settings, routes and registered objects are identical for every config. install_requires is unchanged, faust/web/base.py's __all__ is unchanged, and the set of shipped faust.* packages is byte-identical to master. tests/unit/web/test_blueprint_flags.py::Test_backwards_compatibility pins the served-blueprint set for the default and debug=True cases so this cannot silently regress.

That audit also caught one genuine production-visible change and fixed it: setup.py excluded "examples" but not "examples.*", so find_packages was installing the example sub-packages into site-packages under an examples namespace. Renaming examples/fastapi to examples/fastapi_project therefore changed what the wheel installs. Excluding "examples.*" means no example package ships either way.

Testing

2320 passed, 8 skipped locally. isort, black, flake8 and bandit clean; Sphinx builds.

New tests bind no socket and start no uvicorn, so they stay fast and keep the autouse lingering-thread/task guards happy:

• tests/functional/test_fastapi_cohost.py reproduces the import-time-vs-running-loop split without a broker, by declaring the app in a synchronous fixture and asserting from async tests. Verified these fail when the old code is restored.
• tests/unit/contrib/test_opentelemetry.py asserts against a real InMemorySpanExporter that the process span continues the trace from a traceparent header and that nested spans parent to it.
• tests/unit/web/test_blueprint_flags.py covers the flag matrix and the back-compat cases.
• tests/functional/web/test_metrics.py drives /performance/ through the existing aiohttp test client.

Verified against a real broker

Both directions were run end to end against Kafka 4.0 (KRaft, single node):

faust worker + serve_asgi
  POST -> await topic.send() on the worker's loop -> 200
  agent in the same process consumed the message
  real SIGINT to the process group -> exit 0, 1.4s clean shutdown

uvicorn + faust_lifespan   (the literal #448 scenario)
  POST -> await topic.send() -> 200, zero loop errors
  agent consumed it, including the backlog from a previous run

Signal handling

Checked against real SIGINTs — a mode.Worker hosting AsgiService with uvicorn serving on a real port, signalled at the process group like a terminal Ctrl-C, single and repeated, with and without the override. All four combinations exit 0 and complete a slow service's drain.

That measurement corrected the code comment: mode.Worker registers via loop.add_signal_handler(), which is delivered through asyncio's wakeup fd, so uvicorn's signal.signal() does not displace it. The override is kept to avoid a second concurrent shutdown path, and the docstring now says that rather than something untrue.

Unrelated issues surfaced, not fixed here

• extra/tools/render_configuration_reference.py iterates related_cli_options string values character by character, so make configref rewrites :option:faust --debug`` as :option:faust -`, :option:`faust d`, ...` and drops hand-written prose. The new settings are hand-added to `settingref.txt` in the generator's format instead.
• make configcheck fails on master because web_application_options was never added to settingref.txt. Documented here so the symmetric difference is empty, but configcheck is a tox environment CI never invokes, which is why it drifted.
• The PyPy leg failed on a hypothesis FlakyFailure in the partition-assignor property test (5123ms against a 4000ms deadline on one run, 3455ms on the next). Pre-existing and unrelated; the deadline is raised on PyPy only, so CPython keeps the tighter bound.
• An agent will not consume from a topic that does not exist when the worker starts — messages are written but unprocessed until a restart. Confirmed to be general Faust behaviour (faust worker reproduces it identically), so it is documented in the userguide caveats rather than worked around.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 96.07%. Comparing base (4acc180) to head (0cc470f).

Additional details and impacted files

@@            Coverage Diff             @@
##           master     #748      +/-   ##
==========================================
+ Coverage   95.97%   96.07%   +0.10%     
==========================================
  Files         103      105       +2     
  Lines       11072    11326     +254     
  Branches     1191     1226      +35     
==========================================
+ Hits        10626    10882     +256     
+ Misses        352      350       -2     
  Partials       94       94              

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.