system: Introduce Kueue as an optional alternative approach how to submit user jobs

reana_workflow_controller/config.py

@@ -13,7 +13,7 @@
 
 from reana_commons.config import REANA_COMPONENT_PREFIX, SHARED_VOLUME_PATH
 from reana_db.models import JobStatus, RunStatus
-
+from distutils.util import strtobool
 from reana_workflow_controller.version import __version__
 
 SQLALCHEMY_TRACK_MODIFICATIONS = False
@@ -172,3 +172,5 @@
     RunStatus.pending,
 ]
 """Alive workflow statuses."""
+
+USE_KUEUE = bool(strtobool(os.getenv("KUEUE_ENABLED", "False")))

reana_workflow_controller/workflow_run_manager.py

@@ -9,9 +9,8 @@
 import json
 import logging
 import os
-
 from flask import current_app
-from kubernetes import client
+from kubernetes import client, config
 from kubernetes.client.models.v1_delete_options import V1DeleteOptions
 from kubernetes.client.rest import ApiException
 from reana_commons.config import (
@@ -68,6 +67,7 @@
     REANA_WORKFLOW_ENGINE_IMAGE_YADAGE,
     WORKFLOW_ENGINE_COMMON_ENV_VARS,
     DEBUG_ENV_VARS,
+    USE_KUEUE,
 )
 
 
@@ -491,12 +491,18 @@ def _create_job_spec(
             self.workflow.workspace_path
         )
 
+        labels = {
+            "reana_workflow_mode": "batch",
+            "reana-run-batch-workflow-uuid": str(self.workflow.id_),
+            "kueue.x-k8s.io/queue-name": ("local-queue-batch" if USE_KUEUE else None),
+        }
+
+        # Filter out None values (labels should not include None)
+        labels = {k: v for k, v in labels.items() if v is not None}
+
         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,
         )
 
@@ -524,6 +530,10 @@ def _create_job_spec(
             volume_mounts=[],
             command=["/bin/bash", "-c"],
             args=command,
+            # resources=client.V1ResourceRequirements(
+            #     requests={"cpu": "1", "memory": "2Gi"},
+            #     limits={"cpu": "3", "memory": "4Gi"},
+            # ),
         )
         workflow_engine_env_vars.extend(
             [
@@ -575,6 +585,10 @@ def _create_job_spec(
             command=["/bin/bash", "-c"],
             args=self._create_job_controller_startup_cmd(user),
             ports=[],
+            # resources=client.V1ResourceRequirements(
+            #     requests={"cpu": "1", "memory": "2Gi"},
+            #     limits={"cpu": "3", "memory": "4Gi"},
+            # ),
         )
 
         job_controller_env_vars.extend(
@@ -584,6 +598,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": "USE_KUEUE", "value": str(USE_KUEUE)},
                 {
                     "name": "REANA_SQLALCHEMY_DATABASE_URI",
                     "value": SQLALCHEMY_DATABASE_URI,