feat:add dpcloudserver and openapi kill job (#357)

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

dpdispatcher/dp_cloud_server.py

@@ -250,9 +250,16 @@ def map_dp_job_state(status):
             return JobStatus.unknown
         return map_dict[status]
 
-    # def check_finish_tag(self, job):
-    #     job_tag_finished = job.job_hash + '_job_tag_finished'
-    #     return self.context.check_file_exists(job_tag_finished)
+    def kill(self, job):
+        """Kill the job.
+
+        Parameters
+        ----------
+        job : Job
+            job
+        """
+        job_id = job.job_id
+        self.api.kill(job_id)
 
 
 DpCloudServer = Bohrium

dpdispatcher/dpcloudserver/client.py

@@ -326,3 +326,16 @@ def get_job_result_url(self, job_id):
         except ValueError as e:
             dlog.error(e, stack_info=ENABLE_STACK)
             return None
+
+    def kill(self, job_id):
+        try:
+            if not job_id:
+                return None
+            if "job_group_id" in job_id:
+                ids = job_id.split(":job_group_id:")
+                job_id, _ = int(ids[0]), int(ids[1])
+            ret = self.post(f"/brm/v1/job/kill/{job_id}", {})
+            return ret
+        except ValueError as e:
+            dlog.error(e, stack_info=ENABLE_STACK)
+            return None

dpdispatcher/openapi.py

@@ -193,6 +193,17 @@ def map_dp_job_state(status):
             return JobStatus.unknown
         return map_dict[status]
 
+    def kill(self, job):
+        """Kill the job.
+
+        Parameters
+        ----------
+        job : Job
+            job
+        """
+        job_id = job.job_id
+        self.job.kill(job_id)
+
     # def check_finish_tag(self, job):
     #     job_tag_finished = job.job_hash + '_job_tag_finished'
     #     return self.context.check_file_exists(job_tag_finished)

dpdispatcher/openapi_context.py

@@ -131,7 +131,7 @@ def upload(self, submission):
         result = None
         dlog.info("checking all job has been uploaded")
         for job in submission.belonging_jobs:
-            if job.job_state == JobStatus.unsubmitted:
+            if job.job_state in (JobStatus.unsubmitted, JobStatus.terminated):
                 job_to_be_uploaded.append(job)
         if len(job_to_be_uploaded) == 0:
             dlog.info("all job has been uploaded, continue")