I only found

prefect flow-run logs

but I would need to wait then until the flow run is finished

I tried to write a python function which executes the flow and then prints all the logs afterwards. This sometimes works and sometimes only shows not the full logs. I think it takes a while until all logs are saved and available.

import asyncio

from prefect import get_client
from prefect.deployments import run_deployment
from prefect.server.schemas.core import FlowRun
from prefect.server.schemas.filters import LogFilter, LogFilterFlowRunId


async def print_flow_run_logs(flow_run_id: str):
    log_filter = LogFilter(flow_run_id=LogFilterFlowRunId(any_=[flow_run_id]))

    async with get_client() as client:
        flow_run_logs = await client.read_logs(log_filter=log_filter)

    for flow_run_log in flow_run_logs:
        print(flow_run_log.message)


if __name__ == "__main__":
    flow_run: FlowRun = run_deployment(name="my-flow")
    asyncio.run(print_flow_run_logs(flow_run_id=flow_run.id))

Is there a better method to start a flow and stream the prefect cloud logs to the machine where I start the flow?

Objectively speaking, the flow run is in EKS, and streaming to the worker, and to Prefect cloud. To get the logs locally, would either be a push effort (from the container to your local terminal), or a pull effort (from your local terminal to the container) You could tail / stream the container, or the cloudwatch log group

or the prefect rest logs endpoint

but your local terminal never enters the picture for execution

so you’d need to filter for flow state first to ensure it has completed (or failed or whatever) before attempting to print, because there is a delay in aggregating the logs

it’s near real-time but not exactly

@Christopher Boyd I added that function:

async def wait_until_flow_run_reached_terminal_state(flow_run_id: str):
    while True:
        async with get_client() as client:
            flow_run = await client.read_flow_run(flow_run_id=flow_run_id)
            if flow_run.state.type in TERMINAL_STATES:
                return

asyncio.run(wait_until_flow_run_reached_terminal_state(flow_run_id=flow_run.id))
asyncio.run(print_flow_run_logs(flow_run_id=flow_run.id))

Unfortunately it does not help. The logs are still not always fully available. When

run_deployment

finishes, it is already in a terminal state. And I can see all logs in Prefect Cloud while the flow runs, but when I try to retrieve them, I get a lot less. I added a sleep 10 seconds command but this seems to be not enough. Will try to increase it to a minute then