populate task name cache from worker inspect

celerymon/cli/__init__.py

@@ -48,14 +48,16 @@ def run():
     redis_watcher = RedisWatcher.create_started(
         redis_client, args.queue, args.redis_watch_interval_sec
     )
-    worker_watcher = WorkerWatcher.create_started(app, args.worker_inspect_interval_sec)
     event_watcher = EventWatcher.create_started(
         app,
         buckets,
         queue_wait_buckets,
         args.in_flight_cache_size,
         args.in_flight_ttl_sec,
     )
+    worker_watcher = WorkerWatcher.create_started(
+        app, args.worker_inspect_interval_sec, event_watcher
+    )
     collector = Collector(redis_watcher, worker_watcher, event_watcher)
 
     registry = prometheus_client.CollectorRegistry()

celerymon/event_watcher.py

@@ -128,6 +128,7 @@ def __init__(
         in_flight_ttl_sec: int,
     ):
         self._task_names_by_uuid: OrderedDict[str, str] = OrderedDict()
+        self._task_names_by_uuid_lock = threading.Lock()
         self._worker_last_heartbeat: dict[str, datetime.datetime] = dict()
         self._in_flight: OrderedDict[str, InFlightEntry] = OrderedDict()
         self._in_flight_cache_size = in_flight_cache_size
@@ -171,10 +172,7 @@ def on_event(self, event: dict[str, Any]):
 
         uuid: str = event["uuid"]
         if "name" in event:
-            self._task_names_by_uuid.pop(uuid, None)
-            self._task_names_by_uuid[uuid] = event["name"]
-            while len(self._task_names_by_uuid) > self._TASK_NAMES_CACHE_LIMIT:
-                self._task_names_by_uuid.popitem(last=False)
+            self.record_task_name(uuid, event["name"])
 
         task_name = self._task_names_by_uuid.get(uuid, "(UNKNOWN)")
         self.task_names.add(task_name)
@@ -200,6 +198,13 @@ def on_event(self, event: dict[str, Any]):
                 self._eviction_counts["failed_pre_start"] += 1
             self._record_task_runtime(task_name, "failed", event)
 
+    def record_task_name(self, uuid: str, name: str) -> None:
+        with self._task_names_by_uuid_lock:
+            self._task_names_by_uuid[uuid] = name
+            self._task_names_by_uuid.move_to_end(uuid)
+            while len(self._task_names_by_uuid) > self._TASK_NAMES_CACHE_LIMIT:
+                self._task_names_by_uuid.popitem(last=False)
+
     def _on_worker_event(
         self,
         event_name: str,

celerymon/worker_watcher.py

@@ -8,6 +8,7 @@
 
 import celery
 
+from .event_watcher import EventWatcher
 from .timer import RepeatTimer
 
 
@@ -21,16 +22,18 @@ def create_started(
         cls,
         app: celery.Celery,
         interval: float,
+        event_watcher: EventWatcher,
     ) -> WorkerWatcher:
-        watcher = cls(app)
+        watcher = cls(app, event_watcher)
 
         timer = RepeatTimer(interval, watcher._update)
         timer.start()
 
         return watcher
 
-    def __init__(self, app: celery.Celery):
+    def __init__(self, app: celery.Celery, event_watcher: EventWatcher):
         self._inspect = app.control.inspect()
+        self._event_watcher = event_watcher
         self.last_updated_timestamp = None
         self.oldest_started_task_timestamp = dict()
         self.task_count = dict()
@@ -46,6 +49,7 @@ def _update(self) -> None:
                 else:
                     start_time = datetime.datetime.fromtimestamp(task["time_start"])
                 task_name = task["type"]
+                self._event_watcher.record_task_name(task["id"], task_name)
                 task_count[("active", task_name, hostname)] += 1
                 if task_name not in oldest_timestamp:
                     oldest_timestamp[task_name] = start_time
@@ -55,12 +59,15 @@ def _update(self) -> None:
                     )
         for hostname, tasks in (self._inspect.reserved() or {}).items():
             for task in tasks:
-                task_count[("reserved", task["type"], hostname)] += 1
+                task_name = task["type"]
+                self._event_watcher.record_task_name(task["id"], task_name)
+                task_count[("reserved", task_name, 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"], hostname)
-                ] += 1
+                request = scheduled_task["request"]
+                task_name = request["type"]
+                self._event_watcher.record_task_name(request["id"], task_name)
+                task_count[("scheduled", task_name, hostname)] += 1
 
         self.last_updated_timestamp = datetime.datetime.now(tz=datetime.UTC)
         self.oldest_started_task_timestamp = oldest_timestamp

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "celerymon"
-version = "0.1.10"
+version = "0.1.11"
 description = ""
 authors = [
     { name = "Masaya Suzuki", email = "masaya@aviator.co" },