Simplify lifecycle and telemetry ownership for 2.0 (#444)

* Keep FastStream observability alive through shutdown (#434)

* Simplify lifecycle and telemetry ownership for 2.0

Implements #434, #435, #436, #438 and #440, and clarifies conflicting configuration arguments.

* Roll back OTel acquisitions and preserve borrowed FastAPI instrumentation

* Preserve request handling and borrowed tracing during teardown

Author: lesnik512

docs/adr/0002-cross-cutting-logic-stays-put.md

@@ -13,11 +13,8 @@ framework-locality already exists, because all of a framework's subclasses live
 bootstrapper file. Reconsider if the per-cell bindings start converging, or if a new framework shares
 an existing one's attach mechanism exactly.
 
-`_build_excluded_urls` likewise keeps the whole trace-exclusion policy in one method that reads
-its siblings' paths through `getattr(..., None)`. Those sibling configs are optional, so the
-defensive read is correct. A per-instrument contribution mechanism would
-spread the policy across `PrometheusConfig`, `HealthChecksConfig` and OpenTelemetry while still
-needing `opentelemetry_generate_health_check_spans`, so the coupling would not even disappear. The
-real risk, renaming `prometheus_metrics_path` and silently stopping the exclusion, is covered by a
-test pinning each sibling path in the built set. Reconsider when a user-supplied instrument needs
-paths the policy cannot know about.
+OpenTelemetry's `_build_infrastructure_excluded_paths` owns the exclusion policy for
+the metrics route and the health route. Framework bindings apply those paths alongside
+caller-supplied URL patterns. HTTP-level tests verify the resulting exclusions.
+A per-instrument contribution mechanism would distribute the same policy across
+configuration classes while retaining their dependency on the health span setting.

docs/integrations/faststream.md

@@ -51,6 +51,15 @@ application = bootstrapper.bootstrap()
 
 Read more about available configuration options [here](../introduction/configuration.md).
 
+## Lifecycle
+
+The ASGI lifespan runs `bootstrapper.teardown()` after broker shutdown and user lifespan
+cleanup. Tracing and logging remain active through both. Startup and shutdown failures still
+trigger teardown; cleanup failures are added as notes to the original error.
+
+When using `start()` and `stop()` directly or `async with application`, call
+`bootstrapper.teardown()` in an outer `finally` block after the application stops.
+
 ## Health and metrics routes
 
 When configured, FastStream health and Prometheus metrics routes accept requests with or without

docs/introduction/configuration.md

@@ -183,7 +183,7 @@ Additional parameters for FastAPI integration:
 
 - `prometheus_instrumentator_additional_params` - passed to `prometheus_fastapi_instrumentator.Instrumentator`
 - `prometheus_instrument_additional_params` - passed to `method Instrumentator(...).instrument`
-- `prometheus_expose_additional_params` - passed to `method Instrumentator(...).expose`.
+- `prometheus_expose_additional_params` - passed to `Instrumentator(...).expose`. Set the endpoint with `prometheus_metrics_path`.
 
 ### Prometheus FastMCP
 
@@ -343,7 +343,7 @@ Additional parameters:
 
 - `pyroscope_sample_rate` - CPU profiling sample rate in Hz (default: `100`).
 - `pyroscope_tags` - key/value string pairs attached to all profiles.
-- `pyroscope_additional_params` - additional params passed directly to `pyroscope.configure`.
+- `pyroscope_additional_params` - overrides the prepared arguments passed to `pyroscope.configure`.
 
 When OpenTelemetry is also enabled, a `PyroscopeSpanProcessor` is automatically added to the tracer provider. It tags root spans with a `pyroscope.profile.id` attribute and sets Pyroscope thread tags so that traces and profiles can be linked in the Grafana UI.
 
@@ -511,6 +511,8 @@ To bootstrap swagger, you have the following parameters:
 - `swagger_offline_docs` - option to turn on offline docs.
 - `swagger_additional_params` - Litestar only: extra keyword arguments for `litestar.openapi.OpenAPIConfig`. It cannot repeat `path`, `title`, `version`, `description` or `render_plugins`, which lite-bootstrap already passes.
 
+When lite-bootstrap creates the FastAPI application, set its docs URL with `swagger_path`.
+
 Swagger is on by default at `swagger_path` (default: `/docs`). For Litestar, set `swagger_path` to an empty string to disable it.
 
 ## Health checks

docs/migration/to-2.x.md

@@ -397,6 +397,12 @@ spans and logs. Keep exporters open until that work finishes
 ([#412](https://github.com/modern-python/lite-bootstrap/pull/412),
 [#413](https://github.com/modern-python/lite-bootstrap/pull/413)).
 
+FastStream teardown now runs after broker shutdown and user lifespan shutdown,
+including when startup or shutdown fails. If you drive FastStream with
+`start()` and `stop()` or use `async with application`, explicitly call
+`bootstrapper.teardown()` in an outer `finally` block
+([#434](https://github.com/modern-python/lite-bootstrap/issues/434)).
+
 Failed bootstrap rolls back successfully started instruments and cannot be reused. Retry with
 a fresh application and bootstrapper ([#406](https://github.com/modern-python/lite-bootstrap/pull/406)).
 

lite_bootstrap/bootstrappers/base.py

@@ -1,7 +1,9 @@
 import abc
+import contextlib
 import logging
 import typing
 import warnings
+from collections.abc import Generator
 
 from lite_bootstrap.exceptions import (
     BootstrapperNotReadyError,
@@ -161,6 +163,22 @@ def bootstrap(self) -> ApplicationT:
         self.is_bootstrapped = True
         return self._application
 
+    @contextlib.contextmanager
+    def _teardown_after_lifecycle(self) -> Generator[None]:
+        lifecycle_error: BaseException | None = None
+        try:
+            yield
+        except BaseException as error:
+            lifecycle_error = error
+            raise
+        finally:
+            try:
+                self.teardown()
+            except Exception as teardown_error:
+                if lifecycle_error is None:
+                    raise
+                lifecycle_error.add_note(f"Error tearing down observability: {teardown_error}")
+
     def teardown(self) -> None:
         if not self.is_bootstrapped:
             return

lite_bootstrap/bootstrappers/fastapi_bootstrapper.py

@@ -2,11 +2,12 @@
 import dataclasses
 import time
 import typing
+import weakref
 from collections.abc import AsyncGenerator, Mapping
 
 from lite_bootstrap import import_checker
 from lite_bootstrap.bootstrappers.base import BaseBootstrapper
-from lite_bootstrap.exceptions import ConfigurationError
+from lite_bootstrap.exceptions import ConfigurationError, collect_teardown_errors
 from lite_bootstrap.helpers.fastapi_helpers import enable_offline_docs
 from lite_bootstrap.helpers.warn import warn_at_caller
 from lite_bootstrap.instruments.cors_instrument import CorsConfig, CorsInstrument
@@ -92,6 +93,9 @@ def __post_init__(self) -> None:
             raise ConfigurationError(msg)
 
         if isinstance(self.application, UnsetType):
+            if "docs_url" in self.application_kwargs:
+                msg = "Use swagger_path to configure docs_url when lite-bootstrap creates the FastAPI application"
+                raise ConfigurationError(msg)
             application = fastapi.FastAPI(docs_url=self.swagger_path, **self.application_kwargs)
             # FastAPIConfig stays frozen for user-facing immutability; __post_init__ needs
             # to set application after construction, so we bypass the freeze here.
@@ -208,34 +212,70 @@ def bootstrap(self) -> None:
 
 @dataclasses.dataclass(kw_only=True)
 class FastAPIOpenTelemetryInstrument(OpenTelemetryInstrument[FastAPIConfig]):
-    def bootstrap(self) -> None:
-        super().bootstrap()
-        FastAPIInstrumentor.instrument_app(
-            app=self.bootstrap_config.app,
-            tracer_provider=self.tracer_provider,
-            meter_provider=get_meter_provider(),
-            excluded_urls=",".join(self._build_excluded_url_patterns("FASTAPI")),
-            exclude_spans=self.bootstrap_config.opentelemetry_exclude_spans,
-        )
-        self._install_infrastructure_exclusions()
+    _owns_app_instrumentation: bool = dataclasses.field(default=False, init=False, repr=False, compare=False)
+    _restore_infrastructure_exclusions: typing.Callable[[], None] | None = dataclasses.field(
+        default=None, init=False, repr=False, compare=False
+    )
 
-    def _install_infrastructure_exclusions(self) -> None:
+    def _bootstrap_telemetry(self) -> None:
+        super()._bootstrap_telemetry()
+        app = self.bootstrap_config.app
+        if not getattr(app, "_is_instrumented_by_opentelemetry", False):
+            FastAPIInstrumentor.instrument_app(
+                app=app,
+                tracer_provider=self.tracer_provider,
+                meter_provider=get_meter_provider(),
+                excluded_urls=",".join(self._build_excluded_url_patterns("FASTAPI")),
+                exclude_spans=self.bootstrap_config.opentelemetry_exclude_spans,
+            )
+            self._owns_app_instrumentation = True
+        self._restore_infrastructure_exclusions = self._install_infrastructure_exclusions()
+
+    def _install_infrastructure_exclusions(self) -> typing.Callable[[], None]:
         app = self.bootstrap_config.app
         build_stack = app.build_middleware_stack
         excluded_paths = self._build_infrastructure_excluded_paths()
+        owned_wrappers = weakref.WeakSet[ExcludeInfrastructureTelemetry]()
 
-        def build_middleware_stack() -> "ASGIApp":
-            stack = build_stack()
+        def wrap(stack: "ASGIApp | None") -> None:
             # FastAPIInstrumentor places its OTel middleware inside an outer error handler.
             if isinstance(stack, ServerErrorMiddleware) and isinstance(stack.app, OpenTelemetryMiddleware):
-                stack.app = ExcludeInfrastructureTelemetry(stack.app, excluded_paths)
-            return stack
+                wrapper = ExcludeInfrastructureTelemetry(stack.app, excluded_paths)
+                owned_wrappers.add(wrapper)
+                stack.app = wrapper
 
-        app.build_middleware_stack = build_middleware_stack  # ty: ignore[invalid-assignment]
+        def build_middleware_stack() -> "ASGIApp":
+            stack = build_stack()
+            wrap(stack)
+            return stack
 
-    def teardown(self) -> None:
-        FastAPIInstrumentor.uninstrument_app(self.bootstrap_config.app)
-        super().teardown()
+        def restore() -> None:
+            if app.build_middleware_stack is build_middleware_stack:
+                vars(app)["build_middleware_stack"] = build_stack
+            stack = app.middleware_stack
+            if (
+                isinstance(stack, ServerErrorMiddleware)
+                and isinstance(stack.app, ExcludeInfrastructureTelemetry)
+                and stack.app in owned_wrappers
+            ):
+                stack.app = stack.app.app
+
+        vars(app)["build_middleware_stack"] = build_middleware_stack
+        wrap(app.middleware_stack)
+        return restore
+
+    def _teardown_owned_resources(self) -> None:
+        with collect_teardown_errors() as teardown_errors:
+            if self._restore_infrastructure_exclusions is not None:
+                with teardown_errors.capture("FastAPI infrastructure exclusions"):
+                    self._restore_infrastructure_exclusions()
+                self._restore_infrastructure_exclusions = None
+            if self._owns_app_instrumentation:
+                self._owns_app_instrumentation = False
+                with teardown_errors.capture("FastAPIInstrumentor"):
+                    FastAPIInstrumentor.uninstrument_app(self.bootstrap_config.app)
+            with teardown_errors.capture("OpenTelemetry"):
+                super()._teardown_owned_resources()
 
 
 @dataclasses.dataclass(kw_only=True)
@@ -252,6 +292,9 @@ def dependencies_installed() -> bool:
     def bootstrap(self) -> None:
         config = self.bootstrap_config
         application = config.app
+        if "endpoint" in config.prometheus_expose_additional_params:
+            msg = "Use prometheus_metrics_path to configure the Prometheus endpoint"
+            raise ConfigurationError(msg)
         instrumentator_params = dict(config.prometheus_instrumentator_additional_params)
         instrumentator_params["excluded_handlers"] = [
             *instrumentator_params.get("excluded_handlers", []),
@@ -312,10 +355,8 @@ class FastAPIBootstrapper(BaseBootstrapper["fastapi.FastAPI", FastAPIConfig]):
 
     @contextlib.asynccontextmanager
     async def lifespan_manager(self, _: "fastapi.FastAPI") -> AsyncGenerator[dict[str, typing.Any]]:
-        try:
+        with self._teardown_after_lifecycle():
             yield {}
-        finally:
-            self.teardown()
 
     def __init__(self, bootstrap_config: FastAPIConfig) -> None:
         super().__init__(bootstrap_config)

lite_bootstrap/bootstrappers/fastmcp_bootstrapper.py

@@ -83,7 +83,7 @@ def _postprocess_http_apps(
     def http_app(*args: typing.Any, **kwargs: typing.Any) -> "StarletteWithLifespan":  # noqa: ANN401
         return postprocess(build_http_app(*args, **kwargs))
 
-    application.http_app = http_app  # ty: ignore[invalid-assignment]
+    vars(application)["http_app"] = http_app
 
     def restore() -> None:
         if previous_override is None:
@@ -259,8 +259,8 @@ def _instrument_http_app(self, http_application: "StarletteWithLifespan") -> "St
         setattr(http_application, _OPENTELEMETRY_INSTRUMENTED_MARKER, True)
         return http_application
 
-    def bootstrap(self) -> None:
-        super().bootstrap()
+    def _bootstrap_telemetry(self) -> None:
+        super()._bootstrap_telemetry()
         self._restore_http_app = _postprocess_http_apps(self.bootstrap_config.application, self._instrument_http_app)
 
     def teardown(self) -> None:
@@ -406,11 +406,9 @@ def _wrap_lifespan(self, application: "FastMCP[typing.Any]") -> None:
 
         @contextlib.asynccontextmanager
         async def lifespan(server: "FastMCP[typing.Any]") -> AsyncGenerator[typing.Any]:
-            try:
+            with self._teardown_after_lifecycle():
                 async with user_lifespan(server) as result:
                     yield result
-            finally:
-                self.teardown()
 
         application._lifespan = lifespan  # noqa: SLF001
 

lite_bootstrap/bootstrappers/faststream_bootstrapper.py

@@ -1,9 +1,10 @@
 import asyncio
+import contextlib
 import dataclasses
 import json
 import logging
 import typing
-from collections.abc import Mapping, Sequence
+from collections.abc import AsyncGenerator, Mapping, Sequence
 from types import TracebackType
 
 from lite_bootstrap import import_checker
@@ -219,8 +220,8 @@ def teardown(self) -> None:
 
 @dataclasses.dataclass(kw_only=True)
 class FastStreamOpenTelemetryInstrument(OpenTelemetryInstrument[FastStreamConfig]):
-    def bootstrap(self) -> None:
-        super().bootstrap()
+    def _bootstrap_telemetry(self) -> None:
+        super()._bootstrap_telemetry()
         config = self.bootstrap_config
         if config.opentelemetry_middleware_cls and config.application.brokers:
             for broker, middleware_cls in _middleware_targets(
@@ -280,7 +281,18 @@ def is_ready(self) -> bool:
     def __init__(self, bootstrap_config: FastStreamConfig) -> None:
         super().__init__(bootstrap_config)
         application = self.bootstrap_config.application
-        self._attach_teardown_once(application, lambda: application.on_shutdown(self.teardown))
+        self._attach_teardown_once(application, lambda: self._wrap_lifespan(application))
+
+    def _wrap_lifespan(self, application: "AsgiFastStream") -> None:
+        user_lifespan = application.lifespan_context
+
+        @contextlib.asynccontextmanager
+        async def lifespan(*args: typing.Any, **kwargs: typing.Any) -> AsyncGenerator[typing.Any]:  # noqa: ANN401
+            with self._teardown_after_lifecycle():
+                async with user_lifespan(*args, **kwargs) as result:
+                    yield result
+
+        application.lifespan_context = lifespan
 
     def _prepare_application(self) -> "AsgiFastStream":
         return self.bootstrap_config.application

lite_bootstrap/bootstrappers/litestar_bootstrapper.py

@@ -6,6 +6,7 @@
 
 from lite_bootstrap import import_checker
 from lite_bootstrap.bootstrappers.base import BaseBootstrapper
+from lite_bootstrap.exceptions import ConfigurationError
 from lite_bootstrap.helpers.path import is_valid_path
 from lite_bootstrap.helpers.warn import warn_at_caller
 from lite_bootstrap.instruments.cors_instrument import CorsConfig, CorsInstrument
@@ -375,8 +376,8 @@ def _build_excluded_path_patterns(self) -> list[str]:
         }
         return sorted(anchored_patterns | set(self.bootstrap_config.opentelemetry_excluded_urls))
 
-    def bootstrap(self) -> None:
-        super().bootstrap()
+    def _bootstrap_telemetry(self) -> None:
+        super()._bootstrap_telemetry()
         application_config: typing.Final = self.bootstrap_config.application_config
         application_config.plugins.append(
             OpenTelemetryPlugin(
@@ -457,6 +458,9 @@ def is_configured(cls, bootstrap_config: "LitestarConfig") -> bool:
 
     def bootstrap(self) -> None:
         config = self.bootstrap_config
+        if "path" in config.swagger_additional_params:
+            msg = "Use swagger_path to configure the OpenAPI path"
+            raise ConfigurationError(msg)
         render_plugins: typing.Final = (
             (
                 SwaggerRenderPlugin(

lite_bootstrap/instruments/logging_instrument.py

@@ -192,8 +192,9 @@ def _attach_record_filters(self) -> None:
         for logger_name, record_filters in self.bootstrap_config.logging_record_filters.items():
             target_logger = logging.getLogger(logger_name)
             for record_filter in record_filters:
-                target_logger.addFilter(record_filter)
-                self._attached_filters.append((target_logger, record_filter))
+                if record_filter not in target_logger.filters:
+                    target_logger.addFilter(record_filter)
+                    self._attached_filters.append((target_logger, record_filter))
 
     def _detach_record_filters(self) -> None:
         """Remove only the filters this instrument attached, leaving any others in place."""