diff --git a/addon_service/apps.py b/addon_service/apps.py index c2916203..29de23c6 100644 --- a/addon_service/apps.py +++ b/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) diff --git a/addon_service/templates/admin/dj_celery_panel/queue_detail.html b/addon_service/templates/admin/dj_celery_panel/queue_detail.html new file mode 100644 index 00000000..72765cc4 --- /dev/null +++ b/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 %} +
+
+
{% trans "Routed Tasks" %}
+
+ {% if queue.routed_tasks %} + {{ queue.routed_tasks|length }} {% trans "tasks" %} +
+ {% trans "Show routed tasks" %} + +
+ {% else %} + {% trans "No tasks are routed to this queue" %} + {% endif %} +
+
+
+{% endif %} +{% endblock %} diff --git a/app/celery.py b/app/celery.py index c972607d..51e5930f 100644 --- a/app/celery.py +++ b/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 diff --git a/app/celery_panel.py b/app/celery_panel.py new file mode 100644 index 00000000..a12eee34 --- /dev/null +++ b/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) diff --git a/app/settings.py b/app/settings.py index c471057d..24edeffa 100644 --- a/app/settings.py +++ b/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 = { diff --git a/app/urls.py b/app/urls.py index 47967f90..13c86f14 100644 --- a/app/urls.py +++ b/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", diff --git a/poetry.lock b/poetry.lock index e4ccad45..fdde496d 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 2.2.1 and should not be changed by hand. +# This file is automatically @generated by Poetry 2.5.1 and should not be changed by hand. [[package]] name = "aiohappyeyeballs" @@ -299,7 +299,7 @@ files = [ [package.dependencies] jmespath = ">=0.7.1,<2.0.0" python-dateutil = ">=2.1,<3.0.0" -urllib3 = {version = ">=1.25.4,<2.2.0 || >2.2.0,<3", markers = "python_version >= \"3.10\""} +urllib3 = {version = ">=1.25.4,!=2.2.0,<3", markers = "python_version >= \"3.10\""} [package.extras] crt = ["awscrt (==0.21.2)"] @@ -759,6 +759,45 @@ files = [ {file = "distlib-0.3.8.tar.gz", hash = "sha256:1530ea13e350031b6312d8580ddb6b27a104275a31106523b8f123787f494f64"}, ] +[[package]] +name = "dj-celery-panel" +version = "0.6.0" +description = "A Django Admin panel for browsing and inspecting Celery tasks and workers" +optional = false +python-versions = ">=3.9" +groups = ["main"] +files = [ + {file = "dj_celery_panel-0.6.0-py3-none-any.whl", hash = "sha256:b059af9edaedb64015357a0f520d389543dcf7e125495cbc8a3572c61afb9f8b"}, + {file = "dj_celery_panel-0.6.0.tar.gz", hash = "sha256:de915a8880474954f59f8af0a5462b8f647539c96f271ac26f22ab897218401d"}, +] + +[package.dependencies] +dj-control-room-base = ">=1.5.0" +Django = ">=4.2" + +[package.extras] +build = ["build (>=1.0.0)", "twine (>=4.0.0)"] +dev = ["django-redis (>=5.0.0)", "mkdocs-material (>=9.1.12)", "psycopg2-binary (>=2.9.0)", "pytest (>=7.0.0)", "pytest-cov (>=4.0.0)", "pytest-django (>=4.5.0)", "pytest-xdist (>=3.2.0)"] + +[[package]] +name = "dj-control-room-base" +version = "1.6.0" +description = "Core framework for Django Control Room panels" +optional = false +python-versions = ">=3.9" +groups = ["main"] +files = [ + {file = "dj_control_room_base-1.6.0-py3-none-any.whl", hash = "sha256:54e74e8655ff077f429722b956b2c574e5e5fb390da0882cfb41ebce1cab1551"}, + {file = "dj_control_room_base-1.6.0.tar.gz", hash = "sha256:250c11296baa7e84e4ea988783f70825598523d3fa30f8447db937dabb76b338"}, +] + +[package.dependencies] +Django = ">=4.2" + +[package.extras] +build = ["build (>=1.0.0)", "twine (>=4.0.0)"] +dev = ["django-redis (>=5.0.0)", "mkdocs-material (>=9.1.12)", "psycopg2-binary (>=2.9.0)", "pytest (>=7.0.0)", "pytest-cov (>=4.0.0)", "pytest-django (>=4.5.0)", "pytest-xdist (>=3.2.0)"] + [[package]] name = "django" version = "4.2.26" @@ -846,7 +885,7 @@ files = [ [package.dependencies] autopep8 = "*" Django = ">=4.2" -gprof2dot = ">=2017.09.19" +gprof2dot = ">=2017.9.19" sqlparse = "*" [[package]] @@ -1275,7 +1314,7 @@ files = [ [package.dependencies] attrs = ">=22.2.0" -jsonschema-specifications = ">=2023.03.6" +jsonschema-specifications = ">=2023.3.6" referencing = ">=0.28.4" rpds-py = ">=0.7.1" @@ -2284,10 +2323,10 @@ files = [ ] [package.dependencies] -botocore = ">=1.33.2,<2.0a.0" +botocore = ">=1.33.2,<2.0a0" [package.extras] -crt = ["botocore[crt] (>=1.33.2,<2.0a.0)"] +crt = ["botocore[crt] (>=1.33.2,<2.0a0)"] [[package]] name = "sentry-sdk" @@ -2749,4 +2788,4 @@ testing = ["coverage (>=5.0.3)", "zope.event", "zope.testing"] [metadata] lock-version = "2.1" python-versions = "^3.12" -content-hash = "43f63125b6043f67c908e34bf7af8cbb027afde3ff5375c631772971dd6345ba" +content-hash = "2ac648dad93ba8cb5b776c841ba85149f22ea1903b1f8fc716edb6e8ed39c9c1" diff --git a/pyproject.toml b/pyproject.toml index 9b3870aa..a401e7ae 100644 --- a/pyproject.toml +++ b/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"