from __future__ import absolute_import from __future__ import unicode_literals import datetime import logging from functools import reduce from docker.errors import APIError from docker.errors import NotFound from . import parallel from .config import ConfigurationError from .config.sort_services import get_service_name_from_net from .const import DEFAULT_TIMEOUT from .const import IMAGE_EVENTS from .const import LABEL_ONE_OFF from .const import LABEL_PROJECT from .const import LABEL_SERVICE from .container import Container from .service import ContainerNet from .service import ConvergenceStrategy from .service import Net from .service import Service from .service import ServiceNet from .utils import microseconds_from_time_nano from .volume import Volume log = logging.getLogger(__name__) class Project(object): """ A collection of services. """ def __init__(self, name, services, client, volumes=None, use_networking=False, network_driver=None): self.name = name self.services = services self.client = client self.use_networking = use_networking self.network_driver = network_driver self.volumes = volumes or [] def labels(self, one_off=False): return [ '{0}={1}'.format(LABEL_PROJECT, self.name), '{0}={1}'.format(LABEL_ONE_OFF, "True" if one_off else "False"), ] @classmethod def from_config(cls, name, config_data, client, use_networking=False, network_driver=None): """ Construct a Project from a config.Config object. """ project = cls(name, [], client, use_networking=use_networking, network_driver=network_driver) if use_networking: remove_links(config_data.services) for service_dict in config_data.services: links = project.get_links(service_dict) volumes_from = project.get_volumes_from(service_dict) net = project.get_net(service_dict) project.services.append( Service( client=client, project=name, use_networking=use_networking, links=links, net=net, volumes_from=volumes_from, **service_dict)) if config_data.volumes: for vol_name, data in config_data.volumes.items(): project.volumes.append( Volume( client=client, project=name, name=vol_name, driver=data.get('driver'), driver_opts=data.get('driver_opts') ) ) return project @property def service_names(self): return [service.name for service in self.services] def get_service(self, name): """ Retrieve a service by name. Raises NoSuchService if the named service does not exist. """ for service in self.services: if service.name == name: return service raise NoSuchService(name) def validate_service_names(self, service_names): """ Validate that the given list of service names only contains valid services. Raises NoSuchService if one of the names is invalid. """ valid_names = self.service_names for name in service_names: if name not in valid_names: raise NoSuchService(name) def get_services(self, service_names=None, include_deps=False): """ Returns a list of this project's services filtered by the provided list of names, or all services if service_names is None or []. If include_deps is specified, returns a list including the dependencies for service_names, in order of dependency. Preserves the original order of self.services where possible, reordering as needed to resolve dependencies. Raises NoSuchService if any of the named services do not exist. """ if service_names is None or len(service_names) == 0: return self.get_services( service_names=self.service_names, include_deps=include_deps ) else: unsorted = [self.get_service(name) for name in service_names] services = [s for s in self.services if s in unsorted] if include_deps: services = reduce(self._inject_deps, services, []) uniques = [] [uniques.append(s) for s in services if s not in uniques] return uniques def get_services_without_duplicate(self, service_names=None, include_deps=False): services = self.get_services(service_names, include_deps) for service in services: service.remove_duplicate_containers() return services def get_links(self, service_dict): links = [] if 'links' in service_dict: for link in service_dict.get('links', []): if ':' in link: service_name, link_name = link.split(':', 1) else: service_name, link_name = link, None try: links.append((self.get_service(service_name), link_name)) except NoSuchService: raise ConfigurationError( 'Service "%s" has a link to service "%s" which does not ' 'exist.' % (service_dict['name'], service_name)) del service_dict['links'] return links def get_volumes_from(self, service_dict): volumes_from = [] if 'volumes_from' in service_dict: for volume_from_spec in service_dict.get('volumes_from', []): # Get service try: service = self.get_service(volume_from_spec.source) volume_from_spec = volume_from_spec._replace(source=service) except NoSuchService: try: container = Container.from_id(self.client, volume_from_spec.source) volume_from_spec = volume_from_spec._replace(source=container) except APIError: raise ConfigurationError( 'Service "%s" mounts volumes from "%s", which is ' 'not the name of a service or container.' % ( service_dict['name'], volume_from_spec.source)) volumes_from.append(volume_from_spec) del service_dict['volumes_from'] return volumes_from def get_net(self, service_dict): net = service_dict.pop('net', None) if not net: if self.use_networking: return Net(self.name) return Net(None) net_name = get_service_name_from_net(net) if not net_name: return Net(net) try: return ServiceNet(self.get_service(net_name)) except NoSuchService: pass try: return ContainerNet(Container.from_id(self.client, net_name)) except APIError: raise ConfigurationError( 'Service "%s" is trying to use the network of "%s", ' 'which is not the name of a service or container.' % ( service_dict['name'], net_name)) def start(self, service_names=None, **options): containers = [] for service in self.get_services(service_names): service_containers = service.start(**options) containers.extend(service_containers) return containers def stop(self, service_names=None, **options): parallel.parallel_stop(self.containers(service_names), options) def pause(self, service_names=None, **options): containers = self.containers(service_names) parallel.parallel_pause(reversed(containers), options) return containers def unpause(self, service_names=None, **options): containers = self.containers(service_names) parallel.parallel_unpause(containers, options) return containers def kill(self, service_names=None, **options): parallel.parallel_kill(self.containers(service_names), options) def remove_stopped(self, service_names=None, **options): parallel.parallel_remove(self.containers(service_names, stopped=True), options) def initialize_volumes(self): try: for volume in self.volumes: volume.create() except NotFound: raise ConfigurationError( 'Volume %s specifies nonexistent driver %s' % (volume.name, volume.driver) ) except APIError as e: if 'Choose a different volume name' in str(e): raise ConfigurationError( 'Configuration for volume {0} specifies driver {1}, but ' 'a volume with the same name uses a different driver ' '({3}). If you wish to use the new configuration, please ' 'remove the existing volume "{2}" first:\n' '$ docker volume rm {2}'.format( volume.name, volume.driver, volume.full_name, volume.inspect()['Driver'] ) ) def restart(self, service_names=None, **options): containers = self.containers(service_names, stopped=True) parallel.parallel_restart(containers, options) return containers def build(self, service_names=None, no_cache=False, pull=False, force_rm=False): for service in self.get_services(service_names): if service.can_be_built(): service.build(no_cache, pull, force_rm) else: log.info('%s uses an image, skipping' % service.name) def create(self, service_names=None, strategy=ConvergenceStrategy.changed, do_build=True): services = self.get_services_without_duplicate(service_names, include_deps=True) plans = self._get_convergence_plans(services, strategy) for service in services: service.execute_convergence_plan( plans[service.name], do_build, detached=True, start=False) def events(self): def build_container_event(event, container): time = datetime.datetime.fromtimestamp(event['time']) time = time.replace( microsecond=microseconds_from_time_nano(event['timeNano'])) return { 'time': time, 'type': 'container', 'action': event['status'], 'id': container.id, 'service': container.service, 'attributes': { 'name': container.name, 'image': event['from'], } } service_names = set(self.service_names) for event in self.client.events( filters={'label': self.labels()}, decode=True ): if event['status'] in IMAGE_EVENTS: # We don't receive any image events because labels aren't applied # to images continue # TODO: get labels from the API v1.22 , see github issue 2618 container = Container.from_id(self.client, event['id']) if container.service not in service_names: continue yield build_container_event(event, container) def up(self, service_names=None, start_deps=True, strategy=ConvergenceStrategy.changed, do_build=True, timeout=DEFAULT_TIMEOUT, detached=False): services = self.get_services_without_duplicate(service_names, include_deps=start_deps) plans = self._get_convergence_plans(services, strategy) if self.use_networking and self.uses_default_network(): self.ensure_network_exists() self.initialize_volumes() return [ container for service in services for container in service.execute_convergence_plan( plans[service.name], do_build=do_build, timeout=timeout, detached=detached ) ] def _get_convergence_plans(self, services, strategy): plans = {} for service in services: updated_dependencies = [ name for name in service.get_dependency_names() if name in plans and plans[name].action in ('recreate', 'create') ] if updated_dependencies and strategy.allows_recreate: log.debug('%s has upstream changes (%s)', service.name, ", ".join(updated_dependencies)) plan = service.convergence_plan(ConvergenceStrategy.always) else: plan = service.convergence_plan(strategy) plans[service.name] = plan return plans def pull(self, service_names=None, ignore_pull_failures=False): for service in self.get_services(service_names, include_deps=False): service.pull(ignore_pull_failures) def containers(self, service_names=None, stopped=False, one_off=False): if service_names: self.validate_service_names(service_names) else: service_names = self.service_names containers = list(filter(None, [ Container.from_ps(self.client, container) for container in self.client.containers( all=stopped, filters={'label': self.labels(one_off=one_off)})])) def matches_service_names(container): return container.labels.get(LABEL_SERVICE) in service_names return [c for c in containers if matches_service_names(c)] def get_network(self): try: return self.client.inspect_network(self.name) except NotFound: return None def ensure_network_exists(self): # TODO: recreate network if driver has changed? if self.get_network() is None: driver_name = 'the default driver' if self.network_driver: driver_name = 'driver "{}"'.format(self.network_driver) log.info( 'Creating network "{}" with {}' .format(self.name, driver_name) ) self.client.create_network(self.name, driver=self.network_driver) def remove_network(self): network = self.get_network() if network: self.client.remove_network(network['Id']) def uses_default_network(self): return any(service.net.mode == self.name for service in self.services) def _inject_deps(self, acc, service): dep_names = service.get_dependency_names() if len(dep_names) > 0: dep_services = self.get_services( service_names=list(set(dep_names)), include_deps=True ) else: dep_services = [] dep_services.append(service) return acc + dep_services def remove_links(service_dicts): services_with_links = [s for s in service_dicts if 'links' in s] if not services_with_links: return if len(services_with_links) == 1: prefix = '"{}" defines'.format(services_with_links[0]['name']) else: prefix = 'Some services ({}) define'.format( ", ".join('"{}"'.format(s['name']) for s in services_with_links)) log.warn( '\n{} links, which are not compatible with Docker networking and will be ignored.\n' 'Future versions of Docker will not support links - you should remove them for ' 'forwards-compatibility.\n'.format(prefix)) for s in services_with_links: del s['links'] class NoSuchService(Exception): def __init__(self, name): self.name = name self.msg = "No such service: %s" % self.name def __str__(self): return self.msg