Add ability to stop a running pipeline task #176

Signed-off-by: Thomas Druez <tdruez@nexb.com>

scanpipe/models.py

@@ -48,6 +48,8 @@
 import django_rq
 from packageurl import normalize_qualifiers
 from packageurl.contrib.django.models import PackageURLQuerySetMixin
+from rq.command import send_stop_job_command
+from rq.job import Job
 
 from scancodeio import __version__ as scancodeio_version
 from scanpipe import tasks
@@ -110,32 +112,13 @@ class AbstractTaskFieldsModel(models.Model):
     class Meta:
         abstract = True
 
-    # @property
-    # def task_result(self):
-    #     return AsyncResult(str(self.task_id))
-    #
-    # def task_state(self):
-    #     """
-    #     Possible values include:
-    #     - UNKNOWN (PENDING)
-    #         No history about the task is available.
-    #     - STARTED
-    #         The task has been started.
-    #     - RETRY
-    #         The task is to be re-executed; possibly due to a failure.
-    #     - FAILURE
-    #         The task raised an exception or has exceeded the retry limit.
-    #         The result attribute would contain the exception raised by the task.
-    #     - SUCCESS
-    #         The task executed successfully. The result attribute would contain
-    #         the task's return value.
-    #
-    #     Notes: All tasks are PENDING by default in Celery, so it would make more
-    #     sense if the state was named "unknown". Celery doesn't update the state
-    #     when a task is sent, and any task with no history is assumed to be pending.
-    #     """
-    #     state = self.task_result.state
-    #     return "UNKNOWN" if state == "PENDING" else state
+    @staticmethod
+    def get_job(job_id):
+        return Job.fetch(job_id, connection=django_rq.get_connection())
+
+    @property
+    def job(self):
+        return self.get_job(str(self.task_id))
 
     @property
     def task_succeeded(self):
@@ -151,6 +134,13 @@ def task_staled(self):
         """
         return self.task_exitcode == 99
 
+    @property
+    def task_stopped(self):
+        """
+        Returns True if the task was stopped.
+        """
+        return self.task_exitcode == 88
+
     class Status(models.TextChoices):
         """
         List of Run status.
@@ -161,6 +151,7 @@ class Status(models.TextChoices):
         RUNNING = "running"
         SUCCESS = "success"
         FAILURE = "failure"
+        STOPPED = "stopped"
         STALE = "stale"
 
     @property
@@ -176,6 +167,9 @@ def status(self):
         elif self.task_staled:
             return status.STALE
 
+        elif self.task_stopped:
+            return status.STOPPED
+
         elif self.task_exitcode and self.task_exitcode > 0:
             return status.FAILURE
 
@@ -248,6 +242,15 @@ def set_task_staled(self):
         """
         self.set_task_ended(exitcode=99, output="")
 
+    def stop_task(self):
+        """
+        Stop a "running" task.
+        """
+        send_stop_job_command(
+            connection=django_rq.get_connection(), job_id=str(self.task_id)
+        )
+        self.set_task_ended(exitcode=88, output="Stopped")
+
 
 class ExtraDataFieldMixin(models.Model):
     """

scanpipe/templates/scanpipe/includes/run_modal_content.html

@@ -47,6 +47,12 @@
           </div>
         </div>
       {% endif %}
+      {% if run.status == run.Status.RUNNING %}
+        <a href="{% url 'project_stop_pipeline' run.project.uuid run.uuid %}" class="execute-pipeline-link has-text-danger">
+          <i class="fas fa-ban"></i>
+          Stop Pipeline
+        </a>
+      {% endif %}
     </div>
     <div class="field is-grouped is-grouped-multiline mb-4">
       <div class="control">

scanpipe/templates/scanpipe/includes/run_status_tag.html

@@ -7,7 +7,9 @@
 {% elif run.status == run.Status.QUEUED %}
   <span class="tag is-info">Queued <i class="fas fa-clock ml-1"></i></span>
 {% elif run.status == run.Status.STALE %}
-  <span class="tag is-dark">Stale <i class="fas fa-skull-crossbones ml-1"></i></span>
+  <span class="tag is-dark">Stale</span>
+{% elif run.status == run.Status.STOPPED %}
+  <span class="tag is-danger">Stopped</span>
 {% else %}
   <span class="tag is-light">Not started</span>
 {% endif %}

scanpipe/urls.py

@@ -85,6 +85,11 @@
         views.execute_pipeline_view,
         name="project_execute_pipeline",
     ),
+    path(
+        "project/<uuid:uuid>/stop_pipeline/<uuid:run_uuid>/",
+        views.stop_pipeline_view,
+        name="project_stop_pipeline",
+    ),
     path(
         "project/add/",
         views.ProjectCreateView.as_view(),

scanpipe/views.py

@@ -340,6 +340,18 @@ def execute_pipeline_view(request, uuid, run_uuid):
     return redirect(project)
 
 
+def stop_pipeline_view(request, uuid, run_uuid):
+    project = get_object_or_404(Project, uuid=uuid)
+    run = get_object_or_404(Run, uuid=run_uuid, project=project)
+
+    if run.status != run.Status.RUNNING:
+        raise Http404("Pipeline is not running.")
+
+    run.stop_task()
+    messages.success(request, f'Pipeline "{run.pipeline_name}" stopped.')
+    return redirect(project)
+
+
 def project_results_json_response(project, as_attachment=False):
     """
     Returns the results as JSON compatible with ScanCode data format.