This is something I've seen a few times in different cases but I'm running the OOM example and I i'm seeing parent/driver task fail with

plugin "armada" failed with system error 4 times: failed to read data from dataDir [s3://<redacted>/flytesnacks/development/r2bwqqbc2tr2mmcxmw55/a0/1/outputs.pb]. Error: path:s3://<redacted>/flytesnacks/development/r2bwqqbc2tr2mmcxmw55/a0/1/outputs.pb: not found

import asyncio

import flyte
import flyte.errors

env = flyte.TaskEnvironment(name="fail", resources=flyte.Resources(cpu=1, memory="250Mi"))

@env.task
async def oomer():
    large_list = [0] * 100000000
    print(len(large_list))

@env.task
async def always_succeeds() -> int:
    await asyncio.sleep(1)
    return 42

@env.task
async def parent() -> int:
    try:
        await oomer()
    except flyte.errors.OOMError as e:
        print(f"Failed with oom trying with more resources: {e}, of type {type(e)}, {e.code}")
        try:
            await oomer.override(resources=flyte.Resources(cpu=1, memory="1Gi"))()
        except flyte.errors.OOMError as e:
            print(f"Failed with OOM Again giving up: {e}, of type {type(e)}, {e.code}")
            raise e
    finally:
        await always_succeeds()

I expect that it would upload the results of

always_succeed

?

yeah it should… weird

To be fair we're running this through our Armada plugin that we've ported from V1 but it works in most cases

cc @glamorous-carpet-83516 @faint-pilot-24699 FYI. We can test again on a vanilla Flyte cluster, but I believe all of these examples work

Just a snippet.

func (p Executor) Handle(ctx context.Context, tCtx core.TaskExecutionContext) (core.Transition, error) {
	state, err := p.getOrCreateJob(ctx, tCtx)

	// Return if there was an error
	if err != nil {
		return core.UnknownTransition, err
	}

	// If no error, then infer the new CurrentPhase from the various states
	phaseInfo := mapExecutionStateToPhaseInfo(state, p.cfg, p.clock)

	// Ensure the output reader is set. This is similar to what the k8s pod plugin management does
	if phaseInfo.Phase() == core.PhaseSuccess {
		opReader := ioutils.NewRemoteFileOutputReader(ctx, tCtx.DataStore(), tCtx.OutputWriter(), 0)
		putErr := tCtx.OutputWriter().Put(ctx, opReader)
		if putErr != nil {
			return core.UnknownTransition, putErr
		}
	}

	return core.DoTransition(phaseInfo), nil
}

In this case the phase info is success (the parent task armada jobs exits with code 0 / success) and it should be configuring this correctly.

mind filing an issue? github.com/flyteorg/flyte/issues

Sure

nice! wait, so what changed?

I changed our plugin implementation to treat OOM from our compute scheduler as a retryable user error