feat: add System Status page with Celery queue monitoring and purge

dojo/api_v2/serializers.py

@@ -3051,6 +3051,14 @@ def validate(self, data):
         return data
 
 
+class CeleryStatusSerializer(serializers.Serializer):
+    worker_status = serializers.BooleanField(read_only=True)
+    queue_length = serializers.IntegerField(allow_null=True, read_only=True)
+    task_time_limit = serializers.IntegerField(allow_null=True, read_only=True)
+    task_soft_time_limit = serializers.IntegerField(allow_null=True, read_only=True)
+    task_default_expires = serializers.IntegerField(allow_null=True, read_only=True)
+
+
 class FindingNoteSerializer(serializers.Serializer):
     note_id = serializers.IntegerField()
 

dojo/api_v2/views.py

@@ -33,6 +33,7 @@
 from rest_framework.parsers import MultiPartParser
 from rest_framework.permissions import DjangoModelPermissions, IsAuthenticated
 from rest_framework.response import Response
+from rest_framework.views import APIView
 
 import dojo.finding.helper as finding_helper
 import dojo.jira_link.helper as jira_helper
@@ -179,9 +180,12 @@
 from dojo.utils import (
     async_delete,
     generate_file_response,
+    get_celery_queue_length,
+    get_celery_worker_status,
     get_setting,
     get_system_setting,
     process_tag_notifications,
+    purge_celery_queue,
 )
 
 logger = logging.getLogger(__name__)
@@ -3123,6 +3127,49 @@ def get_queryset(self):
         return System_Settings.objects.all().order_by("id")
 
 
+@extend_schema(
+    responses=serializers.CeleryStatusSerializer,
+    summary="Get Celery worker and queue status",
+    description=(
+        "Returns Celery worker liveness, pending queue length, and the active task "
+        "timeout/expiry configuration. Uses the Celery control channel (pidbox) for "
+        "worker status so it works correctly even when the task queue is clogged."
+    ),
+)
+class CeleryStatusView(APIView):
+    permission_classes = (permissions.IsSuperUser, DjangoModelPermissions)
+    queryset = System_Settings.objects.none()
+
+    def get(self, request):
+        data = {
+            "worker_status": get_celery_worker_status(),
+            "queue_length": get_celery_queue_length(),
+            "task_time_limit": getattr(settings, "CELERY_TASK_TIME_LIMIT", None),
+            "task_soft_time_limit": getattr(settings, "CELERY_TASK_SOFT_TIME_LIMIT", None),
+            "task_default_expires": getattr(settings, "CELERY_TASK_DEFAULT_EXPIRES", None),
+        }
+        return Response(serializers.CeleryStatusSerializer(data).data)
+
+
+@extend_schema(
+    request=None,
+    responses={200: {"type": "object", "properties": {"purged": {"type": "integer"}}}},
+    summary="Purge all pending Celery tasks from the queue",
+    description=(
+        "Removes all pending tasks from the default Celery queue. Tasks already being "
+        "executed by workers are not affected. Note: if deduplication tasks were queued, "
+        "you may need to re-run deduplication manually via `python manage.py dedupe`."
+    ),
+)
+class CeleryQueuePurgeView(APIView):
+    permission_classes = (permissions.IsSuperUser, DjangoModelPermissions)
+    queryset = System_Settings.objects.none()
+
+    def post(self, request):
+        purged = purge_celery_queue()
+        return Response({"purged": purged})
+
+
 # Authorization: superuser
 @extend_schema_view(**schema_with_prefetch())
 class NotificationsViewSet(

dojo/settings/settings.dist.py

@@ -90,6 +90,17 @@
     DD_CELERY_BEAT_SCHEDULE_FILENAME=(str, root("dojo.celery.beat.db")),
     DD_CELERY_TASK_SERIALIZER=(str, "pickle"),
     DD_CELERY_LOG_LEVEL=(str, "INFO"),
+    # Hard ceiling on task runtime. When reached, the worker process is sent SIGKILL — no cleanup
+    # code runs. Always set higher than DD_CELERY_TASK_SOFT_TIME_LIMIT. (0 = disabled, no limit)
+    DD_CELERY_TASK_TIME_LIMIT=(int, 43200),        # default: 12 hours
+    # Raises SoftTimeLimitExceeded inside the task, giving it a chance to clean up before the hard
+    # kill. Set a few seconds below DD_CELERY_TASK_TIME_LIMIT so cleanup has time to finish.
+    # (0 = disabled, no limit)
+    DD_CELERY_TASK_SOFT_TIME_LIMIT=(int, 0),
+    # If a task sits in the broker queue for longer than this without being picked up by a worker,
+    # Celery silently discards it — it is never executed and no exception is raised. Does not
+    # affect tasks that are already running. (0 = disabled, no limit)
+    DD_CELERY_TASK_DEFAULT_EXPIRES=(int, 43200),   # default: 12 hours
     DD_TAG_BULK_ADD_BATCH_SIZE=(int, 1000),
     # Tagulous slug truncate unique setting. Set to -1 to use tagulous internal default (5)
     DD_TAGULOUS_SLUG_TRUNCATE_UNIQUE=(int, -1),
@@ -1249,6 +1260,13 @@ def saml2_attrib_map_format(din):
 CELERY_TASK_SERIALIZER = env("DD_CELERY_TASK_SERIALIZER")
 CELERY_LOG_LEVEL = env("DD_CELERY_LOG_LEVEL")
 
+if env("DD_CELERY_TASK_TIME_LIMIT") > 0:
+    CELERY_TASK_TIME_LIMIT = env("DD_CELERY_TASK_TIME_LIMIT")
+if env("DD_CELERY_TASK_SOFT_TIME_LIMIT") > 0:
+    CELERY_TASK_SOFT_TIME_LIMIT = env("DD_CELERY_TASK_SOFT_TIME_LIMIT")
+if env("DD_CELERY_TASK_DEFAULT_EXPIRES") > 0:
+    CELERY_TASK_DEFAULT_EXPIRES = env("DD_CELERY_TASK_DEFAULT_EXPIRES")
+
 if len(env("DD_CELERY_BROKER_TRANSPORT_OPTIONS")) > 0:
     CELERY_BROKER_TRANSPORT_OPTIONS = json.loads(env("DD_CELERY_BROKER_TRANSPORT_OPTIONS"))
 

dojo/system_settings/urls.py

@@ -8,4 +8,9 @@
         views.SystemSettingsView.as_view(),
         name="system_settings",
     ),
+    re_path(
+        r"^system_status$",
+        views.SystemStatusView.as_view(),
+        name="system_status",
+    ),
 ]

dojo/system_settings/views.py

@@ -1,6 +1,5 @@
 import logging
 
-from django.conf import settings
 from django.contrib import messages
 from django.core.exceptions import PermissionDenied
 from django.http import HttpRequest, HttpResponse
@@ -9,7 +8,7 @@
 
 from dojo.forms import SystemSettingsForm
 from dojo.models import System_Settings
-from dojo.utils import add_breadcrumb, get_celery_queue_length, get_celery_worker_status
+from dojo.utils import add_breadcrumb
 
 logger = logging.getLogger(__name__)
 
@@ -30,15 +29,10 @@ def get_context(
         request: HttpRequest,
     ) -> dict:
         system_settings_obj = self.get_settings_object()
-        # Set the initial context
-        context = {
+        return {
             "system_settings_obj": system_settings_obj,
             "form": self.get_form(request, system_settings_obj),
         }
-        # Check the status of celery
-        self.get_celery_status(context)
-
-        return context
 
     def get_form(
         self,
@@ -95,35 +89,6 @@ def validate_form(
             return request, True
         return request, False
 
-    def get_celery_status(
-        self,
-        context: dict,
-    ) -> None:
-        # Celery needs to be set with the setting: CELERY_RESULT_BACKEND = 'db+sqlite:///dojo.celeryresults.sqlite'
-        if hasattr(settings, "CELERY_RESULT_BACKEND"):
-            # Check the status of Celery by sending calling a celery task
-            context["celery_bool"] = get_celery_worker_status()
-
-            if context["celery_bool"]:
-                context["celery_msg"] = "Celery is processing tasks."
-                context["celery_status"] = "Running"
-            else:
-                context["celery_msg"] = "Celery does not appear to be up and running. Please ensure celery is running."
-                context["celery_status"] = "Not Running"
-
-            q_len = get_celery_queue_length()
-            if q_len is None:
-                context["celery_q_len"] = " It is not possible to identify number of waiting tasks."
-            elif q_len:
-                context["celery_q_len"] = f"{q_len} tasks are waiting to be proccessed."
-            else:
-                context["celery_q_len"] = "No task is waiting to be proccessed."
-
-        else:
-            context["celery_bool"] = False
-            context["celery_msg"] = "Celery needs to have the setting CELERY_RESULT_BACKEND = 'db+sqlite:///dojo.celeryresults.sqlite' set in settings.py."
-            context["celery_status"] = "Unknown"
-
     def get_template(self) -> str:
         return "dojo/system_settings.html"
 
@@ -148,9 +113,19 @@ def post(
         self.permission_check(request)
         # Set up the initial context
         context = self.get_context(request)
-        # Check the status of celery
         request, _ = self.validate_form(request, context)
         # Add some breadcrumbs
         add_breadcrumb(title="System settings", top_level=False, request=request)
         # Render the page
         return render(request, self.get_template(), context)
+
+
+class SystemStatusView(View):
+    def get(
+        self,
+        request: HttpRequest,
+    ) -> HttpResponse:
+        if not request.user.is_superuser:
+            raise PermissionDenied
+        add_breadcrumb(title="System status", top_level=False, request=request)
+        return render(request, "dojo/system_status.html")

dojo/templates/base.html

@@ -613,6 +613,11 @@
                                                             {% trans "System Settings" %}
                                                         </a>
                                                     </li>
+                                                    <li>
+                                                        <a href="{% url 'system_status' %}">
+                                                            {% trans "System Status" %}
+                                                        </a>
+                                                    </li>
                                                 {% endif %}
                                                 {% if "dojo.view_tool_configuration"|has_configuration_permission:request %}
                                                     <li>

dojo/templates/dojo/system_settings.html

@@ -10,29 +10,6 @@
 
 {% block content %}
     {{ block.super }}
-    {% block status %}
-        <div class="row">
-            <h3> System Status </h3>
-            <br>
-            <div id="test-strategy" class="col-md-4">
-                <div class="panel panel-default">
-                    <div class="panel-heading">
-                        {% if celery_bool %}
-                        <h4> Celery <span class="label label-success">{{celery_status}}</span> </h4>
-                        {% else %}
-                        <h4> Celery <span class="label label-danger">{{celery_status}}</span> </h4>
-                        {% endif %}
-                    </div>
-                    <div class="panel-body text-left">
-                        {{celery_msg}}
-                    </div>
-                    <div class="panel-body text-left">
-                        {{celery_q_len}}
-                    </div>
-                </div>
-            </div>
-        </div>
-    {% endblock status %}
     <hr>
     {% block settings %}
     <div class="row">

dojo/templates/dojo/system_status.html

@@ -0,0 +1,132 @@
+{% extends "base.html" %}
+{% load static %}
+
+{% block content %}
+    {{ block.super }}
+    <div class="row">
+        <h3>System Status</h3>
+        <br>
+        <div id="celery-status-panel" class="col-md-5">
+            <div class="panel panel-default">
+                <div class="panel-heading">
+                    <h4>Celery <span id="celery-status-badge" class="label label-default">Loading...</span></h4>
+                </div>
+                <div class="panel-body text-left" id="celery-status-msg">—</div>
+                <div class="panel-body text-left">
+                    <span id="celery-queue-msg">—</span>
+                    <div style="margin-top: 8px;">
+                        <button id="purge-queue-btn" class="btn btn-danger btn-xs" disabled title="Loading...">
+                            Purge queue
+                        </button>
+                        <button id="refresh-status-btn" class="btn btn-default btn-xs" style="margin-left: 6px;">
+                            <span class="glyphicon glyphicon-refresh"></span> Refresh
+                        </button>
+                    </div>
+                    <p class="text-muted" style="margin-top: 8px; font-size: 0.9em">
+                        <strong>Note:</strong> Purging the queue removes pending tasks that have not yet been executed,
+                        including deduplication tasks. If deduplication tasks were in the queue, you may need to
+                        re-run deduplication manually using the <code>dedupe</code> management command:
+                        <code>python manage.py dedupe</code>
+                    </p>
+                </div>
+                <div class="panel-body text-left">
+                    <table class="table table-condensed" style="margin-bottom: 0">
+                        <thead><tr><th>Setting</th><th>Value</th></tr></thead>
+                        <tbody id="celery-config-tbody"></tbody>
+                        <tfoot>
+                            <tr>
+                                <td colspan="2" class="text-muted" style="font-size: 0.9em; border-top: 1px solid #ddd; padding-top: 6px">
+                                    Read-only. Configured via environment variables:
+                                    <code>DD_CELERY_TASK_TIME_LIMIT</code>,
+                                    <code>DD_CELERY_TASK_SOFT_TIME_LIMIT</code>,
+                                    <code>DD_CELERY_TASK_DEFAULT_EXPIRES</code>
+                                </td>
+                            </tr>
+                        </tfoot>
+                    </table>
+                </div>
+            </div>
+        </div>
+    </div>
+{% endblock content %}
+
+{% block postscript %}
+    {{ block.super }}
+    <script>
+        (function() {
+            var purgeUrl = "{% url 'celery_queue_purge_api' %}";
+            var statusUrl = "{% url 'celery_status_api' %}";
+
+            function renderCeleryStatus(data) {
+                var badge = $('#celery-status-badge');
+                if (data.worker_status) {
+                    badge.text('Running').removeClass('label-default label-danger').addClass('label-success');
+                    $('#celery-status-msg').text('Celery is processing tasks.');
+                } else {
+                    badge.text('Not Running').removeClass('label-default label-success').addClass('label-danger');
+                    $('#celery-status-msg').text('Celery does not appear to be running.');
+                }
+
+                var qLen = data.queue_length;
+                if (qLen === null) {
+                    $('#celery-queue-msg').text('It is not possible to identify the number of waiting tasks.');
+                    $('#purge-queue-btn').prop('disabled', true).attr('title', 'Broker unreachable');
+                } else {
+                    $('#celery-queue-msg').text(qLen + ' task(s) waiting to be processed.');
+                    $('#purge-queue-btn').prop('disabled', false).removeAttr('title');
+                }
+
+                function humanDuration(seconds) {
+                    if (seconds === null) return '<em>Not set</em>';
+                    var d = Math.floor(seconds / 86400);
+                    var h = Math.floor((seconds % 86400) / 3600);
+                    var m = Math.floor((seconds % 3600) / 60);
+                    var s = seconds % 60;
+                    var parts = [];
+                    if (d) parts.push(d + 'd');
+                    if (h) parts.push(h + 'h');
+                    if (m) parts.push(m + 'm');
+                    if (s || parts.length === 0) parts.push(s + 's');
+                    return parts.join(' ') + ' <span class="text-muted">(' + seconds + 's)</span>';
+                }
+
+                var cfgRows = [
+                    ['Task time limit (hard kill)', data.task_time_limit],
+                    ['Task soft time limit', data.task_soft_time_limit],
+                    ['Default task expiry (queue)', data.task_default_expires],
+                ];
+                var tbody = $('#celery-config-tbody').empty();
+                cfgRows.forEach(function(row) {
+                    tbody.append('<tr><td>' + row[0] + '</td><td>' + humanDuration(row[1]) + '</td></tr>');
+                });
+            }
+
+            function loadCeleryStatus() {
+                $.get(statusUrl).done(renderCeleryStatus).fail(function() {
+                    $('#celery-status-badge').text('Error').removeClass('label-default label-success').addClass('label-danger');
+                    $('#celery-status-msg').text('Failed to load Celery status.');
+                });
+            }
+
+            $(function() {
+                loadCeleryStatus();
+
+                $('#refresh-status-btn').on('click', loadCeleryStatus);
+
+                $('#purge-queue-btn').on('click', function() {
+                    if (!confirm('Purge all pending tasks from the queue? This cannot be undone.')) return;
+                    $.ajax({
+                        url: purgeUrl,
+                        method: 'POST',
+                        headers: {'X-CSRFToken': '{{ csrf_token }}'},
+                    }).done(function(data) {
+                        location.reload();
+                    }).fail(function() {
+                        alert('Purge failed. Check server logs.');
+                        loadCeleryStatus();
+                    });
+                });
+            });
+        })();
+    </script>
+{% endblock %}