feat: get taskmaster values from config

deployment/config.yaml

@@ -72,6 +72,23 @@ 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
 
 # Logging configuration
 # Cf. https://foca.readthedocs.io/en/latest/modules/foca.models.html#foca.models.config.LogConfig

tesk/custom_config.py

@@ -1,12 +1,83 @@
 """Custom configuration model for the FOCA app."""
 
-from pydantic import BaseModel
+from typing import Dict, Optional
+
+from pydantic import BaseModel, Field
 
 from tesk.api.ga4gh.tes.models import Service
+from tesk.constants import tesk_constants
+
+
+class FtpConfig(BaseModel):
+    """Ftp configuration model for the TESK."""
+
+    secretName: Optional[str] = Field(
+        default=None, description="Name of the secret with FTP account credentials"
+    )
+    enabled: bool = Field(
+        default=False,
+        description="If FTP account enabled (based on non-emptiness of secretName)",
+    )
+
+
+class ExecutorSecret(BaseModel):
+    """Executor secret configuration."""
+
+    name: Optional[str] = Field(
+        default=None,
+        description=(
+            "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: Optional[str] = Field(
+        default=None,
+        alias="mountPath",
+        description="The path where the secret will be mounted to executors",
+    )
+    enabled: bool = Field(
+        default=False, description="Indicates whether the secret is enabled"
+    )
+
+
+class Taskmaster(BaseModel):
+    """Taskmaster environment properties model for the TESK."""
+
+    imageName: str = Field(
+        default=tesk_constants.TASKMASTER_IMAGE_NAME,
+        description="Taskmaster image name",
+    )
+    imageVersion: str = Field(
+        default=tesk_constants.TASKMASTER_IMAGE_VERSION,
+        description="Taskmaster image version",
+    )
+    filerImageName: str = Field(
+        default=tesk_constants.FILER_IMAGE_NAME, description="Filer image name"
+    )
+    filerImageVersion: str = Field(
+        default=tesk_constants.FILER_IMAGE_VERSION, description="Filer image version"
+    )
+    ftp: FtpConfig = Field(default=None, description="Test FTP account settings")
+    debug: bool = Field(
+        default=False,
+        description="If verbose (debug) mode of taskmaster is on (passes additional "
+        "flag to taskmaster and sets image pull policy to Always)",
+    )
+    environment: Optional[Dict[str, str]] = Field(
+        default=None,
+        description="Environment variables, that will be passed to taskmaster",
+    )
+    serviceAccountName: str = Field(
+        default="default", description="Service Account name for taskmaster"
+    )
+    executorSecret: Optional[ExecutorSecret] = Field(
+        default=None, description="Executor secret configuration"
+    )
 
 
 class CustomConfig(BaseModel):
     """Custom configuration model for the FOCA app."""
 
     # Define custom configuration fields here
     service_info: Service
+    taskmaster: Taskmaster

tesk/utils.py

@@ -3,6 +3,28 @@
 import os
 from pathlib import Path
 
+from foca import Foca
+from kubernetes.client.models import (
+    V1Container,
+    V1EnvVar,
+    V1EnvVarSource,
+    V1Job,
+    V1JobSpec,
+    V1ObjectMeta,
+    V1PodSpec,
+    V1PodTemplateSpec,
+    V1SecretKeySelector,
+    V1VolumeMount,
+)
+
+from tesk.constants import TeskConstants
+from tesk.custom_config import (
+    CustomConfig,
+    TaskmasterEnvProperties,
+)
+from tesk.exceptions import ConfigNotFoundError
+from tesk.k8s.constants import TeskK8sConstants
+
 
 def get_config_path() -> Path:
     """Get the configuration path.
@@ -15,3 +37,108 @@ def get_config_path() -> 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 ConfigNotFoundError(
+            "Custom configuration not found in config file."
+        ) from None
+
+
+def get_taskmaster_template() -> V1Job:
+    """Get the taskmaster template from the custom configuration.
+
+    Returns:
+      The taskmaster template.
+    """
+    job = V1Job(
+        api_version="batch/v1",
+        kind="Job",
+        metadata=V1ObjectMeta(
+            name=TeskK8sConstants.label_constants.LABEL_JOBTYPE_VALUE_TASKM,
+            labels={"app": TeskK8sConstants.label_constants.LABEL_JOBTYPE_VALUE_TASKM},
+        ),
+        spec=V1JobSpec(
+            template=V1PodTemplateSpec(
+                metadata=V1ObjectMeta(
+                    name=TeskK8sConstants.label_constants.LABEL_JOBTYPE_VALUE_TASKM
+                ),
+                spec=V1PodSpec(
+                    service_account_name="default",
+                    containers=[
+                        V1Container(
+                            name=TeskK8sConstants.label_constants.LABEL_JOBTYPE_VALUE_TASKM,
+                            image=f"{TeskConstants.TASKMASTER_IMAGE_NAME}:{TeskConstants.TASKMASTER_IMAGE_VERSION}",
+                            args=[
+                                "-f",
+                                f"/jsoninput/{TeskK8sConstants.job_constants.TASKMASTER_INPUT}.gz",
+                            ],
+                            env=[
+                                V1EnvVar(
+                                    name=TeskK8sConstants.ftp_constants.FTP_SECRET_USERNAME_ENV,
+                                    value_from=V1EnvVarSource(
+                                        secret_key_ref=V1SecretKeySelector(
+                                            name="ftp-secret",
+                                            key="username",
+                                            optional=True,
+                                        )
+                                    ),
+                                ),
+                                V1EnvVar(
+                                    name=TeskK8sConstants.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=[],
+                    restart_policy=TeskK8sConstants.k8s_constants.JOB_RESTART_POLICY,
+                ),
+            )
+        ),
+    )
+    return job
+
+
+def get_taskmaster_env_property() -> TaskmasterEnvProperties:
+    """Get the taskmaster env property from the custom configuration.
+
+    Returns:
+      The taskmaster env property.
+    """
+    custom_conf = get_custom_config()
+    try:
+        return custom_conf.taskmaster_env_properties
+    except AttributeError:
+        raise ConfigNotFoundError(
+            "Custom configuration doesn't seem to have taskmaster_env_properties in "
+            "config file."
+            f"Custom config:\n{custom_conf}"
+        ) from None