feat: Make ASGI support span first (#5680)

This PR makes the ASGI integration work both in legacy and span
streaming mode. Some features and attributes will be missing in span
streaming mode for now (see the Out of scope section below).

Best reviewed with whitespace ignored:
https://github.com/getsentry/sentry-python/pull/5680/changes?w=1

---

A bit of a background on migrating integrations to span first. In order
to support both legacy spans and span streaming, most integrations will
follow the same patterns:

### API

We need to use the `start_span` API from `sentry_sdk.traces` if we're in
span streaming mode (`traces_lifecycle="stream"`).

There are no transactions anymore. Top-level spans will also be started
via the `start_span` API in span streaming mode.

### Setting data on spans

If an integration sets data on a span (via `span.set_data`,
`span.set_tag` etc.), it should use `span.set_attribute` when span
streaming is enabled.

The attributes that we set need to be in Sentry conventions. This is
deliberately not the case for most quick ports of integrations like this
one and will follow in [a future
step](#5152).

### Trace propagation

If an integration sits at a service boundary and is capable of
propagating incoming trace information (like WSGI/ASGI or Celery), in
span first mode we need to switch from the old style `with
continue_trace(...) as transaction:` to the [new style
`continue_trace()` and
`new_trace()`](https://sentry-docs-git-ivana-span-first-migration-guide.sentry.dev/platforms/python/migration/span-first/#trace-propagation)
(not context managers).

### `start_span` arguments

You can pass things like `op`, `origin`, `source` to the old
`start_span` API. With the new API, this is no longer possible, and the
individual properties need to be set as attributes directly.

### Out of scope

For now, the following is out of scope and will follow in the future:
- Making sure all attributes are correct and that they're set in Sentry
conventions: #5152
- Migrating event processors (as streaming spans are not events, event
processors are not run on them, meaning some data will not be set yet):
#5152

Author: sentrivana

sentry_sdk/integrations/asgi.py

@@ -22,10 +22,13 @@
     nullcontext,
 )
 from sentry_sdk.sessions import track_session
+from sentry_sdk.traces import StreamedSpan
 from sentry_sdk.tracing import (
     SOURCE_FOR_STYLE,
+    Transaction,
     TransactionSource,
 )
+from sentry_sdk.tracing_utils import has_span_streaming_enabled
 from sentry_sdk.utils import (
     ContextVar,
     event_from_exception,
@@ -35,17 +38,19 @@
     transaction_from_function,
     _get_installed_modules,
 )
-from sentry_sdk.tracing import Transaction
 
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
     from typing import Any
+    from typing import ContextManager
     from typing import Dict
     from typing import Optional
     from typing import Tuple
+    from typing import Union
 
-    from sentry_sdk._types import Event, Hint
+    from sentry_sdk._types import Attributes, Event, Hint
+    from sentry_sdk.tracing import Span
 
 
 _asgi_middleware_applied = ContextVar("sentry_asgi_middleware_applied")
@@ -185,6 +190,9 @@ async def _run_app(
                 self._capture_lifespan_exception(exc)
                 raise exc from None
 
+        client = sentry_sdk.get_client()
+        span_streaming = has_span_streaming_enabled(client.options)
+
         _asgi_middleware_applied.set(True)
         try:
             with sentry_sdk.isolation_scope() as sentry_scope:
@@ -204,48 +212,105 @@ async def _run_app(
                     )
 
                     method = scope.get("method", "").upper()
-                    transaction = None
-                    if ty in ("http", "websocket"):
-                        if ty == "websocket" or method in self.http_methods_to_capture:
-                            transaction = continue_trace(
-                                _get_headers(scope),
-                                op="{}.server".format(ty),
+
+                    span_ctx: "ContextManager[Union[Span, StreamedSpan, None]]"
+                    if span_streaming:
+                        segment: "Optional[StreamedSpan]" = None
+                        attributes: "Attributes" = {
+                            "sentry.span.source": getattr(
+                                transaction_source, "value", transaction_source
+                            ),
+                            "sentry.origin": self.span_origin,
+                            "asgi.type": ty,
+                        }
+
+                        if ty in ("http", "websocket"):
+                            if (
+                                ty == "websocket"
+                                or method in self.http_methods_to_capture
+                            ):
+                                sentry_sdk.traces.continue_trace(_get_headers(scope))
+
+                                sentry_scope.set_custom_sampling_context(
+                                    {"asgi_scope": scope}
+                                )
+
+                                attributes["sentry.op"] = f"{ty}.server"
+                                segment = sentry_sdk.traces.start_span(
+                                    name=transaction_name, attributes=attributes
+                                )
+                        else:
+                            sentry_sdk.traces.new_trace()
+
+                            sentry_scope.set_custom_sampling_context(
+                                {"asgi_scope": scope}
+                            )
+
+                            attributes["sentry.op"] = OP.HTTP_SERVER
+                            segment = sentry_sdk.traces.start_span(
+                                name=transaction_name, attributes=attributes
+                            )
+
+                        span_ctx = segment or nullcontext()
+
+                    else:
+                        transaction = None
+                        if ty in ("http", "websocket"):
+                            if (
+                                ty == "websocket"
+                                or method in self.http_methods_to_capture
+                            ):
+                                transaction = continue_trace(
+                                    _get_headers(scope),
+                                    op="{}.server".format(ty),
+                                    name=transaction_name,
+                                    source=transaction_source,
+                                    origin=self.span_origin,
+                                )
+                        else:
+                            transaction = Transaction(
+                                op=OP.HTTP_SERVER,
                                 name=transaction_name,
                                 source=transaction_source,
                                 origin=self.span_origin,
                             )
-                    else:
-                        transaction = Transaction(
-                            op=OP.HTTP_SERVER,
-                            name=transaction_name,
-                            source=transaction_source,
-                            origin=self.span_origin,
-                        )
 
-                    if transaction:
-                        transaction.set_tag("asgi.type", ty)
+                        if transaction:
+                            transaction.set_tag("asgi.type", ty)
 
-                    transaction_context = (
-                        sentry_sdk.start_transaction(
-                            transaction,
-                            custom_sampling_context={"asgi_scope": scope},
+                        span_ctx = (
+                            sentry_sdk.start_transaction(
+                                transaction,
+                                custom_sampling_context={"asgi_scope": scope},
+                            )
+                            if transaction is not None
+                            else nullcontext()
                         )
-                        if transaction is not None
-                        else nullcontext()
-                    )
-                    with transaction_context:
+
+                    with span_ctx as span:
                         try:
 
                             async def _sentry_wrapped_send(
                                 event: "Dict[str, Any]",
                             ) -> "Any":
-                                if transaction is not None:
+                                if span is not None:
                                     is_http_response = (
                                         event.get("type") == "http.response.start"
                                         and "status" in event
                                     )
                                     if is_http_response:
-                                        transaction.set_http_status(event["status"])
+                                        if isinstance(span, StreamedSpan):
+                                            span.status = (
+                                                "error"
+                                                if event["status"] >= 400
+                                                else "ok"
+                                            )
+                                            span.set_attribute(
+                                                "http.response.status_code",
+                                                event["status"],
+                                            )
+                                        else:
+                                            span.set_http_status(event["status"])
 
                                 return await send(event)