Add config and job to clean up old pipeline runs

vulnerabilities/models.py

@@ -14,6 +14,7 @@
 import xml.etree.ElementTree as ET
 from contextlib import suppress
 from functools import cached_property
+from inspect import cleandoc
 from itertools import groupby
 from operator import attrgetter
 from traceback import format_exc as traceback_format_exc
@@ -2073,7 +2074,7 @@ def pipeline_class(self):
 
     @property
     def job(self):
-        with suppress(NoSuchJobError):
+        with suppress(redis.exceptions.ConnectionError, NoSuchJobError):
             return Job.fetch(
                 str(self.run_id),
                 connection=django_rq.get_connection(),
@@ -2218,21 +2219,21 @@ def stop_run(self):
             job_id=str(self.run_id),
         )
         self.set_run_stopped()
+        self.delete_run_job()
 
-    def delete_run(self, delete_self=True):
+    def delete_run_job(self):
         if job := self.job:
             job.delete()
 
-        if delete_self:
-            self.delete()
-
     def delete(self, *args, **kwargs):
         """
         Before deletion of the run instance, try to stop the run execution.
         """
         with suppress(redis.exceptions.ConnectionError, AttributeError):
             self.stop_run()
 
+        self.delete_run_job()
+
         return super().delete(*args, **kwargs)
 
     def append_to_log(self, message, is_multiline=False):
@@ -2366,16 +2367,20 @@ def pipeline_class(self):
         from vulnerabilities.importers import IMPORTERS_REGISTRY
         from vulnerabilities.improvers import IMPROVERS_REGISTRY
         from vulnerabilities.pipelines.exporters import EXPORTERS_REGISTRY
+        from vulnerabilities.pipelines.management import MANAGEMENT_REGISTRY
 
-        pipeline_registry = IMPORTERS_REGISTRY | IMPROVERS_REGISTRY | EXPORTERS_REGISTRY
+        pipeline_registry = (
+            IMPORTERS_REGISTRY | IMPROVERS_REGISTRY | EXPORTERS_REGISTRY | MANAGEMENT_REGISTRY
+        )
 
         return pipeline_registry[self.pipeline_id]
 
     @property
     def description(self):
         """Return the pipeline class."""
+
         if self.pipeline_class:
-            return self.pipeline_class.__doc__
+            return cleandoc(self.pipeline_class.__doc__ or "")
 
     @property
     def all_runs(self):

vulnerabilities/pipelines/management/__init__.py

@@ -0,0 +1,16 @@
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+from vulnerabilities.pipelines.management import cleanup_pipeline_runs
+from vulnerabilities.utils import create_registry
+
+MANAGEMENT_REGISTRY = create_registry(
+    [
+        cleanup_pipeline_runs.CleanupPipelineRuns,
+    ]
+)

vulnerabilities/pipelines/management/cleanup_pipeline_runs.py

@@ -0,0 +1,64 @@
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+
+from datetime import timedelta
+
+from django.utils import timezone
+
+from vulnerabilities.models import PipelineSchedule
+from vulnerabilities.pipelines import LoopProgress
+from vulnerabilities.pipelines import VulnerableCodePipeline
+from vulnerablecode.settings import (
+    VULNERABLECODE_MINIMUM_PIPELINE_RUNS_TO_RETAIN as minimum_number_of_runs_to_keep,
+)
+from vulnerablecode.settings import VULNERABLECODE_PIPELINE_RUN_RETENTION_DAYS as retention_days
+
+
+class CleanupPipelineRuns(VulnerableCodePipeline):
+    """Remove pipeline runs older than the retention period while preserving the configured minimum number of runs."""
+
+    pipeline_id = "cleanup_pipeline_runs"
+
+    run_interval = 1440
+    run_priority = PipelineSchedule.ExecutionPriority.DEFAULT
+
+    @classmethod
+    def steps(cls):
+        return (cls.cleanup_old_pipeline_runs,)
+
+    def cleanup_old_pipeline_runs(self):
+        """Remove pipeline runs older than the retention period while preserving the configured minimum number of runs."""
+
+        pipelines = PipelineSchedule.objects.all()
+        pipeline_count = 0
+        deleted_run_count = 0
+
+        progress = LoopProgress(
+            total_iterations=pipelines.count(), progress_step=5, logger=self.log
+        )
+        for pipeline in progress.iter(pipelines):
+            cutoff = timezone.now() - timedelta(days=retention_days)
+            runs = pipeline.pipelineruns.filter(run_exitcode__isnull=False).order_by(
+                "-created_date"
+            )
+            runs_to_keep = runs[:minimum_number_of_runs_to_keep]
+            runs_to_delete = runs.filter(created_date__lt=cutoff).exclude(
+                run_id__in=runs_to_keep.values("run_id")
+            )
+
+            if runs_to_delete.exists():
+                pipeline_count += 1
+
+            for run in runs_to_delete.iterator():
+                run.delete()
+                deleted_run_count += 1
+
+        self.log(
+            f"Successfully removed {deleted_run_count:,d} runs from {pipeline_count} pipelines that were older than {retention_days} days."
+        )

vulnerabilities/schedules.py

@@ -91,8 +91,9 @@ def update_pipeline_schedule():
     from vulnerabilities.improvers import IMPROVERS_REGISTRY
     from vulnerabilities.models import PipelineSchedule
     from vulnerabilities.pipelines.exporters import EXPORTERS_REGISTRY
+    from vulnerabilities.pipelines.management import MANAGEMENT_REGISTRY
 
-    pipelines = IMPORTERS_REGISTRY | IMPROVERS_REGISTRY | EXPORTERS_REGISTRY
+    pipelines = IMPORTERS_REGISTRY | IMPROVERS_REGISTRY | EXPORTERS_REGISTRY | MANAGEMENT_REGISTRY
 
     PipelineSchedule.objects.exclude(pipeline_id__in=pipelines.keys()).delete()
     for id, pipeline_class in pipelines.items():
@@ -130,7 +131,7 @@ def mark_stale_runs():
     stale_jobs_count = stale_jobs.count()
 
     for job in stale_jobs.iterator(chunk_size=1000):
-        job.set_run_staled()
+        job.stop_run()
 
     log.info(f"Marked {stale_jobs_count} unfinished jobs as stale.")
 
@@ -150,7 +151,7 @@ def requeue_missing_jobs():
                 enqueue_run(run=job)
                 missing_jobs_count += 1
             else:
-                job.set_run_staled()
+                job.stop_run()
 
     log.info(f"Requeued {missing_jobs_count} missing jobs.")
 

vulnerabilities/tasks.py

@@ -178,7 +178,7 @@ def compute_queue_load_factor():
     """
     field = models.PipelineSchedule._meta.get_field("run_priority")
     label_to_value = {label: value for value, label in field.choices}
-    total_compute_seconds_per_queue = {}
+    total_compute_seconds_per_queue_in_24hr_cycle = {}
     worker_per_queue = {}
     load_per_queue = {}
     seconds_in_24_hr = 86400
@@ -191,26 +191,39 @@ def compute_queue_load_factor():
         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 * 60)))
-            for p in models.PipelineSchedule.objects.filter(
-                is_active=True, run_priority=label_to_value[queue]
-            )
-            if p.latest_successful_run
+        total_compute_seconds = 0
+        active_schedules = models.PipelineSchedule.objects.filter(
+            is_active=True,
+            run_priority=label_to_value[queue],
         )
+        for schedule in active_schedules:
+            if not schedule.latest_successful_run:
+                continue
+
+            runs_per_day = (24 * 60) / schedule.run_interval
+            compute_seconds_per_day = schedule.latest_successful_run.runtime * runs_per_day
+            total_compute_seconds += compute_seconds_per_day
+
+        total_compute_seconds_per_queue_in_24hr_cycle[queue] = total_compute_seconds
+
         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)
+        total_compute = total_compute_seconds_per_queue_in_24hr_cycle.get(queue_name, 0)
+        total_pipeline_for_queue = models.PipelineSchedule.objects.filter(
+            is_active=True, run_priority=label_to_value[queue_name]
+        ).count()
         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)
+        addition_worker_needed = min(
+            max(num_of_worker_for_balanced_queue - worker_count, 0), total_pipeline_for_queue
+        )
 
         if worker_count > 0:
             net_load_on_queue = unit_load_on_queue / worker_count

vulnerabilities/templates/pipeline_run_list.html

@@ -41,14 +41,22 @@ <h1>{{ pipeline_name }} Runs</h1>
     </div>
 
     {% if pipeline_description %}
-      <div class="notification is-info is-light">
-        <p class="is-size-5 has-text-grey-dark has-text-centered">
-          {{ pipeline_description }}
-        </p>
-      </div>
+<div class="notification is-info is-light">
+    <div class="is-size-5 has-text-grey-dark">
+        {{ pipeline_description|linebreaks }}
+    </div>
+</div>
     {% endif %}
 
     <div class="box">
+      <div class="notification  has-text-centered" style="background-color: #f8edb2;">
+          <p class="is-size-6 has-text-grey-dark">
+              <i class="fa fa-exclamation-triangle mr-1"></i>
+              Pipeline runs older than {{ pipeline_run_retention_days }} days are automatically removed.
+              At least the latest {{ minimum_pipeline_runs_retain }} pipeline runs are always retained.
+          </p>
+      </div>
+
       <table class="table is-striped  is-hoverable is-fullwidth">
         <thead>
           <tr>

vulnerabilities/tests/pipelines/management/test_cleanup_pipeline_runs.py

@@ -0,0 +1,60 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+
+from datetime import timedelta
+
+from django.test import TestCase
+from django.utils import timezone
+
+from vulnerabilities import models
+from vulnerabilities.pipelines.management.cleanup_pipeline_runs import CleanupPipelineRuns
+from vulnerabilities.tests.pipelines import TestLogger
+
+
+class TestCleanupPipelineRuns(TestCase):
+    def setUp(self):
+        self.logger = TestLogger()
+
+        self.schedule1 = models.PipelineSchedule.objects.create(pipeline_id="test_pipeline")
+        for _ in range(70):
+            models.PipelineRun.objects.create(
+                pipeline=self.schedule1,
+                run_exitcode=0,
+            )
+
+    def test_pipelines_management_cleanup_pipeline_runs_with_old_and_new_runs(self):
+        cutoff = timezone.now() - timedelta(days=100)
+        old_run_ids = models.PipelineRun.objects.all()[:60].values("run_id")
+        models.PipelineRun.objects.filter(run_id__in=old_run_ids).update(created_date=cutoff)
+
+        self.assertEqual(models.PipelineRun.objects.count(), 70)
+        pipeline = CleanupPipelineRuns()
+        pipeline.log = self.logger.write
+        exit_code, _ = pipeline.execute()
+
+        self.assertEqual(exit_code, 0)
+        self.assertEqual(models.PipelineRun.objects.count(), 60)
+        self.assertIn(
+            "Successfully removed 10 runs from 1 pipelines that were older than 60 days",
+            self.logger.getvalue(),
+        )
+
+    def test_pipelines_management_cleanup_pipeline_runs_with_new_runs_only(self):
+        self.assertEqual(models.PipelineRun.objects.count(), 70)
+        pipeline = CleanupPipelineRuns()
+        pipeline.log = self.logger.write
+        exit_code, _ = pipeline.execute()
+
+        self.assertEqual(exit_code, 0)
+        self.assertEqual(models.PipelineRun.objects.count(), 70)
+        self.assertIn(
+            "Successfully removed 0 runs from 0 pipelines that were older than 60 days",
+            self.logger.getvalue(),
+        )

vulnerabilities/views.py

@@ -66,6 +66,8 @@
 from vulnerabilities.utils import safe_altcha_redirect
 from vulnerablecode import __version__ as VULNERABLECODE_VERSION
 from vulnerablecode.settings import VULNERABLECODE_ALTCHA_SESSION_TIMEOUT
+from vulnerablecode.settings import VULNERABLECODE_MINIMUM_PIPELINE_RUNS_TO_RETAIN
+from vulnerablecode.settings import VULNERABLECODE_PIPELINE_RUN_RETENTION_DAYS
 from vulnerablecode.settings import env
 
 PAGE_SIZE = 10
@@ -1048,6 +1050,8 @@ def get_context_data(self, **kwargs):
         )
         context["pipeline_name"] = pipeline.pipeline_class.__name__
         context["pipeline_description"] = pipeline.description
+        context["pipeline_run_retention_days"] = VULNERABLECODE_PIPELINE_RUN_RETENTION_DAYS
+        context["minimum_pipeline_runs_retain"] = VULNERABLECODE_MINIMUM_PIPELINE_RUNS_TO_RETAIN
         return context
 
 

vulnerablecode/settings.py

@@ -44,6 +44,16 @@
 
 VULNERABLECODE_ALTCHA_SESSION_TIMEOUT = env.int("VULNERABLECODE_ALTCHA_SESSION_TIMEOUT", None)
 
+# Retain pipeline runs for the specified number of days, cannot be less than 10 days.
+VULNERABLECODE_PIPELINE_RUN_RETENTION_DAYS = max(
+    env.int("VULNERABLECODE_PIPELINE_RUN_RETENTION_DAYS", 60), 10
+)
+
+# Minimum number of pipeline runs to retain regardless of the retention period, cannot be less than 15.
+VULNERABLECODE_MINIMUM_PIPELINE_RUNS_TO_RETAIN = max(
+    env.int("VULNERABLECODE_MINIMUM_PIPELINE_RUNS_TO_RETAIN", 60), 15
+)
+
 # SECURITY WARNING: do not run with debug turned on in production
 DEBUG = env.bool("VULNERABLECODE_DEBUG", default=False)