fix: handle big parameters of failed step

fix: add sub_path and slice to download_artifact
fix: data persistent for minio

Signed-off-by: zjgemi <[email protected]>

Author: zjgemi

manifests/quick-start-postgres-stable-cn.yaml

@@ -1692,6 +1692,9 @@ spec:
               - mkdir
               - -p
               - /data/my-bucket
+        volumeMounts:
+        - mountPath: /data
+          name: minio-data
         livenessProbe:
           httpGet:
             path: /minio/health/live
@@ -1710,6 +1713,11 @@ spec:
             port: 9000
           initialDelaySeconds: 5
           periodSeconds: 10
+      volumes:
+      - hostPath:
+          path: /data/minio
+          type: DirectoryOrCreate
+        name: minio-data
 ---
 apiVersion: apps/v1
 kind: Deployment

manifests/quick-start-postgres.yaml

@@ -1692,6 +1692,9 @@ spec:
               - mkdir
               - -p
               - /data/my-bucket
+        volumeMounts:
+        - mountPath: /data
+          name: minio-data
         livenessProbe:
           httpGet:
             path: /minio/health/live
@@ -1710,6 +1713,11 @@ spec:
             port: 9000
           initialDelaySeconds: 5
           periodSeconds: 10
+      volumes:
+      - hostPath:
+          path: /data/minio
+          type: DirectoryOrCreate
+        name: minio-data
 ---
 apiVersion: apps/v1
 kind: Deployment

src/dflow/argo_objects.py

@@ -106,22 +106,26 @@ def handle_big_parameters(self, io):
                     if name[13:] in io.parameters:
                         continue
                     with tempfile.TemporaryDirectory() as tmpdir:
-                        download_artifact(art, path=tmpdir)
-                        fs = os.listdir(tmpdir)
-                        assert len(fs) == 1
-                        with open(os.path.join(tmpdir, fs[0]), "r") as f:
-                            content = jsonpickle.loads(f.read())
-                            param = {"name": name[13:],
-                                     "save_as_artifact": True}
-                            if "type" in content:
-                                param["type"] = content["type"]
-                            if "type" in content and content["type"] != \
-                                    str(str):
-                                param["value"] = jsonpickle.loads(
-                                    content["value"])
-                            else:
-                                param["value"] = content["value"]
-                            io.parameters[name[13:]] = ArgoObjectDict(param)
+                        try:
+                            download_artifact(art, path=tmpdir)
+                            fs = os.listdir(tmpdir)
+                            assert len(fs) == 1
+                            with open(os.path.join(tmpdir, fs[0]), "r") as f:
+                                content = jsonpickle.loads(f.read())
+                                param = {"name": name[13:],
+                                         "save_as_artifact": True}
+                                if "type" in content:
+                                    param["type"] = content["type"]
+                                if "type" in content and content["type"] != \
+                                        str(str):
+                                    param["value"] = jsonpickle.loads(
+                                        content["value"])
+                                else:
+                                    param["value"] = content["value"]
+                                io.parameters[name[13:]
+                                              ] = ArgoObjectDict(param)
+                        except Exception:
+                            pass
 
     def modify_output_parameter(
             self,

src/dflow/utils.py

@@ -35,6 +35,8 @@
 def download_artifact(
         artifact,
         extract: bool = True,
+        sub_path: str = None,
+        slice: int = None,
         **kwargs,
 ) -> List[str]:
     """
@@ -51,34 +53,45 @@ def download_artifact(
         bucket_name: bucket name for Minio
     """
     if hasattr(artifact, "s3"):
-        if hasattr(artifact, "archive") and hasattr(artifact.archive, "none")\
-                and artifact.archive.none is not None:
-            path = download_s3(key=artifact.s3.key, recursive=True, **kwargs)
-        else:
-            path = download_s3(key=artifact.s3.key, recursive=False, **kwargs)
-            if path[-4:] == ".tgz" and extract:
-                tf = tarfile.open(path, "r:gz")
-                with tempfile.TemporaryDirectory() as tmpdir:
-                    tf.extractall(tmpdir)
-                    tf.close()
-
-                    os.remove(path)
-                    path = os.path.dirname(path)
-
-                    # if the artifact contains only one directory, merge the
-                    # directory with the target directory
-                    ld = os.listdir(tmpdir)
-                    if len(ld) == 1 and os.path.isdir(os.path.join(tmpdir,
-                                                                   ld[0])):
-                        merge_dir(os.path.join(tmpdir, ld[0]), path)
-                    else:
-                        merge_dir(tmpdir, path)
-
-        remove_empty_dir_tag(path)
-        return assemble_path_list(path, remove=True)
+        key = artifact.s3.key
+    elif hasattr(artifact, "key"):
+        key = artifact.key
     else:
         raise NotImplementedError()
 
+    if slice is not None:
+        sub_path = path_list_of_artifact(artifact)[slice]
+
+    if sub_path is not None:
+        key = key + "/" + sub_path
+        if "path" in kwargs:
+            kwargs["path"] = os.path.join(kwargs["path"],
+                                          os.path.dirname(sub_path))
+        else:
+            kwargs["path"] = os.path.join(".", os.path.dirname(sub_path))
+
+    path = download_s3(key=key, recursive=True, **kwargs)
+    if key[-4:] == ".tgz" and extract:
+        path = os.path.join(path, os.path.basename(key))
+        tf = tarfile.open(path, "r:gz")
+        with tempfile.TemporaryDirectory() as tmpdir:
+            tf.extractall(tmpdir)
+            tf.close()
+
+            os.remove(path)
+            path = os.path.dirname(path)
+
+            # if the artifact contains only one directory, merge the
+            # directory with the target directory
+            ld = os.listdir(tmpdir)
+            if len(ld) == 1 and os.path.isdir(os.path.join(tmpdir, ld[0])):
+                merge_dir(os.path.join(tmpdir, ld[0]), path)
+            else:
+                merge_dir(tmpdir, path)
+
+    remove_empty_dir_tag(path)
+    return assemble_path_list(path, remove=True)
+
 
 def upload_artifact(
         path: Union[os.PathLike, List[os.PathLike], Set[os.PathLike]],