LOGMLE - Fix argo metaflow retry integration

[dhpikolo]

Problem

https://github.com/deliveryhero/logistics-ds-metaflow-ext/issues/108#issue-2832910761

Solution

This PR,modifies the step cli command to infer retry_count from flow_datastore class. Looks like this class, holds all the information about underlying datastore and run artifiacts.

• Adds a method that infers latest done attempt of a task
• Remove MF_ATTEMPT from mflog.save_logs python script.

New Behaviour

Previously argo retry --node <> would overwrite previously run attepmts, Now it would add artifacts as it was new attempt.

Fixes: https://github.com/deliveryhero/logistics-ds-metaflow-ext/issues/108

Slack Discussion: https://outerbounds-community.slack.com/archives/C020U025QJK/p1737483912784969

Testing and Limitations

• Cannot Retry more than 6 times.

Use this branch on dummy model to test this:

• https://github.com/deliveryhero/logistics-ds-dummy-project/pull/145

Validation & QA

• LOGMLE - Fix argo metaflow retry integration #1 (comment))

Documentation:

• https://github.com/deliveryhero/logistics-ds-metaflow-ext/pull/110

[dhpikolo]

will remove it after testing complete

[dhpikolo]

this is needed because, both kubernetes and argo options run mflog.save_logs after the task is completed. which will upload task_stdout, task_stderr logs to s3 and reflects on the metaflow UI.

Example run without this change, have no logs for argo retried attempts. ie Attempt 3, 4 & 5

[dhpollack]

Why are we explicitly casting this to an int? Why don't we do this in the function?

[dhpikolo]

yeah you are right, i dont think we need to cast it anymore, the function returns a count.

I had it this way to keep the changes minimal.

[dhpikolo]

function returning None integrates well here, since get_task_data_store(..., attempt=None) will not be broken.

https://github.com/Netflix/metaflow/blob/5c960eaff1ae486f503b37177f03cc1419b5571d/metaflow/datastore/flow_datastore.py#L206

[dhpollack]

Where does this come from? I am wondering if we can leave all the code in this file alone and try to set the MF_ATTEMPT with the value that we want.

[dhpikolo]

MF_ATTEMPT comes from the argo workflow template container command.

[dhpollack]

Ok, one last "idea" because I feel like I'm being negative, but don't want to be.

What if instead of creating more retries, we simply set the attempt number to the last available attempt and overwrite all the data on this attempt?

Pros:

• existing code should work as expected
• won't increase the max attempt number

Cons:

• will overwrite the data of the last attempt (which presumably failed so you don't need it anyway)
• there will only be one attempted retry because the next attempt would hit the maximum.

We might even be able to do this in the operator by rewriting the command when we see the annotation that the workflow is a retry.

Even if we try this, we can still have this as a stop gap solution.

[dhpikolo]

I kind of looked into this idea already.

one of the difficulty was:

we see the annotation that the workflow is a retry.

there is no annotation to spot if a workflow is retried.

[dhpikolo]

but you just gave me a hint 😄

if we can change {{retries}} --> {{retries}} + {{limit}}, we could actually make argo retry button work for once.

EDIT: aah no sorry. i was wrong

[dhpikolo]

#1 (comment)

{{retries}} --> {{retries}} + if this_is_retried_flow {{limit}} else 0

[colebaileygit]

will overwrite the data of the last attempt (which presumably failed so you don't need it anyway)
I think we do want to keep data so that we can debug why it failed, but if it failed multiple times overwriting just 1 attempt isn't the end of the world.

Since one of the metaflow maintainers' main concerns is breaking "immutability" so I think overwriting data would stop this from getting merged upstream. The other concern is making changes to the core task logic which has a risk of breaking other systems.

[colebaileygit]

Shouldn't the default be -1 or None?

[dhpikolo]

i will make the default None since, it makes more sense when no done attempts were found.

Also: #1 (comment)

[dhpikolo]

Tracking discussion with metaflow maintainers in a separate PR

• [WIP] feature: support argo retry Netflix/metaflow#2277

[dhpikolo]

FYI, @colebaileygit

MAX_ATTEMPTS vs max_user_code_retries

i think, max_user_code_retries is number of times a task can be retried. this value is same as the value that the retry decorator contains.

sample workflow-template

Where as MAX_ATTEMPTS is number of total attempts that a task can be ran. [here], which is basically the number of retries + initial run

[dhpikolo]

Testing

Hitting argo retry button on the failed argo-workflow

• Linear flow ✅
• Branched flow ✅
• Resume argo retried flow ✅
• Does argo retry works without metaflow retry ? ✅
• Does argo retried attempts appear on metaflow UI ? ✅
• Metaflow API integration ✅

In [1]: from metaflow import Step, current, namespace, Task

In [2]: namespace(None)

In [3]: t = Task("DummyProjectFlow/argo-dummyproject.user.j.kollipara.dummyprojectflow-z6mfl/train_model/t-8586c205")

In [4]: t.data
Out[4]: <MetaflowData: country_code, test_data, train_data, countries, region, country_codes, execution_date, regions, rev_map_regions, skip_create_table, table_run_id, auto_deploy, buffer_ratio, build_static_image, ccs, code_version_override, components, dump_model, ed, merge_strategy, name, register_model, stamps, tri>

In [5]: t.data.test_data
Out[5]: 
   country_code        date  orders
53           at  2025-02-10   27967
54           at  2025-02-11   28230
55           at  2025-02-12   26754
56           at  2025-02-13   30494
57           at  2025-02-14   36382
58           at  2025-02-15   30812
59           at  2025-02-16   38858

In [6]: t.data.country_code
Out[6]: 'at'

In [7]: t.current_attempt
Out[7]: 5

• Downstream task fetches artifacts from the latest attempt ✅

[dhpikolo]

Infer the retry count only if previously done attempts were detected.