diff --git a/microbootstrap/bootstrappers/faststream.py b/microbootstrap/bootstrappers/faststream.py index 9e72305..97d0b37 100644 --- a/microbootstrap/bootstrappers/faststream.py +++ b/microbootstrap/bootstrappers/faststream.py @@ -1,8 +1,10 @@ from __future__ import annotations +import functools import json import typing import prometheus_client +import sentry_sdk import structlog import typing_extensions from faststream._internal.logger.logger_proxy import RealLoggerObject @@ -26,6 +28,29 @@ tracer: typing.Final = trace.get_tracer(__name__) +MessageT = typing.TypeVar("MessageT") +ResponseT = typing.TypeVar("ResponseT") + + +def _with_sentry_isolation_scope( + process_message: typing.Callable[[MessageT], typing.Awaitable[ResponseT]], +) -> typing.Callable[[MessageT], typing.Awaitable[ResponseT]]: + @functools.wraps(process_message) + async def isolated_process_message(message: MessageT) -> ResponseT: + with sentry_sdk.isolation_scope(): + return await process_message(message) + + return isolated_process_message + + +def _isolate_faststream_subscribers(application: AsgiFastStream) -> None: + for broker in application.brokers: + for subscriber in broker.subscribers: + object.__setattr__( + subscriber, + "process_message", + _with_sentry_isolation_scope(subscriber.process_message), + ) class KwargsAsgiFastStream(AsgiFastStream): @@ -51,7 +76,18 @@ def bootstrap_before(self: typing_extensions.Self) -> dict[str, typing.Any]: } -FastStreamBootstrapper.use_instrument()(SentryInstrument) +@FastStreamBootstrapper.use_instrument() +class FastStreamSentryInstrument(SentryInstrument): + def bootstrap_after(self, application: AsgiFastStream) -> AsgiFastStream: # type: ignore[override] + # FastStream logs handler errors after custom broker middlewares exit, so the isolation scope + # must enclose the subscriber's complete processing lifecycle. + if application.brokers: + _isolate_faststream_subscribers(application) + else: + application.on_startup(functools.partial(_isolate_faststream_subscribers, application)) + return application + + FastStreamBootstrapper.use_instrument()(PyroscopeInstrument) diff --git a/microbootstrap/middlewares/fastapi.py b/microbootstrap/middlewares/fastapi.py index ec3fb6d..6fcf5ad 100644 --- a/microbootstrap/middlewares/fastapi.py +++ b/microbootstrap/middlewares/fastapi.py @@ -26,10 +26,7 @@ async def dispatch( return await call_next(request) start_time: typing.Final = time.perf_counter_ns() - try: - response = await call_next(request) - except Exception: # noqa: BLE001 - response = fastapi.Response(status_code=500) + response = await call_next(request) fill_log_message( "exception" if response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR else "info", diff --git a/tests/bootstrappers/test_faststream.py b/tests/bootstrappers/test_faststream.py index 7ec2089..c7fb425 100644 --- a/tests/bootstrappers/test_faststream.py +++ b/tests/bootstrappers/test_faststream.py @@ -1,9 +1,12 @@ +import asyncio +import logging import typing from unittest import mock from unittest.mock import MagicMock import faker import pytest +import sentry_sdk from fastapi import status from fastapi.testclient import TestClient from faststream.redis import RedisBroker, TestRedisBroker @@ -16,6 +19,7 @@ from microbootstrap.instruments.logging_instrument import LoggingConfig from microbootstrap.instruments.opentelemetry_instrument import FastStreamOpentelemetryConfig, OpentelemetryConfig from microbootstrap.instruments.prometheus_instrument import FastStreamPrometheusConfig +from microbootstrap.instruments.sentry_instrument import SentryConfig from microbootstrap.settings import FastStreamSettings @@ -124,3 +128,93 @@ async def test_faststream_logging(broker: RedisBroker, minimal_logging_config: L FastStreamBootstrapper(FastStreamSettings()).configure_application( FastStreamConfig(broker=broker) ).configure_instruments(minimal_logging_config).bootstrap() + + +async def test_faststream_sentry_isolates_concurrent_messages( + broker: RedisBroker, + minimal_sentry_config: SentryConfig, + monkeypatch: pytest.MonkeyPatch, +) -> None: + channel: typing.Final = "test-channel" + conversation_id_tag: typing.Final = "conversation_id" + first_started = asyncio.Event() + second_started = asyncio.Event() + second_logged = asyncio.Event() + captured_tags: dict[str, str | None] = {} + monkeypatch.setattr(sentry_sdk, "init", mock.Mock()) + minimal_sentry_config.sentry_tags = None + + @broker.subscriber(channel) + async def handler(conversation_id: str) -> None: + sentry_sdk.get_isolation_scope().set_tag(conversation_id_tag, conversation_id) + if conversation_id == "first": + first_started.set() + await second_started.wait() + await second_logged.wait() + else: + second_started.set() + await first_started.wait() + raise ValueError(conversation_id) + + FastStreamBootstrapper(FastStreamSettings()).configure_application( + FastStreamConfig(broker=broker) + ).configure_instruments(minimal_sentry_config).bootstrap() + + original_log = broker.config.logger.log + + def record_error_tag(*args: typing.Any, **kwargs: typing.Any) -> None: # noqa: ANN401 + if kwargs.get("log_level") == logging.ERROR: + error_message = typing.cast("str", kwargs["message"]) + captured_tags[error_message] = sentry_sdk.get_isolation_scope()._tags.get(conversation_id_tag) # noqa: SLF001 + if error_message.endswith("second"): + second_logged.set() + original_log(*args, **kwargs) + + monkeypatch.setattr(broker.config.logger, "log", record_error_tag) + event_loop = asyncio.get_running_loop() + previous_exception_handler = event_loop.get_exception_handler() + event_loop.set_exception_handler(lambda *_: None) + try: + async with TestRedisBroker(broker): + errors: typing.Final = await asyncio.gather( + broker.publish("first", channel), + broker.publish("second", channel), + return_exceptions=True, + ) + finally: + event_loop.set_exception_handler(previous_exception_handler) + + assert all(isinstance(error, ValueError) for error in errors) + assert captured_tags == { + "ValueError: first": "first", + "ValueError: second": "second", + } + assert conversation_id_tag not in sentry_sdk.get_isolation_scope()._tags # noqa: SLF001 + + +async def test_faststream_sentry_isolates_broker_configured_on_startup( + broker: RedisBroker, + minimal_sentry_config: SentryConfig, + monkeypatch: pytest.MonkeyPatch, +) -> None: + application: typing.Any + + @broker.subscriber("test-channel") + async def handler(message: str) -> None: + pass + + def set_broker() -> None: + application.set_broker(broker) + + monkeypatch.setattr(sentry_sdk, "init", mock.Mock()) + monkeypatch.setattr(broker, "start", mock.AsyncMock()) + application = ( + FastStreamBootstrapper(FastStreamSettings()) + .configure_application(FastStreamConfig(on_startup=[set_broker])) + .configure_instruments(minimal_sentry_config) + .bootstrap() + ) + + await application.start() + + assert hasattr(broker.subscribers[0].process_message, "__wrapped__") diff --git a/tests/instruments/test_logging.py b/tests/instruments/test_logging.py index 0550afa..87622a4 100644 --- a/tests/instruments/test_logging.py +++ b/tests/instruments/test_logging.py @@ -6,6 +6,7 @@ import fastapi import litestar import pytest +from fastapi.responses import JSONResponse from fastapi.testclient import TestClient as FastAPITestClient from faststream.redis import RedisBroker, TestRedisBroker from litestar.testing import TestClient as LitestarTestClient @@ -181,6 +182,33 @@ async def test_handler() -> str: assert fill_log_mock.call_count == 2 # noqa: PLR2004 +def test_fastapi_logging_bootstrap_propagates_unhandled_exception( + monkeypatch: pytest.MonkeyPatch, + minimal_logging_config: LoggingConfig, +) -> None: + fastapi_application: typing.Final = fastapi.FastAPI() + + @fastapi_application.get("/test-error-handler") + async def test_error_handler() -> None: + raise RuntimeError("test error") + + @fastapi_application.exception_handler(Exception) + async def handle_exception(_request: fastapi.Request, exception: Exception) -> JSONResponse: + return JSONResponse(status_code=418, content={"detail": str(exception)}) + + logging_instrument: typing.Final = FastApiLoggingInstrument(minimal_logging_config) + logging_instrument.bootstrap() + logging_instrument.bootstrap_after(fastapi_application) + monkeypatch.setattr("microbootstrap.middlewares.fastapi.fill_log_message", fill_log_mock := mock.Mock()) + + with FastAPITestClient(app=fastapi_application, raise_server_exceptions=False) as test_client: + response: typing.Final = test_client.get("/test-error-handler") + + assert response.status_code == 418 # noqa: PLR2004 + assert response.json() == {"detail": "test error"} + fill_log_mock.assert_not_called() + + def test_fastapi_logging_bootstrap_ignores_health( monkeypatch: pytest.MonkeyPatch, minimal_logging_config: LoggingConfig ) -> None: diff --git a/tests/instruments/test_sentry.py b/tests/instruments/test_sentry.py index 98f46c5..9e7fa1a 100644 --- a/tests/instruments/test_sentry.py +++ b/tests/instruments/test_sentry.py @@ -4,13 +4,16 @@ import typing from unittest import mock +import fastapi import litestar import pytest import structlog +from fastapi.testclient import TestClient as FastAPITestClient from litestar.testing import TestClient as LitestarTestClient from opentelemetry import baggage from opentelemetry.context import Context, attach, detach +from microbootstrap.bootstrappers.fastapi import FastApiLoggingInstrument from microbootstrap.bootstrappers.litestar import LitestarSentryInstrument from microbootstrap.instruments.logging_instrument import LoggingConfig, LoggingInstrument from microbootstrap.instruments.sentry_instrument import ( @@ -80,6 +83,37 @@ async def error_handler() -> None: assert mock_capture_event.called +def test_fastapi_sentry_captures_unhandled_exception_with_traceback( + minimal_logging_config: LoggingConfig, + minimal_sentry_config: SentryConfig, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr("sentry_sdk.Scope.capture_event", capture_event := mock.Mock()) + SentryInstrument(minimal_sentry_config).bootstrap() + fastapi_application: typing.Final = fastapi.FastAPI() + + @fastapi_application.get("/test-error-handler") + async def error_handler() -> None: + raise RuntimeError("test error") + + logging_instrument: typing.Final = FastApiLoggingInstrument(minimal_logging_config) + logging_instrument.bootstrap() + logging_instrument.bootstrap_after(fastapi_application) + + with FastAPITestClient(app=fastapi_application, raise_server_exceptions=False) as test_client: + response: typing.Final = test_client.get("/test-error-handler") + + assert response.status_code == fastapi.status.HTTP_500_INTERNAL_SERVER_ERROR + captured_events: typing.Final = [call.args[0] for call in capture_event.mock_calls] + exception_events: typing.Final = [event for event in captured_events if event.get("exception")] + assert len(exception_events) == 1 + exception_event: typing.Final = exception_events[0] + exception_value: typing.Final = exception_event["exception"]["values"][-1] + assert exception_value["type"] == "RuntimeError" + assert exception_value["value"] == "test error" + assert exception_value["stacktrace"]["frames"] + + class TestSentryEnrichEventFromStructlog: @pytest.mark.parametrize( "event",