Hi Flyte community, I have a scheduled launchplan of a workflow. I wish to deactivate the launchplan if the workflow fails (in an execution) is there a way to do it?

no i don't think so

While it's correct there is no explicit feature to do exactly that, you can leverage failure nodes to achieve this. You can add a failure node to the workflow, it'll only be invoked in case the workflow fails, you can then inspect the execution using flyte remote to find out which launchpad was used and deactivate it (all using FlyteRemote)

Thanks! That's interesting. How would I accomplish the inspection part?

I'm on my phone atm.. maybe @damp-lion-88352 can help? Or @ancient-wolf-19325 ?

To inspect the execution and determine which launch plan was used, you can use the

FlyteRemote

interface from the

flytekit

library. Here's a general approach: 1. Set Up FlyteRemote: Initialize

FlyteRemote

with your Flyte configuration. 2. Fetch Execution Details: Use

FlyteRemote

to fetch the workflow execution details. 3. Inspect Execution: Check the execution metadata to identify the launch plan used. 4. Deactivate Launch Plan: If the workflow fails, use the

FlyteRemote

to deactivate the launch plan. Would you like a more detailed explanation or a code example? Sources: - FlyteCTL Feature 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

Yes, please write the code to do that

To accomplish this task using

FlyteRemote

, you can follow these steps in Python:

from flytekit import FlyteRemote, Config
from flytekit.models.admin import Execution

# Initialize FlyteRemote with your Flyte configuration
remote = FlyteRemote(
    config=Config.auto(config_file="path/to/your/config.yaml"),
    default_project="your_project",
    default_domain="your_domain"
)

# Fetch the execution details
execution_id = "your_execution_id"  # Replace with your actual execution ID
execution = remote.fetch_execution(name=execution_id)

# Check if the execution failed
if execution.closure.phase == Execution.Phase.FAILED:
    # Fetch the launch plan associated with the execution
    launch_plan = remote.fetch_launch_plan(name=execution.spec.launch_plan.name)

    # Deactivate the launch plan
    remote.deactivate_launch_plan(launch_plan)
    print(f"Launch plan {launch_plan.id.name} has been deactivated.")
else:
    print("Execution did not fail, no action taken.")

This code snippet initializes a

FlyteRemote

instance, fetches the execution details, checks if the execution has failed, and deactivates the associated launch plan if it has. Make sure to replace placeholders like

path/to/your/config.yaml

,

your_project

,

your_domain

, and

your_execution_id

with your actual configuration and execution details. Would you like more information on any specific part of this process? Sources: - https://github.com/flyteorg/flyte/pull/4308 - https://github.com/flyteorg/flyte/issues/1506 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325 how to get the current execution's execution Id

To get the current execution's execution ID within a Flyte task, you can use the

flytekit

library to access the current context. Here's how you can do it:

from flytekit import current_context

# Get the current execution ID
ctx = current_context()
execution_id = ctx.execution_id.name
print(f"Current execution ID: {execution_id}")

This code snippet retrieves the execution ID from the current Flyte context using

flytekit.current_context()

and prints it. Would you like more details on this process? Sources: - Flyte Community Discussion 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

Thank you! Really appreciated

failure node documentation https://docs.flyte.org/en/latest/user_guide/development_lifecycle/failure_node.html#id1