我正在使用 docker-py 构建和运行 Docker 映像。
通过阅读文档,我不清楚我应该如何检测构建图像是否存在错误。 build()
出现错误时不会引发异常。这让我觉得我必须调查回来的回应。
确定 docker-py 的 client.build() 是否失败的最佳方法是什么?
本着分享的精神,我想展示一个实现,它是 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)
看起来“最好”的方法是解码响应并寻找一个名为“错误”的键。
例如:
for response in client.build(path, tag, decode=True):
if response.has_key('error'):
raise Exception("Error building docker image: {}".format(response['error']))
创建 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