feat(manager): add Kueue scheduling option for user workloads (#658)

Introduces Kueue as an alternative way to submit user jobs.
Kueue will schedule Kubernetes jobs representing user runtime
batch and job workloads.

Co-authored-by: Xavier Tintin <[email protected]>.

Closes reanahub/reana#800

Author: CameronMcClymont

AUTHORS.md

@@ -25,3 +25,4 @@ The list of contributors in alphabetical order:
 - [Sinclert Perez](https://www.linkedin.com/in/sinclert)
 - [Tibor Simko](https://orcid.org/0000-0001-7202-5803)
 - [Vladyslav Moisieienkov](https://orcid.org/0000-0001-9717-0775)
+- [Xavier Tintin](https://orcid.org/0000-0002-3150-9112)

reana_workflow_controller/config.py

@@ -17,6 +17,7 @@
     SHARED_VOLUME_PATH,
 )
 from reana_db.models import JobStatus, RunStatus
+from distutils.util import strtobool
 
 from reana_workflow_controller.version import __version__
 
@@ -394,6 +395,12 @@ def _parse_interactive_sessions_environments(env_var):
 ]
 """Alive workflow statuses."""
 
+KUEUE_ENABLED = bool(strtobool(os.getenv("KUEUE_ENABLED", "False")))
+"""Whether to use Kueue for workflow scheduling."""
+
+KUEUE_LOCAL_QUEUE_NAME = "local-queue-batch"
+"""Name of the local queue to be used by Kueue."""
+
 REANA_RUNTIME_BATCH_TERMINATION_GRACE_PERIOD = int(
     os.getenv("REANA_RUNTIME_BATCH_TERMINATION_GRACE_PERIOD", "120")
 )

reana_workflow_controller/workflow_run_manager.py

@@ -92,6 +92,8 @@
     REANA_DASK_CLUSTER_DEFAULT_NUMBER_OF_WORKERS,
     REANA_DASK_CLUSTER_DEFAULT_SINGLE_WORKER_MEMORY,
     REANA_DASK_CLUSTER_DEFAULT_SINGLE_WORKER_THREADS,
+    KUEUE_ENABLED,
+    KUEUE_LOCAL_QUEUE_NAME,
 )
 
 
@@ -623,12 +625,17 @@ def _create_job_spec(
             self.workflow.workspace_path
         )
 
+        labels = {
+            "reana_workflow_mode": "batch",
+            "reana-run-batch-workflow-uuid": str(self.workflow.id_),
+        }
+
+        if KUEUE_ENABLED:
+            labels["kueue.x-k8s.io/queue-name"] = KUEUE_LOCAL_QUEUE_NAME
+
         workflow_metadata = client.V1ObjectMeta(
             name=name,
-            labels={
-                "reana_workflow_mode": "batch",
-                "reana-run-batch-workflow-uuid": str(self.workflow.id_),
-            },
+            labels=labels,
             namespace=REANA_RUNTIME_KUBERNETES_NAMESPACE,
         )
 
@@ -727,6 +734,7 @@ def _create_job_spec(
                 {"name": "USER", "value": user},  # Required by HTCondor
                 {"name": "K8S_CERN_EOS_AVAILABLE", "value": K8S_CERN_EOS_AVAILABLE},
                 {"name": "IMAGE_PULL_SECRETS", "value": ",".join(IMAGE_PULL_SECRETS)},
+                {"name": "KUEUE_ENABLED", "value": str(KUEUE_ENABLED)},
                 {
                     "name": "REANA_SQLALCHEMY_DATABASE_URI",
                     "value": SQLALCHEMY_DATABASE_URI,
@@ -835,7 +843,7 @@ def _create_job_spec(
             spec.template.spec.volumes.append(k8s_code_volume)
 
             for container in spec.template.spec.containers:
-                container.env.extend(current_app.config["DEBUG_ENV_VARS"])
+                # container.env.extend(current_app.config["DEBUG_ENV_VARS"])
                 sub_path = f"reana-{container.name}"
                 if container.name == "workflow-engine":
                     sub_path += f"-{self.workflow.type_}"