Add support for minio copy workflow though API

fridge-job-api/app/main.py

@@ -381,6 +381,45 @@ async def get_object(
     return minio_client.get_object(bucket, file_name, target_file, version)
 
 
+# Trigger Argo workflow
+@app.post("/object/move/{file_name}", tags=["s3"])
+async def move_object(
+    bucket: str,
+    file_name: str,
+    version: str = None,
+    verified: Annotated[bool, "Verify the request with basic auth"] = Depends(
+        verify_request
+    ),
+) -> dict:
+    r = requests.post(
+        f"{ARGO_SERVER}/api/v1/workflows/argo-workflows/submit",
+        verify=VERIFY_TLS,
+        headers={"Authorization": f"Bearer {argo_token()}"},
+        data=json.dumps(
+            {
+                "resourceKind": "WorkflowTemplate",
+                "resourceName": "minio-data-copy",
+                "submitOptions": {
+                    "generateName": "minio-copy-",
+                    "parameters": [
+                        f"bucket={bucket}",
+                        f"file={file_name}",
+                    ],
+                },
+            }
+        ),
+    )
+
+    if r.status_code != 200:
+        raise HTTPException(
+            status_code=r.status_code, detail=parse_argo_error(r.json())
+        )
+    return {
+        "status": r.status_code,
+        "response": r.json(),
+    }
+
+
 @app.post("/object/bucket", tags=["s3"])
 async def create_bucket(
     bucket_name: str,

infra/fridge/__main__.py

@@ -154,6 +154,26 @@ def patch_namespace(name: str, pss: PodSecurityStandard) -> NamespacePatch:
     ),
 )
 
+argo_workflow_templates = ConfigFile(
+    "argo-workflow-templates",
+    file="./k8s/argo_workflows/templates.yaml",
+    transformations=[
+        lambda obj, opts: (
+            obj["spec"]["templates"][0]["volumes"][0].update(
+                {
+                    "persistentVolumeClaim": {"claimName": "workflow-data-ingress"}
+                }  # Replace this with PVC created in block_storage.py
+            )
+            if obj["spec"]["templates"][0]["volumes"][0].get("name")
+            == "workflow-data-ingress"
+            else None
+        ),
+    ],
+    opts=ResourceOptions(
+        depends_on=[argo_workflows],
+    ),
+)
+
 if enable_sso:
     argo_workflows_rbac = components.WorkflowUiRbac(
         "argo-workflows-rbac",

infra/fridge/components/api_server.py

@@ -206,7 +206,7 @@ def __init__(
                                         name="token-vol",
                                         mount_path="/service-account",
                                         read_only=True,
-                                    )
+                                    ),
                                 ],
                             )
                         ],

infra/fridge/components/object_storage.py

@@ -120,7 +120,6 @@ def __init__(
                 "tenant": {
                     "name": "argo-artifacts",
                     "buckets": [
-                        {"name": "argo-artifacts"},
                         {"name": "ingress"},
                         {"name": "egress"},
                     ],

infra/fridge/components/workflow_server.py

@@ -1,6 +1,12 @@
 import pulumi
 from pulumi import ComponentResource, FileAsset, Output, ResourceOptions
-from pulumi_kubernetes.core.v1 import Namespace, Secret
+from pulumi_kubernetes.core.v1 import (
+    Namespace,
+    Secret,
+    PersistentVolumeClaim,
+    PersistentVolumeClaimSpecArgs,
+    VolumeResourceRequirementsArgs,
+)
 from pulumi_kubernetes.helm.v4 import Chart, RepositoryOptsArgs
 from pulumi_kubernetes.meta.v1 import ObjectMetaArgs
 

infra/fridge/k8s/argo_workflows/templates.yaml

@@ -0,0 +1,42 @@
+---
+apiVersion: argoproj.io/v1alpha1
+kind: WorkflowTemplate
+metadata:
+  name: minio-data-copy
+  namespace: argo-workflows
+spec:
+  entrypoint: pull-data-from-minio
+  templates:
+    - name: pull-data-from-minio
+      inputs:
+        parameters:
+          - name: bucket
+          - name: file
+      volumes:
+        - name: workflow-data-ingress
+          persistentVolumeClaim:
+            claimName: replace_me
+      container:
+        name: mc
+        image: minio/mc:latest
+        command: ["/bin/sh", "-c"]
+        args:
+          - |
+            mc --insecure alias set artifacts http://minio.argo-artifacts.svc.cluster.local $MINIO_ACCESS_KEY $MINIO_SECRET_KEY
+            mc cp artifacts/{{inputs.parameters.bucket}}/{{inputs.parameters.file}} /data/
+        env:
+          - name: MC_CONFIG_DIR
+            value: /tmp/.mc
+          - name: MINIO_ACCESS_KEY
+            valueFrom:
+              secretKeyRef:
+                name: argo-artifacts-minio
+                key: accesskey
+          - name: MINIO_SECRET_KEY
+            valueFrom:
+              secretKeyRef:
+                name: argo-artifacts-minio
+                key: secretkey
+        volumeMounts:
+          - name: workflow-data-ingress
+            mountPath: /data