When using the training operator and PyTroch elastic with Flyte, how does Flyte decide which pod's error to propagate at the task level? Say I have a 2 node PyTorch elastic training. And say both pods have a failure reported by PyTorch elastic. Which one propagates at the Flyte task level?

pod with rank-0

That doesn’t match what I’m observing.

aaha, let me share the logic https://github.com/flyteorg/flytekit/blob/16ad748c4cf40459c0b316db12cda2f1db81520c/plugins/flytekit-kf-pytorch/flytekitplugins/kfpytorch/task.py#L461-L470 We do use rank-0

I took rank 0 to mean PyTorch rank 0. I think you mean task with index 0. With c10d rdvz, rank 0 and task 0 are not the same.

🤷‍♂️

@freezing-airport-6809 The code you pointed above is the happy case. I was asking about what happens when there are multiple nodes that throw an exception from elastic_launch.

but there should only ever be one rank-0 or 0 right?

Here is the code that handles error cases:

try:
            out = elastic_launch(
                config=config,
                entrypoint=launcher_target_func,
            )(*launcher_args)
        except ChildFailedError as e:
            _, first_failure = e.get_first_failure()
            if is_recoverable_worker_error(first_failure):
                raise FlyteRecoverableException(e.format_msg())
            else:
                raise RuntimeError(e.format_msg())
        except SignalException as e:
            logger.exception(f"Elastic launch agent process terminating: {e}")
            raise IgnoreOutputs()

It assumes that elastic launch either throws ChildFailedError or SignalException. Here is what happens: • We have 2 node training (0-7 ranks on one node, 8-15 ranks on another node)

i do not think it reports the first error you re right Bugra

It doesn’t return the error collected on rank 0 based on my reading of the code, but I might be mistaken. SignalExceptions are ignored, and if there is only one ChildError across nodes, then all is well. If there are multiple exceptions across nodes, then the handling is somewhere else, on the golang side I assume. We have training runs with 128 nodes or more. It is not uncommon for elastic agent to throw various exceptions on multiple nodes. Reporting the first error based on timestamp is almost always reliable. Would be easy to support, but don’t know where to look yet, as the PyTorch plugin doesn’t seem to deal with the multi node error propagation.

@fierce-oil-47448 you are correct that we don't properly handle this, we discussed this back in this PR 😿 Letting the worker pods (as opposed to the backend) figure out the exception with the earliest time stamp would require opening a new distributed process group + communicating over torch dist. I feel this would be very brittle, especially since some workers might already be down and not able to join the rendezvous. Instead, I think we need to modify flytekit's pod entrypoint to allow choosing different error pb file names depending in this case on the group rank. (Maybe the entrypoint "can ask" the task whether it wants to customize the error file name?) Under the respective bucket uri there would then be multiple error files

error-0.pb

,

error-1.pb

, ... and the backend plugin would need to go through them and pick the one with the earliest time stamp. It would be nice if there was a general solution for plugins to be able to have multiple error files as torch is not the only plugin with this conceptional problem. Would you be willing to colaborate on this with me @fierce-oil-47448? @freezing-airport-6809 maybe we can discuss in the next contrib sync how exactly this should be done.

Hi @cool-lifeguard-49380. Yes, I would be happy to collaborate. And I agree with the technical direction. It would be very fragile to try to do this at the task pod level. I would love to understand how this works currently as well. Are different task pods overwriting the same error file today? In my minimal reproduce, we have 2 pods, each getting an exception from its elastic_launch. How does Flyte decide to report one or the other.

I never debugged this to the end but I would assume they overwrite each other and there is a race condition 😭

Sure, will look into this.

Thank you for looking into this! I replied in the PR.

@cool-lifeguard-49380 I think timestamp is needed in

ContainerError

(here) and not in

Error

, as the former is what entry point uses to capture the error. Everything you described makes a lot of sense. These changes look reasonably straightforward to me. I can help with the implementation, but I'd be slow (due to other priorities at work). Also, I'll need help with a set up where I can try backend changes (never done that).

Also, I'll need help with a set up where I can try backend changes (never done that).

Happy to help, in which time zone are you?

I'm PST

I’ll dm you

The thing is many folks are out this week, they will be back next week

Sorry I missed this. Let me catch up.

The thing is many folks are out this week, they will be back next week

No time pressure

Let me summarize the discussion up to this point. The proposed solution handles error propagation in Flyte's multi-node PyTorch elastic training by using multiple error files and selecting the earliest error based on timestamps. The breakdown: 1. Add Timestamp to Error Messages: Introduce an optional timestamp field to the error proto message to enable sorting errors by their occurrence time. a. The

error.pb

protobuf message is an ErrorDocument b. where we add the timestamp? in

ContainerError

or

ErrorDocument

c. Do we have to worry about time drift? (different containers having different clocks) 2. Custom Error File Naming: Modify the Flytekit pod entrypoint to generate multiple error files (e.g., error-0.pb, error-1.pb) based on the group rank. This avoids overwriting errors from different pods. a. in https://github.com/flyteorg/flytekit/pull/2607/ we're generating random suffixes instead 3. Proposed backend Changes: a. Implement a new output reader (e.g., MultiErrorFileRemoteFileOutputReader) in Flyte's plugin manager to read these multiple error files. This reader will search for files matching the pattern (e.g., error-*.pb) and select the one with the earliest timestamp. i. Flyte turns

ErrorDocument

into an

ExecutionError

here and this is what ends up being reported downstream. We can leverage that in the implementation of

ReadError

in this new output reader. Not sure if we have to turn this into a policy (i.e. does it always make sense to pick the first error sorted by timestamp or should that be configurable?) b. Task Pod Modifications: Ensure the task pods write unique error files based on the group rank or replica index. i. If we want to encode the rank in the error message, the PyTorch operator exposes the rank as an environment variable (see https://github.com/kubeflow/training-operator/blob/master/pkg/controller.v1/pytorch/envvar.go#L89-L96). * This

PET_*

business seems to be a pytorch quirk introduced some time ago (I couldn't find what

PET

stands for) c. Additional updates to the Training Operator might be needed to ensure the replica index is available in the environment variables.

> The proposed solution handles error propagation in Flyte's multi-node PyTorch elastic training by using multiple error files and selecting the earliest error based on timestamps. Yes but if I'm not mistaken this is not only for torch elastic tasks but also for other distributed training plugins where each worker pod might write the error.pb. It's certainly also the case for non-elastic pytorch trainings and potentially for tensorflow, mpi, ...? (I don't know whether the latter come with a mechanism that propagates any errors to a dedicated pod which is the only one to write the error.pb) I think it would be good if we build a general solution for all plugins that rely on multiple pods - nothing too specific to elastic. > c. Do we have to worry about time drift? (different containers having different clocks) I think we can deal with this in a best-effort manner. If time drift might happen, causing us to report the wrong error, one can still combine the logs of all pods in stackdriver logs, filter by severity, and search for the first error(s). But it will save a lot of time, if we can get it right often enough ^^ (Or is there a low hanging fruit for how we can sync the clocks?) > generate multiple error files (e.g., error-0.pb, error-1.pb) based on the group rank. In the case of elastic, we unfortunately don't have easy access to the group rank in the entrypoint. It is not set as an env var, I ran a test yesterday. What you linked here, is only set for non-elastic pytorch tasks. I think it doesn't matter though, if Flyte would inject the pod name as an env var (or we use the downstream api), we could just use that - in the end the group rank was just a proposal to allow easy identification in the flyte console which pod died first. We can use the pod name for that - which would work for all distributed plugins regardless of what concepts they use for ranks etc.

Thanks @cool-lifeguard-49380 for putting up the RFC ❤️. I've added some comments (backward compatibility is an area that may need some work imo).

Thank you for the comments 🙏

@cool-lifeguard-49380 Here is what I'm thinking in terms of the implementation plan: • Set environment variables for FLYTE_INTERNAL_POD_NAME and FLYTE_INTERNAL_ERROR_PROPAGATION • Add entry point support for multi error file upload based on above • Add exception timestamps and timestamp handling in PyTorch plugin • Add MultiErrorFileRemoteFileOutputReader • Add configuration support for enabling MultiErrorFileRemoteFileOutputReader for PyTorch plugin (and other relevant ones) Let me know what you think.

Fully agree 👍

• Add entry point support for multi error file upload based on above

This requires a flyteidl change to add timestamp to error document.

Can I get an approval on this https://github.com/flyteorg/flyte/pull/5616 I think there was a small test file change on the spark support, that still needs someone to approve cc: @high-accountant-32689

@fierce-oil-47448, left one minor comment.

Thanks @high-accountant-32689, addressed

@fierce-oil-47448, I think you committed a few changes by mistake in your last update.

Really sorry for that @high-accountant-32689. Let me fix. Didn't notice I had a dirty branch.

go doesn't like python comments 🙂 Fixed in https://github.com/flyteorg/flyte/pull/5616/commits/63a1faaba0b9b0dc642938bdad484d1fe412a630. Should merge as soon as tests pass.

Haha, Tx. @high-accountant-32689 Would appreciate a look at https://github.com/flyteorg/flyte/pull/5674/files when you get a chance as well. This adds listing support to storage interfaces. I added it for stow storage. One specific feedback I'm looking for is to understand if this needs to be added for all storage kinds.

cool, I will take a look later today.

@high-accountant-32689 One more approval here https://github.com/flyteorg/flyte/pull/5616/files#diff-4798c57d1e223aea7c8724d4401e259fe395da096b0eed7766493e8ae342a336 Had to touch another test file. All tests look healthy now.