feat(sanic): Support span streaming

* Moved span end state cleanup logic to a new `_end_cleanup` sub-method
* we need this because sanic retroactively decides not to queue a span
  and in span-first there is no way to unsample a span like in the older
  system

Author: sl0thentr0py

.claude/settings.json

@@ -30,11 +30,15 @@
       "Bash(mv:*)",
       "Bash(source .venv/bin/activate)",
       "Bash(source tox.venv/bin/activate:*)",
+      "Bash(source .tox/*/bin/activate:*)",
       "Bash(tox:*)",
       "Bash(tox.venv/bin/tox:*)",
       "Bash(.tox/*/bin/python:*)",
       "Bash(.tox/*/bin/pytest:*)",
-      "Bash(.tox/*/bin/ruff:*)"
+      "Bash(.tox/*/bin/ruff:*)",
+      "Bash(ruff format:*)",
+      "Bash(ruff check:*)",
+      "Bash(mypy:*)"
     ],
     "deny": []
   }

sentry_sdk/integrations/sanic.py

@@ -6,11 +6,14 @@
 
 import sentry_sdk
 from sentry_sdk import continue_trace
-from sentry_sdk.consts import OP
+from sentry_sdk.consts import OP, SPANDATA
 from sentry_sdk.integrations import DidNotEnable, Integration, _check_minimum_version
 from sentry_sdk.integrations._wsgi_common import RequestExtractor, _filter_headers
 from sentry_sdk.integrations.logging import ignore_logger
+from sentry_sdk.scope import should_send_default_pii
+from sentry_sdk.traces import SegmentSource, StreamedSpan
 from sentry_sdk.tracing import TransactionSource
+from sentry_sdk.tracing_utils import has_span_streaming_enabled
 from sentry_sdk.utils import (
     CONTEXTVARS_ERROR_MESSAGE,
     HAS_REAL_CONTEXTVARS,
@@ -165,23 +168,43 @@ async def _context_enter(request: "Request") -> None:
     if not request.ctx._sentry_do_integration:
         return
 
+    client = sentry_sdk.get_client()
+    is_span_streaming_enabled = has_span_streaming_enabled(client.options)
+
     weak_request = weakref.ref(request)
     request.ctx._sentry_scope = sentry_sdk.isolation_scope()
     scope = request.ctx._sentry_scope.__enter__()
     scope.clear_breadcrumbs()
     scope.add_event_processor(_make_request_processor(weak_request))
 
-    transaction = continue_trace(
-        dict(request.headers),
-        op=OP.HTTP_SERVER,
-        # Unless the request results in a 404 error, the name and source will get overwritten in _set_transaction
-        name=request.path,
-        source=TransactionSource.URL,
-        origin=SanicIntegration.origin,
-    )
-    request.ctx._sentry_transaction = sentry_sdk.start_transaction(
-        transaction
-    ).__enter__()
+    if is_span_streaming_enabled:
+        sentry_sdk.traces.continue_trace(dict(request.headers))
+        scope.set_custom_sampling_context({"sanic_request": request})
+
+        span = sentry_sdk.traces.start_span(
+            # Unless the request results in a 404 error, the name and source
+            # will get overwritten in _set_transaction
+            name=request.path,
+            attributes={
+                "sentry.op": OP.HTTP_SERVER,
+                "sentry.origin": SanicIntegration.origin,
+                "sentry.span.source": SegmentSource.URL.value,
+            },
+            parent_span=None,
+        )
+        request.ctx._sentry_root_span = span
+    else:
+        transaction = continue_trace(
+            dict(request.headers),
+            op=OP.HTTP_SERVER,
+            # Unless the request results in a 404 error, the name and source will get overwritten in _set_transaction
+            name=request.path,
+            source=TransactionSource.URL,
+            origin=SanicIntegration.origin,
+        )
+        request.ctx._sentry_root_span = sentry_sdk.start_transaction(
+            transaction
+        ).__enter__()
 
 
 async def _context_exit(
@@ -194,16 +217,34 @@ async def _context_exit(
         integration = sentry_sdk.get_client().get_integration(SanicIntegration)
 
         response_status = None if response is None else response.status
+        should_sample = (
+            isinstance(integration, SanicIntegration)
+            and response_status not in integration._unsampled_statuses
+        )
 
         # This capture_internal_exceptions block has been intentionally nested here, so that in case an exception
         # happens while trying to end the transaction, we still attempt to exit the hub.
         with capture_internal_exceptions():
-            request.ctx._sentry_transaction.set_http_status(response_status)
-            request.ctx._sentry_transaction.sampled &= (
-                isinstance(integration, SanicIntegration)
-                and response_status not in integration._unsampled_statuses
-            )
-            request.ctx._sentry_transaction.__exit__(None, None, None)
+            span = request.ctx._sentry_root_span
+            if isinstance(span, StreamedSpan):
+                with capture_internal_exceptions():
+                    for attr, value in _get_request_attributes(request).items():
+                        span.set_attribute(attr, value)
+                if response_status is not None:
+                    span.set_attribute(SPANDATA.HTTP_STATUS_CODE, response_status)
+                    span.status = "error" if response_status >= 400 else "ok"
+
+                # If the status is in unsampled_statuses, skip ending the span so
+                # it is never queued for sending but just cleanup the state.
+                # The orphaned scope is discarded when the isolation scope exits below.
+                if should_sample:
+                    span._end()
+                else:
+                    span._end_cleanup()
+            else:
+                span.set_http_status(response_status)
+                span.sampled &= should_sample
+                span.__exit__(None, None, None)
 
         request.ctx._sentry_scope.__exit__(None, None, None)
 
@@ -315,6 +356,36 @@ def _capture_exception(exception: "Union[ExcInfo, BaseException]") -> None:
         sentry_sdk.capture_event(event, hint=hint)
 
 
+def _get_request_attributes(request: "Request") -> "Dict[str, Any]":
+    """
+    Return span attributes related to the HTTP request from a Sanic request.
+    """
+    attributes = {}  # type: Dict[str, Any]
+
+    if request.method:
+        attributes[SPANDATA.HTTP_REQUEST_METHOD] = request.method.upper()
+
+    headers = _filter_headers(dict(request.headers), use_annotated_value=False)
+    for header, value in headers.items():
+        attributes[f"{SPANDATA.HTTP_REQUEST_HEADER}.{header.lower()}"] = value
+
+    urlparts = urlsplit(request.url)
+
+    if urlparts.query:
+        attributes[SPANDATA.HTTP_QUERY] = urlparts.query
+
+    attributes[SPANDATA.URL_FULL] = request.url
+
+    if urlparts.scheme:
+        attributes[SPANDATA.NETWORK_PROTOCOL_NAME] = urlparts.scheme
+
+    if should_send_default_pii() and request.remote_addr:
+        attributes[SPANDATA.CLIENT_ADDRESS] = request.remote_addr
+        attributes[SPANDATA.USER_IP_ADDRESS] = request.remote_addr
+
+    return attributes
+
+
 def _make_request_processor(weak_request: "Callable[[], Request]") -> "EventProcessor":
     def sanic_processor(event: "Event", hint: "Optional[Hint]") -> "Optional[Event]":
         try:

sentry_sdk/traces.py

@@ -366,17 +366,7 @@ def _end(self, end_timestamp: "Optional[Union[float, datetime]]" = None) -> None
             # This span is already finished, ignore.
             return
 
-        # Stop the profiler
-        if self._is_segment() and self._continuous_profile is not None:
-            with capture_internal_exceptions():
-                self._continuous_profile.stop()
-
-        # Detach from scope
-        if self._active:
-            with capture_internal_exceptions():
-                old_span = self._previous_span_on_scope
-                del self._previous_span_on_scope
-                self._scope.streamed_span = old_span
+        self._end_cleanup()
 
         # Set attributes from the segment. These are set on span end on purpose
         # so that we have the best chance to capture the segment's final name
@@ -412,6 +402,21 @@ def _end(self, end_timestamp: "Optional[Union[float, datetime]]" = None) -> None
         # Finally, queue the span for sending to Sentry
         self._scope._capture_span(self)
 
+
+    def _end_cleanup(self):
+        # Stop the profiler
+        if self._is_segment() and self._continuous_profile is not None:
+            with capture_internal_exceptions():
+                self._continuous_profile.stop()
+
+        # Detach from scope
+        if self._active:
+            with capture_internal_exceptions():
+                old_span = self._previous_span_on_scope
+                del self._previous_span_on_scope
+                self._scope.streamed_span = old_span
+
+
     def get_attributes(self) -> "Attributes":
         return self._attributes
 

tests/integrations/sanic/test_sanic.py

@@ -360,6 +360,7 @@ def __init__(
 @pytest.mark.skipif(
     not PERFORMANCE_SUPPORTED, reason="Performance not supported on this Sanic version"
 )
+@pytest.mark.parametrize("span_streaming", [True, False])
 @pytest.mark.parametrize(
     "test_config",
     [
@@ -371,6 +372,14 @@ def __init__(
             expected_transaction_name="hi",
             expected_source=TransactionSource.COMPONENT,
         ),
+        TransactionTestConfig(
+            # Transaction for successful page load with query string
+            integration_args=(),
+            url="/message?foo=bar",
+            expected_status=200,
+            expected_transaction_name="hi",
+            expected_source=TransactionSource.COMPONENT,
+        ),
         TransactionTestConfig(
             # Transaction still recorded when we have an internal server error
             integration_args=(),
@@ -408,57 +417,121 @@ def test_transactions(
     sentry_init: "Any",
     app: "Any",
     capture_events: "Any",
+    capture_items: "Any",
+    span_streaming: bool,
 ) -> None:
     # Init the SanicIntegration with the desired arguments
     sentry_init(
         integrations=[SanicIntegration(*test_config.integration_args)],
         traces_sample_rate=1.0,
+        _experiments={"trace_lifecycle": "stream" if span_streaming else "static"},
     )
-    events = capture_events()
+
+    if span_streaming:
+        items = capture_items("span")
+    else:
+        events = capture_events()
 
     # Make request to the desired URL
     c = get_client(app)
     with c as client:
         _, response = client.get(test_config.url)
         assert response.status == test_config.expected_status
 
-    # Extract the transaction events by inspecting the event types. We should at most have 1 transaction event.
-    transaction_events = [
-        e for e in events if "type" in e and e["type"] == "transaction"
-    ]
-    assert len(transaction_events) <= 1
-
-    # Get the only transaction event, or set to None if there are no transaction events.
-    (transaction_event, *_) = [*transaction_events, None]
-
-    # We should have no transaction event if and only if we expect no transactions
-    assert (transaction_event is None) == (
-        test_config.expected_transaction_name is None
-    )
+    sentry_sdk.flush()
+
+    if span_streaming:
+        segments = [
+            i.payload
+            for i in items
+            if i.type == "span"
+            and i.payload["attributes"].get("sentry.origin") == "auto.http.sanic"
+        ]
+        assert len(segments) <= 1
+        (segment, *_) = [*segments, None]
+
+        assert (segment is None) == (test_config.expected_transaction_name is None)
+
+        if segment is not None:
+            assert segment["name"] == test_config.expected_transaction_name
+            assert (
+                segment["attributes"]["sentry.span.source"]
+                == test_config.expected_source
+            )
 
-    # If a transaction was expected, ensure it is correct
-    assert (
-        transaction_event is None
-        or transaction_event["transaction"] == test_config.expected_transaction_name
-    )
-    assert (
-        transaction_event is None
-        or transaction_event["transaction_info"]["source"]
-        == test_config.expected_source
-    )
+            attrs = segment["attributes"]
+            assert attrs["http.request.method"] == "GET"
+            assert attrs["url.full"].endswith(test_config.url)
+            if "?" in test_config.url:
+                assert attrs["http.query"] == test_config.url.split("?", 1)[1]
+            assert attrs["network.protocol.name"] == "http"
+            header_keys = {
+                key[len("http.request.header.") :]
+                for key in attrs
+                if key.startswith("http.request.header.")
+            }
+            assert header_keys >= {"accept", "accept-encoding", "host", "user-agent"}
+            assert attrs["http.response.status_code"] == test_config.expected_status
+            assert segment["status"] == (
+                "error" if test_config.expected_status >= 400 else "ok"
+            )
+    else:
+        # Extract the transaction events by inspecting the event types. We should at most have 1 transaction event.
+        transaction_events = [
+            e for e in events if "type" in e and e["type"] == "transaction"
+        ]
+        assert len(transaction_events) <= 1
+
+        # Get the only transaction event, or set to None if there are no transaction events.
+        (transaction_event, *_) = [*transaction_events, None]
+
+        # We should have no transaction event if and only if we expect no transactions
+        assert (transaction_event is None) == (
+            test_config.expected_transaction_name is None
+        )
+
+        # If a transaction was expected, ensure it is correct
+        assert (
+            transaction_event is None
+            or transaction_event["transaction"] == test_config.expected_transaction_name
+        )
+        assert (
+            transaction_event is None
+            or transaction_event["transaction_info"]["source"]
+            == test_config.expected_source
+        )
 
 
 @pytest.mark.skipif(
     not PERFORMANCE_SUPPORTED, reason="Performance not supported on this Sanic version"
 )
-def test_span_origin(sentry_init, app, capture_events):
-    sentry_init(integrations=[SanicIntegration()], traces_sample_rate=1.0)
-    events = capture_events()
+@pytest.mark.parametrize("span_streaming", [True, False])
+def test_span_origin(sentry_init, app, capture_events, capture_items, span_streaming):
+    sentry_init(
+        integrations=[SanicIntegration()],
+        traces_sample_rate=1.0,
+        _experiments={"trace_lifecycle": "stream" if span_streaming else "static"},
+    )
+
+    if span_streaming:
+        items = capture_items("span")
+    else:
+        events = capture_events()
 
     c = get_client(app)
     with c as client:
         client.get("/message?foo=bar")
 
-    (_, event) = events
+    sentry_sdk.flush()
 
-    assert event["contexts"]["trace"]["origin"] == "auto.http.sanic"
+    if span_streaming:
+        (segment,) = [
+            i.payload
+            for i in items
+            if i.type == "span"
+            and i.payload["attributes"].get("sentry.origin") == "auto.http.sanic"
+        ]
+        assert segment["attributes"]["sentry.origin"] == "auto.http.sanic"
+    else:
+        (_, event) = events
+        assert event["contexts"]["trace"]["origin"] == "auto.http.sanic"