fix: kubernetes imagePullSecrets support for argo workflows (#2494)

Argo workflows does not support applying imagePullSecrets on a
per-template basis, only on a workflow template level.

Changing the implementation to leverage podSpecPatch in order to apply
the imagePullSecrets on a per task basis

JobSets were not tested but should be unaffected due to them utilizing
`create` within the spec and simply dumping native k8s job specs

closes #2477

Author: saikonen

metaflow/plugins/argo/argo_workflows.py

@@ -2125,8 +2125,17 @@ def _container_templates(self):
                     .node_selectors(resources.get("node_selector"))
                     # Set tolerations
                     .tolerations(resources.get("tolerations"))
-                    # Set image pull secrets
-                    .image_pull_secrets(resources.get("image_pull_secrets"))
+                    # Set image pull secrets if present. We need to use pod_spec_patch due to Argo not supporting this on a template level.
+                    .pod_spec_patch(
+                        {
+                            "imagePullSecrets": [
+                                {"name": secret}
+                                for secret in resources["image_pull_secrets"]
+                            ]
+                        }
+                        if resources["image_pull_secrets"]
+                        else None
+                    )
                     # Set container
                     .container(
                         # TODO: Unify the logic with kubernetes.py
@@ -3825,6 +3834,14 @@ def pvc_volumes(self, pvcs=None):
             )
         return self
 
+    def pod_spec_patch(self, pod_spec_patch=None):
+        if pod_spec_patch is None:
+            return self
+
+        self.payload["podSpecPatch"] = json.dumps(pod_spec_patch)
+
+        return self
+
     def node_selectors(self, node_selectors):
         if "nodeSelector" not in self.payload:
             self.payload["nodeSelector"] = {}
@@ -3836,10 +3853,6 @@ def tolerations(self, tolerations):
         self.payload["tolerations"] = tolerations
         return self
 
-    def image_pull_secrets(self, image_pull_secrets):
-        self.payload["image_pull_secrets"] = image_pull_secrets
-        return self
-
     def to_json(self):
         return self.payload