feat(celery): Set task ID on span

Set Celery task ID on the "queue.process" span, and add tests for this new behavior.

Closes #2974

sentry_sdk/consts.py

@@ -270,6 +270,11 @@ class SPANDATA:
     e.g. the queue name or topic.
     """
 
+    MESSAGING_MESSAGE_ID = "messaging.message.id"
+    """
+    The message's identifier.
+    """
+
     MESSAGING_MESSAGE_RETRY_COUNT = "messaging.message.retry.count"
     """
     Number of retries/attempts to process a message.

sentry_sdk/integrations/celery/__init__.py

@@ -356,6 +356,8 @@ def _inner(*args, **kwargs):
                 op=OP.QUEUE_PROCESS, description=task.name
             ) as span:
                 _set_messaging_destination_name(task, span)
+                with capture_internal_exceptions():
+                    span.set_data(SPANDATA.MESSAGING_MESSAGE_ID, task.request.id)
                 with capture_internal_exceptions():
                     span.set_data(
                         SPANDATA.MESSAGING_MESSAGE_RETRY_COUNT, task.request.retries

tests/integrations/celery/test_celery.py

@@ -661,6 +661,20 @@ def task(): ...
     assert "messaging.destination.name" not in span["data"]
 
 
+def test_messaging_id(init_celery, capture_events):
+    celery = init_celery(enable_tracing=True)
+    events = capture_events()
+
+    @celery.task
+    def example_task(): ...
+
+    example_task.apply_async()
+
+    (event,) = events
+    (span,) = event["spans"]
+    assert "messaging.message.id" in span["data"]
+
+
 def test_retry_count_zero(init_celery, capture_events):
     celery = init_celery(enable_tracing=True)
     events = capture_events()