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

sentry_sdk/integrations/arq.py

@@ -79,6 +79,9 @@ async def _sentry_enqueue_job(
             return await old_enqueue_job(self, function, *args, **kwargs)
 
         if has_span_streaming_enabled(client.options):
+            if sentry_sdk.traces.get_current_span() is None:
+                return await old_enqueue_job(self, function, *args, **kwargs)
+
             with sentry_sdk.traces.start_span(
                 name=function,
                 attributes={

sentry_sdk/integrations/celery/__init__.py

@@ -419,6 +419,9 @@ def _inner(*args: "Any", **kwargs: "Any") -> "Any":
         span_streaming = has_span_streaming_enabled(client.options)
 
         try:
+            if span_streaming and get_current_span() is None:
+                return f(*args, **kwargs)
+
             span: "Union[Span, StreamedSpan]"
             if span_streaming:
                 span = sentry_sdk.traces.start_span(
@@ -462,7 +465,8 @@ def _inner(*args: "Any", **kwargs: "Any") -> "Any":
 
                 with capture_internal_exceptions():
                     set_on_span(
-                        SPANDATA.MESSAGING_MESSAGE_RETRY_COUNT, task.request.retries
+                        SPANDATA.MESSAGING_MESSAGE_RETRY_COUNT,
+                        task.request.retries,
                     )
 
                 with capture_internal_exceptions():

sentry_sdk/integrations/huey.py

@@ -79,7 +79,18 @@ def _sentry_enqueue(
             sentry_sdk.get_client().options
         )
 
-        span_ctx = None
+        no_headers_types = (PeriodicTask,) + tuple(
+            t for t in [HueyGroup, HueyChord] if t is not None
+        )
+
+        if is_span_streaming_enabled and sentry_sdk.traces.get_current_span() is None:
+            if not isinstance(item, no_headers_types):
+                item.kwargs["sentry_headers"] = {
+                    BAGGAGE_HEADER_NAME: get_baggage(),
+                    SENTRY_TRACE_HEADER_NAME: get_traceparent(),
+                }
+            return old_enqueue(self, item)
+
         if is_span_streaming_enabled:
             span_ctx = sentry_sdk.traces.start_span(
                 name=span_name,
@@ -97,9 +108,6 @@ def _sentry_enqueue(
             )
             span_ctx.set_data(SPANDATA.MESSAGING_DESTINATION_NAME, self.name)
 
-        no_headers_types = (PeriodicTask,) + tuple(
-            t for t in [HueyGroup, HueyChord] if t is not None
-        )
         with span_ctx:
             if not isinstance(item, no_headers_types):
                 # Attach trace propagation data to task kwargs. We do

sentry_sdk/integrations/ray.py

@@ -156,6 +156,23 @@ def _remote_method_with_header_propagation(
                 )
                 if span_streaming:
                     function_name = qualname_from_function(user_f)
+
+                    if sentry_sdk.traces.get_current_span() is None:
+                        tracing = {
+                            k: v
+                            for k, v in sentry_sdk.get_current_scope().iter_trace_propagation_headers()
+                        }
+                        try:
+                            result = old_remote_method(
+                                *args, **kwargs, _sentry_tracing=tracing
+                            )
+                        except Exception:
+                            exc_info = sys.exc_info()
+                            _capture_exception(exc_info)
+                            reraise(*exc_info)
+
+                        return result
+
                     with sentry_sdk.traces.start_span(
                         name="unknown Ray task"
                         if function_name is None