feat: create task

deployment/config.yaml

@@ -72,6 +72,25 @@ custom:
     tesResources_backend_parameters:
       - VmSize
       - ParamToRecogniseDataComingFromConfig
+  taskmaster:
+    imageName: docker.io/elixircloud/tesk-core-taskmaster
+    imageVersion: v0.10.2
+    filerImageName: docker.io/elixircloud/tesk-core-filer
+    filerImageVersion: v0.10.2
+    ftp:
+      # Name of the secret with FTP account credentials
+      secretName: account-secret
+      # If FTP account enabled (based on non-emptiness of secretName)
+      enabled: true
+    # If verbose (debug) mode of taskmaster is on (passes additional flag to taskmaster and sets image pull policy to Always)
+    debug: false
+    # Environment variables, that will be passed to taskmaster
+    environment:
+      key: value
+    # Service Account name for taskmaster
+    serviceAccountName: taskmaster
+    filerBackoffLimit: 2
+    executorBackoffLimit: 2
 
 # Logging configuration
 # Cf. https://foca.readthedocs.io/en/latest/modules/foca.models.html#foca.models.config.LogConfig

tesk/api/ga4gh/tes/controllers.py

@@ -1,11 +1,15 @@
 """Controllers for GA4GH TES API endpoints."""
 
 import logging
+from typing import Any
 
-# from connexion import request  # type: ignore
 from foca.utils.logging import log_traffic  # type: ignore
+from pydantic import ValidationError
 
+from tesk.api.ga4gh.tes.models import TesTask
 from tesk.api.ga4gh.tes.service_info.service_info import ServiceInfo
+from tesk.api.ga4gh.tes.task.create_task import CreateTesTask
+from tesk.exceptions import BadRequest, InternalServerError
 
 # Get logger instance
 logger = logging.getLogger(__name__)
@@ -26,14 +30,21 @@ def CancelTask(id, *args, **kwargs) -> dict:  # type: ignore
 
 # POST /tasks
 @log_traffic
-def CreateTask(*args, **kwargs) -> dict:  # type: ignore
+def CreateTask(**kwargs) -> dict:
     """Create task.
 
     Args:
-        *args: Variable length argument list.
         **kwargs: Arbitrary keyword arguments.
     """
-    pass
+    try:
+        request_body: Any = kwargs.get("body")
+        try:
+            tes_task = TesTask(**request_body)
+        except ValidationError as e:
+            raise BadRequest(str(e)) from e
+        return CreateTesTask(tes_task).response()
+    except Exception as e:
+        raise InternalServerError from e
 
 
 # GET /tasks/service-info

tesk/api/ga4gh/tes/models.py

@@ -275,7 +275,7 @@ class TesResources(BaseModel):
         example={"VmSize": "Standard_D64_v3"},
     )
     backend_parameters_strict: Optional[bool] = Field(
-        False,
+        default=False,
         description="If set to true, backends should fail the task if any "
         "backend_parameters\nkey/values are unsupported, otherwise, backends should "
         "attempt to run the task",

tesk/api/ga4gh/tes/task/__init__.py

@@ -0,0 +1 @@
+"""Task API controller logic."""

tesk/api/ga4gh/tes/task/create_task.py

@@ -0,0 +1,81 @@
+"""TESK API module for creating a task."""
+
+import logging
+from http import HTTPStatus
+
+from tesk.api.ga4gh.tes.models import TesCreateTaskResponse, TesResources, TesTask
+from tesk.api.ga4gh.tes.task.task_request import TesTaskRequest
+from tesk.exceptions import KubernetesError
+
+logger = logging.getLogger(__name__)
+
+
+class CreateTesTask(TesTaskRequest):
+    """Create TES task."""
+
+    def __init__(
+        self,
+        task: TesTask,
+    ):
+        """Initialize the CreateTask class.
+
+        Args:
+          task: TES task to create.
+        """
+        super().__init__()
+        self.task = task
+
+    def handle_request(self) -> TesCreateTaskResponse:
+        """Create TES task."""
+        attempts_no: int = 0
+        total_attempts_no: int = (
+            self.tesk_k8s_constants.job_constants.JOB_CREATE_ATTEMPTS_NO
+        )
+
+        while (
+            attempts_no < self.tesk_k8s_constants.job_constants.JOB_CREATE_ATTEMPTS_NO
+        ):
+            try:
+                logger.debug(
+                    f"Creating K8s job, attempt no: {attempts_no}/{total_attempts_no}."
+                )
+                attempts_no += 1
+                minimum_ram_gb = self.kubernetes_client_wrapper.minimum_ram_gb()
+
+                if self.task.resources is None:
+                    self.task.resources = TesResources(cpu_cores=int(minimum_ram_gb))
+                elif (
+                    self.task.resources.ram_gb is None
+                    or self.task.resources.ram_gb < minimum_ram_gb
+                ):
+                    self.task.resources.ram_gb = minimum_ram_gb
+
+                taskmaster_job = self.tes_kubernetes_converter.from_tes_task_to_k8s_job(
+                    self.task,
+                )
+                taskmaster_config_map = (
+                    self.tes_kubernetes_converter.from_tes_task_to_k8s_config_map(
+                        self.task,
+                        taskmaster_job,
+                    )
+                )
+
+                _ = self.kubernetes_client_wrapper.create_config_map(
+                    taskmaster_config_map
+                )
+                created_job = self.kubernetes_client_wrapper.create_job(taskmaster_job)
+
+                assert created_job.metadata is not None
+                assert created_job.metadata.name is not None
+
+                return TesCreateTaskResponse(id=created_job.metadata.name)
+
+            except KubernetesError as e:
+                if (
+                    e.status != HTTPStatus.CONFLICT
+                    or attempts_no
+                    >= self.tesk_k8s_constants.job_constants.JOB_CREATE_ATTEMPTS_NO
+                ):
+                    raise e
+
+        return TesCreateTaskResponse(id="")  # To silence mypy, should never be reached

tesk/api/ga4gh/tes/task/task_request.py

@@ -0,0 +1,38 @@
+"""Base class for tesk request."""
+
+import json
+import logging
+from abc import ABC, abstractmethod
+
+from pydantic import BaseModel
+
+from tesk.k8s.constants import tesk_k8s_constants
+from tesk.k8s.converter.converter import TesKubernetesConverter
+from tesk.k8s.wrapper import KubernetesClientWrapper
+
+logger = logging.getLogger(__name__)
+
+
+class TesTaskRequest(ABC):
+    """Base class for tesk request ecapsulating common methods and members."""
+
+    def __init__(self):
+        """Initialise base class for tesk request."""
+        self.kubernetes_client_wrapper = KubernetesClientWrapper()
+        self.tes_kubernetes_converter = TesKubernetesConverter()
+        self.tesk_k8s_constants = tesk_k8s_constants
+
+    @abstractmethod
+    def handle_request(self) -> BaseModel:
+        """Business logic for the request."""
+        pass
+
+    def response(self) -> dict:
+        """Get response for the request."""
+        response: BaseModel = self.handle_request()
+        try:
+            res: dict = json.loads(json.dumps(response))
+            return res
+        except (TypeError, ValueError) as e:
+            logger.info(e)
+            return response.dict()

tesk/constants.py

@@ -19,55 +19,16 @@ class TeskConstants(BaseModel):
         TASKMASTER_ENVIRONMENT_EXECUTOR_BACKOFF_LIMIT:  Backoff limit for taskmaster env
         FILER_BACKOFF_LIMIT: Backoff limit got filer job
         EXECUTOR_BACKOFF_LIMIT: Backoff limit for executor job
-
-    Note:
-        Below are the mentioned environment variable with which these constants can be
-        configured, otherwise mentioned default will be assigned.
-
-        variable:
-            ENV_VARIABLE = default
-
-        FILER_IMAGE_NAME:
-            TESK_API_TASKMASTER_FILER_IMAGE_NAME = docker.io/elixircloud/tesk-core-filer
-        FILER_IMAGE_VERSION:
-            TESK_API_TASKMASTER_FILER_IMAGE_VERSION = latest
-        TASKMASTER_IMAGE_NAME:
-            TESK_API_TASKMASTER_IMAGE_NAME = docker.io/elixircloud/tesk-core-taskmaster
-        TASKMASTER_IMAGE_VERSION:
-            TESK_API_TASKMASTER_IMAGE_VERSION = latest
-        TESK_NAMESPACE:
-            TESK_API_K8S_NAMESPACE = tesk
-        TASKMASTER_SERVICE_ACCOUNT_NAME:
-            TESK_API_TASKMASTER_SERVICE_ACCOUNT_NAME = taskmaster
-        TASKMASTER_ENVIRONMENT_EXECUTOR_BACKOFF_LIMIT:
-            ENVIRONMENT_EXECUTOR_BACKOFF_LIMIT = 6
-        FILER_BACKOFF_LIMIT:
-            FILER_BACKOFF_LIMIT = 2
-        EXECUTOR_BACKOFF_LIMIT:
-            EXECUTOR_BACKOFF_LIMIT = 2
     """
 
-    FILER_IMAGE_NAME: str = os.getenv(
-        "TESK_API_TASKMASTER_FILER_IMAGE_NAME", "docker.io/elixircloud/tesk-core-filer"
-    )
-    FILER_IMAGE_VERSION: str = os.getenv(
-        "TESK_API_TASKMASTER_FILER_IMAGE_VERSION", "latest"
-    )
-    TASKMASTER_IMAGE_NAME: str = os.getenv(
-        "TESK_API_TASKMASTER_IMAGE_NAME", "docker.io/elixircloud/tesk-core-taskmaster"
-    )
-    TASKMASTER_IMAGE_VERSION: str = os.getenv(
-        "TESK_API_TASKMASTER_IMAGE_VERSION", "latest"
-    )
     TESK_NAMESPACE: str = os.getenv("TESK_API_K8S_NAMESPACE", "tesk")
-    TASKMASTER_SERVICE_ACCOUNT_NAME: str = os.getenv(
-        "TESK_API_TASKMASTER_SERVICE_ACCOUNT_NAME", "taskmaster"
-    )
-    TASKMASTER_ENVIRONMENT_EXECUTOR_BACKOFF_LIMIT: str = os.getenv(
-        "ENVIRONMENT_EXECUTOR_BACKOFF_LIMIT", "6"
-    )
-    FILER_BACKOFF_LIMIT: str = os.getenv("FILER_BACKOFF_LIMIT", "2")
-    EXECUTOR_BACKOFF_LIMIT: str = os.getenv("EXECUTOR_BACKOFF_LIMIT", "2")
+    FILER_IMAGE_NAME: str = "docker.io/elixircloud/tesk-core-filer"
+    FILER_IMAGE_VERSION: str = "latest"
+    TASKMASTER_IMAGE_NAME: str = "docker.io/elixircloud/tesk-core-taskmaster"
+    TASKMASTER_IMAGE_VERSION: str = "latest"
+    TASKMASTER_SERVICE_ACCOUNT_NAME: str = "taskmaster"
+    FILER_BACKOFF_LIMIT: str = "2"
+    EXECUTOR_BACKOFF_LIMIT: str = "2"
 
     class Config:
         """Configuration for class."""

tesk/custom_config.py

@@ -1,12 +1,77 @@
 """Custom configuration model for the FOCA app."""
 
+from typing import Dict, Optional
+
 from pydantic import BaseModel
 
 from tesk.api.ga4gh.tes.models import Service
+from tesk.constants import tesk_constants
+
+
+class FtpConfig(BaseModel):
+    """Ftp configuration model for the TESK.
+
+    Args:
+        secretName: Name of the secret with FTP account credentials.
+        enabled: If FTP account enabled (based on non-emptiness of secretName).
+    """
+
+    secretName: Optional[str] = None
+    enabled: bool = False
+
+
+class ExecutorSecret(BaseModel):
+    """Executor secret configuration.
+
+    Args:
+        name: Name of a secret that will be mounted as volume to each executor. The same
+            name will be used for the secret and the volume.
+        mountPath: The path where the secret will be mounted to executors.
+        enabled: Indicates whether the secret is enabled.
+    """
+
+    name: Optional[str] = None
+    mountPath: Optional[str] = None
+    enabled: bool = False
+
+
+class Taskmaster(BaseModel):
+    """Taskmaster's configuration model for the TESK.
+
+    Args:
+        imageName: Taskmaster image name.
+        imageVersion: Taskmaster image version.
+        filerImageName: Filer image name.
+        filerImageVersion: Filer image version.
+        ftp: FTP account settings.
+        debug: If verbose (debug) mode of taskmaster is on (passes additional flag to
+            taskmaster and sets image pull policy to Always).
+        environment: Environment variables, that will be passed to taskmaster.
+        serviceAccountName: Service Account name for taskmaster.
+        executorSecret: Executor secret configuration
+    """
+
+    imageName: str = tesk_constants.TASKMASTER_IMAGE_NAME
+    imageVersion: str = tesk_constants.TASKMASTER_IMAGE_VERSION
+    filerImageName: str = tesk_constants.FILER_IMAGE_NAME
+    filerImageVersion: str = tesk_constants.FILER_IMAGE_VERSION
+    ftp: FtpConfig = FtpConfig()
+    debug: bool = False
+    environment: Optional[Dict[str, str]] = None
+    serviceAccountName: str = tesk_constants.TASKMASTER_SERVICE_ACCOUNT_NAME
+    executorSecret: Optional[ExecutorSecret] = None
+    filerBackoffLimit: str = tesk_constants.FILER_BACKOFF_LIMIT
+    executorBackoffLimit: str = tesk_constants.EXECUTOR_BACKOFF_LIMIT
 
 
 class CustomConfig(BaseModel):
-    """Custom configuration model for the FOCA app."""
+    """Custom configuration model for the FOCA app.
+
+    Args:
+        service_info: Service information.
+        taskmaster: Taskmaster environment.
+    """
 
     # Define custom configuration fields here
     service_info: Service
+    taskmaster: Taskmaster = Taskmaster()

tesk/exceptions.py

@@ -19,6 +19,10 @@ class ConfigNotFoundError(FileNotFoundError):
     """Configuration file not found error."""
 
 
+class ConfigInvalidError(ValueError):
+    """Configuration file is invalid."""
+
+
 class KubernetesError(ApiException):
     """Kubernetes error."""
 

tesk/k8s/converter/__init__.py

@@ -0,0 +1 @@
+"""Module for converting Kubernetes objects to Task objects."""