feat: create TES task as k8s job

deployment/config.yaml

@@ -48,6 +48,78 @@ api:
  swagger_ui: True
  serve_spec: True
 
+custom:
+ 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: eu.gcr.io/tes-wes/taskmaster:v0.0.5.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_env_properties:
+ # Taskmaster image name
+ imageName: eu.gcr.io/tes-wes/taskmaster
+ # Taskmaster image version
+ imageVersion: v0.10.0
+ # Filer image name
+ filerImageName: eu.gcr.io/tes-wes/filer
+ # Filer image version
+ filerImageVersion: v0.10.0
+ # 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
 log:

pyproject.toml

@@ -30,7 +30,7 @@ version = "0.1.0"
 [tool.poetry.dependencies]
 boto3 = "1.34.104"
 foca = "^0.13.0"
-kubernetes = "^29.0.0"
+kubernetes = "^30.1.0"
 python = "^3.11"
 requests = ">=2.20.0"
 urllib3 = "^2.2.1"

tesk/api/ga4gh/tes/controllers.py

@@ -2,9 +2,15 @@
 
 import logging
 
-# from connexion import request # type: ignore
 from foca.utils.logging import log_traffic # type: ignore
 
+from tesk.api.kubernetes.template import KubernetesTemplateSupplier
+from tesk.api.ga4gh.tes.models import TesTask
+from tesk.utils import get_custom_config
+from tesk.api.kubernetes.converter import TesKubernetesConverter
+
+# from tesk.api.ga4gh.tes.task.create_task import CreateTask as TaskCreater
+
 # Get logger instance
 logger = logging.getLogger(__name__)
 
@@ -24,13 +30,22 @@ 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.
  """
+ request_body = kwargs["body"]
+ create_task_request = TesTask(**request_body)
+ user = {
+ "username": "test",
+ "is_member": True,
+ "any_group": "test_group",
+ }
+ tkc = TesKubernetesConverter().from_tes_task_to_k8s_job(create_task_request, user)
+ print(tkc)
  pass
 
 

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,77 @@
+"""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.converter import TesKubernetesConverter
+from tesk.exceptions import KubernetesError
+
+logger = logging.getLogger(__name__)
+
+
+class CreateTesTask:
+ """Create TES task."""
+
+ def __init__(self, task: TesTask, user=None, namespace="default"):
+ """Initialize the CreateTask class.
+
+ Args:
+ task: TES task to create.
+ user: User who creates the task.
+ 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 True:
+ try:
+ 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
+ )
+ )
+
+ # TODO: Create ConfigMap
+ # TODO: Create Job
+ # TODO Return created job
+ # task_master_config_map = converter.from_tes_task_to_k8s_config_map(
+ # task, user, task_master_job
+ # )
+ # created_config_map = kubernetes_client_wrapper.create_config_map(
+ # task_master_config_map
+ # )
+ # created_job = kubernetes_client_wrapper.create_job(task_master_job)
+ # return converter.from_k8s_job_to_tes_create_task_response(created_job)
+
+ except KubernetesError as e:
+ # Handle Kubernetes specific exceptions
+ if (
+ not e.is_object_name_duplicated()
+ or attempts_no >= self.constants.job_create_attempts_no
+ ):
+ raise e
+ attempts_no += 1
+
+ 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."""