View to multiple queues

Signed-off-by: kiblik <5609770+kiblik@users.noreply.github.com>

Author: kiblik

dojo/celery.py

@@ -45,7 +45,7 @@ def __call__(self, *args, **kwargs):
 
 
 @app.task(bind=True)
-def debug_task(self, priority=0):
+def debug_task(self, priority=1):
     logger.info(f"Request: {self.request!r}")
 
 

dojo/jira_link/helper.py

@@ -777,7 +777,7 @@ def push_to_jira(obj, *args, **kwargs):
 
 # we need thre separate celery tasks due to the decorators we're using to map to/from ids
 @dojo_async_task
-@app.task(priority=3)
+@app.task(priority=4)
 def push_finding_to_jira(finding_id, *args, **kwargs):
     finding = get_object_or_none(Finding, id=finding_id)
     if not finding:
@@ -790,7 +790,7 @@ def push_finding_to_jira(finding_id, *args, **kwargs):
 
 
 @dojo_async_task
-@app.task(priority=3)
+@app.task(priority=4)
 def push_finding_group_to_jira(finding_group_id, *args, **kwargs):
     finding_group = get_object_or_none(Finding_Group, id=finding_group_id)
     if not finding_group:
@@ -807,7 +807,7 @@ def push_finding_group_to_jira(finding_group_id, *args, **kwargs):
 
 
 @dojo_async_task
-@app.task(priority=3)
+@app.task(priority=4)
 def push_engagement_to_jira(engagement_id, *args, **kwargs):
     engagement = get_object_or_none(Engagement, id=engagement_id)
     if not engagement:
@@ -1397,7 +1397,7 @@ def jira_check_attachment(issue, source_file_name):
 
 
 @dojo_async_task
-@app.task(priority=3)
+@app.task(priority=4)
 def close_epic(engagement_id, push_to_jira, **kwargs):
     engagement = get_object_or_none(Engagement, id=engagement_id)
     if not engagement:
@@ -1446,7 +1446,7 @@ def close_epic(engagement_id, push_to_jira, **kwargs):
 
 
 @dojo_async_task
-@app.task(priority=3)
+@app.task(priority=4)
 def update_epic(engagement_id, **kwargs):
     engagement = get_object_or_none(Engagement, id=engagement_id)
     if not engagement:
@@ -1493,7 +1493,7 @@ def update_epic(engagement_id, **kwargs):
 
 
 @dojo_async_task
-@app.task(priority=3)
+@app.task(priority=4)
 def add_epic(engagement_id, **kwargs):
     engagement = get_object_or_none(Engagement, id=engagement_id)
     if not engagement:

dojo/notifications/helper.py

@@ -200,7 +200,7 @@ class SlackNotificationManger(NotificationManagerHelpers):
     """Manger for slack notifications and their helpers."""
 
     @dojo_async_task
-    @app.task(priority=3)
+    @app.task(priority=4)
     def send_slack_notification(
         self,
         event: str,
@@ -318,7 +318,7 @@ class MSTeamsNotificationManger(NotificationManagerHelpers):
     """Manger for Microsoft Teams notifications and their helpers."""
 
     @dojo_async_task
-    @app.task(priority=3)
+    @app.task(priority=4)
     def send_msteams_notification(
         self,
         event: str,
@@ -369,7 +369,7 @@ class EmailNotificationManger(NotificationManagerHelpers):
     """Manger for email notifications and their helpers."""
 
     @dojo_async_task
-    @app.task(priority=3)
+    @app.task(priority=4)
     def send_mail_notification(
         self,
         event: str,
@@ -421,7 +421,7 @@ class WebhookNotificationManger(NotificationManagerHelpers):
     ERROR_TEMPORARY = "temporary"
 
     @dojo_async_task
-    @app.task(priority=3)
+    @app.task(priority=4)
     def send_webhooks_notification(
         self,
         event: str,
@@ -559,7 +559,7 @@ def _test_webhooks_notification(self, endpoint: Notification_Webhooks) -> None:
         # in "send_webhooks_notification", we are doing deeper analysis, why it failed
         # for now, "raise_for_status" should be enough
 
-    @app.task(ignore_result=True, priority=1)
+    @app.task(ignore_result=True, priority=2)
     def _webhook_reactivation(self, endpoint_id: int, **_kwargs: dict):
         endpoint = Notification_Webhooks.objects.get(pk=endpoint_id)
         # User already changed status of endpoint
@@ -875,7 +875,7 @@ def _process_notifications(
                 )
 
 
-@app.task(ignore_result=True, priority=1)
+@app.task(ignore_result=True, priority=2)
 def webhook_status_cleanup(*_args: list, **_kwargs: dict):
     # If some endpoint was affected by some outage (5xx, 429, Timeout) but it was clean during last 24 hours,
     # we consider this endpoint as healthy so need to reset it

dojo/risk_acceptance/helper.py

@@ -167,7 +167,7 @@ def add_findings_to_risk_acceptance(user: Dojo_User, risk_acceptance: Risk_Accep
     post_jira_comments(risk_acceptance, findings, accepted_message_creator)
 
 
-@app.task(priority=3)
+@app.task(priority=4)
 def expiration_handler(*args, **kwargs):
     """
     Creates a notification upon risk expiration and X days beforehand if configured.

dojo/settings/settings.dist.py

@@ -17,7 +17,6 @@
 import environ
 import pghistory
 from celery.schedules import crontab
-from kombu import Queue
 from netaddr import IPNetwork, IPSet
 
 from dojo import __version__
@@ -1255,28 +1254,27 @@ def saml2_attrib_map_format(din):
 else:
     CELERY_BROKER_TRANSPORT_OPTIONS = {}
 
+# There are 5 queues. Lower number = higher priority. Plus one legacy (for backward compatibility)
+CELERY_QUEUE_TYPES = {
+    "celery": "legacy",
+    "celery:1": "live checks",
+    "celery:2": "keep system consistent (e.g. activation/deactivation webhooks)",
+    "celery:3": "regular tasks",
+    "celery:4": "notifications and jira",
+    "celery:5": "garbidge collectors",
+}
+
 if not CELERY_BROKER_URL.startswith("sqla+sqlite"):  # Priority queues are not supported by sqlite based engine
     if "queue_order_strategy" not in CELERY_BROKER_TRANSPORT_OPTIONS:
         CELERY_BROKER_TRANSPORT_OPTIONS["queue_order_strategy"] = "priority"
 
     if "priority_steps" not in CELERY_BROKER_TRANSPORT_OPTIONS:
-        # There are 4 queues. Lower number = higher priority
-        # 0 - live checks
-        # 1 - keep system consistent (e.g. activation/deavtivation webhooks)
-        # 2 - regular tasks
-        # 3 - notifications and jira
-        # 4 - garbidge collectors
-        CELERY_BROKER_TRANSPORT_OPTIONS["priority_steps"] = list(range(5))
-    CELERY_TASK_DEFAULT_PRIORITY = 2
+        CELERY_BROKER_TRANSPORT_OPTIONS["priority_steps"] = list(range(len(CELERY_QUEUE_TYPES)))
+    CELERY_TASK_DEFAULT_PRIORITY = 3
 
     if "sep" not in CELERY_BROKER_TRANSPORT_OPTIONS:
         CELERY_BROKER_TRANSPORT_OPTIONS["sep"] = ":"
 
-    # TODO: needs to be tested
-    CELERY_TASK_QUEUES = [
-        Queue("celery"),  # This handles BOTH legacy AND is the base for priority queues
-    ]
-
 
 CELERY_IMPORTS = ("dojo.tools.tool_issue_updater", )
 

dojo/system_settings/views.py

@@ -113,9 +113,11 @@ def get_celery_status(
 
             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."
+                context["celery_q_len"] = "It is not possible to identify number of waiting tasks."
+            elif len(q_len):
+                context["celery_q_len"] = f"{sum(q_len.values())} tasks are waiting to be proccessed."
+                context["celery_stats"] = q_len
+                context["celery_stats_desc"] = settings.CELERY_QUEUE_TYPES
             else:
                 context["celery_q_len"] = "No task is waiting to be proccessed."
 

dojo/tasks.py

@@ -30,7 +30,7 @@ def log_generic_alert(source, title, description):
                         icon="bullseye", source=source)
 
 
-@app.task(bind=True, priority=3)
+@app.task(bind=True, priority=4)
 def add_alerts(self, runinterval):
     now = timezone.now()
 
@@ -76,7 +76,7 @@ def add_alerts(self, runinterval):
             calculate_grade(product.id)
 
 
-@app.task(bind=True, priority=4)
+@app.task(bind=True, priority=5)
 def cleanup_alerts(*args, **kwargs):
     try:
         max_alerts_per_user = settings.MAX_ALERTS_PER_USER
@@ -94,12 +94,12 @@ def cleanup_alerts(*args, **kwargs):
         logger.info("total number of alerts deleted: %s", total_deleted_count)
 
 
-@app.task(bind=True, priority=4)
+@app.task(bind=True, priority=5)
 def flush_auditlog(*args, **kwargs):
     run_flush_auditlog()
 
 
-@app.task(bind=True, priority=4)
+@app.task(bind=True, priority=5)
 def async_dupe_delete(*args, **kwargs):
     # Wrap with pghistory context for audit trail
     with pghistory.context(source="dupe_delete_task"):
@@ -173,7 +173,7 @@ def _async_dupe_delete_impl():
                     calculate_grade(product.id)
 
 
-@app.task(ignore_result=False, priority=0)
+@app.task(ignore_result=False, priority=1, expires=10)  # It is expected to respond in 5 seconds. If it is in a queue more than 10, it is not necessary to respond anymore.
 def celery_status():
     return True
 
@@ -237,13 +237,13 @@ def evaluate_pro_proposition(*args, **kwargs):
     announcement.save()
 
 
-@app.task(priority=4)
+@app.task(priority=5)
 def clear_sessions(*args, **kwargs):
     call_command("clearsessions")
 
 
 @dojo_async_task
-@app.task(priority=4)
+@app.task(priority=5)
 def update_watson_search_index_for_model(model_name, pk_list, *args, **kwargs):
     """
     Async task to update watson search indexes for a specific model type.

dojo/templates/dojo/system_settings.html

@@ -28,6 +28,24 @@ <h4> Celery <span class="label label-danger">{{celery_status}}</span> </h4>
                     </div>
                     <div class="panel-body text-left">
                         {{celery_q_len}}
+                        {% if celery_stats %}
+                            <table id="general" class="tablesorter-bootstrap table table-condensed table-striped">
+                                <thead>
+                                    <tr>
+                                        <th>Name of the Quene</th>
+                                        <th>Number of Tasks</th>
+                                        <th>Description of the Queue</th>
+                                    </tr>
+                                </thead>
+                                {% for key, value in celery_stats.items %}
+                                    <tr>
+                                        <td>{{ key | first }}</td>
+                                        <td>{{ value }}</td>
+                                        <td>{{ key | last }}</td>
+                                    </tr>
+                                {% endfor %}
+                            </table>
+                        {% endif %}
                     </div>
                 </div>
             </div>

dojo/utils.py

@@ -1332,15 +1332,21 @@ def get_celery_worker_status():
 
 
 def get_celery_queue_length():
+    queue_lens = {}
     try:
-        with Connection(settings.CELERY_BROKER_URL) as conn, conn.SimpleQueue("celery") as queue:
-            return queue.qsize()
-    except ChannelError as e:
-        if "NOT_FOUND" in str(e):
-            return 0
-        return None
+        with Connection(settings.CELERY_BROKER_URL) as conn:
+            for queue_name in settings.CELERY_QUEUE_TYPES:
+                try:
+                    with conn.SimpleQueue(queue_name) as queue:
+                        if size := queue.qsize():
+                            queue_lens[queue_name, settings.CELERY_QUEUE_TYPES[queue_name]] = size
+                except ChannelError as e:
+                    if "NOT_FOUND" in str(e):
+                        continue
+                    raise
     except:
         return None
+    return queue_lens
 
 
 # Used to display the counts and enabled tabs in the product view