docker compose logs with time ordering

gistfile1.txt

#!/usr/bin/env python3
import os
import sys
import threading

import docker
from docker.types import CancellableStream

compose_ = "com.docker.compose"
compose_working_dir = compose_ + ".project.working_dir"
compose_container_number = compose_ + ".container-number"
compose_service = compose_ + ".service"


def iterate_over_it(logs_elements):
    buffer = b''
    for char in logs_elements:
        char = bytes([char])
        buffer += char
        if char == b'\n':
            yield buffer
            buffer = b''


cursor = None
cursor_lock = threading.Lock()


def reader(timestamps, width, container, logs_item):
    global cursor
    for line in logs_item:
        with cursor_lock:
            ts, _line = line.decode('utf-8').split(' ', maxsplit=1)

            if cursor is None or (cursor is not None and cursor < ts):
                label = f"{container.labels[compose_service]}-{container.labels[compose_container_number]}"
                if timestamps:
                    print(f"{label:<{width}} | {ts} {_line}", end='')
                else:
                    print(f"{label:<{width}} | {_line}", end='')

                cursor = ts


def main():
    client = docker.from_env()
    containers = client.containers.list(all=True)
    containers = [
        container for container in containers
        if compose_working_dir in container.labels and container.labels[compose_working_dir] == os.getcwd()
    ]

    kwargs = {}
    if "-f" in sys.argv:
        kwargs['follow'] = True
        kwargs['stream'] = True
    timestamps = "-t" in sys.argv
    # always get the log timestamp from the API
    kwargs['timestamps'] = True

    logs = []
    for container in containers:
        cont_logs = container.logs(**kwargs)
        if not isinstance(cont_logs, CancellableStream):
            cont_logs = iterate_over_it(cont_logs)
        logs.append((container, cont_logs))

    width = max(
        len(f"{container.labels[compose_service]}-{container.labels[compose_container_number]}") for container in
        containers
    )

    threads = []
    for container, logs_item in logs:
        thread = threading.Thread(target=reader, args=(timestamps, width, container, logs_item))
        thread.start()
        threads.append(thread)

    for thread in threads:
        thread.join()


if __name__ == '__main__':
    main()