feat(sanic): Support span streaming

.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(
@@ -198,12 +221,24 @@ async def _context_exit(
  # 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"
+
+ span.end()
+ else:
+ span.set_http_status(response_status)
+ span.sampled &= (
+ isinstance(integration, SanicIntegration)
+ and response_status not in integration._unsampled_statuses
+ )
+
+ span.__exit__(None, None, None)
 
  request.ctx._sentry_scope.__exit__(None, None, None)
 
@@ -315,6 +350,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:

tests/integrations/sanic/test_sanic.py

@@ -346,6 +346,7 @@ def __init__(
  expected_status: int,
  expected_transaction_name: "Optional[str]",
  expected_source: "Optional[str]" = None,
+ streaming_compatible: bool = True,
  ) -> None:
  """
  expected_transaction_name of None indicates we expect to not receive a transaction
@@ -355,11 +356,13 @@ def __init__(
  self.expected_status = expected_status
  self.expected_transaction_name = expected_transaction_name
  self.expected_source = expected_source
+ self.streaming_compatible = streaming_compatible
 
 
 @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 +374,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=(),
@@ -385,6 +396,7 @@ def __init__(
  url="/404",
  expected_status=404,
  expected_transaction_name=None,
+ streaming_compatible=False,
  ),
  TransactionTestConfig(
  # With no ignored HTTP statuses, we should get transactions for 404 errors
@@ -400,6 +412,7 @@ def __init__(
  url="/message",
  expected_status=200,
  expected_transaction_name=None,
+ streaming_compatible=False,
  ),
  ],
 )
@@ -408,57 +421,123 @@ def test_transactions(
  sentry_init: "Any",
  app: "Any",
  capture_events: "Any",
+ capture_items: "Any",
+ span_streaming: bool,
 ) -> None:
+ if span_streaming and not test_config.streaming_compatible:
+ pytest.skip("unsampled_statuses is not supported in span streaming mode")
+
  # 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.payload["attributes"].get("sentry.origin") == "auto.http.sanic"
+ and i.payload["is_segment"]
+ ]
+ 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.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"