add failed runtime, online worker count, hostname label

Author: tulioz

README.md

@@ -63,16 +63,17 @@ on the finished tasks. celerymon uses all these three to get the data.
 
 ## Metrics
 
-| Name                                                      | Type      | Labels                    |
-|-----------------------------------------------------------|-----------|---------------------------|
-| `celerymon_redis_last_updated_timestamp_seconds`          | Gauge     |                           |
-| `celerymon_redis_queue_item_count`                        | Gauge     | `queue_name`, `priority`  |
-| `celerymon_inspect_last_updated_timestamp_seconds`        | Gauge     |                           |
-| `celerymon_inspect_oldest_started_task_timestamp_seconds` | Gauge     | `task_name`               |
-| `celerymon_inspect_worker_held_task_count`                | Gauge     | `task_name`, `state`      |
-| `celerymon_events_last_received_timestamp_seconds`        | Gauge     | `task_name`, `event_name` |
-| `celerymon_events_count`                                  | Counter   | `task_name`, `event_name` |
-| `celerymon_events_success_task_runtime_seconds`           | Histogram | `task_name`               |
+| Name                                                      | Type      | Labels                           |
+|-----------------------------------------------------------|-----------|----------------------------------|
+| `celerymon_redis_last_updated_timestamp_seconds`          | Gauge     |                                  |
+| `celerymon_redis_queue_item_count`                        | Gauge     | `queue_name`, `priority`         |
+| `celerymon_inspect_last_updated_timestamp_seconds`        | Gauge     |                                  |
+| `celerymon_inspect_oldest_started_task_timestamp_seconds` | Gauge     | `task_name`                      |
+| `celerymon_inspect_worker_held_task_count`                | Gauge     | `task_name`, `state`, `hostname` |
+| `celerymon_events_last_received_timestamp_seconds`        | Gauge     | `task_name`, `event_name`        |
+| `celerymon_events_count`                                  | Counter   | `task_name`, `event_name`        |
+| `celerymon_events_task_runtime_seconds`                   | Histogram | `task_name`, `result`            |
+| `celerymon_events_online_worker_count`                    | Gauge     |                                  |
 
 There are timestamp metrics. These are meant to be used for checking the
 monitoring health. If this stops updating, it means that the monitoring cannot

celerymon/event_watcher.py

@@ -23,8 +23,8 @@ class EventWatcher:
     num_events_per_task_count: dict[tuple[str, str], int]
     upper_bounds: list[float]
     task_names: set[str]
-    succeeded_task_runtime_sec: list[dict[str, int]]
-    succeeded_task_runtime_sec_sum: dict[str, float]
+    task_runtime_sec: list[dict[tuple[str, str], int]]
+    task_runtime_sec_sum: dict[tuple[str, str], float]
 
     @classmethod
     def create_started(
@@ -71,6 +71,7 @@ def update_enable_event() -> None:
 
     def __init__(self, buckets: Sequence[float | str]):
         self._task_names_by_uuid: OrderedDict[str, str] = OrderedDict()
+        self._worker_last_heartbeat: dict[str, datetime.datetime] = dict()
 
         self.upper_bounds = [float(b) for b in buckets]
         if self.upper_bounds and self.upper_bounds[-1] != float("inf"):
@@ -81,16 +82,21 @@ def __init__(self, buckets: Sequence[float | str]):
         self.last_received_timestamp = None
         self.last_received_timestamp_per_task_event = dict()
         self.num_events_per_task_count = defaultdict(int)
-        self.succeeded_task_runtime_sec = []
-        self.succeeded_task_runtime_sec_sum = defaultdict(float)
-        for _ in range(0, len(self.upper_bounds)):
-            self.succeeded_task_runtime_sec.append(defaultdict(int))
+        self.task_runtime_sec = [
+            defaultdict(int) for _ in range(len(self.upper_bounds))
+        ]
+        self.task_runtime_sec_sum = defaultdict(float)
 
     def on_event(self, event: dict[str, Any]):
         now = datetime.datetime.now(tz=datetime.UTC)
         self.last_received_timestamp = now
 
         event_name: str = event["type"]
+
+        if event_name.startswith("worker-"):
+            self._on_worker_event(event_name, event, now)
+            return
+
         if not event_name.startswith("task-"):
             return
 
@@ -108,10 +114,52 @@ def on_event(self, event: dict[str, Any]):
         self.num_events_per_task_count[(task_name, event_name)] += 1
 
         if event_name == "task-succeeded":
-            # Not documented, but looking into the Celery codebase, the runtime
-            # looks like seconds.
-            runtime_sec = event["runtime"]
-            self.succeeded_task_runtime_sec_sum[task_name] += runtime_sec
-            for i, bound in enumerate(self.upper_bounds):
-                if runtime_sec <= bound:
-                    self.succeeded_task_runtime_sec[i][task_name] += 1
+            self._record_task_runtime(task_name, "success", event)
+
+        if event_name == "task-failed":
+            self._record_task_runtime(task_name, "failed", event)
+
+    def _on_worker_event(
+        self,
+        event_name: str,
+        event: dict[str, Any],
+        now: datetime.datetime,
+    ) -> None:
+        hostname = event.get("hostname")
+        if not hostname:
+            return
+        if event_name == "worker-offline":
+            self._worker_last_heartbeat.pop(hostname, None)
+        else:
+            self._worker_last_heartbeat[hostname] = now
+
+    def online_worker_count(
+        self,
+        now: datetime.datetime,
+        ttl_sec: int = 120,
+    ) -> int:
+        cutoff = now - datetime.timedelta(seconds=ttl_sec)
+        return sum(1 for ts in self._worker_last_heartbeat.values() if ts > cutoff)
+
+    def _record_task_runtime(
+        self,
+        task_name: str,
+        result: str,
+        event: dict[str, Any],
+    ) -> None:
+        # Celery sets `runtime` (seconds) on task-succeeded consistently, and on
+        # task-failed when the task actually ran. Defensive `.get()` handles
+        # failure modes where the task never executed.
+        runtime_sec = event.get("runtime")
+        if runtime_sec is None:
+            logger.debug(
+                "task event missing runtime field; task_name=%s result=%s",
+                task_name,
+                result,
+            )
+            return
+        key = (task_name, result)
+        self.task_runtime_sec_sum[key] += runtime_sec
+        for i, bound in enumerate(self.upper_bounds):
+            if runtime_sec <= bound:
+                self.task_runtime_sec[i][key] += 1

celerymon/metrics.py

@@ -3,6 +3,7 @@
 
 from __future__ import annotations
 
+import datetime
 from typing import Iterable
 
 import prometheus_client
@@ -85,7 +86,7 @@ def worker_metrics(watcher: WorkerWatcher) -> list[prometheus_client.Metric]:
     active_task_count_metric = GaugeMetricFamily(
         name="celerymon_inspect_worker_held_task_count",
         documentation="Number of tasks held in all workers",
-        labels=["task_name", "state"],
+        labels=["task_name", "state", "hostname"],
     )
     if watcher.last_updated_timestamp is not None:
         last_updated_timestamp_seconds_metric.add_metric(
@@ -100,9 +101,9 @@ def worker_metrics(watcher: WorkerWatcher) -> list[prometheus_client.Metric]:
                 timestamp=watcher.last_updated_timestamp.timestamp(),
             )
         for key, count in watcher.task_count.items():
-            state, task_name = key
+            state, task_name, hostname = key
             active_task_count_metric.add_metric(
-                labels=[task_name, state],
+                labels=[task_name, state, hostname],
                 value=count,
                 timestamp=watcher.last_updated_timestamp.timestamp(),
             )
@@ -125,12 +126,21 @@ def event_metrics(watcher: EventWatcher) -> list[prometheus_client.Metric]:
         documentation="The task event count per task name and event name.",
         labels=["task_name", "event_name"],
     )
-    success_task_runtime_seconds_metric = HistogramMetricFamily(
-        name="celerymon_events_success_task_runtime_seconds",
-        documentation="The task runtime per task name for finished success tasks.",
-        labels=["task_name"],
+    task_runtime_seconds_metric = HistogramMetricFamily(
+        name="celerymon_events_task_runtime_seconds",
+        documentation=(
+            "Task runtime per task name, labeled by result (success|failed)."
+        ),
+        labels=["task_name", "result"],
         unit="seconds",
     )
+    online_worker_count_metric = GaugeMetricFamily(
+        name="celerymon_events_online_worker_count",
+        documentation=(
+            "Number of workers currently online, derived from worker-online, "
+            "worker-heartbeat, and worker-offline events."
+        ),
+    )
     if watcher.last_received_timestamp is not None:
         for key, timestamp in watcher.last_received_timestamp_per_task_event.items():
             count = watcher.num_events_per_task_count[key]
@@ -147,19 +157,28 @@ def event_metrics(watcher: EventWatcher) -> list[prometheus_client.Metric]:
                 timestamp=timestamp.timestamp(),
             )
         for task_name in watcher.task_names:
-            acc = 0.0
-            buckets = []
-            for i, bound in enumerate(watcher.upper_bounds):
-                acc += watcher.succeeded_task_runtime_sec[i][task_name]
-                buckets.append((floatToGoString(bound), acc))
-            success_task_runtime_seconds_metric.add_metric(
-                labels=[task_name],
-                buckets=buckets,
-                sum_value=watcher.succeeded_task_runtime_sec_sum[task_name],
-                timestamp=watcher.last_received_timestamp.timestamp(),
-            )
+            for result in ("success", "failed"):
+                key = (task_name, result)
+                acc = 0.0
+                buckets = []
+                for i, bound in enumerate(watcher.upper_bounds):
+                    acc += watcher.task_runtime_sec[i].get(key, 0)
+                    buckets.append((floatToGoString(bound), acc))
+                task_runtime_seconds_metric.add_metric(
+                    labels=[task_name, result],
+                    buckets=buckets,
+                    sum_value=watcher.task_runtime_sec_sum.get(key, 0.0),
+                    timestamp=watcher.last_received_timestamp.timestamp(),
+                )
+        now = datetime.datetime.now(tz=datetime.UTC)
+        online_worker_count_metric.add_metric(
+            labels=[],
+            value=watcher.online_worker_count(now),
+            timestamp=watcher.last_received_timestamp.timestamp(),
+        )
     return [
         last_received_timestamp_seconds_metric,
         events_count_metric,
-        success_task_runtime_seconds_metric,
+        task_runtime_seconds_metric,
+        online_worker_count_metric,
     ]

celerymon/worker_watcher.py

@@ -14,7 +14,7 @@
 class WorkerWatcher:
     last_updated_timestamp: datetime.datetime | None
     oldest_started_task_timestamp: dict[str, datetime.datetime]
-    task_count: dict[tuple[str, str], int]
+    task_count: dict[tuple[str, str, str], int]
 
     @classmethod
     def create_started(
@@ -37,28 +37,30 @@ def __init__(self, app: celery.Celery):
 
     def _update(self) -> None:
         oldest_timestamp: dict[str, datetime.datetime] = dict()
-        task_count: dict[tuple[str, str], int] = defaultdict(int)
+        task_count: dict[tuple[str, str, str], int] = defaultdict(int)
 
-        for tasks in (self._inspect.active() or {}).values():
+        for hostname, tasks in (self._inspect.active() or {}).items():
             for task in tasks:
                 if isinstance(task["time_start"], str):
                     start_time = datetime.datetime.fromisoformat(task["time_start"])
                 else:
                     start_time = datetime.datetime.fromtimestamp(task["time_start"])
                 task_name = task["type"]
-                task_count[("active", task_name)] += 1
+                task_count[("active", task_name, hostname)] += 1
                 if task_name not in oldest_timestamp:
                     oldest_timestamp[task_name] = start_time
                 else:
                     oldest_timestamp[task_name] = min(
                         oldest_timestamp[task_name], start_time
                     )
-        for tasks in (self._inspect.reserved() or {}).values():
+        for hostname, tasks in (self._inspect.reserved() or {}).items():
             for task in tasks:
-                task_count[("reserved", task["type"])] += 1
-        for scheduled_tasks in (self._inspect.scheduled() or {}).values():
+                task_count[("reserved", task["type"], hostname)] += 1
+        for hostname, scheduled_tasks in (self._inspect.scheduled() or {}).items():
             for scheduled_task in scheduled_tasks:
-                task_count[("scheduled", scheduled_task["request"]["type"])] += 1
+                task_count[
+                    ("scheduled", scheduled_task["request"]["type"], hostname)
+                ] += 1
 
         self.last_updated_timestamp = datetime.datetime.now(tz=datetime.UTC)
         self.oldest_started_task_timestamp = oldest_timestamp