Stream and Display Remote Execution Logs

When working with remote executions in flyte-sdk, you often need to inspect logs to understand the progress and troubleshoot issues. The flyte.remote.Action and flyte.remote.Run classes provide convenient methods for retrieving and displaying these logs. A Run object internally manages an Action object and delegates its log-related operations to this Action instance.

Retrieving Logs Programmatically with get_logs()

To access execution logs as an iterable stream for custom processing, use the get_logs() method available on both Action and Run objects. This method returns an asynchronous generator, allowing you to process log lines as they become available.

import asyncio
from flyte.remote import Action, Run
from flyte.remote._logs import LogsNotYetAvailableError

async def get_and_print_logs(action: Action):
    print(f"Fetching logs for action: {action.name}")
    try:
        async for line in action.get_logs.aio():
            print(line.strip())
    except LogsNotYetAvailableError as e:
        print(f"Error: {e}")

# In a real scenario, you would obtain an Action or Run object from a remote client.
# For demonstration, let's assume 'my_action' is an existing Action instance.
# For example, you might get an action from a run:
# run = await Run.get(name="my-run-name")
# my_action = run.action

# Example usage (replace with your actual Action/Run instance)
# await get_and_print_logs(my_action)

Filtering System-Generated Lines

To focus on user-generated output and exclude system messages (e.g., those prefixed with [flyte]), set the filter_system parameter to True.

import asyncio
from flyte.remote import Action

async def get_filtered_logs(action: Action):
    print(f"Fetching filtered logs for action: {action.name}")
    async for line in action.get_logs.aio(filter_system=True):
        print(line.strip())

# await get_filtered_logs(my_action)

Including Timestamps

To prefix each log line with an ISO-8601 timestamp, set the show_ts parameter to True.

import asyncio
from flyte.remote import Action

async def get_timestamped_logs(action: Action):
    print(f"Fetching timestamped logs for action: {action.name}")
    async for line in action.get_logs.aio(show_ts=True):
        print(line.strip())

# await get_timestamped_logs(my_action)

Retrieving Logs for a Specific Attempt

Actions can have multiple attempts. To retrieve logs for a particular attempt, specify the attempt parameter. If not specified, logs for the latest attempt are returned.

import asyncio
from flyte.remote import Action

async def get_logs_for_attempt(action: Action, attempt_number: int):
    print(f"Fetching logs for action: {action.name}, attempt: {attempt_number}")
    async for line in action.get_logs.aio(attempt=attempt_number):
        print(line.strip())

# await get_logs_for_attempt(my_action, 2) # Get logs for the second attempt

Displaying Logs Interactively with show_logs()

For an interactive, real-time view of logs in your console or Jupyter notebook, use the show_logs() method. This method provides a dynamic viewer that updates as new log lines arrive.

import asyncio
from flyte.remote import Action

async def show_interactive_logs(action: Action):
    print(f"Displaying interactive logs for action: {action.name}")
    await action.show_logs.aio()

# await show_interactive_logs(my_action)

Customizing Interactive Display

The show_logs() method offers several parameters to customize the interactive viewer:

• attempt (int): The attempt number to show logs for (defaults to the latest attempt).
• max_lines (int): Maximum number of log lines to display in the viewer. The viewer scrolls and keeps only max_lines in view (default: 30).
• show_ts (bool): Whether to show timestamps with each log line (default: False).
• raw (bool): If True, logs are printed directly to the console without the interactive viewer. This is useful in non-interactive environments (default: False).
• filter_system (bool): If True, system-generated log lines are filtered out (default: False).

Here's an example demonstrating raw output with timestamps and filtered system lines:

import asyncio
from flyte.remote import Action

async def show_raw_filtered_timestamped_logs(action: Action):
    print(f"Displaying raw, filtered, timestamped logs for action: {action.name}")
    await action.show_logs.aio(raw=True, filter_system=True, show_ts=True)

# await show_raw_filtered_timestamped_logs(my_action)

Handling Log Availability and Asynchronous Operations

Both get_logs() and show_logs() are designed to handle scenarios where logs might not be immediately available. If an action is not yet running or has not completed, these methods will automatically wait for logs to become ready before streaming them. This ensures a robust experience without needing manual polling.

It is crucial to remember that all log retrieval methods (get_logs(), show_logs()) are asynchronous. Therefore, they must be called with await (for show_logs()) or iterated using async for (for get_logs()) within an async function or an asyncio event loop. The .aio() suffix is used to explicitly call the asynchronous version of these methods. If logs are not available after a certain number of retries, a LogsNotYetAvailableError may be raised, which you can catch to handle such cases gracefully.