Merge branch 'main' into codex/issue-1295-package-release-date

Author: abhey8

CHANGELOG.rst

@@ -1,6 +1,12 @@
 Release notes
 =============
 
+Version v38.4.0
+---------------------
+
+- fix: run pipeline scheduling jobs in respective queues (https://github.com/aboutcode-org/vulnerablecode/pull/2263)
+- feat: show queue load factors on the pipeline dashboard (https://github.com/aboutcode-org/vulnerablecode/pull/2264)
+
 Version v38.3.0
 ---------------------
 

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = vulnerablecode
-version = 38.3.0
+version = 38.4.0
 license = Apache-2.0 AND CC-BY-SA-4.0
 
 # description must be on ONE line https://github.com/pypa/setuptools/issues/1390

vulnerabilities/migrations/0121_advisoryv2_is_latest_alter_advisoryv2_advisory_id_and_more.py

@@ -0,0 +1,65 @@
+# Generated by Django 5.2.11 on 2026-04-13 19:05
+
+from django.db import migrations
+from django.db import models
+from django.db.models import F
+
+
+class Migration(migrations.Migration):
+    def add_is_latest_on_existing_advisory(apps, schema_editor):
+        Advisory = apps.get_model("vulnerabilities", "AdvisoryV2")
+
+        print(f"\nUpdating is_latest on existing V2 Advisory.")
+        latest_qs = Advisory.objects.order_by(
+            "avid",
+            F("date_collected").desc(nulls_last=True),
+            "-id",
+        ).distinct("avid")
+
+        Advisory.objects.filter(id__in=latest_qs.values("id")).update(is_latest=True)
+
+    dependencies = [
+        ("vulnerabilities", "0120_impactedpackage_last_range_unfurl_at_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="advisoryv2",
+            name="is_latest",
+            field=models.BooleanField(
+                db_index=True,
+                default=False,
+                help_text="Indicates whether this is the latest version of the advisory identified by its AVID.",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="advisoryv2",
+            name="advisory_id",
+            field=models.CharField(
+                db_index=True,
+                help_text="An advisory is a unique vulnerability identifier in some database, such as PYSEC-2020-2233",
+                max_length=500,
+            ),
+        ),
+        migrations.AlterField(
+            model_name="advisoryv2",
+            name="datasource_id",
+            field=models.CharField(
+                db_index=True,
+                help_text="Unique ID for the datasource used for this advisory .e.g.: nginx_importer_v2",
+                max_length=200,
+            ),
+        ),
+        migrations.AddConstraint(
+            model_name="advisoryv2",
+            constraint=models.UniqueConstraint(
+                condition=models.Q(("is_latest", True)),
+                fields=("avid",),
+                name="unique_latest_per_avid",
+            ),
+        ),
+        migrations.RunPython(
+            code=add_is_latest_on_existing_advisory,
+            reverse_code=migrations.RunPython.noop,
+        ),
+    ]

vulnerabilities/models.py

@@ -2358,7 +2358,11 @@ def save(self, *args, **kwargs):
         if not self.pk:
             self.schedule_work_id = self.create_new_job(execute_now=True)
         elif self.pk and (existing := PipelineSchedule.objects.get(pk=self.pk)):
-            if existing.is_active != self.is_active or existing.run_interval != self.run_interval:
+            if (
+                existing.is_active != self.is_active
+                or existing.run_interval != self.run_interval
+                or existing.run_priority != self.run_priority
+            ):
                 self.schedule_work_id = self.create_new_job()
         self.full_clean()
         return super().save(*args, **kwargs)
@@ -2390,6 +2394,11 @@ def all_runs(self):
     def latest_run(self):
         return self.pipelineruns.first() if self.pipelineruns.exists() else None
 
+    @property
+    def latest_successful_run(self):
+        successful_runs = self.pipelineruns.filter(run_end_date__isnull=False, run_exitcode=0)
+        return successful_runs.first() if successful_runs.exists() else None
+
     @property
     def earliest_run(self):
         return self.pipelineruns.earliest("run_start_date") if self.pipelineruns.exists() else None
@@ -2874,21 +2883,10 @@ def to_dict(self):
 
 class AdvisoryV2QuerySet(BaseQuerySet):
     def latest_for_avid(self, avid: str):
-        return (
-            self.filter(avid=avid)
-            .order_by(
-                F("date_collected").desc(nulls_last=True),
-                "-id",
-            )
-            .first()
-        )
+        return self.get(avid=avid, is_latest=True)
 
     def latest_per_avid(self):
-        return self.order_by(
-            "avid",
-            F("date_collected").desc(nulls_last=True),
-            "-id",
-        ).distinct("avid")
+        return self.filter(is_latest=True)
 
     def latest_for_avids(self, avids):
         return self.filter(avid__in=avids).latest_per_avid()
@@ -3005,6 +3003,7 @@ class AdvisoryV2(models.Model):
         max_length=200,
         blank=False,
         null=False,
+        db_index=True,
         help_text="Unique ID for the datasource used for this advisory ." "e.g.: nginx_importer_v2",
     )
 
@@ -3014,6 +3013,7 @@ class AdvisoryV2(models.Model):
         blank=False,
         null=False,
         unique=False,
+        db_index=True,
         help_text="An advisory is a unique vulnerability identifier in some database, "
         "such as PYSEC-2020-2233",
     )
@@ -3088,6 +3088,14 @@ class AdvisoryV2(models.Model):
         help_text="UTC Date on which the advisory was collected",
     )
 
+    is_latest = models.BooleanField(
+        default=False,
+        blank=False,
+        null=False,
+        db_index=True,
+        help_text="Indicates whether this is the latest version of the advisory identified by its AVID.",
+    )
+
     original_advisory_text = models.TextField(
         blank=True,
         null=True,
@@ -3140,6 +3148,11 @@ class AdvisoryV2(models.Model):
     class Meta:
         unique_together = ["datasource_id", "advisory_id", "unique_content_id"]
         ordering = ["datasource_id", "advisory_id", "date_published", "unique_content_id"]
+        constraints = [
+            models.UniqueConstraint(
+                fields=["avid"], condition=Q(is_latest=True), name="unique_latest_per_avid"
+            )
+        ]
         indexes = [
             models.Index(
                 fields=["avid", "-date_collected", "-id"],

vulnerabilities/pipes/advisory.py

@@ -334,6 +334,13 @@ def insert_advisory_v2(
     if not created:
         return advisory_obj
 
+    AdvisoryV2.objects.filter(
+        avid=f"{pipeline_id}/{advisory.advisory_id}",
+        is_latest=True,
+    ).update(is_latest=False)
+    advisory_obj.is_latest = True
+    advisory_obj.save()
+
     aliases = get_or_create_advisory_aliases(aliases=advisory.aliases)
     references = get_or_create_advisory_references(references=advisory.references)
     severities = get_or_create_advisory_severities(severities=advisory.severities)

vulnerabilities/schedules.py

@@ -24,6 +24,7 @@ def schedule_execution(pipeline_schedule, execute_now=False):
     Takes a `PackageSchedule` object as input and schedule a
     recurring job using `rq_scheduler` to execute the pipeline.
     """
+    queue_name = pipeline_schedule.get_run_priority_display()
     first_execution = datetime.datetime.now(tz=datetime.timezone.utc)
     if not execute_now:
         first_execution = pipeline_schedule.next_run_date
@@ -36,6 +37,7 @@ def schedule_execution(pipeline_schedule, execute_now=False):
         args=[pipeline_schedule.pipeline_id],
         interval=interval_in_seconds,
         repeat=None,
+        queue_name=queue_name,
     )
     return job._id
 

vulnerabilities/tasks.py

@@ -9,24 +9,23 @@
 
 
 import logging
+from collections import Counter
+from contextlib import suppress
 from io import StringIO
 from traceback import format_exc as traceback_format_exc
 
 import django_rq
+from redis.exceptions import ConnectionError
+from rq import Worker
 
 from vulnerabilities import models
 from vulnerabilities.importer import Importer
 from vulnerabilities.improver import Improver
+from vulnerablecode.settings import RQ_QUEUES
 
 logger = logging.getLogger(__name__)
 
-default_queue = django_rq.get_queue("default")
-high_queue = django_rq.get_queue("high")
-
-queues = {
-    "default": django_rq.get_queue("default"),
-    "high": django_rq.get_queue("high"),
-}
+queues = {queue: django_rq.get_queue(queue) for queue in RQ_QUEUES.keys()}
 
 
 def execute_pipeline(pipeline_id, run_id):
@@ -151,3 +150,61 @@ def dequeue_job(job_id):
     for queue in queues.values():
         if job_id in queue.jobs:
             queue.remove(job_id)
+
+
+def compute_queue_load_factor():
+    """
+    Compute worker load per queue.
+
+    Load factor is the ratio of the total compute required to run all active pipelines
+    in a queue to the available worker capacity for that queue over a 24-hour period.
+    A value greater than 1 indicates that the number of workers is insufficient to
+    run all pipelines within the schedule.
+
+    Also compute the additional workers needed to balance each queue
+    """
+    field = models.PipelineSchedule._meta.get_field("run_priority")
+    label_to_value = {label: value for value, label in field.choices}
+    total_compute_seconds_per_queue = {}
+    worker_per_queue = {}
+    load_per_queue = {}
+    seconds_in_24_hr = 86400
+
+    with suppress(ConnectionError):
+        redis_conn = django_rq.get_connection()
+        queue_names = [
+            w.queue_names()[0] for w in Worker.all(connection=redis_conn) if w.queue_names()
+        ]
+        worker_per_queue = dict(Counter(queue_names))
+
+    for queue in RQ_QUEUES.keys():
+        total_compute_seconds_per_queue[queue] = sum(
+            (p.latest_successful_run.runtime / (p.run_interval / 24))
+            for p in models.PipelineSchedule.objects.filter(
+                is_active=True, run_priority=label_to_value[queue]
+            )
+            if p.latest_successful_run
+        )
+        if queue not in worker_per_queue:
+            worker_per_queue[queue] = 0
+
+    for queue_name, worker_count in worker_per_queue.items():
+        net_load_on_queue = "no_worker"
+        total_compute = total_compute_seconds_per_queue.get(queue_name, 0)
+        if total_compute == 0:
+            continue
+
+        unit_load_on_queue = total_compute / seconds_in_24_hr
+
+        num_of_worker_for_balanced_queue = round(unit_load_on_queue)
+        addition_worker_needed = max(num_of_worker_for_balanced_queue - worker_count, 0)
+
+        if worker_count > 0:
+            net_load_on_queue = unit_load_on_queue / worker_count
+
+        load_per_queue[queue_name] = {
+            "load_factor": net_load_on_queue,
+            "additional_worker": addition_worker_needed,
+        }
+
+    return dict(sorted(load_per_queue.items(), key=lambda x: x[0], reverse=True))

vulnerabilities/templates/pipeline_dashboard.html

@@ -1,5 +1,7 @@
 {% extends "base.html" %}
 
+{% load utils %}
+
 {% block title %}
 Pipeline Dashboard
 {% endblock %}
@@ -22,6 +24,18 @@
     .column {
         word-break: break-word;
     }
+
+    .has-text-orange {
+        color: #ff8c42 !important;
+    }
+
+    .has-tooltip-orange::before {
+        background-color: #ff8c42 !important;
+    }
+
+    .has-tooltip-orange::after {
+        border-top-color: #ff8c42 !important;
+    }
 </style>
 {% endblock %}
 
@@ -48,11 +62,63 @@ <h1>Pipeline Dashboard</h1>
         </form>
 
         <div class="box">
-            <div class="column has-text-right">
-                <p class="has-text-weight-semibold">
-                    {{ active_pipeline_count|default:0 }} active pipeline{{ active_pipeline_count|default:0|pluralize }},
-                    {{ disabled_pipeline_count|default:0 }} disabled pipeline{{ disabled_pipeline_count|default:0|pluralize }}
-                </p>
+            <div class="columns is-multiline is-vcentered mb-0">
+                <div class="column is-half has-text-left">
+                    {% if load_per_queue %}
+                    <p class="ml-3">
+                        <span class="has-text-weight-bold has-text-black is-size-6 has-tooltip-arrow has-tooltip-multiline"
+                            data-tooltip="Load factor is the ratio of the total compute required to run all active pipelines
+                                in a queue to the available worker capacity for that queue over a 24-hour period.
+                                A value greater than 1 indicates that the number of workers is insufficient to
+                                run all pipelines within the schedule.">
+                            Load Factor:
+                        </span>
+                        {% for queue_name, values in load_per_queue.items %}
+
+                            <span class="has-text-weight-bold is-size-6 has-tooltip-arrow has-tooltip-multiline"
+                                data-tooltip="{{ queue_name|capfirst }} priority pipeline queue.">
+                                {{ queue_name| capfirst }}
+                            </span>
+                            {% with load_factor=values|get_item:"load_factor" additional=values|get_item:"additional_worker" %}
+                                {% if load_factor == "no_worker" %}
+                                    <span class="has-text-weight-bold is-size-6 has-text-danger has-tooltip-arrow has-tooltip-multiline has-tooltip-danger"
+                                            data-tooltip="All workers in the {{ queue_name }} queue are down. Please run {{ additional }}
+                                            worker{{ additional|pluralize }} for the {{ queue_name }} queue.">
+                                        <span class="icon"><i class="fa fa-exclamation-triangle"></i></span>
+                                    </span>
+                                {% elif load_factor < 1 %}
+                                    <span class="has-text-weight-bold is-size-6 has-text-success has-tooltip-arrow has-tooltip-multiline has-tooltip-success"
+                                            data-tooltip="{{ queue_name|capfirst }} queue perfectly balanced.">
+                                        {{ load_factor|floatformat:2 }}
+                                        <span class="icon"><i class="fa fa-check-circle"></i></span>
+                                    </span>
+                                {% elif load_factor < 1.6 %}
+                                    <span class="has-text-weight-bold is-size-6 has-text-orange has-tooltip-arrow has-tooltip-multiline has-tooltip-orange"
+                                    data-tooltip="Consider adding {{ additional }} additional worker{{ additional|pluralize }} to the {{ queue_name }} queue.">
+                                        {{ load_factor|floatformat:2 }}
+                                        <span class="icon"><i class="fa fa-info-circle"></i></span>
+                                    </span>
+                                {% else %}
+                                    <span class="has-text-weight-bold is-size-6 has-text-danger has-tooltip-arrow has-tooltip-multiline has-tooltip-danger"
+                                            data-tooltip="Consider adding {{ additional }} additional worker{{ additional|pluralize }} to the {{ queue_name }} queue.">
+                                        {{ load_factor|floatformat:2 }}
+                                        <span class="icon"><i class="fa fa-exclamation-circle"></i></span>
+                                    </span>
+                                {% endif %}
+                            {% endwith %}
+
+                            {% if not forloop.last %} &bull; {% endif %}
+
+                        {% endfor %}
+                    </p>
+                    {% endif %}
+                </div>
+                <div class="column is-half has-text-right">
+                    <p class="has-text-grey-dark has-text-weight-semibold mr-3">
+                        {{ active_pipeline_count|default:0 }} active pipeline{{ active_pipeline_count|default:0|pluralize }},
+                        {{ disabled_pipeline_count|default:0 }} disabled pipeline{{ disabled_pipeline_count|default:0|pluralize }}
+                    </p>
+                </div>
             </div>
             <table class="table is-striped  is-hoverable is-fullwidth">
                 <thead>