Add Queue Time Metrics for All Detected Queues

pyproject.toml

@@ -43,6 +43,7 @@ Flask = "^3.1.1"
 waitress = "^3.0.2"
 arrow = "^1.3.0"
 timy = "^0.4.2"
+cachetools = "^6.2.0"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^8.2.2"
@@ -64,6 +65,7 @@ pytest-celery = "^0.0.0"
 pylint = "^3.3.1"
 certifi = "^2024.8.30"
 idna = "^3.7"
+types-cachetools = "^6.2.0"
 
 [build-system]
 requires = ["poetry-core>=1.0.0a5"]

src/exporter.py

@@ -4,8 +4,9 @@
 import sys
 import time
 from collections import defaultdict
-from typing import Callable, Optional
+from typing import Callable, Dict, Optional
 
+from cachetools import LRUCache
 from celery import Celery
 from celery.events.state import State  # type: ignore
 from celery.utils import nodesplit  # type: ignore
@@ -31,6 +32,7 @@ def __init__(
         metric_prefix="celery_",
         default_queue_name="celery",
         static_label=None,
+        max_tasks_in_memory=50000,
     ):
         self.registry = CollectorRegistry(auto_describe=True)
         self.queue_cache = set(initial_queues or [])
@@ -46,6 +48,11 @@ def __init__(
         self.static_label = static_label or {}
         self.static_label_keys = self.static_label.keys()
 
+        # Track task received timestamps for latency calculation
+        self.task_received_times: Dict[str, float] = LRUCache(
+            maxsize=max_tasks_in_memory
+        )
+
         self.state_counters = {
             "task-sent": Counter(
                 f"{metric_prefix}task_sent",
@@ -122,6 +129,13 @@ def __init__(
             registry=self.registry,
             buckets=buckets or Histogram.DEFAULT_BUCKETS,
         )
+        self.celery_task_latency = Histogram(
+            f"{metric_prefix}task_latency",
+            "Histogram of task latency measurements (time between received and started).",
+            ["name", "hostname", "queue_name", *self.static_label_keys],
+            registry=self.registry,
+            buckets=buckets or Histogram.DEFAULT_BUCKETS,
+        )
         self.celery_queue_length = Gauge(
             f"{metric_prefix}queue_length",
             "The number of message in broker queue.",
@@ -194,6 +208,10 @@ def purge_worker_metrics(self, hostname):
             if hostname in label_seq:
                 self.celery_task_runtime.remove(*label_seq)
 
+        for label_seq in list(self.celery_task_latency._metrics.keys()):
+            if hostname in label_seq:
+                self.celery_task_latency.remove(*label_seq)
+
         del self.worker_last_seen[hostname]
 
     def track_timed_out_workers(self):
@@ -287,6 +305,29 @@ def track_task_event(self, event):
         if event["type"] == "task-sent" and self.generic_hostname_task_sent_metric:
             labels["hostname"] = "generic"
 
+        # Store timestamp when task is received for latency calculation
+        if event["type"] == "task-received":
+            self.task_received_times[event["uuid"]] = event.get(
+                "local_received"
+            ) or event.get("timestamp")
+            logger.debug("Stored received timestamp for task uuid='{}'", event["uuid"])
+
+        # Calculate and observe latency when task starts
+        if event["type"] == "task-started":
+            received_time = self.task_received_times.get(event["uuid"])
+            if received_time:
+                started_time = event.get("local_received") or event.get("timestamp")
+                latency = started_time - received_time
+                self.celery_task_latency.labels(**labels).observe(latency)
+                logger.debug(
+                    "Observed metric='{}' labels='{}': {}s",
+                    self.celery_task_latency._name,
+                    labels,
+                    latency,
+                )
+                # Clean up stored timestamp
+                del self.task_received_times[event["uuid"]]
+
         for counter_name, counter in self.state_counters.items():
             _labels = labels.copy()
 

src/test_metrics.py

@@ -231,3 +231,40 @@ def succeed():
             )
             is None
         )
+
+
+@pytest.mark.celery()
+def test_task_latency(threaded_exporter, celery_app, hostname):
+    time.sleep(5)
+
+    @celery_app.task
+    def latency_task():
+        time.sleep(0.1)
+
+    with start_worker(celery_app, without_heartbeat=False):
+        latency_task.delay().get()
+        time.sleep(2)
+
+        # Check that latency metric was recorded
+        latency_count = threaded_exporter.registry.get_sample_value(
+            "celery_task_latency_count",
+            labels={
+                "hostname": hostname,
+                "name": "src.test_metrics.latency_task",
+                "queue_name": "celery",
+            },
+        )
+        print(latency_count)
+        assert latency_count == 1.0
+
+        # Check that latency value is reasonable (should be very small for in-memory broker)
+        latency_sum = threaded_exporter.registry.get_sample_value(
+            "celery_task_latency_sum",
+            labels={
+                "hostname": hostname,
+                "name": "src.test_metrics.latency_task",
+                "queue_name": "celery",
+            },
+        )
+        assert latency_sum is not None
+        assert latency_sum >= 0  # Latency should be non-negative