Add pending metrics and age filter to scheduler allstatus

Author: [email protected]

src/memos/api/handlers/scheduler_handler.py

@@ -6,6 +6,7 @@
 """
 
 import json
+from datetime import datetime, timezone
 import time
 import traceback
 from collections import Counter
@@ -59,18 +60,22 @@ def _summarize_tasks(task_details: list[dict[str, Any]]) -> TaskSummary:
             waiting=counter.get("waiting", 0),
             in_progress=counter.get("in_progress", 0),
             completed=counter.get("completed", 0),
+            pending=counter.get("pending", counter.get("in_progress", 0)),
             failed=counter.get("failed", 0),
             cancelled=counter.get("cancelled", 0),
             total=total,
         )
 
-    def _aggregate_counts_from_redis(tracker: TaskStatusTracker) -> TaskSummary | None:
+    def _aggregate_counts_from_redis(
+        tracker: TaskStatusTracker, max_age_seconds: float = 86400
+    ) -> TaskSummary | None:
         """Stream status counts directly from Redis to avoid loading all task payloads."""
         redis_client = getattr(tracker, "redis", None)
         if not redis_client:
             return None
 
         counter = Counter()
+        now = datetime.now(timezone.utc).timestamp()
 
         # Scan task_meta keys, then hscan each hash in batches
         cursor: int | str = 0
@@ -83,6 +88,16 @@ def _aggregate_counts_from_redis(tracker: TaskStatusTracker) -> TaskSummary | No
                     for value in fields.values():
                         try:
                             payload = json.loads(value.decode("utf-8") if isinstance(value, bytes) else value)
+                            # Skip stale entries to reduce noise and load
+                            ts = payload.get("submitted_at") or payload.get("started_at")
+                            if ts:
+                                try:
+                                    ts_dt = datetime.fromisoformat(ts)
+                                    ts_seconds = ts_dt.timestamp()
+                                except Exception:
+                                    ts_seconds = None
+                                if ts_seconds and (now - ts_seconds) > max_age_seconds:
+                                    continue
                             status = payload.get("status")
                             if status:
                                 counter[status] += 1
@@ -101,6 +116,7 @@ def _aggregate_counts_from_redis(tracker: TaskStatusTracker) -> TaskSummary | No
             waiting=counter.get("waiting", 0),
             in_progress=counter.get("in_progress", 0),
             completed=counter.get("completed", 0),
+            pending=counter.get("pending", counter.get("in_progress", 0)),
             failed=counter.get("failed", 0),
             cancelled=counter.get("cancelled", 0),
             total=total,
@@ -120,6 +136,7 @@ def _aggregate_counts_from_redis(tracker: TaskStatusTracker) -> TaskSummary | No
         # Scheduler view: assume tracker contains scheduler tasks; overlay queue monitor for live queue depth
         sched_waiting = all_tasks_summary.waiting
         sched_in_progress = all_tasks_summary.in_progress
+        sched_pending = all_tasks_summary.pending
         sched_completed = all_tasks_summary.completed
         sched_failed = all_tasks_summary.failed
         sched_cancelled = all_tasks_summary.cancelled
@@ -129,17 +146,21 @@ def _aggregate_counts_from_redis(tracker: TaskStatusTracker) -> TaskSummary | No
             queue_status_data = mem_scheduler.task_schedule_monitor.get_tasks_status() or {}
             scheduler_waiting = 0
             scheduler_in_progress = 0
+            scheduler_pending = 0
             for key, value in queue_status_data.items():
                 if not key.startswith("scheduler:"):
                     continue
                 scheduler_in_progress += int(value.get("running", 0) or 0)
+                scheduler_pending += int(value.get("pending", value.get("running", 0)) or 0)
                 scheduler_waiting += int(value.get("remaining", 0) or 0)
             sched_waiting = scheduler_waiting
             sched_in_progress = scheduler_in_progress
+            sched_pending = scheduler_pending
 
         scheduler_summary = TaskSummary(
             waiting=sched_waiting,
             in_progress=sched_in_progress,
+            pending=sched_pending,
             completed=sched_completed,
             failed=sched_failed,
             cancelled=sched_cancelled,

src/memos/api/product_models.py

@@ -872,6 +872,7 @@ class TaskSummary(BaseModel):
 
     waiting: int = Field(0, description="Number of tasks waiting to run")
     in_progress: int = Field(0, description="Number of tasks currently running")
+    pending: int = Field(0, description="Number of tasks fetched by workers but not yet acknowledged")
     completed: int = Field(0, description="Number of tasks completed")
     failed: int = Field(0, description="Number of tasks failed")
     cancelled: int = Field(0, description="Number of tasks cancelled")

src/memos/mem_scheduler/monitors/task_schedule_monitor.py

@@ -29,7 +29,7 @@ def __init__(
 
     @staticmethod
     def init_task_status() -> dict:
-        return {"running": 0, "remaining": 0}
+        return {"running": 0, "remaining": 0, "pending": 0}
 
     def get_tasks_status(self) -> dict:
         if isinstance(self.queue, SchedulerRedisQueue):
@@ -158,6 +158,7 @@ def _get_local_tasks_status(self) -> dict:
             # running from dispatcher if available
             if self.dispatcher and hasattr(self.dispatcher, "get_running_task_count"):
                 task_status["running"] = int(self.dispatcher.get_running_task_count())
+                task_status["pending"] = task_status["running"]
         except Exception as e:
             logger.warning(f"Failed to collect local queue status: {e}")
         return task_status
@@ -200,11 +201,13 @@ async def _collect_async() -> dict:
                                 if group.get("name") == self.queue.consumer_group:
                                     pending = int(group.get("pending", 0))
                                     break
-                        # Remaining = total messages (xlen) - pending for our group
-                        remaining = max(0, int(xlen_val or 0))
+                        total_messages = max(0, int(xlen_val or 0))
+                        remaining = max(0, total_messages - pending)
                         local[stream_key]["running"] += pending
+                        local[stream_key]["pending"] += pending
                         local[stream_key]["remaining"] += remaining
                         local["running"] += pending
+                        local["pending"] += pending
                         local["remaining"] += remaining
                     return local
 
@@ -234,10 +237,12 @@ async def _collect_async() -> dict:
                 for group in groups_info:
                     if group.get("name") == self.queue.consumer_group:
                         pending = int(group.get("pending", 0))
-                        remaining = max(0, xlen_val)
+                        remaining = max(0, xlen_val - pending)
                         task_status[stream_key]["running"] += pending
+                        task_status[stream_key]["pending"] += pending
                         task_status[stream_key]["remaining"] += remaining
                         task_status["running"] += pending
+                        task_status["pending"] += pending
                         task_status["remaining"] += remaining
                         break