diff --git a/microbootstrap/instruments/opentelemetry_instrument.py b/microbootstrap/instruments/opentelemetry_instrument.py index 6c40e0f..0a90602 100644 --- a/microbootstrap/instruments/opentelemetry_instrument.py +++ b/microbootstrap/instruments/opentelemetry_instrument.py @@ -21,6 +21,7 @@ from opentelemetry.util._importlib_metadata import entry_points from microbootstrap.instruments.base import BaseInstrumentConfig, Instrument +from microbootstrap.instruments.sentry_instrument import snapshot_sentry_opentelemetry_baggage LOGGER_OBJ: typing.Final = structlog.get_logger(__name__) @@ -55,6 +56,10 @@ def opentelemetry_baggage_scope( token: typing.Final = context.attach(baggage_context) try: yield + except Exception as exc: + with contextlib.suppress(Exception): + snapshot_sentry_opentelemetry_baggage(exc) + raise finally: context.detach(token) diff --git a/microbootstrap/instruments/sentry_instrument.py b/microbootstrap/instruments/sentry_instrument.py index 51bcc0b..3f5d8b7 100644 --- a/microbootstrap/instruments/sentry_instrument.py +++ b/microbootstrap/instruments/sentry_instrument.py @@ -3,13 +3,14 @@ import functools import typing import urllib.parse +from collections.abc import Mapping import orjson import pydantic import sentry_sdk from opentelemetry import baggage from sentry_sdk import _types as sentry_types -from sentry_sdk.integrations import Integration # noqa: TC002 +from sentry_sdk.integrations import Integration from microbootstrap.instruments.base import BaseInstrumentConfig, Instrument @@ -66,6 +67,71 @@ def enrich_sentry_event_from_structlog_log(event: sentry_types.Event, _hint: sen SENTRY_EXTRA_OTEL_TRACE_ID_KEY: typing.Final = "otelTraceID" SENTRY_EXTRA_OTEL_TRACE_URL_KEY: typing.Final = "otelTraceURL" +SENTRY_OTEL_BAGGAGE_SNAPSHOT_ATTRIBUTE: typing.Final = "__microbootstrap_sentry_opentelemetry_baggage__" + + +@typing.final +class SentryOpentelemetryBaggageIntegration(Integration): + identifier = "microbootstrap_opentelemetry_baggage" + + def __init__(self, baggage_keys: set[str], baggage_url_templates: dict[str, str]) -> None: + self.baggage_keys: typing.Final = frozenset(baggage_keys) + self.baggage_url_templates: typing.Final = dict(baggage_url_templates) + + @staticmethod + def setup_once() -> None: + pass + + +def snapshot_sentry_opentelemetry_baggage(exception: BaseException) -> None: + integration = sentry_sdk.get_client().get_integration(SentryOpentelemetryBaggageIntegration) + if not isinstance(integration, SentryOpentelemetryBaggageIntegration) or hasattr( + exception, + SENTRY_OTEL_BAGGAGE_SNAPSHOT_ATTRIBUTE, + ): + return + + configured_keys: typing.Final = integration.baggage_keys.union(integration.baggage_url_templates) + snapshot: typing.Final = {key: baggage.get_baggage(key) for key in configured_keys} + with contextlib.suppress(AttributeError, TypeError): + setattr(exception, SENTRY_OTEL_BAGGAGE_SNAPSHOT_ATTRIBUTE, snapshot) + + +def _find_sentry_opentelemetry_baggage_snapshot(hint: sentry_types.Hint) -> Mapping[str, object | None] | None: + if not isinstance(hint, Mapping): + return None + + exc_info: typing.Final = hint.get("exc_info") + if not isinstance(exc_info, tuple): + return None + try: + _, exception_value, _ = exc_info + except ValueError: + return None + if not isinstance(exception_value, BaseException): + return None + + exceptions_to_visit: list[BaseException] = [exception_value] + visited_exceptions: set[int] = set() + while exceptions_to_visit: + exception = exceptions_to_visit.pop() + if id(exception) in visited_exceptions: + continue + visited_exceptions.add(id(exception)) + exception_snapshot = getattr(exception, SENTRY_OTEL_BAGGAGE_SNAPSHOT_ATTRIBUTE, None) + if isinstance(exception_snapshot, Mapping): + return exception_snapshot + + nested_exceptions = getattr(exception, "exceptions", ()) + if isinstance(nested_exceptions, tuple): + exceptions_to_visit.extend( + reversed([nested for nested in nested_exceptions if isinstance(nested, BaseException)]) + ) + if chained_exception := exception.__cause__ or ( + None if exception.__suppress_context__ else exception.__context__ + ): + exceptions_to_visit.append(chained_exception) + return None def add_trace_url_to_event( @@ -80,27 +146,30 @@ def enrich_sentry_event_from_opentelemetry_baggage( baggage_keys: set[str], baggage_url_templates: dict[str, str], event: sentry_types.Event, - _hint: sentry_types.Hint, + hint: sentry_types.Hint, ) -> sentry_types.Event: - baggage_values = { - key: value - for key in baggage_keys.union(baggage_url_templates) - if (value := baggage.get_baggage(key)) is not None - } - if not baggage_values: - return event - - if tag_values := {key: str(baggage_values[key]) for key in baggage_keys if key in baggage_values}: - event.setdefault("tags", {}).update(tag_values) + configured_keys: typing.Final = baggage_keys.union(baggage_url_templates) + snapshot: typing.Final = _find_sentry_opentelemetry_baggage_snapshot(hint) + baggage_values: typing.Final = ( + snapshot if snapshot is not None else {key: baggage.get_baggage(key) for key in configured_keys} + ) + + for key in baggage_keys: + if (value := baggage_values.get(key)) is not None: + event.setdefault("tags", {})[key] = str(value) + elif tags := event.get("tags"): + tags.pop(key, None) for key, url_template in baggage_url_templates.items(): placeholder = f"{{{key}}}" - if key not in baggage_values or placeholder not in url_template: - continue - event.setdefault("extra", {})[f"{key}_url"] = url_template.replace( - placeholder, - urllib.parse.quote(str(baggage_values[key]), safe=""), - ) + extra_key = f"{key}_url" + if (value := baggage_values.get(key)) is not None and placeholder in url_template: + event.setdefault("extra", {})[extra_key] = url_template.replace( + placeholder, + urllib.parse.quote(str(value), safe=""), + ) + elif extra := event.get("extra"): + extra.pop(extra_key, None) return event @@ -126,6 +195,17 @@ def is_ready(self) -> bool: return bool(self.instrument_config.sentry_dsn) def bootstrap(self) -> None: + baggage_integration: typing.Final = ( + SentryOpentelemetryBaggageIntegration( + self.instrument_config.sentry_opentelemetry_baggage_keys, + self.instrument_config.sentry_opentelemetry_baggage_url_templates, + ) + if ( + self.instrument_config.sentry_opentelemetry_baggage_keys + or self.instrument_config.sentry_opentelemetry_baggage_url_templates + ) + else None + ) sentry_sdk.init( dsn=self.instrument_config.sentry_dsn, sample_rate=self.instrument_config.sentry_sample_rate, @@ -153,7 +233,10 @@ def bootstrap(self) -> None: else None, self.instrument_config.sentry_before_send, ), - integrations=self.instrument_config.sentry_integrations, + integrations=[ + *self.instrument_config.sentry_integrations, + *([baggage_integration] if baggage_integration else []), + ], **self.instrument_config.sentry_additional_params, ) if self.instrument_config.sentry_tags: diff --git a/tests/bootstrappers/test_faststream.py b/tests/bootstrappers/test_faststream.py index c7fb425..a3d5e46 100644 --- a/tests/bootstrappers/test_faststream.py +++ b/tests/bootstrappers/test_faststream.py @@ -13,6 +13,7 @@ from faststream.redis.opentelemetry import RedisTelemetryMiddleware from faststream.redis.prometheus import RedisPrometheusMiddleware +from microbootstrap import opentelemetry_baggage_scope from microbootstrap.bootstrappers.faststream import FastStreamBootstrapper from microbootstrap.config.faststream import FastStreamConfig from microbootstrap.instruments.health_checks_instrument import HealthChecksConfig @@ -192,6 +193,81 @@ def record_error_tag(*args: typing.Any, **kwargs: typing.Any) -> None: # noqa: assert conversation_id_tag not in sentry_sdk.get_isolation_scope()._tags # noqa: SLF001 +async def test_faststream_sentry_automatic_errors_use_concurrent_baggage_snapshots( + 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_captured = asyncio.Event() + captured_tags: dict[str, str | None] = {} + + init = mock.Mock() + monkeypatch.setattr(sentry_sdk, "init", init) + minimal_sentry_config.sentry_tags = None + minimal_sentry_config.sentry_opentelemetry_baggage_keys = {conversation_id_tag} + + @broker.subscriber(channel) + async def handler(conversation_id: str) -> None: + with opentelemetry_baggage_scope({conversation_id_tag: conversation_id}): + if conversation_id == "first": + first_started.set() + await second_started.wait() + await second_captured.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() + + baggage_integration: typing.Final = next( + integration + for integration in init.call_args.kwargs["integrations"] + if integration.identifier == "microbootstrap_opentelemetry_baggage" + ) + client = mock.Mock() + client.get_integration.return_value = baggage_integration + monkeypatch.setattr(sentry_sdk, "get_client", mock.Mock(return_value=client)) + before_send: typing.Final = init.call_args.kwargs["before_send"] + original_log = broker.config.logger.log + + def record_automatic_error(*args: typing.Any, **kwargs: typing.Any) -> None: # noqa: ANN401 + if kwargs.get("log_level") == logging.ERROR: + exception = typing.cast("ValueError", kwargs["exc_info"]) + event = before_send( + {}, + {"exc_info": (type(exception), exception, exception.__traceback__)}, + ) + captured_tags[str(exception)] = event.get("tags", {}).get(conversation_id_tag) + if str(exception) == "second": + second_captured.set() + original_log(*args, **kwargs) + + monkeypatch.setattr(broker.config.logger, "log", record_automatic_error) + + 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 == {"first": "first", "second": "second"} + + async def test_faststream_sentry_isolates_broker_configured_on_startup( broker: RedisBroker, minimal_sentry_config: SentryConfig, diff --git a/tests/instruments/test_opentelemetry.py b/tests/instruments/test_opentelemetry.py index dd928f4..0ebc7c1 100644 --- a/tests/instruments/test_opentelemetry.py +++ b/tests/instruments/test_opentelemetry.py @@ -48,6 +48,22 @@ def test_opentelemetry_baggage_scope_overrides_removes_and_restores_values() -> context.detach(outer_token) +def test_opentelemetry_baggage_scope_does_not_replace_exception_when_snapshot_fails( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + opentelemetry_instrument, + "snapshot_sentry_opentelemetry_baggage", + Mock(side_effect=RuntimeError("snapshot failed")), + ) + + with ( + pytest.raises(ValueError, match="application error"), + opentelemetry_baggage_scope({"conversation_id": "conversation-1"}), + ): + raise ValueError("application error") + + @pytest.mark.parametrize( ("span_kind", "expected_attribute"), [ diff --git a/tests/instruments/test_sentry.py b/tests/instruments/test_sentry.py index 9e7fa1a..dfeccc5 100644 --- a/tests/instruments/test_sentry.py +++ b/tests/instruments/test_sentry.py @@ -1,4 +1,5 @@ from __future__ import annotations +import builtins import copy import logging import typing @@ -7,19 +8,23 @@ import fastapi import litestar import pytest +import sentry_sdk 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 import opentelemetry_baggage_scope 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 ( SENTRY_EXTRA_OTEL_TRACE_ID_KEY, SENTRY_EXTRA_OTEL_TRACE_URL_KEY, + SENTRY_OTEL_BAGGAGE_SNAPSHOT_ATTRIBUTE, SentryInstrument, + SentryOpentelemetryBaggageIntegration, add_trace_url_to_event, enrich_sentry_event_from_opentelemetry_baggage, enrich_sentry_event_from_structlog_log, @@ -159,6 +164,17 @@ def test_modify(self, event_before: sentry_types.Event, event_after: sentry_type TRACE_URL_TEMPLATE = "https://example.com/traces/{trace_id}" +CONVERSATION_LOGS_URL_TEMPLATE = "https://example.com/logs/{conversation_id}" + + +def _configure_sentry_baggage_integration(monkeypatch: pytest.MonkeyPatch) -> None: + integration: typing.Final = SentryOpentelemetryBaggageIntegration( + {"conversation_id"}, + {"conversation_id": CONVERSATION_LOGS_URL_TEMPLATE}, + ) + client = mock.Mock() + client.get_integration.return_value = integration + monkeypatch.setattr(sentry_sdk, "get_client", mock.Mock(return_value=client)) class TestSentryAddTraceUrlToEvent: @@ -271,6 +287,206 @@ def test_keeps_attached_contexts_isolated(self) -> None: assert [result["tags"]["conversation_id"] for result in results] == ["first", "second"] + @pytest.mark.parametrize( + "hint", + [ + {"exc_info": ()}, + {"exc_info": (RuntimeError, "not-an-exception", None)}, + ], + ) + def test_invalid_exception_hint_falls_back_to_live_baggage(self, hint: sentry_types.Hint) -> None: + with opentelemetry_baggage_scope({"conversation_id": "live"}): + result = enrich_sentry_event_from_opentelemetry_baggage( + {"conversation_id"}, + {}, + {}, + hint, + ) + + assert result["tags"]["conversation_id"] == "live" + + def test_uses_exception_snapshot_after_baggage_scope_detaches(self, monkeypatch: pytest.MonkeyPatch) -> None: + _configure_sentry_baggage_integration(monkeypatch) + + with ( + pytest.raises(RuntimeError) as exc_info, + opentelemetry_baggage_scope( + { + "conversation_id": "conversation/id +", + "not_allowed": "secret", + } + ), + ): + raise RuntimeError("test error") + + assert baggage.get_baggage("conversation_id") is None + result = enrich_sentry_event_from_opentelemetry_baggage( + {"conversation_id"}, + {"conversation_id": CONVERSATION_LOGS_URL_TEMPLATE}, + {}, + {"exc_info": (RuntimeError, exc_info.value, exc_info.value.__traceback__)}, + ) + + assert result["tags"] == {"conversation_id": "conversation/id +"} + assert result["extra"] == { + "conversation_id_url": "https://example.com/logs/conversation%2Fid%20%2B", + } + assert getattr(exc_info.value, SENTRY_OTEL_BAGGAGE_SNAPSHOT_ATTRIBUTE) == { + "conversation_id": "conversation/id +", + } + + def test_nested_baggage_scopes_keep_innermost_snapshot(self, monkeypatch: pytest.MonkeyPatch) -> None: + _configure_sentry_baggage_integration(monkeypatch) + + with ( + pytest.raises(RuntimeError) as exc_info, + opentelemetry_baggage_scope({"conversation_id": "outer"}), + opentelemetry_baggage_scope({"conversation_id": "inner"}), + ): + raise RuntimeError("test error") + + result = enrich_sentry_event_from_opentelemetry_baggage( + {"conversation_id"}, + {}, + {}, + {"exc_info": (RuntimeError, exc_info.value, exc_info.value.__traceback__)}, + ) + + assert result["tags"]["conversation_id"] == "inner" + + def test_unhashable_baggage_value_does_not_replace_original_exception( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + _configure_sentry_baggage_integration(monkeypatch) + baggage_value: list[str] = ["conversation-1"] + + with ( + pytest.raises(RuntimeError, match="test error") as exc_info, + opentelemetry_baggage_scope({"conversation_id": baggage_value}), + ): + raise RuntimeError("test error") + + assert getattr(exc_info.value, SENTRY_OTEL_BAGGAGE_SNAPSHOT_ATTRIBUTE) == { + "conversation_id": baggage_value, + } + + def test_missing_snapshot_value_removes_stale_event_fields(self, monkeypatch: pytest.MonkeyPatch) -> None: + _configure_sentry_baggage_integration(monkeypatch) + outer_token = attach(baggage.set_baggage("conversation_id", "parent", context=Context())) + + try: + with pytest.raises(RuntimeError) as exc_info, opentelemetry_baggage_scope({"conversation_id": None}): + raise RuntimeError("test error") + + assert baggage.get_baggage("conversation_id") == "parent" + result = enrich_sentry_event_from_opentelemetry_baggage( + {"conversation_id"}, + {"conversation_id": CONVERSATION_LOGS_URL_TEMPLATE}, + { + "tags": {"conversation_id": "stale", "existing": "tag"}, + "extra": {"conversation_id_url": "stale", "existing": "extra"}, + }, + {"exc_info": (RuntimeError, exc_info.value, exc_info.value.__traceback__)}, + ) + finally: + detach(outer_token) + + assert result["tags"] == {"existing": "tag"} + assert result["extra"] == {"existing": "extra"} + + def test_caught_exception_snapshot_does_not_contaminate_later_event( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + _configure_sentry_baggage_integration(monkeypatch) + + with pytest.raises(RuntimeError), opentelemetry_baggage_scope({"conversation_id": "caught"}): + raise RuntimeError("caught error") + + with opentelemetry_baggage_scope({"conversation_id": "later"}): + result = enrich_sentry_event_from_opentelemetry_baggage( + {"conversation_id"}, + {}, + {}, + {}, + ) + + assert result["tags"]["conversation_id"] == "later" + + def test_wrapped_exception_uses_first_snapshot_in_visible_chain(self) -> None: + cause = RuntimeError("cause") + setattr(cause, SENTRY_OTEL_BAGGAGE_SNAPSHOT_ATTRIBUTE, {"conversation_id": "cause"}) + wrapper = ValueError("wrapper") + wrapper.__cause__ = cause + + cause_result = enrich_sentry_event_from_opentelemetry_baggage( + {"conversation_id"}, + {}, + {}, + {"exc_info": (ValueError, wrapper, None)}, + ) + setattr(wrapper, SENTRY_OTEL_BAGGAGE_SNAPSHOT_ATTRIBUTE, {"conversation_id": "wrapper"}) + wrapper_result = enrich_sentry_event_from_opentelemetry_baggage( + {"conversation_id"}, + {}, + {}, + {"exc_info": (ValueError, wrapper, None)}, + ) + + assert cause_result["tags"]["conversation_id"] == "cause" + assert wrapper_result["tags"]["conversation_id"] == "wrapper" + + @pytest.mark.skipif(not hasattr(builtins, "ExceptionGroup"), reason="ExceptionGroup requires Python 3.11+") + def test_exception_group_uses_first_nested_snapshot(self) -> None: + first_exception = RuntimeError("first") + setattr(first_exception, SENTRY_OTEL_BAGGAGE_SNAPSHOT_ATTRIBUTE, {"conversation_id": "first"}) + second_exception = RuntimeError("second") + setattr(second_exception, SENTRY_OTEL_BAGGAGE_SNAPSHOT_ATTRIBUTE, {"conversation_id": "second"}) + exception_group_type = vars(builtins)["ExceptionGroup"] + exception_group: BaseException = exception_group_type( + "concurrent failures", [first_exception, second_exception] + ) + + result = enrich_sentry_event_from_opentelemetry_baggage( + {"conversation_id"}, + {}, + {}, + {"exc_info": (type(exception_group), exception_group, None)}, + ) + + assert result["tags"]["conversation_id"] == "first" + + def test_exception_chain_cycle_falls_back_to_live_baggage(self) -> None: + exception = RuntimeError("cycle") + exception.__cause__ = exception + + with opentelemetry_baggage_scope({"conversation_id": "live"}): + result = enrich_sentry_event_from_opentelemetry_baggage( + {"conversation_id"}, + {}, + {}, + {"exc_info": (RuntimeError, exception, None)}, + ) + + assert result["tags"]["conversation_id"] == "live" + + def test_scope_without_sentry_integration_does_not_attach_snapshot( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + client = mock.Mock() + client.get_integration.return_value = None + monkeypatch.setattr(sentry_sdk, "get_client", mock.Mock(return_value=client)) + + with ( + pytest.raises(RuntimeError) as exc_info, + opentelemetry_baggage_scope({"conversation_id": "conversation-1"}), + ): + raise RuntimeError("test error") + + assert not hasattr(exc_info.value, SENTRY_OTEL_BAGGAGE_SNAPSHOT_ATTRIBUTE) + def test_sentry_bootstrap_composes_baggage_enrichment_before_custom_callback( minimal_sentry_config: SentryConfig, @@ -278,6 +494,9 @@ def test_sentry_bootstrap_composes_baggage_enrichment_before_custom_callback( ) -> None: custom_before_send = mock.Mock(side_effect=lambda event, _hint: event) minimal_sentry_config.sentry_opentelemetry_baggage_keys = {"conversation_id"} + minimal_sentry_config.sentry_opentelemetry_baggage_url_templates = { + "conversation_id": CONVERSATION_LOGS_URL_TEMPLATE, + } minimal_sentry_config.sentry_before_send = custom_before_send init = mock.Mock() monkeypatch.setattr("sentry_sdk.init", init) @@ -291,6 +510,16 @@ def test_sentry_bootstrap_composes_baggage_enrichment_before_custom_callback( assert result["tags"]["conversation_id"] == "conversation-1" assert custom_before_send.call_args.args[0]["tags"]["conversation_id"] == "conversation-1" + baggage_integration: typing.Final = next( + integration + for integration in init.call_args.kwargs["integrations"] + if isinstance(integration, SentryOpentelemetryBaggageIntegration) + ) + assert baggage_integration.baggage_keys == {"conversation_id"} + assert baggage_integration.baggage_url_templates == { + "conversation_id": CONVERSATION_LOGS_URL_TEMPLATE, + } + assert baggage_integration.setup_once() is None @pytest.mark.parametrize("logger_instance", [structlog.get_logger(__name__), logging.getLogger(__name__)])