https://outerbounds.com/ logo
Join Slack
Powered by
Hey all, I am currently facing a weird issue where...
# ask-metaflow
b
Hey all, I am currently facing a weird issue where running on argo workflows with more than 4 branching steps causes error like this, weirdly this is not happening in Step Functions, need some guidance on this thanks: split logic:
Copy code
@step
    def my_step(self):
        self.next(
            self.branch1
            , self.branch2
            , self.branch3
            , self.branch4
            , self.branch5
            , self.branch6
            , self.branch7
        )
error:
Copy code
Setting up task environment.
Downloading code package...
Code package downloaded.
Task is starting.
    Data missing:
    Some input datastores are missing. Expected: 7 Actual: 0
āœ… 1
tried updating to 
metaflow==2.12.3
but still facing this issue
šŸ‘€ 1
s
we are triaging this issue
šŸ‘ 1
b
trying to reproduce the issue now, @brief-kite-90012 can you give me some more info on the structure of your flow, as I'm not getting any failures with a top-level slit of 7 branches at least. ā€¢ are the 7 branches inside a nested split, or on the top level of a flow? ā€¢ assuming branches are not actually named "branch", are the names extremely long? ā€¢ just to confirm, less than 7 branches work fine with your argo deployment?
b
the dag flow looks something like this in general, seems like the first span is fine but the second starts failing:
b
hmm, running a flow with two stages of 7-splits now and still no failures on my end. Which version of Argo Workflows are you running on? Also just for validation purposes, if you have access to the Argo UI, could you check the workflow's failing join step if it is receiving the correct task-id's in its input-paths parameter? Should look something like this
b
the input-paths seems to be wrong here, im on argo v3.4.11
a
@brief-kite-90012 are you able to try it out with a newer version of argo?
it could be an issue with the argo deployment
b
sure but it might take me some time to upgrade the argo version, any recommended minimum version for it?
a
you could try to reproduce it inside the metaflow sandbox too - metaflow.org/sandbox
šŸ‘ 1
b
tried running a dummy flow that branches into 7 steps twice on the sandbox but my sandbox env ran into: 
503 Service Temporarily Unavailable
after triggering the job in argo, and seems like some of the steps are stuck
let me try from my end after upgrading the argo workflows version
šŸ‘ 1
think i figured out the reason for this issue, as we maintain a separate version of plugins for argo, after updating the changes from the latest metaflow version and deploying new version of argo workflows seemed to fixed it, thanks for the help: ā€¢ 
argo_workflows.py
ā€¢ 
argo_client.py
@ancient-application-36103 @bulky-afternoon-92433 sorry after some deeper digging seems like this issue still persist, it works fine on the first level where each branch step is just having 1 artifacts to join downstrea, but on the second split when each branch step has more than 1 artifacts to pass down, maybe like around 2-5 dfs that are passed to self, the join step input path starts acting weird and fails with the similar error log:
Copy code
Data missing:
    Some input datastores are missing. Expected: 7 Actual: 0
also seems like it is broken even on the 
workflow-template
level in argo workflows, the 
input-paths
parameter for these steps are long strings of ids that doesnt make sense unlike those comma separated jinjja values string
a
can you try with the sandboxes now? they should be stable at the moment. i believe there is some customization in your argo deployment that is hashing the values which metaflow doesn't expect
b
@square-wire-39606 seems like its coming from here where the hashing values is happening when the 
input-paths
are too long (>500), https://github.com/Netflix/metaflow/blob/master/metaflow/plugins/argo/argo_workflows.py#L899-L904, and i've tested decompressing it using 
decompress_list
function in utils the return list looks fine, not sure what is happening here
was able to reproduce in sandbox using this flow on argo:
Copy code
from metaflow import step, FlowSpec, conda_base, kubernetes


class HelloFlow(FlowSpec):
    @kubernetes
    @step
    def start(self):
        print("Starting šŸ‘‹")
        self.next(
            self.some_very_long_step_name_here,
            self.another_long_name_of_step_here,
            self.yes_another_long_long_step_name_at_here,
            self.do_not_question_how_i_name_my_step,
            self.yes_yes_long_long_name_again_here,
            self.ran_out_of_idea_long_step_name,
            self.last_dummy_long_long_long_name_here,
            )

    @kubernetes
    @step
    def some_very_long_step_name_here(self):
        from sklearn.datasets import load_iris
        iris = load_iris()
        self.some_very_long_step_name_here1 =  iris
        self.some_very_long_step_name_here2 =  iris
        self.some_very_long_step_name_here3 =  iris
        self.some_very_long_step_name_here4 =  iris
        self.next(self.join)

    @kubernetes
    @step
    def another_long_name_of_step_here(self):
        from sklearn.datasets import load_iris
        iris = load_iris()
        self.another_long_name_of_step_here1 =  iris
        self.another_long_name_of_step_here2 =  iris
        self.another_long_name_of_step_here3 =  iris
        self.another_long_name_of_step_here4 =  iris
        self.next(self.join)

    @kubernetes
    @step
    def yes_another_long_long_step_name_at_here(self):
        from sklearn.datasets import load_iris
        iris = load_iris()
        self.yes_another_long_long_step_name_at_here1 =  iris
        self.yes_another_long_long_step_name_at_here2 =  iris
        self.yes_another_long_long_step_name_at_here3 =  iris
        self.yes_another_long_long_step_name_at_here4 =  iris
        self.next(self.join)

    @kubernetes
    @step
    def do_not_question_how_i_name_my_step(self):
        from sklearn.datasets import load_iris
        iris = load_iris()
        self.do_not_question_how_i_name_my_step1 =  iris
        self.do_not_question_how_i_name_my_step2 =  iris
        self.do_not_question_how_i_name_my_step3 =  iris
        self.do_not_question_how_i_name_my_step4 =  iris
        self.next(self.join)

    @kubernetes
    @step
    def yes_yes_long_long_name_again_here(self):
        from sklearn.datasets import load_iris
        iris = load_iris()
        self.yes_yes_long_long_name_again_here1 =  iris
        self.yes_yes_long_long_name_again_here2 =  iris
        self.yes_yes_long_long_name_again_here3 =  iris
        self.yes_yes_long_long_name_again_here4 =  iris
        self.next(self.join)

    @kubernetes
    @step
    def ran_out_of_idea_long_step_name(self):
        from sklearn.datasets import load_iris
        iris = load_iris()
        self.ran_out_of_idea_long_step_name1 =  iris
        self.ran_out_of_idea_long_step_name2 =  iris
        self.ran_out_of_idea_long_step_name3 =  iris
        self.ran_out_of_idea_long_step_name4 =  iris
        self.next(self.join)

    @kubernetes
    @step
    def last_dummy_long_long_long_name_here(self):
        from sklearn.datasets import load_iris
        iris = load_iris()
        self.last_dummy_long_long_long_name_here1 =  iris
        self.last_dummy_long_long_long_name_here2 =  iris
        self.last_dummy_long_long_long_name_here3 =  iris
        self.last_dummy_long_long_long_name_here4 =  iris
        self.next(self.join)

    @kubernetes
    @step
    def join(self, inputs):
        print("joining")
        self.next(self.end)

    @kubernetes
    @step
    def end(self):
        print("Done! šŸ")


if __name__ == "__main__":
    HelloFlow()
šŸ‘ 1
image.png
b
thank you for figuring out the cause, and the working example. this should be of great help with fixing the issue. I'll look into what has introduced the bug and a possible fix during today.
b
thanks @bulky-afternoon-92433 !
b
https://github.com/Netflix/metaflow/releases/tag/2.12.6 went out yesterday which contains a fix for this issue.
b
thanks! will test it from my end
b
the cause was actually that the list compression is not compatible with Argo workflows to begin with, as Argo replaces values for tags 
{{}}
during template evaluation/pod scheduling, and we perform the list decompress during runtime, which just results in a bunch of tokens without actual values
the release disables compression of input-paths completely on Argo, as this should realistically not be required for static splits
b
right, just a followup question, is there a limit of the input-paths which kinda limits how many splits or artifacts we can have?
b
the main limiting factor is the size of 
ARGO_TEMPLATE
env var on the pods where argo keeps the complete workflow manifest with values. There was an issue in the past with extremely wide foreach steps running into the limit, but even these could hold ~1k step names in there, which is why I'm expecting that any static flow written by hand should not be running against any of the limits.
šŸ‘ 1
(moreover the foreach split issue has been completely alleviated now so they are not eating space in the env var either)
šŸ‘ 1
there have been some efforts on Argos side as well to support massive workflow templates as of late. if I recall one proposal was to try offload the template to configmap after a threshold, but I haven't kept up with the recent developments to see if they released anything for this yet
šŸ‘ 1
18 Views
Open in Slack
PreviousNext
Powered by