feat: set taskmaster manifest from FOCA config (#206)

Author: JaeAeich

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/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/utils.py

@@ -3,15 +3,162 @@
 import os
 from pathlib import Path
 
+from foca import Foca
+from kubernetes.client.models import (
+    V1Container,
+    V1DownwardAPIVolumeFile,
+    V1DownwardAPIVolumeSource,
+    V1EnvVar,
+    V1EnvVarSource,
+    V1Job,
+    V1JobSpec,
+    V1ObjectFieldSelector,
+    V1ObjectMeta,
+    V1PodSpec,
+    V1PodTemplateSpec,
+    V1SecretKeySelector,
+    V1Volume,
+    V1VolumeMount,
+)
+
+from tesk.custom_config import (
+    CustomConfig,
+    Taskmaster,
+)
+from tesk.exceptions import ConfigInvalidError
+from tesk.k8s.constants import tesk_k8s_constants
+
 
 def get_config_path() -> Path:
     """Get the configuration path.
 
     Returns:
-      The path of the config file.
+        The path of the config file.
     """
     # Determine the configuration path
     if config_path_env := os.getenv("TESK_FOCA_CONFIG_PATH"):
         return Path(config_path_env).resolve()
     else:
         return (Path(__file__).parents[1] / "deployment" / "config.yaml").resolve()
+
+
+def get_custom_config() -> CustomConfig:
+    """Get the custom configuration.
+
+    Returns:
+        The custom configuration.
+    """
+    conf = Foca(config_file=get_config_path()).conf
+    try:
+        return CustomConfig(**conf.custom)
+    except AttributeError:
+        raise ConfigInvalidError(
+            "Custom configuration not found in config file."
+        ) from None
+
+
+def get_taskmaster_config() -> Taskmaster:
+    """Get the taskmaster env property from the custom configuration.
+
+    Returns:
+        The taskmaster env property.
+    """
+    custom_conf = get_custom_config()
+    try:
+        return custom_conf.taskmaster
+    except AttributeError:
+        raise ConfigInvalidError(
+            "Custom configuration doesn't seem to have taskmaster_env_properties in "
+            "config file."
+            f"Custom config:\n{custom_conf}"
+        ) from None
+
+
+def get_taskmaster_template() -> V1Job:
+    """Get the taskmaster template from the custom configuration.
+
+    This will be used to create the taskmaster job, API will inject values
+    into the template, depending upon the type of job and request.
+
+    Returns:
+        The taskmaster template.
+    """
+    taskmaster_conf: Taskmaster = get_taskmaster_config()
+
+    return V1Job(
+        api_version=tesk_k8s_constants.k8s_constants.K8S_BATCH_API_VERSION,
+        kind=tesk_k8s_constants.k8s_constants.K8S_BATCH_API_JOB_TYPE,
+        metadata=V1ObjectMeta(
+            name=tesk_k8s_constants.label_constants.LABEL_JOBTYPE_VALUE_TASKM,
+        ),
+        spec=V1JobSpec(
+            template=V1PodTemplateSpec(
+                metadata=V1ObjectMeta(
+                    name=tesk_k8s_constants.label_constants.LABEL_JOBTYPE_VALUE_TASKM
+                ),
+                spec=V1PodSpec(
+                    service_account_name=taskmaster_conf.serviceAccountName,
+                    containers=[
+                        V1Container(
+                            name=tesk_k8s_constants.label_constants.LABEL_JOBTYPE_VALUE_TASKM,
+                            image=f"{taskmaster_conf.imageName}:{taskmaster_conf.imageVersion}",
+                            args=[
+                                "-f",
+                                f"/jsoninput/{tesk_k8s_constants.job_constants.TASKMASTER_INPUT}.gz",
+                            ],
+                            env=[
+                                V1EnvVar(
+                                    name=tesk_k8s_constants.ftp_constants.FTP_SECRET_USERNAME_ENV,
+                                    value_from=V1EnvVarSource(
+                                        secret_key_ref=V1SecretKeySelector(
+                                            name="ftp-secret",
+                                            key="username",
+                                            optional=True,
+                                        )
+                                    ),
+                                ),
+                                V1EnvVar(
+                                    name=tesk_k8s_constants.ftp_constants.FTP_SECRET_PASSWORD_ENV,
+                                    value_from=V1EnvVarSource(
+                                        secret_key_ref=V1SecretKeySelector(
+                                            name="ftp-secret",
+                                            key="password",
+                                            optional=True,
+                                        )
+                                    ),
+                                ),
+                            ],
+                            volume_mounts=[
+                                V1VolumeMount(
+                                    name="podinfo",
+                                    mount_path="/podinfo",
+                                    read_only=True,
+                                ),
+                                V1VolumeMount(
+                                    name="jsoninput",
+                                    mount_path="/jsoninput",
+                                    read_only=True,
+                                ),
+                            ],
+                        )
+                    ],
+                    volumes=[
+                        V1Volume(
+                            name="podinfo",
+                            downward_api=V1DownwardAPIVolumeSource(
+                                items=[
+                                    V1DownwardAPIVolumeFile(
+                                        path="labels",
+                                        field_ref=V1ObjectFieldSelector(
+                                            field_path="metadata.labels"
+                                        ),
+                                    ),
+                                ]
+                            ),
+                        ),
+                    ],
+                    restart_policy=tesk_k8s_constants.k8s_constants.JOB_RESTART_POLICY,
+                ),
+            )
+        ),
+    )