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