feat(sanic): Support span streaming

Author: sl0thentr0py

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,31 @@ 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. The orphaned scope is discarded
+                # when the isolation scope exits below.
+                if should_sample:
+                    span.__exit__(None, None, None)
+            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 +353,40 @@ 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] = "%s://%s%s" % (
+        urlparts.scheme,
+        urlparts.netloc,
+        urlparts.path,
+    )
+
+    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:

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",
     [
@@ -408,57 +409,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)
+            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"