How can an Executor communicate task failure to Prefect?

[jbweston]

Hi! First off thanks for making Prefect, it's an awesome project.

I have a question about how prefect handles a certain failure mode.

My understanding from reading the FlowRunner.run implementation is that (assuming there's no mapped edges or anything) it essentially does a bunch of calls to Executor.submit, submitting the tasks in topological order without waiting for any of them to finish, and relying on the Executor to manage task dependencies (as indicated in the README).

FlowRunner.run then does a single call to Executor.wait on all the futures that it needs to determine the final flow state.

My question is: how can the Executor communicate to Prefect that a task was unable to run for some reason?

For example, the Dask Executor makes a call to client.gather, and this method will AFAIK raise an exception if one of the gathered Dask tasks fail too many times. This could happen if the usercode (Task.run) fails in a way that Prefect does not handle (e.g. Segfault, SIGKILL, OOM, ...).

Another example would be a hypothetical executor that runs jobs on some cloud computing platform "directly". If you run out of money / allotted computing time for the month then how can the executor communicate an OutOfMoney error to Prefect?

It seems to me that API specified in the Executor README simply does not allow for this possibility.

[cicdw]

Hi @jbweston - this is such a great question! You are correct that there isn't really a first-class way to capture this failure mode using Prefect Core alone. This is precisely where Prefect Server / Cloud shine - in that case, if any tasks are mid run and killed, they will stop sending a heartbeat to the API. The Zombie Killer will eventually mark such a task as "Failed", from which point either the full flow run will resume (via Lazarus) or retry (if the task has retry settings configured).

[jbweston]

@cicdw thanks for the rapid response!

I'll be sure to check out the capabilities of Prefect Server.