Worker status/monitoring for competition organizers (from rabbit)

src/settings/base.py

@@ -276,10 +276,6 @@
         'task': 'profiles.tasks.clean_non_activated_users',
         'schedule': timedelta(days=1),  # Run every 24 hours
     },
-    "refresh_compute_worker_health": {
-        "task": "competitions.tasks.refresh_compute_worker_health",
-        "schedule": 60,
-    },
 }
 CELERY_TIMEZONE = 'UTC'
 CELERY_WORKER_PREFETCH_MULTIPLIER = 1

src/static/riot/competitions/detail/_header.tag

@@ -36,13 +36,11 @@
                             Migrate
                         </button>
 
-                        <!--
                         <worker-monitor-toggle
                             if="{competition.admin}"
                             can_view_workers_panel="true"
                             competition_id="{ competition.id }">
                         </worker-monitor-toggle>
-                        -->
 
                     </div>
                     <div class="row">

src/templates/pages/monitor_queues.html

@@ -62,5 +62,21 @@ <h1>Monitor queues</h1>
             </div>
         </div>
     {% endif %}
+
+    {% if user.is_superuser %}
+    <div class="ui container">
+
+        <div class="ui segment">
+            <worker-monitor-toggle
+                can_view_workers_panel="true"
+                all_workers="true"
+                inline_mode="true">
+            </worker-monitor-toggle>
+        </div>
+
+        <div id="external_monitors" class="ui two column grid">
+        </div>
+    </div>
+    {% endif %}
 
 {% endblock %}

src/utils/consumers.py

@@ -1,59 +1,20 @@
 import asyncio
-import json
 import logging
-import time
-
-from competitions.models import Competition
 
 from asgiref.sync import sync_to_async
 from channels.generic.websocket import AsyncJsonWebsocketConsumer
-from django_redis import get_redis_connection
 
-from utils.worker_utils import WORKER_HEARTBEAT_TTL, WORKERS_REGISTRY_KEY
+from utils.worker_utils import fetch_compute_workers
 
 logger = logging.getLogger(__name__)
 
-r = get_redis_connection("default")
-
-
-def _load_snapshot(competition_queue_name=None):
-    """
-    Charge les workers depuis Redis.
-    - workers par défaut : toujours inclus (queue_source == 'default')
-    - workers privés : inclus uniquement si leur queue_source correspond
-      à la queue de la compétition courante
-    """
-    raw = r.hgetall(WORKERS_REGISTRY_KEY)
-    workers = []
-    private_workers = []
-    now = time.time()
-
-    for _, value in raw.items():
-        try:
-            worker = json.loads(value)
-        except Exception:
-            continue
-
-        if now - worker.get("last_seen", 0) > WORKER_HEARTBEAT_TTL:
-            continue
-
-        if worker.get("queue_source") == "default":
-            workers.append(worker)
-        else:
-            # Worker privé : n'afficher que si la queue correspond à la compétition
-            if competition_queue_name and worker.get("queue_source") == competition_queue_name:
-                private_workers.append(worker)
-
-    workers.sort(key=lambda x: x.get("hostname", ""))
-    private_workers.sort(key=lambda x: (x.get("queue_source", ""), x.get("hostname", "")))
-    return workers, private_workers
-
 
 def _get_competition_queue_name(competition_id):
-    """Retourne le nom de la queue de la compétition, ou None."""
     if not competition_id:
         return None
     try:
+        from competitions.models import Competition
+
         competition = Competition.objects.select_related("queue").get(pk=competition_id)
         if competition.queue and competition.queue.name:
             return competition.queue.name
@@ -62,6 +23,30 @@ def _get_competition_queue_name(competition_id):
     return None
 
 
+def _load_snapshot(competition_queue_name=None, show_all=False):
+    workers, private_workers, queue_stats = fetch_compute_workers()
+
+    if show_all:
+        pass
+    elif competition_queue_name:
+        private_workers = [
+            w
+            for w in private_workers
+            if w.get("queue_source") == competition_queue_name
+        ]
+        queue_stats = [
+            q
+            for q in queue_stats
+            if q.get("source_name") == competition_queue_name
+            or q.get("source_name") == "default"
+        ]
+    else:
+        private_workers = []
+        queue_stats = [q for q in queue_stats if q.get("source_name") == "default"]
+
+    return workers, private_workers, queue_stats
+
+
 class ComputeWorkersConsumer(AsyncJsonWebsocketConsumer):
 
     async def connect(self):
@@ -72,6 +57,7 @@ async def connect(self):
         await self.accept()
         await self.channel_layer.group_add("compute_workers", self.channel_name)
         self._competition_queue_name = None
+        self._show_all = False
         self._running = True
         self._subscribed = asyncio.Event()
         self._task = asyncio.create_task(self._push_workers_loop())
@@ -88,55 +74,41 @@ async def disconnect(self, close_code):
                 pass
 
     async def receive_json(self, content):
-        logger.debug("WebSocket received: %s", content)
         if content.get("type") == "subscribe":
-            competition_id = content.get("competition_id")
-            self._competition_queue_name = await sync_to_async(_get_competition_queue_name)(
-                competition_id)
+            if content.get("all_workers"):
+                self._show_all = True
+            else:
+                competition_id = content.get("competition_id")
+                self._competition_queue_name = await sync_to_async(
+                    _get_competition_queue_name
+                )(competition_id)
             self._subscribed.set()
 
     async def _push_workers_loop(self):
         try:
             try:
                 await asyncio.wait_for(self._subscribed.wait(), timeout=5.0)
             except asyncio.TimeoutError:
-                logger.warning("WebSocket subscribe timeout, proceeding without competition filter")
+                logger.warning("WebSocket subscribe timeout, proceeding without filter")
 
             while self._running:
-                workers, private_workers = await sync_to_async(_load_snapshot)(
-                    self._competition_queue_name
+                workers, private_workers, queue_stats = await sync_to_async(_load_snapshot)(
+                    competition_queue_name=self._competition_queue_name,
+                    show_all=self._show_all,
                 )
                 if not self._running:
                     break
                 try:
-                    await self.send_json({
-                        "type": "workers.snapshot",
-                        "workers": workers,
-                        "private_workers": private_workers,
-                    })
+                    await self.send_json(
+                        {
+                            "type": "workers.snapshot",
+                            "workers": workers,
+                            "private_workers": private_workers,
+                            "queue_stats": queue_stats,
+                        }
+                    )
                 except RuntimeError:
                     break
                 await asyncio.sleep(3)
         except asyncio.CancelledError:
             pass
-
-    async def worker_health(self, event):
-        worker = event["worker"]
-        is_default = worker.get("queue_source") == "default"
-        is_mine = (
-            self._competition_queue_name is not None
-            and worker.get("queue_source") == self._competition_queue_name
-        )
-        if not is_default and not is_mine:
-            return
-        try:
-            workers, private_workers = await sync_to_async(_load_snapshot)(
-                self._competition_queue_name
-            )
-            await self.send_json({
-                "type": "workers.snapshot",
-                "workers": workers,
-                "private_workers": private_workers,
-            })
-        except RuntimeError:
-            pass