Reading Logging Data in a Prefect Task

[AlexanderKri]

Hello everyone,

I am currently working on a prefect flow and I want to create a prefect task that reads all the available logging data of the flow that is being executed at the moment. My goal is to search for specific keywords such as "ERROR" in the logs and take appropriate action based on the results.

Is it possible to read the logging data of the currently running flow? If so, how can I achieve this?

Any suggestions or insights would be greatly appreciated. Thank you in advance for your help!

[zanieb]

You could read the logs from the API with the client

import asyncio

from prefect import flow, task, get_client, runtime, get_run_logger
from prefect.server.schemas.filters import LogFilter
from prefect.logging.handlers import APILogHandler


@flow
def example():
    check_logs.submit()
    logger = get_run_logger()
    logger.info("test")


@task
async def check_logs():
    async with get_client() as client:
        for _ in range(5):
            # Ensure any pending logs are sent
            await APILogHandler.aflush()
            logs = await client.read_logs(
                LogFilter(flow_run_id={"any_": [runtime.flow_run.id]})
            )

            for log in logs:
                print(log)

            await asyncio.sleep(1)


example()

[AlexanderKri]

Dear madkinsz,

Firstly, I would like to express my gratitude for your response to my previous question. Your answer was helpful, and I appreciate your effort in providing a solution!

Upon further review, I realized that my question was not precise enough, and I apologize for any confusion caused. To clarify, I have a Prefect flow consisting of three tasks that execute sequentially. What I am trying to achieve is to create a fourth Prefect task that will execute after the third task in my flow and store all available logging data of the flow in a variable. Later I want to search for specific keywords, such as "ERROR," in the variable and create a new output in the Prefect UI logging window.

I have adapted the code a bit and store logging data in the variable logs. Basically this is exactly what I wanted to have. Thanks again! Would you say that the custom code is the right one for my task? For example, the code is executed asynchronously, which I don't really need since the task is executed at the very end and I only need the logs up to that fourth task.

@task
async def check_logs():
    async with get_client() as client:
        # Ensure any pending logs are sent
        await APILogHandler.aflush()
        logs = await client.read_logs(
            LogFilter(flow_run_id={"any_": [runtime.flow_run.id]})
        )

        print(logs)

        await asyncio.sleep(1)

Thank you again for your response!

Best regards,
Alex

[zanieb]

The code is written in async because we do not supply a sync client at this time. You can call the task without using submit after your other tasks. I wrote this example to run in the background while other tasks run, but you definitely don't have to do it that way. You can remove the loop / sleep. For production, you'll need to read the logs in batches as we will only return 200 at a time. See the prefect flow-run logs CLI implementation for an example of that.