feat(dask): create Kerberos sidecars for Dask workflows (#632)

If we detect that the user asked for Kerberos resources, we create the
necessary Kerberos sidecar and add relevant credentials to Dask pods as
environment variables. No additional changes were needed as first
implementation of adding kerberos credentials works just fine.

Closes reanahub/reana#871

reana_workflow_controller/dask.py

@@ -59,6 +59,7 @@ def __init__(
         user_id,
         num_of_workers,
         single_worker_memory,
+        kerberos=False,
     ):
         """Instantiate Dask resource manager.
 
@@ -93,6 +94,8 @@ def __init__(
         )
         self.kubernetes_uid = WORKFLOW_RUNTIME_USER_UID
 
+        self.kerberos = kerberos
+
         if DASK_AUTOSCALER_ENABLED:
             self.autoscaler_name = get_dask_component_name(workflow_id, "autoscaler")
             self.autoscaler_body = self._load_dask_autoscaler_template()
@@ -212,12 +215,10 @@ 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:
+        if self.kerberos:
             self._add_krb5_containers()
         if voms_proxy:
             self._add_voms_proxy_init_container()

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.
@@ -399,6 +399,7 @@ def start_batch_workflow_run(
                         "single_worker_memory",
                         REANA_DASK_CLUSTER_DEFAULT_SINGLE_WORKER_MEMORY,
                     ),
+                    kerberos=self.requires_kerberos(),
                 ).create_dask_resources()
 
             current_k8s_batchv1_api_client.create_namespaced_job(