From 05478cfde73d7bd14c49b3993bbb98a95256642c Mon Sep 17 00:00:00 2001 From: Keshav Priyadarshi Date: Wed, 23 Sep 2026 14:44:40 +0530 Subject: [PATCH 1/6] feat: add config and job to clean up old pipeline runs Signed-off-by: Keshav Priyadarshi --- .../pipelines/management/__init__.py | 16 +++++ .../management/cleanup_pipeline_runs.py | 64 +++++++++++++++++++ vulnerabilities/schedules.py | 3 +- vulnerablecode/settings.py | 10 +++ 4 files changed, 92 insertions(+), 1 deletion(-) create mode 100644 vulnerabilities/pipelines/management/__init__.py create mode 100644 vulnerabilities/pipelines/management/cleanup_pipeline_runs.py diff --git a/vulnerabilities/pipelines/management/__init__.py b/vulnerabilities/pipelines/management/__init__.py new file mode 100644 index 000000000..b549c5827 --- /dev/null +++ b/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, + ] +) diff --git a/vulnerabilities/pipelines/management/cleanup_pipeline_runs.py b/vulnerabilities/pipelines/management/cleanup_pipeline_runs.py new file mode 100644 index 000000000..4045cc0bd --- /dev/null +++ b/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." + ) diff --git a/vulnerabilities/schedules.py b/vulnerabilities/schedules.py index 038e9a927..4cb0a7fa3 100644 --- a/vulnerabilities/schedules.py +++ b/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(): diff --git a/vulnerablecode/settings.py b/vulnerablecode/settings.py index ba760beee..965b15af8 100644 --- a/vulnerablecode/settings.py +++ b/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) From e74d2c743e24cbf31323f6d557c0bf8f3e08ce04 Mon Sep 17 00:00:00 2001 From: Keshav Priyadarshi Date: Wed, 23 Sep 2026 14:47:17 +0530 Subject: [PATCH 2/6] feat: remove job from redis before deleting the run Signed-off-by: Keshav Priyadarshi --- vulnerabilities/models.py | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/vulnerabilities/models.py b/vulnerabilities/models.py index 6b115b1e2..e35a5a34b 100644 --- a/vulnerabilities/models.py +++ b/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(), @@ -2219,13 +2220,10 @@ def stop_run(self): ) self.set_run_stopped() - def delete_run(self, delete_self=True): + def delete_run(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. @@ -2233,6 +2231,8 @@ def delete(self, *args, **kwargs): with suppress(redis.exceptions.ConnectionError, AttributeError): self.stop_run() + self.delete_run() + return super().delete(*args, **kwargs) def append_to_log(self, message, is_multiline=False): @@ -2366,16 +2366,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): From a7386471dc5ce5db1024533ac5c56f393268dc98 Mon Sep 17 00:00:00 2001 From: Keshav Priyadarshi Date: Wed, 23 Sep 2026 14:49:35 +0530 Subject: [PATCH 3/6] test: add coverage for the run cleanup job Signed-off-by: Keshav Priyadarshi --- .../management/test_cleanup_pipeline_runs.py | 60 +++++++++++++++++++ 1 file changed, 60 insertions(+) create mode 100644 vulnerabilities/tests/pipelines/management/test_cleanup_pipeline_runs.py diff --git a/vulnerabilities/tests/pipelines/management/test_cleanup_pipeline_runs.py b/vulnerabilities/tests/pipelines/management/test_cleanup_pipeline_runs.py new file mode 100644 index 000000000..19eefe00f --- /dev/null +++ b/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(), + ) From aa53acf312bfd800ea59715b8a5e23d5aa233fdb Mon Sep 17 00:00:00 2001 From: Keshav Priyadarshi Date: Wed, 23 Sep 2026 14:51:01 +0530 Subject: [PATCH 4/6] ui: display pipeline run retention period and minimum runs retained on dashboard Signed-off-by: Keshav Priyadarshi --- .../templates/pipeline_run_list.html | 18 +++++++++++++----- vulnerabilities/views.py | 4 ++++ 2 files changed, 17 insertions(+), 5 deletions(-) diff --git a/vulnerabilities/templates/pipeline_run_list.html b/vulnerabilities/templates/pipeline_run_list.html index c9e109b64..72e6e767b 100644 --- a/vulnerabilities/templates/pipeline_run_list.html +++ b/vulnerabilities/templates/pipeline_run_list.html @@ -41,14 +41,22 @@

{{ pipeline_name }} Runs

{% if pipeline_description %} -
-

- {{ pipeline_description }} -

-
+
+
+ {{ pipeline_description|linebreaks }} +
+
{% endif %}
+
+

+ + 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. +

+
+ diff --git a/vulnerabilities/views.py b/vulnerabilities/views.py index 16d183993..b5ba74214 100644 --- a/vulnerabilities/views.py +++ b/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 From 23e8bd9a3e617c148d1d2e11ae6efe5464b7dcbe Mon Sep 17 00:00:00 2001 From: Keshav Priyadarshi Date: Wed, 23 Sep 2026 18:01:11 +0530 Subject: [PATCH 5/6] fix: remove stale runs from redis jobs Signed-off-by: Keshav Priyadarshi --- vulnerabilities/models.py | 5 +++-- vulnerabilities/schedules.py | 4 ++-- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/vulnerabilities/models.py b/vulnerabilities/models.py index e35a5a34b..38afa1511 100644 --- a/vulnerabilities/models.py +++ b/vulnerabilities/models.py @@ -2219,8 +2219,9 @@ def stop_run(self): job_id=str(self.run_id), ) self.set_run_stopped() + self.delete_run_job() - def delete_run(self): + def delete_run_job(self): if job := self.job: job.delete() @@ -2231,7 +2232,7 @@ def delete(self, *args, **kwargs): with suppress(redis.exceptions.ConnectionError, AttributeError): self.stop_run() - self.delete_run() + self.delete_run_job() return super().delete(*args, **kwargs) diff --git a/vulnerabilities/schedules.py b/vulnerabilities/schedules.py index 4cb0a7fa3..2e684167b 100644 --- a/vulnerabilities/schedules.py +++ b/vulnerabilities/schedules.py @@ -131,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.") @@ -151,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.") From 991377530491a8713a5ed480e4d6e50328f5487a Mon Sep 17 00:00:00 2001 From: Keshav Priyadarshi Date: Wed, 23 Sep 2026 20:23:23 +0530 Subject: [PATCH 6/6] fix: limit recommended workers to the number of pipelines Signed-off-by: Keshav Priyadarshi --- vulnerabilities/tasks.py | 31 ++++++++++++++++++++++--------- 1 file changed, 22 insertions(+), 9 deletions(-) diff --git a/vulnerabilities/tasks.py b/vulnerabilities/tasks.py index 0188511ed..d4670040f 100644 --- a/vulnerabilities/tasks.py +++ b/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