feat: Add enrichers for attribute-based telemetry

sentry_sdk/_types.py

@@ -112,6 +112,8 @@ def substituted_because_contains_sensitive_data(cls) -> "AnnotatedValue":
     from typing import Type
     from typing_extensions import Literal, TypedDict
 
+    from sentry_sdk.traces import StreamedSpan
+
     class SDKInfo(TypedDict):
         name: str
         version: str
@@ -293,11 +295,15 @@ class SDKInfo(TypedDict):
     # TODO: Make a proper type definition for this (PRs welcome!)
     SamplingContext = Dict[str, Any]
 
+    # New-style, attribute-based telemetry
+    Telemetry = Union[Metric, Log, StreamedSpan]
+
     EventProcessor = Callable[[Event, Hint], Optional[Event]]
     ErrorProcessor = Callable[[Event, ExcInfo], Optional[Event]]
     BreadcrumbProcessor = Callable[[Breadcrumb, BreadcrumbHint], Optional[Breadcrumb]]
     TransactionProcessor = Callable[[Event, Hint], Optional[Event]]
     LogProcessor = Callable[[Log, Hint], Optional[Log]]
+    Enricher = Callable[[Telemetry], Telemetry]
 
     TracesSampler = Callable[[SamplingContext], Union[float, int, bool]]
 

sentry_sdk/integrations/_asgi_common.py

@@ -12,6 +12,7 @@
     from typing import Union
     from typing_extensions import Literal
 
+    from sentry_sdk._types import Attributes
     from sentry_sdk.utils import AnnotatedValue
 
 
@@ -105,3 +106,32 @@ def _get_request_data(asgi_scope: "Any") -> "Dict[str, Any]":
         request_data["env"] = {"REMOTE_ADDR": _get_ip(asgi_scope)}
 
     return request_data
+
+
+def _get_request_attributes(asgi_scope: "Any") -> "Dict[str, Any]":
+    """
+    Return attributes related to the HTTP request from the ASGI scope.
+    """
+    attributes: "Attributes" = {}
+
+    ty = asgi_scope["type"]
+    if ty in ("http", "websocket"):
+        attributes["http.request.method"] = asgi_scope.get("method")
+
+        headers = _filter_headers(_get_headers(asgi_scope))
+        # TODO[span-first]: Correctly merge headers if duplicate
+        for header, value in headers.items():
+            attributes[f"http.request.headers.{header.lower()}"] = [value]
+
+        attributes["http.query"] = _get_query(asgi_scope)
+
+        attributes["url.full"] = _get_url(
+            asgi_scope, "http" if ty == "http" else "ws", headers.get("host")
+        )
+
+    # TODO[span-first]: Figure out where to put REMOTE_ADDR
+    # client = asgi_scope.get("client")
+    # if client and should_send_default_pii():
+    #    request_data["env"] = {"REMOTE_ADDR": _get_ip(asgi_scope)}
+
+    return attributes

sentry_sdk/integrations/asgi.py

@@ -15,6 +15,7 @@
 from sentry_sdk.consts import OP
 from sentry_sdk.integrations._asgi_common import (
     _get_headers,
+    _get_request_attributes,
     _get_request_data,
     _get_url,
 )
@@ -23,7 +24,7 @@
     nullcontext,
 )
 from sentry_sdk.sessions import track_session
-from sentry_sdk.traces import StreamedSpan
+from sentry_sdk.traces import SegmentSource, StreamedSpan
 from sentry_sdk.tracing import (
     SOURCE_FOR_STYLE,
     Transaction,
@@ -52,7 +53,7 @@
     from typing import Tuple
     from typing import Union
 
-    from sentry_sdk._types import Attributes, Event, Hint
+    from sentry_sdk._types import Attributes, Event, Hint, Telemetry
     from sentry_sdk.tracing import Span
 
 
@@ -214,7 +215,9 @@
                     sentry_scope.clear_breadcrumbs()
                     sentry_scope._name = "asgi"
                     processor = partial(self.event_processor, asgi_scope=scope)
+                    enricher = partial(self.enricher, asgi_scope=scope)
                     sentry_scope.add_event_processor(processor)
+                    sentry_scope._add_enricher(enricher)
 
                     ty = scope["type"]
                     (
@@ -382,6 +385,32 @@
 
         return event
 
+    def enricher(self, telemetry: "Telemetry", asgi_scope: "Any") -> "Telemetry":
+        request_attributes = _get_request_attributes(asgi_scope)
+        telemetry.set_attributes(request_attributes)
+
+        # Only set transaction name if not already set by Starlette or FastAPI (or other frameworks)
+        segment = telemetry.segment
+        source = segment.get_attributes.get("sentry.span.source")
+        already_set = (
+            segment is not None
+            and segment.name != _DEFAULT_TRANSACTION_NAME
+            and source
+            in [
+                SegmentSource.COMPONENT,
+                SegmentSource.ROUTE,
+                SegmentSource.CUSTOM,
+            ]
+        )
+        if not already_set:
+            name, source = self._get_transaction_name_and_source(
+                self.transaction_style, asgi_scope
+            )
+            segment.name = name
+            segment.set_attribute("sentry.span.source", source)
+
+        return telemetry
+
     # Helper functions.
     #
     # Note: Those functions are not public API. If you want to mutate request

sentry_sdk/integrations/stdlib.py

@@ -7,7 +7,7 @@
 import sentry_sdk
 from sentry_sdk.consts import OP, SPANDATA
 from sentry_sdk.integrations import Integration
-from sentry_sdk.scope import add_global_event_processor
+from sentry_sdk.scope import add_global_event_processor, global_enrichers
 from sentry_sdk.tracing_utils import (
     EnvironHeaders,
     should_propagate_trace,
@@ -61,6 +61,22 @@ def add_python_runtime_context(
 
             return event
 
+        def add_python_runtime_context_enricher(telemetry):
+            if sentry_sdk.get_client().get_integration(StdlibIntegration) is not None:
+                telemetry.set_attribute(
+                    "process.runtime.name", _RUNTIME_CONTEXT["name"]
+                )
+                telemetry.set_attribute(
+                    "process.runtime.version", _RUNTIME_CONTEXT["version"]
+                )
+                telemetry.set_attribute(
+                    "process.runtime.description", _RUNTIME_CONTEXT["build"]
+                )
+
+            return telemetry
+
+        global_enrichers.append(add_python_runtime_context_enricher)
+
 
 def _install_httplib() -> None:
     real_putrequest = HTTPConnection.putrequest

sentry_sdk/scope.py

@@ -81,6 +81,7 @@
         AttributeValue,
         Breadcrumb,
         BreadcrumbHint,
+        Enricher,
         ErrorProcessor,
         Event,
         EventProcessor,
@@ -121,6 +122,7 @@
 _current_scope = ContextVar("current_scope", default=None)
 
 global_event_processors: "List[EventProcessor]" = []
+global_enrichers: "List[Enricher]" = []
 
 # A function returning a (trace_id, span_id) tuple
 # from an external tracing source (such as otel)
@@ -226,6 +228,7 @@ class Scope:
         "_gen_ai_conversation_id",
         "_event_processors",
         "_error_processors",
+        "_enrichers",
         "_should_capture",
         "_span",
         "_session",
@@ -249,6 +252,7 @@ def __init__(
 
         self._event_processors: "List[EventProcessor]" = []
         self._error_processors: "List[ErrorProcessor]" = []
+        self._enrichers: "List[Enricher]" = []
 
         self._name: "Optional[str]" = None
         self._propagation_context: "Optional[PropagationContext]" = None
@@ -289,6 +293,7 @@ def __copy__(self) -> "Scope":
         rv._n_breadcrumbs_truncated = self._n_breadcrumbs_truncated
         rv._gen_ai_original_message_count = self._gen_ai_original_message_count.copy()
         rv._event_processors = self._event_processors.copy()
+        rv._enrichers = self._enrichers.copy()
         rv._error_processors = self._error_processors.copy()
         rv._propagation_context = self._propagation_context
 
@@ -1573,6 +1578,20 @@ def add_event_processor(
 
         self._event_processors.append(func)
 
+    def _add_enricher(
+        self,
+        func: "Enricher",
+    ) -> None:
+        """Register a scope local enricher on the scope."""
+        if len(self._enrichers) > 20:
+            logger.warning(
+                "Too many enrichers on scope! Clearing list to free up some memory: %r",
+                self._enrichers,
+            )
+            del self._enrichers[:]
+
+        self._enrichers.append(func)
+
     def add_error_processor(
         self,
         func: "ErrorProcessor",
@@ -1842,6 +1861,38 @@ def apply_to_telemetry(self, telemetry: "Union[Log, Metric, StreamedSpan]") -> N
         self._apply_scope_attributes_to_telemetry(telemetry)
         self._apply_user_attributes_to_telemetry(telemetry)
 
+        self._run_enrichers(telemetry)
+
+    def _run_enrichers(
+        self, telemetry: "Union[Log, Metric, StreamedSpan]"
+    ) -> "Union[Log, Metric, StreamedSpan]":
+        """
+        Runs enrichers on the telemetry and returns the modified telemetry.
+        """
+        if not isinstance(telemetry, StreamedSpan):
+            # Currently we don't support enrichers for non-span telemetry
+            return
+
+        # Get scopes without creating them to prevent infinite recursion
+        isolation_scope = _isolation_scope.get()
+        current_scope = _current_scope.get()
+
+        enrichers = chain(
+            global_enrichers,
+            _global_scope and _global_scope._enrichers or [],
+            isolation_scope and isolation_scope._enrichers or [],
+            current_scope and current_scope._enrichers or [],
+        )
+
+        for enricher in enrichers:
+            new_telemetry = telemetry
+            with capture_internal_exceptions():
+                new_telemetry = enricher(telemetry)
+            if new_telemetry is not None:
+                telemetry = new_telemetry
+
+        return telemetry
+
     def update_from_scope(self, scope: "Scope") -> None:
         """Update the scope with another scope's data."""
         if scope._level is not None:

tests/tracing/test_span_streaming.py

@@ -1503,6 +1503,42 @@ def test_profile_stops_when_segment_ends(
     assert get_profiler_id() is None, "profiler should have stopped"
 
 
+def test_enrichers(sentry_init, capture_envelopes):
+    sentry_init(
+        traces_sample_rate=1.0,
+        _experiments={
+            "trace_lifecycle": "stream",
+        },
+    )
+
+    envelopes = capture_envelopes()
+
+    def enricher(telemetry):
+        telemetry.set_attribute("enriched", True)
+
+    with sentry_sdk.new_scope() as scope:
+        scope._add_enricher(enricher)
+
+        with sentry_sdk.traces.start_span(name="enriched segment"):
+            pass
+
+    with sentry_sdk.traces.start_span(name="not enriched segment"):
+        pass
+
+    sentry_sdk.get_client().flush()
+
+    spans = envelopes_to_spans(envelopes)
+    assert len(spans) == 2
+    span1, span2 = spans
+
+    assert span1["name"] == "enriched segment"
+    assert "enriched" in span1["attributes"]
+    assert span1["attributes"]["enriched"] is True
+
+    assert span2["name"] == "not enriched segment"
+    assert "enriched" not in span2["attributes"]
+
+
 def test_transport_format(sentry_init, capture_envelopes):
     sentry_init(
         server_name="test-server",
@@ -1539,6 +1575,12 @@ def test_transport_format(sentry_init, capture_envelopes):
                 "start_timestamp": mock.ANY,
                 "end_timestamp": mock.ANY,
                 "attributes": {
+                    "process.runtime.name": {"value": mock.ANY, "type": "string"},
+                    "process.runtime.description": {
+                        "value": mock.ANY,
+                        "type": "string",
+                    },
+                    "process.runtime.version": {"value": mock.ANY, "type": "string"},
                     "sentry.span.source": {"value": "custom", "type": "string"},
                     "thread.id": {"value": mock.ANY, "type": "string"},
                     "thread.name": {"value": "MainThread", "type": "string"},