Viewing file:  config.py (48.92 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
import functools
import logging
import os
import re
import string
import sys
from collections import namedtuple
from itertools import chain
from operator import attrgetter
from operator import itemgetter
import yaml
try:
from functools import cached_property
except ImportError:
from cached_property import cached_property
from . import types
from ..const import COMPOSE_SPEC as VERSION
from ..const import COMPOSEFILE_V1 as V1
from ..utils import build_string_dict
from ..utils import json_hash
from ..utils import parse_bytes
from ..utils import parse_nanoseconds_int
from ..utils import splitdrive
from ..version import ComposeVersion
from .environment import env_vars_from_file
from .environment import Environment
from .environment import split_env
from .errors import CircularReference
from .errors import ComposeFileNotFound
from .errors import ConfigurationError
from .errors import DuplicateOverrideFileFound
from .errors import VERSION_EXPLANATION
from .interpolation import interpolate_environment_variables
from .sort_services import get_container_name_from_network_mode
from .sort_services import get_service_name_from_network_mode
from .sort_services import sort_service_dicts
from .types import MountSpec
from .types import parse_extra_hosts
from .types import parse_restart_spec
from .types import SecurityOpt
from .types import ServiceLink
from .types import ServicePort
from .types import VolumeFromSpec
from .types import VolumeSpec
from .validation import match_named_volumes
from .validation import validate_against_config_schema
from .validation import validate_config_section
from .validation import validate_cpu
from .validation import validate_credential_spec
from .validation import validate_depends_on
from .validation import validate_extends_file_path
from .validation import validate_healthcheck
from .validation import validate_ipc_mode
from .validation import validate_links
from .validation import validate_network_mode
from .validation import validate_pid_mode
from .validation import validate_service_constraints
from .validation import validate_top_level_object
from .validation import validate_ulimits
DOCKER_CONFIG_KEYS = [
'cap_add',
'cap_drop',
'cgroup_parent',
'command',
'cpu_count',
'cpu_percent',
'cpu_period',
'cpu_quota',
'cpu_rt_period',
'cpu_rt_runtime',
'cpu_shares',
'cpus',
'cpuset',
'detach',
'device_cgroup_rules',
'devices',
'dns',
'dns_search',
'dns_opt',
'domainname',
'entrypoint',
'env_file',
'environment',
'extra_hosts',
'group_add',
'hostname',
'healthcheck',
'image',
'ipc',
'isolation',
'labels',
'links',
'mac_address',
'mem_limit',
'mem_reservation',
'memswap_limit',
'mem_swappiness',
'net',
'oom_score_adj',
'oom_kill_disable',
'pid',
'ports',
'privileged',
'read_only',
'restart',
'runtime',
'secrets',
'security_opt',
'shm_size',
'pids_limit',
'stdin_open',
'stop_signal',
'sysctls',
'tty',
'user',
'userns_mode',
'volume_driver',
'volumes',
'volumes_from',
'working_dir',
]
ALLOWED_KEYS = DOCKER_CONFIG_KEYS + [
'blkio_config',
'build',
'container_name',
'credential_spec',
'dockerfile',
'init',
'log_driver',
'log_opt',
'logging',
'network_mode',
'platform',
'profiles',
'scale',
'stop_grace_period',
]
DOCKER_VALID_URL_PREFIXES = (
'http://',
'https://',
'git://',
'github.com/',
'git@',
)
SUPPORTED_FILENAMES = [
'docker-compose.yml',
'docker-compose.yaml',
'compose.yml',
'compose.yaml',
]
DEFAULT_OVERRIDE_FILENAMES = ('docker-compose.override.yml',
'docker-compose.override.yaml',
'compose.override.yml',
'compose.override.yaml')
log = logging.getLogger(__name__)
class ConfigDetails(namedtuple('_ConfigDetails', 'working_dir config_files environment')):
"""
:param working_dir: the directory to use for relative paths in the config
:type working_dir: string
:param config_files: list of configuration files to load
:type config_files: list of :class:`ConfigFile`
:param environment: computed environment values for this project
:type environment: :class:`environment.Environment`
"""
def __new__(cls, working_dir, config_files, environment=None):
if environment is None:
environment = Environment.from_env_file(working_dir)
return super().__new__(
cls, working_dir, config_files, environment
)
class ConfigFile(namedtuple('_ConfigFile', 'filename config')):
"""
:param filename: filename of the config file
:type filename: string
:param config: contents of the config file
:type config: :class:`dict`
"""
@classmethod
def from_filename(cls, filename):
return cls(filename, load_yaml(filename))
@cached_property
def config_version(self):
version = self.config.get('version', None)
if isinstance(version, dict):
return V1
return ComposeVersion(version) if version else self.version
@cached_property
def version(self):
version = self.config.get('version', None)
if not version:
# no version is specified in the config file
services = self.config.get('services', None)
networks = self.config.get('networks', None)
volumes = self.config.get('volumes', None)
if services or networks or volumes:
# validate V2/V3 structure
for section in ['services', 'networks', 'volumes']:
validate_config_section(
self.filename, self.config.get(section, {}), section)
return VERSION
# validate V1 structure
validate_config_section(
self.filename, self.config, 'services')
return V1
if isinstance(version, dict):
log.warning('Unexpected type for "version" key in "{}". Assuming '
'"version" is the name of a service, and defaulting to '
'Compose file version {}.'.format(self.filename, V1))
return V1
if not isinstance(version, str):
raise ConfigurationError(
'Version in "{}" is invalid - it should be a string.'
.format(self.filename))
if isinstance(version, str):
version_pattern = re.compile(r"^[1-3]+(\.\d+)?$")
if not version_pattern.match(version):
raise ConfigurationError(
'Version "{}" in "{}" is invalid.'
.format(version, self.filename))
if version.startswith("1"):
raise ConfigurationError(
'Version in "{}" is invalid. {}'
.format(self.filename, VERSION_EXPLANATION)
)
return VERSION
def get_service(self, name):
return self.get_service_dicts()[name]
def get_service_dicts(self):
if self.version == V1:
return self.config
return self.config.get('services', {})
def get_volumes(self):
return {} if self.version == V1 else self.config.get('volumes', {})
def get_networks(self):
return {} if self.version == V1 else self.config.get('networks', {})
def get_secrets(self):
return {} if self.version == V1 else self.config.get('secrets', {})
def get_configs(self):
return {} if self.version == V1 else self.config.get('configs', {})
class Config(namedtuple('_Config', 'config_version version services volumes networks secrets configs')):
"""
:param config_version: configuration file version
:type config_version: int
:param version: configuration version
:type version: int
:param services: List of service description dictionaries
:type services: :class:`list`
:param volumes: Dictionary mapping volume names to description dictionaries
:type volumes: :class:`dict`
:param networks: Dictionary mapping network names to description dictionaries
:type networks: :class:`dict`
:param secrets: Dictionary mapping secret names to description dictionaries
:type secrets: :class:`dict`
:param configs: Dictionary mapping config names to description dictionaries
:type configs: :class:`dict`
"""
class ServiceConfig(namedtuple('_ServiceConfig', 'working_dir filename name config')):
@classmethod
def with_abs_paths(cls, working_dir, filename, name, config):
if not working_dir:
raise ValueError("No working_dir for ServiceConfig.")
return cls(
os.path.abspath(working_dir),
os.path.abspath(filename) if filename else filename,
name,
config)
def find(base_dir, filenames, environment, override_dir=None):
if filenames == ['-']:
return ConfigDetails(
os.path.abspath(override_dir) if override_dir else os.getcwd(),
[ConfigFile(None, yaml.safe_load(sys.stdin))],
environment
)
if filenames:
filenames = [os.path.join(base_dir, f) for f in filenames]
else:
# search for compose files in the base dir and its parents
filenames = get_default_config_files(base_dir)
if not filenames and not override_dir:
# none found in base_dir and no override_dir defined
raise ComposeFileNotFound(SUPPORTED_FILENAMES)
if not filenames:
# search for compose files in the project directory and its parents
filenames = get_default_config_files(override_dir)
if not filenames:
raise ComposeFileNotFound(SUPPORTED_FILENAMES)
log.debug("Using configuration files: {}".format(",".join(filenames)))
return ConfigDetails(
override_dir if override_dir else os.path.dirname(filenames[0]),
[ConfigFile.from_filename(f) for f in filenames],
environment
)
def validate_config_version(config_files):
main_file = config_files[0]
validate_top_level_object(main_file)
for next_file in config_files[1:]:
validate_top_level_object(next_file)
if main_file.version != next_file.version:
raise ConfigurationError(
"Version mismatch: file {} specifies version {} but "
"extension file {} uses version {}".format(
main_file.filename,
main_file.version,
next_file.filename,
next_file.version))
def get_default_config_files(base_dir):
(candidates, path) = find_candidates_in_parent_dirs(SUPPORTED_FILENAMES, base_dir)
if not candidates:
return None
winner = candidates[0]
if len(candidates) > 1:
log.warning("Found multiple config files with supported names: %s", ", ".join(candidates))
log.warning("Using %s\n", winner)
return [os.path.join(path, winner)] + get_default_override_file(path)
def get_default_override_file(path):
override_files_in_path = [os.path.join(path, override_filename) for override_filename
in DEFAULT_OVERRIDE_FILENAMES
if os.path.exists(os.path.join(path, override_filename))]
if len(override_files_in_path) > 1:
raise DuplicateOverrideFileFound(override_files_in_path)
return override_files_in_path
def find_candidates_in_parent_dirs(filenames, path):
"""
Given a directory path to start, looks for filenames in the
directory, and then each parent directory successively,
until found.
Returns tuple (candidates, path).
"""
candidates = [filename for filename in filenames
if os.path.exists(os.path.join(path, filename))]
if not candidates:
parent_dir = os.path.join(path, '..')
if os.path.abspath(parent_dir) != os.path.abspath(path):
return find_candidates_in_parent_dirs(filenames, parent_dir)
return (candidates, path)
def check_swarm_only_config(service_dicts):
warning_template = (
"Some services ({services}) use the '{key}' key, which will be ignored. "
"Compose does not support '{key}' configuration - use "
"`docker stack deploy` to deploy to a swarm."
)
key = 'configs'
services = [s for s in service_dicts if s.get(key)]
if services:
log.warning(
warning_template.format(
services=", ".join(sorted(s['name'] for s in services)),
key=key
)
)
def load(config_details, interpolate=True):
"""Load the configuration from a working directory and a list of
configuration files. Files are loaded in order, and merged on top
of each other to create the final configuration.
Return a fully interpolated, extended and validated configuration.
"""
# validate against latest version and if fails do it against v1 schema
validate_config_version(config_details.config_files)
processed_files = [
process_config_file(config_file, config_details.environment, interpolate=interpolate)
for config_file in config_details.config_files
]
config_details = config_details._replace(config_files=processed_files)
main_file = config_details.config_files[0]
volumes = load_mapping(
config_details.config_files, 'get_volumes', 'Volume'
)
networks = load_mapping(
config_details.config_files, 'get_networks', 'Network'
)
secrets = load_mapping(
config_details.config_files, 'get_secrets', 'Secret', config_details.working_dir
)
configs = load_mapping(
config_details.config_files, 'get_configs', 'Config', config_details.working_dir
)
service_dicts = load_services(config_details, main_file, interpolate=interpolate)
if main_file.version != V1:
for service_dict in service_dicts:
match_named_volumes(service_dict, volumes)
check_swarm_only_config(service_dicts)
return Config(main_file.config_version, main_file.version,
service_dicts, volumes, networks, secrets, configs)
def load_mapping(config_files, get_func, entity_type, working_dir=None):
mapping = {}
for config_file in config_files:
for name, config in getattr(config_file, get_func)().items():
mapping[name] = config or {}
if not config:
continue
external = config.get('external')
if external:
validate_external(entity_type, name, config, config_file.version)
if isinstance(external, dict):
config['name'] = external.get('name')
elif not config.get('name'):
config['name'] = name
if 'labels' in config:
config['labels'] = parse_labels(config['labels'])
if 'file' in config:
config['file'] = expand_path(working_dir, config['file'])
if 'driver_opts' in config:
config['driver_opts'] = build_string_dict(
config['driver_opts']
)
device = format_device_option(entity_type, config)
if device:
config['driver_opts']['device'] = device
return mapping
def format_device_option(entity_type, config):
if entity_type != 'Volume':
return
# default driver is 'local'
driver = config.get('driver', 'local')
if driver != 'local':
return
o = config['driver_opts'].get('o')
device = config['driver_opts'].get('device')
if o and o == 'bind' and device:
fullpath = os.path.abspath(os.path.expanduser(device))
return fullpath
def validate_external(entity_type, name, config, version):
for k in config.keys():
if entity_type == 'Network' and k == 'driver':
continue
if k not in ['external', 'name']:
raise ConfigurationError(
"{} {} declared as external but specifies additional attributes "
"({}).".format(
entity_type, name, ', '.join(k for k in config if k != 'external')))
def load_services(config_details, config_file, interpolate=True):
def build_service(service_name, service_dict, service_names):
service_config = ServiceConfig.with_abs_paths(
config_details.working_dir,
config_file.filename,
service_name,
service_dict)
resolver = ServiceExtendsResolver(
service_config, config_file, environment=config_details.environment
)
service_dict = process_service(resolver.run())
service_config = service_config._replace(config=service_dict)
validate_service(service_config, service_names, config_file)
service_dict = finalize_service(
service_config,
service_names,
config_file.version,
config_details.environment,
interpolate
)
return service_dict
def build_services(service_config):
service_names = service_config.keys()
return sort_service_dicts([
build_service(name, service_dict, service_names)
for name, service_dict in service_config.items()
])
def merge_services(base, override):
all_service_names = set(base) | set(override)
return {
name: merge_service_dicts_from_files(
base.get(name, {}),
override.get(name, {}),
config_file.version)
for name in all_service_names
}
service_configs = [
file.get_service_dicts() for file in config_details.config_files
]
service_config = functools.reduce(merge_services, service_configs)
return build_services(service_config)
def interpolate_config_section(config_file, config, section, environment):
return interpolate_environment_variables(
config_file.version,
config,
section,
environment
)
def process_config_section(config_file, config, section, environment, interpolate):
validate_config_section(config_file.filename, config, section)
if interpolate:
return interpolate_environment_variables(
config_file.version,
config,
section,
environment)
else:
return config
def process_config_file(config_file, environment, service_name=None, interpolate=True):
services = process_config_section(
config_file,
config_file.get_service_dicts(),
'service',
environment,
interpolate,
)
if config_file.version > V1:
processed_config = dict(config_file.config)
processed_config['services'] = services
processed_config['volumes'] = process_config_section(
config_file,
config_file.get_volumes(),
'volume',
environment,
interpolate,
)
processed_config['networks'] = process_config_section(
config_file,
config_file.get_networks(),
'network',
environment,
interpolate,
)
processed_config['secrets'] = process_config_section(
config_file,
config_file.get_secrets(),
'secret',
environment,
interpolate,
)
processed_config['configs'] = process_config_section(
config_file,
config_file.get_configs(),
'config',
environment,
interpolate,
)
else:
processed_config = services
config_file = config_file._replace(config=processed_config)
validate_against_config_schema(config_file, config_file.version)
if service_name and service_name not in services:
raise ConfigurationError(
"Cannot extend service '{}' in {}: Service not found".format(
service_name, config_file.filename))
return config_file
class ServiceExtendsResolver:
def __init__(self, service_config, config_file, environment, already_seen=None):
self.service_config = service_config
self.working_dir = service_config.working_dir
self.already_seen = already_seen or []
self.config_file = config_file
self.environment = environment
@property
def signature(self):
return self.service_config.filename, self.service_config.name
def detect_cycle(self):
if self.signature in self.already_seen:
raise CircularReference(self.already_seen + [self.signature])
def run(self):
self.detect_cycle()
if 'extends' in self.service_config.config:
service_dict = self.resolve_extends(*self.validate_and_construct_extends())
return self.service_config._replace(config=service_dict)
return self.service_config
def validate_and_construct_extends(self):
extends = self.service_config.config['extends']
if not isinstance(extends, dict):
extends = {'service': extends}
config_path = self.get_extended_config_path(extends)
service_name = extends['service']
if config_path == os.path.abspath(self.config_file.filename):
try:
service_config = self.config_file.get_service(service_name)
except KeyError:
raise ConfigurationError(
"Cannot extend service '{}' in {}: Service not found".format(
service_name, config_path)
)
else:
extends_file = ConfigFile.from_filename(config_path)
validate_config_version([self.config_file, extends_file])
extended_file = process_config_file(
extends_file, self.environment, service_name=service_name
)
service_config = extended_file.get_service(service_name)
return config_path, service_config, service_name
def resolve_extends(self, extended_config_path, service_dict, service_name):
resolver = ServiceExtendsResolver(
ServiceConfig.with_abs_paths(
os.path.dirname(extended_config_path),
extended_config_path,
service_name,
service_dict),
self.config_file,
already_seen=self.already_seen + [self.signature],
environment=self.environment
)
service_config = resolver.run()
other_service_dict = process_service(service_config)
validate_extended_service_dict(
other_service_dict,
extended_config_path,
service_name)
return merge_service_dicts(
other_service_dict,
self.service_config.config,
self.config_file.version)
def get_extended_config_path(self, extends_options):
"""Service we are extending either has a value for 'file' set, which we
need to obtain a full path too or we are extending from a service
defined in our own file.
"""
filename = self.service_config.filename
validate_extends_file_path(
self.service_config.name,
extends_options,
filename)
if 'file' in extends_options:
return expand_path(self.working_dir, extends_options['file'])
return filename
def resolve_environment(service_dict, environment=None, interpolate=True):
"""Unpack any environment variables from an env_file, if set.
Interpolate environment values if set.
"""
env = {}
for env_file in service_dict.get('env_file', []):
env.update(env_vars_from_file(env_file, interpolate))
env.update(parse_environment(service_dict.get('environment')))
return dict(resolve_env_var(k, v, environment) for k, v in env.items())
def resolve_build_args(buildargs, environment):
args = parse_build_arguments(buildargs)
return dict(resolve_env_var(k, v, environment) for k, v in args.items())
def validate_extended_service_dict(service_dict, filename, service):
error_prefix = "Cannot extend service '{}' in {}:".format(service, filename)
if 'links' in service_dict:
raise ConfigurationError(
"%s services with 'links' cannot be extended" % error_prefix)
if 'volumes_from' in service_dict:
raise ConfigurationError(
"%s services with 'volumes_from' cannot be extended" % error_prefix)
if 'net' in service_dict:
if get_container_name_from_network_mode(service_dict['net']):
raise ConfigurationError(
"%s services with 'net: container' cannot be extended" % error_prefix)
if 'network_mode' in service_dict:
if get_service_name_from_network_mode(service_dict['network_mode']):
raise ConfigurationError(
"%s services with 'network_mode: service' cannot be extended" % error_prefix)
if 'depends_on' in service_dict:
raise ConfigurationError(
"%s services with 'depends_on' cannot be extended" % error_prefix)
def validate_service(service_config, service_names, config_file):
def build_image():
args = sys.argv[1:]
if 'pull' in args:
return False
if '--no-build' in args:
return False
return True
service_dict, service_name = service_config.config, service_config.name
validate_service_constraints(service_dict, service_name, config_file)
if build_image():
# We only care about valid paths when actually building images
validate_paths(service_dict)
validate_cpu(service_config)
validate_ulimits(service_config)
validate_ipc_mode(service_config, service_names)
validate_network_mode(service_config, service_names)
validate_pid_mode(service_config, service_names)
validate_depends_on(service_config, service_names)
validate_links(service_config, service_names)
validate_healthcheck(service_config)
validate_credential_spec(service_config)
if not service_dict.get('image') and has_uppercase(service_name):
raise ConfigurationError(
"Service '{name}' contains uppercase characters which are not valid "
"as part of an image name. Either use a lowercase service name or "
"use the `image` field to set a custom name for the service image."
.format(name=service_name))
def process_service(service_config):
working_dir = service_config.working_dir
service_dict = dict(service_config.config)
if 'env_file' in service_dict:
service_dict['env_file'] = [
expand_path(working_dir, path)
for path in to_list(service_dict['env_file'])
]
if 'build' in service_dict:
process_build_section(service_dict, working_dir)
if 'volumes' in service_dict and service_dict.get('volume_driver') is None:
service_dict['volumes'] = resolve_volume_paths(working_dir, service_dict)
if 'sysctls' in service_dict:
service_dict['sysctls'] = build_string_dict(parse_sysctls(service_dict['sysctls']))
if 'labels' in service_dict:
service_dict['labels'] = parse_labels(service_dict['labels'])
service_dict = process_depends_on(service_dict)
for field in ['dns', 'dns_search', 'tmpfs']:
if field in service_dict:
service_dict[field] = to_list(service_dict[field])
service_dict = process_security_opt(process_blkio_config(process_ports(
process_healthcheck(service_dict)
)))
return service_dict
def process_build_section(service_dict, working_dir):
if isinstance(service_dict['build'], str):
service_dict['build'] = resolve_build_path(working_dir, service_dict['build'])
elif isinstance(service_dict['build'], dict):
if 'context' in service_dict['build']:
path = service_dict['build']['context']
service_dict['build']['context'] = resolve_build_path(working_dir, path)
if 'labels' in service_dict['build']:
service_dict['build']['labels'] = parse_labels(service_dict['build']['labels'])
def process_ports(service_dict):
if 'ports' not in service_dict:
return service_dict
ports = []
for port_definition in service_dict['ports']:
if isinstance(port_definition, ServicePort):
ports.append(port_definition)
else:
ports.extend(ServicePort.parse(port_definition))
service_dict['ports'] = ports
return service_dict
def process_depends_on(service_dict):
if 'depends_on' in service_dict and not isinstance(service_dict['depends_on'], dict):
service_dict['depends_on'] = {
svc: {'condition': 'service_started'} for svc in service_dict['depends_on']
}
return service_dict
def process_blkio_config(service_dict):
if not service_dict.get('blkio_config'):
return service_dict
for field in ['device_read_bps', 'device_write_bps']:
if field in service_dict['blkio_config']:
for v in service_dict['blkio_config'].get(field, []):
rate = v.get('rate', 0)
v['rate'] = parse_bytes(rate)
if v['rate'] is None:
raise ConfigurationError('Invalid format for bytes value: "{}"'.format(rate))
for field in ['device_read_iops', 'device_write_iops']:
if field in service_dict['blkio_config']:
for v in service_dict['blkio_config'].get(field, []):
try:
v['rate'] = int(v.get('rate', 0))
except ValueError:
raise ConfigurationError(
'Invalid IOPS value: "{}". Must be a positive integer.'.format(v.get('rate'))
)
return service_dict
def process_healthcheck(service_dict):
if 'healthcheck' not in service_dict:
return service_dict
hc = service_dict['healthcheck']
if 'disable' in hc:
del hc['disable']
hc['test'] = ['NONE']
for field in ['interval', 'timeout', 'start_period']:
if field not in hc or isinstance(hc[field], int):
continue
hc[field] = parse_nanoseconds_int(hc[field])
return service_dict
def finalize_service_volumes(service_dict, environment):
if 'volumes' in service_dict:
finalized_volumes = []
normalize = environment.get_boolean('COMPOSE_CONVERT_WINDOWS_PATHS')
win_host = environment.get_boolean('COMPOSE_FORCE_WINDOWS_HOST')
for v in service_dict['volumes']:
if isinstance(v, dict):
finalized_volumes.append(MountSpec.parse(v, normalize, win_host))
else:
finalized_volumes.append(VolumeSpec.parse(v, normalize, win_host))
duplicate_mounts = []
mounts = [v.as_volume_spec() if isinstance(v, MountSpec) else v for v in finalized_volumes]
for mount in mounts:
if list(map(attrgetter('internal'), mounts)).count(mount.internal) > 1:
duplicate_mounts.append(mount.repr())
if duplicate_mounts:
raise ConfigurationError("Duplicate mount points: [%s]" % (
', '.join(duplicate_mounts)))
service_dict['volumes'] = finalized_volumes
return service_dict
def finalize_service(service_config, service_names, version, environment,
interpolate=True):
service_dict = dict(service_config.config)
if 'environment' in service_dict or 'env_file' in service_dict:
service_dict['environment'] = resolve_environment(service_dict, environment, interpolate)
service_dict.pop('env_file', None)
if 'volumes_from' in service_dict:
service_dict['volumes_from'] = [
VolumeFromSpec.parse(vf, service_names, version)
for vf in service_dict['volumes_from']
]
service_dict = finalize_service_volumes(service_dict, environment)
if 'net' in service_dict:
network_mode = service_dict.pop('net')
container_name = get_container_name_from_network_mode(network_mode)
if container_name and container_name in service_names:
service_dict['network_mode'] = 'service:{}'.format(container_name)
else:
service_dict['network_mode'] = network_mode
if 'networks' in service_dict:
service_dict['networks'] = parse_networks(service_dict['networks'])
if 'restart' in service_dict:
service_dict['restart'] = parse_restart_spec(service_dict['restart'])
if 'secrets' in service_dict:
service_dict['secrets'] = [
types.ServiceSecret.parse(s) for s in service_dict['secrets']
]
if 'configs' in service_dict:
service_dict['configs'] = [
types.ServiceConfig.parse(c) for c in service_dict['configs']
]
normalize_build(service_dict, service_config.working_dir, environment)
service_dict['name'] = service_config.name
return normalize_v1_service_format(service_dict)
def normalize_v1_service_format(service_dict):
if 'log_driver' in service_dict or 'log_opt' in service_dict:
if 'logging' not in service_dict:
service_dict['logging'] = {}
if 'log_driver' in service_dict:
service_dict['logging']['driver'] = service_dict['log_driver']
del service_dict['log_driver']
if 'log_opt' in service_dict:
service_dict['logging']['options'] = service_dict['log_opt']
del service_dict['log_opt']
if 'dockerfile' in service_dict:
service_dict['build'] = service_dict.get('build', {})
service_dict['build'].update({
'dockerfile': service_dict.pop('dockerfile')
})
return service_dict
def merge_service_dicts_from_files(base, override, version):
"""When merging services from multiple files we need to merge the `extends`
field. This is not handled by `merge_service_dicts()` which is used to
perform the `extends`.
"""
new_service = merge_service_dicts(base, override, version)
if 'extends' in override:
new_service['extends'] = override['extends']
elif 'extends' in base:
new_service['extends'] = base['extends']
return new_service
class MergeDict(dict):
"""A dict-like object responsible for merging two dicts into one."""
def __init__(self, base, override):
self.base = base
self.override = override
def needs_merge(self, field):
return field in self.base or field in self.override
def merge_field(self, field, merge_func, default=None):
if not self.needs_merge(field):
return
self[field] = merge_func(
self.base.get(field, default),
self.override.get(field, default))
def merge_mapping(self, field, parse_func=None):
if not self.needs_merge(field):
return
if parse_func is None:
def parse_func(m):
return m or {}
self[field] = parse_func(self.base.get(field))
self[field].update(parse_func(self.override.get(field)))
def merge_sequence(self, field, parse_func):
def parse_sequence_func(seq):
return to_mapping((parse_func(item) for item in seq), 'merge_field')
if not self.needs_merge(field):
return
merged = parse_sequence_func(self.base.get(field, []))
merged.update(parse_sequence_func(self.override.get(field, [])))
self[field] = [item.repr() for item in sorted(merged.values())]
def merge_scalar(self, field):
if self.needs_merge(field):
self[field] = self.override.get(field, self.base.get(field))
def merge_service_dicts(base, override, version):
md = MergeDict(base, override)
md.merge_mapping('environment', parse_environment)
md.merge_mapping('labels', parse_labels)
md.merge_mapping('ulimits', parse_flat_dict)
md.merge_mapping('sysctls', parse_sysctls)
md.merge_mapping('depends_on', parse_depends_on)
md.merge_mapping('storage_opt', parse_flat_dict)
md.merge_sequence('links', ServiceLink.parse)
md.merge_sequence('secrets', types.ServiceSecret.parse)
md.merge_sequence('configs', types.ServiceConfig.parse)
md.merge_sequence('security_opt', types.SecurityOpt.parse)
md.merge_mapping('extra_hosts', parse_extra_hosts)
md.merge_field('networks', merge_networks, default={})
for field in ['volumes', 'devices']:
md.merge_field(field, merge_path_mappings)
for field in [
'cap_add', 'cap_drop', 'expose', 'external_links',
'volumes_from', 'device_cgroup_rules', 'profiles',
]:
md.merge_field(field, merge_unique_items_lists, default=[])
for field in ['dns', 'dns_search', 'env_file', 'tmpfs']:
md.merge_field(field, merge_list_or_string)
md.merge_field('logging', merge_logging, default={})
merge_ports(md, base, override)
md.merge_field('blkio_config', merge_blkio_config, default={})
md.merge_field('healthcheck', merge_healthchecks, default={})
md.merge_field('deploy', merge_deploy, default={})
for field in set(ALLOWED_KEYS) - set(md):
md.merge_scalar(field)
if version == V1:
legacy_v1_merge_image_or_build(md, base, override)
elif md.needs_merge('build'):
md['build'] = merge_build(md, base, override)
return dict(md)
def merge_unique_items_lists(base, override):
override = (str(o) for o in override)
base = (str(b) for b in base)
return sorted(set(chain(base, override)))
def merge_healthchecks(base, override):
if override.get('disabled') is True:
return override
result = base.copy()
result.update(override)
return result
def merge_ports(md, base, override):
def parse_sequence_func(seq):
acc = [s for item in seq for s in ServicePort.parse(item)]
return to_mapping(acc, 'merge_field')
field = 'ports'
if not md.needs_merge(field):
return
merged = parse_sequence_func(md.base.get(field, []))
merged.update(parse_sequence_func(md.override.get(field, [])))
md[field] = [item for item in sorted(merged.values(), key=attrgetter("target"))]
def merge_build(output, base, override):
def to_dict(service):
build_config = service.get('build', {})
if isinstance(build_config, str):
return {'context': build_config}
return build_config
md = MergeDict(to_dict(base), to_dict(override))
md.merge_scalar('context')
md.merge_scalar('dockerfile')
md.merge_scalar('network')
md.merge_scalar('target')
md.merge_scalar('shm_size')
md.merge_scalar('isolation')
md.merge_mapping('args', parse_build_arguments)
md.merge_field('cache_from', merge_unique_items_lists, default=[])
md.merge_mapping('labels', parse_labels)
md.merge_mapping('extra_hosts', parse_extra_hosts)
return dict(md)
def merge_deploy(base, override):
md = MergeDict(base or {}, override or {})
md.merge_scalar('mode')
md.merge_scalar('endpoint_mode')
md.merge_scalar('replicas')
md.merge_mapping('labels', parse_labels)
md.merge_mapping('update_config')
md.merge_mapping('rollback_config')
md.merge_mapping('restart_policy')
if md.needs_merge('resources'):
resources_md = MergeDict(md.base.get('resources') or {}, md.override.get('resources') or {})
resources_md.merge_mapping('limits')
resources_md.merge_field('reservations', merge_reservations, default={})
md['resources'] = dict(resources_md)
if md.needs_merge('placement'):
placement_md = MergeDict(md.base.get('placement') or {}, md.override.get('placement') or {})
placement_md.merge_scalar('max_replicas_per_node')
placement_md.merge_field('constraints', merge_unique_items_lists, default=[])
placement_md.merge_field('preferences', merge_unique_objects_lists, default=[])
md['placement'] = dict(placement_md)
return dict(md)
def merge_networks(base, override):
merged_networks = {}
all_network_names = set(base) | set(override)
base = {k: {} for k in base} if isinstance(base, list) else base
override = {k: {} for k in override} if isinstance(override, list) else override
for network_name in all_network_names:
md = MergeDict(base.get(network_name) or {}, override.get(network_name) or {})
md.merge_field('aliases', merge_unique_items_lists, [])
md.merge_field('link_local_ips', merge_unique_items_lists, [])
md.merge_scalar('priority')
md.merge_scalar('ipv4_address')
md.merge_scalar('ipv6_address')
merged_networks[network_name] = dict(md)
return merged_networks
def merge_reservations(base, override):
md = MergeDict(base, override)
md.merge_scalar('cpus')
md.merge_scalar('memory')
md.merge_sequence('generic_resources', types.GenericResource.parse)
md.merge_field('devices', merge_unique_objects_lists, default=[])
return dict(md)
def merge_unique_objects_lists(base, override):
result = {json_hash(i): i for i in base + override}
return [i[1] for i in sorted(((k, v) for k, v in result.items()), key=itemgetter(0))]
def merge_blkio_config(base, override):
md = MergeDict(base, override)
md.merge_scalar('weight')
def merge_blkio_limits(base, override):
get_path = itemgetter('path')
index = {get_path(b): b for b in base}
index.update((get_path(o), o) for o in override)
return sorted(index.values(), key=get_path)
for field in [
"device_read_bps", "device_read_iops", "device_write_bps",
"device_write_iops", "weight_device",
]:
md.merge_field(field, merge_blkio_limits, default=[])
return dict(md)
def merge_logging(base, override):
md = MergeDict(base, override)
md.merge_scalar('driver')
if md.get('driver') == base.get('driver') or base.get('driver') is None:
md.merge_mapping('options', lambda m: m or {})
elif override.get('options'):
md['options'] = override.get('options', {})
return dict(md)
def legacy_v1_merge_image_or_build(output, base, override):
output.pop('image', None)
output.pop('build', None)
if 'image' in override:
output['image'] = override['image']
elif 'build' in override:
output['build'] = override['build']
elif 'image' in base:
output['image'] = base['image']
elif 'build' in base:
output['build'] = base['build']
def merge_environment(base, override):
env = parse_environment(base)
env.update(parse_environment(override))
return env
def merge_labels(base, override):
labels = parse_labels(base)
labels.update(parse_labels(override))
return labels
def split_kv(kvpair):
if '=' in kvpair:
return kvpair.split('=', 1)
else:
return kvpair, ''
def parse_dict_or_list(split_func, type_name, arguments):
if not arguments:
return {}
if isinstance(arguments, list):
return dict(split_func(e) for e in arguments)
if isinstance(arguments, dict):
return dict(arguments)
raise ConfigurationError(
"%s \"%s\" must be a list or mapping," %
(type_name, arguments)
)
parse_build_arguments = functools.partial(parse_dict_or_list, split_env, 'build arguments')
parse_environment = functools.partial(parse_dict_or_list, split_env, 'environment')
parse_labels = functools.partial(parse_dict_or_list, split_kv, 'labels')
parse_networks = functools.partial(parse_dict_or_list, lambda k: (k, None), 'networks')
parse_sysctls = functools.partial(parse_dict_or_list, split_kv, 'sysctls')
parse_depends_on = functools.partial(
parse_dict_or_list, lambda k: (k, {'condition': 'service_started'}), 'depends_on'
)
def parse_flat_dict(d):
if not d:
return {}
if isinstance(d, dict):
return dict(d)
raise ConfigurationError("Invalid type: expected mapping")
def resolve_env_var(key, val, environment):
if val is not None:
return key, val
elif environment and key in environment:
return key, environment[key]
else:
return key, None
def resolve_volume_paths(working_dir, service_dict):
return [
resolve_volume_path(working_dir, volume)
for volume in service_dict['volumes']
]
def resolve_volume_path(working_dir, volume):
if isinstance(volume, dict):
if volume.get('source', '').startswith(('.', '~')) and volume['type'] == 'bind':
volume['source'] = expand_path(working_dir, volume['source'])
return volume
mount_params = None
container_path, mount_params = split_path_mapping(volume)
if mount_params is not None:
host_path, mode = mount_params
if host_path is None:
return container_path
if host_path.startswith('.'):
host_path = expand_path(working_dir, host_path)
host_path = os.path.expanduser(host_path)
return "{}:{}{}".format(host_path, container_path, (':' + mode if mode else ''))
return container_path
def normalize_build(service_dict, working_dir, environment):
if 'build' in service_dict:
build = {}
# Shortcut where specifying a string is treated as the build context
if isinstance(service_dict['build'], str):
build['context'] = service_dict.pop('build')
else:
build.update(service_dict['build'])
if 'args' in build:
build['args'] = build_string_dict(
resolve_build_args(build.get('args'), environment)
)
service_dict['build'] = build
def resolve_build_path(working_dir, build_path):
if is_url(build_path):
return build_path
return expand_path(working_dir, build_path)
def is_url(build_path):
return build_path.startswith(DOCKER_VALID_URL_PREFIXES)
def validate_paths(service_dict):
if 'build' in service_dict:
build = service_dict.get('build', {})
if isinstance(build, str):
build_path = build
elif isinstance(build, dict) and 'context' in build:
build_path = build['context']
else:
# We have a build section but no context, so nothing to validate
return
if (
not is_url(build_path) and
(not os.path.exists(build_path) or not os.access(build_path, os.R_OK))
):
raise ConfigurationError(
"build path %s either does not exist, is not accessible, "
"or is not a valid URL." % build_path)
def merge_path_mappings(base, override):
d = dict_from_path_mappings(base)
d.update(dict_from_path_mappings(override))
return path_mappings_from_dict(d)
def dict_from_path_mappings(path_mappings):
if path_mappings:
return dict(split_path_mapping(v) for v in path_mappings)
else:
return {}
def path_mappings_from_dict(d):
return [join_path_mapping(v) for v in sorted(d.items())]
def split_path_mapping(volume_path):
"""
Ascertain if the volume_path contains a host path as well as a container
path. Using splitdrive so windows absolute paths won't cause issues with
splitting on ':'.
"""
if isinstance(volume_path, dict):
return (volume_path.get('target'), volume_path)
drive, volume_config = splitdrive(volume_path)
if ':' in volume_config:
(host, container) = volume_config.split(':', 1)
container_drive, container_path = splitdrive(container)
mode = None
if ':' in container_path:
container_path, mode = container_path.rsplit(':', 1)
return (container_drive + container_path, (drive + host, mode))
else:
return (volume_path, None)
def process_security_opt(service_dict):
security_opts = service_dict.get('security_opt', [])
result = []
for value in security_opts:
result.append(SecurityOpt.parse(value))
if result:
service_dict['security_opt'] = result
return service_dict
def join_path_mapping(pair):
(container, host) = pair
if isinstance(host, dict):
return host
elif host is None:
return container
else:
host, mode = host
result = ":".join((host, container))
if mode:
result += ":" + mode
return result
def expand_path(working_dir, path):
return os.path.abspath(os.path.join(working_dir, os.path.expanduser(path)))
def merge_list_or_string(base, override):
return to_list(base) + to_list(override)
def to_list(value):
if value is None:
return []
elif isinstance(value, str):
return [value]
else:
return value
def to_mapping(sequence, key_field):
return {getattr(item, key_field): item for item in sequence}
def has_uppercase(name):
return any(char in string.ascii_uppercase for char in name)
def load_yaml(filename, encoding=None, binary=True):
try:
with open(filename, 'rb' if binary else 'r', encoding=encoding) as fh:
return yaml.safe_load(fh)
except (OSError, yaml.YAMLError, UnicodeDecodeError) as e:
if encoding is None:
# Sometimes the user's locale sets an encoding that doesn't match
# the YAML files. Im such cases, retry once with the "default"
# UTF-8 encoding
return load_yaml(filename, encoding='utf-8-sig', binary=False)
error_name = getattr(e, '__module__', '') + '.' + e.__class__.__name__
raise ConfigurationError("{}: {}".format(error_name, e))
|