Monitor scheduled executions and handle failures
"""
Monitoring Scheduled Executions and Handling Failures
Flyte-sdk provides robust mechanisms for monitoring the lifecycle of your scheduled executions and building resilient automation through retries and timeouts. Understanding the different phases an execution goes through, how to inspect its details, and how to configure failure handling is crucial for reliable workflows.
Understanding Action Lifecycle with ActionPhase
Every execution in Flyte, referred to as an "action" or "run," progresses through a defined set of phases. The flyte.models.ActionPhase enum captures these states, providing a clear representation of an execution's current status.
from flyte.models import ActionPhase
print(ActionPhase.QUEUED)
print(ActionPhase.RUNNING)
print(ActionPhase.SUCCEEDED)
These phases include:
QUEUED: The action is waiting to be scheduled.WAITING_FOR_RESOURCES: The action is waiting for necessary compute resources to become available.INITIALIZING: The action is being set up and prepared for execution.RUNNING: The action is actively executing its logic.SUCCEEDED: The action completed successfully.FAILED: The action encountered an error and failed during execution.ABORTED: The action was manually terminated by a user.TIMED_OUT: The action exceeded its configured timeout limit and was terminated.
Flyte-sdk categorizes SUCCEEDED, FAILED, ABORTED, and TIMED_OUT as terminal phases. Once an action enters a terminal phase, it will not transition to any other state. You can check if a phase is terminal using the is_terminal property:
from flyte.models import ActionPhase
print(ActionPhase.SUCCEEDED.is_terminal) # True
print(ActionPhase.RUNNING.is_terminal) # False
The ActionPhase enum is instrumental for filtering runs and checking execution status, as demonstrated in the flyte.models module's docstring:
from flyte.models import ActionPhase
from flyte.remote import Run
# Filter runs by phase
# You can pass ActionPhase enum members or their string representations
# In a real scenario, this would be an async operation.
# runs_in_terminal_state = Run.listall(in_phase=(ActionPhase.SUCCEEDED, ActionPhase.FAILED))
# Check if a specific run succeeded
# Assuming 'my-run' is an existing run name
# run = await Run.get.aio("my-run")
# if run.phase == ActionPhase.SUCCEEDED:
# print("Success!")
Monitoring Runs with Run and RunDetails
The flyte.remote.Run class serves as the primary interface for interacting with and managing task executions on the remote Flyte platform. It allows you to list, retrieve, and monitor the high-level status of your runs.
Listing Runs
You can retrieve a list of all runs for the current project and domain using Run.listall(). This method supports various filters, including filtering by ActionPhase:
from flyte.models import ActionPhase
from flyte.remote import Run
from datetime import datetime, timedelta
# from flyte.remote._helpers import TimeFilter # TimeFilter is an internal helper, not directly exposed
# List runs that succeeded or failed
# The in_phase parameter accepts a tuple of ActionPhase members or their string values.
# In a real scenario, this would be an async operation.
# async for run in Run.listall(in_phase=(ActionPhase.SUCCEEDED, ActionPhase.FAILED)):
# print(f"Run {run.name} is in phase {run.phase}")
# You can also use string representations for phases
# async for run in Run.listall(in_phase=("succeeded",)):
# print(f"Succeeded run: {run.name}")
# Combine with other filters, e.g., by creation time
# Note: TimeFilter is an internal helper. For user-facing code, you'd typically pass datetime objects directly
# after = datetime.now() - timedelta(days=7)
# async for run in Run.listall(in_phase=("succeeded",), created_at=TimeFilter(after=after)):
# print(f"Recent succeeded run: {run.name}")
Getting Specific Run Details
To get a specific run, use Run.get() with the run's name. This returns a Run object, which provides access to its current phase, name, and other high-level properties.
from flyte.remote import Run
# Assuming 'my-specific-run' is an existing run name
# run = await Run.get.aio("my-specific-run")
# print(f"Run name: {run.name}, Current phase: {run.phase}")
# print(f"Is run done? {run.done()}")
For more comprehensive details about a run, you can access the RunDetails object associated with it via run.details():
from flyte.remote import Run, RunDetails
# Assuming 'my-specific-run' is an existing run name
# run: Run = await Run.get.aio("my-specific-run")
# run_details: RunDetails = await run.details.aio()
# print(f"Run details for task: {run_details.task_name}")
# output = await run_details.outputs() # Access outputs after the run is complete
Alternatively, you can fetch RunDetails directly if you know the run's name:
from flyte.remote import RunDetails
# details = await RunDetails.get.aio(name="my-specific-run")
# print(f"Retrieved details for run: {details.name}")
Retrieving Logs
Run objects allow you to retrieve logs for an execution. The get_logs() method returns an asynchronous iterator of log lines, while show_logs() prints them directly to the console. You can specify the attempt number, filter_system logs, and show_ts (timestamps).
from flyte.remote import Run
# Assuming 'my-run-with-logs' is an existing run name
# run = await Run.get.aio("my-run-with-logs")
# Get logs for the second attempt
# async for line in run.get_logs.aio(attempt=2):
# print(line)
# Get logs for the latest attempt, filtering system logs and showing timestamps
# async for line in run.get_logs.aio(filter_system=True, show_ts=True):
# print(line)
# To display logs directly
# await run.show_logs.aio(max_lines=50, show_ts=True)
Waiting for Run Completion
To pause execution until a run reaches a terminal state, use the wait() method. This method updates the Run object's internal state, ensuring properties like run.action.phase reflect the final status.
from flyte.remote import Run
# Assuming 'my-long-running-run' is an existing run name
# run = await Run.get.aio("my-long-running-run")
# print(f"Waiting for run {run.name} to complete...")
# await run.wait()
# print(f"Run {run.name} finished with phase: {run.phase}")
Deep Dive into Action Details with ActionDetails
For more granular insights into an execution's lifecycle, the flyte.remote.ActionDetails class provides detailed information, including phase transitions and durations. ActionDetails is typically accessed via a RunDetails object.
Inspecting Phase Transitions and Durations
ActionDetails allows you to examine the time spent in each phase of an execution. The get_phase_transitions() method provides a list of PhaseTransitionInfo objects, detailing the start and end times for each phase. The phase_durations property offers a convenient dictionary mapping ActionPhase to timedelta objects, showing the total time spent in each phase for the latest attempt.
from flyte.remote import ActionDetails, RunDetails, Run
from flyte.models import ActionPhase
import datetime
# To get ActionDetails, you typically retrieve a Run, then its RunDetails
# and then access action_details from there.
# For demonstration, let's assume we have an ActionDetails object.
# In a real scenario, you'd get it like this:
# run = await Run.get.aio("my-run-name")
# run_details = await run.details.aio()
# action_details = run_details.action_details
# Example of how to get and use ActionDetails (assuming 'action_details' is already obtained)
# transitions = action_details.get_phase_transitions()
# for t in transitions:
# print(f"Phase: {t.phase}, Duration: {t.duration}")
# Accessing phase durations directly
# durations = action_details.phase_durations
# print(f"Queued time: {durations.get(ActionPhase.QUEUED, datetime.timedelta(0))}")
# print(f"Running time: {durations.get(ActionPhase.RUNNING, datetime.timedelta(0))}")
# Specific properties for common phase durations
# print(f"Queued time (property): {action_details.queued_time}")
# print(f"Running time (property): {action_details.running_time}")
Building Resilient Automation: Retries and Timeouts
Flyte-sdk allows you to configure tasks to automatically retry upon failure and to enforce execution time limits, enhancing the resilience of your workflows.
Configuring Retries
The retries parameter in the @env.task decorator allows you to specify how many times a task should be retried if it fails. You can provide an integer directly for the retry count, or use a flyte.RetryStrategy object for more advanced configurations (though the direct integer approach is shown here due to example simplification).
from flyte import TaskEnvironment
env = TaskEnvironment(name="my_task_env")
@env.task(retries=5)
def my_retriable_task_int():
# This task will retry up to 5 times if it fails
print("Attempting my_retriable_task_int...")
# raise ValueError("Simulated failure")
Defining Timeouts
The timeout parameter in the @env.task decorator allows you to set maximum execution times for your tasks. This prevents tasks from running indefinitely and consuming resources. You can provide a datetime.timedelta object or an integer (interpreted as seconds) directly for the maximum runtime. For more granular control, including maximum queued time, a flyte.Timeout object can be used (though the direct timedelta or integer approach is shown here due to example simplification).
from flyte import TaskEnvironment
from datetime import timedelta
import asyncio
env = TaskEnvironment(name="my_timeout_env")
@env.task(timeout=timedelta(minutes=5))
async def my_task_with_timedelta_timeout():
# This task has a max runtime of 5 minutes
print("Running with timedelta timeouts...")
await asyncio.sleep(30) # Simulate work
@env.task(timeout=300) # 300 seconds = 5 minutes
async def my_task_with_int_timeout():
# This task has a max runtime of 5 minutes (integer input)
print("Running with integer timeout...")
await asyncio.sleep(60) # Simulate work
Configuring Notifications
Flyte-sdk integrates with notification systems, allowing you to receive alerts based on the ActionPhase of your runs. You can configure notifications to trigger on specific phases, such as SUCCEEDED or FAILED, using classes like flyte.notify.Slack or flyte.notify.Email.
This example could not be verified against this version of the codebase and may not work as shown. Validator finding: Parsed source module 'flyte.notify' has no member 'notify'
import os
from flyte.models import ActionPhase
import flyte.notify
from flyte import TaskEnvironment
# Assuming SLACK_WEBHOOK_URL and NOTIFICATION_EMAIL are set as environment variables
SLACK_WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL", "https://hooks.slack.com/services/...")
NOTIFICATION_EMAIL = os.getenv("NOTIFICATION_EMAIL", "your_email@example.com")
slack_success = flyte.notify.Slack(
on_phase=ActionPhase.SUCCEEDED,
webhook_url=SLACK_WEBHOOK_URL,
blocks=[
{
"type": "header",
"text": {"type": "plain_text", "text": "Task Succeeded"},
},
{
"type": "section",
"fields": [
{"type": "mrkdwn", "text": "*Run:*\\n{{.Run.Name}}"},
{"type": "mrkdwn", "text": "*Phase:*\\n{{.Phase}}"},
],
},
{"type": "divider"},
{
"type": "context",
"elements": [
{"type": "mrkdwn", "text": "{{.Run.Project}}/{{.Run.Domain}}"},
],
},
],
)
email_failure = flyte.notify.Email(
on_phase=ActionPhase.FAILED,
recipients=[NOTIFICATION_EMAIL],
subject="ALERT: Run {{.Run.Name}} failed",
body=("Run: {{.Run.Name}}\\nError: {{.Error}}\\n"),
)
# These notification objects would then be associated with a task or workflow.
# For example:
# env = TaskEnvironment(name="my_notified_env")
# @env.task(notifications=[slack_success, email_failure])
# def my_notified_task():
# pass
By leveraging ActionPhase for status tracking, Run and ActionDetails for deep monitoring, and RetryStrategy and Timeout for robust failure handling, flyte-sdk empowers you to build and manage highly reliable and observable automated workflows.
"""