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 base64
import datetime
import shlex
import re
import os

    import simplejson as json
except Exception:
    import json

from libcloud.utils.py3 import httplib
from libcloud.utils.py3 import b

from libcloud.common.base import JsonResponse, ConnectionUserAndKey
from libcloud.common.base import KeyCertificateConnection
from libcloud.common.types import InvalidCredsError

from libcloud.container.base import Container, ContainerDriver, ContainerImage

from libcloud.container.providers import Provider
from libcloud.container.types import ContainerState


[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 "%s %s" % (self.code, self.message) def __repr__(self): return "DockerException %s %s" % (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("%s:%s" % (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(DockertlsConnection, self).__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(DockerContainerDriver, self).__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%s/images/create?fromImage=%s" % (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%s/containers/json%s" % (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%s/containers/%s/start" % (self.version, id_), method="POST" ) else: result = self.connection.request( "/v%s/containers/%s/start" % (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%s/containers/%s/json" % (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%s/containers/%s/start" % (self.version,, method="POST" ) else: payload = { "Binds": [], "PublishAllPorts": True, } data = json.dumps(payload) result = self.connection.request( "/v%s/containers/%s/start" % (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%s/containers/%s/stop" % (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%s/containers/%s/restart" % (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%s/containers/%s" % (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%s/containers/%s/top" % (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%s/containers/%s/rename?name=%s" % (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%s/images/search?term=%s" % (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%s/images/%s" % (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