#2505 [Core feature] Add ability to await first task of pyflyte run workflow Issue created by ggydush-fn Motivation: Why do you think this is important? Pyflyte run is an amazing improvement, but users have to leave the context of their terminal to check Flyte workflow run, or run flytectl commands repetitively to check the status. A design decision we've made, is to make the first task of a workflow lightweight (spec validation), and if this succeeds, a larger subworkflow/task will be executed. It may be useful to add the ability to wait for this first task to succeed before exiting pyflyte's CLI. Goal: What should the final outcome look like, ideally? I think 2 features to pyflyte run could be useful: 1. Print out flytectl command that can be used to monitor the workflow directly from CLI 2. Add an optional

--await-first-task

parameter that would wait for the first task to succeed before exiting the CLI. An indicator showing that the job was successful pushed to remote and is in running state would also be useful. For #2, I use something like this currently, and would love for this functionality to be added to flyte remote!

import time

from flytekit.clients.friendly import SynchronousFlyteClient
from flytekit.models.core.execution import NodeExecutionPhase
from flytekit.models.node_execution import NodeExecution
from flytekit.remote.executions import FlyteWorkflowExecution

from flyte_common.remote.remote import get_flyte_client
from flyte_common.remote.ui.status import status


class NodeMonitor:
    """Class for monitoring nodes in a workflow execution."""

    client: SynchronousFlyteClient = get_flyte_client()

    def __init__(self, workflow_execution: FlyteWorkflowExecution):
        self.workflow_execution = workflow_execution

    @status(message="Validating spec on Flyte")
    def wait_for_first_node_to_complete(self) -> None:
        """Wait for first node to complete before returning.

        Note: this is currently an experimental feature we've added to help validate
        specs remotely and notify user of any spec errors. It is recommended to have a
        lightweight first task so that your workflow will fail fast if there are problems
        with the spec.
        """
        first_node = self.get_first_node()
        while True:
            node = self.client.get_node_execution(first_node.id)
            status_string = NodeExecutionPhase.enum_to_string(node.closure.phase)
            if status_string in ["FAILED", "FAILING", "ABORTED", "TIMED_OUT"]:
                raise ValueError(
                    f"Spec validation failed. {node.closure.error.message}"
                )
            if status_string in ["SUCCEEDING", "SUCCEEDED"]:
                return
            time.sleep(1)  # Slow down requests to Flyte

    def get_first_node(self) -> NodeExecution:
        """Return the first node part of a workflow execution."""
        # Note: this functionality is not on current `FlyteRemote`
        while True:
            nodes = self.client.list_node_executions(
                self.workflow_execution.id,
                sort_by=Sort(key="started_at", direction="ASCENDING"),
            )[0]
            if len(nodes) > 1:
                # Wait until first node after "start-node"
                return nodes[0]
            time.sleep(1)  # Slow down requests to Flyte

Describe alternatives you've considered N/A Propose: Link/Inline OR Additional context No response Are you sure this issue hasn't been raised already? ☑︎ Yes Have you read the Code of Conduct? ☑︎ Yes flyteorg/flyte