diff --git a/CHANGELOG.md b/CHANGELOG.md index 18b85cc5e..6f7706cf8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,48 @@ https://github.com/faust-streaming/faust/releases. The v0.12.0 entry below resumes the Keep a Changelog format. --> +## [Unreleased] + +### Added +- Per-endpoint feature flags for the built-in web endpoints: + `web_stats_enabled`, `web_graph_enabled`, `web_router_enabled`, + `web_tables_enabled` and `web_metrics_enabled`. Previously `debug` was the + only control, and it enabled the statistics and graph endpoints together, + while `/router` and `/table` could not be turned off at all — even though + `/table` serves table *data* over HTTP. The statistics and graph flags take + their default from `debug`, so behaviour is unchanged unless you set them. +- New `/performance/` endpoint (`web_metrics_enabled`, off by default) + returning throughput, latency percentiles, consumer lag and table statistics + as JSON. Consumer lag and latency percentiles are computed here — `Monitor` + tracks read and log-end offsets but never derives lag, and keeps raw latency + deques rather than summaries. Needs no extra dependency, and is independent + of both `debug` and `faust.sensors.prometheus`. +- `faust.contrib.fastapi`: co-host a FastAPI (or any ASGI) application with the + worker, in one process and one event loop. `faust_lifespan()` runs Faust from + an ASGI lifespan, `serve_asgi()` serves your app from inside `faust worker`. + New `faust[fastapi]` extra. +- `faust.contrib.opentelemetry`: OpenTelemetry tracing. `setup_opentelemetry()` + continues a trace from Kafka message headers into your agents — the hop + `opentelemetry-instrumentation-aiokafka` cannot bridge, because Faust's + consumer runs in its own thread. FastAPI apps are instrumented automatically + when an SDK is configured. New `faust[opentelemetry]` extra. +- New userguide page: *FastAPI and other ASGI applications*. + +### Fixed +- Faust apps no longer resolve an event loop when agents, tables or the + transport are declared at import time. Previously that pinned the app to a + loop that was never run, so starting it from `asyncio.run()` — as uvicorn + does — failed with "Please create objects with the same loop as running with" + or "Task ... got Future ... attached to a different loop" (#322, #435, #448). +- `faust[aerospike]` installed nothing: `requirements/extras/aerospike.txt` + shipped without the matching `BUNDLES` entry in `setup.py`, despite being + advertised in the README. A new test guards both directions of that mapping. + +### Changed +- The `examples/fastapi/` directory is now `examples/fastapi_project/`. The old + name shadowed the real `fastapi` package when running the sibling + `examples/fastapi_example.py`, so neither example could be run as documented. + ## [v0.12.1](https://github.com/faust-streaming/faust/releases/tag/v0.12.1) - 2026-07-19 [Compare with v0.12.0](https://github.com/faust-streaming/faust/compare/v0.12.0...v0.12.1) diff --git a/docs/includes/installation.txt b/docs/includes/installation.txt index a44f485d6..00a32ae82 100644 --- a/docs/includes/installation.txt +++ b/docs/includes/installation.txt @@ -84,6 +84,23 @@ Sensors :``faust[sentry]``: for reporting worker errors to Sentry via :pypi:`sentry-sdk`. +:``faust[opentracing]``: + for distributed tracing via :pypi:`opentracing`. Deprecated upstream in + March 2026; prefer ``faust[opentelemetry]`` for new work. + +:``faust[opentelemetry]``: + for distributed tracing via :pypi:`opentelemetry-api`, including + continuing a trace from Kafka message headers into your agents. See + :ref:`guide-fastapi`. + +Web +~~~ + +:``faust[fastapi]``: + for co-hosting a FastAPI (or any ASGI) application in the same process + and event loop as the Faust worker, via :pypi:`uvicorn`. + See :ref:`guide-fastapi`. + Event Loops ~~~~~~~~~~~ diff --git a/docs/includes/settingref.txt b/docs/includes/settingref.txt index 669894cff..8467a3df0 100644 --- a/docs/includes/settingref.txt +++ b/docs/includes/settingref.txt @@ -1904,6 +1904,39 @@ Advanced Web Server Settings Web server driver to use. +.. setting:: web_application_options + +``web_application_options`` +--------------------------- + +.. versionadded:: 0.11.4 + +:type: :class:`~typing.Mapping` [ :class:`str`, :class:`~typing.Any` ] +:default: :const:`None` + +Extra keyword arguments passed to the web framework's application. + +Use this to configure the underlying web application object that the +web driver creates. For the default :pypi:`aiohttp` driver these are +forwarded straight to :class:`aiohttp.web.Application`, so you can set +things like ``client_max_size`` or install middlewares: + +.. sourcecode:: python + + from aiohttp.web import middleware + + @middleware + async def error_middleware(request, handler): + ... + + app = App(..., web_application_options={ + 'client_max_size': 1024 ** 2 * 20, + 'middlewares': [error_middleware], + }) + +The accepted keys depend on the configured web driver. + + .. setting:: web_bind ``web_bind`` @@ -1984,6 +2017,132 @@ Enable web server and other web components. This option can also be set using :option:`faust worker --without-web`. +.. setting:: web_graph_enabled + +``web_graph_enabled`` +--------------------- + +.. versionadded:: 0.13 + +:type: :class:`bool` +:default (alias to setting): :setting:`debug` +:environment: :envvar:`APP_WEB_GRAPH_ENABLED` + +Enable/disable the ``/graph`` dependency graph endpoint. + +Renders the worker's service dependency graph as a PNG. + +If not set, this follows :setting:`debug`, which is how this endpoint +was gated before this setting existed. + +.. warning:: + + The graph describes the entire internal service tree of the + worker. + + +.. setting:: web_metrics_enabled + +``web_metrics_enabled`` +----------------------- + +.. versionadded:: 0.13 + +:type: :class:`bool` +:default: :const:`False` +:environment: :envvar:`APP_WEB_METRICS_ENABLED` + +Enable/disable the ``/performance/`` metrics endpoint. + +Serves throughput, latency, consumer lag and table statistics as +JSON, gathered from :setting:`Monitor`. Unlike the statistics +endpoints this is independent of :setting:`debug`, so it can be left +on in production. + +Disabled by default: it is a new endpoint, and enabling it should be +a deliberate choice. + +.. seealso:: + + :mod:`faust.sensors.prometheus` serves the same underlying data + in Prometheus format on its own ``/metrics`` path. + + +.. setting:: web_router_enabled + +``web_router_enabled`` +---------------------- + +.. versionadded:: 0.13 + +:type: :class:`bool` +:default: :const:`True` +:environment: :envvar:`APP_WEB_ROUTER_ENABLED` + +Enable/disable the ``/router`` endpoints. + +These report which worker in the cluster owns a given table key, and +are what makes :meth:`@table_route` work across nodes. + +.. warning:: + + Disabling this breaks :meth:`@table_route` for multi-node + deployments. Only turn it off if you route entirely within a + single worker. + + Left enabled, it exposes your cluster topology -- the URLs of + every other worker. + + +.. setting:: web_stats_enabled + +``web_stats_enabled`` +--------------------- + +.. versionadded:: 0.13 + +:type: :class:`bool` +:default (alias to setting): :setting:`debug` +:environment: :envvar:`APP_WEB_STATS_ENABLED` + +Enable/disable the built-in statistics endpoints. + +Serves sensor statistics at ``/`` and the current partition +assignment at ``/assignment/``. + +If not set, this follows :setting:`debug`, which is how these +endpoints were gated before this setting existed. + +When disabled, ``/`` serves the plain production index instead. + +.. warning:: + + These endpoints expose internal state: every registered sensor's + counters, and which partitions this worker is handling. + + +.. setting:: web_tables_enabled + +``web_tables_enabled`` +---------------------- + +.. versionadded:: 0.13 + +:type: :class:`bool` +:default: :const:`True` +:environment: :envvar:`APP_WEB_TABLES_ENABLED` + +Enable/disable the ``/table`` endpoints. + +These list the tables defined by this app and allow reading +individual keys over HTTP. + +.. warning:: + + This exposes table *data*, not just table names. If your tables + hold anything sensitive, turn this off. + + .. setting:: web_host ``web_host`` diff --git a/docs/reference/faust.contrib.fastapi.rst b/docs/reference/faust.contrib.fastapi.rst new file mode 100644 index 000000000..4d963f8ca --- /dev/null +++ b/docs/reference/faust.contrib.fastapi.rst @@ -0,0 +1,11 @@ +===================================================== + ``faust.contrib.fastapi`` +===================================================== + +.. contents:: + :local: +.. currentmodule:: faust.contrib.fastapi + +.. automodule:: faust.contrib.fastapi + :members: + :undoc-members: diff --git a/docs/reference/faust.contrib.opentelemetry.rst b/docs/reference/faust.contrib.opentelemetry.rst new file mode 100644 index 000000000..e440f3460 --- /dev/null +++ b/docs/reference/faust.contrib.opentelemetry.rst @@ -0,0 +1,11 @@ +===================================================== + ``faust.contrib.opentelemetry`` +===================================================== + +.. contents:: + :local: +.. currentmodule:: faust.contrib.opentelemetry + +.. automodule:: faust.contrib.opentelemetry + :members: + :undoc-members: diff --git a/docs/reference/faust.web.apps.metrics.rst b/docs/reference/faust.web.apps.metrics.rst new file mode 100644 index 000000000..c07b7667b --- /dev/null +++ b/docs/reference/faust.web.apps.metrics.rst @@ -0,0 +1,11 @@ +===================================================== + ``faust.web.apps.metrics`` +===================================================== + +.. contents:: + :local: +.. currentmodule:: faust.web.apps.metrics + +.. automodule:: faust.web.apps.metrics + :members: + :undoc-members: diff --git a/docs/reference/index.rst b/docs/reference/index.rst index 979616bda..d54cd5387 100644 --- a/docs/reference/index.rst +++ b/docs/reference/index.rst @@ -54,6 +54,8 @@ Contrib :maxdepth: 1 faust.contrib + faust.contrib.fastapi + faust.contrib.opentelemetry faust.contrib.sentry Fixups @@ -241,6 +243,7 @@ Web :maxdepth: 1 faust.web.apps.graph + faust.web.apps.metrics faust.web.apps.router faust.web.apps.stats faust.web.base diff --git a/docs/userguide/fastapi.rst b/docs/userguide/fastapi.rst new file mode 100644 index 000000000..fc223bb42 --- /dev/null +++ b/docs/userguide/fastapi.rst @@ -0,0 +1,265 @@ +.. _guide-fastapi: + +================================================= + FastAPI and other ASGI applications +================================================= + +.. module:: faust.contrib.fastapi + +.. contents:: + :local: + :depth: 2 + +.. _fastapi-basics: + +Basics +====== + +Faust and an ASGI server can share a single process, so an HTTP endpoint can +produce to a Kafka topic directly: + +.. sourcecode:: python + + import faust + from fastapi import FastAPI + from faust.contrib.fastapi import faust_lifespan + + faust_app = faust.App("hello", broker="kafka://localhost:9092") + greetings = faust_app.topic("greetings", value_type=str) + + api = FastAPI(lifespan=faust_lifespan(faust_app)) + + @faust_app.agent(greetings) + async def print_greetings(stream): + async for greeting in stream: + print(greeting) + + @api.post("/greet") + async def greet(text: str): + await greetings.send(value=text) + return {"ok": True} + +Run it with: + +.. sourcecode:: console + + $ uvicorn myapp:api + +Install the optional dependencies with: + +.. sourcecode:: console + + $ pip install "faust-streaming[fastapi]" + +Nothing in :mod:`faust.contrib.fastapi` imports :pypi:`fastapi` or +:pypi:`starlette`, so it works equally well with Starlette, Quart or Litestar. + +.. _fastapi-one-loop: + +One process, one event loop +=========================== + +This is the whole reason the module exists. + +A Faust app resolves its event loop the first time something asks for it, and +everything it builds afterwards -- producers, consumers, timers, table +managers -- belongs to that loop. An ASGI server such as :pypi:`uvicorn` +creates its *own* loop with :func:`asyncio.run`. If the app has already +resolved a different loop by then, every one of those objects is attached to a +loop that will never run, and you get:: + + AssertionError: Please create objects with the same loop as running with + RuntimeError: Task ... got Future ... attached to a different loop + +:func:`faust_lifespan` and :func:`faust_app_running` bind the app to the loop +that is actually running before starting it, so this cannot happen. + +.. admonition:: Do not touch ``app.loop`` at import time + + Declare the app, its topics, agents and tables at module scope as usual -- + that is supported and is what the examples do. What you must *not* do is + ask the app for its event loop before the server starts. In practice that + means avoiding, at module scope: + + * ``app.loop`` + * ``app.web``, ``app.transport``, ``app.producer`` + * passing your own loop to ``faust.App(loop=...)`` + + If the app is already bound to another loop, + :func:`bind_to_running_loop` raises :exc:`LoopMismatch` with a message + naming the likely cause, rather than letting it fail later and less + legibly. + +.. _fastapi-composing: + +Composing with your own lifespan +================================ + +When you have setup of your own, use :func:`faust_app_running` -- it is the +context manager :func:`faust_lifespan` is built from: + +.. sourcecode:: python + + from contextlib import asynccontextmanager + from faust.contrib.fastapi import faust_app_running + + @asynccontextmanager + async def lifespan(api: FastAPI): + async with faust_app_running(faust_app): + ml_models["answer"] = load_model() + yield + ml_models.clear() + + api = FastAPI(lifespan=lifespan) + +The app is started with ``maybe_start()``, so this composes safely with an app +that is already running, and will not stop one it did not start. + +.. _fastapi-worker: + +Running under ``faust worker`` +============================== + +The other direction: keep ``faust worker`` as your entry point and let it serve +your ASGI application on its own loop. + +.. sourcecode:: python + + from faust.contrib.fastapi import serve_asgi + + api = FastAPI() + serve_asgi(faust_app, api, port=8000) + +.. sourcecode:: console + + $ faust -A myapp worker -l info + +:func:`serve_asgi` registers the server as an app service, so it starts only +once the app is up -- after table recovery has finished, which is when it is +actually safe to serve traffic. Extra keyword arguments are passed through to +:class:`uvicorn.Config`. + +.. admonition:: ``faust -A`` looks for ``app`` + + The ``-A`` option imports the module and looks for an attribute named + ``app``. Bind that name to the **Faust** app, not to the ``FastAPI`` + object:: + + app = faust_app = faust.App("myapp", ...) + api = FastAPI() + + Naming the FastAPI object ``app`` -- the FastAPI convention -- is what + makes ``faust -A`` fail to find your app. + +.. _fastapi-web-server: + +Faust's own web server +====================== + +Faust ships its own :pypi:`aiohttp` server for ``@app.page``, table routing and +``/metrics``. It is independent of anything here and keeps listening on +``web_port`` (6066 by default). Two servers in one process is fine; if you do +not want Faust's, turn it off: + +.. sourcecode:: python + + app = faust.App("myapp", web_enabled=False) + +There is no ASGI *driver* for Faust's own views -- ``@app.page`` and +``@app.table_route`` are still served by aiohttp. This page is about +co-hosting your application, not about replacing that. + +.. _fastapi-opentelemetry: + +OpenTelemetry +============= + +.. sourcecode:: console + + $ pip install "faust-streaming[opentelemetry]" + +With the instrumentation packages installed, most of a distributed trace works +already: :pypi:`opentelemetry-instrumentation-aiokafka` wraps the same +``AIOKafkaProducer.send`` and ``AIOKafkaConsumer.getmany`` calls that Faust's +driver uses, so an HTTP request that produces to Kafka carries its trace +context onto the wire in a ``traceparent`` header. + +Two things Faust adds: + +**Your FastAPI application is instrumented automatically.** +:func:`faust_lifespan` and :func:`serve_asgi` attach +``FastAPIInstrumentor`` when OpenTelemetry is installed *and* the application +has configured a real ``TracerProvider``. Until an SDK is configured the +OpenTelemetry API is a no-op, so nothing is enabled behind your back. Pass +``opentelemetry=False`` to opt out, or ``True`` to force it. Applications +already instrumented (for example under ``opentelemetry-instrument``) are left +alone rather than double-wrapped. + +**The consumer-to-agent hop is bridged.** Faust's consumer runs in its own +thread, and :mod:`contextvars` do not cross threads -- so the ``receive`` span +created inside ``getmany`` is closed before your agent ever runs, leaving the +agent's work unparented. Register the sensor to close that gap: + +.. sourcecode:: python + + from faust.contrib.opentelemetry import setup_opentelemetry + + setup_opentelemetry(app) + +It extracts the trace context from each message's headers and opens a +``{topic} process`` span that stays current for as long as the stream is +processing the event, so everything the agent does nests underneath it. The +resulting trace reads: + +.. sourcecode:: text + + FastAPI server span + └─ {topic} send (PRODUCER, from opentelemetry-instrumentation-aiokafka) + └─ {topic} process (CONSUMER, from faust.contrib.opentelemetry) + └─ ...whatever your agent does + +.. admonition:: Do not run this alongside the opentracing sensor + + :class:`faust.sensors.distributed_tracing.TracingSensor` also injects trace + headers into Kafka messages. Running both puts two ``traceparent`` headers + on the wire. :func:`setup_opentelemetry` warns if it sees one registered. + Note also that the OpenTracing bridge was deprecated upstream in March + 2026; new work should target OpenTelemetry directly. + +.. _fastapi-examples: + +Examples +======== + +Two complete examples ship with Faust: + +``examples/fastapi_example.py`` + A single file -- ``hello_world.py`` plus a FastAPI application. + +``examples/fastapi_project/`` + The same thing as a package, with agents, tables, timers and routers in + their own modules. ``main.py`` is served by uvicorn; ``worker_main.py`` + shows the same API served from inside ``faust worker``. + +.. _fastapi-caveats: + +Caveats +======= + +* **A topic must exist when the worker starts for its agent to consume from + it.** This bites co-hosted apps particularly often, because they typically + produce to and consume from the same topic: on a first run against a fresh + cluster the topic does not exist yet, the agent subscribes to nothing, and + messages produced by your endpoints are written but never processed until + the process is restarted. Nothing is lost -- the backlog is picked up on + the next start -- but the first run looks like the agent is broken. + + This is not specific to co-hosting; ``faust worker`` behaves the same way. + Create your topics ahead of time (or restart once) when bootstrapping a new + environment. + +* ``producer_threaded=True`` spawns a producer with its own thread and loop. + It is untested in a co-hosted process and is not supported here yet. +* ``uvicorn --reload`` and ``--workers`` fork or re-exec the process. Faust + is not designed to be forked; use a single worker. +* Faust's own web views stay on aiohttp, as described above. diff --git a/docs/userguide/index.rst b/docs/userguide/index.rst index c7737aa79..bbb00cbe6 100644 --- a/docs/userguide/index.rst +++ b/docs/userguide/index.rst @@ -26,3 +26,4 @@ kafka debugging workers + fastapi diff --git a/examples/fastapi/main.py b/examples/fastapi/main.py deleted file mode 100644 index 82e3c6b2c..000000000 --- a/examples/fastapi/main.py +++ /dev/null @@ -1,42 +0,0 @@ -from contextlib import asynccontextmanager -from fastapi import FastAPI -from api import router as api_router - -from my_faust.timer import router as timer_router -from my_faust.app import faust_app - - -# This is just hello_world.py integrated with a FastAPI application - - -def fake_answer_to_everything_ml_model(x: float): - return x * 42 - - -ml_models = {} - - -@asynccontextmanager -async def lifespan(app: FastAPI): - faust_app.discover() - await faust_app.start() - yield - await faust_app.stop() - - -# You MUST have "app" defined in order for Faust to discover the app -# if you're using "faust" on CLI, but this doesn't work yet -app = fastapi_app = FastAPI( - lifespan=lifespan, -) - -# For now, run via "uvicorn fastapi_example:app" -# then visit http://127.0.0.1:8000/docs - -app.include_router(router=api_router) -app.include_router(router=timer_router) - - -@app.get("/") -def read_root(): - return {"Hello": "World"} diff --git a/examples/fastapi/my_faust/app.py b/examples/fastapi/my_faust/app.py deleted file mode 100644 index 8a5b5f79b..000000000 --- a/examples/fastapi/my_faust/app.py +++ /dev/null @@ -1,18 +0,0 @@ -import faust - - -def get_all_packages_to_scan(): - return ["my_faust"] - - -# You MUST have "app" defined in order for Faust to discover the app -# if you're using "faust" on CLI, but this doesn't work yet -# autodiscover https://faust-streaming.github.io/faust/userguide/settings.html#autodiscover -app = faust_app = faust.App( - 'hello-world-fastapi', - broker='kafka://localhost:9092', - web_enabled=False, - autodiscover=get_all_packages_to_scan, -) - -# For now, run via "faust -A my_faust.app worker -l info" diff --git a/examples/fastapi/my_faust/timer/my_timer.py b/examples/fastapi/my_faust/timer/my_timer.py deleted file mode 100644 index 57261aafd..000000000 --- a/examples/fastapi/my_faust/timer/my_timer.py +++ /dev/null @@ -1,14 +0,0 @@ -from uuid import uuid4 -from fastapi import APIRouter - -from my_faust.app import faust_app -from my_faust.topic.my_topic import greetings_topic - -router = APIRouter() - - -@faust_app.timer(5) # make sure you *always* add the timer above if you're using one -@router.get("/produce") -async def produce(): - await greetings_topic.send(value=uuid4().hex) - return {"success": True} diff --git a/examples/fastapi_example.py b/examples/fastapi_example.py index 3e666fcda..85ab08ad8 100755 --- a/examples/fastapi_example.py +++ b/examples/fastapi_example.py @@ -1,14 +1,27 @@ #!/usr/bin/env python -import asyncio +"""hello_world.py, co-hosted with a FastAPI application. + +Faust and the web server share one process and one event loop, so an endpoint +can produce to a topic directly:: + + $ uvicorn fastapi_example:api --reload + + # ...then visit http://127.0.0.1:8000/docs + +The same file also works as a plain worker, because ``app`` is the Faust app:: + + $ faust -A fastapi_example worker -l info + +Requires ``pip install "faust-streaming[fastapi]"``. +""" + from contextlib import asynccontextmanager from typing import Union from fastapi import FastAPI import faust - - -# This is just hello_world.py integrated with a FastAPI application +from faust.contrib.fastapi import faust_app_running def fake_answer_to_everything_ml_model(x: float): @@ -17,43 +30,38 @@ def fake_answer_to_everything_ml_model(x: float): ml_models = {} - -# You MUST have "app" defined in order for Faust to discover the app -# if you're using "faust" on CLI, but this doesn't work yet -faust_app = faust.App( - 'hello-world-fastapi', - broker='kafka://localhost:9092', +# ``web_enabled=False`` turns off Faust's own aiohttp server, since uvicorn is +# already serving. Leave it on if you also want ``@app.page`` views -- they +# are served separately, on ``web_port`` (6066 by default). +app = faust_app = faust.App( + "hello-world-fastapi", + broker="kafka://localhost:9092", web_enabled=False, ) -# app = faust_app -greetings_topic = faust_app.topic('greetings', value_type=str) +greetings_topic = faust_app.topic("greetings", value_type=str) @asynccontextmanager -async def lifespan(app: FastAPI): - # Load the ML model - ml_models["answer_to_everything"] = fake_answer_to_everything_ml_model - await faust_app.start() - yield - # Clean up the ML models and release the resources - ml_models.clear() - await faust_app.stop() - - -app = fastapi_app = FastAPI( - lifespan=lifespan, -) -# For now, run via "uvicorn fastapi_example:app" -# then visit http://127.0.0.1:8000/docs +async def lifespan(api: FastAPI): + # ``faust_app_running`` binds the app to uvicorn's event loop, starts it, + # and stops it again on shutdown. With no setup of your own to do, use + # ``FastAPI(lifespan=faust_lifespan(faust_app))`` instead. + async with faust_app_running(faust_app): + ml_models["answer_to_everything"] = fake_answer_to_everything_ml_model + yield + ml_models.clear() + + +api = FastAPI(lifespan=lifespan) -@fastapi_app.get("/") +@api.get("/") def read_root(): return {"Hello": "World"} -@fastapi_app.get("/items/{item_id}") +@api.get("/items/{item_id}") def read_item(item_id: int, q: Union[str, None] = None): return {"item_id": item_id, "q": q} @@ -64,9 +72,20 @@ async def print_greetings(greetings): print(greeting) -@faust_app.timer(5) # make sure you *always* add the timer above if you're using one -@fastapi_app.get("/produce") +async def produce_greetings(count: int = 100) -> None: + for i in range(count): + await greetings_topic.send(value=f"hello {i}") + + +@api.get("/produce") async def produce(): - for i in range(100): - await greetings_topic.send(value=f'hello {i}') + await produce_greetings() return {"success": True} + + +# Register the timer separately rather than stacking it on the route. Stacking +# registers the undecorated function as the HTTP route and the timer-wrapped +# one as the timer, which is rarely what people mean. +@faust_app.timer(5) +async def produce_periodically() -> None: + await produce_greetings() diff --git a/examples/fastapi/__init__.py b/examples/fastapi_project/__init__.py similarity index 100% rename from examples/fastapi/__init__.py rename to examples/fastapi_project/__init__.py diff --git a/examples/fastapi/api/__init__.py b/examples/fastapi_project/api/__init__.py similarity index 99% rename from examples/fastapi/api/__init__.py rename to examples/fastapi_project/api/__init__.py index 251300510..414b1664f 100644 --- a/examples/fastapi/api/__init__.py +++ b/examples/fastapi_project/api/__init__.py @@ -1,6 +1,5 @@ -from fastapi import APIRouter - from api.my_api import router as my_api_router +from fastapi import APIRouter router = APIRouter() diff --git a/examples/fastapi/api/my_api.py b/examples/fastapi_project/api/my_api.py similarity index 100% rename from examples/fastapi/api/my_api.py rename to examples/fastapi_project/api/my_api.py index e03859834..159ec720b 100644 --- a/examples/fastapi/api/my_api.py +++ b/examples/fastapi_project/api/my_api.py @@ -1,6 +1,6 @@ from typing import Union -from fastapi import APIRouter +from fastapi import APIRouter from my_faust.table.my_table import greetings_table router = APIRouter() diff --git a/examples/fastapi_project/main.py b/examples/fastapi_project/main.py new file mode 100644 index 000000000..4b1927e87 --- /dev/null +++ b/examples/fastapi_project/main.py @@ -0,0 +1,37 @@ +"""hello_world.py as a package-structured FastAPI + Faust application. + +Serve it with uvicorn -- Faust starts and stops from the ASGI lifespan:: + + $ uvicorn main:api --reload + + # ...then visit http://127.0.0.1:8000/docs + +The Faust worker is a separate entry point, since the app lives in its own +module and is found by autodiscovery:: + + $ faust -A my_faust.app worker -l info + +See ``worker_main.py`` for the other direction: one ``faust worker`` process +serving this same API on its own event loop. + +Requires ``pip install "faust-streaming[fastapi]"``. +""" + +from api import router as api_router +from fastapi import FastAPI +from my_faust.app import faust_app +from my_faust.timer import router as timer_router + +from faust.contrib.fastapi import faust_lifespan + +# ``faust_lifespan`` binds the app to uvicorn's event loop, runs autodiscovery +# (the app sets ``autodiscover``), starts it, and stops it on shutdown. +api = FastAPI(lifespan=faust_lifespan(faust_app)) + +api.include_router(router=api_router) +api.include_router(router=timer_router) + + +@api.get("/") +def read_root(): + return {"Hello": "World"} diff --git a/examples/fastapi/my_faust/__init__.py b/examples/fastapi_project/my_faust/__init__.py similarity index 100% rename from examples/fastapi/my_faust/__init__.py rename to examples/fastapi_project/my_faust/__init__.py diff --git a/examples/fastapi/my_faust/agent/__init__.py b/examples/fastapi_project/my_faust/agent/__init__.py similarity index 100% rename from examples/fastapi/my_faust/agent/__init__.py rename to examples/fastapi_project/my_faust/agent/__init__.py diff --git a/examples/fastapi/my_faust/agent/my_agent.py b/examples/fastapi_project/my_faust/agent/my_agent.py similarity index 100% rename from examples/fastapi/my_faust/agent/my_agent.py rename to examples/fastapi_project/my_faust/agent/my_agent.py diff --git a/examples/fastapi_project/my_faust/app.py b/examples/fastapi_project/my_faust/app.py new file mode 100644 index 000000000..02c71d47f --- /dev/null +++ b/examples/fastapi_project/my_faust/app.py @@ -0,0 +1,20 @@ +import faust + + +def get_all_packages_to_scan(): + return ["my_faust"] + + +# ``faust -A `` looks for an attribute named ``app``, so bind that name +# to the Faust app -- not to the FastAPI object. That is the whole trick. +# autodiscover: +# https://faust-streaming.github.io/faust/userguide/settings.html#autodiscover +app = faust_app = faust.App( + "hello-world-fastapi", + broker="kafka://localhost:9092", + web_enabled=False, + autodiscover=get_all_packages_to_scan, +) + +# Run the worker with "faust -A my_faust.app worker -l info", +# or serve the API with "uvicorn main:api" (see ../main.py). diff --git a/examples/fastapi/my_faust/table/__init__.py b/examples/fastapi_project/my_faust/table/__init__.py similarity index 100% rename from examples/fastapi/my_faust/table/__init__.py rename to examples/fastapi_project/my_faust/table/__init__.py diff --git a/examples/fastapi/my_faust/table/my_table.py b/examples/fastapi_project/my_faust/table/my_table.py similarity index 100% rename from examples/fastapi/my_faust/table/my_table.py rename to examples/fastapi_project/my_faust/table/my_table.py diff --git a/examples/fastapi/my_faust/timer/__init__.py b/examples/fastapi_project/my_faust/timer/__init__.py similarity index 100% rename from examples/fastapi/my_faust/timer/__init__.py rename to examples/fastapi_project/my_faust/timer/__init__.py diff --git a/examples/fastapi_project/my_faust/timer/my_timer.py b/examples/fastapi_project/my_faust/timer/my_timer.py new file mode 100644 index 000000000..dcf22796e --- /dev/null +++ b/examples/fastapi_project/my_faust/timer/my_timer.py @@ -0,0 +1,26 @@ +from uuid import uuid4 + +from fastapi import APIRouter +from my_faust.app import faust_app +from my_faust.topic.my_topic import greetings_topic + +router = APIRouter() + + +async def produce_greeting() -> None: + await greetings_topic.send(value=uuid4().hex) + + +@router.get("/produce") +async def produce(): + await produce_greeting() + return {"success": True} + + +# Keep the timer separate from the route. Stacking ``@faust_app.timer`` on top +# of ``@router.get`` registers the undecorated function as the HTTP route and +# the timer-wrapped one as the timer -- two different callables, which is +# rarely what people mean. +@faust_app.timer(5) +async def produce_periodically() -> None: + await produce_greeting() diff --git a/examples/fastapi/my_faust/topic/__init__.py b/examples/fastapi_project/my_faust/topic/__init__.py similarity index 100% rename from examples/fastapi/my_faust/topic/__init__.py rename to examples/fastapi_project/my_faust/topic/__init__.py diff --git a/examples/fastapi/my_faust/topic/my_topic.py b/examples/fastapi_project/my_faust/topic/my_topic.py similarity index 100% rename from examples/fastapi/my_faust/topic/my_topic.py rename to examples/fastapi_project/my_faust/topic/my_topic.py diff --git a/examples/fastapi_project/worker_main.py b/examples/fastapi_project/worker_main.py new file mode 100644 index 000000000..cbc41494a --- /dev/null +++ b/examples/fastapi_project/worker_main.py @@ -0,0 +1,35 @@ +"""Serve the API from inside the Faust worker, on the worker's event loop. + +This is the mirror image of ``main.py``: instead of uvicorn starting Faust from +an ASGI lifespan, ``faust worker`` starts uvicorn as one of its services:: + + $ faust -A worker_main worker -l info + +The API is then on http://127.0.0.1:8000 and the worker is a single process. +``serve_asgi`` registers the server as an app service, so it starts only after +table recovery finishes -- which is when it is actually safe to serve traffic. + +Requires ``pip install "faust-streaming[fastapi]"``. +""" + +from api import router as api_router +from fastapi import FastAPI +from my_faust.app import faust_app +from my_faust.timer import router as timer_router + +from faust.contrib.fastapi import serve_asgi + +# ``faust -A worker_main`` looks for an attribute named ``app``. +app = faust_app + +api = FastAPI() +api.include_router(router=api_router) +api.include_router(router=timer_router) + + +@api.get("/") +def read_root(): + return {"Hello": "World"} + + +serve_asgi(faust_app, api, port=8000) diff --git a/faust/agents/agent.py b/faust/agents/agent.py index ceff3949e..d92c2ec8b 100644 --- a/faust/agents/agent.py +++ b/faust/agents/agent.py @@ -227,7 +227,14 @@ def __init__( "Agent concurrency must be 1 when using isolated partitions" ) self.use_reply_headers = use_reply_headers - Service.__init__(self, loop=app.loop) + # Do *not* pass ``loop=app.loop`` here. Agents are declared at module + # scope (``@app.agent(...)``), so reading ``app.loop`` at this point + # pins the App to whatever loop ``mode``'s ``get_event_loop()`` finds -- + # at import time that is a loop nobody will ever run. ``mode.Service`` + # binds ``self.loop`` lazily on first access, which happens inside + # ``_default_start()``, i.e. in the loop that actually runs the app. + # See issues #322, #435 and #448. + Service.__init__(self) def on_init_dependencies(self) -> Iterable[ServiceT]: """Return list of services dependencies required to start agent.""" diff --git a/faust/app/base.py b/faust/app/base.py index 581253325..f754b53cc 100644 --- a/faust/app/base.py +++ b/faust/app/base.py @@ -1818,13 +1818,16 @@ def _new_conductor(self) -> ConductorT: return self.transport.create_conductor(beacon=None) def _new_transport(self) -> TransportT: + # No ``loop=`` argument: the transport is reachable from import-time + # code (``app.topics`` -> ``_new_conductor`` -> ``app.transport``), so + # it resolves its loop lazily instead. See ``faust.transport.base``. return transport.by_url(self.conf.broker_consumer[0])( - self.conf.broker_consumer, self, loop=self.loop + self.conf.broker_consumer, self ) def _new_producer_transport(self) -> TransportT: return transport.by_url(self.conf.broker_producer[0])( - self.conf.broker_producer, self, loop=self.loop + self.conf.broker_producer, self ) def _new_cache_backend(self) -> CacheBackendT: @@ -2002,9 +2005,10 @@ def cache(self, cache: CacheBackendT) -> None: @cached_property def tables(self) -> TableManagerT: """Map of available tables, and the table manager service.""" + # No ``loop=``: tables are declared at module scope, so this property + # runs before any loop is running. ``mode.Service`` late-binds. manager = self.conf.TableManager( # type: ignore app=self, - loop=self.loop, beacon=self.beacon, ) return cast(TableManagerT, manager) diff --git a/faust/cli/base.py b/faust/cli/base.py index 8e57b6485..933023d57 100644 --- a/faust/cli/base.py +++ b/faust/cli/base.py @@ -37,6 +37,7 @@ from mode.utils import text from mode.utils.compat import want_bytes from mode.utils.imports import import_from_cwd, symbol_by_name +from mode.utils.loops import get_event_loop from mode.worker import exiting from faust.types import AppT, CodecArg, ModelT @@ -622,7 +623,11 @@ def __call__(self, *args: Any, **kwargs: Any) -> NoReturn: def run_using_worker(self, *args: Any, **kwargs: Any) -> NoReturn: """Execute command using :class:`faust.Worker`.""" - loop = asyncio.get_event_loop_policy().get_event_loop() + # ``get_event_loop_policy().get_event_loop()`` raises on Python 3.14 + # when no loop is current, and nothing has created one by this point: + # declaring agents and tables no longer resolves a loop. mode's + # helper keeps the historical get-or-create behaviour across versions. + loop = get_event_loop() args = self.args + args kwargs = {**self.kwargs, **kwargs} service = self.as_service(loop, *args, **kwargs) @@ -641,7 +646,7 @@ def as_service( return Service.from_awaitable( self.execute(*args, **kwargs), name=type(self).__name__, - loop=loop or asyncio.get_event_loop_policy().get_event_loop(), + loop=loop or get_event_loop(), ) def worker_for_service( @@ -660,7 +665,7 @@ def worker_for_service( console_port=self.console_port, redirect_stdouts=self.redirect_stdouts or False, redirect_stdouts_level=self.redirect_stdouts_level, - loop=loop or asyncio.get_event_loop_policy().get_event_loop(), + loop=loop or get_event_loop(), daemon=self.daemon, ) diff --git a/faust/contrib/fastapi.py b/faust/contrib/fastapi.py new file mode 100644 index 000000000..7ce239dd4 --- /dev/null +++ b/faust/contrib/fastapi.py @@ -0,0 +1,379 @@ +"""Co-host a Faust app with FastAPI (or any other ASGI application). + +Faust and an ASGI server must share a single event loop: a Faust producer +created on one loop cannot be awaited from another. This module provides the +two directions that need: + +1. **The ASGI server drives.** Your own :class:`~fastapi.FastAPI` application + is served by uvicorn, and Faust is started and stopped from its lifespan:: + + import faust + from fastapi import FastAPI + from faust.contrib.fastapi import faust_lifespan + + faust_app = faust.App("hello", broker="kafka://localhost:9092") + greetings = faust_app.topic("greetings", value_type=str) + + api = FastAPI(lifespan=faust_lifespan(faust_app)) + + @api.post("/greet") + async def greet(text: str): + await greetings.send(value=text) + return {"ok": True} + + Run it with ``uvicorn myapp:api``. + +2. **The Faust worker drives.** ``faust worker`` runs as usual and your ASGI + application is served from inside it, on the worker's own loop:: + + from faust.contrib.fastapi import serve_asgi + + api = FastAPI() + serve_asgi(faust_app, api, port=8000) + + Run it with ``faust -A myapp worker -l info``. + +Nothing here imports :pypi:`fastapi` or :pypi:`starlette`, so it works just as +well with Starlette, Quart or Litestar. :pypi:`uvicorn` is imported lazily and +only needed for :func:`serve_asgi`. + +Install the optional dependencies with ``pip install faust-streaming[fastapi]``. +""" + +import asyncio +import contextlib +import typing +from contextlib import asynccontextmanager +from typing import Any, AsyncIterator, Callable, Mapping, Optional, Type + +from mode import Service, get_logger + +from faust.exceptions import ImproperlyConfigured +from faust.types import AppT + +if typing.TYPE_CHECKING: + from asyncio import AbstractEventLoop +else: + AbstractEventLoop = Any + +__all__ = [ + "LoopMismatch", + "AsgiService", + "bind_to_running_loop", + "faust_app_running", + "faust_lifespan", + "serve_asgi", +] + +logger = get_logger(__name__) + +#: How long to wait for the ASGI server to finish serving during shutdown. +DEFAULT_SERVER_SHUTDOWN_TIMEOUT = 10.0 + +LOOP_MISMATCH_HELP = """\ +The Faust app {app!r} is bound to a different event loop than the one now \ +running, so anything it creates (producers, consumers, timers) would be \ +unusable from here. + +This almost always means something read the app's event loop while no loop was \ +running -- usually at import time. The common causes are: + + * accessing ``app.loop`` at module scope + * accessing ``app.web``, ``app.transport`` or ``app.producer`` at module + scope + * calling ``asyncio.get_event_loop()`` yourself and passing it to + ``faust.App(loop=...)`` + +Declare the app, its topics, agents and tables at module scope as usual, but \ +leave the event loop alone -- Faust binds it when the app starts.\ +""" + + +class LoopMismatch(ImproperlyConfigured): + """The Faust app is bound to an event loop other than the running one.""" + + +def bind_to_running_loop(app: AppT) -> AbstractEventLoop: + """Bind ``app`` to the event loop that is currently running. + + Returns the running loop. + + Raises: + RuntimeError: if there is no running event loop. + LoopMismatch: if the app is already bound to a *different* loop. + """ + running = asyncio.get_running_loop() + # Read the private attribute on purpose: the public ``app.loop`` property + # resolves -- and caches -- a loop as a side effect of being read, which is + # exactly what we are trying to detect. + current = getattr(app, "_loop", None) + if current is None: + app.loop = running + elif current is not running: + raise LoopMismatch(LOOP_MISMATCH_HELP.format(app=app.conf.id)) + return running + + +@asynccontextmanager +async def faust_app_running( + app: AppT, + *, + finalize: bool = True, + discover: Optional[bool] = None, + stop_timeout: Optional[float] = None, +) -> AsyncIterator[AppT]: + """Start ``app`` on the running loop, and stop it on exit. + + Use this when you have a lifespan of your own to compose with:: + + @asynccontextmanager + async def lifespan(api: FastAPI): + async with faust_app_running(faust_app): + ml_models["answer"] = load_model() + yield + ml_models.clear() + + Arguments: + app: the Faust app to run. + finalize: call :meth:`~faust.App.finalize` before starting. + discover: run autodiscovery. The default (:const:`None`) discovers + when the app is configured with ``autodiscover``. + stop_timeout: seconds to wait for a graceful stop. :const:`None` + waits indefinitely. + + The app is started with ``maybe_start()``, so this composes with an app + that is already running (and will not stop one it did not start). + """ + bind_to_running_loop(app) + if finalize: + app.finalize() + if discover is None: + discover = bool(app.conf.autodiscover) + if discover: + app.discover() + + started = await app.maybe_start() + try: + yield app + finally: + if started: + if stop_timeout is None: + await app.stop() + else: + await asyncio.wait_for(app.stop(), timeout=stop_timeout) + + +def maybe_instrument_opentelemetry( + asgi_app: Any, enabled: Optional[bool] = None +) -> bool: + """Instrument ``asgi_app`` with OpenTelemetry when that makes sense. + + ``enabled=None`` (the default) auto-detects: instrumentation is attached + only when :pypi:`opentelemetry-instrumentation-fastapi` is installed *and* + the application has configured a real ``TracerProvider``. Until an SDK is + configured the OpenTelemetry API is a no-op, so this never turns on + telemetry an operator did not ask for. + + Pass ``False`` to opt out entirely, or ``True`` to instrument even when no + SDK has been configured yet (useful if you configure it later). + """ + if enabled is False: + return False + try: + from faust.contrib.opentelemetry import instrument_asgi_app + except Exception as exc: # pragma: no cover + logger.debug("OpenTelemetry: integration unavailable: %r", exc) + return False + return instrument_asgi_app(asgi_app, force=bool(enabled)) + + +def faust_lifespan( + app: AppT, *, opentelemetry: Optional[bool] = None, **kwargs: Any +) -> Callable[..., Any]: + """Build an ASGI ``lifespan`` handler that runs ``app``. + + Accepts the same keyword arguments as :func:`faust_app_running`:: + + api = FastAPI(lifespan=faust_lifespan(faust_app)) + + If OpenTelemetry is installed and configured, the ASGI application is + instrumented automatically; pass ``opentelemetry=False`` to opt out. + """ + + @asynccontextmanager + async def lifespan(*args: Any, **lifespan_kwargs: Any) -> AsyncIterator[None]: + # ASGI hands the application object to the lifespan handler, which is + # the thing OpenTelemetry needs to wrap. + if args: + maybe_instrument_opentelemetry(args[0], opentelemetry) + async with faust_app_running(app, **kwargs): + # Yield None, not the app: Starlette treats a non-None lifespan + # value as a state mapping to merge into the ASGI scope. + yield + + return lifespan + + +@contextlib.contextmanager +def _no_signal_handlers() -> Any: + """Stand-in for ``uvicorn.Server.capture_signals`` that installs nothing.""" + yield + + +def _disable_signal_handling(server: Any) -> None: + """Leave process signals to :class:`mode.Worker`. + + ``mode.Worker`` registers SIGINT/SIGTERM via ``loop.add_signal_handler()``, + which on Unix is delivered through asyncio's wakeup file descriptor. + uvicorn installs its own handlers with ``signal.signal()`` afterwards, + which replaces the OS-level handler but *not* the wakeup fd -- so both + still fire, and the worker shuts down gracefully either way. (Checked + with real SIGINTs, single and repeated, against a live uvicorn.) + + What this avoids is a second, concurrent shutdown path: uvicorn reacting + to the same Ctrl-C, logging a duplicate interrupt, and -- on a repeated + signal -- setting ``force_exit`` while the worker is still draining + in-flight work. One owner for process signals keeps shutdown predictable. + + The hook moved in uvicorn 0.27 -- older versions call + ``install_signal_handlers()``, newer ones use the ``capture_signals()`` + context manager -- so neutralize whichever is present. + """ + if hasattr(server, "capture_signals"): + server.capture_signals = _no_signal_handlers + if hasattr(server, "install_signal_handlers"): + server.install_signal_handlers = lambda *args, **kwargs: None + + +class AsgiService(Service): + """Serve an ASGI application with uvicorn, on the Faust worker's loop. + + Usually created for you by :func:`serve_asgi`. + """ + + #: The ASGI application to serve. + asgi_app: Any = None + + #: Interface to bind to. + host: str = "0.0.0.0" # nosec: B104 + + #: Port to bind to. + port: int = 8000 + + #: Extra keyword arguments for :class:`uvicorn.Config`. + uvicorn_options: Mapping[str, Any] = {} + + #: Seconds to wait for the server to finish serving on shutdown. + server_shutdown_timeout: float = DEFAULT_SERVER_SHUTDOWN_TIMEOUT + + #: Instrument the ASGI app with OpenTelemetry. :const:`None` auto-detects. + opentelemetry: Optional[bool] = None + + def __init__( + self, + asgi_app: Any = None, + *, + host: Optional[str] = None, + port: Optional[int] = None, + uvicorn_options: Optional[Mapping[str, Any]] = None, + **kwargs: Any, + ) -> None: + if asgi_app is not None: + self.asgi_app = asgi_app + if host is not None: + self.host = host + if port is not None: + self.port = port + if uvicorn_options is not None: + self.uvicorn_options = uvicorn_options + self._server: Any = None + self._serve_fut: Optional[asyncio.Future] = None + Service.__init__(self, **kwargs) + + def _create_server(self) -> Any: + """Build the uvicorn server (overridden in tests).""" + try: + import uvicorn + except ImportError as exc: # pragma: no cover + raise ImproperlyConfigured( + "serve_asgi() requires uvicorn: " + 'pip install "faust-streaming[fastapi]"' + ) from exc + + options = dict(self.uvicorn_options) + # ``loop="none"`` keeps uvicorn from installing its own event loop + # policy -- we are already running inside the worker's loop. + options.setdefault("loop", "none") + options.setdefault("lifespan", "on") + options.setdefault("log_config", None) + config = uvicorn.Config( + self.asgi_app, host=self.host, port=self.port, **options + ) + server = uvicorn.Server(config) + _disable_signal_handling(server) + return server + + async def on_start(self) -> None: + """Start serving.""" + if self.asgi_app is None: + raise ImproperlyConfigured("AsgiService requires an ASGI application") + maybe_instrument_opentelemetry(self.asgi_app, self.opentelemetry) + self._server = self._create_server() + self._serve_fut = self.add_future(self._server.serve()) + + async def on_stop(self) -> None: + """Ask the server to exit and wait for it.""" + server, fut = self._server, self._serve_fut + self._server = self._serve_fut = None + if server is not None: + server.should_exit = True + if fut is not None: + try: + await asyncio.wait_for(fut, timeout=self.server_shutdown_timeout) + except asyncio.TimeoutError: + logger.warning( + "ASGI server did not stop within %ss", + self.server_shutdown_timeout, + ) + except asyncio.CancelledError: # pragma: no cover + pass + + @property + def label(self) -> str: + """Return description of this service, used in logs.""" + return f"{type(self).__name__}: http://{self.host}:{self.port}" + + +def serve_asgi( + app: AppT, + asgi_app: Any, + *, + host: str = "0.0.0.0", # nosec: B104 + port: int = 8000, + **uvicorn_options: Any, +) -> Type[AsgiService]: + """Serve ``asgi_app`` from inside the Faust worker. + + The server is registered as an extra app service, so it starts once the + app is up -- after table recovery has finished, which is when it is + actually safe to serve traffic:: + + api = FastAPI() + serve_asgi(faust_app, api, port=8000) + + Note this is separate from Faust's own web server (``@app.page`` and + friends), which keeps running on ``web_port``. Set ``web_enabled=False`` + if you do not want it. + """ + cls: Type[AsgiService] = type( + "FaustAsgiService", + (AsgiService,), + { + "asgi_app": asgi_app, + "host": host, + "port": port, + "uvicorn_options": uvicorn_options, + }, + ) + app.service(cls) + return cls diff --git a/faust/contrib/opentelemetry.py b/faust/contrib/opentelemetry.py new file mode 100644 index 000000000..6867788af --- /dev/null +++ b/faust/contrib/opentelemetry.py @@ -0,0 +1,337 @@ +"""OpenTelemetry tracing for Faust, and for a co-hosted ASGI application. + +Why this exists +=============== + +:pypi:`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 (:class:`~faust.transport.consumer.ConsumerThread`), and +:mod:`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. + +:class:`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 +===== + +.. sourcecode:: python + + 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 :pypi:`opentracing` support + (:class:`faust.sensors.distributed_tracing.TracingSensor`) *does* inject into + Kafka headers. Do not run both; :func:`setup_opentelemetry` warns if it + sees one already registered. +""" + +import typing +from typing import Any, Dict, List, Mapping, Optional, Sequence + +from mode import get_logger + +from faust.sensors.base import Sensor +from faust.types import TP, AppT, EventT, StreamT + +if typing.TYPE_CHECKING: + from opentelemetry.trace import Tracer +else: + Tracer = Any + +__all__ = [ + "OpenTelemetrySensor", + "instrument_asgi_app", + "opentelemetry_available", + "sdk_is_configured", + "setup_opentelemetry", +] + +logger = get_logger(__name__) + +#: Instrumentation scope name reported for spans created here. +INSTRUMENTATION_NAME = "faust" + +#: Provider class names that mean "the user never configured an SDK". +_NOOP_PROVIDERS = frozenset( + {"ProxyTracerProvider", "NoOpTracerProvider", "DefaultTracerProvider"} +) + + +def opentelemetry_available() -> bool: + """Return :const:`True` if the OpenTelemetry API is importable.""" + try: + import opentelemetry.trace # noqa: F401 + except Exception: # pragma: no cover + return False + return True + + +def sdk_is_configured() -> bool: + """Return :const:`True` if a real :class:`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. + """ + try: + from opentelemetry import trace + except Exception: # pragma: no cover + return False + return type(trace.get_tracer_provider()).__name__ not in _NOOP_PROVIDERS + + +def _kafka_headers_as_list( + headers: Optional[Any], +) -> List[Any]: + """Normalize Faust's ``HeadersArg`` to a list of ``(str, bytes)``.""" + if not headers: + return [] + if isinstance(headers, Mapping): + return list(headers.items()) + return list(headers) + + +def _build_getter() -> Any: + from opentelemetry.propagators import textmap + + class KafkaHeadersGetter(textmap.Getter): + """Read W3C trace context out of Kafka record headers. + + Kafka headers are a sequence of ``(str, bytes)`` pairs which may repeat + a key; OpenTelemetry carriers are string-keyed. The first match wins, + matching the behaviour of the aiokafka instrumentation. + """ + + def get(self, carrier: Any, key: str) -> Optional[Sequence[str]]: + for item_key, value in _kafka_headers_as_list(carrier): + if item_key == key and value is not None: + if isinstance(value, bytes): + return [value.decode("utf-8", "replace")] + return [str(value)] + return None + + def keys(self, carrier: Any) -> List[str]: + return [key for key, _ in _kafka_headers_as_list(carrier)] + + return KafkaHeadersGetter() + + +class OpenTelemetrySensor(Sensor): + """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. + """ + + def __init__(self, *, tracer: Optional[Tracer] = None, **kwargs: Any) -> None: + super().__init__(**kwargs) + self._explicit_tracer = tracer + self._getter = _build_getter() + + @property + def tracer(self) -> Tracer: + """Tracer used to create spans (resolved lazily).""" + if self._explicit_tracer is not None: + return self._explicit_tracer + from opentelemetry import trace + + # Resolved on every call on purpose: applications commonly configure + # the SDK after importing their Faust app. + return trace.get_tracer(INSTRUMENTATION_NAME) + + def _span_attributes( + self, stream: StreamT, event: EventT, tp: TP, offset: int + ) -> Dict[str, Any]: + message = event.message + attributes: Dict[str, Any] = { + "messaging.system": "kafka", + "messaging.operation.name": "process", + "messaging.operation.type": "process", + "messaging.destination.name": tp.topic, + "messaging.destination.partition.id": str(tp.partition), + "messaging.kafka.message.offset": offset, + } + # Sensors are not given the app, but the stream knows it. + app = getattr(stream, "app", None) + consumer_group = getattr(getattr(app, "conf", None), "id", None) + if consumer_group: + attributes["messaging.consumer.group.name"] = str(consumer_group) + key = getattr(message, "key", None) + # The spec says this MUST NOT be set when the key is null. + if key is not None: + attributes["messaging.kafka.message.key"] = ( + key.decode("utf-8", "replace") if isinstance(key, bytes) else str(key) + ) + return attributes + + def on_stream_event_in( + self, tp: TP, offset: int, stream: StreamT, event: EventT + ) -> Optional[Dict]: + """Start a ``process`` span and make it current.""" + try: + from opentelemetry import context as otel_context, propagate, trace + except Exception: # pragma: no cover + return None + + try: + headers = getattr(event.message, "headers", None) + parent = propagate.extract(headers, getter=self._getter) + span = self.tracer.start_span( + f"{tp.topic} process", + context=parent, + kind=trace.SpanKind.CONSUMER, + attributes=self._span_attributes(stream, event, tp, offset), + ) + token = otel_context.attach(trace.set_span_in_context(span)) + except Exception as exc: # pragma: no cover + # Telemetry must never break message processing. + logger.debug("OpenTelemetry: could not start process span: %r", exc) + return None + return {"span": span, "token": token} + + def on_stream_event_out( + self, + tp: TP, + offset: int, + stream: StreamT, + event: EventT, + state: Dict = None, + ) -> None: + """Detach the context and end the span.""" + if not state: + # ``Stream.ack()`` calls this without sensor state; the span is + # then closed by the ``__aiter__`` path that opened it. + return + span = state.pop("span", None) + token = state.pop("token", None) + try: + from opentelemetry import context as otel_context + except Exception: # pragma: no cover + return + try: + if token is not None: + otel_context.detach(token) + except Exception as exc: # pragma: no cover + logger.debug("OpenTelemetry: could not detach context: %r", exc) + if span is not None: + try: + span.end() + except Exception as exc: # pragma: no cover + logger.debug("OpenTelemetry: could not end span: %r", exc) + + +def instrument_asgi_app( + asgi_app: Any, *, tracer_provider: Any = None, force: bool = False +) -> bool: + """Attach OpenTelemetry instrumentation to a FastAPI/Starlette app. + + Returns :const:`True` if instrumentation was attached. + + Does nothing -- and returns :const:`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``). + """ + if asgi_app is None: + return False + if getattr(asgi_app, "_is_instrumented_by_opentelemetry", False): + logger.debug("OpenTelemetry: ASGI app already instrumented, skipping") + return False + if not force and not sdk_is_configured(): + return False + + try: + # A ``try/except ImportError`` rather than ``find_spec``: the + # instrumentation packages pin their siblings exactly, so an + # importable-but-broken install is a realistic failure mode. + from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor + except Exception as exc: + logger.debug("OpenTelemetry: FastAPI instrumentation unavailable: %r", exc) + return False + + try: + FastAPIInstrumentor.instrument_app( + asgi_app, + tracer_provider=tracer_provider, + exclude_spans=["receive", "send"], + ) + except TypeError: + # ``exclude_spans`` was added in a later release. + try: + FastAPIInstrumentor.instrument_app( + asgi_app, tracer_provider=tracer_provider + ) + except Exception as exc: + logger.debug("OpenTelemetry: could not instrument ASGI app: %r", exc) + return False + except Exception as exc: + logger.debug("OpenTelemetry: could not instrument ASGI app: %r", exc) + return False + logger.info("OpenTelemetry: instrumented ASGI application") + return True + + +def setup_opentelemetry( + app: AppT, *, tracer: Optional[Tracer] = None +) -> Optional[OpenTelemetrySensor]: + """Register the OpenTelemetry sensor on ``app``. + + Returns the sensor, or :const:`None` if OpenTelemetry is not installed. + """ + if not opentelemetry_available(): + logger.debug("OpenTelemetry: API not installed, tracing sensor not registered") + return None + + if _has_opentracing_sensor(app): + logger.warning( + "OpenTelemetry: a TracingSensor (opentracing) is already " + "registered. Both inject/extract Kafka trace headers; running " + "them together produces duplicate traceparent headers and " + "confusing traces. Register only one." + ) + + sensor = OpenTelemetrySensor(tracer=tracer) + app.sensors.add(sensor) + return sensor + + +def _has_opentracing_sensor(app: AppT) -> bool: + try: + from faust.sensors.distributed_tracing import TracingSensor + except Exception: # pragma: no cover + return False + return any(isinstance(s, TracingSensor) for s in app.sensors) diff --git a/faust/livecheck/case.py b/faust/livecheck/case.py index 771d89050..d54bbc3c7 100644 --- a/faust/livecheck/case.py +++ b/faust/livecheck/case.py @@ -186,7 +186,11 @@ def __init__( # signal attributes have the correct signal instance. self.__dict__.update(self.signals) - Service.__init__(self, loop=app.loop, **kwargs) + # Do *not* pass ``loop=app.loop`` here: cases are declared at module + # scope, and reading ``app.loop`` before a loop is running pins the App + # to a loop that will never be run. ``mode.Service`` late-binds the + # loop on first access. See the note in ``faust.agents.agent``. + Service.__init__(self, **kwargs) @Service.timer(10.0) async def _sampler(self) -> None: diff --git a/faust/tables/base.py b/faust/tables/base.py index ca26c0f11..71324206c 100644 --- a/faust/tables/base.py +++ b/faust/tables/base.py @@ -126,7 +126,11 @@ def __init__( synchronize_all_active_partitions: bool = False, **kwargs: Any, ) -> None: - Service.__init__(self, loop=app.loop, **kwargs) + # Do *not* pass ``loop=app.loop`` here: tables are declared at module + # scope, and reading ``app.loop`` before a loop is running pins the App + # to a loop that will never be run. ``mode.Service`` late-binds the + # loop on first access. See the note in ``faust.agents.agent``. + Service.__init__(self, **kwargs) self.app = app self.name = cast(str, name) # set lazily so CAN BE NONE! self.default = default diff --git a/faust/transport/base.py b/faust/transport/base.py index b3e478379..c5e42fd25 100644 --- a/faust/transport/base.py +++ b/faust/transport/base.py @@ -13,6 +13,7 @@ from typing import Any, ClassVar, List, Optional, Type from mode.services import ServiceT +from mode.utils.loops import get_event_loop from yarl import URL from faust.types import AppT @@ -62,7 +63,26 @@ def __init__( ) -> None: self.url = url self.app = app - self.loop = loop or asyncio.get_event_loop_policy().get_event_loop() + self._loop = loop + + @property + def loop(self) -> asyncio.AbstractEventLoop: + """Event loop this transport belongs to (resolved lazily). + + A transport can be constructed at import time -- declaring an agent + touches ``app.topics``, which builds the conductor and therefore + ``app.transport`` -- so resolving the loop in ``__init__`` would pin + the App to a loop that is never run. Resolving on first *access* + instead means the loop is picked up from whichever loop is actually + running the app. See the note in ``faust.agents.agent``. + """ + if self._loop is None: + self._loop = get_event_loop() + return self._loop + + @loop.setter + def loop(self, loop: Optional[asyncio.AbstractEventLoop]) -> None: + self._loop = loop def create_consumer(self, callback: ConsumerCallback, **kwargs: Any) -> ConsumerT: """Create new consumer.""" @@ -85,4 +105,8 @@ def create_transaction_manager( def create_conductor(self, **kwargs: Any) -> ConductorT: """Create new consumer conductor.""" - return self.Conductor(app=self.app, loop=self.loop, **kwargs) + # No ``loop=``: the conductor is built from ``app.topics``, which is + # reached at import time when an agent is declared. Reading + # ``self.loop`` here would resolve -- and cache -- a loop that is + # never run. ``Conductor`` is a ``mode.Service`` and late-binds. + return self.Conductor(app=self.app, **kwargs) diff --git a/faust/types/settings/settings.py b/faust/types/settings/settings.py index cf080dab8..1d319084e 100644 --- a/faust/types/settings/settings.py +++ b/faust/types/settings/settings.py @@ -161,10 +161,15 @@ def __init__( web_application_options: typing.Mapping[str, typing.Any] = None, web_cors_options: typing.Mapping[str, ResourceOptions] = None, web_enabled: Optional[bool] = None, + web_graph_enabled: Optional[bool] = None, web_host: Optional[str] = None, web_in_thread: Optional[bool] = None, + web_metrics_enabled: Optional[bool] = None, web_port: Optional[int] = None, + web_router_enabled: Optional[bool] = None, web_ssl_context: ssl.SSLContext = None, + web_stats_enabled: Optional[bool] = None, + web_tables_enabled: Optional[bool] = None, web_transport: URLArg = None, # Worker settings: worker_redirect_stdouts: Optional[bool] = None, @@ -1934,6 +1939,112 @@ def web_enabled(self) -> bool: This option can also be set using :option:`faust worker --without-web`. """ + @sections.WebServer.setting( + params.Bool, + version_introduced="0.13", + env_name="APP_WEB_STATS_ENABLED", + default_alias="debug", + ) + def web_stats_enabled(self) -> bool: + """Enable/disable the built-in statistics endpoints. + + Serves sensor statistics at ``/`` and the current partition + assignment at ``/assignment/``. + + If not set, this follows :setting:`debug`, which is how these + endpoints were gated before this setting existed. + + When disabled, ``/`` serves the plain production index instead. + + .. warning:: + + These endpoints expose internal state: every registered sensor's + counters, and which partitions this worker is handling. + """ + + @sections.WebServer.setting( + params.Bool, + version_introduced="0.13", + env_name="APP_WEB_GRAPH_ENABLED", + default_alias="debug", + ) + def web_graph_enabled(self) -> bool: + """Enable/disable the ``/graph`` dependency graph endpoint. + + Renders the worker's service dependency graph as a PNG. + + If not set, this follows :setting:`debug`, which is how this endpoint + was gated before this setting existed. + + .. warning:: + + The graph describes the entire internal service tree of the + worker. + """ + + @sections.WebServer.setting( + params.Bool, + version_introduced="0.13", + env_name="APP_WEB_ROUTER_ENABLED", + default=True, + ) + def web_router_enabled(self) -> bool: + """Enable/disable the ``/router`` endpoints. + + These report which worker in the cluster owns a given table key, and + are what makes :meth:`@table_route` work across nodes. + + .. warning:: + + Disabling this breaks :meth:`@table_route` for multi-node + deployments. Only turn it off if you route entirely within a + single worker. + + Left enabled, it exposes your cluster topology -- the URLs of + every other worker. + """ + + @sections.WebServer.setting( + params.Bool, + version_introduced="0.13", + env_name="APP_WEB_TABLES_ENABLED", + default=True, + ) + def web_tables_enabled(self) -> bool: + """Enable/disable the ``/table`` endpoints. + + These list the tables defined by this app and allow reading + individual keys over HTTP. + + .. warning:: + + This exposes table *data*, not just table names. If your tables + hold anything sensitive, turn this off. + """ + + @sections.WebServer.setting( + params.Bool, + version_introduced="0.13", + env_name="APP_WEB_METRICS_ENABLED", + default=False, + ) + def web_metrics_enabled(self) -> bool: + """Enable/disable the ``/performance/`` metrics endpoint. + + Serves throughput, latency, consumer lag and table statistics as + JSON, gathered from :setting:`Monitor`. Unlike the statistics + endpoints this is independent of :setting:`debug`, so it can be left + on in production. + + Disabled by default: it is a new endpoint, and enabling it should be + a deliberate choice. + + .. seealso:: + + :mod:`faust.sensors.prometheus` serves the same underlying data + in Prometheus format on its own ``/metrics`` path. + """ + @sections.WebServer.setting( params.Str, version_introduced="1.2", diff --git a/faust/web/apps/metrics.py b/faust/web/apps/metrics.py new file mode 100644 index 000000000..b024b4e7d --- /dev/null +++ b/faust/web/apps/metrics.py @@ -0,0 +1,148 @@ +"""HTTP endpoint exposing performance metrics. + +Enabled with the :setting:`web_metrics_enabled` setting, which is off by +default:: + + app = faust.App("myapp", web_metrics_enabled=True) + +Unlike the statistics endpoints in :mod:`faust.web.apps.stats` this is +independent of :setting:`debug`, so it can be left on in production. + +The payload is grouped by concern rather than being a flat dump of +:meth:`Monitor.asdict() `, and adds two +things that monitor does not compute: + +* **consumer lag** -- derived from the log end offsets and the offsets this + worker has actually read, which is the number you usually want to alert on. +* **latency percentiles** -- monitor keeps raw deques of up to a few thousand + samples; those are summarized here instead of serialized. + +:mod:`faust.sensors.prometheus` serves the same underlying data in Prometheus +format on its own path; the two are independent. +""" + +from typing import Any, Iterable, Mapping, MutableMapping, Optional + +from faust import web + +__all__ = ["Metrics", "blueprint"] + +blueprint = web.Blueprint("metrics") + + +def _percentile(values: Iterable[float], percentile: float) -> Optional[float]: + """Nearest-rank percentile, or :const:`None` when there is no data. + + Deliberately not :mod:`statistics.quantiles`: that raises on fewer than + two samples, and a metrics endpoint should never fail because a worker has + only just started. + """ + ordered = sorted(values) + if not ordered: + return None + index = int(round(percentile * (len(ordered) - 1))) + return ordered[index] + + +def _summarize(values: Iterable[float]) -> Mapping[str, Optional[float]]: + ordered = list(values) + return { + "count": len(ordered), + "p50": _percentile(ordered, 0.50), + "p95": _percentile(ordered, 0.95), + "max": max(ordered) if ordered else None, + } + + +@blueprint.route("/", name="index") +class Metrics(web.View): + """ + --- + description: Worker performance metrics. + tags: + - Faust + produces: + - application/json + """ + + async def get(self, request: web.Request) -> web.Response: + """Return JSON response with performance metrics.""" + return self.json(self.metrics()) + + def metrics(self) -> Mapping[str, Any]: + """Build the metrics payload.""" + app = self.app + monitor = app.monitor + return { + "app": { + "id": app.conf.id, + "web_port": app.conf.web_port, + }, + "throughput": self._throughput(monitor), + "latency": self._latency(monitor), + "consumer": self._consumer(monitor), + "tables": {name: state.asdict() for name, state in monitor.tables.items()}, + "errors": { + "send_errors": monitor.send_errors, + "assignments_completed": monitor.assignments_completed, + "assignments_failed": monitor.assignments_failed, + }, + "topic_buffer_full": monitor._topic_buffer_full_dict(), + } + + def _throughput(self, monitor: Any) -> Mapping[str, Any]: + return { + "messages_active": monitor.messages_active, + "messages_received_total": monitor.messages_received_total, + "messages_s": monitor.messages_s, + "messages_sent": monitor.messages_sent, + "events_active": monitor.events_active, + "events_total": monitor.events_total, + "events_s": monitor.events_s, + } + + def _latency(self, monitor: Any) -> Mapping[str, Any]: + return { + "events_runtime_avg": monitor.events_runtime_avg, + "commit_latency": _summarize(monitor.commit_latency), + "send_latency": _summarize(monitor.send_latency), + "assignment_latency": _summarize(monitor.assignment_latency), + "rebalance_return_avg": monitor.rebalance_return_avg, + "rebalance_end_avg": monitor.rebalance_end_avg, + "http_response_latency_avg": monitor.http_response_latency_avg, + } + + def _consumer(self, monitor: Any) -> Mapping[str, Any]: + read = monitor._tp_read_offsets_dict() + committed = monitor._tp_committed_offsets_dict() + end = monitor._tp_end_offsets_dict() + lag, lag_total = self._lag(read, end) + return { + "lag_total": lag_total, + "lag_by_partition": lag, + "read_offsets": read, + "committed_offsets": committed, + "end_offsets": end, + "rebalances": monitor.rebalances, + } + + @classmethod + def _lag(cls, read: Mapping, end: Mapping) -> Any: + """Consumer lag per partition, and the total across all partitions. + + A partition is skipped when either offset is unknown -- reporting it + as zero lag would be a lie, and reporting it as the full end offset + would spike alerts every time a worker starts. + """ + lag: MutableMapping[str, MutableMapping[int, int]] = {} + total = 0 + for topic, end_partitions in end.items(): + read_partitions = read.get(topic) or {} + for partition, end_offset in end_partitions.items(): + read_offset = read_partitions.get(partition) + if read_offset is None or end_offset is None: + continue + behind = max(0, end_offset - read_offset) + lag.setdefault(topic, {})[partition] = behind + total += behind + return lag, total diff --git a/faust/web/base.py b/faust/web/base.py index f62fb2955..7ec6a4c32 100644 --- a/faust/web/base.py +++ b/faust/web/base.py @@ -55,6 +55,23 @@ ("", "faust.web.apps.stats:blueprint"), ] +#: Blueprints that are off unless explicitly enabled. +OPTIONAL_BLUEPRINTS: _BPList = [ + ("/performance", "faust.web.apps.metrics:blueprint"), +] + +#: Maps a blueprint to the setting that enables it. +#: +#: A blueprint not listed here is always enabled, so user-supplied blueprints +#: are unaffected. +BLUEPRINT_FLAGS: Mapping[str, str] = { + "faust.web.apps.router:blueprint": "web_router_enabled", + "faust.web.apps.tables.blueprint": "web_tables_enabled", + "faust.web.apps.graph:blueprint": "web_graph_enabled", + "faust.web.apps.stats:blueprint": "web_stats_enabled", + "faust.web.apps.metrics:blueprint": "web_metrics_enabled", +} + CONTENT_SEPARATOR: bytes = b"\r\n\r\n" HEADER_SEPARATOR: bytes = b"\r\n" HEADER_KEY_VALUE_SEPARATOR: bytes = b": " @@ -163,6 +180,8 @@ class Web(Service): default_blueprints: ClassVar[_BPList] = DEFAULT_BLUEPRINTS # noqa: E704 production_blueprints: ClassVar[_BPList] = PRODUCTION_BLUEPRINTS debug_blueprints: ClassVar[_BPList] = DEBUG_BLUEPRINTS + optional_blueprints: ClassVar[_BPList] = OPTIONAL_BLUEPRINTS + blueprint_flags: ClassVar[Mapping[str, str]] = BLUEPRINT_FLAGS app: AppT @@ -181,13 +200,38 @@ def __init__(self, app: AppT, **kwargs: Any) -> None: self.app = app self.views = {} self.reverse_names = {} - blueprints = list(self.default_blueprints) - if self.app.conf.debug: - blueprints.extend(self.debug_blueprints) - else: - blueprints.extend(self.production_blueprints) - self.blueprints = BlueprintManager(blueprints) - Service.__init__(self, loop=app.loop, **kwargs) + self.blueprints = BlueprintManager(self._enabled_blueprints()) + # Do *not* pass ``loop=app.loop`` here: ``app.web`` is a cached + # property that is commonly touched before the loop is running (the + # ``faust worker`` banner does so), and reading ``app.loop`` there pins + # the App to a loop that will never be run. ``mode.Service`` + # late-binds the loop. See the note in ``faust.agents.agent``. + Service.__init__(self, **kwargs) + + def _enabled_blueprints(self) -> List[Tuple[str, _BPArg]]: + """Select which built-in blueprints to serve. + + Each built-in blueprint has its own setting (see + :attr:`blueprint_flags`). The statistics and graph endpoints take + their default from :setting:`debug`, so an app that does not set the + new settings serves exactly what it served before. + """ + conf = self.app.conf + candidates: List[Tuple[str, _BPArg]] = list(self.default_blueprints) + candidates.extend(self.debug_blueprints) + candidates.extend(self.optional_blueprints) + if not conf.web_stats_enabled: + # Statistics and the production index both mount at "/", so + # exactly one of them is served. + candidates.extend(self.production_blueprints) + return [(prefix, bp) for prefix, bp in candidates if self._is_enabled(bp, conf)] + + def _is_enabled(self, blueprint: _BPArg, conf: Any) -> bool: + setting = ( + self.blueprint_flags.get(blueprint) if isinstance(blueprint, str) else None + ) + # Blueprints with no flag -- including any a subclass adds -- are on. + return True if setting is None else bool(getattr(conf, setting)) @abc.abstractmethod def text( diff --git a/faust/worker.py b/faust/worker.py index d17f1491f..3a8ceb612 100644 --- a/faust/worker.py +++ b/faust/worker.py @@ -162,7 +162,7 @@ async def start_worker(worker: Worker) -> None: await worker.start() def manage_loop(): - loop = asyncio.get_event_loop_policy().get_event_loop() + loop = asyncio.new_event_loop() worker = Worker(app, loop=loop) loop.run_until_complete(start_worker(worker)) diff --git a/requirements/extras/fastapi.txt b/requirements/extras/fastapi.txt new file mode 100644 index 000000000..94a0ce577 --- /dev/null +++ b/requirements/extras/fastapi.txt @@ -0,0 +1,2 @@ +fastapi>=0.100.0 +uvicorn>=0.27.0 diff --git a/requirements/extras/opentelemetry.txt b/requirements/extras/opentelemetry.txt new file mode 100644 index 000000000..da5a66612 --- /dev/null +++ b/requirements/extras/opentelemetry.txt @@ -0,0 +1,3 @@ +opentelemetry-api>=1.20.0 +opentelemetry-instrumentation-fastapi>=0.45b0 +opentelemetry-instrumentation-aiokafka>=0.45b0 diff --git a/requirements/test.txt b/requirements/test.txt index b06081cb4..0d1916a27 100644 --- a/requirements/test.txt +++ b/requirements/test.txt @@ -29,6 +29,16 @@ intervaltree -r requirements.txt -r extras/datadog.txt -r extras/opentracing.txt +# The API+SDK only (pure Python, no fastapi): enough to cover +# faust.contrib.opentelemetry's sensor. The instrumentation packages in +# extras/opentelemetry.txt pull in fastapi and are left to that extra. +opentelemetry-api>=1.20.0 +opentelemetry-sdk>=1.20.0 +# uvicorn's only dependencies are click (already required by faust) and h11, +# both pure Python -- cheap enough to cover faust.contrib.fastapi's server +# construction on every matrix leg. fastapi itself is not added here: it +# pulls in pydantic-core, which has no wheel for every leg. +uvicorn>=0.27.0 -r extras/redis.txt -r extras/statsd.txt -r extras/yaml.txt diff --git a/setup.py b/setup.py index 7a3190b70..428f55954 100644 --- a/setup.py +++ b/setup.py @@ -21,6 +21,7 @@ NAME = "faust" BUNDLES = { + "aerospike", "aiodns", "aiomonitor", "cchardet", @@ -30,6 +31,8 @@ "datadog", "debug", "fast", + "fastapi", + "opentelemetry", "opentracing", "orjson", "prometheus", @@ -195,7 +198,12 @@ def do_setup(**kwargs): description=meta["doc"], long_description=long_description, long_description_content_type="text/markdown", - packages=find_packages(exclude=["examples", "ez_setup", "tests", "tests.*"]), + # "examples.*" matters as much as "examples": without it the example + # sub-packages are installed into site-packages under an ``examples`` + # namespace, which is not something a library should claim. + packages=find_packages( + exclude=["examples", "examples.*", "ez_setup", "tests", "tests.*"] + ), # PEP-561: https://www.python.org/dev/peps/pep-0561/ package_data={"faust": ["py.typed"]}, include_package_data=True, diff --git a/tests/functional/test_fastapi_cohost.py b/tests/functional/test_fastapi_cohost.py new file mode 100644 index 000000000..c1e9cd19c --- /dev/null +++ b/tests/functional/test_fastapi_cohost.py @@ -0,0 +1,117 @@ +"""Regression tests for co-hosting Faust with an ASGI server (FastAPI). + +Faust apps are declared at module scope, but under an ASGI server such as +uvicorn the app is *started* from a loop created later by :func:`asyncio.run`. +If declaring an agent/table binds the App to whatever loop happens to exist at +import time, every service built during startup inherits that dead loop and the +worker fails with either:: + + AssertionError: Please create objects with the same loop as running with + RuntimeError: Task ... got Future ... attached to a different loop + +See issues #322, #435 and #448. + +These tests reproduce that split without needing Kafka: the ``declared_app`` +fixture is *synchronous*, so it runs outside the running loop exactly like an +import does, while the ``async def`` tests run inside the per-test loop. +""" + +import asyncio + +import pytest + +import faust + + +@pytest.fixture() +def declared_app(): + """Build an app the way a module does at import time -- no loop running. + + Returns a ``(app, topic, agent, table)`` tuple. Declaring the agent is + what matters most: ``@app.agent`` reaches ``app.topics`` and therefore + builds the conductor and the transport, which is the path that used to + pin the app. + """ + app = faust.App( + "test-fastapi-cohost", + store="memory://", + cache="memory://", + ) + topic = app.topic("greetings", value_type=str) + + @app.agent(topic) + async def printer(stream): # pragma: no cover - never started + async for greeting in stream: + yield greeting + + table = app.Table("cohost-tbl", default=int) + return app, topic, printer, table + + +def test_declaration_does_not_bind_loop(declared_app): + """Declaring agents/tables must not resolve an event loop. + + This is the regression lock. Every other failure mode in #448 follows + from the app being bound here, at import time, to a loop that will never + be run. + """ + app, _topic, agent, table = declared_app + + assert app._loop is None + assert agent._loop is None + assert table._loop is None + assert app.tables._loop is None + assert app.agents._loop is None + + +def test_declaration_does_not_bind_transport_loop(declared_app): + """The transport is reachable from import-time code, so it must be lazy.""" + app, *_ = declared_app + + # Touching ``app.topics`` builds the conductor -> transport. Neither may + # resolve a loop, and neither may pin the app. + assert app.topics is not None + assert app.transport._loop is None + assert app._loop is None + + +async def test_binds_to_running_loop(declared_app): + """First access from inside a running loop must resolve to *that* loop.""" + app, _topic, agent, table = declared_app + running = asyncio.get_running_loop() + + assert app.loop is running + assert agent.loop is running + assert table.loop is running + + +async def test_transport_uses_running_loop(declared_app): + """The transport -- and what it builds -- must land on the running loop. + + Constructing a ``Transport`` performs no I/O and opens no socket, so this + can assert the exact invariant that #448 violated without a broker. + """ + app, *_ = declared_app + running = asyncio.get_running_loop() + + assert app._new_transport().loop is running + assert app._new_producer_transport().loop is running + + +async def test_transport_loop_is_settable(declared_app): + """``Transport.loop`` stayed writable when it became lazy.""" + app, *_ = declared_app + transport = app._new_transport() + sentinel = asyncio.get_running_loop() + + transport.loop = sentinel + assert transport.loop is sentinel + + +async def test_explicit_loop_argument_still_wins(declared_app): + """Passing ``loop=`` to a Transport must still override the lazy lookup.""" + app, *_ = declared_app + running = asyncio.get_running_loop() + transport = type(app.transport)(app.conf.broker_consumer, app, loop=running) + + assert transport.loop is running diff --git a/tests/functional/web/test_metrics.py b/tests/functional/web/test_metrics.py new file mode 100644 index 000000000..772083518 --- /dev/null +++ b/tests/functional/web/test_metrics.py @@ -0,0 +1,102 @@ +import pytest + +from faust.types import TP + +pytestmark = pytest.mark.app(web_metrics_enabled=True) + + +@pytest.fixture() +def traffic(app): + """Populate the monitor as if the worker had been running.""" + monitor = app.monitor + monitor.messages_received_total = 1_048_576 + monitor.messages_s = 812 + monitor.messages_sent = 1_048_570 + monitor.events_total = 1_048_576 + monitor.events_s = 812 + monitor.events_runtime_avg = 0.0031 + monitor.commit_latency.extend([0.010, 0.014, 0.031, 0.012]) + monitor.send_latency.extend([0.002, 0.004, 0.0021]) + monitor.tp_read_offsets[TP("withdrawals", 0)] = 91_240 + monitor.tp_end_offsets[TP("withdrawals", 0)] = 92_042 + monitor.tp_committed_offsets[TP("withdrawals", 0)] = 91_180 + return monitor + + +async def test_metrics(web_client, traffic): + async with await web_client as client: + resp = await client.get("/performance/") + assert resp.status == 200 + payload = await resp.json() + + assert payload["app"]["id"] + assert payload["throughput"]["messages_received_total"] == 1_048_576 + assert payload["throughput"]["events_s"] == 812 + assert payload["latency"]["events_runtime_avg"] == 0.0031 + + +async def test_latency_is_summarized_not_dumped(web_client, traffic): + """Monitor keeps raw deques; the endpoint must not serialize them.""" + async with await web_client as client: + payload = await (await client.get("/performance/")).json() + + commit = payload["latency"]["commit_latency"] + assert commit["count"] == 4 + assert commit["p50"] == 0.014 + assert commit["max"] == 0.031 + assert not isinstance(commit, list) + + +async def test_latency_with_no_samples(web_client): + """A worker that just started must not 500.""" + async with await web_client as client: + resp = await client.get("/performance/") + assert resp.status == 200 + payload = await resp.json() + + assert payload["latency"]["commit_latency"] == { + "count": 0, + "p50": None, + "p95": None, + "max": None, + } + + +async def test_consumer_lag(web_client, traffic): + async with await web_client as client: + payload = await (await client.get("/performance/")).json() + + consumer = payload["consumer"] + assert consumer["lag_total"] == 802 + assert consumer["lag_by_partition"] == {"withdrawals": {"0": 802}} + assert consumer["read_offsets"] == {"withdrawals": {"0": 91_240}} + assert consumer["committed_offsets"] == {"withdrawals": {"0": 91_180}} + + +async def test_lag_skips_partitions_with_unknown_read_offset(web_client, traffic): + """Reporting a fresh partition as fully lagged would spike alerts.""" + traffic.tp_end_offsets[TP("withdrawals", 1)] = 500 + + async with await web_client as client: + payload = await (await client.get("/performance/")).json() + + lag = payload["consumer"]["lag_by_partition"]["withdrawals"] + assert "1" not in lag + assert payload["consumer"]["lag_total"] == 802 + + +async def test_tables_are_included(web_client, app): + app.Table("counts") + app.monitor.on_table_get(app.tables["counts"], "k") + + async with await web_client as client: + payload = await (await client.get("/performance/")).json() + + assert payload["tables"]["counts"]["keys_retrieved"] == 1 + + +@pytest.mark.app(web_metrics_enabled=False) +async def test_disabled_by_default(web_client): + async with await web_client as client: + resp = await client.get("/performance/") + assert resp.status == 404 diff --git a/tests/integration/fastapi/test_cohost.py b/tests/integration/fastapi/test_cohost.py new file mode 100644 index 000000000..a834da593 --- /dev/null +++ b/tests/integration/fastapi/test_cohost.py @@ -0,0 +1,78 @@ +"""End-to-end co-hosting check against a real FastAPI application. + +Skipped unless the ``faust-streaming[fastapi]`` extra (and :pypi:`httpx`) are +installed, so it costs nothing in the default CI matrix. No socket is bound: +``httpx.ASGITransport`` calls the ASGI application in-process, which is enough +to exercise the lifespan and prove that an endpoint can produce to a topic on +the same event loop. +""" + +from unittest.mock import Mock + +import pytest + +import faust +from faust.contrib.fastapi import faust_lifespan + +fastapi = pytest.importorskip("fastapi") +httpx = pytest.importorskip("httpx") + + +@pytest.fixture() +def cohosted(): + """Declare app, topic, agent and API the way a module would.""" + app = faust.App( + "test-fastapi-integration", + store="memory://", + cache="memory://", + web_enabled=False, + ) + greetings = app.topic("greetings", value_type=str) + + @app.agent(greetings) + async def printer(stream): # pragma: no cover - never started + async for greeting in stream: + yield greeting + + api = fastapi.FastAPI(lifespan=faust_lifespan(app, discover=False)) + + @api.post("/greet") + async def greet(text: str): + await greetings.send(value=text) + return {"ok": True} + + @api.get("/loop") + async def loop_identity(): + import asyncio + + return {"same_loop": app.loop is asyncio.get_running_loop()} + + return app, api, greetings + + +async def test_endpoint_produces_on_the_same_loop(cohosted): + app, api, greetings = cohosted + # Stub the send path: this test is about loop identity and wiring, not + # about talking to a broker. + sent = [] + + async def fake_send(**kwargs): + sent.append(kwargs) + return Mock(name="record_metadata") + + greetings.send = fake_send + + transport = httpx.ASGITransport(app=api) + async with httpx.AsyncClient( + transport=transport, base_url="http://testserver" + ) as client: + response = await client.post("/greet", params={"text": "hello"}) + assert response.status_code == 200 + assert response.json() == {"ok": True} + + loop_response = await client.get("/loop") + assert loop_response.json() == {"same_loop": True} + + assert sent == [{"value": "hello"}] + # The lifespan must have stopped the app again on exit. + assert not app.started diff --git a/tests/meticulous/assignor/test_copartitioned_assignor.py b/tests/meticulous/assignor/test_copartitioned_assignor.py index e0fb1f66b..3c5d07d86 100644 --- a/tests/meticulous/assignor/test_copartitioned_assignor.py +++ b/tests/meticulous/assignor/test_copartitioned_assignor.py @@ -1,4 +1,5 @@ import copy +import platform from collections import Counter from typing import MutableMapping @@ -8,7 +9,13 @@ from faust.assignor.client_assignment import CopartitionedAssignment from faust.assignor.copartitioned_assignor import CopartitionedAssignor -TEST_DEADLINE = 4000 +# PyPy runs these property tests several times slower than CPython, and the +# largest generated cases (hundreds of clients x hundreds of partitions) land +# either side of a 4s deadline from run to run. Hypothesis reports that as +# FlakyFailure ("failed on the first call but did not on a subsequent one"), +# so the deadline is raised there rather than dropped everywhere -- CPython +# keeps the tighter bound that makes it a useful performance guard. +TEST_DEADLINE = 20000 if platform.python_implementation() == "PyPy" else 4000 _topics = {"foo", "bar", "baz"} diff --git a/tests/unit/app/test_base.py b/tests/unit/app/test_base.py index 0d4278b68..b79992269 100644 --- a/tests/unit/app/test_base.py +++ b/tests/unit/app/test_base.py @@ -137,9 +137,9 @@ def test_new_transport(self, broker_url, broker_consumer_url, *, app, patching): assert app._new_transport() is by_url.return_value.return_value assert app.transport is by_url.return_value.return_value by_url.assert_called_with(app.conf.broker_consumer[0]) - by_url.return_value.assert_called_with( - app.conf.broker_consumer, app, loop=app.loop - ) + # No ``loop=`` is passed: the transport resolves its loop lazily so + # that building it at import time cannot pin the app to a dead loop. + by_url.return_value.assert_called_with(app.conf.broker_consumer, app) app.transport = 10 assert app.transport == 10 @@ -161,9 +161,8 @@ def test_new_producer_transport( assert transport is by_url.return_value.return_value assert app.producer_transport is by_url.return_value.return_value by_url.assert_called_with(app.conf.broker_producer[0]) - by_url.return_value.assert_called_with( - app.conf.broker_producer, app, loop=app.loop - ) + # See ``test_new_transport``: no ``loop=`` argument by design. + by_url.return_value.assert_called_with(app.conf.broker_producer, app) app.producer_transport = 10 assert app.producer_transport == 10 diff --git a/tests/unit/cli/test_base.py b/tests/unit/cli/test_base.py index 8b81b1c04..0c72accbf 100644 --- a/tests/unit/cli/test_base.py +++ b/tests/unit/cli/test_base.py @@ -1,3 +1,4 @@ +import asyncio import io import json import os @@ -9,6 +10,7 @@ import click import pytest from mode import Worker +from mode.utils import loops as mode_loops from faust.cli import AppCommand, Command, call_command from faust.cli.base import ( @@ -350,6 +352,44 @@ def test_run_using_worker(self, *, command): ) worker.execute_from_commandline.assert_called_once_with() + def test_run_using_worker_without_a_current_loop(self, *, command): + """The CLI must create its own loop when none is set. + + Nothing sets a current event loop before the CLI runs -- declaring + agents and tables deliberately does not resolve one -- and from Python + 3.14 ``get_event_loop_policy().get_event_loop()`` raises instead of + creating one. Emulated here by clearing the current loop, which makes + every supported version raise on the old code path. + """ + command.as_service = Mock() + command.worker_for_service = Mock() + worker = command.worker_for_service.return_value + worker.execute_from_commandline.side_effect = KeyError() + + # Clearing the current loop is global state, and mode caches its own + # resolved loop in a thread local. Both have to be put back, and any + # loop this test causes to be created has to be closed, or later tests + # inherit a loop that is not running and hang on it. + previous_loop = asyncio.get_event_loop_policy().get_event_loop() + previous_mode_loop = getattr(mode_loops._current_loop, "loop", None) + asyncio.set_event_loop(None) + if hasattr(mode_loops._current_loop, "loop"): + del mode_loops._current_loop.loop + try: + with pytest.raises(KeyError): # i.e. got as far as the worker + command.run_using_worker() + used_loop = command.as_service.call_args[0][0] + assert isinstance(used_loop, asyncio.AbstractEventLoop) + finally: + if used_loop is not previous_loop and not used_loop.is_closed(): + used_loop.close() + if previous_mode_loop is None: + if hasattr(mode_loops._current_loop, "loop"): + del mode_loops._current_loop.loop + else: + mode_loops._current_loop.loop = previous_mode_loop + asyncio.set_event_loop(previous_loop) + def test_on_worker_created(self, *, command): assert command.on_worker_created(Mock(name="worker")) is None diff --git a/tests/unit/contrib/test_fastapi.py b/tests/unit/contrib/test_fastapi.py new file mode 100644 index 000000000..62f3d7183 --- /dev/null +++ b/tests/unit/contrib/test_fastapi.py @@ -0,0 +1,368 @@ +import asyncio +from unittest.mock import Mock, patch + +import pytest + +import faust +from faust.contrib.fastapi import ( + AsgiService, + LoopMismatch, + _disable_signal_handling, + bind_to_running_loop, + faust_app_running, + faust_lifespan, + maybe_instrument_opentelemetry, + serve_asgi, +) +from faust.exceptions import ImproperlyConfigured + + +@pytest.fixture() +def app(): + """An app declared the way a module declares one -- no loop running.""" + return faust.App("test-contrib-fastapi", store="memory://", cache="memory://") + + +def _completed_future(result=None): + """A future that is already done -- awaitable without scheduling a task.""" + fut = asyncio.get_running_loop().create_future() + fut.set_result(result) + return fut + + +def _started(value): + """Stand-in for ``App.maybe_start()``, which returns "did I start it?".""" + return _completed_future(value) + + +class Test_bind_to_running_loop: + def test_requires_a_running_loop(self, *, app): + with pytest.raises(RuntimeError): + bind_to_running_loop(app) + + async def test_binds_unbound_app(self, *, app): + running = asyncio.get_running_loop() + assert app._loop is None + + assert bind_to_running_loop(app) is running + assert app._loop is running + + async def test_is_idempotent(self, *, app): + running = asyncio.get_running_loop() + bind_to_running_loop(app) + + assert bind_to_running_loop(app) is running + assert app._loop is running + + async def test_rejects_a_foreign_loop(self, *, app): + # A Mock, not a real loop: an unclosed loop would trip the + # error::ResourceWarning filter when it is garbage collected. + app.loop = Mock(name="other_loop") + + with pytest.raises(LoopMismatch) as excinfo: + bind_to_running_loop(app) + # The message must name what people actually did wrong. + assert "app.loop" in str(excinfo.value) + + async def test_does_not_pin_while_checking(self, *, app): + """Reading ``app.loop`` would itself bind -- it must read ``_loop``.""" + running = asyncio.get_running_loop() + bind_to_running_loop(app) + assert app._loop is running + + +class Test_faust_app_running: + async def test_starts_and_stops(self, *, app): + app.maybe_start = Mock(side_effect=lambda: _started(True)) + app.stop = Mock(side_effect=lambda: _completed_future()) + + async with faust_app_running(app, discover=False) as running_app: + assert running_app is app + assert app._loop is asyncio.get_running_loop() + app.stop.assert_not_called() + + app.maybe_start.assert_called_once_with() + app.stop.assert_called_once_with() + + async def test_does_not_stop_an_app_it_did_not_start(self, *, app): + app.maybe_start = Mock(side_effect=lambda: _started(False)) + app.stop = Mock(side_effect=lambda: _completed_future()) + + async with faust_app_running(app, discover=False): + pass + + app.stop.assert_not_called() + + async def test_finalizes_by_default(self, *, app): + app.maybe_start = Mock(side_effect=lambda: _started(False)) + app.finalize = Mock() + + async with faust_app_running(app, discover=False): + pass + + app.finalize.assert_called_once_with() + + async def test_finalize_can_be_disabled(self, *, app): + app.maybe_start = Mock(side_effect=lambda: _started(False)) + app.finalize = Mock() + + async with faust_app_running(app, finalize=False, discover=False): + pass + + app.finalize.assert_not_called() + + async def test_discovers_when_asked(self, *, app): + app.maybe_start = Mock(side_effect=lambda: _started(False)) + app.discover = Mock() + + async with faust_app_running(app, discover=True): + pass + + app.discover.assert_called_once_with() + + async def test_discover_defaults_to_the_app_setting(self, *, app): + app.maybe_start = Mock(side_effect=lambda: _started(False)) + app.discover = Mock() + + async with faust_app_running(app): + pass + + # This app is not configured with autodiscover. + app.discover.assert_not_called() + + async def test_stop_timeout_is_honoured(self, *, app): + app.maybe_start = Mock(side_effect=lambda: _started(True)) + app.stop = Mock(side_effect=lambda: _completed_future()) + + with patch("asyncio.wait_for", side_effect=asyncio.wait_for) as wait_for: + async with faust_app_running(app, discover=False, stop_timeout=30): + pass + + assert wait_for.call_args.kwargs["timeout"] == 30 + app.stop.assert_called_once_with() + + async def test_slow_stop_raises_when_it_exceeds_the_timeout(self, *, app): + async def never_stops(): + await asyncio.Event().wait() + + app.maybe_start = Mock(side_effect=lambda: _started(True)) + app.stop = Mock(side_effect=never_stops) + + with pytest.raises(asyncio.TimeoutError): + async with faust_app_running(app, discover=False, stop_timeout=0.01): + pass + + async def test_propagates_loop_mismatch(self, *, app): + app.loop = Mock(name="other_loop") + + with pytest.raises(LoopMismatch): + async with faust_app_running(app, discover=False): + pass # pragma: no cover + + +class Test_faust_lifespan: + async def test_yields_none_for_starlette(self, *, app): + """Starlette merges a non-None lifespan value into the ASGI scope.""" + app.maybe_start = Mock(side_effect=lambda: _started(False)) + lifespan = faust_lifespan(app, discover=False) + + async with lifespan(Mock(name="asgi_app")) as value: + assert value is None + + async def test_starts_the_app(self, *, app): + app.maybe_start = Mock(side_effect=lambda: _started(True)) + app.stop = Mock(side_effect=lambda: _completed_future()) + lifespan = faust_lifespan(app, discover=False) + + async with lifespan(Mock(name="asgi_app")): + app.maybe_start.assert_called_once_with() + app.stop.assert_called_once_with() + + async def test_works_without_an_asgi_app_argument(self, *, app): + """Not every ASGI server passes the app to the lifespan handler.""" + app.maybe_start = Mock(side_effect=lambda: _started(False)) + lifespan = faust_lifespan(app, discover=False) + + async with lifespan(): + app.maybe_start.assert_called_once_with() + + +class Test_serve_asgi: + def test_registers_an_extra_service(self, *, app): + asgi_app = Mock(name="asgi_app") + + cls = serve_asgi(app, asgi_app, host="127.0.0.1", port=9001, workers=2) + + assert issubclass(cls, AsgiService) + assert cls.asgi_app is asgi_app + assert cls.host == "127.0.0.1" + assert cls.port == 9001 + assert cls.uvicorn_options == {"workers": 2} + assert cls in app._extra_services + + def test_defaults(self, *, app): + cls = serve_asgi(app, Mock(name="asgi_app")) + + assert cls.port == 8000 + assert cls.uvicorn_options == {} + + +class Test_AsgiService: + def test_init_overrides_class_attributes(self): + asgi_app = Mock(name="asgi_app") + service = AsgiService( + asgi_app, host="1.2.3.4", port=99, uvicorn_options={"a": 1} + ) + + assert service.asgi_app is asgi_app + assert service.host == "1.2.3.4" + assert service.port == 99 + assert service.uvicorn_options == {"a": 1} + + def test_label(self): + service = AsgiService(Mock(), host="1.2.3.4", port=99) + + assert "1.2.3.4" in service.label + assert "99" in service.label + + async def test_on_start_requires_an_app(self): + service = AsgiService() + + with pytest.raises(ImproperlyConfigured): + await service.on_start() + + async def test_on_start_serves(self): + service = AsgiService(Mock(name="asgi_app")) + server = Mock(name="server") + server.serve = Mock(return_value=_completed_future()) + service._create_server = Mock(return_value=server) + # Patch add_future so no task is created and the lingering-task guard + # in tests/conftest.py stays happy. + service.add_future = Mock(side_effect=lambda coro: coro) + + await service.on_start() + + service._create_server.assert_called_once_with() + server.serve.assert_called_once_with() + assert service._server is server + + async def test_on_stop_asks_the_server_to_exit(self): + service = AsgiService(Mock(name="asgi_app")) + server = Mock(name="server") + server.serve = Mock(return_value=_completed_future()) + service._create_server = Mock(return_value=server) + service.add_future = Mock(side_effect=lambda coro: coro) + + await service.on_start() + await service.on_stop() + + assert server.should_exit is True + assert service._server is None + assert service._serve_fut is None + + async def test_on_stop_is_safe_when_never_started(self): + service = AsgiService(Mock(name="asgi_app")) + + await service.on_stop() # must not raise + + async def test_on_stop_warns_but_does_not_hang(self): + service = AsgiService(Mock(name="asgi_app")) + server = Mock(name="server") + never = asyncio.get_running_loop().create_future() + server.serve = Mock(return_value=never) + service._create_server = Mock(return_value=server) + service.add_future = Mock(side_effect=lambda coro: coro) + service.server_shutdown_timeout = 0.01 + + await service.on_start() + await service.on_stop() + + assert server.should_exit is True + never.cancel() + + +class Test_disable_signal_handling: + def test_neutralizes_capture_signals(self): + server = Mock(name="server", spec=["capture_signals"]) + + _disable_signal_handling(server) + + with server.capture_signals(): + pass # must be a no-op context manager + + def test_neutralizes_legacy_install_signal_handlers(self): + server = Mock(name="server", spec=["install_signal_handlers"]) + + _disable_signal_handling(server) + + assert server.install_signal_handlers() is None + + def test_create_server_applies_it_to_a_real_uvicorn_server(self): + """The override must actually reach the server ``on_start`` builds. + + Testing ``_disable_signal_handling`` in isolation would not catch + ``_create_server`` forgetting to call it. + """ + import signal + + uvicorn = pytest.importorskip("uvicorn") + + service = AsgiService(Mock(name="asgi_app"), port=0) + server = service._create_server() + + assert isinstance(server, uvicorn.Server) + before = signal.getsignal(signal.SIGINT) + with server.capture_signals(): + # A server left to itself would have replaced the SIGINT handler + # here; ours must leave mode.Worker's in place. + assert signal.getsignal(signal.SIGINT) is before + assert signal.getsignal(signal.SIGINT) is before + + +class Test_maybe_instrument_opentelemetry: + def test_opt_out(self): + assert maybe_instrument_opentelemetry(Mock(name="asgi_app"), False) is False + + def test_delegates_to_the_contrib_module(self): + asgi_app = Mock(name="asgi_app") + with patch( + "faust.contrib.opentelemetry.instrument_asgi_app", return_value=True + ) as instrument: + assert maybe_instrument_opentelemetry(asgi_app, None) is True + + instrument.assert_called_once_with(asgi_app, force=False) + + def test_force(self): + asgi_app = Mock(name="asgi_app") + with patch( + "faust.contrib.opentelemetry.instrument_asgi_app", return_value=True + ) as instrument: + maybe_instrument_opentelemetry(asgi_app, True) + + instrument.assert_called_once_with(asgi_app, force=True) + + async def test_lifespan_instruments_the_asgi_app(self, *, app): + app.maybe_start = Mock(side_effect=lambda: _started(False)) + asgi_app = Mock(name="asgi_app") + lifespan = faust_lifespan(app, discover=False) + + with patch( + "faust.contrib.fastapi.maybe_instrument_opentelemetry" + ) as instrument: + async with lifespan(asgi_app): + pass + + instrument.assert_called_once_with(asgi_app, None) + + async def test_lifespan_opt_out_is_propagated(self, *, app): + app.maybe_start = Mock(side_effect=lambda: _started(False)) + asgi_app = Mock(name="asgi_app") + lifespan = faust_lifespan(app, discover=False, opentelemetry=False) + + with patch( + "faust.contrib.fastapi.maybe_instrument_opentelemetry" + ) as instrument: + async with lifespan(asgi_app): + pass + + instrument.assert_called_once_with(asgi_app, False) diff --git a/tests/unit/contrib/test_opentelemetry.py b/tests/unit/contrib/test_opentelemetry.py new file mode 100644 index 000000000..a773c34d8 --- /dev/null +++ b/tests/unit/contrib/test_opentelemetry.py @@ -0,0 +1,345 @@ +import sys +import types +from unittest.mock import Mock, patch + +import pytest + +import faust +from faust.types import TP + +pytest.importorskip("opentelemetry") + +from opentelemetry import propagate, 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 faust.contrib.opentelemetry import ( # noqa: E402 + OpenTelemetrySensor, + _build_getter, + _kafka_headers_as_list, + instrument_asgi_app, + opentelemetry_available, + sdk_is_configured, + setup_opentelemetry, +) + +TOPIC = "greetings" +TP1 = TP(TOPIC, 3) + + +@pytest.fixture() +def exporter(): + return InMemorySpanExporter() + + +@pytest.fixture() +def provider(exporter): + """A private TracerProvider -- the global one can only be set once.""" + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + return provider + + +@pytest.fixture() +def sensor(provider): + return OpenTelemetrySensor(tracer=provider.get_tracer("test")) + + +@pytest.fixture() +def app(): + return faust.App("test-contrib-otel", store="memory://", cache="memory://") + + +@pytest.fixture() +def stream(app): + return Mock(name="stream", app=app) + + +def _event(headers=None, key=None): + event = Mock(name="event") + event.message = Mock(headers=headers, key=key) + return event + + +def _traceparent_headers(provider): + """Produce Kafka headers carrying a real upstream span context.""" + carrier = {} + tracer = provider.get_tracer("producer") + with tracer.start_as_current_span("upstream send") as span: + propagate.inject(carrier) + trace_id = span.get_span_context().trace_id + return [(k, v.encode()) for k, v in carrier.items()], trace_id + + +class Test_kafka_headers_as_list: + def test_none_and_empty(self): + assert _kafka_headers_as_list(None) == [] + assert _kafka_headers_as_list([]) == [] + + def test_passes_through_a_list(self): + headers = [("traceparent", b"x")] + assert _kafka_headers_as_list(headers) == headers + + def test_normalizes_a_mapping(self): + assert _kafka_headers_as_list({"traceparent": b"x"}) == [("traceparent", b"x")] + + +class Test_getter: + def test_decodes_bytes(self): + getter = _build_getter() + assert getter.get([("traceparent", b"abc")], "traceparent") == ["abc"] + + def test_missing_key_is_none(self): + getter = _build_getter() + assert getter.get([("other", b"abc")], "traceparent") is None + + def test_ignores_null_values(self): + getter = _build_getter() + assert getter.get([("traceparent", None)], "traceparent") is None + + def test_first_match_wins(self): + getter = _build_getter() + headers = [("traceparent", b"first"), ("traceparent", b"second")] + assert getter.get(headers, "traceparent") == ["first"] + + def test_keys(self): + getter = _build_getter() + assert getter.keys([("a", b"1"), ("b", b"2")]) == ["a", "b"] + + def test_works_with_a_mapping(self): + getter = _build_getter() + assert getter.get({"traceparent": b"abc"}, "traceparent") == ["abc"] + + def test_accepts_non_bytes_values(self): + """Some clients hand back ``str`` headers rather than ``bytes``.""" + getter = _build_getter() + assert getter.get([("traceparent", "already-a-str")], "traceparent") == [ + "already-a-str" + ] + + +class Test_OpenTelemetrySensor: + def test_creates_a_process_span(self, *, sensor, stream, exporter): + state = sensor.on_stream_event_in(TP1, 42, stream, _event()) + sensor.on_stream_event_out(TP1, 42, stream, _event(), state) + + (span,) = exporter.get_finished_spans() + assert span.name == f"{TOPIC} process" + assert span.kind is trace.SpanKind.CONSUMER + + def test_semantic_attributes(self, *, sensor, stream, exporter, app): + state = sensor.on_stream_event_in(TP1, 42, stream, _event(key=b"k1")) + sensor.on_stream_event_out(TP1, 42, stream, _event(), state) + + (span,) = exporter.get_finished_spans() + assert span.attributes["messaging.system"] == "kafka" + assert span.attributes["messaging.operation.type"] == "process" + assert span.attributes["messaging.destination.name"] == TOPIC + assert span.attributes["messaging.destination.partition.id"] == "3" + assert span.attributes["messaging.kafka.message.offset"] == 42 + assert span.attributes["messaging.consumer.group.name"] == app.conf.id + assert span.attributes["messaging.kafka.message.key"] == "k1" + + def test_null_key_is_omitted(self, *, sensor, stream, exporter): + """The spec says the key attribute MUST NOT be set when key is null.""" + state = sensor.on_stream_event_in(TP1, 42, stream, _event(key=None)) + sensor.on_stream_event_out(TP1, 42, stream, _event(), state) + + (span,) = exporter.get_finished_spans() + assert "messaging.kafka.message.key" not in span.attributes + + def test_continues_the_trace_from_message_headers( + self, *, sensor, stream, exporter, provider + ): + """This is the gap the aiokafka instrumentation cannot close.""" + headers, upstream_trace_id = _traceparent_headers(provider) + + state = sensor.on_stream_event_in(TP1, 42, stream, _event(headers=headers)) + sensor.on_stream_event_out(TP1, 42, stream, _event(), state) + + spans = {s.name: s for s in exporter.get_finished_spans()} + process = spans[f"{TOPIC} process"] + assert process.context.trace_id == upstream_trace_id + assert process.parent is not None + + def test_span_is_current_while_processing( + self, *, sensor, stream, exporter, provider + ): + """Work done by the agent must nest under the process span.""" + state = sensor.on_stream_event_in(TP1, 42, stream, _event()) + with provider.get_tracer("agent").start_as_current_span("db query"): + pass + sensor.on_stream_event_out(TP1, 42, stream, _event(), state) + + spans = {s.name: s for s in exporter.get_finished_spans()} + assert ( + spans["db query"].parent.span_id + == spans[f"{TOPIC} process"].context.span_id + ) + + def test_context_is_detached_afterwards(self, *, sensor, stream, provider): + state = sensor.on_stream_event_in(TP1, 42, stream, _event()) + sensor.on_stream_event_out(TP1, 42, stream, _event(), state) + + assert trace.get_current_span() is trace.INVALID_SPAN + + def test_missing_state_is_ignored(self, *, sensor, stream, exporter): + """``Stream.ack()`` calls the hook without sensor state.""" + sensor.on_stream_event_out(TP1, 42, stream, _event(), None) + + assert exporter.get_finished_spans() == () + + def test_out_is_idempotent(self, *, sensor, stream, exporter): + state = sensor.on_stream_event_in(TP1, 42, stream, _event()) + sensor.on_stream_event_out(TP1, 42, stream, _event(), state) + sensor.on_stream_event_out(TP1, 42, stream, _event(), state) + + assert len(exporter.get_finished_spans()) == 1 + + def test_state_without_a_span_or_token(self, *, sensor, stream, exporter): + """A truthy state that carries neither must not raise.""" + sensor.on_stream_event_out(TP1, 42, stream, _event(), {"other": 1}) + + assert exporter.get_finished_spans() == () + + def test_consumer_group_is_omitted_when_unknown(self, *, sensor, exporter): + """Sensors are not given the app; the stream may not expose one.""" + state = sensor.on_stream_event_in(TP1, 42, Mock(spec=[]), _event()) + sensor.on_stream_event_out(TP1, 42, Mock(spec=[]), _event(), state) + + (span,) = exporter.get_finished_spans() + assert "messaging.consumer.group.name" not in span.attributes + + def test_garbage_headers_do_not_break_processing(self, *, sensor, stream, exporter): + event = _event(headers=[("traceparent", b"not-a-traceparent")]) + + state = sensor.on_stream_event_in(TP1, 42, stream, event) + sensor.on_stream_event_out(TP1, 42, stream, event, state) + + # Still traced, just not continuing a (nonexistent) upstream trace. + (span,) = exporter.get_finished_spans() + assert span.name == f"{TOPIC} process" + + def test_tracer_defaults_to_the_global_provider(self): + sensor = OpenTelemetrySensor() + assert sensor.tracer is not None + + +class Test_sdk_is_configured: + def test_true_for_a_real_provider(self, *, provider): + with patch.object(trace, "get_tracer_provider", return_value=provider): + assert sdk_is_configured() is True + + def test_false_for_the_proxy_provider(self): + proxy = Mock() + type(proxy).__name__ = "ProxyTracerProvider" + with patch.object(trace, "get_tracer_provider", return_value=proxy): + assert sdk_is_configured() is False + + def test_api_is_available(self): + assert opentelemetry_available() is True + + +class Test_instrument_asgi_app: + def test_none_app(self): + assert instrument_asgi_app(None) is False + + def test_skips_an_already_instrumented_app(self): + asgi_app = Mock(_is_instrumented_by_opentelemetry=True) + + assert instrument_asgi_app(asgi_app, force=True) is False + + def test_skips_when_no_sdk_is_configured(self): + proxy = Mock() + type(proxy).__name__ = "ProxyTracerProvider" + asgi_app = Mock(_is_instrumented_by_opentelemetry=False) + with patch.object(trace, "get_tracer_provider", return_value=proxy): + assert instrument_asgi_app(asgi_app) is False + + def test_instruments_when_forced(self, *, provider): + pytest.importorskip("opentelemetry.instrumentation.fastapi") + fastapi = pytest.importorskip("fastapi") + api = fastapi.FastAPI() + + assert instrument_asgi_app(api, tracer_provider=provider, force=True) is True + assert api._is_instrumented_by_opentelemetry is True + # Second call is a no-op, not a duplicate middleware stack. + assert instrument_asgi_app(api, tracer_provider=provider, force=True) is False + + # The tests below stub the instrumentation package through sys.modules, so + # they exercise our own branching everywhere -- including CI, where + # opentelemetry-instrumentation-fastapi is deliberately not installed (it + # would drag fastapi and pydantic-core into all 15 matrix legs). + + @staticmethod + def _fake_instrumentation(instrument_app): + module = types.ModuleType("opentelemetry.instrumentation.fastapi") + module.FastAPIInstrumentor = Mock(instrument_app=instrument_app) + return {"opentelemetry.instrumentation.fastapi": module} + + def test_returns_false_when_the_package_is_missing(self, *, provider): + # A None entry in sys.modules makes the import raise ImportError. + with patch.dict(sys.modules, {"opentelemetry.instrumentation.fastapi": None}): + assert instrument_asgi_app(Mock(spec=[]), force=True) is False + + def test_instruments_via_the_instrumentor(self, *, provider): + instrument_app = Mock() + asgi_app = Mock(spec=[]) + with patch.dict(sys.modules, self._fake_instrumentation(instrument_app)): + assert ( + instrument_asgi_app(asgi_app, tracer_provider=provider, force=True) + is True + ) + + instrument_app.assert_called_once_with( + asgi_app, tracer_provider=provider, exclude_spans=["receive", "send"] + ) + + def test_retries_without_exclude_spans_on_older_releases(self, *, provider): + """``exclude_spans`` was added after the initial release.""" + instrument_app = Mock(side_effect=[TypeError("unexpected kwarg"), None]) + asgi_app = Mock(spec=[]) + with patch.dict(sys.modules, self._fake_instrumentation(instrument_app)): + assert instrument_asgi_app(asgi_app, force=True) is True + + assert instrument_app.call_count == 2 + assert "exclude_spans" not in instrument_app.call_args.kwargs + + def test_gives_up_when_the_retry_also_fails(self, *, provider): + instrument_app = Mock(side_effect=[TypeError("nope"), RuntimeError("nope")]) + with patch.dict(sys.modules, self._fake_instrumentation(instrument_app)): + assert instrument_asgi_app(Mock(spec=[]), force=True) is False + + def test_never_raises_out_of_instrumentation(self, *, provider): + """Telemetry setup must not take the application down.""" + instrument_app = Mock(side_effect=RuntimeError("boom")) + with patch.dict(sys.modules, self._fake_instrumentation(instrument_app)): + assert instrument_asgi_app(Mock(spec=[]), force=True) is False + + +class Test_setup_opentelemetry: + def test_registers_the_sensor(self, *, app): + sensor = setup_opentelemetry(app) + + assert isinstance(sensor, OpenTelemetrySensor) + assert sensor in list(app.sensors) + + def test_returns_none_without_opentelemetry(self, *, app): + with patch( + "faust.contrib.opentelemetry.opentelemetry_available", return_value=False + ): + assert setup_opentelemetry(app) is None + + def test_warns_when_opentracing_sensor_is_registered(self, *, app): + from faust.sensors.distributed_tracing import TracingSensor + + app.sensors.add(TracingSensor()) + + with patch("faust.contrib.opentelemetry.logger") as logger: + setup_opentelemetry(app) + + assert logger.warning.called + assert "traceparent" in logger.warning.call_args[0][0] diff --git a/tests/unit/test_packaging.py b/tests/unit/test_packaging.py new file mode 100644 index 000000000..d59ea78df --- /dev/null +++ b/tests/unit/test_packaging.py @@ -0,0 +1,56 @@ +"""Guard the hand-maintained parts of ``setup.py``. + +``extras_require()`` is driven entirely by the ``BUNDLES`` set, so a +``requirements/extras/.txt`` file whose name is missing from ``BUNDLES`` +is a dead extra: ``pip install faust-streaming[]`` silently installs +nothing. That is exactly what happened to ``aerospike``, which was advertised +in the README while resolving to an empty list. +""" + +import ast +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +SETUP_PY = ROOT / "setup.py" +EXTRAS_DIR = ROOT / "requirements" / "extras" + + +def _bundles_from_setup_py(): + """Read ``BUNDLES`` without importing (and thus running) ``setup.py``.""" + tree = ast.parse(SETUP_PY.read_text()) + for node in tree.body: + if isinstance(node, ast.Assign) and any( + isinstance(t, ast.Name) and t.id == "BUNDLES" for t in node.targets + ): + return set(ast.literal_eval(node.value)) + raise AssertionError("BUNDLES not found in setup.py") + + +@pytest.mark.skipif( + not SETUP_PY.exists(), reason="running against an installed package, not a checkout" +) +def test_every_extras_file_is_declared_in_bundles(): + bundles = _bundles_from_setup_py() + on_disk = {path.stem for path in EXTRAS_DIR.glob("*.txt")} + + undeclared = on_disk - bundles + assert not undeclared, ( + f"requirements/extras/{{{','.join(sorted(undeclared))}}}.txt exist but are " + f"missing from BUNDLES in setup.py, so those extras install nothing" + ) + + +@pytest.mark.skipif( + not SETUP_PY.exists(), reason="running against an installed package, not a checkout" +) +def test_every_bundle_has_a_requirements_file(): + bundles = _bundles_from_setup_py() + on_disk = {path.stem for path in EXTRAS_DIR.glob("*.txt")} + + missing = bundles - on_disk + assert not missing, ( + f"BUNDLES declares {sorted(missing)} but " + f"requirements/extras/.txt is missing for them" + ) diff --git a/tests/unit/web/test_blueprint_flags.py b/tests/unit/web/test_blueprint_flags.py new file mode 100644 index 000000000..88eeabb2a --- /dev/null +++ b/tests/unit/web/test_blueprint_flags.py @@ -0,0 +1,104 @@ +"""Which built-in blueprints get served, for each combination of flags. + +Before these settings existed the choice was driven entirely by +:setting:`debug`: ``/router`` and ``/table`` were always on, and ``debug`` +swapped the production index for the statistics and graph endpoints. The +back-compat cases below pin that exact behaviour for apps that set none of +the new settings. +""" + +import pytest + +import faust + +ROUTER = "faust.web.apps.router:blueprint" +TABLES = "faust.web.apps.tables.blueprint" +GRAPH = "faust.web.apps.graph:blueprint" +STATS = "faust.web.apps.stats:blueprint" +METRICS = "faust.web.apps.metrics:blueprint" +INDEX = "faust.web.apps.production_index:blueprint" + + +def enabled(**settings): + app = faust.App("t-flags", store="memory://", cache="memory://", **settings) + return {bp for _prefix, bp in app.web._enabled_blueprints()} + + +class Test_backwards_compatibility: + def test_defaults_match_the_old_non_debug_behaviour(self): + assert enabled() == {ROUTER, TABLES, INDEX} + + def test_debug_matches_the_old_debug_behaviour(self): + assert enabled(debug=True) == {ROUTER, TABLES, GRAPH, STATS} + + def test_metrics_is_off_by_default(self): + assert METRICS not in enabled() + assert METRICS not in enabled(debug=True) + + +class Test_flags: + @pytest.mark.parametrize( + "setting,blueprint", + [ + ("web_router_enabled", ROUTER), + ("web_tables_enabled", TABLES), + ], + ) + def test_on_by_default_and_can_be_turned_off(self, setting, blueprint): + assert blueprint in enabled() + assert blueprint not in enabled(**{setting: False}) + + @pytest.mark.parametrize( + "setting,blueprint", + [ + ("web_graph_enabled", GRAPH), + ("web_stats_enabled", STATS), + ("web_metrics_enabled", METRICS), + ], + ) + def test_off_by_default_and_can_be_turned_on(self, setting, blueprint): + assert blueprint not in enabled() + assert blueprint in enabled(**{setting: True}) + + def test_graph_and_stats_are_independent(self): + """The point of the change: debug used to enable both or neither.""" + only_graph = enabled(debug=True, web_stats_enabled=False) + assert GRAPH in only_graph + assert STATS not in only_graph + + only_stats = enabled(debug=True, web_graph_enabled=False) + assert STATS in only_stats + assert GRAPH not in only_stats + + def test_router_and_tables_can_be_locked_down(self): + """Neither could be disabled at all before these settings.""" + assert enabled(web_router_enabled=False, web_tables_enabled=False) == {INDEX} + + +class Test_root_path: + """Stats and the production index both mount at "/" -- exactly one wins.""" + + def test_index_serves_root_when_stats_is_off(self): + served = enabled(web_stats_enabled=False) + assert INDEX in served + assert STATS not in served + + def test_stats_replaces_the_index(self): + served = enabled(web_stats_enabled=True) + assert STATS in served + assert INDEX not in served + + def test_root_is_never_unserved(self): + for settings in ({}, {"debug": True}, {"web_stats_enabled": True}): + served = enabled(**settings) + assert (STATS in served) != (INDEX in served) + + +class Test_unknown_blueprints: + def test_blueprints_without_a_flag_are_always_enabled(self): + """A subclass adding blueprints must not need to register a flag.""" + app = faust.App("t-extra", store="memory://", cache="memory://") + web = app.web + web.optional_blueprints = [("/custom", "proj.web:blueprint")] + + assert "proj.web:blueprint" in {bp for _p, bp in web._enabled_blueprints()}