feat(celery): Send Celery retry count

sentry_sdk/consts.py

@@ -270,6 +270,21 @@ 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.
+    """
+
+    MESSAGING_SYSTEM = "messaging.system"
+    """
+    The messaging system's name, e.g. `kafka`, `aws_sqs`
+    """
+
     SERVER_ADDRESS = "server.address"
     """
     Name of the database host.

sentry_sdk/integrations/celery/__init__.py

@@ -356,6 +356,18 @@ 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
+                    )
+                with capture_internal_exceptions():
+                    span.set_data(
+                        SPANDATA.MESSAGING_SYSTEM,
+                        task.app.connection().transport.driver_type,
+                    )
+
                 return f(*args, **kwargs)
         except Exception:
             exc_info = sys.exc_info()

tests/integrations/celery/test_celery.py

@@ -28,7 +28,7 @@ def init_celery(sentry_init, request):
     def inner(propagate_traces=True, backend="always_eager", **kwargs):
         sentry_init(
             integrations=[CeleryIntegration(propagate_traces=propagate_traces)],
-            **kwargs
+            **kwargs,
         )
         celery = Celery(__name__)
 
@@ -659,3 +659,66 @@ def task(): ...
     (event,) = events
     (span,) = event["spans"]
     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()
+
+    @celery.task()
+    def task(): ...
+
+    task.apply_async()
+
+    (event,) = events
+    (span,) = event["spans"]
+    assert span["data"]["messaging.message.retry.count"] == 0
+
+
+@mock.patch("celery.app.task.Task.request")
+def test_retry_count_nonzero(mock_request, init_celery, capture_events):
+    mock_request.retries = 3
+
+    celery = init_celery(enable_tracing=True)
+    events = capture_events()
+
+    @celery.task()
+    def task(): ...
+
+    task.apply_async()
+
+    (event,) = events
+    (span,) = event["spans"]
+    assert span["data"]["messaging.message.retry.count"] == 3
+
+
+@pytest.mark.parametrize("system", ("redis", "amqp"))
+def test_messaging_system(system, init_celery, capture_events):
+    celery = init_celery(enable_tracing=True)
+    events = capture_events()
+
+    # Does not need to be a real URL, since we use always eager
+    celery.conf.broker_url = f"{system}://example.com"  # noqa: E231
+
+    @celery.task()
+    def task(): ...
+
+    task.apply_async()
+
+    (event,) = events
+    (span,) = event["spans"]
+    assert span["data"]["messaging.system"] == system