Abort running suites using DELETE request

manifests/base/role.yaml

@@ -9,11 +9,21 @@ metadata:
     rbac.authorization.kubernetes.io/autoupdate: "true"
   name: etos-api:sa:pod-reader
 rules:
-- apiGroups:
-  - ""
-  resources:
-  - pods
-  verbs:
-  - get
-  - list
-  - watch
+   # apiGroups batch shall be defined before apiGroups ""
+  - apiGroups:
+    - "batch"
+    resources:
+    - jobs
+    verbs:
+    - get
+    - delete
+    - list
+    - watch
+  - apiGroups:
+    - ""
+    resources:
+    - pods
+    verbs:
+    - get
+    - list
+    - watch

python/src/etos_api/routers/etos/router.py

@@ -16,26 +16,31 @@
 """ETOS API router."""
 import logging
 import os
+import time
 from uuid import uuid4
 
 from eiffellib.events import EiffelTestExecutionRecipeCollectionCreatedEvent
 from etos_lib import ETOS
 from fastapi import APIRouter, HTTPException
+from kubernetes import client, config
 from opentelemetry import trace
 
 from etos_api.library.utilities import sync_to_async
 from etos_api.library.validator import SuiteValidator
 from etos_api.routers.environment_provider.router import configure_environment_provider
 from etos_api.routers.environment_provider.schemas import ConfigureEnvironmentProviderRequest
+from etos_api.routers.logs.router import namespace
 
-from .schemas import StartEtosRequest, StartEtosResponse
+from .schemas import AbortEtosResponse, StartEtosRequest, StartEtosResponse
 from .utilities import wait_for_artifact_created
 
 ROUTER = APIRouter()
 TRACER = trace.get_tracer("etos_api.routers.etos.router")
 LOGGER = logging.getLogger(__name__)
 logging.getLogger("pika").setLevel(logging.WARNING)
 
+# Wait time between receiving DELETE request via /etos endpoint and deleting the k8s job.
+ABORT_WAIT_TIME = 300
 
 async def validate_suite(test_suite_url: str) -> None:
     """Validate the ETOS test suite through the SuiteValidator.
@@ -141,6 +146,31 @@ async def _start(etos: StartEtosRequest, span: "Span") -> dict:
         "event_repository": etos_library.debug.graphql_server,
     }
 
+async def _abort(suite_id: str, span: str = "Span") -> dict:
+    ns = namespace()
+
+    batch_api = client.BatchV1Api()
+    jobs = batch_api.list_namespaced_job(namespace=ns)
+
+    delete_options = client.V1DeleteOptions(
+        propagation_policy = "Background"  # asynchronous cascading deletion
+    )
+
+    for job in jobs.items:
+        if job.metadata.labels.get("app") == "suite-runner" and job.metadata.labels.get("id") == suite_id:
+            LOGGER.info(f"Waiting {ABORT_WAIT_TIME} seconds before deleting the suite-runner job: {suite_id}")
+            time.sleep(ABORT_WAIT_TIME)
+            batch_api.delete_namespaced_job(name=job.metadata.name, namespace=ns, body=delete_options)
+            LOGGER.info("Deleted suite-runner job: {suite_id}")
+            break
+    else:
+        raise HTTPException(
+            status_code=404,
+            detail="Suite ID not found.")
+
+    return {
+        "message": f"Abort triggered for suite id: {suite_id}."
+    }
 
 @ROUTER.post("/etos", tags=["etos"], response_model=StartEtosResponse)
 async def start_etos(etos: StartEtosRequest):
@@ -153,3 +183,15 @@ async def start_etos(etos: StartEtosRequest):
     """
     with TRACER.start_as_current_span("start-etos") as span:
         return await _start(etos, span)
+
+@ROUTER.delete("/etos/{suite_id}", tags=["etos"], response_model=AbortEtosResponse)
+async def abort_etos(suite_id: str):
+    """Abort ETOS execution on delete.
+
+    :param suite_id: ETOS suite id
+    :type suite_id: str
+    :return: JSON dictionary with response.
+    :rtype: dict
+    """
+    with TRACER.start_as_current_span("abort-etos") as span:
+        return await _abort(suite_id, span)

python/src/etos_api/routers/etos/schemas.py

@@ -25,8 +25,16 @@
 # pylint: disable=no-self-argument
 
 
-class StartEtosRequest(BaseModel):
-    """Request model for the ETOS start API."""
+class EtosRequest(BaseModel):
+    """Base class for ETOS request models."""
+
+
+class EtosResponse(BaseModel):
+    """Base class for ETOS response models."""
+
+
+class StartEtosRequest(EtosRequest):
+    """Request model for the start endpoint of the ETOS API."""
 
     artifact_identity: Optional[str]
     artifact_id: Optional[UUID]
@@ -56,10 +64,16 @@ def validate_id_or_identity(cls, artifact_id, values):
         return artifact_id
 
 
-class StartEtosResponse(BaseModel):
-    """Response model for the ETOS start API."""
+class StartEtosResponse(EtosResponse):
+    """Response model for the start endpoint of the ETOS API."""
 
     event_repository: str
     tercc: UUID
     artifact_id: UUID
     artifact_identity: str
+
+
+class AbortEtosResponse(EtosResponse):
+    """Response model for the abort endpoint of the ETOS API."""
+
+    message: str

python/src/etos_api/routers/lib/__init__.py

@@ -0,0 +1,15 @@
+# Copyright Axis Communications AB.
+#
+# For a full list of individual contributors, please see the commit history.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.

python/src/etos_api/routers/lib/kubernetes.py

@@ -0,0 +1,42 @@
+# Copyright Axis Communications AB.
+#
+# For a full list of individual contributors, please see the commit history.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import logging
+import os
+
+from kubernetes import config
+
+NAMESPACE_FILE = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"
+LOGGER = logging.getLogger(__name__)
+
+try:
+    config.load_incluster_config()
+except config.ConfigException:
+    try:
+        config.load_config()
+    except config.ConfigException:
+        LOGGER.warning("Could not load a Kubernetes config")
+
+def namespace() -> str:
+    """Get current namespace if available."""
+    if not os.path.isfile(NAMESPACE_FILE):
+        LOGGER.warning(
+            "Not running in Kubernetes. Cannot figure out namespace. "
+            "Defaulting to environment variable 'ETOS_NAMESPACE'."
+        )
+        return os.getenv("ETOS_NAMESPACE")
+    with open(NAMESPACE_FILE, encoding="utf-8") as namespace_file:
+        return namespace_file.read()
+

python/src/etos_api/routers/logs/router.py

@@ -25,30 +25,12 @@
 from sse_starlette.sse import EventSourceResponse
 from starlette.requests import Request
 
+from etos_api.routers.lib.kubernetes import namespace
+
 NAMESPACE_FILE = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"
 LOGGER = logging.getLogger(__name__)
 ROUTER = APIRouter()
 
-try:
-    config.load_incluster_config()
-except config.ConfigException:
-    try:
-        config.load_config()
-    except config.ConfigException:
-        LOGGER.warning("Could not load a Kubernetes config")
-
-
-def namespace() -> str:
-    """Get current namespace if available."""
-    if not os.path.isfile(NAMESPACE_FILE):
-        LOGGER.warning(
-            "Not running in Kubernetes. Cannot figure out namespace. "
-            "Defaulting to environment variable 'ETOS_NAMESPACE'."
-        )
-        return os.getenv("ETOS_NAMESPACE")
-    with open(NAMESPACE_FILE, encoding="utf-8") as namespace_file:
-        return namespace_file.read()
-
 
 @ROUTER.get("/logs/{uuid}", tags=["logs"])
 async def get_logs(uuid: UUID, request: Request):