Add Celery receive latency

sentry_sdk/consts.py

@@ -301,6 +301,11 @@ class SPANDATA:
     Number of retries/attempts to process a message.
     """
 
+    MESSAGING_MESSAGE_RECEIVE_LATENCY = "messaging.message.receive.latency"
+    """
+    The latency between when the task was enqueued and when it was started to be processed.
+    """
+
     MESSAGING_SYSTEM = "messaging.system"
     """
     The messaging system's name, e.g. `kafka`, `aws_sqs`

sentry_sdk/integrations/celery/__init__.py

@@ -181,6 +181,12 @@ def _update_celery_task_headers(original_headers, span, monitor_beat_tasks):
                 }
             )
 
+        # Add the time the task was enqueued to the headers
+        # This is used in the consumer to calculate the latency
+        updated_headers.update(
+            {"sentry-task-enqueued-time": _now_seconds_since_epoch()}
+        )
+
         if headers:
             existing_baggage = updated_headers.get(BAGGAGE_HEADER_NAME)
             sentry_baggage = headers.get(BAGGAGE_HEADER_NAME)
@@ -360,12 +366,25 @@ def _inner(*args, **kwargs):
                 op=OP.QUEUE_PROCESS, description=task.name
             ) as span:
                 _set_messaging_destination_name(task, span)
+
+                latency = None
+                with capture_internal_exceptions():
+                    if "sentry-task-enqueued-time" in task.request.headers:
+                        latency = _now_seconds_since_epoch() - task.request.headers.pop(
+                            "sentry-task-enqueued-time"
+                        )
+
+                if latency is not None:
+                    span.set_data(SPANDATA.MESSAGING_MESSAGE_RECEIVE_LATENCY, latency)
+
                 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,

tests/integrations/celery/test_celery.py

@@ -754,3 +754,18 @@ def task(): ...
     assert span["data"]["messaging.message.retry.count"] == 0
 
     monkeypatch.setattr(kombu.messaging.Producer, "_publish", old_publish)
+
+
+def test_receive_latency(init_celery, capture_events):
+    celery = init_celery(traces_sample_rate=1.0)
+    events = capture_events()
+
+    @celery.task()
+    def task(): ...
+
+    task.apply_async()
+
+    (event,) = events
+    (span,) = event["spans"]
+    assert "messaging.message.receive.latency" in span["data"]
+    assert span["data"]["messaging.message.receive.latency"] > 0