diff --git a/docs/adr/0002-cross-cutting-logic-stays-put.md b/docs/adr/0002-cross-cutting-logic-stays-put.md index fcc522a..850bcd6 100644 --- a/docs/adr/0002-cross-cutting-logic-stays-put.md +++ b/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. diff --git a/docs/integrations/faststream.md b/docs/integrations/faststream.md index bdb35bc..d3b1e44 100644 --- a/docs/integrations/faststream.md +++ b/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 diff --git a/docs/introduction/configuration.md b/docs/introduction/configuration.md index 6383f99..25715a9 100644 --- a/docs/introduction/configuration.md +++ b/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 diff --git a/docs/migration/to-2.x.md b/docs/migration/to-2.x.md index 5dc6633..5249662 100644 --- a/docs/migration/to-2.x.md +++ b/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)). diff --git a/lite_bootstrap/bootstrappers/base.py b/lite_bootstrap/bootstrappers/base.py index d15e3a5..b24feff 100644 --- a/lite_bootstrap/bootstrappers/base.py +++ b/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 diff --git a/lite_bootstrap/bootstrappers/fastapi_bootstrapper.py b/lite_bootstrap/bootstrappers/fastapi_bootstrapper.py index 5e05714..d1a5ecc 100644 --- a/lite_bootstrap/bootstrappers/fastapi_bootstrapper.py +++ b/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) diff --git a/lite_bootstrap/bootstrappers/fastmcp_bootstrapper.py b/lite_bootstrap/bootstrappers/fastmcp_bootstrapper.py index dcbadae..57c89e6 100644 --- a/lite_bootstrap/bootstrappers/fastmcp_bootstrapper.py +++ b/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 diff --git a/lite_bootstrap/bootstrappers/faststream_bootstrapper.py b/lite_bootstrap/bootstrappers/faststream_bootstrapper.py index 1657333..9ec91a3 100644 --- a/lite_bootstrap/bootstrappers/faststream_bootstrapper.py +++ b/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 diff --git a/lite_bootstrap/bootstrappers/litestar_bootstrapper.py b/lite_bootstrap/bootstrappers/litestar_bootstrapper.py index 01e2290..d1fad80 100644 --- a/lite_bootstrap/bootstrappers/litestar_bootstrapper.py +++ b/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( diff --git a/lite_bootstrap/instruments/logging_instrument.py b/lite_bootstrap/instruments/logging_instrument.py index 42e67f8..d38d372 100644 --- a/lite_bootstrap/instruments/logging_instrument.py +++ b/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.""" diff --git a/lite_bootstrap/instruments/opentelemetry_instrument.py b/lite_bootstrap/instruments/opentelemetry_instrument.py index 7127ec3..00d052f 100644 --- a/lite_bootstrap/instruments/opentelemetry_instrument.py +++ b/lite_bootstrap/instruments/opentelemetry_instrument.py @@ -1,11 +1,18 @@ +import contextlib import dataclasses import logging import os import typing import urllib.parse +from collections.abc import Callable, Generator from lite_bootstrap import import_checker -from lite_bootstrap.exceptions import ConfigurationError, InstrumentDependencyMissingWarning, collect_teardown_errors +from lite_bootstrap.exceptions import ( + ConfigurationError, + InstrumentDependencyMissingWarning, + TeardownErrorCollector, + collect_teardown_errors, +) from lite_bootstrap.helpers.warn import warn_at_caller from lite_bootstrap.instruments.base import BaseConfig, BaseInstrument @@ -52,6 +59,19 @@ import pyroscope +@contextlib.contextmanager +def _rollback_on_error(shutdown: Callable[[], None]) -> Generator[None]: + try: + yield + except BaseException as error: + teardown_errors = TeardownErrorCollector() + with teardown_errors.capture("OpenTelemetry"): + shutdown() + for name, failure in teardown_errors.errors: + error.add_note(f"Error rolling back {name}: {failure}") + raise + + def _format_span(readable_span: "ReadableSpan") -> str: return typing.cast("str", readable_span.to_json(indent=None)) + "\n" @@ -210,6 +230,9 @@ class OpenTelemetryInstrument(BaseInstrument[OpenTelemetryConfigT]): _meter_provider: "MeterProvider | None" = dataclasses.field( default_factory=lambda: None, init=False, repr=False, compare=False ) + _owned_instrumentors: list[tuple["BaseInstrumentor", dict[str, typing.Any]]] = dataclasses.field( + default_factory=list, init=False, repr=False, compare=False + ) _prior_logger_disabled: dict[str, bool] = dataclasses.field( default_factory=dict, init=False, repr=False, compare=False ) @@ -250,9 +273,6 @@ def _build_infrastructure_excluded_paths(self) -> set[str]: excluded_paths.add(health_path) return excluded_paths - def _build_excluded_urls(self) -> set[str]: - return set(self.bootstrap_config.opentelemetry_excluded_urls) | self._build_infrastructure_excluded_paths() - def _build_excluded_url_patterns(self, environment_key: str) -> list[str]: """Combine caller-supplied URL regexes with the OpenTelemetry environment defaults.""" environment_urls: typing.Final = os.environ.get( @@ -340,70 +360,80 @@ def _bootstrap_metrics(self, resource: "resources.Resource") -> None: metric_exporter = self._build_metric_exporter() if metric_exporter is None: return - meter_provider = MeterProvider( - resource=resource, metric_readers=[PeriodicExportingMetricReader(metric_exporter)] - ) + with _rollback_on_error(metric_exporter.shutdown): + reader = PeriodicExportingMetricReader(metric_exporter) + with _rollback_on_error(reader.shutdown): + meter_provider = MeterProvider(resource=resource, metric_readers=[reader]) + self._meter_provider = meter_provider set_meter_provider(meter_provider) if get_meter_provider() is not meter_provider: warn_at_caller( "a MeterProvider is already installed; the configured metrics exporter and resource will not be used" ) - self._meter_provider = meter_provider def _apply_instrumentors(self, tracer_provider: "APITracerProvider") -> None: - for one_instrumentor in self.bootstrap_config.opentelemetry_instrumentors: - if isinstance(one_instrumentor, InstrumentorWithParams): - one_instrumentor.instrumentor.instrument( - tracer_provider=tracer_provider, - **one_instrumentor.additional_params, - ) + for configured in self.bootstrap_config.opentelemetry_instrumentors: + if isinstance(configured, InstrumentorWithParams): + instrumentor = configured.instrumentor + params = dict(configured.additional_params) else: - one_instrumentor.instrument(tracer_provider=tracer_provider) - - def _build_span_processors(self) -> "list[SpanProcessor]": + instrumentor = configured + params = {} + if instrumentor.is_instrumented_by_opentelemetry: + continue + instrumentor.instrument(tracer_provider=tracer_provider, **params) + if instrumentor.is_instrumented_by_opentelemetry: + self._owned_instrumentors.append((instrumentor, params)) + + def _bootstrap_traces(self, resource: "resources.Resource") -> None: config = self.bootstrap_config - processors: list[SpanProcessor] = [] - if import_checker.is_pyroscope_installed and getattr(config, "pyroscope_endpoint", None): - processors.append(PyroscopeSpanProcessor()) - if config.opentelemetry_log_traces: - processors.append(SimpleSpanProcessor(ConsoleSpanExporter(formatter=_format_span))) - if config.opentelemetry_endpoint and (span_exporter := self._build_span_exporter()): - processors.append(BatchSpanProcessor(span_exporter)) - return processors + profile_spans = import_checker.is_pyroscope_installed and getattr(config, "pyroscope_endpoint", None) + exporter = self._build_span_exporter() if config.opentelemetry_endpoint else None + if not (profile_spans or config.opentelemetry_log_traces or exporter is not None): + existing_provider = get_tracer_provider() + self.tracer_provider = ( + NoOpTracerProvider() if isinstance(existing_provider, ProxyTracerProvider) else existing_provider + ) + return - def bootstrap(self) -> None: - config = self.bootstrap_config - self._silence_otel_loggers() - resource = self._build_resource() - processors = self._build_span_processors() - if processors: + with _rollback_on_error(exporter.shutdown) if exporter is not None else contextlib.nullcontext(): tracer_provider = TracerProvider(resource=resource, sampler=config.opentelemetry_sampler) self._tracer_provider = tracer_provider - for processor in processors: + if profile_spans: + tracer_provider.add_span_processor(PyroscopeSpanProcessor()) + if config.opentelemetry_log_traces: + tracer_provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter(formatter=_format_span))) + processor = BatchSpanProcessor(exporter) if exporter is not None else None + if processor is not None: + with _rollback_on_error(processor.shutdown): tracer_provider.add_span_processor(processor) - set_tracer_provider(tracer_provider) - if get_tracer_provider() is not tracer_provider: - warn_at_caller( - "a TracerProvider is already installed; " - "the configured exporter, sampler and resource will not be used" - ) - self.tracer_provider = get_tracer_provider() - else: - existing_provider = get_tracer_provider() - self.tracer_provider = ( - NoOpTracerProvider() if isinstance(existing_provider, ProxyTracerProvider) else existing_provider + set_tracer_provider(tracer_provider) + if get_tracer_provider() is not tracer_provider: + warn_at_caller( + "a TracerProvider is already installed; the configured exporter, sampler and resource will not be used" ) + self.tracer_provider = get_tracer_provider() + + def bootstrap(self) -> None: + with _rollback_on_error(self._teardown_owned_resources): + self._bootstrap_telemetry() + + def _bootstrap_telemetry(self) -> None: + self._silence_otel_loggers() + resource = self._build_resource() + self._bootstrap_traces(resource) self._bootstrap_metrics(resource) self._apply_instrumentors(self.tracer_provider) def teardown(self) -> None: + self._teardown_owned_resources() + + def _teardown_owned_resources(self) -> None: with collect_teardown_errors() as teardown_errors: - for one_instrumentor in self.bootstrap_config.opentelemetry_instrumentors: - with teardown_errors.capture(type(one_instrumentor).__name__): - if isinstance(one_instrumentor, InstrumentorWithParams): - one_instrumentor.instrumentor.uninstrument(**one_instrumentor.additional_params) - else: - one_instrumentor.uninstrument() + while self._owned_instrumentors: + instrumentor, params = self._owned_instrumentors.pop() + with teardown_errors.capture(type(instrumentor).__name__): + instrumentor.uninstrument(**params) for logger_name, prior in self._prior_logger_disabled.items(): logging.getLogger(logger_name).disabled = prior self._prior_logger_disabled.clear() diff --git a/lite_bootstrap/instruments/pyroscope_instrument.py b/lite_bootstrap/instruments/pyroscope_instrument.py index 3f02e30..60e5c17 100644 --- a/lite_bootstrap/instruments/pyroscope_instrument.py +++ b/lite_bootstrap/instruments/pyroscope_instrument.py @@ -40,13 +40,14 @@ def bootstrap(self) -> None: raise RuntimeError(msg) namespace = config.opentelemetry_namespace tags = ({"service_namespace": namespace} if namespace else {}) | config.pyroscope_tags - pyroscope.configure( - application_name=config.opentelemetry_service_name or config.service_name_or_default, - server_address=config.pyroscope_endpoint, - sample_rate=config.pyroscope_sample_rate, - tags=tags, + params = { + "application_name": config.opentelemetry_service_name or config.service_name_or_default, + "server_address": config.pyroscope_endpoint, + "sample_rate": config.pyroscope_sample_rate, + "tags": tags, **config.pyroscope_additional_params, - ) + } + pyroscope.configure(**params) def teardown(self) -> None: pyroscope.shutdown() diff --git a/lite_bootstrap/instruments/sentry_instrument.py b/lite_bootstrap/instruments/sentry_instrument.py index 4c80492..d07c141 100644 --- a/lite_bootstrap/instruments/sentry_instrument.py +++ b/lite_bootstrap/instruments/sentry_instrument.py @@ -120,7 +120,7 @@ def is_sentry_auto_enabling_integrations_on(self) -> bool: def should_collect_request_body(content_length: str | None) -> bool: client = sentry_sdk.get_client() - if "incoming_request" not in client.options["data_collection"]["http_bodies"]: + if not client.is_active() or "incoming_request" not in client.options["data_collection"]["http_bodies"]: return False try: size = int(content_length or "0") @@ -140,9 +140,12 @@ def capture_request_exception( handled: bool, enrich: typing.Callable[["sentry_types.Event"], "sentry_types.Event"] = lambda event: event, ) -> None: + client = sentry_sdk.get_client() + if not client.is_active(): + return event, hint = event_from_exception( exception, - client_options=sentry_sdk.get_client().options, + client_options=client.options, mechanism={"type": mechanism_type, "handled": handled}, ) sentry_sdk.capture_event(enrich(event), hint=hint) @@ -310,11 +313,11 @@ def bootstrap(self) -> None: def teardown(self) -> None: """Flush pending events and reset the SDK to a no-op state. - Calling ``sentry_sdk.init()`` with no DSN disables further event capture. This + Clear the global client to disable capture even when SENTRY_DSN is set. This cleans up after a bootstrap so the same process can be torn down and re-tested without leaking the previous DSN/transport into subsequent code. """ try: sentry_sdk.flush(timeout=2) finally: - sentry_sdk.init() + sentry_sdk.get_global_scope().set_client(None) diff --git a/tests/instruments/test_logging_record_filters.py b/tests/instruments/test_logging_record_filters.py index 35b3570..12ab9ee 100644 --- a/tests/instruments/test_logging_record_filters.py +++ b/tests/instruments/test_logging_record_filters.py @@ -391,3 +391,21 @@ def test_every_bootstrapper_installs_configured_filters( assert [record.getMessage() for record in configured_filter.seen] == ["through the bootstrapper"] assert target_logger.filters == [] + + +def test_teardown_preserves_preexisting_configured_record_filter() -> None: + target = logging.getLogger(__name__ + ".preexisting") + existing = RecordingFilter() + owned = RecordingFilter() + target.addFilter(existing) + try: + instrument = LoggingInstrument( + bootstrap_config=LoggingConfig(logging_record_filters={target.name: (existing, owned, owned)}) + ) + instrument.bootstrap() + assert target.filters == [existing, owned] + instrument.teardown() + assert target.filters == [existing] + finally: + target.removeFilter(existing) + target.removeFilter(owned) diff --git a/tests/instruments/test_opentelemetry_instrument.py b/tests/instruments/test_opentelemetry_instrument.py index 350935c..07e9620 100644 --- a/tests/instruments/test_opentelemetry_instrument.py +++ b/tests/instruments/test_opentelemetry_instrument.py @@ -812,3 +812,77 @@ def test_configured_exporter_still_records_and_exports_spans(protocol: typing.Li assert [span.name for span in exporter.get_finished_spans()] == ["exported"] finally: instrument.teardown() + + +@pytest.mark.parametrize("wrapped", [False, True]) +def test_otel_preserves_preexisting_instrumentation(wrapped: bool) -> None: + external = CustomInstrumentor() + external.instrument() + try: + configured = InstrumentorWithParams(instrumentor=external) if wrapped else external + instrument = OpenTelemetryInstrument( + bootstrap_config=OpenTelemetryConfig(opentelemetry_instrumentors=[configured]) + ) + instrument.bootstrap() + instrument.teardown() + assert external.is_instrumented_by_opentelemetry + finally: + if external.is_instrumented_by_opentelemetry: + external.uninstrument() + + +@pytest.mark.parametrize("cleanup_fails", [False, True]) +def test_otel_rolls_back_earlier_instrumentation_when_startup_fails( + monkeypatch: pytest.MonkeyPatch, cleanup_fails: bool +) -> None: + primary = ValueError("instrumentation startup failed") + earlier = CustomInstrumentor() + + class FailingInstrumentor(BaseInstrumentor): + def instrumentation_dependencies(self) -> typing.Collection[str]: + return [] + + def _uninstrument(self, **kwargs: object) -> None: ... + + def _instrument(self, **_kwargs: object) -> None: + raise primary + + logger = logging.getLogger("opentelemetry.trace") + monkeypatch.setattr(logger, "disabled", False) + instrument = OpenTelemetryInstrument( + bootstrap_config=OpenTelemetryConfig( + opentelemetry_log_traces=True, + opentelemetry_instrumentors=[InstrumentorWithParams(instrumentor=earlier), FailingInstrumentor()], + ) + ) + shutdown_provider = SDKTracerProvider.shutdown + + def shutdown_then_fail(provider: SDKTracerProvider) -> None: + shutdown_provider(provider) + if cleanup_fails: + message = "provider shutdown failed" + raise RuntimeError(message) + + with ( + patch.object(SDKTracerProvider, "shutdown", autospec=True, side_effect=shutdown_then_fail) as shutdown, + pytest.raises(ValueError, match="instrumentation startup failed") as caught, + ): + instrument.bootstrap() + + assert caught.value is primary + if cleanup_fails: + assert any("provider shutdown failed" in note for note in caught.value.__notes__) + shutdown.assert_called_once_with(trace.get_tracer_provider()) + assert not earlier.is_instrumented_by_opentelemetry + assert not logger.disabled + + +def test_otel_releases_exporter_when_tracer_provider_construction_fails(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", "invalid") + instrument = OpenTelemetryInstrument(bootstrap_config=OpenTelemetryConfig(opentelemetry_endpoint="localhost:4317")) + with ( + patch.object(otel_module, "OTLPGrpcSpanExporter", autospec=True) as exporter, + pytest.raises(ValueError, match="invalid"), + ): + instrument.bootstrap() + exporter.return_value.shutdown.assert_called_once() diff --git a/tests/instruments/test_pyroscope_instrument.py b/tests/instruments/test_pyroscope_instrument.py index a41f533..a5c4706 100644 --- a/tests/instruments/test_pyroscope_instrument.py +++ b/tests/instruments/test_pyroscope_instrument.py @@ -246,3 +246,13 @@ async def test_handler() -> None: ... mock_module.call("span_name", "GET /test-handler"), ] ) + + +def test_pyroscope_additional_params_override_sample_rate() -> None: + rate = 50 + config = PyroscopeConfig( + pyroscope_endpoint="http://pyroscope:4040", pyroscope_additional_params={"sample_rate": rate} + ) + with patch(_PYROSCOPE_PYROSCOPE) as sdk: + PyroscopeInstrument(bootstrap_config=config).bootstrap() + assert sdk.configure.call_args.kwargs["sample_rate"] == rate diff --git a/tests/instruments/test_sentry_instrument.py b/tests/instruments/test_sentry_instrument.py index 92de7de..1c0c511 100644 --- a/tests/instruments/test_sentry_instrument.py +++ b/tests/instruments/test_sentry_instrument.py @@ -72,14 +72,20 @@ def test_sentry_instrument_with_structlog_error( sentry_instrument.teardown() -def test_sentry_teardown_disables_sdk(minimal_sentry_config: SentryConfig) -> None: +def test_sentry_teardown_disables_sdk( + minimal_sentry_config: SentryConfig, sentry_mock: SentryTestTransport, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("SENTRY_DSN", "https://synthetic@example.invalid/1") + monkeypatch.setattr("sentry_sdk.client.make_transport", lambda options: sentry_mock if options["dsn"] else None) instrument = SentryInstrument(bootstrap_config=minimal_sentry_config) instrument.bootstrap() assert sentry_sdk.get_client().dsn == minimal_sentry_config.sentry_dsn instrument.teardown() - assert sentry_sdk.get_client().dsn is None + assert not sentry_sdk.get_client().is_active() + assert sentry_sdk.capture_message("after teardown") is None + assert sentry_mock.mock_envelopes == [] def test_sentry_instrument_empty_dsn() -> None: @@ -138,7 +144,11 @@ def test_modify_lifts_message_and_attaches_extra( assert enrich_sentry_event_from_structlog_log(event_before, {}) == event_after -def test_sentry_teardown_runs_init_when_flush_raises(minimal_sentry_config: SentryConfig) -> None: +def test_sentry_teardown_disables_sdk_when_flush_raises( + minimal_sentry_config: SentryConfig, sentry_mock: SentryTestTransport, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("SENTRY_DSN", "https://synthetic@example.invalid/1") + monkeypatch.setattr("sentry_sdk.client.make_transport", lambda options: sentry_mock if options["dsn"] else None) instrument = SentryInstrument(bootstrap_config=minimal_sentry_config) instrument.bootstrap() @@ -148,8 +158,9 @@ def test_sentry_teardown_runs_init_when_flush_raises(minimal_sentry_config: Sent ): instrument.teardown() - # init() still ran (in the finally), so SDK is now disabled. - assert sentry_sdk.get_client().dsn is None + assert not sentry_sdk.get_client().is_active() + assert sentry_sdk.capture_message("after teardown") is None + assert sentry_mock.mock_envelopes == [] def installed_logging_integration() -> LoggingIntegration: diff --git a/tests/test_fastapi_bootstrap.py b/tests/test_fastapi_bootstrap.py index 08366aa..53c1416 100644 --- a/tests/test_fastapi_bootstrap.py +++ b/tests/test_fastapi_bootstrap.py @@ -12,6 +12,7 @@ import fastapi import pytest import structlog +from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor from opentelemetry.metrics import get_meter_provider from opentelemetry.sdk.metrics import MeterProvider as SDKMeterProvider from opentelemetry.sdk.trace import TracerProvider as SDKTracerProvider @@ -824,3 +825,88 @@ def test_empty_observability_paths_disable_routes() -> None: assert client.get("/health/").status_code == status.HTTP_404_NOT_FOUND assert client.get("/metrics").status_code == status.HTTP_404_NOT_FOUND assert client.get("/").status_code == status.HTTP_404_NOT_FOUND + + +def test_fastapi_docs_url_conflict_names_canonical_setting() -> None: + with pytest.raises(ConfigurationError, match="swagger_path"): + FastAPIConfig(application_kwargs={"docs_url": "/custom"}) + + +def test_fastapi_shutdown_keeps_lifespan_error_when_exporter_shutdown_fails() -> None: + primary = ValueError("user shutdown failed") + + @contextlib.asynccontextmanager + async def lifespan(_: fastapi.FastAPI) -> AsyncGenerator[None]: + yield + raise primary + + config = FastAPIConfig(application=fastapi.FastAPI(lifespan=lifespan), opentelemetry_log_traces=True) + app = FastAPIBootstrapper(config).bootstrap() + provider = get_tracer_provider() + with ( + patch.object(provider, "shutdown", side_effect=RuntimeError("exporter shutdown failed")), + pytest.raises(ValueError, match="user shutdown failed") as caught, + TestClient(app), + ): + pass + assert caught.value is primary + assert any("exporter shutdown failed" in note for note in caught.value.__notes__) + + +def test_fastapi_metrics_endpoint_conflict_names_canonical_setting() -> None: + config = FastAPIConfig(prometheus_expose_additional_params={"endpoint": "/custom"}) + with pytest.raises(ConfigurationError, match="prometheus_metrics_path"): + FastAPIBootstrapper(config).bootstrap() + + +@pytest.mark.parametrize("stack_already_built", [False, True]) +def test_fastapi_excludes_infrastructure_temporarily_from_borrowed_tracing(stack_already_built: bool) -> None: + app = fastapi.FastAPI() + + @app.get("/work") + async def work() -> str: + return "ok" + + provider = SDKTracerProvider() + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + FastAPIInstrumentor.instrument_app(app, tracer_provider=provider, exclude_spans=["receive", "send"]) + external_builder = app.build_middleware_stack + client = TestClient(app) + if stack_already_built: + assert client.get("/work").status_code == status.HTTP_200_OK + exporter.clear() + + config = FastAPIConfig( + application=app, + logging_enabled=False, + opentelemetry_log_traces=True, + prometheus_http_metrics_enabled=False, + ) + bootstrapper = FastAPIBootstrapper(config) + try: + bootstrapper.bootstrap() + for path in (config.health_checks_path, config.prometheus_metrics_path, "/work"): + assert client.get(path).status_code == status.HTTP_200_OK + assert [span.name for span in exporter.get_finished_spans()] == ["GET /work"] + + bootstrapper.teardown() + assert app.build_middleware_stack is external_builder + exporter.clear() + assert client.get(config.prometheus_metrics_path).status_code == status.HTTP_200_OK + assert [span.name for span in exporter.get_finished_spans()] == [f"GET {config.prometheus_metrics_path}"] + finally: + bootstrapper.teardown() + FastAPIInstrumentor.uninstrument_app(app) + provider.shutdown() + + +def test_fastapi_instrumentation_failure_rolls_back_tracer_provider() -> None: + config = FastAPIConfig(logging_enabled=False, opentelemetry_log_traces=True, opentelemetry_excluded_urls=["["]) + bootstrapper = FastAPIBootstrapper(config) + with ( + patch.object(SDKTracerProvider, "shutdown", autospec=True, side_effect=SDKTracerProvider.shutdown) as shutdown, + pytest.raises(re.error), + ): + bootstrapper.bootstrap() + shutdown.assert_called_once_with(get_tracer_provider()) diff --git a/tests/test_faststream_bootstrap.py b/tests/test_faststream_bootstrap.py index cbb9940..d22c029 100644 --- a/tests/test_faststream_bootstrap.py +++ b/tests/test_faststream_bootstrap.py @@ -1,3 +1,4 @@ +import contextlib import dataclasses import json import logging @@ -5,6 +6,7 @@ import typing import uuid import warnings +from collections.abc import AsyncGenerator from types import TracebackType from unittest.mock import AsyncMock, patch @@ -38,7 +40,7 @@ FastStreamOpenTelemetryInstrument, FastStreamPrometheusInstrument, ) -from lite_bootstrap.exceptions import ConfigurationError +from lite_bootstrap.exceptions import ConfigurationError, TeardownError from lite_bootstrap.instruments.healthchecks_instrument import HealthProbe from tests.conftest import ( HEALTH_PROBE_CASES, @@ -55,6 +57,21 @@ logger = structlog.getLogger(__name__) +class _ShutdownCountingSpanExporter(InMemorySpanExporter): + shutdown_calls = 0 + + def __init__(self, shutdown_error: Exception | None = None) -> None: + super().__init__() + self.shutdown_error = shutdown_error + + def shutdown(self) -> None: + self.shutdown_calls += 1 + super().shutdown() + if self.shutdown_error is not None: + error, self.shutdown_error = self.shutdown_error, None + raise error + + @pytest.fixture def broker() -> RedisBroker: return RedisBroker() @@ -128,6 +145,215 @@ async def test_faststream_bootstrap(broker: RedisBroker) -> None: assert not bootstrapper.is_bootstrapped +def test_faststream_exports_broker_and_user_shutdown_spans(broker: RedisBroker) -> None: + @contextlib.asynccontextmanager + async def lifespan() -> AsyncGenerator[None]: + yield + with get_tracer(__name__).start_as_current_span("user-shutdown"): + pass + + async def stop_broker() -> None: + with get_tracer(__name__).start_as_current_span("broker-drain"): + pass + + application = faststream.asgi.AsgiFastStream(broker, lifespan=lifespan) + bootstrapper = FastStreamBootstrapper( + FastStreamConfig(application=application, logging_enabled=False, opentelemetry_log_traces=True) + ) + bootstrapper.bootstrap() + provider = get_tracer_provider() + assert isinstance(provider, SDKTracerProvider) + exporter = _ShutdownCountingSpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + + with ( + patch.object(broker, "start", new=AsyncMock()), + patch.object(broker, "stop", new=AsyncMock(side_effect=stop_broker)), + TestClient(application), + get_tracer(__name__).start_as_current_span("during-service"), + ): + pass + + assert [span.name for span in exporter.get_finished_spans()] == [ + "during-service", + "broker-drain", + "user-shutdown", + ] + assert not bootstrapper.is_bootstrapped + assert exporter.shutdown_calls == 1 + bootstrapper.teardown() + assert exporter.shutdown_calls == 1 + + +@pytest.mark.parametrize( + "failure_stage", ["lifespan-startup", "startup-hook", "shutdown-hook", "broker-stop", "lifespan-shutdown"] +) +@pytest.mark.parametrize("teardown_fails", [False, True]) +def test_faststream_lifecycle_failure_preserves_error_and_tears_down( + broker: RedisBroker, failure_stage: str, teardown_fails: bool +) -> None: + error = ValueError(failure_stage) + shutdown_seen: list[bool] = [] + + def fail_at(stage: str) -> None: + if failure_stage == stage: + raise error + + @contextlib.asynccontextmanager + async def lifespan() -> AsyncGenerator[None]: + fail_at("lifespan-startup") + try: + yield + finally: + shutdown_seen.append(bootstrapper.is_bootstrapped) + fail_at("lifespan-shutdown") + + async def startup() -> None: + fail_at("startup-hook") + + async def shutdown() -> None: + fail_at("shutdown-hook") + + async def stop_broker() -> None: + fail_at("broker-stop") + + application = faststream.asgi.AsgiFastStream( + broker, lifespan=lifespan, on_startup=[startup], on_shutdown=[shutdown] + ) + bootstrapper = FastStreamBootstrapper( + FastStreamConfig(application=application, logging_enabled=False, opentelemetry_log_traces=True) + ) + bootstrapper.bootstrap() + provider = get_tracer_provider() + assert isinstance(provider, SDKTracerProvider) + exporter = _ShutdownCountingSpanExporter(ValueError("exporter shutdown failed") if teardown_fails else None) + provider.add_span_processor(SimpleSpanProcessor(exporter)) + + with ( + patch.object(broker, "start", new=AsyncMock()), + patch.object(broker, "stop", new=AsyncMock(side_effect=stop_broker)), + pytest.raises(ValueError, match=failure_stage) as caught, + TestClient(application), + ): + pass + + assert caught.value is error + assert shutdown_seen == ([] if failure_stage == "lifespan-startup" else [True]) + assert not bootstrapper.is_bootstrapped + assert exporter.shutdown_calls == 1 + if teardown_fails: + assert any("exporter shutdown failed" in note for note in caught.value.__notes__) + + +def test_faststream_reports_teardown_failure_after_successful_lifespan(broker: RedisBroker) -> None: + application = faststream.asgi.AsgiFastStream(broker) + bootstrapper = FastStreamBootstrapper( + FastStreamConfig(application=application, logging_enabled=False, opentelemetry_log_traces=True) + ) + bootstrapper.bootstrap() + provider = get_tracer_provider() + assert isinstance(provider, SDKTracerProvider) + exporter = _ShutdownCountingSpanExporter(ValueError("exporter shutdown failed")) + provider.add_span_processor(SimpleSpanProcessor(exporter)) + + with ( + patch.object(broker, "start", new=AsyncMock()), + patch.object(broker, "stop", new=AsyncMock()), + pytest.raises(TeardownError, match="exporter shutdown failed"), + TestClient(application), + ): + pass + + assert not bootstrapper.is_bootstrapped + assert exporter.shutdown_calls == 1 + + +async def test_faststream_lifespan_preserves_run_extra_options(broker: RedisBroker) -> None: + seen: list[tuple[str, int]] = [] + + @contextlib.asynccontextmanager + async def lifespan(service: str, workers: int) -> AsyncGenerator[None]: + seen.append((service, workers)) + yield + seen.append((service, workers)) + + application = faststream.asgi.AsgiFastStream(broker, lifespan=lifespan) + bootstrapper = FastStreamBootstrapper(FastStreamConfig(application=application, logging_enabled=False)) + bootstrapper.bootstrap() + + with patch.object(broker, "start", new=AsyncMock()), patch.object(broker, "stop", new=AsyncMock()): + async with application.start_lifespan_context(run_extra_options={"service": "worker", "workers": 3}): + assert seen == [("worker", 3)] + + assert seen == [("worker", 3), ("worker", 3)] + assert not bootstrapper.is_bootstrapped + + +async def test_faststream_lifespan_preserves_positional_and_keyword_arguments(broker: RedisBroker) -> None: + @contextlib.asynccontextmanager + async def lifespan(service: str, *, workers: int) -> AsyncGenerator[None]: + assert service == "worker" + expected_workers = 3 + assert workers == expected_workers + yield + + application = faststream.asgi.AsgiFastStream(broker, lifespan=lifespan) + bootstrapper = FastStreamBootstrapper(FastStreamConfig(application=application, logging_enabled=False)) + bootstrapper.bootstrap() + + async with application.lifespan_context("worker", workers=3) as result: + assert result is None + assert bootstrapper.is_bootstrapped + + assert not bootstrapper.is_bootstrapped + + +@pytest.mark.parametrize( + "lifecycle", + [ + pytest.param( + "context-manager", + marks=pytest.mark.skipif( + not hasattr(faststream.asgi.AsgiFastStream, "__aexit__"), + reason="This FastStream version has no application context manager", + ), + ), + "start-stop", + ], +) +async def test_faststream_manual_lifecycle_requires_explicit_teardown(broker: RedisBroker, lifecycle: str) -> None: + async def stop_broker() -> None: + with get_tracer(__name__).start_as_current_span("broker-drain"): + pass + + application = faststream.asgi.AsgiFastStream(broker) + bootstrapper = FastStreamBootstrapper( + FastStreamConfig(application=application, logging_enabled=False, opentelemetry_log_traces=True) + ) + bootstrapper.bootstrap() + provider = get_tracer_provider() + assert isinstance(provider, SDKTracerProvider) + exporter = _ShutdownCountingSpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + + with ( + patch.object(broker, "start", new=AsyncMock()), + patch.object(broker, "stop", new=AsyncMock(side_effect=stop_broker)), + ): + if lifecycle == "context-manager": + async with application: + assert bootstrapper.is_bootstrapped + else: + await application.start() + await application.stop() + + assert [span.name for span in exporter.get_finished_spans()] == ["broker-drain"] + assert bootstrapper.is_bootstrapped + assert exporter.shutdown_calls == 0 + bootstrapper.teardown() + assert exporter.shutdown_calls == 1 + + @pytest.mark.parametrize("path", ["/health", "/health/", "/metrics", "/metrics/"]) async def test_faststream_route_paths_accept_both_slash_forms(broker: RedisBroker, path: str) -> None: config = dataclasses.replace( @@ -448,37 +674,6 @@ async def test_faststream_health_check_pings_every_broker(broker: RedisBroker) - bootstrapper.teardown() -def test_faststream_opentelemetry_excluded_urls_in_built_set(broker: RedisBroker) -> None: - bootstrap_config = dataclasses.replace( - build_faststream_config(broker=broker), - opentelemetry_excluded_urls=["/foo", "/bar"], - ) - instrument = FastStreamOpenTelemetryInstrument(bootstrap_config=bootstrap_config) - excluded = instrument._build_excluded_urls() # noqa: SLF001 - assert "/foo" in excluded - assert "/bar" in excluded - - -def test_faststream_build_excluded_urls_covers_prometheus_and_health_paths(broker: RedisBroker) -> None: - # Pins the genuine cross-instrument reads in _build_excluded_urls: a rename of - # prometheus_metrics_path / health_checks_path (or a regression in the health - # conditional) breaks this loudly instead of silently dropping the exclusion. - config_no_health_spans = dataclasses.replace( - build_faststream_config(broker=broker), - opentelemetry_generate_health_check_spans=False, - ) - excluded = FastStreamOpenTelemetryInstrument(bootstrap_config=config_no_health_spans)._build_excluded_urls() # noqa: SLF001 - assert config_no_health_spans.prometheus_metrics_path in excluded # always excluded - assert config_no_health_spans.health_checks_path in excluded # excluded when health spans are off - - config_with_health_spans = dataclasses.replace( - build_faststream_config(broker=broker), opentelemetry_generate_health_check_spans=True - ) - excluded_with = FastStreamOpenTelemetryInstrument(bootstrap_config=config_with_health_spans)._build_excluded_urls() # noqa: SLF001 - assert config_with_health_spans.prometheus_metrics_path in excluded_with - assert config_with_health_spans.health_checks_path not in excluded_with # kept when health spans are on - - @pytest.mark.parametrize( "middleware_cls", [RedisPrometheusMiddleware, None], ids=["with_middleware", "without_middleware"] ) diff --git a/tests/test_litestar_bootstrap.py b/tests/test_litestar_bootstrap.py index 6343b6e..dceb9dc 100644 --- a/tests/test_litestar_bootstrap.py +++ b/tests/test_litestar_bootstrap.py @@ -898,3 +898,9 @@ def test_litestar_openapi_and_prometheus_fall_back_to_default_service_name() -> with TestClient(app=application) as client: client.get("/health/") assert 'app_name="micro-service"' in client.get("/metrics").text + + +def test_litestar_swagger_path_conflict_names_canonical_setting() -> None: + config = LitestarConfig(swagger_additional_params={"path": "/custom"}) + with pytest.raises(ConfigurationError, match="swagger_path"): + LitestarBootstrapper(config).bootstrap() diff --git a/tests/test_litestar_sentry_capture.py b/tests/test_litestar_sentry_capture.py index 7af26da..a5ccb02 100644 --- a/tests/test_litestar_sentry_capture.py +++ b/tests/test_litestar_sentry_capture.py @@ -534,3 +534,12 @@ def test_integration_capture_does_not_warn_about_lean_settings(sentry_mock: Sent sentry_traces_sample_rate=1.0, sentry_integrations=[LitestarIntegration()], ) + + +def test_litestar_handles_request_errors_after_sentry_teardown(sentry_mock: SentryTestTransport) -> None: + bootstrapper = LitestarBootstrapper(_build_config(sentry_mock, logging_enabled=False)) + application = bootstrapper.bootstrap() + with TestClient(app=application) as client: + bootstrapper.teardown() + assert client.get("/mapped-503").status_code == status_codes.HTTP_503_SERVICE_UNAVAILABLE + assert sentry_mock.mock_envelopes == [] diff --git a/tests/test_starlette_sentry_capture.py b/tests/test_starlette_sentry_capture.py index a19afbf..ef12ea7 100644 --- a/tests/test_starlette_sentry_capture.py +++ b/tests/test_starlette_sentry_capture.py @@ -911,3 +911,14 @@ async def fail(request: Request) -> Response: [event] = _error_events(sentry_mock) assert event["transaction"] == "/items/{item_id}" assert event["transaction_info"] == {"source": "route"} + + +def test_fastapi_requests_continue_after_sentry_teardown(sentry_mock: SentryTestTransport) -> None: + bootstrapper = FastAPIBootstrapper(_build_config(sentry_mock, logging_enabled=False)) + client = TestClient(bootstrapper.bootstrap()) + assert client.get("/logs-then-ok").status_code == status.HTTP_200_OK + bootstrapper.teardown() + + assert client.get("/logs-then-ok").status_code == status.HTTP_200_OK + assert client.get("/mapped-503").status_code == status.HTTP_503_SERVICE_UNAVAILABLE + assert sentry_mock.mock_envelopes == []