feat(dask): detect and create rucio resources for Dask cluster (#634)

We detect if rucio is necessary for Dask cluster by checking top
level rucio statement under resources field in reana.yaml.

Closes reanahub/reana#873

reana_workflow_controller/dask.py

@@ -59,6 +59,7 @@ def __init__(
         user_id,
         num_of_workers,
         single_worker_memory,
+        rucio=False,
     ):
         """Instantiate Dask resource manager.
 
@@ -93,6 +94,8 @@ def __init__(
         )
         self.kubernetes_uid = WORKFLOW_RUNTIME_USER_UID
 
+        self.rucio = rucio
+
         if DASK_AUTOSCALER_ENABLED:
             self.autoscaler_name = get_dask_component_name(workflow_id, "autoscaler")
             self.autoscaler_body = self._load_dask_autoscaler_template()
@@ -212,16 +215,14 @@ def _prepare_cluster(self):
             self.secrets_store.get_file_secrets_volume_as_k8s_specs()
         )
 
-        # FIXME: Decide how to detect if krb5, rucio and voms_proxy are needed
         kerberos = False
-        rucio = False
         voms_proxy = False
 
         if kerberos:
             self._add_krb5_containers()
         if voms_proxy:
             self._add_voms_proxy_init_container()
-        if rucio:
+        if self.rucio:
             self._add_rucio_init_container()
 
     def _prepare_autoscaler(self):

reana_workflow_controller/workflow_run_manager.py

@@ -1,5 +1,5 @@
 # This file is part of REANA.
-# Copyright (C) 2019, 2020, 2021, 2022, 2023, 2024 CERN.
+# Copyright (C) 2019, 2020, 2021, 2022, 2023, 2024, 2025 CERN.
 #
 # REANA is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -353,6 +353,14 @@ def requires_kerberos(self) -> bool:
             .get("kerberos", False)
         )
 
+    def requires_rucio(self) -> bool:
+        """Check whether Rucio is necessary to run the workflow engine."""
+        return (
+            self.workflow.reana_specification["workflow"]
+            .get("resources", {})
+            .get("rucio", False)
+        )
+
 
 class KubernetesWorkflowRunManager(WorkflowRunManager):
     """Implementation of WorkflowRunManager for Kubernetes."""
@@ -399,6 +407,7 @@ def start_batch_workflow_run(
                         "single_worker_memory",
                         REANA_DASK_CLUSTER_DEFAULT_SINGLE_WORKER_MEMORY,
                     ),
+                    rucio=self.requires_rucio(),
                 ).create_dask_resources()
 
             current_k8s_batchv1_api_client.create_namespaced_job(