Issue 474 | Gateway: stop jobs

client/quantum_serverless/core/job.py

@@ -30,6 +30,7 @@
 import json
 import logging
 import os
+import tarfile
 from typing import Dict, Any, Optional
 from uuid import uuid4
 
@@ -47,6 +48,8 @@
     GATEWAY_PROVIDER_VERSION_DEFAULT,
 )
 from quantum_serverless.core.program import Program
+from quantum_serverless.exception import QuantumServerlessException
+from quantum_serverless.serializers.program_serializers import QiskitObjectsEncoder
 from quantum_serverless.utils.json import is_jsonable
 
 RuntimeEnv = ray.runtime_env.RuntimeEnv
@@ -148,6 +151,43 @@ def __init__(self, host: str, token: str, version: str):
         self.version = version
         self._token = token
 
+    def run_program(
+        self, program: Program, arguments: Optional[Dict[str, Any]] = None
+    ) -> "Job":
+        url = f"{self.host}/api/{self.version}/programs/run/"
+        artifact_file_path = os.path.join(program.working_dir, "artifact.tar")
+
+        with tarfile.open(artifact_file_path, "w") as tar:
+            for filename in os.listdir(program.working_dir):
+                fpath = os.path.join(program.working_dir, filename)
+                tar.add(fpath, arcname=filename)
+
+        with open(artifact_file_path, "rb") as file:
+            response = requests.post(
+                url=url,
+                data={
+                    "title": program.title,
+                    "entrypoint": program.entrypoint,
+                    "arguments": json.dumps(arguments or {}, cls=QiskitObjectsEncoder),
+                    "dependencies": json.dumps(program.dependencies or []),
+                },
+                files={"artifact": file},
+                headers={"Authorization": f"Bearer {self._token}"},
+                timeout=REQUESTS_TIMEOUT,
+            )
+            if not response.ok:
+                raise QuantumServerlessException(
+                    f"Something went wrong with program execution. {response.text}"
+                )
+
+            json_response = json.loads(response.text)
+            job_id = json_response.get("id")
+
+        if os.path.exists(artifact_file_path):
+            os.remove(artifact_file_path)
+
+        return Job(job_id, job_client=self)
+
     def status(self, job_id: str):
         default_status = "Unknown"
         status = default_status
@@ -165,7 +205,19 @@ def status(self, job_id: str):
         return status
 
     def stop(self, job_id: str):
-        raise NotImplementedError
+        message = ""
+        response = requests.post(
+            f"{self.host}/api/{self.version}/jobs/{job_id}/stop/",
+            headers={"Authorization": f"Bearer {self._token}"},
+            timeout=REQUESTS_TIMEOUT,
+        )
+        if response.ok:
+            message = json.loads(response.text).get("message", None)
+        else:
+            logging.warning(
+                "Something went wrong during job stopping. %s", response.text
+            )
+        return message
 
     def logs(self, job_id: str):
         result = None

client/quantum_serverless/core/provider.py

@@ -29,7 +29,6 @@
 import json
 import logging
 import os.path
-import tarfile
 from dataclasses import dataclass
 from typing import Optional, List, Dict, Any
 
@@ -53,7 +52,6 @@
 from quantum_serverless.core.program import Program
 from quantum_serverless.core.tracing import _trace_env_vars
 from quantum_serverless.exception import QuantumServerlessException
-from quantum_serverless.serializers.program_serializers import QiskitObjectsEncoder
 from quantum_serverless.utils import JsonSerializable
 
 TIMEOUT = 30
@@ -520,6 +518,8 @@ def __init__(
         if token is None:
             self._fetch_token(username, password)
 
+        self._job_client = GatewayJobClient(self.host, self._token, self.version)
+
     def get_compute_resources(self) -> List[ComputeResource]:
         raise NotImplementedError("GatewayProvider does not support resources api yet.")
 
@@ -541,7 +541,7 @@ def get_job_by_id(self, job_id: str) -> Optional[Job]:
             data = json.loads(response.text)
             job = Job(
                 job_id=data.get("id"),
-                job_client=GatewayJobClient(self.host, self._token, self.version),
+                job_client=self._job_client,
             )
         else:
             logging.warning(response.text)
@@ -551,41 +551,7 @@ def get_job_by_id(self, job_id: str) -> Optional[Job]:
     def run_program(
         self, program: Program, arguments: Optional[Dict[str, Any]] = None
     ) -> Job:
-        url = f"{self.host}/api/{self.version}/programs/run/"
-        artifact_file_path = os.path.join(program.working_dir, "artifact.tar")
-
-        with tarfile.open(artifact_file_path, "w") as tar:
-            for filename in os.listdir(program.working_dir):
-                fpath = os.path.join(program.working_dir, filename)
-                tar.add(fpath, arcname=filename)
-
-        with open(artifact_file_path, "rb") as file:
-            response = requests.post(
-                url=url,
-                data={
-                    "title": program.title,
-                    "entrypoint": program.entrypoint,
-                    "arguments": json.dumps(arguments or {}, cls=QiskitObjectsEncoder),
-                    "dependencies": json.dumps(program.dependencies or []),
-                },
-                files={"artifact": file},
-                headers={"Authorization": f"Bearer {self._token}"},
-                timeout=REQUESTS_TIMEOUT,
-            )
-            if not response.ok:
-                raise QuantumServerlessException(
-                    f"Something went wrong with program execution. {response.text}"
-                )
-
-            json_response = json.loads(response.text)
-            job_id = json_response.get("id")
-
-        if os.path.exists(artifact_file_path):
-            os.remove(artifact_file_path)
-
-        return Job(
-            job_id, job_client=GatewayJobClient(self.host, self._token, self.version)
-        )
+        return self._job_client.run_program(program, arguments)
 
     def get_jobs(self, **kwargs) -> List[Job]:
         jobs = []
@@ -599,7 +565,7 @@ def get_jobs(self, **kwargs) -> List[Job]:
             jobs = [
                 Job(
                     job_id=job.get("id"),
-                    job_client=GatewayJobClient(self.host, self._token, self.version),
+                    job_client=self._job_client,
                 )
                 for job in json.loads(response.text).get("results", [])
             ]

gateway/api/migrations/0001_initial.py

@@ -9,7 +9,6 @@
 
 
 class Migration(migrations.Migration):
-
     initial = True
 
     dependencies = [

gateway/api/views.py

@@ -175,6 +175,19 @@ def logs(self, request, pk=None):  # pylint: disable=invalid-name,unused-argumen
         ray_client = JobSubmissionClient(job.compute_resource.host)
         return Response({"logs": ray_client.get_job_logs(job.ray_job_id)})
 
+    @action(methods=["POST"], detail=True)
+    def stop(self, request, pk=None):  # pylint: disable=invalid-name,unused-argument
+        """Stops job"""
+        job = self.get_object()
+        ray_client = JobSubmissionClient(job.compute_resource.host)
+        was_running = ray_client.stop_job(job.ray_job_id)
+        message = (
+            "Job has been stopped successfully."
+            if not was_running
+            else "Job was already not running."
+        )
+        return Response({"message": message})
+
 
 class KeycloakLogin(SocialLoginView):
     """KeycloakLogin."""