@Marvin I have spot instances on gcp, when spot is being taken, my flow runs are stuck on running, how can I send exit code 1 to the flow run when it occurs

You can handle the termination of your flow runs gracefully using state change hooks. Here's a basic strategy to send an exit code or trigger some actions when a spot instance is preempted: 1. Leverage State Change Hooks: Prefect allows you to define hooks that can be executed on state changes, such as when the flow run fails or is terminated. 2. Implement State Change Hook: You can create a state change hook that gets triggered when your flow enters a

Crashed

or

Failed

state and handle the exit actions within the hook. Here’s an example of how you might implement such a hook:

import os
from prefect import flow, task
from prefect.client import get_client
import prefect.runtime

async def handle_flow_crashed(flow, flow_run, state):
    """
    State change hook that is triggered when the flow enters a crashed state.
    This can be used to log information, clean up resources, or set an exit code.
    """
    # Implement your logic here
    print(f"Flow {flow_run.flow_id} has crashed. Setting exit code 1.")
    os._exit(1)  # Immediately terminate the process with exit code 1

@flow(state_handlers=[handle_flow_crashed])
def my_flow():
    # Your flow logic here
    pass

if __name__ == "__main__":
    my_flow()

3. Running Your Flow: Deploy and run your flow. The state change hook will handle setting the exit code when the flow crashes or fails. ### Links to Documentation 1. State Change Hooks Guide This approach should help you manage your flow runs more gracefully when using spot instances on GCP. Is there anything else you would like to know or any other problems you are facing with Prefect?