如何用 Python 操作 Docker？

作者：肖恩頓

來源：游戲不存在

docker-py是Docker SDK for Python。docker-py主要利用了requests，使用http/socket協(xié)議連接本地的docker engine進(jìn)行操作。對(duì) docker 感興趣，苦于工作中只用到 http 協(xié)議的同學(xué)，都建議閱讀一下本文。話不多說，一起了解docker-py的實(shí)現(xiàn)，本文分下面幾個(gè)部分:

• docker-py項(xiàng)目結(jié)構(gòu)
• docker-py API示例
• DockerClient的實(shí)現(xiàn)
• docker-version命令跟蹤
• UnixHTTPAdapter的實(shí)現(xiàn)
• docker-ps命令跟蹤
• docker-logs命令跟蹤
• docker-exec 命令跟蹤
• 小結(jié)
• 小技巧

docker-py項(xiàng)目結(jié)構(gòu)

本次代碼閱讀，使用的版本是 4.2.0， 項(xiàng)目目錄結(jié)構(gòu)大概如下:

還有一些目錄和類，因?yàn)椴辉谶@次介紹中，所以就沒有羅列。

docker-py API示例

docker-py API上手非常簡(jiǎn)單:

import docker
client = docker.from_env()

result = client.version()
print(result)
# {'Platform': {'Name': 'Docker Engine - Community'},...}

client.containers.list()
# [<Container '45e6d2de7c54'>, <Container 'db18e4f20eaa'>, ...]

client.images.pull('nginx:1.10-alpine')
# <Image: 'nginx:1.10-alpine'>

client.images.list()
[<Image 'ubuntu'>, <Image 'nginx:1.10-alpine'>, ...]

上面示例展示了:

• 使用環(huán)境變量，創(chuàng)建client連接本地docker-engine服務(wù)
• 獲取版本號(hào)，等同 docker version
• 獲取正在運(yùn)行的容器列表，等同 docker container list(別名是 docker ps)
• 拉取 nginx:1.10-alpin 鏡像，等同 docker image pull nginx:1.10-alpine(別名是docker pull nginx:1.10-alpine)
• 獲取鏡像列表, 等同 docker image list

我們可以看到，docker-py的操作和docker的標(biāo)準(zhǔn)命令基本一致。

DockerClient的實(shí)現(xiàn)

DockerClient的構(gòu)造函數(shù)和工廠方法展示docker-client對(duì)象包裝了APIClient對(duì)象：

# client.py

class DockerClient(object):
    def __init__(self, *args, **kwargs):
        self.api = APIClient(*args, **kwargs)
    
    @classmethod
    def from_env(cls, **kwargs):
        timeout = kwargs.pop('timeout', DEFAULT_TIMEOUT_SECONDS)
        max_pool_size = kwargs.pop('max_pool_size', DEFAULT_MAX_POOL_SIZE)
        version = kwargs.pop('version', None)
        use_ssh_client = kwargs.pop('use_ssh_client', False)
        return cls(
            timeout=timeout,
            max_pool_size=max_pool_size,
            version=version,
            use_ssh_client=use_ssh_client,
            **kwargs_from_env(**kwargs)
        )

DockerClient的API分2中，一種是屬性方法，比如常用的 containers，images，networks 和 volumes 等子命令，因?yàn)橐獙⒎祷刂蛋b成對(duì)應(yīng)模型對(duì)象:

@property
def containers(self):
    """
    An object for managing containers on the server. See the
    :doc:`containers documentation <containers>` for full details.
    """
    return ContainerCollection(client=self)

@property
def images(self):
    return ImageCollection(client=self)

@property
def networks(self):
    return NetworkCollection(client=self)

@property
def volumes(self):
    return VolumeCollection(client=self)
    
    ...

另一種是不需要模型包裝，可以直接使用APIClient返回結(jié)果的 info, version 等方法:

# Top-level methods
def info(self, *args, **kwargs):
        return self.api.info(*args, **kwargs)
    info.__doc__ = APIClient.info.__doc__

def version(self, *args, **kwargs):
        return self.api.version(*args, **kwargs)
    version.__doc__ = APIClient.version.__doc__
    
    ...

DockerClient類工廠方法的全局引用:

from_env = DockerClient.from_env

docker-version命令跟蹤

我們先從簡(jiǎn)單的 docker version 命令跟蹤查看APIClient如何工作的。APIClient的構(gòu)造函數(shù):

# api/client.py

import requests

class APIClient(
        requests.Session,
        BuildApiMixin,
        ConfigApiMixin,
        ContainerApiMixin,
        DaemonApiMixin,
        ExecApiMixin,
        ImageApiMixin,
        NetworkApiMixin,
        PluginApiMixin,
        SecretApiMixin,
        ServiceApiMixin,
        SwarmApiMixin,
        VolumeApiMixin):
        
    def __init__(self, base_url=None, version=None,
             timeout=DEFAULT_TIMEOUT_SECONDS, tls=False,
             user_agent=DEFAULT_USER_AGENT, num_pools=None,
             credstore_env=None, use_ssh_client=False,
             max_pool_size=DEFAULT_MAX_POOL_SIZE):
        super(APIClient, self).__init__()
        
        base_url = utils.parse_host(
            base_url, IS_WINDOWS_PLATFORM, tls=bool(tls)
        )
        
        if base_url.startswith('http+unix://'):
            self._custom_adapter = UnixHTTPAdapter(
                base_url, timeout, pool_connections=num_pools,
                max_pool_size=max_pool_size
            )
            self.mount('http+docker://', self._custom_adapter)
            self._unmount('http://', 'https://')
            # host part of URL should be unused, but is resolved by requests
            # module in proxy_bypass_macosx_sysconf()
            self.base_url = 'http+docker://localhost'

上面代碼可見:

• APIClient繼承自 requests.Session
• APIClient使用Mixin方式組合了多個(gè)API，比如ContainerApiMixin提供container的api操作;NetWorkApiMixin提供network的api操作
• 使用mount方法加載不同協(xié)議的適配器adapter，unix系的docker是unix-socket;windows則是npipe

關(guān)于requests的使用，可以參看之前的博文 requests 源碼閱讀

默認(rèn)的服務(wù)URL實(shí)現(xiàn):

DEFAULT_UNIX_SOCKET = "http+unix:///var/run/docker.sock"
DEFAULT_NPIPE = 'npipe:////./pipe/docker_engine'

def parse_host(addr, is_win32=False, tls=False):
    path = ''
    port = None
    host = None

    # Sensible defaults
    if not addr and is_win32:
        return DEFAULT_NPIPE
    if not addr or addr.strip() == 'unix://':
        return DEFAULT_UNIX_SOCKET

version 請(qǐng)求在 DaemonApiMixin 中實(shí)現(xiàn):

class DaemonApiMixin(object):


    def version(self, api_version=True):
        url = self._url("/version", versioned_api=api_version)
        return self._result(self._get(url), json=True)

底層的請(qǐng)求和響應(yīng)在主類APIClient中提供:

class APIClient
    
    def _url(self, pathfmt, *args, **kwargs):
        ...
        return '{0}{1}'.format(self.base_url, pathfmt.format(*args))

    @update_headers
    def _get(self, url, **kwargs):
        return self.get(url, **self._set_request_timeout(kwargs))
            
    def _result(self, response, json=False, binary=False):
        assert not (json and binary)
        self._raise_for_status(response)
    
        if json:
            return response.json()
        if binary:
            return response.content
        return response.text

get和result，response都是requests提供。get發(fā)送請(qǐng)求，response.json將請(qǐng)求格式化成json后返回。

UnixHTTPAdapter的實(shí)現(xiàn)

/var/run/docker.sock是Docker守護(hù)程序偵聽的UNIX套接字,其連接使用UnixHTTPAdapter處理:

# transport/unixconn.py

import requests.adapters

RecentlyUsedContainer = urllib3._collections.RecentlyUsedContainer

class UnixHTTPAdapter(BaseHTTPAdapter):
    def __init__(self, socket_url, timeout=60,
                 pool_connections=constants.DEFAULT_NUM_POOLS,
                 max_pool_size=constants.DEFAULT_MAX_POOL_SIZE):
        socket_path = socket_url.replace('http+unix://', '')
        if not socket_path.startswith('/'):
            socket_path = '/' + socket_path
        self.socket_path = socket_path
        self.timeout = timeout
        self.max_pool_size = max_pool_size
        self.pools = RecentlyUsedContainer(
            pool_connections, dispose_func=lambda p: p.close()
        )
        super(UnixHTTPAdapter, self).__init__()
    
    def get_connection(self, url, proxies=None):
        with self.pools.lock:
            pool = self.pools.get(url)
            if pool:
                return pool

            pool = UnixHTTPConnectionPool(
                url, self.socket_path, self.timeout,
                maxsize=self.max_pool_size
            )
            self.pools[url] = pool

        return pool

UnixHTTPAdapter主要使用urllib3提供的鏈接池管理UnixHTTPConnection連接:

class UnixHTTPConnection(httplib.HTTPConnection, object):

    def __init__(self, base_url, unix_socket, timeout=60):
        super(UnixHTTPConnection, self).__init__(
            'localhost', timeout=timeout
        )
        self.base_url = base_url
        self.unix_socket = unix_socket
        self.timeout = timeout
        self.disable_buffering = False

    def connect(self):
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        sock.settimeout(self.timeout)
        sock.connect(self.unix_socket)
        self.sock = sock

    def putheader(self, header, *values):
        super(UnixHTTPConnection, self).putheader(header, *values)
        if header == 'Connection' and 'Upgrade' in values:
            self.disable_buffering = True

    def response_class(self, sock, *args, **kwargs):
        if self.disable_buffering:
            kwargs['disable_buffering'] = True

        return UnixHTTPResponse(sock, *args, **kwargs)


class UnixHTTPConnectionPool(urllib3.connectionpool.HTTPConnectionPool):
    def __init__(self, base_url, socket_path, timeout=60, maxsize=10):
        super(UnixHTTPConnectionPool, self).__init__(
            'localhost', timeout=timeout, maxsize=maxsize
        )
        self.base_url = base_url
        self.socket_path = socket_path
        self.timeout = timeout

    def _new_conn(self):
        return UnixHTTPConnection(
            self.base_url, self.socket_path, self.timeout
        )

connect展示了socket類型是 socket.AF_UNIX, 這一部分的實(shí)現(xiàn)都非常基礎(chǔ) 。

關(guān)于socket，可以參看之前的博文 python http 源碼閱讀

docker-ps命令跟蹤

接著我們跟蹤稍微復(fù)雜點(diǎn)的命令 client.containers.list()， 也就是 docker ps。前面介紹了，container 會(huì)組裝結(jié)果為數(shù)據(jù)模型，下面是模型的父類:

class Model(object):
    """
    A base class for representing a single object on the server.
    """
    id_attribute = 'Id'

    def __init__(self, attrs=None, client=None, collection=None):
        self.client = client
        # 集合
        self.collection = collection

        self.attrs = attrs

Model是單個(gè)模型抽象，Collection則是模型集合的抽象，使用集合的prepare_model構(gòu)建各種對(duì)象:

class Collection(object):
    """
    A base class for representing all objects of a particular type on the
    server.
    """

    model = None

    def __init__(self, client=None):
        self.client = client
    
    ...
    
    def prepare_model(self, attrs):
        """
        Create a model from a set of attributes.
        """
        if isinstance(attrs, Model):
            attrs.client = self.client
            # 雙向引用
            attrs.collection = self
            return attrs
        elif isinstance(attrs, dict):
            return self.model(attrs=attrs, client=self.client, collection=self)
        else:
            raise Exception("Can't create %s from %s" %
                            (self.model.__name__, attrs))

Container和ContainerCollection的實(shí)現(xiàn)

class Container(Model):
    pass
    
class ContainerCollection(Collection):
    model = Container
    
    def get(self, container_id):
        resp = self.client.api.inspect_container(container_id)
        return self.prepare_model(resp)
        
    def list(self, all=False, before=None, filters=None, limit=-1, since=None,
             sparse=False, ignore_removed=False):
        resp = self.client.api.containers(all=all, before=before,
                                          filters=filters, limit=limit,
                                          since=since)
        containers = []
        for r in resp:
            containers.append(self.get(r['Id']))
        return containers

其中l(wèi)ist函數(shù)主要有下面幾個(gè)步驟

• 使用api的containers接口得到resp，就是container-id列表
• 逐個(gè)循環(huán)使用api的inspect_container請(qǐng)求container的詳細(xì)信息
• 將結(jié)果封裝成Container對(duì)象
• 返回容器Container對(duì)象列表

api.containers和api.inspect_container在ContainerApiMixin中提供, 非常簡(jiǎn)單清晰:

class ContainerApiMixin(object):

    def containers(self, quiet=False, all=False, trunc=False, latest=False,
                   since=None, before=None, limit=-1, size=False,
                   filters=None):
        params = {
            'limit': 1 if latest else limit,
            'all': 1 if all else 0,
            'size': 1 if size else 0,
            'trunc_cmd': 1 if trunc else 0,
            'since': since,
            'before': before
        }
        if filters:
            params['filters'] = utils.convert_filters(filters)
        u = self._url("/containers/json")
        res = self._result(self._get(u, params=params), True)

        if quiet:
            return [{'Id': x['Id']} for x in res]
        if trunc:
            for x in res:
                x['Id'] = x['Id'][:12]
        return res
    
    @utils.check_resource('container')
    def inspect_container(self, container):
        return self._result(
            self._get(self._url("/containers/{0}/json", container)), True
        )

docker-logs命令跟蹤

前面的命令都是request-response的模式，我們?cè)倏纯床灰粯拥模诹鞯膁ocker-logs命令。我們先啟動(dòng)一個(gè)容器:

docker run -d bfirsh/reticulate-splines

查看容器列表

# docker ps
CONTAINER ID   IMAGE                       COMMAND                  CREATED          STATUS          PORTS             NAMES
61709b0ed4b8   bfirsh/reticulate-splines   "/usr/local/bin/run.…"   22 seconds ago   Up 21 seconds                     festive_pare

實(shí)時(shí)跟蹤容器運(yùn)行日志:

# docker logs -f 6170
Reticulating spline 1...
Reticulating spline 2...
....

可以看到reticulate-splines容器就是不停的打印行數(shù)數(shù)據(jù)。可以用下面的代碼實(shí)現(xiàn) docker logs 相同的功能:

logs = client.containers.get('61709b0ed4b8').logs(stream=True)
 try:
  while True:
   line = next(logs).decode("utf-8")
   print(line)
 except StopIteration:
  print(f'log stream ended for {container_name}')   

代碼執(zhí)行結(jié)果和前面的類似:

# python sample.py
...
Reticulating spline 14...

Reticulating spline 15...
...

logs的實(shí)現(xiàn)中返回一個(gè)CancellableStream，而不是一個(gè)result，利用這個(gè)stream，就可以持續(xù)的讀取輸出:

# models/Container

def logs(self, **kwargs):
    return self.client.api.logs(self.id, **kwargs)
    
# api/continer
def logs(self, container, stdout=True, stderr=True, stream=False,
             timestamps=False, tail='all', since=None, follow=None,
             until=None):
    ...
    
    url = self._url("/containers/{0}/logs", container)
        res = self._get(url, params=params, stream=stream)
        output = self._get_result(container, stream, res)

        if stream:
            return CancellableStream(output, res)
        else:
            return output

比較特別的是下面對(duì)于stream的處理:

# api/client

def _multiplexed_response_stream_helper(self, response):
    """A generator of multiplexed data blocks coming from a response
    stream."""

    # Disable timeout on the underlying socket to prevent
    # Read timed out(s) for long running processes
    socket = self._get_raw_response_socket(response)
    self._disable_socket_timeout(socket)

    while True:
        header = response.raw.read(STREAM_HEADER_SIZE_BYTES)
        if not header:
            break
        _, length = struct.unpack('>BxxxL', header)
        if not length:
            continue
        data = response.raw.read(length)
        if not data:
            break
        yield data

def _disable_socket_timeout(self, socket):
    sockets = [socket, getattr(socket, '_sock', None)]

    for s in sockets:
        if not hasattr(s, 'settimeout'):
            continue

        timeout = -1

        if hasattr(s, 'gettimeout'):
            timeout = s.gettimeout()

        # Don't change the timeout if it is already disabled.
        if timeout is None or timeout == 0.0:
            continue

        s.settimeout(None)

上面代碼展示了:

• 流的讀取方式是每次讀取STREAM_HEADER_SIZE_BYTES長(zhǎng)度的數(shù)據(jù)作為協(xié)議頭
• 協(xié)議頭結(jié)構(gòu)體格式解壓后得到后面的數(shù)據(jù)包長(zhǎng)度
• 繼續(xù)讀取指定長(zhǎng)度的數(shù)據(jù)包
• 重復(fù)執(zhí)行上面的數(shù)據(jù)讀取過程
• 流式讀取的時(shí)候還需要關(guān)閉socket的超時(shí)機(jī)制，確保流一直保持，知道手動(dòng)(ctl+c)關(guān)閉

而 attach 則是采用了websocket的實(shí)現(xiàn), 因?yàn)槲覀円话阃扑]使用exec命令，所以這里簡(jiǎn)單了解即可:

def _attach_websocket(self, container, params=None):
    url = self._url("/containers/{0}/attach/ws", container)
    req = requests.Request("POST", url, params=self._attach_params(params))
    full_url = req.prepare().url
    full_url = full_url.replace("http://", "ws://", 1)
    full_url = full_url.replace("https://", "wss://", 1)
    return self._create_websocket_connection(full_url)

def _create_websocket_connection(self, url):
    return websocket.create_connection(url)

docker-exec 命令跟蹤

docker-exec是我們的重頭戲，因?yàn)槌丝梢灾苯荧@取docker是輸出外，還可以和docker進(jìn)行交互。先簡(jiǎn)單回顧一下exec的使用:

# docker exec -it 2075 ping www.weibo.cn
PING www.weibo.cn (123.125.22.241): 56 data bytes
64 bytes from 123.125.22.241: seq=0 ttl=37 time=6.797 ms
64 bytes from 123.125.22.241: seq=1 ttl=37 time=39.279 ms
64 bytes from 123.125.22.241: seq=2 ttl=37 time=29.635 ms
64 bytes from 123.125.22.241: seq=3 ttl=37 time=27.737 ms

上面示例可以用下面代碼完全模擬:

result = client.containers.get("2075").exec_run("ping www.weibo.cn", tty=True, stream=True)
try:
 while True:
  line = next(result[1]).decode("utf-8")
  print(line)
except StopIteration:
 print(f'exec stream ended for {container_name}')

使用tty偽裝終端和容器進(jìn)行交互，就是我們最常用的方式了:

# docker exec -it 2075 sh
/ # ls -la
total 64
drwxr-xr-x    1 root     root          4096 Mar 24 13:16 .
drwxr-xr-x    1 root     root          4096 Mar 24 13:16 ..
-rwxr-xr-x    1 root     root             0 Mar 24 13:16 .dockerenv
drwxr-xr-x    2 root     root          4096 Mar  3  2017 bin
drwxr-xr-x    5 root     root           340 Mar 24 13:16 dev
drwxr-xr-x    1 root     root          4096 Mar 24 13:16 etc
drwxr-xr-x    2 root     root          4096 Mar  3  2017 home
drwxr-xr-x    1 root     root          4096 Mar  3  2017 lib
lrwxrwxrwx    1 root     root            12 Mar  3  2017 linuxrc -> /bin/busybox
drwxr-xr-x    5 root     root          4096 Mar  3  2017 media
drwxr-xr-x    2 root     root          4096 Mar  3  2017 mnt
dr-xr-xr-x  156 root     root             0 Mar 24 13:16 proc
drwx------    1 root     root          4096 Mar 25 08:17 root
drwxr-xr-x    2 root     root          4096 Mar  3  2017 run
drwxr-xr-x    2 root     root          4096 Mar  3  2017 sbin
drwxr-xr-x    2 root     root          4096 Mar  3  2017 srv
dr-xr-xr-x   13 root     root             0 Mar 24 13:16 sys
drwxrwxrwt    1 root     root          4096 Mar  3  2017 tmp
drwxr-xr-x    1 root     root          4096 Mar  3  2017 usr
drwxr-xr-x    1 root     root          4096 Mar  3  2017 var
/ # exit

同樣這個(gè)過程也可以使用docker-py實(shí)現(xiàn):

_, socket = client.containers.get("2075").exec_run("sh", stdin=True, socket=True)
print(socket)
socket._sock.sendall(b"ls -la\n")
try:
 unknown_byte=socket._sock.recv(docker.constants.STREAM_HEADER_SIZE_BYTES)
 print(unknown_byte)

 buffer_size = 4096 # 4 KiB
 data = b''
 while True:
  part = socket._sock.recv(buffer_size)
  data += part
  if len(part) < buffer_size:
   # either 0 or end of data
   break
 print(data.decode("utf8"))

except Exception: 
 pass
socket._sock.send(b"exit\n")

示例演示的過程是:

• 獲取一個(gè)已經(jīng)存在的容器2075
• 對(duì)容器執(zhí)行exec命令，注意需要開啟stdin和socket
• 向容器發(fā)送 ls -lah 展示目錄列表
• 讀區(qū)socket上的結(jié)果。(這里我們偷懶，沒有解析頭，直接硬取，這樣不夠健壯)
• 繼續(xù)發(fā)送 exit 退出容器

程序的輸出和上面使用命令方式完全一致，就不在張貼了。進(jìn)入核心的exec_run函數(shù)的實(shí)現(xiàn):

# model/containers

def exec_run(self, cmd, stdout=True, stderr=True, stdin=False, tty=False,
                 privileged=False, user='', detach=False, stream=False,
                 socket=False, environment=None, workdir=None, demux=False):
    resp = self.client.api.exec_create(
            self.id, cmd, stdout=stdout, stderr=stderr, stdin=stdin, tty=tty,
            privileged=privileged, user=user, environment=environment,
            workdir=workdir,
        )
    exec_output = self.client.api.exec_start(
        resp['Id'], detach=detach, tty=tty, stream=stream, socket=socket,
        demux=demux
    )
    if socket or stream:
        return ExecResult(None, exec_output)

主要使用API的exec_create和exec_start兩個(gè)函數(shù), 先看第一個(gè)exec_create函數(shù):

# api/exec_api

def exec_create(self, container, cmd, stdout=True, stderr=True,
                    stdin=False, tty=False, privileged=False, user='',
                    environment=None, workdir=None, detach_keys=None):

    if isinstance(cmd, six.string_types):
        cmd = utils.split_command(cmd)

    if isinstance(environment, dict):
        environment = utils.utils.format_environment(environment)

    data = {
        'Container': container,
        'User': user,
        'Privileged': privileged,
        'Tty': tty,
        'AttachStdin': stdin,
        'AttachStdout': stdout,
        'AttachStderr': stderr,
        'Cmd': cmd,
        'Env': environment,
    }

    if detach_keys:
        data['detachKeys'] = detach_keys
    elif 'detachKeys' in self._general_configs:
        data['detachKeys'] = self._general_configs['detachKeys']

    url = self._url("/containers/{0}/exec", container)
    res = self._post_json(url, data=data)
    return self._result(res, True)

exec_create相對(duì)還是比較簡(jiǎn)單，就是post-json數(shù)據(jù)到 /containers/{0}/exec 接口。然后是exec_start函數(shù):

def exec_start(self, exec_id, detach=False, tty=False, stream=False,
               socket=False, demux=False):

    # we want opened socket if socket == True

    data = {
        'Tty': tty,
        'Detach': detach
    }

    headers = {} if detach else {
        'Connection': 'Upgrade',
        'Upgrade': 'tcp'
    }

    res = self._post_json(
        self._url("/exec/{0}/start", exec_id),
        headers=headers,
        data=data,
        stream=True
    )
    if detach:
        return self._result(res)
    if socket:
        return self._get_raw_response_socket(res)
    return self._read_from_socket(res, stream, tty=tty, demux=demux)

exec_start是post-json到 /exec/{0}/start 接口，注意這個(gè)接口看起來不是到容器，而是到exec。然后如果socket參數(shù)是true則返回socket，可以進(jìn)行寫入；否則僅僅讀取數(shù)據(jù)。

使用curl訪問docker-api

docker-engine的REST-api也可以直接使用 curl 訪問:

$ curl --unix-socket /var/run/docker.sock -H "Content-Type: application/json" \
  -d '{"Image": "alpine", "Cmd": ["echo", "hello world"]}' \
  -X POST http://localhost/v1.41/containers/create
{"Id":"1c6594faf5","Warnings":null}

$ curl --unix-socket /var/run/docker.sock -X POST http://localhost/v1.41/containers/1c6594faf5/start

$ curl --unix-socket /var/run/docker.sock -X POST http://localhost/v1.41/containers/1c6594faf5/wait
{"StatusCode":0}

$ curl --unix-socket /var/run/docker.sock "http://localhost/v1.41/containers/1c6594faf5/logs?stdout=1"
hello world

可以通過修改/etc/docker/daemon.json更改為http服務(wù)方式的api

{
  "debug": true,
  "hosts": ["tcp://192.168.59.3:2376"]
}

然后 curl 命令可以直接訪問docker的api

curl http://127.0.0.1:2375/info
curl http://127.0.0.1:2375/version
curl http://127.0.0.1:2375/images/json
curl http://127.0.0.1:2375/images/alpine/json
curl http://127.0.0.1:2375/containers/json
curl http://127.0.0.1:2375/containers/25c5805a06b6/json

小結(jié)

利用docker-py可以完全操作docker，這得益docker提供的REST-api操作。同時(shí)也發(fā)現(xiàn)requests的設(shè)計(jì)很強(qiáng)大，不僅僅可以用來做http請(qǐng)求，還可以用來做socket請(qǐng)求。學(xué)習(xí)docker-py后，相信大家對(duì)docker的理解一定有那么一點(diǎn)點(diǎn)加深，也希望下面這張圖可以幫助你記憶:

小技巧

使用 check_resource 裝飾器，對(duì)函數(shù)的參數(shù)進(jìn)行預(yù)先處理:

def check_resource(resource_name):
    def decorator(f):
        @functools.wraps(f)
        def wrapped(self, resource_id=None, *args, **kwargs):
            if resource_id is None and kwargs.get(resource_name):
                resource_id = kwargs.pop(resource_name)
            if isinstance(resource_id, dict):
                resource_id = resource_id.get('Id', resource_id.get('ID'))
            if not resource_id:
                raise errors.NullResource(
                    'Resource ID was not provided'
                )
            return f(self, resource_id, *args, **kwargs)
        return wrapped
    return decorator

代碼版本比較工具:

from distutils.version import StrictVersion


def compare_version(v1, v2):
    """Compare docker versions

    >>> v1 = '1.9'
    >>> v2 = '1.10'
    >>> compare_version(v1, v2)
    1
    >>> compare_version(v2, v1)
    -1
    >>> compare_version(v2, v2)
    0
    """
    s1 = StrictVersion(v1)
    s2 = StrictVersion(v2)
    if s1 == s2:
        return 0
    elif s1 > s2:
        return -1
    else:
        return 1


def version_lt(v1, v2):
    return compare_version(v1, v2) > 0


def version_gte(v1, v2):
    return not version_lt(v1, v2)

參考鏈接

• https://docs.docker.com/engine/api/sdk/examples/
• https://docker-py.readthedocs.io/en/stable/