import os import logging from typing import List from contextlib import contextmanager import docker logger = logging.getLogger(__name__) TRUE_VALUES = ['1', 'true', 'True', True, 1] def docker_client(): """ Create a docker client from the following environment variables:: DOCKER_HOST=unix://tmp/docker.sock DOCKER_TLS_VERIFY=1 DOCKER_CERT_PATH='' """ # NOTE: Remove this fallback in 1.0 if not os.environ.get('DOCKER_HOST'): os.environ['DOCKER_HOST'] = 'unix://tmp/docker.sock' return docker.from_env() def list_containers() -> List[dict]: """ List all containers. Returns: List of raw container json data from the api """ client = docker_client() all_containers = client.containers.list(all=True) client.close() return [c.attrs for c in all_containers] def get_swarm_nodes(): client = docker_client() # NOTE: If not a swarm node docker.errors.APIError is raised # 503 Server Error: Service Unavailable # ("This node is not a swarm manager. Use "docker swarm init" or # "docker swarm join" to connect this node to swarm and try again.") try: return client.nodes.list() except docker.errors.APIError: return [] def remove_containers(containers: List['Container']): client = docker_client() logger.info('Attempting to delete stale backup process containers') for container in containers: logger.info(' -> deleting %s', container.name) try: c = client.containers.get(container.name) c.remove() except Exception as ex: logger.exception(ex) def is_true(value): """ Evaluates the truthfullness of a bool value in container labels """ return value in TRUE_VALUES def strip_root(path): """ Removes the root slash in a path. Example: /srv/data becomes srv/data """ path = path.strip() if path.startswith('/'): return path[1:] return path @contextmanager def environment(name, value): """Tempset env var""" old_val = os.environ.get(name) os.environ[name] = value try: yield finally: if old_val is None: del os.environ[name] else: os.environ[name] = old_val