fix(tracing): Skip child span creation in streaming path when no current span (databases)

sentry_sdk/integrations/aiomysql.py

@@ -179,33 +179,35 @@ async def _inner(self: "Connection") -> T:
                 "sentry.origin": AioMySQLIntegration.origin,
             } | breadcrumb_data
 
+            with capture_internal_exceptions():
+                sentry_sdk.add_breadcrumb(
+                    message="connect", category="query", data=breadcrumb_data
+                )
+
+            if sentry_sdk.traces.get_current_span() is None:
+                return await f(self)
+
             with sentry_sdk.traces.start_span(
                 name="connect", attributes=span_attributes
-            ) as span:
-                with capture_internal_exceptions():
-                    sentry_sdk.add_breadcrumb(
-                        message="connect", category="query", data=breadcrumb_data
-                    )
-                res = await f(self)
-        else:
-            connect_data = _get_connect_data(self)
-
-            with sentry_sdk.start_span(
-                op=OP.DB,
-                name="connect",
-                origin=AioMySQLIntegration.origin,
-            ) as span:
-                _set_db_data(span, self)
+            ):
+                return await f(self)
 
-                with capture_internal_exceptions():
-                    sentry_sdk.add_breadcrumb(
-                        message="connect",
-                        category="query",
-                        data=connect_data,
-                    )
-                res = await f(self)
+        connect_data = _get_connect_data(self)
 
-        return res
+        with sentry_sdk.start_span(
+            op=OP.DB,
+            name="connect",
+            origin=AioMySQLIntegration.origin,
+        ) as span:
+            _set_db_data(span, self)
+
+            with capture_internal_exceptions():
+                sentry_sdk.add_breadcrumb(
+                    message="connect",
+                    category="query",
+                    data=connect_data,
+                )
+            return await f(self)
 
     return _inner
 

sentry_sdk/integrations/asyncpg.py

@@ -235,39 +235,40 @@ async def _inner(*args: "Any", **kwargs: "Any") -> "T":
                 except IndexError:
                     pass
 
+            with capture_internal_exceptions():
+                sentry_sdk.add_breadcrumb(
+                    message="connect", category="query", data=span_attributes
+                )
+
+            if sentry_sdk.traces.get_current_span() is None:
+                return await f(*args, **kwargs)
+
             with sentry_sdk.traces.start_span(
                 name="connect", attributes=span_attributes
-            ) as span:
-                with capture_internal_exceptions():
-                    sentry_sdk.add_breadcrumb(
-                        message="connect", category="query", data=span_attributes
-                    )
-                res = await f(*args, **kwargs)
+            ):
+                return await f(*args, **kwargs)
 
-        else:
-            with sentry_sdk.start_span(
-                op=OP.DB,
-                name="connect",
-                origin=AsyncPGIntegration.origin,
-            ) as span:
-                span.set_data(SPANDATA.DB_SYSTEM, "postgresql")
-                if addr:
-                    try:
-                        span.set_data(SPANDATA.SERVER_ADDRESS, addr[0])
-                        span.set_data(SPANDATA.SERVER_PORT, addr[1])
-                    except IndexError:
-                        pass
-                span.set_data(SPANDATA.DB_NAME, database)
-                span.set_data(SPANDATA.DB_USER, user)
-                span.set_data(SPANDATA.DB_DRIVER_NAME, "asyncpg")
-
-                with capture_internal_exceptions():
-                    sentry_sdk.add_breadcrumb(
-                        message="connect", category="query", data=span._data
-                    )
-                res = await f(*args, **kwargs)
+        with sentry_sdk.start_span(
+            op=OP.DB,
+            name="connect",
+            origin=AsyncPGIntegration.origin,
+        ) as span:
+            span.set_data(SPANDATA.DB_SYSTEM, "postgresql")
+            if addr:
+                try:
+                    span.set_data(SPANDATA.SERVER_ADDRESS, addr[0])
+                    span.set_data(SPANDATA.SERVER_PORT, addr[1])
+                except IndexError:
+                    pass
+            span.set_data(SPANDATA.DB_NAME, database)
+            span.set_data(SPANDATA.DB_USER, user)
+            span.set_data(SPANDATA.DB_DRIVER_NAME, "asyncpg")
 
-        return res
+            with capture_internal_exceptions():
+                sentry_sdk.add_breadcrumb(
+                    message="connect", category="query", data=span._data
+                )
+            return await f(*args, **kwargs)
 
     return _inner
 

sentry_sdk/integrations/clickhouse_driver.py

@@ -85,14 +85,16 @@ def _inner(*args: "P.args", **kwargs: "P.kwargs") -> "T":
         params = args[3] if len(args) > 3 else kwargs.get("params")
 
         if has_span_streaming_enabled(client.options):
-            span = sentry_sdk.traces.start_span(
-                name=query,  # type: ignore
-                attributes={
-                    "sentry.op": OP.DB,
-                    "sentry.origin": ClickhouseDriverIntegration.origin,
-                    SPANDATA.DB_QUERY_TEXT: str(query),
-                },
-            )
+            span = None
+            if sentry_sdk.traces.get_current_span() is not None:
+                span = sentry_sdk.traces.start_span(
+                    name=query,  # type: ignore
+                    attributes={
+                        "sentry.op": OP.DB,
+                        "sentry.origin": ClickhouseDriverIntegration.origin,
+                        SPANDATA.DB_QUERY_TEXT: str(query),
+                    },
+                )
         else:
             span = sentry_sdk.start_span(
                 op=OP.DB,
@@ -110,7 +112,8 @@ def _inner(*args: "P.args", **kwargs: "P.kwargs") -> "T":
 
         connection._sentry_span = span  # type: ignore[attr-defined]
 
-        _set_db_data(span, connection)
+        if span is not None:
+            _set_db_data(span, connection)
 
         # run the original code
         ret = f(*args, **kwargs)

sentry_sdk/integrations/pymongo.py

@@ -161,10 +161,6 @@ def started(self, event: "CommandStartedEvent") -> None:
                     **db_data,
                 }
 
-                span = sentry_sdk.traces.start_span(
-                    name=query, attributes=span_first_data
-                )
-
                 with capture_internal_exceptions():
                     sentry_sdk.add_breadcrumb(
                         message=query,
@@ -173,6 +169,13 @@ def started(self, event: "CommandStartedEvent") -> None:
                         data=span_first_data,
                     )
 
+                if sentry_sdk.traces.get_current_span() is None:
+                    return
+
+                span = sentry_sdk.traces.start_span(
+                    name=query, attributes=span_first_data
+                )
+
             else:
                 tags = {
                     "db.name": db_name,

sentry_sdk/integrations/redis/_async_common.py

@@ -46,6 +46,8 @@ async def _sentry_execute(self: "Any", *args: "Any", **kwargs: "Any") -> "Any":
 
         span: "Union[Span, StreamedSpan]"
         if span_streaming:
+            if sentry_sdk.traces.get_current_span() is None:
+                return await old_execute(self, *args, **kwargs)
             span = sentry_sdk.traces.start_span(
                 name="redis.pipeline.execute",
                 attributes={
@@ -103,6 +105,9 @@ async def _sentry_execute_command(
 
         span_streaming = has_span_streaming_enabled(client.options)
 
+        if span_streaming and sentry_sdk.traces.get_current_span() is None:
+            return await old_execute_command(self, name, *args, **kwargs)
+
         cache_properties = _compile_cache_span_properties(
             name,
             args,

sentry_sdk/integrations/redis/_sync_common.py

@@ -43,6 +43,8 @@ def sentry_patched_execute(self: "Any", *args: "Any", **kwargs: "Any") -> "Any":
 
         span: "Union[Span, StreamedSpan]"
         if span_streaming:
+            if sentry_sdk.traces.get_current_span() is None:
+                return old_execute(self, *args, **kwargs)
             span = sentry_sdk.traces.start_span(
                 name="redis.pipeline.execute",
                 attributes={
@@ -102,6 +104,9 @@ def sentry_patched_execute_command(
 
         span_streaming = has_span_streaming_enabled(client.options)
 
+        if span_streaming and sentry_sdk.traces.get_current_span() is None:
+            return old_execute_command(self, name, *args, **kwargs)
+
         cache_properties = _compile_cache_span_properties(
             name,
             args,