如何检测 docker-py client.build() 何时失败答案

【问题标题】：How can I detect when docker-py client.build() fails如何检测 docker-py client.build() 何时失败
【发布时间】：2016-02-07 18:57:59
【问题描述】：

我正在使用 docker-py 构建和运行 Docker 映像。

从阅读the documentation 开始，我不清楚我应该如何检测构建图像时是否存在错误。 build() 出现错误时不会引发异常。这让我觉得我必须调查回来的回应。

确定 docker-py 的 client.build() 是否失败的最佳方法是什么？

【问题讨论】：

标签： python docker dockerpy

【解决方案1】：

本着分享的精神，我想展示一个实现，它是 Fabien 答案的进展（这非常有用）。这会执行打印、清理等操作，并在出现问题时引发异常：

import json
import logging
import re

import docker

log = logging.getLogger(__name__)


class StreamLineBuildGenerator(object):
    def __init__(self, json_data):
        self.__dict__ = json.loads(json_data)


class Docker(object):
    REGISTRY = "some_docker_registry"

    def __init__(self):
        self.client = docker.from_env()
        self.api_client = docker.APIClient()

    def build(self, path, repository):
        tag = "{}/{}".format(Docker.REGISTRY, repository)
        output = self.api_client.build(path=path, tag=tag)
        self._process_output(output)
        log.info("done building {}".format(repository))

    def push(self, repository):
        tag = "{}/{}".format(Docker.REGISTRY, repository)
        output = self.client.images.push(tag)
        self._process_output(output)
        log.info("done pushing {}".format(tag))

    def _process_output(self, output):
        if type(output) == str:
            output = output.split("\n")

        for line in output:
            if line:
                errors = set()
                try:
                    stream_line = StreamLineBuildGenerator(line)

                    if hasattr(stream_line, "status"):
                        log.info(stream_line.status)

                    elif hasattr(stream_line, "stream"):
                        stream = re.sub("^\n", "", stream_line.stream)
                        stream = re.sub("\n$", "", stream)
                        # found after newline to close (red) "error" blocks: 27 91 48 109
                        stream = re.sub("\n(\x1B\[0m)$", "\\1", stream)
                        if stream:
                            log.info(stream)

                    elif hasattr(stream_line, "aux"):
                        if hasattr(stream_line.aux, "Digest"):
                            log.info("digest: {}".format(stream_line.aux["Digest"]))

                        if hasattr(stream_line.aux, "ID"):
                            log.info("ID: {}".format(stream_line.aux["ID"]))

                    else:
                        log.info("not recognized (1): {}".format(line))

                    if hasattr(stream_line, "error"):
                        errors.add(stream_line.error)

                    if hasattr(stream_line, "errorDetail"):
                        errors.add(stream_line.errorDetail["message"])

                        if hasattr(stream_line.errorDetail, "code"):
                            error_code = stream_line.errorDetail["code"]
                            errors.add("Error code: {}".format(error_code))

                except ValueError as e:
                    log.error("not recognized (2): {}".format(line))

                if errors:
                    message = "problem executing Docker: {}".format(". ".join(errors))
                    raise SystemError(message)

【讨论】：

【解决方案2】：

创建 StreamLineBuilder 生成器：

import json
class StreamLineBuildGenerator(object):
    def __init__(self, json_data):
        self.__dict__ = json.loads(json_data)

然后使用这个生成器来解析你的流：

import docker
docker_client = docker.Client(version="1.18", base_url="unix:///var/run/docker.sock")

generator = docker_client.build(nocache=False, rm=True, stream=True, tag="my_image_tag", path="my_path")

for line in generator:
    try:
        stream_line = StreamLineBuildGenerator(line)
        if hasattr(stream_line, "error"):
            print(stream_line.error)
            if hasattr(stream_line, "errorDetail"):
                if not stream_line.error == stream_line.errorDetail["message"]:
                    if hasattr(stream_line.errorDetail, "code"):
                        print("[" + stream_line.errorDetail["code"] + "] ", False)
                    print(stream_line.errorDetail["message"])
    except ValueError:
        # If we are not able to deserialize the received line as JSON object, just print it out
        print(line)
        continue

【讨论】：

【解决方案3】：

看起来“最好”的方法是解码响应并查找名为“错误”的键。

例如：

for response in client.build(path, tag, decode=True):
    if response.has_key('error'):
        raise Exception("Error building docker image: {}".format(response['error']))

【讨论】：