Also, if the function is intended to clean up resources, I'd recommend checking out this article.

For example, I might run some jobs on a Kubernetes cluster using individual flow runs, which run using

KubernetesRun

on different nodes:

def cleanup(flow, state):
   # cleanup resources in K8s
   # kubectl delete jobs --all --namespace $USER

with Flow("myflow", ..., on_failure=cleanup) a flow:
   # do some work, e.g. run various child flow runs on K8s
   create_flow_run("kubernetes_job")
   create_flow_run("kubernetes_job")
   # ...

Now, if I cancel the "main" flow named "myflow" in this example, I would like to run the

cleanup

function to make sure all resources are released. Does this make sense?

I imagine you could leverage that cleanup task with a state handler as well (although I haven't tried it personally, just a thought). The

on_failure

flow parameter included in the code snippet you provided is called when a flow enters a failed state, and not a

Cancelled

state. You could try creating your

resource_manager

task, passing it to a state handler function which checks to see if the the flow

is_finished

(finished captures the

Cancelled

state), and then passing the state handler to the flow.

Ah, okay, I found this article here which gives a good summary for taking action on flow run cancellation. TL;DR "reacting to a Cancelled state allows to send notifications on a flow run cancellation or to shut down remote execution separate from the agent, but it doesn’t allow to clean up the agent’s / flow run’s compute resources."