feat: create TES task as k8s job

README.md

@@ -55,6 +55,32 @@ After the last executor, the `filer` is called once more to process the outputs
 and push them to remote locations from the PVC. The PVC is the scrubbed, deleted
 and the taskmaster ends, completing the task.
 
+┌─────────────────────────────────────────────────────────┐
+│                         Kubernetes                      │
+│                                                         │
+│  ┌────────────────────────────┐  ┌───────────────────┐  │
+│  │ Secret: ftp-secret         │  │ ConfigMap/PVC     │  │
+│  │ - username                 │  │ - JSON_INPUT.gz   │  │
+│  │ - password                 │  │                   │  │
+│  └──────────▲─────────────────┘  └───────▲───────────┘  │
+│             │                            |              │
+│             │                            |              │
+│             │                            |              │
+│   ┌─────────┴────────────────────────────┴────────────┐ │
+│   │                Job: taskmaster                    │ │
+│   │ ┌───────────────────────────────────────────────┐ │ │
+│   │ │ Pod: taskmaster                               │ │ │
+│   │ │ - Container: taskmaster                       │ │ │
+│   │ │   - Env: TESK_FTP_USERNAME                    │ │ │
+│   │ │   - Env: TESK_FTP_PASSWORD                    │ │ │
+│   │ │   - Args: -f /jsoninput/JSON_INPUT.gz         │ │ │
+│   │ │   - Mounts: /podinfo                          │ │ │
+│   │ │             /jsoninput                        │ │ │
+│   │ └───────────────────────────────────────────────┘ │ │
+│   └───────────────────────────────────────────────────┘ │
+│                                                         │
+└─────────────────────────────────────────────────────────┘
+
 ## Requirements
 
 - A working [Kubernetes](https://kubernetes.io/) cluster version 1.9 and later.

deployment/charts/tesk/values.yaml

@@ -9,7 +9,7 @@ host_name: ""
 #
 
 # 'openstack' or 's3'
-storage: none
+storage: s3
 
 # Configurable storage class.
 storageClass:

deployment/config.yaml

@@ -72,6 +72,78 @@ custom:
     tesResources_backend_parameters:
       - VmSize
       - ParamToRecogniseDataComingFromConfig
+  # Taskmaster configuration
+  taskmaster_template:
+    apiVersion: batch/v1
+    kind: Job
+    metadata:
+      name: taskmaster
+      labels:
+        app: taskmaster
+    spec:
+      template:
+        metadata:
+          name: taskmaster
+        spec:
+          serviceAccountName: default
+          containers:
+          - name: taskmaster
+            image: docker.io/elixircloud/tesk-core-taskmaster:v0.10.2
+            args:
+            - -f
+            - /jsoninput/JSON_INPUT.gz
+            env:
+            - name: TESK_FTP_USERNAME
+              valueFrom:
+                secretKeyRef:
+                  name: ftp-secret
+                  key: username
+                  optional: true
+            - name: TESK_FTP_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: ftp-secret
+                  key: password
+                  optional: true
+            volumeMounts:
+            - name: podinfo
+              mountPath: /podinfo
+              readOnly: true
+            - name: jsoninput
+              mountPath: /jsoninput
+              readOnly: true
+          volumes:
+          - name: podinfo
+            downwardAPI:
+              items:
+              - path: labels
+                fieldRef:
+                  fieldPath: metadata.labels
+          restartPolicy: Never
+  # Taskmaster environment properties
+  taskmaster_env_properties:
+    # Taskmaster image name
+    imageName: docker.io/elixircloud/tesk-core-taskmaster
+    # Taskmaster image version
+    imageVersion: latest
+    # Filer image name
+    filerImageName: docker.io/elixircloud/tesk-core-filer
+    # Filer image version
+    filerImageVersion: latest
+    # Test FTP account settings
+    ftp:
+      # Name of the secret with FTP account credentials
+      secretName: account-secret
+      # If FTP account enabled (based on non-emptiness of secretName)
+      enabled: false
+    # 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: default
+
 
 # Logging configuration
 # Cf. https://foca.readthedocs.io/en/latest/modules/foca.models.html#foca.models.config.LogConfig

tesk/api/ga4gh/tes/controllers.py

@@ -2,10 +2,15 @@
 
 import logging
 
-# from connexion import request  # type: ignore
 from foca.utils.logging import log_traffic  # type: ignore
 
+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.api.kubernetes.convert.converter import TesKubernetesConverter
+from tesk.api.kubernetes.convert.template import KubernetesTemplateSupplier
+from tesk.exceptions import BadRequest, InternalServerError
+from tesk.utils import get_custom_config
 
 # Get logger instance
 logger = logging.getLogger(__name__)
@@ -26,14 +31,24 @@ def CancelTask(id, *args, **kwargs) -> dict:  # type: ignore
 
 # POST /tasks
 @log_traffic
-def CreateTask(*args, **kwargs) -> dict:  # type: ignore
+def CreateTask(**kwargs) -> dict:  # type: ignore
     """Create task.
 
     Args:
         *args: Variable length argument list.
         **kwargs: Arbitrary keyword arguments.
     """
-    pass
+    try:
+        request_body = kwargs.get("body")
+        if request_body is None:
+            logger("Nothing recieved in request body.")
+            raise BadRequest("No request body recieved.")
+        tes_task = TesTask(**request_body)
+        namespace = "tesk"
+        task_creater = CreateTesTask(tes_task, namespace)
+        task_creater.response()
+    except Exception as e:
+        raise InternalServerError from e
 
 
 # GET /tasks/service-info

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 tesk.api.ga4gh.tes.models import TesTask
+from tesk.api.kubernetes.client_wrapper import KubernetesClientWrapper
+from tesk.api.kubernetes.constants import Constants
+from tesk.api.kubernetes.convert.converter import TesKubernetesConverter
+from tesk.constants import TeskConstants
+from tesk.exceptions import KubernetesError
+
+logger = logging.getLogger(__name__)
+
+
+class CreateTesTask:
+    """Create TES task."""
+
+    # TODO: Add user to the class when auth implemented in FOCA
+    def __init__(self, task: TesTask, namespace=TeskConstants.tesk_namespace):
+        """Initialize the CreateTask class.
+
+        Args:
+          task: TES task to create.
+          namespace: Kubernetes namespace where the task is created.
+        """
+        self.task = task
+        # self.user = user
+        self.kubernetes_client_wrapper = KubernetesClientWrapper()
+        self.namespace = namespace
+        self.tes_kubernetes_converter = TesKubernetesConverter(self.namespace)
+        self.constants = Constants()
+
+    def create_task(self):
+        """Create TES task."""
+        attempts_no = 0
+        while attempts_no < self.constants.job_create_attempts_no:
+            try:
+                attempts_no += 1
+                resources = self.task.resources
+
+                if resources and resources.ram_gb:
+                    minimum_ram_gb = self.kubernetes_client_wrapper.minimum_ram_gb()
+                    if resources.ram_gb < minimum_ram_gb:
+                        self.task.resources.ram_gb = minimum_ram_gb
+
+                task_master_job = (
+                    self.tes_kubernetes_converter.from_tes_task_to_k8s_job(
+                        self.task,
+                        # self.user
+                    )
+                )
+
+                task_master_config_map = (
+                    self.tes_kubernetes_converter.from_tes_task_to_k8s_config_map(
+                        self.task,
+                        task_master_job,
+                        # user
+                    )
+                )
+                _ = self.kubernetes_client_wrapper.create_config_map(
+                    task_master_config_map
+                )
+                created_job = self.kubernetes_client_wrapper.create_job(task_master_job)
+                print(task_master_config_map)
+                print(task_master_job)
+                return created_job.metadata.name
+
+            except KubernetesError as e:
+                if (
+                    not e.is_object_name_duplicated()
+                    or attempts_no >= self.constants.job_create_attempts_no
+                ):
+                    raise e
+
+            except Exception as exc:
+                logging.error("ERROR: In createTask", exc_info=True)
+                raise exc
+
+    def response(self) -> dict:
+        """Create response."""
+        return self.create_task()

tesk/api/kubernetes/__init__.py

@@ -0,0 +1 @@
+"""Kubernetes API module."""