feat(dask): create VOMS proxy sidecars for Dask workflows (#633)

We detect if VOMS_proxy is necessary for Dask cluster by checking top
level voms_proxy statement under resources field in reana.yaml. Some
amendments were needed in order to fetch the secrets from secrets store
instead of environment variables since workflow controller does not have
access to runtime environment variables just like job controller.

Closes reanahub/reana#872

Author: Alputer

reana_workflow_controller/dask.py

@@ -60,6 +60,7 @@ def __init__(
         num_of_workers,
         single_worker_memory,
         kerberos=False,
+        voms_proxy=False,
     ):
         """Instantiate Dask resource manager.
 
@@ -95,6 +96,7 @@ def __init__(
         self.kubernetes_uid = WORKFLOW_RUNTIME_USER_UID
 
         self.kerberos = kerberos
+        self.voms_proxy = voms_proxy
 
         if DASK_AUTOSCALER_ENABLED:
             self.autoscaler_name = get_dask_component_name(workflow_id, "autoscaler")
@@ -216,11 +218,10 @@ def _prepare_cluster(self):
         )
 
         rucio = False
-        voms_proxy = False
 
         if self.kerberos:
             self._add_krb5_containers()
-        if voms_proxy:
+        if self.voms_proxy:
             self._add_voms_proxy_init_container()
         if rucio:
             self._add_rucio_init_container()
@@ -343,6 +344,28 @@ def _add_krb5_containers(self):
             f"trap 'touch {KRB5_STATUS_FILE_LOCATION}' EXIT; " + existing_args[0]
         ]
 
+    def _get_voms_proxy_secrets(self, secrets_store):
+        """Get VOMS proxy secrets from secrets store.
+
+        Args:
+            secrets_store: User secrets store instance
+
+        Returns:
+            dict: Dictionary containing VOMS proxy secrets with empty string defaults
+        """
+        secret_keys = ["VONAME", "VOMSPROXY_FILE", "VOMSPROXY_PASS"]
+        secrets = {}
+
+        for key in secret_keys:
+            secret = secrets_store.get_secret(key)
+            secrets[key.lower()] = secret.value_str if secret else ""
+
+        return {
+            "vo": secrets["voname"],
+            "file": secrets["vomsproxy_file"],
+            "pass": secrets["vomsproxy_pass"],
+        }
+
     def _add_voms_proxy_init_container(self):
         """Add sidecar container for Dask workers."""
         ticket_cache_volume = {"name": "voms-proxy-cache", "emptyDir": {}}
@@ -358,8 +381,10 @@ def _add_voms_proxy_init_container(self):
             current_app.config["VOMSPROXY_CERT_CACHE_FILENAME"],
         )
 
-        voms_proxy_vo = os.environ.get("VONAME", "")
-        voms_proxy_user_file = os.environ.get("VOMSPROXY_FILE", "")
+        voms_proxy_secrets = self._get_voms_proxy_secrets(self.secrets_store)
+        voms_proxy_vo = voms_proxy_secrets["vo"]
+        voms_proxy_user_file = voms_proxy_secrets["file"]
+        voms_proxy_pass = voms_proxy_secrets["pass"]
 
         if voms_proxy_user_file:
             # multi-user deployment mode, where we rely on VOMS proxy file supplied by the user
@@ -399,23 +424,24 @@ def _add_voms_proxy_init_container(self):
                         echo "[ERROR] File usercert.pem does not exist in user secrets."; \
                         exit; \
                      fi; \
-                     if [ -z "$VOMSPROXY_PASS" ]; then \
+                     if [ -z {voms_proxy_pass} ]; then \
                         echo "[ERROR] Environment variable VOMSPROXY_PASS is not set in user secrets."; \
                         exit; \
                      fi; \
-                     if [ -z "$VONAME" ]; then \
+                     if [ -z {voms_proxy_vo} ]; then \
                         echo "[ERROR] Environment variable VONAME is not set in user secrets."; \
                         exit; \
                      fi; \
                      cp /etc/reana/secrets/userkey.pem /tmp/userkey.pem; \
                          chmod 400 /tmp/userkey.pem; \
-                         echo $VOMSPROXY_PASS | base64 -d | voms-proxy-init \
+                         echo {voms_proxy_pass} | base64 -d | voms-proxy-init \
                          --voms {voms_proxy_vo} --key /tmp/userkey.pem \
                          --cert $(readlink -f /etc/reana/secrets/usercert.pem) \
                          --pwstdin --out {voms_proxy_file_path}; \
                          chown {kubernetes_uid} {voms_proxy_file_path}'.format(
                         voms_proxy_vo=voms_proxy_vo.lower(),
                         voms_proxy_file_path=voms_proxy_file_path,
+                        voms_proxy_pass=voms_proxy_pass,
                         kubernetes_uid=self.kubernetes_uid,
                     ),
                 ],

reana_workflow_controller/workflow_run_manager.py

@@ -353,6 +353,14 @@ def requires_kerberos(self) -> bool:
             .get("kerberos", False)
         )
 
+    def requires_voms_proxy(self) -> bool:
+        """Check whether Voms_proxy is necessary to run the workflow engine."""
+        return (
+            self.workflow.reana_specification["workflow"]
+            .get("resources", {})
+            .get("voms_proxy", False)
+        )
+
 
 class KubernetesWorkflowRunManager(WorkflowRunManager):
     """Implementation of WorkflowRunManager for Kubernetes."""
@@ -400,6 +408,7 @@ def start_batch_workflow_run(
                         REANA_DASK_CLUSTER_DEFAULT_SINGLE_WORKER_MEMORY,
                     ),
                     kerberos=self.requires_kerberos(),
+                    voms_proxy=self.requires_voms_proxy(),
                 ).create_dask_resources()
 
             current_k8s_batchv1_api_client.create_namespaced_job(

setup.py

@@ -51,7 +51,7 @@
     "marshmallow>2.13.0,<3.0.0",  # same upper pin as reana-server
     "opensearch-py>=2.7.0,<2.8.0",
     "packaging>=18.0",
-    "reana-commons[kubernetes]>=0.95.0a7,<0.96.0",
+    "reana-commons[kubernetes]>=0.95.0a8,<0.96.0",
     "reana-db>=0.95.0a5,<0.96.0",
     "requests>=2.25.0",
     "sqlalchemy-utils>=0.31.0",