Merge branch 'py-2298-migrate-aiohttp-to-span-first' of github.com:getsentry/sentry-python into py-2298-migrate-aiohttp-to-span-first

Author: ericapisani

.github/workflows/update-tox.yml

@@ -9,6 +9,7 @@ on:
 jobs:
   update-tox:
     name: Update test matrix
+    if: github.ref == 'refs/heads/master'
     runs-on: ubuntu-latest
     timeout-minutes: 10
 
@@ -55,11 +56,15 @@ jobs:
 
       - name: Create pull request
         uses: actions/github-script@3a2844b7e9c422d3c10d287c895573f7108da1b3 # v9.0.0
+        env:
+          BRANCH_NAME: ${{ steps.create-branch.outputs.branch_name }}
+          COMMIT_TITLE: ${{ steps.create-branch.outputs.commit_title }}
+          DATE: ${{ steps.create-branch.outputs.date }}
         with:
           script: |
-            const branchName = '${{ steps.create-branch.outputs.branch_name }}';
-            const commitTitle = '${{ steps.create-branch.outputs.commit_title }}';
-            const date = '${{ steps.create-branch.outputs.date }}';
+            const branchName = process.env.BRANCH_NAME;
+            const commitTitle = process.env.COMMIT_TITLE;
+            const date = process.env.DATE;
             const prBody = `Update our test matrix with new releases of integrated frameworks and libraries.
 
             ## How it works
@@ -100,7 +105,7 @@ jobs:
                 repo: context.repo.repo,
                 title: commitTitle + ' (' + date + ')',
                 head: branchName,
-                base: '${{ github.ref_name }}',
+                base: 'master',
                 body: prBody,
             });
 

scripts/populate_tox/package_dependencies.jsonl

@@ -3,6 +3,7 @@
 import threading
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING, TypeVar, Generic
+import weakref
 
 from sentry_sdk.utils import format_timestamp
 from sentry_sdk.envelope import Envelope, Item, PayloadRef
@@ -38,6 +39,26 @@ def __init__(
         self._flusher: "Optional[threading.Thread]" = None
         self._flusher_pid: "Optional[int]" = None
 
+        # See https://github.com/getsentry/sentry-python/blob/051cc01640a29bfd64b1f1e2e3414c02f027dd1b/sentry_sdk/monitor.py#L41-L50
+        if hasattr(os, "register_at_fork"):
+            weak_reset = weakref.WeakMethod(self._reset_thread_state)
+
+            def _reset_in_child() -> None:
+                method = weak_reset()
+                if method is not None:
+                    method()
+
+            os.register_at_fork(after_in_child=_reset_in_child)
+
+    def _reset_thread_state(self) -> None:
+        self._buffer = []
+        self._running = True
+        self._lock = threading.Lock()
+        self._active = threading.local()
+        self._flush_event = threading.Event()
+        self._flusher = None
+        self._flusher_pid = None
+
     def _ensure_thread(self) -> bool:
         """For forking processes we might need to restart this thread.
         This ensures that our process actually has that thread running.

scripts/populate_tox/releases.jsonl

@@ -1,7 +1,9 @@
-import threading
 from collections import defaultdict
 from datetime import datetime, timezone
+import os
+import threading
 from typing import TYPE_CHECKING
+import weakref
 
 from sentry_sdk._batcher import Batcher
 from sentry_sdk.envelope import Envelope, Item, PayloadRef
@@ -50,6 +52,30 @@ def __init__(
         self._flusher: "Optional[threading.Thread]" = None
         self._flusher_pid: "Optional[int]" = None
 
+        # See https://github.com/getsentry/sentry-python/blob/051cc01640a29bfd64b1f1e2e3414c02f027dd1b/sentry_sdk/monitor.py#L41-L50
+        if hasattr(os, "register_at_fork"):
+            weak_reset = weakref.WeakMethod(self._reset_thread_state)
+
+            def _reset_in_child() -> None:
+                method = weak_reset()
+                if method is not None:
+                    method()
+
+            os.register_at_fork(after_in_child=_reset_in_child)
+
+    def _reset_thread_state(self) -> None:
+        self._span_buffer = defaultdict(list)
+        self._running_size = defaultdict(lambda: 0)
+        self._running = True
+
+        self._lock = threading.Lock()
+        self._active = threading.local()
+
+        self._flush_event = threading.Event()
+
+        self._flusher = None
+        self._flusher_pid = None
+
     def add(self, span: "StreamedSpan") -> None:
         # Bail out if the current thread is already executing batcher code.
         # This prevents deadlocks when code running inside the batcher (e.g.

sentry_sdk/_batcher.py

@@ -6,7 +6,7 @@
 from sentry_sdk._init_implementation import init
 from sentry_sdk.consts import INSTRUMENTER
 from sentry_sdk.scope import Scope, _ScopeManager, new_scope, isolation_scope
-from sentry_sdk.traces import StreamedSpan
+from sentry_sdk.traces import StreamedSpan, _get_current_streamed_span
 from sentry_sdk.tracing import NoOpSpan, Transaction, trace
 from sentry_sdk.crons import monitor
 
@@ -422,7 +422,7 @@ def set_measurement(name: str, value: float, unit: "MeasurementUnit" = "") -> No
 
 def get_current_span(
     scope: "Optional[Scope]" = None,
-) -> "Optional[Union[Span, StreamedSpan]]":
+) -> "Optional[Span]":
     """
     Returns the currently active span if there is one running, otherwise `None`
     """
@@ -533,12 +533,7 @@ def update_current_span(
             attributes={"user_id": 123, "batch_size": 50}
         )
     """
-    current_span = get_current_span()
-
-    if current_span is None:
-        return
-
-    if isinstance(current_span, StreamedSpan):
+    if isinstance(_get_current_streamed_span(), StreamedSpan):
         warnings.warn(
             "The `update_current_span` API isn't available in streaming mode. "
             "Retrieve the current span with get_current_span() and use its API "
@@ -548,6 +543,11 @@ def update_current_span(
         )
         return
 
+    current_span = get_current_span()
+
+    if current_span is None:
+        return
+
     if op is not None:
         current_span.op = op
 

sentry_sdk/_span_batcher.py

@@ -881,6 +881,12 @@ class SPANDATA:
     Example: "tcp", "udp", "unix"
     """
 
+    PROCESS_PID = "process.pid"
+    """
+    The process ID of the running process.
+    Example: 12345
+    """
+
     PROFILER_ID = "profiler_id"
     """
     Label identifying the profiler id that the span occurred in. This should be a string.
@@ -924,6 +930,24 @@ class SPANDATA:
     Example: "MainThread"
     """
 
+    URL_FULL = "url.full"
+    """
+    The URL of the resource that was fetched.
+    Example: "https://example.com/test?foo=bar#buzz"
+    """
+
+    URL_FRAGMENT = "url.fragment"
+    """
+    The fragments present in the URI. Note that this does not contain the leading # character, while the `http.fragment` attribute does.
+    Example: "details"
+    """
+
+    URL_QUERY = "url.query"
+    """
+    The query string present in the URL. Note that this does not contain the leading ? character, while the `http.query` attribute does.
+    Example: "foo=bar&bar=baz"
+    """
+
     MCP_TOOL_NAME = "mcp.tool.name"
     """
     The name of the MCP tool being called.

sentry_sdk/api.py

@@ -31,6 +31,7 @@
     "HTTP_SET_COOKIE",
     "HTTP_COOKIE",
     "HTTP_AUTHORIZATION",
+    "HTTP_PROXY_AUTHORIZATION",
     "HTTP_X_API_KEY",
     "HTTP_X_FORWARDED_FOR",
     "HTTP_X_REAL_IP",

sentry_sdk/consts.py

@@ -15,7 +15,7 @@
 from sentry_sdk.integrations.celery.utils import _now_seconds_since_epoch
 from sentry_sdk.integrations.logging import ignore_logger
 from sentry_sdk.scope import should_send_default_pii
-from sentry_sdk.traces import StreamedSpan
+from sentry_sdk.traces import StreamedSpan, _get_current_streamed_span
 from sentry_sdk.tracing import BAGGAGE_HEADER_NAME, Span, TransactionSource
 from sentry_sdk.tracing_utils import Baggage, has_span_streaming_enabled
 from sentry_sdk.utils import (
@@ -98,13 +98,16 @@ def setup_once() -> None:
 
 
 def _set_status(status: str) -> None:
+    client = sentry_sdk.get_client()
+    span_streaming = has_span_streaming_enabled(client.options)
+
     with capture_internal_exceptions():
         scope = sentry_sdk.get_current_scope()
-        if scope.span is not None:
-            if isinstance(scope.span, Span):
-                scope.span.set_status(status)
-            else:
-                scope.span.status = "ok" if status == "ok" else "error"
+
+        if span_streaming and scope.streamed_span is not None:
+            scope.streamed_span.status = "ok" if status == "ok" else "error"
+        elif not span_streaming and scope.span is not None:
+            scope.span.set_status(status)
 
 
 def _capture_exception(task: "Any", exc_info: "ExcInfo") -> None:
@@ -289,7 +292,7 @@ def apply_async(*args: "Any", **kwargs: "Any") -> "Any":
 
         span_mgr: "Union[StreamedSpan, Span, NoOpMgr]" = NoOpMgr()
         if span_streaming:
-            if not task_started_from_beat and sentry_sdk.get_current_span() is not None:
+            if not task_started_from_beat and _get_current_streamed_span() is not None:
                 span_mgr = sentry_sdk.traces.start_span(
                     name=task_name,
                     attributes={
@@ -570,7 +573,7 @@ def sentry_publish(self: "Producer", *args: "Any", **kwargs: "Any") -> "Any":
 
         span: "Union[StreamedSpan, Span, None]" = None
         if span_streaming:
-            if sentry_sdk.get_current_span() is not None:
+            if _get_current_streamed_span() is not None:
                 span = sentry_sdk.traces.start_span(
                     name=task_name,
                     attributes={

sentry_sdk/integrations/_wsgi_common.py

@@ -90,13 +90,17 @@ def _sentry_get_request_handler(*args: "Any", **kwargs: "Any") -> "Any":
             @wraps(old_call)
             def _sentry_call(*args: "Any", **kwargs: "Any") -> "Any":
                 current_scope = sentry_sdk.get_current_scope()
-                current_span = current_scope.span
 
-                if isinstance(current_span, StreamedSpan) and not isinstance(
-                    current_span, NoOpStreamedSpan
-                ):
-                    segment = current_span._segment
-                    segment._update_active_thread()
+                client = sentry_sdk.get_client()
+                if has_span_streaming_enabled(client.options):
+                    current_span = current_scope.streamed_span
+
+                    if isinstance(current_span, StreamedSpan) and not isinstance(
+                        current_span, NoOpStreamedSpan
+                    ):
+                        segment = current_span._segment
+                        segment._update_active_thread()
+
                 elif current_scope.transaction is not None:
                     current_scope.transaction.update_active_thread()