feat(rq): Set messaging.destination.name on consumer spans (#6774)

Set `Queue.name` as the `messaging.destination.name` attribute to support span description inference.

Author: alexander-alderman-webb

sentry_sdk/integrations/rq.py

@@ -65,11 +65,11 @@ def setup_once() -> None:
 
         @functools.wraps(old_perform_job)
         def sentry_patched_perform_job(
-            self: "Any", job: "Job", *args: "Queue", **kwargs: "Any"
+            self: "Any", job: "Job", queue: "Queue", *args: "Any", **kwargs: "Any"
         ) -> bool:
             client = sentry_sdk.get_client()
             if client.get_integration(RqIntegration) is None:
-                return old_perform_job(self, job, *args, **kwargs)
+                return old_perform_job(self, job, queue, *args, **kwargs)
 
             with sentry_sdk.new_scope() as scope:
                 scope.clear_breadcrumbs()
@@ -93,13 +93,14 @@ def sentry_patched_perform_job(
                             "sentry.origin": RqIntegration.origin,
                             "sentry.span.source": SegmentSource.TASK,
                             SPANDATA.MESSAGING_MESSAGE_ID: job.id,
+                            SPANDATA.MESSAGING_DESTINATION_NAME: queue.name,
                         },
                         parent_span=None,
                     ) as span:
                         if func_name is not None:
                             span.set_attribute(SPANDATA.CODE_FUNCTION_NAME, func_name)
 
-                        rv = old_perform_job(self, job, *args, **kwargs)
+                        rv = old_perform_job(self, job, queue, *args, **kwargs)
                 else:
                     transaction = continue_trace(
                         job.meta.get("_sentry_trace_headers") or {},
@@ -115,8 +116,10 @@ def sentry_patched_perform_job(
                     with sentry_sdk.start_transaction(
                         transaction,
                         custom_sampling_context={"rq_job": job},
-                    ):
-                        rv = old_perform_job(self, job, *args, **kwargs)
+                    ) as span:
+                        span.set_data(SPANDATA.MESSAGING_DESTINATION_NAME, queue.name)
+
+                        rv = old_perform_job(self, job, queue, *args, **kwargs)
 
             if self.is_horse:
                 # We're inside of a forked process and RQ is

tests/integrations/rq/test_rq.py

@@ -416,6 +416,7 @@ def test_transaction_no_error(
 
         assert span["attributes"]["sentry.op"] == "queue.task.rq"
         assert span["name"] == "tests.integrations.rq.test_rq.do_trick"
+        assert span["attributes"][SPANDATA.MESSAGING_DESTINATION_NAME] == queue.name
     else:
         events = capture_events()
 
@@ -427,6 +428,10 @@ def test_transaction_no_error(
         assert envelope["type"] == "transaction"
         assert envelope["contexts"]["trace"]["op"] == "queue.task.rq"
         assert envelope["transaction"] == "tests.integrations.rq.test_rq.do_trick"
+        assert (
+            envelope["contexts"]["trace"]["data"][SPANDATA.MESSAGING_DESTINATION_NAME]
+            == queue.name
+        )
         assert envelope["extra"]["rq-job"] == DictionaryContaining(
             {
                 "args": ["Maisey"] if send_default_pii else SENSITIVE_DATA_SUBSTITUTE,