Add READY task age metrics to Redis metrics collector

Author: suda

django_tasks_redis/metrics/collectors.py

@@ -6,6 +6,7 @@
 """
 
 import logging
+from datetime import datetime, timezone
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
@@ -31,6 +32,8 @@ class RedisTaskMetricsCollector(Collector):
 
     Metrics exposed:
     - django_tasks_queue_length: Gauge for current queue length by status
+    - django_tasks_queue_oldest_ready_age_seconds: Age of oldest READY task in seconds
+    - django_tasks_queue_newest_ready_age_seconds: Age of newest READY task in seconds
     """
 
     def __init__(self, backend: "RedisTaskBackend"):
@@ -82,6 +85,61 @@ def collect(self):
 
             yield queue_length
 
+            # Calculate age metrics for READY tasks
+            ready_count = status_counts.get("READY", 0)
+            if ready_count > 0:
+                ready_tasks, _ = self.backend.get_all_tasks(
+                    status="READY",
+                    limit=ready_count  # Get all READY tasks
+                )
+
+                if ready_tasks:
+                    now = datetime.now(timezone.utc)
+                    ages = []
+
+                    for task in ready_tasks:
+                        enqueued_at_str = task.get("enqueued_at", "")
+                        if enqueued_at_str:
+                            try:
+                                from django_tasks_redis.utils import deserialize_datetime
+                                enqueued_at = deserialize_datetime(enqueued_at_str)
+                                if enqueued_at:
+                                    # Make sure enqueued_at is timezone-aware
+                                    if enqueued_at.tzinfo is None:
+                                        enqueued_at = enqueued_at.replace(tzinfo=timezone.utc)
+                                    age_seconds = (now - enqueued_at).total_seconds()
+                                    ages.append(age_seconds)
+                            except Exception as e:
+                                logger.warning(
+                                    "Failed to parse enqueued_at for task: %s",
+                                    e
+                                )
+
+                    if ages:
+                        # Create gauge for oldest READY task age
+                        oldest_age = GaugeMetricFamily(
+                            "django_tasks_queue_oldest_ready_age_seconds",
+                            "Age of the oldest READY task in seconds",
+                            labels=["backend"],
+                        )
+                        oldest_age.add_metric(
+                            labels=[self.backend_name],
+                            value=max(ages),
+                        )
+                        yield oldest_age
+
+                        # Create gauge for newest READY task age
+                        newest_age = GaugeMetricFamily(
+                            "django_tasks_queue_newest_ready_age_seconds",
+                            "Age of the newest READY task in seconds",
+                            labels=["backend"],
+                        )
+                        newest_age.add_metric(
+                            labels=[self.backend_name],
+                            value=min(ages),
+                        )
+                        yield newest_age
+
             logger.debug(
                 "Collected metrics for backend %s: %s",
                 self.backend_name,

tests/test_metrics.py

@@ -2,6 +2,7 @@
 Tests for Prometheus metrics integration.
 """
 
+from datetime import datetime, timedelta, timezone
 from unittest.mock import Mock, patch
 
 import pytest
@@ -135,3 +136,142 @@ def test_collect_handles_errors(self):
 
         # Should return empty list on error
         assert len(metrics) == 0
+
+    @pytest.mark.skipif(
+        not _prometheus_available(),
+        reason="prometheus-client not installed",
+    )
+    def test_collect_with_ready_tasks_age_metrics(self):
+        """Test that collect() includes age metrics for READY tasks."""
+        from django_tasks_redis.metrics.collectors import RedisTaskMetricsCollector
+        from django_tasks_redis.utils import serialize_datetime
+
+        mock_backend = Mock()
+        mock_backend.alias = "test"
+        mock_backend.get_status_counts.return_value = {
+            TaskResultStatus.READY: 3,
+            TaskResultStatus.RUNNING: 1,
+        }
+
+        # Create mock tasks with different ages
+        now = datetime.now(timezone.utc)
+        oldest_task_time = now - timedelta(minutes=10)
+        middle_task_time = now - timedelta(minutes=5)
+        newest_task_time = now - timedelta(minutes=1)
+
+        mock_backend.get_all_tasks.return_value = (
+            [
+                {"enqueued_at": serialize_datetime(oldest_task_time)},
+                {"enqueued_at": serialize_datetime(middle_task_time)},
+                {"enqueued_at": serialize_datetime(newest_task_time)},
+            ],
+            3,
+        )
+
+        collector = RedisTaskMetricsCollector(mock_backend)
+
+        # Call collect
+        metrics = list(collector.collect())
+
+        # Verify get_all_tasks was called with READY status
+        mock_backend.get_all_tasks.assert_called_once_with(
+            status="READY",
+            limit=3
+        )
+
+        # Should return 3 metrics: queue_length, oldest_age, newest_age
+        assert len(metrics) == 3
+
+        metric_names = {m.name for m in metrics}
+        assert "django_tasks_queue_length" in metric_names
+        assert "django_tasks_queue_oldest_ready_age_seconds" in metric_names
+        assert "django_tasks_queue_newest_ready_age_seconds" in metric_names
+
+        # Check the age metrics values
+        oldest_metric = next(
+            m for m in metrics
+            if m.name == "django_tasks_queue_oldest_ready_age_seconds"
+        )
+        newest_metric = next(
+            m for m in metrics
+            if m.name == "django_tasks_queue_newest_ready_age_seconds"
+        )
+
+        # Get the metric values (samples are stored in the metric family)
+        oldest_samples = list(oldest_metric.samples)
+        newest_samples = list(newest_metric.samples)
+
+        assert len(oldest_samples) == 1
+        assert len(newest_samples) == 1
+
+        # Oldest should be ~600 seconds (10 minutes)
+        assert 590 <= oldest_samples[0].value <= 610
+
+        # Newest should be ~60 seconds (1 minute)
+        assert 50 <= newest_samples[0].value <= 70
+
+    @pytest.mark.skipif(
+        not _prometheus_available(),
+        reason="prometheus-client not installed",
+    )
+    def test_collect_without_ready_tasks_no_age_metrics(self):
+        """Test that collect() doesn't include age metrics when no READY tasks."""
+        from django_tasks_redis.metrics.collectors import RedisTaskMetricsCollector
+
+        mock_backend = Mock()
+        mock_backend.alias = "test"
+        mock_backend.get_status_counts.return_value = {
+            TaskResultStatus.RUNNING: 5,
+            TaskResultStatus.SUCCESSFUL: 10,
+        }
+
+        collector = RedisTaskMetricsCollector(mock_backend)
+
+        # Call collect
+        metrics = list(collector.collect())
+
+        # Should only return queue_length metric, no age metrics
+        assert len(metrics) == 1
+        assert metrics[0].name == "django_tasks_queue_length"
+
+        # get_all_tasks should not be called when no READY tasks
+        mock_backend.get_all_tasks.assert_not_called()
+
+    @pytest.mark.skipif(
+        not _prometheus_available(),
+        reason="prometheus-client not installed",
+    )
+    def test_collect_with_invalid_enqueued_at(self):
+        """Test that collect() handles tasks with invalid enqueued_at gracefully."""
+        from django_tasks_redis.metrics.collectors import RedisTaskMetricsCollector
+        from django_tasks_redis.utils import serialize_datetime
+
+        mock_backend = Mock()
+        mock_backend.alias = "test"
+        mock_backend.get_status_counts.return_value = {
+            TaskResultStatus.READY: 3,
+        }
+
+        now = datetime.now(timezone.utc)
+        valid_task_time = now - timedelta(minutes=5)
+
+        # Mix of valid and invalid tasks
+        mock_backend.get_all_tasks.return_value = (
+            [
+                {"enqueued_at": serialize_datetime(valid_task_time)},
+                {"enqueued_at": "invalid-datetime"},
+                {"enqueued_at": ""},
+            ],
+            3,
+        )
+
+        collector = RedisTaskMetricsCollector(mock_backend)
+
+        # Call collect - should not raise exception
+        metrics = list(collector.collect())
+
+        # Should still return metrics based on valid tasks
+        metric_names = {m.name for m in metrics}
+        assert "django_tasks_queue_length" in metric_names
+        assert "django_tasks_queue_oldest_ready_age_seconds" in metric_names
+        assert "django_tasks_queue_newest_ready_age_seconds" in metric_names