Address PR Review Comments #1953

Signed-off-by: Michael Ehab Mikhail <michael.ehab@hotmail.com>

Author: michaelehab

docker-compose.yml

@@ -23,9 +23,9 @@ services:
   vulnerablecode:
     build: .
     command: /bin/sh -c "
-        ./manage.py migrate &&
-        ./manage.py collectstatic --no-input --verbosity 0 --clear &&
-        gunicorn vulnerablecode.wsgi:application -u nobody -g nogroup --bind :8000 --timeout 600 --workers 8"
+      ./manage.py migrate &&
+      ./manage.py collectstatic --no-input --verbosity 0 --clear &&
+      gunicorn vulnerablecode.wsgi:application -u nobody -g nogroup --bind :8000 --timeout 600 --workers 8"
     env_file:
       - docker.env
     expose:
@@ -60,6 +60,17 @@ services:
       - db
       - vulnerablecode
 
+  vulnerablecode_rqworker_live:
+    build: .
+    command: wait-for-it web:8000 -- python ./manage.py rqworker live
+    env_file:
+      - docker.env
+    volumes:
+      - /etc/vulnerablecode/:/etc/vulnerablecode/
+    depends_on:
+      - vulnerablecode_redis
+      - db
+      - vulnerablecode
 
   nginx:
     image: nginx
@@ -75,9 +86,7 @@ services:
     depends_on:
       - vulnerablecode
 
-
 volumes:
   db_data:
   static:
   vulnerablecode_redis_data:
-

docker.env

@@ -5,4 +5,6 @@ POSTGRES_PASSWORD=vulnerablecode
 VULNERABLECODE_DB_HOST=db
 VULNERABLECODE_STATIC_ROOT=/var/vulnerablecode/static/
 
-VULNERABLECODE_REDIS_HOST=vulnerablecode_redis
+VULNERABLECODE_REDIS_HOST=vulnerablecode_redis
+
+VULNERABLECODE_ENABLE_LIVE_EVALUATION_API=false

vulnerabilities/api_v2.py

@@ -1304,6 +1304,7 @@ class LiveEvaluationSerializer(serializers.Serializer):
 
 class LiveEvaluationViewSet(viewsets.GenericViewSet):
     serializer_class = LiveEvaluationSerializer
+    throttle_classes = [AnonRateThrottle, PermissionBasedUserRateThrottle]
 
     @extend_schema(
         request=LiveEvaluationSerializer,
@@ -1346,34 +1347,15 @@ def evaluate(self, request):
                 status=status.HTTP_400_BAD_REQUEST,
             )
 
-        # Create a single LivePipelineRun to represent this evaluation
-        from vulnerabilities.models import LivePipelineRun
-
-        live_run = LivePipelineRun.objects.create(purl=purl_string)
-        runs = []
-        for importer in importers:
-            importer_name = getattr(importer, "pipeline_id", importer.__name__)
-            run_id = enqueue_ad_hoc_pipeline(importer_name, inputs={"purl": purl})
-            # Attach each PipelineRun to the LivePipelineRun
-            from vulnerabilities.models import PipelineRun
-
-            try:
-                run_obj = PipelineRun.objects.get(run_id=run_id)
-                run_obj.live_pipeline = live_run
-                run_obj.save()
-            except PipelineRun.DoesNotExist:
-                pass
-            runs.append(
-                {
-                    "importer": importer_name,
-                    "run_id": str(run_id) if run_id else None,
-                }
-            )
+        # Enqueue all selected importers together and link runs to a new LivePipelineRun
+        importer_ids = [getattr(imp, "pipeline_id", imp.__name__) for imp in importers]
+        live_run_id, run_ids = enqueue_ad_hoc_pipeline(importer_ids, inputs={"purl": purl})
+        runs = [
+            {"importer": importer_ids[idx], "run_id": str(rid)} for idx, rid in enumerate(run_ids)
+        ]
 
         request_obj = request
-        status_path = reverse(
-            "live-evaluation-status", kwargs={"live_run_id": str(live_run.run_id)}
-        )
+        status_path = reverse("live-evaluation-status", kwargs={"live_run_id": str(live_run_id)})
 
         if hasattr(request_obj, "build_absolute_uri"):
             status_url = request_obj.build_absolute_uri(status_path)
@@ -1382,7 +1364,7 @@ def evaluate(self, request):
 
         return Response(
             {
-                "live_run_id": str(live_run.run_id),
+                "live_run_id": str(live_run_id),
                 "runs": runs,
                 "status_url": status_url,
             },
@@ -1431,6 +1413,7 @@ def status(self, request, live_run_id=None):
             "live_run_id": str(live_run.run_id),
             "overall_status": live_run.status,
             "created_date": live_run.created_date,
+            "started_date": getattr(live_run, "started_date", None),
             "completed_date": live_run.completed_date,
             "purl": live_run.purl,
             "importers": importer_statuses,

vulnerabilities/models.py

@@ -1985,17 +1985,37 @@ def is_finished(self):
         return self.status == "finished"
 
     def update_status(self):
-        if not self.pipelineruns.exists():
+        runs = list(self.pipelineruns.all())
+        if not runs:
             self.status = "queued"
-        elif all(run.status == PipelineRun.Status.SUCCESS for run in self.pipelineruns.all()):
+            self.completed_date = None
+            self.save(update_fields=["status", "completed_date"])
+            return
+
+        # Determine aggregate status
+        if all(r.status == PipelineRun.Status.SUCCESS for r in runs):
             self.status = "finished"
-            self.completed_date = timezone.now()
-        elif any(run.status == PipelineRun.Status.FAILURE for run in self.pipelineruns.all()):
+        elif any(r.status == PipelineRun.Status.FAILURE for r in runs):
             self.status = "failed"
-            self.completed_date = timezone.now()
-        else:
+        elif any(r.status == PipelineRun.Status.RUNNING for r in runs):
             self.status = "running"
-        self.save()
+        else:
+            # queued or mixed queued
+            self.status = "queued"
+
+        end_times = [r.run_end_date for r in runs if r.run_end_date]
+        completed = None
+        if self.status in ("finished", "failed") and end_times:
+            completed = max(end_times)
+        self.completed_date = completed
+        self.save(update_fields=["status", "completed_date"])
+
+    @property
+    def started_date(self):
+        """Return earliest run_start_date among child runs, if any."""
+        runs = self.pipelineruns.all()
+        start_times = [r.run_start_date for r in runs if r.run_start_date]
+        return min(start_times) if start_times else None
 
     class Meta:
         ordering = ["-created_date"]
@@ -2281,10 +2301,8 @@ def append_to_log(self, message, is_multiline=False):
         message = message.strip()
         if not is_multiline:
             message = message.replace("\n", "").replace("\r", "")
-
-        new_log = (self.log or "") + message + "\n"
-        type(self).objects.filter(run_id=self.run_id).update(log=new_log)
-        self.log = new_log
+        self.log = (self.log or "") + message + "\n"
+        self.save(update_fields=["log"])
 
     def dequeue(self):
         from vulnerabilities.tasks import dequeue_job

vulnerabilities/tasks.py

@@ -22,6 +22,7 @@
 logger = logging.getLogger(__name__)
 
 default_queue = django_rq.get_queue("default")
+live_queue = django_rq.get_queue("live")
 
 
 def execute_pipeline(pipeline_id, run_id, inputs=None):
@@ -132,35 +133,59 @@ def enqueue_pipeline(pipeline_id):
     )
 
 
-def enqueue_ad_hoc_pipeline(pipeline_id, *, inputs=None):
-    """Enqueue a one-off execution for the given pipeline_id with optional inputs.
+def enqueue_ad_hoc_pipeline(pipeline_ids, *, inputs=None):
+    """Enqueue one-off executions for the given pipeline_ids with optional inputs.
 
-    Returns the created run_id or None if the pipeline cannot be enqueued.
+    When multiple pipeline IDs are provided, this will create a single LivePipelineRun and attach
+    each created PipelineRun to it. Returns a tuple of (live_run_id, run_ids).
+
+    If a single pipeline ID (str) is provided, it will be wrapped into a list.
     """
+    inputs = inputs or {}
+    # Normalize to list
+    if isinstance(pipeline_ids, str):
+        pipeline_ids = [pipeline_ids]
+
+    # Create a LivePipelineRun to group these ad-hoc runs, if any inputs (such as purl) are given
+    purl_val = inputs.get("purl")
     try:
-        pipeline_schedule = models.PipelineSchedule.objects.get(pipeline_id=pipeline_id)
-    except models.PipelineSchedule.DoesNotExist:
-        pipeline_schedule = models.PipelineSchedule.objects.create(
-            pipeline_id=pipeline_id,
-            is_active=False,
-        )
+        # accept PackageURL instance as well as string
+        purl_str = str(purl_val) if purl_val is not None else None
+    except Exception:
+        purl_str = None
+
+    live_run = models.LivePipelineRun.objects.create(purl=purl_str)
+
+    run_ids = []
+    for pipeline_id in pipeline_ids:
+        try:
+            pipeline_schedule = models.PipelineSchedule.objects.get(pipeline_id=pipeline_id)
+        except models.PipelineSchedule.DoesNotExist:
+            pipeline_schedule = models.PipelineSchedule.objects.create(
+                pipeline_id=pipeline_id,
+                is_active=False,
+            )
 
-    run = models.PipelineRun.objects.create(pipeline=pipeline_schedule)
+        run = models.PipelineRun.objects.create(pipeline=pipeline_schedule, live_pipeline=live_run)
+
+        # Enqueue on the dedicated live queue
+        live_queue.enqueue(
+            execute_pipeline,
+            pipeline_id,
+            run.run_id,
+            inputs,
+            job_id=str(run.run_id),
+            on_failure=set_run_failure,
+            job_timeout=f"{pipeline_schedule.execution_timeout}h",
+        )
+        run_ids.append(run.run_id)
 
-    live_queue = django_rq.get_queue("live")
-    job = live_queue.enqueue(
-        execute_pipeline,
-        pipeline_id,
-        run.run_id,
-        inputs or {},
-        job_id=str(run.run_id),
-        on_failure=set_run_failure,
-        job_timeout=f"{pipeline_schedule.execution_timeout}h",
-    )
-    return run.run_id
+    return live_run.run_id, run_ids
 
 
 def dequeue_job(job_id):
     """Remove a job from queue if it hasn't been executed yet."""
     if job_id in default_queue.jobs:
         default_queue.remove(job_id)
+    if job_id in live_queue.jobs:
+        live_queue.remove(job_id)

vulnerabilities/tests/test_api_v2.py

@@ -913,27 +913,15 @@ def setUp(self):
 
     @patch("vulnerabilities.api_v2.LIVE_IMPORTERS_REGISTRY")
     @patch("vulnerabilities.api_v2.enqueue_ad_hoc_pipeline")
-    @patch("vulnerabilities.models.PipelineRun.objects.get")
-    @patch("vulnerabilities.models.LivePipelineRun.objects.create")
     @patch("django.urls.reverse")
-    def test_evaluate_success(
-        self, mock_reverse, mock_live_create, mock_pipeline_get, mock_enqueue, mock_registry
-    ):
+    def test_evaluate_success(self, mock_reverse, mock_enqueue, mock_registry):
         class MockImporter:
             pipeline_id = "pypa_live_importer_v2"
             supported_types = ["pypi"]
 
         mock_registry.values.return_value = [MockImporter]
         valid_uuid = "00000000-0000-0000-0000-000000000001"
-        mock_live_run = type("MockLiveRun", (), {"run_id": valid_uuid})()
-        mock_live_create.return_value = mock_live_run
-        mock_enqueue.return_value = "mock-run-id"
-        mock_pipeline_run = type(
-            "MockPipelineRun",
-            (),
-            {"run_id": "mock-run-id", "live_pipeline": None, "save": lambda self: None},
-        )()
-        mock_pipeline_get.return_value = mock_pipeline_run
+        mock_enqueue.return_value = (valid_uuid, ["mock-run-id"])
         mock_reverse.return_value = f"/api/v2/live-evaluation/status/{valid_uuid}"
 
         data = {"purl": "pkg:pypi/django@3.2"}
@@ -942,7 +930,7 @@ class MockImporter:
         assert isinstance(response.data, dict)
         assert response.data["live_run_id"] is not None
         assert response.data["runs"][0]["importer"] == "pypa_live_importer_v2"
-        assert response.data["runs"][0]["run_id"] is not None
+        assert response.data["runs"][0]["run_id"] == "mock-run-id"
         assert "status_url" in response.data
         assert response.data["status_url"].endswith(f"/api/v2/live-evaluation/status/{valid_uuid}")
 

vulnerabilities/views.py

@@ -697,6 +697,15 @@ def get_context_data(self, **kwargs):
         context["disabled_pipeline_count"] = PipelineSchedule.objects.filter(
             pipeline_id__in=live_pipeline_ids, is_active=False
         ).count()
+
+        # Update status of recent LivePipelineRun groups for freshness
+        try:
+            from vulnerabilities.models import LivePipelineRun
+
+            for lpr in LivePipelineRun.objects.order_by("-created_date")[:50]:
+                lpr.update_status()
+        except Exception:
+            pass
         return context
 
 

vulnerablecode/settings.py

@@ -181,6 +181,10 @@
     "VULNERABLECODEIO_REQUIRE_AUTHENTICATION", default=False
 )
 
+VULNERABLECODE_ENABLE_LIVE_EVALUATION_API = env.bool(
+    "VULNERABLECODE_ENABLE_LIVE_EVALUATION_API", default=False
+)
+
 LOGIN_REDIRECT_URL = "/"
 LOGOUT_REDIRECT_URL = "/"
 
@@ -383,12 +387,11 @@
         "PORT": env.str("VULNERABLECODE_REDIS_PORT", default="6379"),
         "PASSWORD": env.str("VULNERABLECODE_REDIS_PASSWORD", default=""),
         "DEFAULT_TIMEOUT": env.int("VULNERABLECODE_REDIS_DEFAULT_TIMEOUT", default=3600),
-    }
-}
-
-RQ_QUEUES["live"] = {
-    "HOST": env.str("VULNERABLECODE_REDIS_HOST", default=RQ_QUEUES["default"]["HOST"]),
-    "PORT": env.str("VULNERABLECODE_REDIS_PORT", default=RQ_QUEUES["default"]["PORT"]),
-    "PASSWORD": env.str("VULNERABLECODE_REDIS_PASSWORD", default=RQ_QUEUES["default"]["PASSWORD"]),
-    "DEFAULT_TIMEOUT": env.int("VULNERABLECODE_LIVE_REDIS_DEFAULT_TIMEOUT", default=3600),
+    },
+    "live": {
+        "HOST": env.str("VULNERABLECODE_REDIS_HOST", default="localhost"),
+        "PORT": env.str("VULNERABLECODE_REDIS_PORT", default="6379"),
+        "PASSWORD": env.str("VULNERABLECODE_REDIS_PASSWORD", default=""),
+        "DEFAULT_TIMEOUT": env.int("VULNERABLECODE_LIVE_REDIS_DEFAULT_TIMEOUT", default=3600),
+    },
 }

vulnerablecode/urls.py

@@ -46,6 +46,7 @@
 from vulnerabilities.views import VulnerabilitySearch
 from vulnerablecode.settings import DEBUG
 from vulnerablecode.settings import DEBUG_TOOLBAR
+from vulnerablecode.settings import VULNERABLECODE_ENABLE_LIVE_EVALUATION_API
 
 
 # See the comment at https://stackoverflow.com/a/46163870.
@@ -71,7 +72,9 @@ def __init__(self, *args, **kwargs):
 api_v2_router.register("codefixes", CodeFixViewSet, basename="codefix")
 api_v2_router.register("pipelines", PipelineScheduleV2ViewSet, basename="pipelines")
 api_v2_router.register("advisory-codefixes", CodeFixV2ViewSet, basename="advisory-codefix")
-api_v2_router.register("live-evaluation", LiveEvaluationViewSet, basename="live-evaluation")
+
+if VULNERABLECODE_ENABLE_LIVE_EVALUATION_API:
+    api_v2_router.register("live-evaluation", LiveEvaluationViewSet, basename="live-evaluation")
 
 
 urlpatterns = [