Merge pull request #1871 from aboutcode-org/1783-task-queue

Support running pipelines in scheduled task queue

Author: keshav-space

.VERSION

@@ -1,3 +1,3 @@
 refs=$Format:%D$
-commit=$Format:%H$
+commit=$Format:%h$
 abbrev_commit=$Format:%H$

.dockerignore

@@ -6,7 +6,6 @@ docker-compose.yml
 
 
 # Ignore Git directory and files and github directory.
-**/.git
 **/.gitignore
 **/.gitattributes
 **/.gitmodules

.gitattributes

@@ -0,0 +1 @@
+.VERSION export-subst

Dockerfile

@@ -17,8 +17,21 @@ ENV PYTHONDONTWRITEBYTECODE 1
 
 RUN mkdir -p /var/vulnerablecode/static
 
+RUN apt-get update \
+ && apt-get install -y --no-install-recommends \
+       wait-for-it \
+ && apt-get clean \
+ && rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/*
+
 # Keep the dependencies installation before the COPY of the app/ for proper caching
 COPY setup.cfg setup.py requirements.txt pyproject.toml /app/
 RUN pip install . -c requirements.txt
 
 COPY . /app
+
+# Store commit hash for docker deployment from local checkout.
+RUN if [ -d ".git" ]; then \
+  GIT_COMMIT=$(git rev-parse --short HEAD) && \
+  echo "VULNERABLECODE_GIT_COMMIT=\"$GIT_COMMIT\"" >> /app/vulnerablecode/settings.py; \
+  rm -rf .git; \
+fi

docker-compose.yml

@@ -10,6 +10,15 @@ services:
       - db_data:/var/lib/postgresql/data/
       - ./etc/postgresql/postgresql.conf:/etc/postgresql/postgresql.conf
 
+  vulnerablecode_redis:
+    image: redis
+    # Enable redis data persistence using the "Append Only File" with the
+    # default policy of fsync every second. See https://redis.io/topics/persistence
+    command: redis-server --appendonly yes
+    volumes:
+      - vulnerablecode_redis_data:/data
+    restart: always
+
   vulnerablecode:
     build: .
     command: /bin/sh -c "
@@ -26,6 +35,31 @@ services:
     depends_on:
       - db
 
+  vulnerablecode_scheduler:
+    build: .
+    command: wait-for-it web:8000 -- python ./manage.py run_scheduler
+    env_file:
+      - docker.env
+    volumes:
+      - /etc/vulnerablecode/:/etc/vulnerablecode/
+    depends_on:
+      - vulnerablecode_redis
+      - db
+      - vulnerablecode
+
+  vulnerablecode_rqworker:
+    build: .
+    command: wait-for-it web:8000 -- python ./manage.py rqworker default
+    env_file:
+      - docker.env
+    volumes:
+      - /etc/vulnerablecode/:/etc/vulnerablecode/
+    depends_on:
+      - vulnerablecode_redis
+      - db
+      - vulnerablecode
+
+
   nginx:
     image: nginx
     ports:
@@ -44,4 +78,5 @@ services:
 volumes:
   db_data:
   static:
+  vulnerablecode_redis_data:
 

docker.env

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

docs/source/conf.py

@@ -38,6 +38,8 @@
     "https://example.org/api/non-existent-packages",
     "https://github.com/aboutcode-org/vulnerablecode/pull/495/commits",
     "https://nvd.nist.gov/products/cpe",
+    "https://ftp.suse.com/pub/projects/security/yaml/suse-cvss-scores.yaml",
+    "http://ftp.suse.com/pub/projects/security/yaml/",
 ]
 
 # Add any Sphinx extension module names here, as strings. They can be

requirements.txt

@@ -92,8 +92,10 @@ python-dateutil==2.8.2
 python-dotenv==0.20.0
 pytz==2022.1
 PyYAML==6.0.1
+redis==5.0.1
 requests==2.32.0
 restructuredtext-lint==1.4.0
+rq==1.15.1
 saneyaml==0.6.0
 semantic-version==2.9.0
 six==1.16.0

setup.cfg

@@ -94,6 +94,8 @@ install_requires =
 
     #pipeline
     aboutcode.pipeline>=0.1.0
+    django-rq==2.10.1
+    rq-scheduler==0.13.1
 
     #vulntotal
     python-dotenv

vulnerabilities/api_v2.py

@@ -14,15 +14,20 @@
 from drf_spectacular.utils import extend_schema
 from drf_spectacular.utils import extend_schema_view
 from packageurl import PackageURL
+from rest_framework import mixins
 from rest_framework import serializers
 from rest_framework import status
 from rest_framework import viewsets
+from rest_framework.authentication import SessionAuthentication
 from rest_framework.decorators import action
+from rest_framework.permissions import BasePermission
 from rest_framework.response import Response
 from rest_framework.reverse import reverse
 
 from vulnerabilities.models import CodeFix
 from vulnerabilities.models import Package
+from vulnerabilities.models import PipelineRun
+from vulnerabilities.models import PipelineSchedule
 from vulnerabilities.models import Vulnerability
 from vulnerabilities.models import VulnerabilityReference
 from vulnerabilities.models import VulnerabilitySeverity
@@ -606,3 +611,146 @@ def get_queryset(self):
                 affected_package_vulnerability__vulnerability__vulnerability_id=vulnerability_id
             )
         return queryset
+
+
+class CreateListRetrieveUpdateViewSet(
+    mixins.CreateModelMixin,
+    mixins.ListModelMixin,
+    mixins.RetrieveModelMixin,
+    mixins.UpdateModelMixin,
+    viewsets.GenericViewSet,
+):
+    """
+    A viewset that provides `create`, `list, `retrieve`, and `update` actions.
+    To use it, override the class and set the `.queryset` and
+    `.serializer_class` attributes.
+    """
+
+    pass
+
+
+class IsAdminWithSessionAuth(BasePermission):
+    """Permit only staff users authenticated via session (not token)."""
+
+    def has_permission(self, request, view):
+        is_authenticated = request.user and request.user.is_authenticated
+        is_staff = request.user and request.user.is_staff
+        is_session_auth = isinstance(request.successful_authenticator, SessionAuthentication)
+
+        return is_authenticated and is_staff and is_session_auth
+
+
+class PipelineRunAPISerializer(serializers.HyperlinkedModelSerializer):
+    status = serializers.SerializerMethodField()
+    execution_time = serializers.SerializerMethodField()
+    log = serializers.SerializerMethodField()
+
+    class Meta:
+        model = PipelineRun
+        fields = [
+            "run_id",
+            "status",
+            "execution_time",
+            "run_start_date",
+            "run_end_date",
+            "run_exitcode",
+            "run_output",
+            "created_date",
+            "vulnerablecode_version",
+            "vulnerablecode_commit",
+            "log",
+        ]
+
+    def get_status(self, run):
+        return run.status
+
+    def get_execution_time(self, run):
+        if run.execution_time:
+            return round(run.execution_time, 2)
+
+    def get_log(self, run):
+        """Return only last 5000 character of log."""
+        return run.log[-5000:]
+
+
+class PipelineScheduleAPISerializer(serializers.HyperlinkedModelSerializer):
+    url = serializers.HyperlinkedIdentityField(
+        view_name="schedule-detail", lookup_field="pipeline_id"
+    )
+    latest_run = serializers.SerializerMethodField()
+    next_run_date = serializers.SerializerMethodField()
+
+    class Meta:
+        model = PipelineSchedule
+        fields = [
+            "url",
+            "pipeline_id",
+            "is_active",
+            "live_logging",
+            "run_interval",
+            "execution_timeout",
+            "created_date",
+            "schedule_work_id",
+            "next_run_date",
+            "latest_run",
+        ]
+
+    def get_next_run_date(self, schedule):
+        return schedule.next_run_date
+
+    def get_latest_run(self, schedule):
+        if latest := schedule.pipelineruns.first():
+            return PipelineRunAPISerializer(latest).data
+        return None
+
+
+class PipelineScheduleCreateSerializer(serializers.ModelSerializer):
+    class Meta:
+        model = PipelineSchedule
+        fields = [
+            "pipeline_id",
+            "is_active",
+            "run_interval",
+            "live_logging",
+            "execution_timeout",
+        ]
+        extra_kwargs = {
+            field: {"initial": PipelineSchedule._meta.get_field(field).get_default()}
+            for field in [
+                "is_active",
+                "run_interval",
+                "live_logging",
+                "execution_timeout",
+            ]
+        }
+
+
+class PipelineScheduleUpdateSerializer(serializers.ModelSerializer):
+    class Meta:
+        model = PipelineSchedule
+        fields = [
+            "is_active",
+            "run_interval",
+            "live_logging",
+            "execution_timeout",
+        ]
+
+
+class PipelineScheduleV2ViewSet(CreateListRetrieveUpdateViewSet):
+    queryset = PipelineSchedule.objects.prefetch_related("pipelineruns").all()
+    serializer_class = PipelineScheduleAPISerializer
+    lookup_field = "pipeline_id"
+    lookup_value_regex = r"[\w.]+"
+
+    def get_serializer_class(self):
+        if self.action == "create":
+            return PipelineScheduleCreateSerializer
+        elif self.action == "update":
+            return PipelineScheduleUpdateSerializer
+        return super().get_serializer_class()
+
+    def get_permissions(self):
+        """Restrict addition and modifications to staff users authenticated via session."""
+        if self.action not in ["list", "retrieve"]:
+            return [IsAdminWithSessionAuth()]
+        return super().get_permissions()