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

zaqlnxep  于 2023-04-19  发布在  Python
关注(0)|答案(8)|浏览(214)

我正在使用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说:
返回一个阻塞生成器,您可以迭代它以检索生成输出
当stream=True时。
如何在python中获取这些日志?

ttisahbt

ttisahbt1#

根据文档,映像构建现在返回一个包含映像和构建日志的元组
第一项是构建的图像的Image对象。第二项是构建日志的生成器,作为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)
mtb9vblg

mtb9vblg2#

可以使用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)
jgwigjjp

jgwigjjp3#

您可以使用底层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)
46scxncf

46scxncf4#

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)

希望这能帮上忙。

iih3973s

iih3973s5#

如果你对上面的答案有问题,我可以用这个来得到我需要的:

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
3xiyfsfu

3xiyfsfu6#

使用较新版本的python docker sdk,您仍然可以使用高级API函数,因为它们在需要时返回[object_created, generator]形式的元组。
这已经在python 3.10中进行了测试,运行docker==4.2.2click==8.0.3
以下是我正在使用的函数,用于记录我正在构建的cli的docker输出,基于这个答案:

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")
sg24os4d

sg24os4d7#

我遇到了同样的问题,这个解决方案对我很有效:environment={"PYTHONUNBUFFERED": "1"}

kulphzqa

kulphzqa8#

docs状态…
如果您想获得构建的原始输出,请使用低级API中的build()方法。
你试过吗

相关问题