如何在 docker python API 中流式传输日志？

Question

我正在使用 docker python API 从 Dockerfile 构建图像。

import os
import sys
import os.path
import docker


client = docker.from_env()
try:
    here = os.path.dirname(__file__)
    no_cache = False
    dockerfile = os.path.join(here, 'app', 'nextdir')
    image = client.images.build(path=dockerfile, tag='app:v.2.4', nocache=no_cache, stream=True)

操作成功完成，但是我无法流式传输日志。 API 说：

返回一个阻塞生成器，您可以对其进行迭代以检索构建 发生时输出

当流=真时。

如何在 python 中获取这些日志？

Answer 1

可以使用docker-py 中给出的低级 API 来流式传输 docker 构建日志，如下所示，

        here = os.path.dirname(__file__)
        dockerfile = os.path.join(here, 'app', 'nextdir')
        docker_client = docker.APIClient(base_url='unix://var/run/docker.sock')
        generator = docker_client.build(path=dockerfile, tag='app:v.2.4', rm=True)
        while True:
            try:
                output = generator.__next__
                output = output.strip('\r\n')
                json_output = json.loads(output)
                if 'stream' in json_output:
                    click.echo(json_output['stream'].strip('\n'))
            except StopIteration:
                click.echo("Docker image build complete.")
                break
            except ValueError:
                click.echo("Error parsing output from docker image build: %s" % output)

Answer 2

根据docs，镜像构建现在返回一个包含镜像和构建日志的元组

第一项是图像对象 构建的图像。第二项是将构建日志生成为 JSON 解码对象。

并相应地修改@havock 解决方案：

import docker

client = docker.from_env()
image, build_logs = client.images.build(**your_build_kwargs)

for chunk in build_logs:
    if 'stream' in chunk:
        for line in chunk['stream'].splitlines():
            log.debug(line)

Answer 3

您可以使用low-level API 客户端。 build() 函数将返回一个生成器，您可以对其进行迭代以获取构建日志的块。

生成器将生成一个包含 JSON 对象的字符串，您可以在其上调用 json.loads()，或者您可以使用 build() 函数中的 decode=True 参数为您执行此操作。

一旦你从产生的字典中获取'stream' 键，你可能只是print() 它，但如果你需要将它发送到记录器，最好逐行执行，因为收到的块将包含不止一行。

此类代码的一个选项如下：

from docker import APIClient

client = APIClient(base_url='unix://var/run/docker.sock')

# Build docker image
log.info('Building docker image ...')
streamer = client.build(
    decode=True,
    path=args.path,
    tag=args.tag,
)

for chunk in streamer:
    if 'stream' in chunk:
        for line in chunk['stream'].splitlines():
            log.debug(line)

Answer 4

Docker 低级构建能够直接将输出解码为字典。为此使用decode=True。

我使用 mashumaro 将字典转换为数据类。如果您喜欢类型提示，这很有用。要使用数据类，您至少需要 python3.7。

安装 mashumaro：

pip install mashumaro

数据类：

@dataclass
class DockerStreamOutput(DataClassDictMixin):
    stream: str = None
    aux: str = None

建立你的形象：

from mashumaro import DataClassDictMixin, MissingField
generator = client.api.build(
            path=container_path,
            tag=tag_name,
            dockerfile=dockerfile_path,
            decode=True  # encode to dict
        )

for line in generator:
    try:
        stream_output: DockerStreamOutput = DockerStreamOutput.from_dict(line)
        stream: str = stream_output.stream
        if stream:
            logging.info(stream.strip())
    except MissingField:
        logging.error(line)

希望这会有所帮助。

Answer 5

如果您对上述答案有疑问，我可以使用以下方法获得所需的内容：

def _build(self, **kwargs) -> bool:
    """**kwargs should contain the same values expected by the docker.from_env().api.build() method"""
    print("building...")
    try:
        response = [line for line in docker.from_env().api.build(decode=True, **kwargs)]
        if not response:
            print("Error! Failed to get a response from docker client")
            return False
        for result in response:
            for (key, value) in result.items():
                if isinstance(value, str):
                    print(value)
        return True
    except (docker_error.BuildError, docker_error.APIError) as error:
        print("Error! {}")
    return False

Answer 6

使用较新版本的 python docker sdk，您仍然可以使用高级 api 函数，因为它们会在需要时返回 [object_created, generator] 形式的元组。

这已在python 3.10 中测试，运行docker==4.2.2 和click==8.0.3

这是我用来记录我正在构建的 cli 的 docker 输出的函数，基于 this answer:

def log_docker_output(generator, task_name: str = 'docker command execution') -> None:
    """
    Log output to console from a generator returned from docker client
    :param Any generator: The generator to log the output of
    :param str task_name: A name to give the task, i.e. 'Build database image', used for logging
    """
    while True:
        try:
            output = generator.__next__()
            if 'stream' in output:
                output_str = output['stream'].strip('\r\n').strip('\n')
                click.echo(output_str)
        except StopIteration:
            click.echo(f'{task_name} complete.')
            break
        except ValueError:
            click.echo(f'Error parsing output from {task_name}: {output}')

示例用法：

import docker
from docker.client import DockerClient

docker_client: DockerClient = docker.from_env()

image_obj, log_generator = docker_client.images.build(path='path/to/my_api_dockerfile')
log_docker_output(log_generator, "Building Django API image")

Answer 7

docs 状态...

如果您想获取构建的原始输出，请使用低级 API 中的 build() 方法。

你试过了吗？