Fan-out with independent Volume for each branch

[eduardohenriquearnold]

Hi, I would like to create a workflow with a fan-out where subsequent tasks share a volume (unique to each branch).

The DAG would look something like:

             Task A
               │
   ┌───────────┼───────────┐
   │           │           │
   ▼           ▼           ▼
Task B1     Task B2     Task B3
   │           │           │
   ▼           ▼           ▼
Task C1     Task C2     Task C3

For example, Tasks B1, C1 share Volume 1; Tasks B2, C2 share Volume 2, etc.

I know I can define a volume to be used by two subsequent tasks as follows:

from hera import Task, Workflow, Volume,
with Workflow("volumes-pvc-hera-", generate_name=True) as w:
    volumes = [Volume(size="1Gi", mount_path="/mnt/vol")]
    generate = Task(
        "generate",
        image="docker/whalesay:latest",
        command=["sh", "-c"],
        args=[
            "echo generating message in volume; cowsay hello world | tee /mnt/vol/hello_world.txt"
        ],
        volumes=volumes,
    )
​
    print = Task(
        "print",
        image="alpine:latest",
        command=["sh", "-c"],
        args=[
            "echo getting message from volume; find /mnt/vol; cat /mnt/vol/hello_world.txt"
        ],
        volumes=volumes,
    )
    generate >> print

As noted above, one needs the reference to the volume for each subsequent task. My question is: how to generalise the example above for the fan-out case? Thanks!

[eduardohenriquearnold]

I have also a minimal example below:

def A():
    import json
    import sys

    # this can be anything! e.g fetch from some API, then in parallel process all entities; chunk database records
    # and process them in parallel, etc.
    json.dump([i for i in range(10)], sys.stdout)


def B(value: int):
    print(f"Received value: {value}!")
    with open(f"/mnt/vol/{value}.txt", 'w') as f:
        f.write("Values!")

def C(value: int):
    import pathlib
    # do something with value and files in /mnt/vol

    print("Files in /mnt/vol:")
    print(tuple(pathlib.Path("/mnt/vol").iterdir()))


# assumes you used `hera.set_global_token` and `hera.set_global_host` so that the workflow can be submitted
with Workflow("dynamic-fanout-", generate_name=True) as w:
    volumes = [Volume(size="1Gi", mount_path="/mnt/vol")]
    A_task = Task("a", A)
    B_task = Task("b", B, with_param=A_task.get_result(), volumes=volumes)
    C_task = Task("c", C, with_param=A_task.get_result(), volumes=volumes)

    A_task >> B_task >> C_task

When I execute this, the shared volume is the same across the branches, and Task C only starts once all Task B's have completed. My questions are:

1. How to make Task C start immediately once it's corresponding B task completes (and not wait for all B tasks to complete).
2. How to make the volume independent for different branches of tasks B, C. So each Task C should only see a single file in /mnt/vol

[eduardohenriquearnold]

I think I was able to solve question 1 using a parallel DAG as follows

def A():
    import json
    import sys

    # this can be anything! e.g fetch from some API, then in parallel process all entities; chunk database records
    # and process them in parallel, etc.
    json.dump([i for i in range(10)], sys.stdout)


def B(value: int):
    import pathlib
    pathlib.Path("/mnt/vol").mkdir(parents=True, exist_ok=True)
    print(f"Received value: {value}!")
    with open(f"/mnt/vol/{value}.txt", "w") as f:
        f.write("Values!")


def C(value: int):
    import pathlib

    # do something with value and files in /mnt/vol

    print("Files in /mnt/vol:")
    print(tuple(pathlib.Path("/mnt/vol").iterdir()))


# assumes you used `hera.set_global_token` and `hera.set_global_host` so that the workflow can be submitted
with Workflow("dynamic-fanout-", generate_name=True, service=service, service_account_name=sa) as w:
    with DAG("pipeline", inputs=[Parameter("value")], outputs=[Parameter("value")]) as pipeline:
        volumes = [Volume(size="1Gi", mount_path="/mnt/vol")]
        B_task = Task("b", B, inputs=[pipeline.get_parameter("value")], volumes=volumes)
        C_task = Task("c", C, inputs=[pipeline.get_parameter("value")], volumes=volumes)
        B_task >> C_task
        pipeline.outputs = [C_task.get_result_as("output")]

    A_task = Task("a", A)
    parallel_task = Task("parallel-pipeline", dag=pipeline, with_param=A_task.get_result())
    A_task >> parallel_task

Still, the same volume is shared across all branches, so question (2) remains.

[flaviuvadan]

Hi @eduardohenriquearnold! Thanks for the question :) I think this is actually a limitation of Argo Workflows. When volumes are created dynamically they are set at the workflow level. So, Argo tells K8S about the PV/PVCs it needs to create. Those PV/PVCs, depending on your cluster setup, either instantly create the volumes when the workflow is created or when the pod is created/ready to mount the volume. To my knowledge, there's no way for Argo to know how many tasks there will be in the fanout, so it does not know what volumes to create. Now, I think there might be ways around this. I have not experimented with this so take it with a grain of salt. This idea is also conditioned on your application knowing how many tasks there will be a fanout over:

• take the N volumes required based on the AOT known fanout elements and create that many volumes in the workflow
• set a K8S storage class on those volumes that says "only create those volumes when a pod is ready to mount them". We use this at Dyno Therapeutics by having a custom storage class in K8S with VolumeBindingModeset to WaitForFirstConsumer and that storage class is set on volumes by default
• pass the volume IDs as a parameter to the main task that fans out
• perform a zip between the fanout elements and the volume IDs
• make the fanout task mount the volume identified by ID parameters.volume.volume_id or something like that

I think the above might work? I don't think this is fully possible in Hera at the moment because we had an accidental regression and we cannot set volumes at the workflow level anymore. However, it will for sure be possible in the upcoming Hera V5, where you will get full feature parity with Argo Workflows + Events!

[flaviuvadan]

Thanks for outlining the above @Mastergalen! I think another option for cleaning up is to use exit handlers, either at the task/step level or the workflow level. I think these will run irrespective of result! Can also run as a normal step but then make the step "continue on fail"

[Mastergalen]

@flaviuvadan Indeed, that is what I have been trying with lifecycle hooks, but ran into issues with the workflow.status not being accessible in template-level hooks haha 😅 .

See argoproj/argo-workflows#11269

[flaviuvadan]

That's annoying :( thank you for explaining!

[Muralikrishna2009]

@Mastergalen In the below - The volume doesnot work in Container(). Can you please verify it once. May be a bug?

volumes=m.Volume(
name="workdir",
persistent_volume_claim={
"claim_name": "{{inputs.parameters.pvc-name}}"
},

whalesay = Container(
    name="whalesay",
    image="docker/whalesay:latest",
    command=["sh", "-c"],
    inputs=[
        Parameter(name="pvc-name"),
    ],
    args=[
        "echo generating message in volume {{inputs.parameters.pvc-name}}; cowsay hello world | tee /mnt/vol/hello_world.txt"
    ],
    volume_mounts=[m.VolumeMount(name="workdir", mount_path="/mnt/vol")],
    volumes=m.Volume(
        name="workdir",
        persistent_volume_claim={
            "claim_name": "{{inputs.parameters.pvc-name}}"
        },
    ),
)

[Muralikrishna2009]

@eduardohenriquearnold - Can you please share how you made this work, will be very useful for my fanout use case
An example will help