feat(job-manager): add Kueue scheduling option for user workloads (#492)

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_job_controller/config.py

@@ -177,6 +177,12 @@
 SLURM_SSH_AUTH_TIMEOUT = float(os.getenv("SLURM_SSH_AUTH_TIMEOUT", "60"))
 """Seconds to wait for SLURM SSH authentication response."""
 
+USE_KUEUE = bool(strtobool(os.getenv("USE_KUEUE", "False")))
+"""Whether to use Kueue to manage job execution."""
+
+KUEUE_LOCAL_QUEUE_NAME = "local-queue-job"
+"""Name of the local queue to be used by Kueue."""
+
 REANA_USER_ID = os.getenv("REANA_USER_ID")
 """User UUID of the owner of the workflow."""
 

reana_job_controller/kubernetes_job_manager.py

@@ -52,6 +52,8 @@
     REANA_KUBERNETES_JOBS_MEMORY_LIMIT,
     REANA_KUBERNETES_JOBS_MAX_USER_MEMORY_LIMIT,
     REANA_USER_ID,
+    USE_KUEUE,
+    KUEUE_LOCAL_QUEUE_NAME,
 )
 from reana_job_controller.errors import ComputingBackendSubmissionError
 from reana_job_controller.job_manager import JobManager
@@ -155,12 +157,18 @@ def secrets(self):
     def execute(self):
         """Execute a job in Kubernetes."""
         backend_job_id = build_unique_component_name("run-job")
+
         self.job = {
             "kind": "Job",
             "apiVersion": "batch/v1",
             "metadata": {
                 "name": backend_job_id,
                 "namespace": REANA_RUNTIME_KUBERNETES_NAMESPACE,
+                "labels": (
+                    {"kueue.x-k8s.io/queue-name": KUEUE_LOCAL_QUEUE_NAME}
+                    if USE_KUEUE
+                    else {}
+                ),
             },
             "spec": {
                 "backoffLimit": KubernetesJobManager.MAX_NUM_JOB_RESTARTS,