chore: Replace framework-specific queue ops (#7959)

Author: alexander-alderman-webb

MIGRATION_GUIDE.md

@@ -44,6 +44,7 @@ Span ops are now a smaller, framework-neutral set.
 - The Cohere integration now emits `gen_ai.chat` instead of `ai.chat_completions.create.cohere`, and `gen_ai.embeddings` instead of `ai.embeddings.create.cohere`.
 - The Django integration now emits `view.render` instead of `view.response.render`, `view` instead of `view.authenticate`, and `function` instead of `event.django`.
 - The Django, Starlette, Starlite, and Litestar integrations now emit `middleware` instead of `middleware.django`, `middleware.starlette`, `middleware.starlette.receive`, `middleware.starlette.send`, `middleware.starlite`, `middleware.starlite.receive`, `middleware.starlite.send`, `middleware.litestar`, `middleware.litestar.receive`, and `middleware.litestar.send`.
+- The ARQ, Celery, Django, Dramatiq, Huey, RQ, and Ray integrations now emit `queue.publish` and `queue.process` instead of `queue.submit.arq`, `queue.task.arq`, `queue.submit.celery`, `queue.task.celery`, `queue.submit.django`, `queue.task.dramatiq`, `queue.submit.huey`, `queue.task.huey`, `queue.task.rq`, `queue.submit.ray`, and `queue.task.ray`.
 
 #### Django
 

sentry_sdk/integrations/arq.py

@@ -80,7 +80,7 @@ async def _sentry_enqueue_job(
         with sentry_sdk.start_span(
             name=function,
             attributes={
-                "sentry.op": OP.QUEUE_SUBMIT_ARQ,
+                "sentry.op": OP.QUEUE_PUBLISH,
                 "sentry.origin": ArqIntegration.origin,
             },
         ):
@@ -105,7 +105,7 @@ async def _sentry_run_job(self: "Worker", job_id: str, score: int) -> None:
             with sentry_sdk.start_span(
                 name="unknown arq task",
                 attributes={
-                    "sentry.op": OP.QUEUE_TASK_ARQ,
+                    "sentry.op": OP.QUEUE_PROCESS,
                     "sentry.origin": ArqIntegration.origin,
                     "sentry.segment.name.source": SegmentNameSource.TASK,
                     SPANDATA.MESSAGING_MESSAGE_ID: job_id,

sentry_sdk/integrations/celery/__init__.py

@@ -267,7 +267,7 @@ def apply_async(*args: "Any", **kwargs: "Any") -> "Any":
             span = sentry_sdk.start_span(
                 name=task_name,
                 attributes={
-                    "sentry.op": OP.QUEUE_SUBMIT_CELERY,
+                    "sentry.op": OP.QUEUE_PUBLISH,
                     "sentry.origin": CeleryIntegration.origin,
                 },
             )
@@ -329,7 +329,7 @@ def _inner(*args: "Any", **kwargs: "Any") -> "Any":
                     attributes={
                         "sentry.origin": CeleryIntegration.origin,
                         "sentry.segment.name.source": SegmentNameSource.TASK.value,
-                        "sentry.op": OP.QUEUE_TASK_CELERY,
+                        "sentry.op": OP.QUEUE_PROCESS,
                     },
                 )
 

sentry_sdk/integrations/django/tasks.py

@@ -38,7 +38,7 @@ def _sentry_enqueue(self: "Any", *args: "Any", **kwargs: "Any") -> "Any":
         with sentry_sdk.start_span(
             name=name,
             attributes={
-                "sentry.op": OP.QUEUE_SUBMIT_DJANGO,
+                "sentry.op": OP.QUEUE_PUBLISH,
                 "sentry.origin": DjangoIntegration.origin,
             },
         ):

sentry_sdk/integrations/dramatiq.py

@@ -141,7 +141,7 @@ def before_process_message(self, broker: "Broker", message: "Message[R]") -> Non
         span = sentry_sdk.start_span(
             name=message.actor_name,
             attributes={
-                "sentry.op": OP.QUEUE_TASK_DRAMATIQ,
+                "sentry.op": OP.QUEUE_PROCESS,
                 "sentry.origin": DramatiqIntegration.origin,
                 "sentry.segment.name.source": SegmentNameSource.TASK.value,
                 SPANDATA.MESSAGING_DESTINATION_NAME: message.queue_name,

sentry_sdk/integrations/huey.py

@@ -93,7 +93,7 @@ def _sentry_enqueue(
         with sentry_sdk.start_span(
             name=span_name,
             attributes={
-                "sentry.op": OP.QUEUE_SUBMIT_HUEY,
+                "sentry.op": OP.QUEUE_PUBLISH,
                 "sentry.origin": HueyIntegration.origin,
                 SPANDATA.MESSAGING_DESTINATION_NAME: self.name,
             },
@@ -191,7 +191,7 @@ def _sentry_execute(
             span_ctx = sentry_sdk.start_span(
                 name=task.name,
                 attributes={
-                    "sentry.op": OP.QUEUE_TASK_HUEY,
+                    "sentry.op": OP.QUEUE_PROCESS,
                     "sentry.origin": HueyIntegration.origin,
                     "sentry.segment.name.source": SegmentNameSource.TASK,
                     SPANDATA.MESSAGING_DESTINATION_NAME: self.name,

sentry_sdk/integrations/ray.py

@@ -95,7 +95,7 @@ def new_func(
                 with sentry_sdk.start_span(
                     name="unknown Ray task" if function_name is None else function_name,
                     attributes={
-                        "sentry.op": OP.QUEUE_TASK_RAY,
+                        "sentry.op": OP.QUEUE_PROCESS,
                         "sentry.origin": RayIntegration.origin,
                         "sentry.segment.name.source": SegmentNameSource.TASK,
                     },
@@ -145,7 +145,7 @@ def _remote_method_with_header_propagation(
                 with sentry_sdk.start_span(
                     name="unknown Ray task" if function_name is None else function_name,
                     attributes={
-                        "sentry.op": OP.QUEUE_SUBMIT_RAY,
+                        "sentry.op": OP.QUEUE_PUBLISH,
                         "sentry.origin": RayIntegration.origin,
                     },
                 ):

sentry_sdk/integrations/rq.py

@@ -82,7 +82,7 @@ def sentry_patched_perform_job(
                 with sentry_sdk.start_span(
                     name="unknown RQ task" if func_name is None else func_name,
                     attributes={
-                        "sentry.op": OP.QUEUE_TASK_RQ,
+                        "sentry.op": OP.QUEUE_PROCESS,
                         "sentry.origin": RqIntegration.origin,
                         "sentry.segment.name.source": SegmentNameSource.TASK,
                         SPANDATA.MESSAGING_MESSAGE_ID: job.id,

tests/integrations/arq/test_arq.py

@@ -11,7 +11,7 @@
 
 import sentry_sdk
 from sentry_sdk import get_client
-from sentry_sdk.consts import SPANDATA
+from sentry_sdk.consts import OP, SPANDATA
 from sentry_sdk.integrations.arq import ArqIntegration
 from tests.integrations.utils import DATA_COLLECTION_QUEUES_CASES
 
@@ -244,10 +244,10 @@ async def retry_job(ctx):
     spans = [item.payload for item in items]
 
     # The retry re-enqueue happens without an active span, so no producer
-    # (queue.submit.arq) span is created for it; only the consumer segment
+    # (queue.publish) 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 2.
-    assert spans[2]["attributes"]["sentry.op"] == "queue.task.arq"
+    assert spans[2]["attributes"]["sentry.op"] == OP.QUEUE_PROCESS
     assert spans[2]["status"] == "ok"
     assert spans[2]["name"] == "retry_job"
 
@@ -256,7 +256,7 @@ async def retry_job(ctx):
     sentry_sdk.flush()
     spans = [item.payload for item in items]
 
-    assert spans[5]["attributes"]["sentry.op"] == "queue.task.arq"
+    assert spans[5]["attributes"]["sentry.op"] == OP.QUEUE_PROCESS
     assert spans[5]["status"] == "ok"
     assert spans[5]["name"] == "retry_job"
 
@@ -329,7 +329,7 @@ async def division(_, a, b=0):
     task_spans = [
         span
         for span in spans
-        if span["attributes"].get("sentry.op") == "queue.task.arq"
+        if span["attributes"].get("sentry.op") == OP.QUEUE_PROCESS
     ]
 
     division_span = next(span for span in task_spans if span["name"] == "division")
@@ -413,7 +413,7 @@ async def dummy_job(_):
     assert spans[2]["trace_id"] == span.trace_id
     assert spans[2]["span_id"] == span.span_id
 
-    assert spans[1]["attributes"]["sentry.op"] == "queue.submit.arq"
+    assert spans[1]["attributes"]["sentry.op"] == OP.QUEUE_PUBLISH
     assert spans[1]["name"] == "dummy_job"
 
 
@@ -497,10 +497,10 @@ async def job(ctx):
     sentry_sdk.flush()
     spans = [item.payload for item in items]
 
-    # No producer (queue.submit.arq) span is created for the re-enqueue
+    # No producer (queue.publish) 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 2.
-    assert spans[2]["attributes"]["sentry.op"] == "queue.task.arq"
+    assert spans[2]["attributes"]["sentry.op"] == OP.QUEUE_PROCESS
     assert spans[2]["attributes"]["sentry.origin"] == "auto.queue.arq"
     assert spans[1]["attributes"]["sentry.origin"] == "auto.db.redis"
     assert spans[0]["attributes"]["sentry.origin"] == "auto.db.redis"

tests/integrations/celery/test_celery.py

@@ -9,6 +9,7 @@
 from celery.exceptions import Ignore, Reject, Retry
 
 import sentry_sdk
+from sentry_sdk.consts import OP
 from sentry_sdk.integrations.celery import (
     CeleryIntegration,
     _wrap_task_run,
@@ -297,14 +298,14 @@ def dummy_task(x, y):
 
     assert process_span["name"] == "dummy_task"
     assert process_span["trace_id"] == span.trace_id
-    assert process_span["attributes"]["sentry.op"] == "queue.process"
+    assert process_span["attributes"]["sentry.op"] == OP.QUEUE_PROCESS
     assert process_span["parent_span_id"] == execution_span["span_id"]
 
     assert submission_span["name"] == "submission"
     assert submission_span["is_segment"] is True
 
     assert submit_span["name"] == "dummy_task"
-    assert submit_span["attributes"]["sentry.op"] == "queue.submit.celery"
+    assert submit_span["attributes"]["sentry.op"] == OP.QUEUE_PUBLISH
     assert submit_span["attributes"]["sentry.origin"] == "auto.queue.celery"
     assert submit_span["parent_span_id"] == submission_span["span_id"] == span.span_id
     assert submit_span["trace_id"] == span.trace_id
@@ -434,7 +435,7 @@ def dummy_task():
     assert not [item for item in items if item.type == "event"]
 
     process_span, execution_span = [item.payload for item in items]
-    assert process_span["attributes"]["sentry.op"] == "queue.process"
+    assert process_span["attributes"]["sentry.op"] == OP.QUEUE_PROCESS
     assert process_span["status"] == "ok"
     assert execution_span["is_segment"] is True
     assert execution_span["status"] == "ok"
@@ -809,7 +810,7 @@ def task(): ...
 
     span_items = [item.payload for item in items]
     publish_span = next(
-        s for s in span_items if s["attributes"].get("sentry.op") == "queue.publish"
+        s for s in span_items if s["attributes"].get("sentry.op") == OP.QUEUE_PUBLISH
     )
 
     assert publish_span["attributes"]["messaging.system"] == system
@@ -941,7 +942,7 @@ def test_send_task_wrapped(
     assert outer["is_segment"] is True
 
     assert submit_span["name"] == "very_creative_task_name"
-    assert submit_span["attributes"]["sentry.op"] == "queue.submit.celery"
+    assert submit_span["attributes"]["sentry.op"] == OP.QUEUE_PUBLISH
     assert submit_span["trace_id"] == outer_span.trace_id
     assert submit_span["trace_id"] == kwargs["headers"]["sentry-trace"].split("-")[0]