feat(asgi): Make integration fully span first compatible (#5920)

In span first, there are no event processors. Therefore, we need to be
able to set the data we were setting in event processors differently.

As we migrate our integrations one by one, this will be an exercise in
whether it's possible to achieve this without some sort of
callback/lifecycle hooks. So far, in ASGI, it seems we can get by by
simply using `scope.set_attribute()` for setting request related data,
and updating the segment name/source just before the span ends.

Adding this enables us to actually test the new functionality.

To sum up, this PR:
- makes the ASGI integration fully span-first-compatible: data set on
spans currently should appear as attributes in span first
- introduces a test helper that makes it easier to unpack envelope items
(useful especially when a test checks both old- and new-style telemetry)
- fully tests the ASGI span streaming implementation

Like in other span first PRs, there is quite a bit of intentional code
duplication so that it's easier to remove the legacy implementation in
the next major.


#### Issues
* Closes #5722
* Closes
https://linear.app/getsentry/issue/PY-2152/event-processor-replacement-for-streamed-spans

Author: sentrivana

sentry_sdk/integrations/_asgi_common.py

@@ -105,3 +105,33 @@ 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: "dict[str, Any]" = {}
+
+    ty = asgi_scope["type"]
+    if ty in ("http", "websocket"):
+        if asgi_scope.get("method"):
+            attributes["http.request.method"] = asgi_scope["method"].upper()
+
+        headers = _filter_headers(_get_headers(asgi_scope), use_annotated_value=False)
+        for header, value in headers.items():
+            attributes[f"http.request.header.{header.lower()}"] = value
+
+        query = _get_query(asgi_scope)
+        if query:
+            attributes["http.query"] = query
+
+        attributes["url.full"] = _get_url(
+            asgi_scope, "http" if ty == "http" else "ws", headers.get("host")
+        )
+
+    client = asgi_scope.get("client")
+    if client and should_send_default_pii():
+        attributes["client.address"] = _get_ip(asgi_scope)
+
+    return attributes

sentry_sdk/integrations/_wsgi_common.py

@@ -3,6 +3,7 @@
 from copy import deepcopy
 
 import sentry_sdk
+from sentry_sdk._types import SENSITIVE_DATA_SUBSTITUTE
 from sentry_sdk.scope import should_send_default_pii
 from sentry_sdk.utils import AnnotatedValue, logger
 
@@ -211,16 +212,19 @@ def _is_json_content_type(ct: "Optional[str]") -> bool:
 
 def _filter_headers(
     headers: "Mapping[str, str]",
+    use_annotated_value: bool = True,
 ) -> "Mapping[str, Union[AnnotatedValue, str]]":
     if should_send_default_pii():
         return headers
 
+    substitute: "Union[AnnotatedValue, str]"
+    if use_annotated_value:
+        substitute = AnnotatedValue.removed_because_over_size_limit()
+    else:
+        substitute = SENSITIVE_DATA_SUBSTITUTE
+
     return {
-        k: (
-            v
-            if k.upper().replace("-", "_") not in SENSITIVE_HEADERS
-            else AnnotatedValue.removed_because_over_size_limit()
-        )
+        k: (v if k.upper().replace("-", "_") not in SENSITIVE_HEADERS else substitute)
         for k, v in headers.items()
     }
 

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,11 @@
     nullcontext,
 )
 from sentry_sdk.sessions import track_session
-from sentry_sdk.traces import StreamedSpan
+from sentry_sdk.traces import (
+    StreamedSpan,
+    SegmentSource,
+    SOURCE_FOR_STYLE as SEGMENT_SOURCE_FOR_STYLE,
+)
 from sentry_sdk.tracing import (
     SOURCE_FOR_STYLE,
     Transaction,
@@ -40,6 +45,7 @@
     _get_installed_modules,
     reraise,
     capture_internal_exceptions,
+    qualname_from_function,
 )
 
 from typing import TYPE_CHECKING
@@ -235,7 +241,7 @@ async def _run_app(
                                 transaction_source, "value", transaction_source
                             ),
                             "sentry.origin": self.span_origin,
-                            "asgi.type": ty,
+                            "network.protocol.name": ty,
                         }
 
                         if ty in ("http", "websocket"):
@@ -302,6 +308,12 @@ async def _run_app(
                         )
 
                     with span_ctx as span:
+                        if isinstance(span, StreamedSpan):
+                            for attribute, value in _get_request_attributes(
+                                scope
+                            ).items():
+                                span.set_attribute(attribute, value)
+
                         try:
 
                             async def _sentry_wrapped_send(
@@ -336,6 +348,7 @@ async def _sentry_wrapped_send(
                                 return await self.app(
                                     scope, receive, _sentry_wrapped_send
                                 )
+
                         except Exception as exc:
                             suppress_chained_exceptions = (
                                 sentry_sdk.get_client()
@@ -350,6 +363,28 @@ async def _sentry_wrapped_send(
                             with capture_internal_exceptions():
                                 self._capture_request_exception(exc)
                             reraise(*exc_info)
+
+                        finally:
+                            if isinstance(span, StreamedSpan):
+                                already_set = (
+                                    span is not None
+                                    and span.name != _DEFAULT_TRANSACTION_NAME
+                                    and span.get_attributes().get("sentry.span.source")
+                                    in [
+                                        SegmentSource.COMPONENT.value,
+                                        SegmentSource.ROUTE.value,
+                                        SegmentSource.CUSTOM.value,
+                                    ]
+                                )
+                                with capture_internal_exceptions():
+                                    if not already_set:
+                                        name, source = (
+                                            self._get_segment_name_and_source(
+                                                self.transaction_style, scope
+                                            )
+                                        )
+                                        span.name = name
+                                        span.set_attribute("sentry.span.source", source)
         finally:
             _asgi_middleware_applied.set(False)
 
@@ -424,3 +459,40 @@ def _get_transaction_name_and_source(
             return name, source
 
         return name, source
+
+    def _get_segment_name_and_source(
+        self: "SentryAsgiMiddleware", segment_style: str, asgi_scope: "Any"
+    ) -> "Tuple[str, str]":
+        name = None
+        source = SEGMENT_SOURCE_FOR_STYLE[segment_style].value
+        ty = asgi_scope.get("type")
+
+        if segment_style == "endpoint":
+            endpoint = asgi_scope.get("endpoint")
+            # Webframeworks like Starlette mutate the ASGI env once routing is
+            # done, which is sometime after the request has started. If we have
+            # an endpoint, overwrite our generic transaction name.
+            if endpoint:
+                name = qualname_from_function(endpoint) or ""
+            else:
+                name = _get_url(asgi_scope, "http" if ty == "http" else "ws", host=None)
+                source = SegmentSource.URL.value
+
+        elif segment_style == "url":
+            # FastAPI includes the route object in the scope to let Sentry extract the
+            # path from it for the transaction name
+            route = asgi_scope.get("route")
+            if route:
+                path = getattr(route, "path", None)
+                if path is not None:
+                    name = path
+            else:
+                name = _get_url(asgi_scope, "http" if ty == "http" else "ws", host=None)
+                source = SegmentSource.URL.value
+
+        if name is None:
+            name = _DEFAULT_TRANSACTION_NAME
+            source = SegmentSource.ROUTE.value
+            return name, source
+
+        return name, source

sentry_sdk/traces.py

@@ -259,6 +259,7 @@ def __init__(
         self._name: str = name
         self._active: bool = active
         self._attributes: "Attributes" = {}
+
         if attributes:
             for attribute, value in attributes.items():
                 self.set_attribute(attribute, value)
@@ -287,7 +288,6 @@ def __init__(
         self._span_id: "Optional[str]" = None
 
         self._status = SpanStatus.OK.value
-        self.set_attribute("sentry.span.source", SegmentSource.CUSTOM.value)
 
         self._update_active_thread()
 

tests/conftest.py

@@ -7,6 +7,7 @@
 import brotli
 import gzip
 import io
+from dataclasses import dataclass
 from threading import Thread
 from contextlib import contextmanager
 from http.server import BaseHTTPRequestHandler, HTTPServer
@@ -320,6 +321,52 @@ def append_envelope(envelope):
     return inner
 
 
+@dataclass
+class UnwrappedItem:
+    type: str
+    payload: dict
+
+
+@pytest.fixture
+def capture_items(monkeypatch):
+    """
+    Capture envelope payload, unfurling individual items.
+
+    Makes it easier to work with both events and attribute-based telemetry in
+    one test.
+    """
+
+    def inner(*types):
+        telemetry = []
+        test_client = sentry_sdk.get_client()
+        old_capture_envelope = test_client.transport.capture_envelope
+
+        def append_envelope(envelope):
+            for item in envelope:
+                if types and item.type not in types:
+                    continue
+
+                if item.type in ("metric", "log", "span"):
+                    for i in item.payload.json["items"]:
+                        t = {k: v for k, v in i.items() if k != "attributes"}
+                        t["attributes"] = {
+                            k: v["value"] for k, v in i["attributes"].items()
+                        }
+                        telemetry.append(UnwrappedItem(type=item.type, payload=t))
+                else:
+                    telemetry.append(
+                        UnwrappedItem(type=item.type, payload=item.payload.json)
+                    )
+
+            return old_capture_envelope(envelope)
+
+        monkeypatch.setattr(test_client.transport, "capture_envelope", append_envelope)
+
+        return telemetry
+
+    return inner
+
+
 @pytest.fixture
 def capture_record_lost_event_calls(monkeypatch):
     def inner():