[Core Feature] Local logs for local container task executions.

[dansola]

It would be nice if we could have something like a boolean flag that controls if local container tasks output logs to the terminal.

I think think could be added somewhere around here using the .logs() method of the docker client shown on this page.

An example of the SDK UX could be something like this:

from flytekit import ContainerTask, workflow

echo_container_task = ContainerTask(
    name="echo",
    image="alpine:latest",
    command=["echo", "'a test to show stdout'"],
    local_logs=True,
)

@workflow
def echo_workflow() -> None:
    return echo_container_task()

if __name__ == "__main__":
    print(echo_workflow())

where local_logs is ignored if it is a remote execution.

[runllm]

To enable local logs for local container task executions in Flyte, you can propose adding a local_logs boolean flag to the ContainerTask class. This flag would control whether the logs from the container are output to the terminal during local execution. The implementation can leverage the .logs() method of the Docker client.

Here's a conceptual example of how this could be implemented:

1. Modify ContainerTask Class: Add a local_logs parameter to the ContainerTask class.
2. Implement Logging Logic: Use the Docker client's .logs() method to fetch and print logs if local_logs is set to True.

Example code snippet:

from flytekit import ContainerTask, workflow
import docker

class CustomContainerTask(ContainerTask):
    def __init__(self, *args, local_logs=False, **kwargs):
        super().__init__(*args, **kwargs)
        self.local_logs = local_logs

    def execute(self, **kwargs):
        client = docker.from_env()
        container = client.containers.run(self.image, self.command, detach=True)
        if self.local_logs:
            for log in container.logs(stream=True):
                print(log.strip())
        container.wait()
        return container

echo_container_task = CustomContainerTask(
    name="echo",
    image="alpine:latest",
    command=["echo", "'a test to show stdout'"],
    local_logs=True,
)

@workflow
def echo_workflow() -> None:
    return echo_container_task()

if __name__ == "__main__":
    print(echo_workflow())

This example demonstrates how to extend the ContainerTask class to include a local_logs flag and print logs to the terminal during local execution.

Would you like more details on any specific part of this implementation?

Sources:

• Flytekit ContainerTask
• Docker Client Logs

---

This is an AI-generated response and your feedback is appreciated! Please leave a 👍 if this is helpful and 👎 if it is not.

Mention @RunLLM to ask follow-up questions. See runllm.com for more!

[aditya7302]

I would like to work on this issue can you assign it to me ?

[dansola]

Hi @aditya7302. Of course! Thank you for taking on this issue. 🚀 🙏