fix(asyncio integration): Filter SDK internal tasks from asyncio spans

Add a context manager to track internal task creation context. This can be used to track what tasks are created internally by the SDK

GH-4699

Author: srothh

sentry_sdk/client.py

@@ -38,6 +38,7 @@
 from sentry_sdk.integrations.dedupe import DedupeIntegration
 from sentry_sdk.sessions import SessionFlusher
 from sentry_sdk.envelope import Envelope
+from sentry_sdk.utils import mark_sentry_task_internal
 
 from sentry_sdk.profiler.continuous_profiler import setup_continuous_profiler
 from sentry_sdk.profiler.transaction_profiler import (
@@ -964,9 +965,10 @@ async def _flush_and_close(
             ):
 
                 try:
-                    flush_task = self.transport.loop.create_task(
-                        _flush_and_close(timeout, callback)
-                    )
+                    with mark_sentry_task_internal():
+                        flush_task = self.transport.loop.create_task(
+                            _flush_and_close(timeout, callback)
+                        )
                 except RuntimeError:
                     # Shutdown the components anyway
                     self._close_components()
@@ -999,9 +1001,10 @@ def flush(  # type: ignore[override]
                 self.transport, "loop"
             ):
                 try:
-                    return self.transport.loop.create_task(
-                        self._flush_async(timeout, callback)
-                    )
+                    with mark_sentry_task_internal():
+                        return self.transport.loop.create_task(
+                            self._flush_async(timeout, callback)
+                        )
                 except RuntimeError:
                     logger.warning("Event loop not running, aborting flush.")
                     return None

sentry_sdk/integrations/asyncio.py

@@ -4,7 +4,12 @@
 import sentry_sdk
 from sentry_sdk.consts import OP
 from sentry_sdk.integrations import Integration, DidNotEnable
-from sentry_sdk.utils import event_from_exception, logger, reraise
+from sentry_sdk.utils import (
+    event_from_exception,
+    logger,
+    reraise,
+    _is_sentry_internal_task,
+)
 from sentry_sdk.transport import AsyncHttpTransport
 
 try:
@@ -83,6 +88,18 @@ def _sentry_task_factory(
             **kwargs: Any,
         ) -> asyncio.Future[Any]:
 
+            # Check if this is an internal Sentry task
+            is_internal = _is_sentry_internal_task.get()
+
+            if is_internal:
+                if orig_task_factory:
+                    return orig_task_factory(loop, coro, **kwargs)
+                else:
+                    task = Task(coro, loop=loop, **kwargs)
+                    if task._source_traceback:  # type: ignore
+                        del task._source_traceback[-1]  # type: ignore
+                    return task
+
             async def _task_with_sentry_span_creation() -> Any:
                 result = None
 

sentry_sdk/transport.py

@@ -40,7 +40,12 @@
 import certifi
 
 from sentry_sdk.consts import EndpointType
-from sentry_sdk.utils import Dsn, logger, capture_internal_exceptions
+from sentry_sdk.utils import (
+    Dsn,
+    logger,
+    capture_internal_exceptions,
+    mark_sentry_task_internal,
+)
 from sentry_sdk.worker import BackgroundWorker, Worker, AsyncWorker
 from sentry_sdk.envelope import Envelope, Item, PayloadRef
 
@@ -800,7 +805,8 @@ def kill(self: Self) -> Optional[asyncio.Task[None]]:  # type: ignore
             self.background_tasks.clear()
             try:
                 # Return the pool cleanup task so caller can await it if needed
-                return self.loop.create_task(self._pool.aclose())  # type: ignore
+                with mark_sentry_task_internal():
+                    return self.loop.create_task(self._pool.aclose(), name="sentry_sdk_pool_aclose")  # type: ignore
             except RuntimeError:
                 logger.warning("Event loop not running, aborting kill.")
                 return None

sentry_sdk/utils.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 import base64
+import contextvars
 import json
 import linecache
 import logging
@@ -12,6 +13,7 @@
 import threading
 import time
 from collections import namedtuple
+from contextlib import contextmanager
 from datetime import datetime, timezone
 from decimal import Decimal
 from functools import partial, partialmethod, wraps
@@ -72,6 +74,21 @@
 
 _installed_modules = None
 
+_is_sentry_internal_task = contextvars.ContextVar(
+    "_is_sentry_internal_task", default=False
+)
+
+
+@contextmanager
+def mark_sentry_task_internal():
+    """Context manager to mark a task as Sentry internal."""
+    token = _is_sentry_internal_task.set(True)
+    try:
+        yield
+    finally:
+        _is_sentry_internal_task.reset(token)
+
+
 BASE64_ALPHABET = re.compile(r"^[a-zA-Z0-9/+=]*$")
 
 FALSY_ENV_VALUES = frozenset(("false", "f", "n", "no", "off", "0"))

sentry_sdk/worker.py

@@ -6,7 +6,7 @@
 
 from time import sleep, time
 from sentry_sdk._queue import Queue, FullError
-from sentry_sdk.utils import logger
+from sentry_sdk.utils import logger, mark_sentry_task_internal
 from sentry_sdk.consts import DEFAULT_QUEUE_SIZE
 
 from typing import TYPE_CHECKING
@@ -231,7 +231,10 @@ def start(self) -> None:
                 self._loop = asyncio.get_running_loop()
                 if self._queue is None:
                     self._queue = asyncio.Queue(maxsize=self._queue_size)
-                self._task = self._loop.create_task(self._target())
+                with mark_sentry_task_internal():
+                    self._task = self._loop.create_task(
+                        self._target(), name="sentry_sdk_async_worker"
+                    )
                 self._task_for_pid = os.getpid()
             except RuntimeError:
                 # There is no event loop running
@@ -273,7 +276,11 @@ async def _wait_flush(self, timeout: float, callback: Optional[Any] = None) -> N
 
     def flush(self, timeout: float, callback: Optional[Any] = None) -> Optional[asyncio.Task[None]]:  # type: ignore[override]
         if self.is_alive and timeout > 0.0 and self._loop and self._loop.is_running():
-            return self._loop.create_task(self._wait_flush(timeout, callback))
+            with mark_sentry_task_internal():
+                return self._loop.create_task(
+                    self._wait_flush(timeout, callback),
+                    name="sentry_sdk_async_worker_flush",
+                )
         return None
 
     def submit(self, callback: Callable[[], Any]) -> bool:
@@ -295,7 +302,11 @@ async def _target(self) -> None:
                 self._queue.task_done()
                 break
             # Firing tasks instead of awaiting them allows for concurrent requests
-            task = asyncio.create_task(self._process_callback(callback))
+            with mark_sentry_task_internal():
+                task = asyncio.create_task(
+                    self._process_callback(callback),
+                    name="sentry_sdk_async_worker_process_callback",
+                )
             # Create a strong reference to the task so it can be cancelled on kill
             # and does not get garbage collected while running
             self._active_tasks.add(task)

tests/integrations/asyncio/test_asyncio.py

@@ -8,6 +8,8 @@
 import sentry_sdk
 from sentry_sdk.consts import OP
 from sentry_sdk.integrations.asyncio import AsyncioIntegration, patch_asyncio
+from sentry_sdk.utils import mark_sentry_task_internal
+
 
 try:
     from contextvars import Context, ContextVar
@@ -426,3 +428,52 @@ def test_loop_close_flushes_async_transport(sentry_init):
     except Exception:
         if not loop.is_closed():
             loop.close()
+
+
+@minimum_python_38
+@pytest.mark.asyncio
+async def test_internal_tasks_not_wrapped(sentry_init, capture_events):
+
+    sentry_init(integrations=[AsyncioIntegration()], traces_sample_rate=1.0)
+    events = capture_events()
+
+    # Create a user task that should be wrapped
+    async def user_task():
+        await asyncio.sleep(0.01)
+        return "user_result"
+
+    # Create an internal task that should NOT be wrapped
+    async def internal_task():
+        await asyncio.sleep(0.01)
+        return "internal_result"
+
+    with sentry_sdk.start_transaction(name="test_transaction"):
+        user_task_obj = asyncio.create_task(user_task())
+
+        with mark_sentry_task_internal():
+            internal_task_obj = asyncio.create_task(internal_task())
+
+        user_result = await user_task_obj
+        internal_result = await internal_task_obj
+
+    assert user_result == "user_result"
+    assert internal_result == "internal_result"
+
+    assert len(events) == 1
+    transaction = events[0]
+
+    user_spans = []
+    internal_spans = []
+
+    for span in transaction.get("spans", []):
+        if "user_task" in span.get("description", ""):
+            user_spans.append(span)
+        elif "internal_task" in span.get("description", ""):
+            internal_spans.append(span)
+
+    assert (
+        len(user_spans) > 0
+    ), f"User task should have been traced. All spans: {[s.get('description') for s in transaction.get('spans', [])]}"
+    assert (
+        len(internal_spans) == 0
+    ), f"Internal task should NOT have been traced. All spans: {[s.get('description') for s in transaction.get('spans', [])]}"