diff --git a/sentry_sdk/integrations/arq.py b/sentry_sdk/integrations/arq.py index 3488457e20..011c5e51bf 100644 --- a/sentry_sdk/integrations/arq.py +++ b/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={ diff --git a/sentry_sdk/integrations/celery/__init__.py b/sentry_sdk/integrations/celery/__init__.py index 532b13539b..a4abac4618 100644 --- a/sentry_sdk/integrations/celery/__init__.py +++ b/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(): diff --git a/sentry_sdk/integrations/huey.py b/sentry_sdk/integrations/huey.py index a1c9d6a357..70814d4b7f 100644 --- a/sentry_sdk/integrations/huey.py +++ b/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 diff --git a/sentry_sdk/integrations/ray.py b/sentry_sdk/integrations/ray.py index f723a96f3c..f48bd59791 100644 --- a/sentry_sdk/integrations/ray.py +++ b/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 diff --git a/tests/integrations/arq/test_arq.py b/tests/integrations/arq/test_arq.py index fb40018d2a..013ea83167 100644 --- a/tests/integrations/arq/test_arq.py +++ b/tests/integrations/arq/test_arq.py @@ -245,18 +245,22 @@ async def retry_job(ctx): sentry_sdk.flush() spans = [item.payload for item in items] - assert spans[4]["attributes"]["sentry.op"] == "queue.task.arq" - assert spans[4]["status"] == "ok" - assert spans[4]["name"] == "retry_job" + # The retry re-enqueue happens without an active span, so no producer + # (queue.submit.arq) span is created for it; only the consumer segment + # is emitted. The consumer segment is preceded by the redis spans for + # the re-enqueue, so it lands at index 3. + assert spans[3]["attributes"]["sentry.op"] == "queue.task.arq" + assert spans[3]["status"] == "ok" + assert spans[3]["name"] == "retry_job" await worker.run_job(job.job_id, timestamp_ms()) sentry_sdk.flush() spans = [item.payload for item in items] - assert spans[7]["attributes"]["sentry.op"] == "queue.task.arq" - assert spans[7]["status"] == "ok" - assert spans[7]["name"] == "retry_job" + assert spans[6]["attributes"]["sentry.op"] == "queue.task.arq" + assert spans[6]["status"] == "ok" + assert spans[6]["name"] == "retry_job" else: events = capture_events() @@ -594,10 +598,13 @@ async def job(ctx): sentry_sdk.flush() spans = [item.payload for item in items] - assert spans[4]["attributes"]["sentry.op"] == "queue.task.arq" - assert spans[4]["attributes"]["sentry.origin"] == "auto.queue.arq" - assert spans[3]["attributes"]["sentry.origin"] == "auto.db.redis" + # No producer (queue.submit.arq) span is created for the re-enqueue + # triggered by the retry, since it happens without an active span, so + # the consumer segment lands at index 3. + assert spans[3]["attributes"]["sentry.op"] == "queue.task.arq" + assert spans[3]["attributes"]["sentry.origin"] == "auto.queue.arq" assert spans[2]["attributes"]["sentry.origin"] == "auto.db.redis" + assert spans[1]["attributes"]["sentry.origin"] == "auto.db.redis" else: job = await pool.enqueue_job("retry_job") diff --git a/tests/integrations/huey/test_huey.py b/tests/integrations/huey/test_huey.py index bfd04791dd..93e24a96d6 100644 --- a/tests/integrations/huey/test_huey.py +++ b/tests/integrations/huey/test_huey.py @@ -93,13 +93,12 @@ def division(a, b): sentry_sdk.get_client().flush() payloads = [i.payload for i in items] - assert len(payloads) == 2 - enqueue_span, execute_span = payloads + # The task is enqueued without a wrapping span, so in streaming mode no + # producer (queue.submit.huey) span is created (see the early return in + # patch_enqueue). Only the consumer segment is emitted. + assert len(payloads) == 1 + (execute_span,) = payloads - assert enqueue_span["attributes"]["sentry.op"] == OP.QUEUE_SUBMIT_HUEY - assert ( - enqueue_span["attributes"][SPANDATA.MESSAGING_DESTINATION_NAME] == huey.name - ) assert execute_span["is_segment"] assert execute_span["attributes"]["sentry.op"] == OP.QUEUE_TASK_HUEY assert ( @@ -157,12 +156,12 @@ def retry_task(context): sentry_sdk.get_client().flush() payloads = [i.payload for i in items] - assert len(payloads) == 3 - - enqueue_span, re_enqueue_span, execute_span = payloads + # The initial enqueue happens without a wrapping span, so no producer + # span is created for it. The re-enqueue triggered by RetryTask happens + # inside the running consumer segment, so it does get a child span. + assert len(payloads) == 2 - assert enqueue_span["attributes"]["sentry.op"] == OP.QUEUE_SUBMIT_HUEY - assert enqueue_span["is_segment"] + re_enqueue_span, execute_span = payloads assert re_enqueue_span["attributes"]["sentry.op"] == OP.QUEUE_SUBMIT_HUEY assert not re_enqueue_span["is_segment"] @@ -181,8 +180,8 @@ def retry_task(context): all_payloads = [i.payload for i in items] - assert len(all_payloads) == 4 - retry_span = all_payloads[3] + assert len(all_payloads) == 3 + retry_span = all_payloads[2] assert retry_span["is_segment"] assert retry_span["name"] == "retry_task" @@ -224,10 +223,10 @@ def cancel_task(): sentry_sdk.get_client().flush() payloads = [i.payload for i in items] - assert len(payloads) == 2 - enqueue_span, execute_span = payloads + # Enqueued without a wrapping span -> no producer span in streaming mode. + assert len(payloads) == 1 + (execute_span,) = payloads - assert enqueue_span["attributes"]["sentry.op"] == OP.QUEUE_SUBMIT_HUEY assert execute_span["attributes"]["sentry.op"] == OP.QUEUE_TASK_HUEY assert execute_span["is_segment"] assert execute_span["name"] == "cancel_task" @@ -270,10 +269,10 @@ def maybe_locked_task(): sentry_sdk.get_client().flush() payloads = [i.payload for i in items] - assert len(payloads) == 2 - enqueue_span, execute_span = payloads + # Enqueued without a wrapping span -> no producer span in streaming mode. + assert len(payloads) == 1 + (execute_span,) = payloads - assert enqueue_span["attributes"]["sentry.op"] == OP.QUEUE_SUBMIT_HUEY assert execute_span["attributes"]["sentry.op"] == OP.QUEUE_TASK_HUEY assert execute_span["is_segment"] @@ -391,29 +390,36 @@ def task2(): if has_span_streaming: items = capture_items("span") - huey.enqueue(group([task1.s(), task2.s()])) + with sentry_sdk.traces.start_span(name="submission"): + huey.enqueue(group([task1.s(), task2.s()])) for _ in range(2): task = huey.dequeue() huey.execute(task) sentry_sdk.get_client().flush() - assert len(items) == 5 + assert len(items) == 6 ( task1_enqueue_span, task2_enqueue_span, group_span, + submission_span, task1_execute_span, task2_execute_span, ) = [i.payload for i in items] - assert group_span["is_segment"] + # The enqueue happens inside a wrapping span, so the group producer + # tree is created and parented under that segment. + assert submission_span["is_segment"] + assert submission_span["name"] == "submission" + assert not group_span["is_segment"] assert not task1_enqueue_span["is_segment"] assert not task2_enqueue_span["is_segment"] assert task1_execute_span["is_segment"] assert task2_execute_span["is_segment"] + assert group_span["parent_span_id"] == submission_span["span_id"] assert group_span["name"] == "Huey Task Group" assert group_span["status"] == "ok" assert group_span["attributes"]["sentry.op"] == "queue.submit.huey" @@ -422,18 +428,14 @@ def task2(): assert task1_enqueue_span["name"] == "task1" assert task1_enqueue_span["status"] == "ok" assert task1_enqueue_span["parent_span_id"] == group_span["span_id"] - assert ( - task1_enqueue_span["attributes"]["sentry.segment.name"] == "Huey Task Group" - ) + assert task1_enqueue_span["attributes"]["sentry.segment.name"] == "submission" assert task1_enqueue_span["attributes"]["sentry.op"] == "queue.submit.huey" assert task1_enqueue_span["attributes"]["sentry.origin"] == "auto.queue.huey" assert task2_enqueue_span["name"] == "task2" assert task2_enqueue_span["status"] == "ok" assert task2_enqueue_span["parent_span_id"] == group_span["span_id"] - assert ( - task2_enqueue_span["attributes"]["sentry.segment.name"] == "Huey Task Group" - ) + assert task2_enqueue_span["attributes"]["sentry.segment.name"] == "submission" assert task2_enqueue_span["attributes"]["sentry.op"] == "queue.submit.huey" assert task2_enqueue_span["attributes"]["sentry.origin"] == "auto.queue.huey" @@ -518,29 +520,36 @@ def task2(results): if has_span_streaming: items = capture_items("span") - huey.enqueue(chord([task1.s()], task2.s())) + with sentry_sdk.traces.start_span(name="submission"): + huey.enqueue(chord([task1.s()], task2.s())) for _ in range(2): task = huey.dequeue() huey.execute(task) sentry_sdk.get_client().flush() - assert len(items) == 5 + assert len(items) == 6 ( task1_enqueue_span, chord_span, + submission_span, task2_enqueue_span, task1_execute_span, task2_execute_span, ) = [i.payload for i in items] - assert chord_span["is_segment"] + # The enqueue happens inside a wrapping span, so the chord producer + # tree is created and parented under that segment. + assert submission_span["is_segment"] + assert submission_span["name"] == "submission" + assert not chord_span["is_segment"] assert not task1_enqueue_span["is_segment"] assert not task2_enqueue_span["is_segment"] assert task1_execute_span["is_segment"] assert task2_execute_span["is_segment"] + assert chord_span["parent_span_id"] == submission_span["span_id"] assert chord_span["name"] == "Huey Chord" assert chord_span["status"] == "ok" assert chord_span["attributes"]["sentry.op"] == "queue.submit.huey" @@ -549,7 +558,7 @@ def task2(results): assert task1_enqueue_span["name"] == "task1" assert task1_enqueue_span["status"] == "ok" assert task1_enqueue_span["parent_span_id"] == chord_span["span_id"] - assert task1_enqueue_span["attributes"]["sentry.segment.name"] == "Huey Chord" + assert task1_enqueue_span["attributes"]["sentry.segment.name"] == "submission" assert task1_enqueue_span["attributes"]["sentry.op"] == "queue.submit.huey" assert task1_enqueue_span["attributes"]["sentry.origin"] == "auto.queue.huey"