Source code for libcloud.container.drivers.docker

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.

import os
import re
import shlex
import base64
import datetime

from libcloud.utils.py3 import b, httplib
from libcloud.common.base import JsonResponse, ConnectionUserAndKey, KeyCertificateConnection
from libcloud.common.types import InvalidCredsError
from libcloud.container.base import Container, ContainerImage, ContainerDriver
from libcloud.container.types import ContainerState
from libcloud.container.providers import Provider

    import simplejson as json
except Exception:
    import json


[docs]class DockerResponse(JsonResponse): valid_response_codes = [ httplib.OK, httplib.ACCEPTED, httplib.CREATED, httplib.NO_CONTENT, ]
[docs] def parse_body(self): if len(self.body) == 0 and not self.parse_zero_length_body: return self.body try: # error responses are tricky in Docker. Eg response could be # an error, but response status could still be 200 content_type = self.headers.get("content-type", "application/json") if content_type == "application/json" or content_type == "": if ( self.headers.get("transfer-encoding") == "chunked" and "fromImage" in self.request.url ): body = [ json.loads(chunk) for chunk in self.body.strip().replace("\r", "").split("\n") ] else: body = json.loads(self.body) else: body = self.body except ValueError: m ='Error: (.+?)"', self.body) if m: error_msg = raise Exception(error_msg) else: raise Exception("ConnectionError: Failed to parse JSON response") return body
[docs] def parse_error(self): if self.status == 401: raise InvalidCredsError("Invalid credentials") return self.body
[docs] def success(self): return self.status in self.valid_response_codes
[docs]class DockerException(Exception): def __init__(self, code, message): self.code = code self.message = message self.args = (code, message) def __str__(self): return "{} {}".format(self.code, self.message) def __repr__(self): return "DockerException {} {}".format(self.code, self.message)
[docs]class DockerConnection(ConnectionUserAndKey): responseCls = DockerResponse timeout = 60
[docs] def add_default_headers(self, headers): """ Add parameters that are necessary for every request If user and password are specified, include a base http auth header """ headers["Content-Type"] = "application/json" if self.user_id and self.key: user_b64 = base64.b64encode(b("{}:{}".format(self.user_id, self.key))) headers["Authorization"] = "Basic %s" % (user_b64.decode("utf-8")) return headers
[docs]class DockertlsConnection(KeyCertificateConnection): responseCls = DockerResponse def __init__( self, key, secret, secure=True, host="localhost", port=4243, key_file="", cert_file="", **kwargs, ): super().__init__( key_file=key_file, cert_file=cert_file, secure=secure, host=host, port=port, url=None, proxy_url=None, timeout=None, backoff=None, retry_delay=None, ) if key_file: keypath = os.path.expanduser(key_file) is_file_path = os.path.exists(keypath) and os.path.isfile(keypath) if not is_file_path: raise InvalidCredsError( "You need an key PEM file to authenticate with " "Docker tls. This can be found in the server." ) self.key_file = key_file certpath = os.path.expanduser(cert_file) is_file_path = os.path.exists(certpath) and os.path.isfile(certpath) if not is_file_path: raise InvalidCredsError( "You need an certificate PEM file to authenticate with " "Docker tls. This can be found in the server." ) self.cert_file = cert_file
[docs] def add_default_headers(self, headers): headers["Content-Type"] = "application/json" return headers
[docs]class DockerContainerDriver(ContainerDriver): """ Docker container driver class. >>> from libcloud.container.providers import get_driver >>> driver = get_driver('docker') >>> conn = driver(host='', port=4243) >>> conn.list_containers() or connecting to http basic auth protected https host: >>> conn = driver('user', 'pass', host='', port=443) connect with tls authentication, by providing a hostname, port, a private key file (.pem) and certificate (.pem) file >>> conn = driver(host='', >>> port=4243, key_file='key.pem', cert_file='cert.pem') """ type = Provider.DOCKER name = "Docker" website = "" connectionCls = DockerConnection supports_clusters = False version = "1.24" def __init__( self, key="", secret="", secure=False, host="localhost", port=4243, key_file=None, cert_file=None, ): """ :param key: API key or username to used (required) :type key: ``str`` :param secret: Secret password to be used (required) :type secret: ``str`` :param secure: Whether to use HTTPS or HTTP. Note: Some providers only support HTTPS, and it is on by default. :type secure: ``bool`` :param host: Override hostname used for connections. :type host: ``str`` :param port: Override port used for connections. :type port: ``int`` :param key_file: Path to private key for TLS connection (optional) :type key_file: ``str`` :param cert_file: Path to public key for TLS connection (optional) :type cert_file: ``str`` :return: ``None`` """ if key_file: self.connectionCls = DockertlsConnection self.key_file = key_file self.cert_file = cert_file secure = True if host.startswith("https://"): secure = True # strip the prefix prefixes = ["http://", "https://"] for prefix in prefixes: if host.startswith(prefix): host = host.strip(prefix) super().__init__( key=key, secret=secret, secure=secure, host=host, port=port, key_file=key_file, cert_file=cert_file, ) if key_file or cert_file: # docker tls authentication- # # We pass two files, a key_file with the # private key and cert_file with the certificate # libcloud will handle them through LibcloudHTTPSConnection if not (key_file and cert_file): raise Exception( "Needs both private key file and " "certificate file for tls authentication" ) = secure = host self.connection.port = port # set API version self.version = self._get_api_version() def _ex_connection_class_kwargs(self): kwargs = {} if hasattr(self, "key_file"): kwargs["key_file"] = self.key_file if hasattr(self, "cert_file"): kwargs["cert_file"] = self.cert_file return kwargs
[docs] def install_image(self, path): """ Install a container image from a remote path. :param path: Path to the container image :type path: ``str`` :rtype: :class:`libcloud.container.base.ContainerImage` """ payload = {} data = json.dumps(payload) result = self.connection.request( "/v{}/images/create?fromImage={}".format(self.version, path), data=data, method="POST", ) if "errorDetail" in result.body: raise DockerException(None, result.body) image_id = None # the response is slightly different if the image is already present # and it's not downloaded. both messages below indicate that the image # is available for use to the daemon if"Downloaded newer image", result.body) or r'"Status: Image is up to date', result.body ): if"sha256:(?P<id>[a-z0-9]{64})", result.body): image_id = re.findall(r"sha256:(?P<id>[a-z0-9]{64})", result.body)[-1] # if there is a failure message or if there is not an image id in the # response then throw an exception. if image_id is None: raise DockerException(None, "failed to install image") image = ContainerImage( id=image_id, name=path, path=path, version=None, driver=self.connection.driver, extra={}, ) return image
[docs] def list_images(self): """ List the installed container images :rtype: ``list`` of :class:`libcloud.container.base.ContainerImage` """ result = self.connection.request("/v%s/images/json" % (self.version)).object images = [] for image in result: try: name = image.get("RepoTags")[0] except Exception: name = image.get("Id") images.append( ContainerImage( id=image.get("Id"), name=name, path=name, version=None, driver=self.connection.driver, extra={ "created": image.get("Created"), "size": image.get("Size"), "virtual_size": image.get("VirtualSize"), }, ) ) return images
[docs] def list_containers(self, image=None, all=True): """ List the deployed container images :param image: Filter to containers with a certain image :type image: :class:`libcloud.container.base.ContainerImage` :param all: Show all container (including stopped ones) :type all: ``bool`` :rtype: ``list`` of :class:`libcloud.container.base.Container` """ if all: ex = "?all=1" else: ex = "" try: result = self.connection.request( "/v{}/containers/json{}".format(self.version, ex) ).object except Exception as exc: errno = getattr(exc, "errno", None) if errno == 111: raise DockerException( errno, "Make sure docker host is accessible" "and the API port is correct", ) raise containers = [self._to_container(value) for value in result] return containers
[docs] def deploy_container( self, name, image, parameters=None, start=True, command=None, hostname=None, user="", stdin_open=True, tty=True, mem_limit=0, ports=None, environment=None, dns=None, volumes=None, volumes_from=None, network_disabled=False, entrypoint=None, cpu_shares=None, working_dir="", domainname=None, memswap_limit=0, port_bindings=None, network_mode="bridge", labels=None, ): """ Deploy an installed container image For details on the additional parameters see : :param name: The name of the new container :type name: ``str`` :param image: The container image to deploy :type image: :class:`libcloud.container.base.ContainerImage` :param parameters: Container Image parameters :type parameters: ``str`` :param start: Start the container on deployment :type start: ``bool`` :rtype: :class:`Container` """ command = shlex.split(str(command)) if port_bindings is None: port_bindings = {} params = {"name": name} payload = { "Hostname": hostname, "Domainname": domainname, "ExposedPorts": ports, "User": user, "Tty": tty, "OpenStdin": stdin_open, "StdinOnce": False, "Memory": mem_limit, "AttachStdin": True, "AttachStdout": True, "AttachStderr": True, "Env": environment, "Cmd": command, "Dns": dns, "Image":, "Volumes": volumes, "VolumesFrom": volumes_from, "NetworkDisabled": network_disabled, "Entrypoint": entrypoint, "CpuShares": cpu_shares, "WorkingDir": working_dir, "MemorySwap": memswap_limit, "PublishAllPorts": True, "PortBindings": port_bindings, "NetworkMode": network_mode, "Labels": labels, } data = json.dumps(payload) try: result = self.connection.request( "/v%s/containers/create" % (self.version), data=data, params=params, method="POST", ) except Exception as e: message = e.message or str(e) # pylint: disable=no-member if message.startswith("No such image:"): raise DockerException(None, "No such image: %s" % else: raise DockerException(None, e) id_ = result.object["Id"] payload = { "Binds": [], "PublishAllPorts": True, "PortBindings": port_bindings, } data = json.dumps(payload) if start: if float(self._get_api_version()) > 1.22: result = self.connection.request( "/v{}/containers/{}/start".format(self.version, id_), method="POST" ) else: result = self.connection.request( "/v{}/containers/{}/start".format(self.version, id_), data=data, method="POST", ) return self.get_container(id_)
[docs] def get_container(self, id): """ Get a container by ID :param id: The ID of the container to get :type id: ``str`` :rtype: :class:`libcloud.container.base.Container` """ result = self.connection.request("/v{}/containers/{}/json".format(self.version, id)).object return self._to_container(result)
[docs] def start_container(self, container): """ Start a container :param container: The container to be started :type container: :class:`libcloud.container.base.Container` :return: The container refreshed with current data :rtype: :class:`libcloud.container.base.Container` """ if float(self._get_api_version()) > 1.22: result = self.connection.request( "/v{}/containers/{}/start".format(self.version,, method="POST" ) else: payload = { "Binds": [], "PublishAllPorts": True, } data = json.dumps(payload) result = self.connection.request( "/v{}/containers/{}/start".format(self.version,, method="POST", data=data, ) if result.status in VALID_RESPONSE_CODES: return self.get_container( else: raise DockerException(result.status, "failed to start container")
[docs] def stop_container(self, container): """ Stop a container :param container: The container to be stopped :type container: :class:`libcloud.container.base.Container` :return: The container refreshed with current data :rtype: :class:`libcloud.container.base.Container` """ result = self.connection.request( "/v{}/containers/{}/stop".format(self.version,, method="POST" ) if result.status in VALID_RESPONSE_CODES: return self.get_container( else: raise DockerException(result.status, "failed to stop container")
[docs] def restart_container(self, container): """ Restart a container :param container: The container to be stopped :type container: :class:`libcloud.container.base.Container` :return: The container refreshed with current data :rtype: :class:`libcloud.container.base.Container` """ data = json.dumps({"t": 10}) # number of seconds to wait before killing the container result = self.connection.request( "/v{}/containers/{}/restart".format(self.version,, data=data, method="POST", ) if result.status in VALID_RESPONSE_CODES: return self.get_container( else: raise DockerException(result.status, "failed to restart container")
[docs] def destroy_container(self, container): """ Remove a container :param container: The container to be destroyed :type container: :class:`libcloud.container.base.Container` :return: True if the destroy was successful, False otherwise. :rtype: ``bool`` """ result = self.connection.request( "/v{}/containers/{}".format(self.version,, method="DELETE" ) return result.status in VALID_RESPONSE_CODES
[docs] def ex_list_processes(self, container): """ List processes running inside a container :param container: The container to list processes for. :type container: :class:`libcloud.container.base.Container` :rtype: ``str`` """ result = self.connection.request( "/v{}/containers/{}/top".format(self.version, ).object return result
[docs] def ex_rename_container(self, container, name): """ Rename a container :param container: The container to be renamed :type container: :class:`libcloud.container.base.Container` :param name: The new name :type name: ``str`` :rtype: :class:`libcloud.container.base.Container` """ result = self.connection.request( "/v{}/containers/{}/rename?name={}".format(self.version,, name), method="POST", ) if result.status in VALID_RESPONSE_CODES: return self.get_container(
[docs] def ex_get_logs(self, container, stream=False): """ Get container logs If stream == True, logs will be yielded as a stream From Api Version 1.11 and above we need a GET request to get the logs Logs are in different format of those of Version 1.10 and below :param container: The container to list logs for :type container: :class:`libcloud.container.base.Container` :param stream: Stream the output :type stream: ``bool`` :rtype: ``bool`` """ payload = {} data = json.dumps(payload) if float(self._get_api_version()) > 1.10: result = self.connection.request( "/v%s/containers/%s/logs?follow=%s&stdout=1&stderr=1" % (self.version,, str(stream)) ).object logs = result else: result = self.connection.request( "/v%s/containers/%s/attach?logs=1&stream=%s&stdout=1&stderr=1" % (self.version,, str(stream)), method="POST", data=data, ) logs = result.body return logs
[docs] def ex_search_images(self, term): """Search for an image on Returns a list of ContainerImage objects >>> images = conn.ex_search_images(term='mistio') >>> images [<ContainerImage: id=rolikeusch/docker-mistio...>, <ContainerImage: id=mist/mistio, name=mist/mistio, driver=Docker ...>] :param term: The search term :type term: ``str`` :rtype: ``list`` of :class:`libcloud.container.base.ContainerImage` """ term = term.replace(" ", "+") result = self.connection.request( "/v{}/images/search?term={}".format(self.version, term) ).object images = [] for image in result: name = image.get("name") images.append( ContainerImage( id=name, path=name, version=None, name=name, driver=self.connection.driver, extra={ "description": image.get("description"), "is_official": image.get("is_official"), "is_trusted": image.get("is_trusted"), "star_count": image.get("star_count"), }, ) ) return images
[docs] def ex_delete_image(self, image): """ Remove image from the filesystem :param image: The image to remove :type image: :class:`libcloud.container.base.ContainerImage` :rtype: ``bool`` """ result = self.connection.request( "/v{}/images/{}".format(self.version,, method="DELETE" ) return result.status in VALID_RESPONSE_CODES
def _to_container(self, data): """ Convert container in Container instances """ try: name = data.get("Name").strip("/") except Exception: try: name = data.get("Names")[0].strip("/") except Exception: name = data.get("Id") state = data.get("State") if isinstance(state, dict): status = data.get("Status", state.get("Status") if state is not None else None) else: status = data.get("Status") if "Exited" in status: state = ContainerState.STOPPED elif status.startswith("Up "): state = ContainerState.RUNNING elif "running" in status: state = ContainerState.RUNNING else: state = ContainerState.STOPPED image = data.get("Image") ports = data.get("Ports", []) created = data.get("Created") if isinstance(created, float): created = ts_to_str(created) extra = { "id": data.get("Id"), "status": data.get("Status"), "created": created, "image": image, "ports": ports, "command": data.get("Command"), "sizerw": data.get("SizeRw"), "sizerootfs": data.get("SizeRootFs"), } ips = [] if ports is not None: for port in ports: if port.get("IP") is not None: ips.append(port.get("IP")) return Container( id=data["Id"], name=name, image=ContainerImage( id=data.get("ImageID", None), path=image, name=image, version=None, driver=self.connection.driver, ), ip_addresses=ips, state=state, driver=self.connection.driver, extra=extra, ) def _get_api_version(self): """ Get the docker API version information """ result = self.connection.request("/version").object result = result or {} api_version = result.get("ApiVersion") return api_version
[docs]def ts_to_str(timestamp): """ Return a timestamp as a nicely formated datetime string. """ date = datetime.datetime.fromtimestamp(timestamp) date_string = date.strftime("%d/%m/%Y %H:%M %Z") return date_string