[ENG-12524] Integrate dj-celery-panel - Part 1

addon_service/apps.py

@@ -1,9 +1,24 @@
+import importlib
+
 from django.apps import AppConfig
 
+from app.celery import app as celery_app
+
 
 class AddonServiceConfig(AppConfig):
     name = "addon_service"
 
     def ready(self):
+        from dj_control_room_base.models import BasePanelPlaceholder
+        from django.contrib import admin
+
         # need to import openapi extensions here for them to be registered
         import addon_service.common.openapi_extensions  # noqa: F401
+
+        # Workers import celery's `include` modules on startup; import them here too,
+        # so this process registers the same tasks (e.g. for the dj-celery-panel overview)
+        for module_name in celery_app.conf.include:
+            importlib.import_module(module_name)
+
+        # dj_control_room_base is installed only for dj_celery_panel's templates/static; hide its demo panel
+        admin.site.unregister(BasePanelPlaceholder)

addon_service/templates/admin/dj_celery_panel/queue_detail.html

@@ -0,0 +1,32 @@
+{% extends "admin/dj_celery_panel/queue_detail.html" %}
+{% load i18n %}
+
+{% block panel_page_content %}
+{{ block.super }}
+{% if queue %}
+<dl class="dcr-dl dcr-mt-sm">
+  <div class="dcr-dl__row">
+    <dt class="dcr-dl__label">{% trans "Routed Tasks" %}</dt>
+    <dd class="dcr-dl__value">
+      {% if queue.routed_tasks %}
+      <strong>{{ queue.routed_tasks|length }}</strong> {% trans "tasks" %}
+      <details class="dcr-mt-sm">
+        <summary>{% trans "Show routed tasks" %}</summary>
+        <ul class="dcr-mt-sm">
+          {% for task in queue.routed_tasks %}
+          <li>
+            <a href="{% url 'dj_celery_panel:tasks' %}?search={{ task|urlencode }}" class="dcr-data-table__link">
+              <code class="dcr-code">{{ task }}</code>
+            </a>
+          </li>
+          {% endfor %}
+        </ul>
+      </details>
+      {% else %}
+      <span class="dcr-text-muted">{% trans "No tasks are routed to this queue" %}</span>
+      {% endif %}
+    </dd>
+  </div>
+</dl>
+{% endif %}
+{% endblock %}

app/celery.py

@@ -60,6 +60,8 @@ def queue_name(self) -> str:
 )
 
 app.config_from_object("django.conf:settings", namespace="CELERY")
+# Make this app the fallback for celery.current_app in every thread, not only the importing one
+app.set_default()
 
 ###
 # additional consumer for custom (non-celery) messages from osf

app/celery_panel.py

@@ -0,0 +1,74 @@
+from dj_celery_panel.celery_utils import (
+    CeleryQueuesInspectBackend,
+    QueueDetailPage,
+    QueueListPage,
+)
+
+
+class RoutedTasksQueuesBackend(CeleryQueuesInspectBackend):
+    """dj-celery-panel queues backend that also lists the tasks routed to each queue
+    and the configured queues that no active worker consumes.
+
+    Routes are resolved the same way Celery does when sending a task, so explicit
+    queue= overrides passed to apply_async() at call time are not reflected.
+    """
+
+    BACKEND_DESCRIPTION = (
+        "Queue bindings, routing configuration, routed tasks and unconsumed queues"
+    )
+
+    def get_queues(self):
+        page = super().get_queues()
+        # Workers report only the queues they consume, so a configured queue nobody consumes
+        # would be missing from the list while tasks routed to it pile up
+        listed = {queue["name"] for queue in page.queues}
+        unconsumed = [
+            self.describe_unconsumed_queue(queue)
+            for name, queue in sorted(self.app.amqp.queues.items())
+            if name not in listed
+        ]
+        return QueueListPage(queues=page.queues + unconsumed, error=page.error)
+
+    def get_queue_detail(self, queue_name):
+        page = super().get_queue_detail(queue_name)
+        queue, error = page.queue, page.error
+        if queue is None:
+            # membership check first: indexing Queues creates missing ones
+            if queue_name not in self.app.amqp.queues:
+                return page
+            queue = self.describe_unconsumed_queue(self.app.amqp.queues[queue_name])
+            error = "No active worker consumes this queue"
+        return QueueDetailPage(
+            queue={**queue, "routed_tasks": self.get_routed_tasks(queue_name)},
+            error=error,
+        )
+
+    def describe_unconsumed_queue(self, queue):
+        broker_info = self._get_queue_length_from_broker(queue.name)
+        return {
+            "name": queue.name,
+            "exchange": queue.exchange.name,
+            "exchange_type": queue.exchange.type,
+            "routing_key": queue.routing_key,
+            "durable": queue.durable,
+            "auto_delete": queue.auto_delete,
+            "exclusive": queue.exclusive,
+            "arguments": queue.queue_arguments,
+            "workers": [],
+            "worker_details": [],
+            "message_count": broker_info.get("length"),
+            "consumer_count": broker_info.get("consumer_count"),
+            "broker_query_error": broker_info.get("error"),
+        }
+
+    def get_routed_tasks(self, queue_name):
+        # AddonServiceConfig.ready() imports every task module, so the local registry is complete
+        router = self.app.amqp.router
+        routed_tasks = []
+        for name, task in self.app.tasks.items():
+            if name.startswith("celery."):
+                continue
+            options = dict(task._get_exec_options())
+            if router.route(options, name)["queue"].name == queue_name:
+                routed_tasks.append(name)
+        return sorted(routed_tasks)

app/settings.py

@@ -111,6 +111,8 @@
     "rest_framework_json_api",
     "addon_service",
     "django_celery_beat",
+    "dj_control_room_base",
+    "dj_celery_panel",
     "drf_spectacular",
 ]
 
@@ -270,6 +272,14 @@
 OSF_BACKCHANNEL_QUEUE_NAME = env.OSF_BACKCHANNEL_QUEUE_NAME
 GV_QUEUE_NAME_PREFIX = env.GV_QUEUE_NAME_PREFIX
 
+DJ_CELERY_PANEL_SETTINGS = {
+    "queues_backend": "app.celery_panel.RoutedTasksQueuesBackend",
+    # No result backend is configured, so list the tasks workers are running right now
+    "tasks_backend": "dj_celery_panel.celery_utils.CeleryTasksInspectBackend",
+    # Beat runs DatabaseScheduler, so the database holds the effective schedule
+    "periodic_tasks_backend": "dj_celery_panel.celery_utils.CeleryPeriodicTasksDjangoCeleryBeatBackend",
+}
+
 # Celery Beat
 CELERY_BEAT_SCHEDULER = "django_celery_beat.schedulers:DatabaseScheduler"
 CELERY_BEAT_SCHEDULE = {

app/urls.py

@@ -14,6 +14,7 @@
 
 urlpatterns = [
     path("v1/", include("addon_service.urls")),
+    path("admin/dj-celery-panel/", include("dj_celery_panel.urls")),
     path("admin/", admin.site.urls),
     path(
         "docs",

pyproject.toml

@@ -24,6 +24,7 @@ boto3 = "1.34"
 boa-api = "^0.1.14"
 django-silk = "^5.3.2"
 django-celery-beat = "^2.7.0"
+dj-celery-panel = "0.6.0"
 tqdm = "^4.67.1"
 itsdangerous = "^2.2.0"
 redis = "^5.2.1"