project.py 38 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094
  1. import datetime
  2. import enum
  3. import logging
  4. import operator
  5. import re
  6. from functools import reduce
  7. from os import path
  8. from docker.errors import APIError
  9. from docker.errors import ImageNotFound
  10. from docker.errors import NotFound
  11. from docker.utils import version_lt
  12. from . import parallel
  13. from .cli.errors import UserError
  14. from .config import ConfigurationError
  15. from .config.config import V1
  16. from .config.sort_services import get_container_name_from_network_mode
  17. from .config.sort_services import get_service_name_from_network_mode
  18. from .const import LABEL_ONE_OFF
  19. from .const import LABEL_PROJECT
  20. from .const import LABEL_SERVICE
  21. from .container import Container
  22. from .network import build_networks
  23. from .network import get_networks
  24. from .network import ProjectNetworks
  25. from .progress_stream import read_status
  26. from .service import BuildAction
  27. from .service import ContainerIpcMode
  28. from .service import ContainerNetworkMode
  29. from .service import ContainerPidMode
  30. from .service import ConvergenceStrategy
  31. from .service import IpcMode
  32. from .service import NetworkMode
  33. from .service import NoSuchImageError
  34. from .service import parse_repository_tag
  35. from .service import PidMode
  36. from .service import Service
  37. from .service import ServiceIpcMode
  38. from .service import ServiceNetworkMode
  39. from .service import ServicePidMode
  40. from .utils import microseconds_from_time_nano
  41. from .utils import truncate_string
  42. from .volume import ProjectVolumes
  43. log = logging.getLogger(__name__)
  44. @enum.unique
  45. class OneOffFilter(enum.Enum):
  46. include = 0
  47. exclude = 1
  48. only = 2
  49. @classmethod
  50. def update_labels(cls, value, labels):
  51. if value == cls.only:
  52. labels.append('{}={}'.format(LABEL_ONE_OFF, "True"))
  53. elif value == cls.exclude:
  54. labels.append('{}={}'.format(LABEL_ONE_OFF, "False"))
  55. elif value == cls.include:
  56. pass
  57. else:
  58. raise ValueError("Invalid value for one_off: {}".format(repr(value)))
  59. class Project:
  60. """
  61. A collection of services.
  62. """
  63. def __init__(self, name, services, client, networks=None, volumes=None, config_version=None):
  64. self.name = name
  65. self.services = services
  66. self.client = client
  67. self.volumes = volumes or ProjectVolumes({})
  68. self.networks = networks or ProjectNetworks({}, False)
  69. self.config_version = config_version
  70. def labels(self, one_off=OneOffFilter.exclude, legacy=False):
  71. name = self.name
  72. if legacy:
  73. name = re.sub(r'[_-]', '', name)
  74. labels = ['{}={}'.format(LABEL_PROJECT, name)]
  75. OneOffFilter.update_labels(one_off, labels)
  76. return labels
  77. @classmethod
  78. def from_config(cls, name, config_data, client, default_platform=None, extra_labels=None):
  79. """
  80. Construct a Project from a config.Config object.
  81. """
  82. extra_labels = extra_labels or []
  83. use_networking = (config_data.version and config_data.version != V1)
  84. networks = build_networks(name, config_data, client)
  85. project_networks = ProjectNetworks.from_services(
  86. config_data.services,
  87. networks,
  88. use_networking)
  89. volumes = ProjectVolumes.from_config(name, config_data, client)
  90. project = cls(name, [], client, project_networks, volumes, config_data.version)
  91. for service_dict in config_data.services:
  92. service_dict = dict(service_dict)
  93. if use_networking:
  94. service_networks = get_networks(service_dict, networks)
  95. else:
  96. service_networks = {}
  97. service_dict.pop('networks', None)
  98. links = project.get_links(service_dict)
  99. ipc_mode = project.get_ipc_mode(service_dict)
  100. network_mode = project.get_network_mode(
  101. service_dict, list(service_networks.keys())
  102. )
  103. pid_mode = project.get_pid_mode(service_dict)
  104. volumes_from = get_volumes_from(project, service_dict)
  105. if config_data.version != V1:
  106. service_dict['volumes'] = [
  107. volumes.namespace_spec(volume_spec)
  108. for volume_spec in service_dict.get('volumes', [])
  109. ]
  110. secrets = get_secrets(
  111. service_dict['name'],
  112. service_dict.pop('secrets', None) or [],
  113. config_data.secrets)
  114. service_dict['scale'] = project.get_service_scale(service_dict)
  115. service_dict = translate_credential_spec_to_security_opt(service_dict)
  116. service_dict, ignored_keys = translate_deploy_keys_to_container_config(
  117. service_dict
  118. )
  119. if ignored_keys:
  120. log.warning(
  121. 'The following deploy sub-keys are not supported and have'
  122. ' been ignored: {}'.format(', '.join(ignored_keys))
  123. )
  124. project.services.append(
  125. Service(
  126. service_dict.pop('name'),
  127. client=client,
  128. project=name,
  129. use_networking=use_networking,
  130. networks=service_networks,
  131. links=links,
  132. network_mode=network_mode,
  133. volumes_from=volumes_from,
  134. secrets=secrets,
  135. pid_mode=pid_mode,
  136. ipc_mode=ipc_mode,
  137. platform=service_dict.pop('platform', None),
  138. default_platform=default_platform,
  139. extra_labels=extra_labels,
  140. **service_dict)
  141. )
  142. return project
  143. @property
  144. def service_names(self):
  145. return [service.name for service in self.services]
  146. def get_service(self, name):
  147. """
  148. Retrieve a service by name. Raises NoSuchService
  149. if the named service does not exist.
  150. """
  151. for service in self.services:
  152. if service.name == name:
  153. return service
  154. raise NoSuchService(name)
  155. def validate_service_names(self, service_names):
  156. """
  157. Validate that the given list of service names only contains valid
  158. services. Raises NoSuchService if one of the names is invalid.
  159. """
  160. valid_names = self.service_names
  161. for name in service_names:
  162. if name not in valid_names:
  163. raise NoSuchService(name)
  164. def get_services(self, service_names=None, include_deps=False):
  165. """
  166. Returns a list of this project's services filtered
  167. by the provided list of names, or all services if service_names is None
  168. or [].
  169. If include_deps is specified, returns a list including the dependencies for
  170. service_names, in order of dependency.
  171. Preserves the original order of self.services where possible,
  172. reordering as needed to resolve dependencies.
  173. Raises NoSuchService if any of the named services do not exist.
  174. """
  175. if service_names is None or len(service_names) == 0:
  176. service_names = self.service_names
  177. unsorted = [self.get_service(name) for name in service_names]
  178. services = [s for s in self.services if s in unsorted]
  179. if include_deps:
  180. services = reduce(self._inject_deps, services, [])
  181. uniques = []
  182. [uniques.append(s) for s in services if s not in uniques]
  183. return uniques
  184. def get_services_without_duplicate(self, service_names=None, include_deps=False):
  185. services = self.get_services(service_names, include_deps)
  186. for service in services:
  187. service.remove_duplicate_containers()
  188. return services
  189. def get_links(self, service_dict):
  190. links = []
  191. if 'links' in service_dict:
  192. for link in service_dict.get('links', []):
  193. if ':' in link:
  194. service_name, link_name = link.split(':', 1)
  195. else:
  196. service_name, link_name = link, None
  197. try:
  198. links.append((self.get_service(service_name), link_name))
  199. except NoSuchService:
  200. raise ConfigurationError(
  201. 'Service "%s" has a link to service "%s" which does not '
  202. 'exist.' % (service_dict['name'], service_name))
  203. del service_dict['links']
  204. return links
  205. def get_network_mode(self, service_dict, networks):
  206. network_mode = service_dict.pop('network_mode', None)
  207. if not network_mode:
  208. if self.networks.use_networking:
  209. return NetworkMode(networks[0]) if networks else NetworkMode('none')
  210. return NetworkMode(None)
  211. service_name = get_service_name_from_network_mode(network_mode)
  212. if service_name:
  213. return ServiceNetworkMode(self.get_service(service_name))
  214. container_name = get_container_name_from_network_mode(network_mode)
  215. if container_name:
  216. try:
  217. return ContainerNetworkMode(Container.from_id(self.client, container_name))
  218. except APIError:
  219. raise ConfigurationError(
  220. "Service '{name}' uses the network stack of container '{dep}' which "
  221. "does not exist.".format(name=service_dict['name'], dep=container_name))
  222. return NetworkMode(network_mode)
  223. def get_pid_mode(self, service_dict):
  224. pid_mode = service_dict.pop('pid', None)
  225. if not pid_mode:
  226. return PidMode(None)
  227. service_name = get_service_name_from_network_mode(pid_mode)
  228. if service_name:
  229. return ServicePidMode(self.get_service(service_name))
  230. container_name = get_container_name_from_network_mode(pid_mode)
  231. if container_name:
  232. try:
  233. return ContainerPidMode(Container.from_id(self.client, container_name))
  234. except APIError:
  235. raise ConfigurationError(
  236. "Service '{name}' uses the PID namespace of container '{dep}' which "
  237. "does not exist.".format(name=service_dict['name'], dep=container_name)
  238. )
  239. return PidMode(pid_mode)
  240. def get_ipc_mode(self, service_dict):
  241. ipc_mode = service_dict.pop('ipc', None)
  242. if not ipc_mode:
  243. return IpcMode(None)
  244. service_name = get_service_name_from_network_mode(ipc_mode)
  245. if service_name:
  246. return ServiceIpcMode(self.get_service(service_name))
  247. container_name = get_container_name_from_network_mode(ipc_mode)
  248. if container_name:
  249. try:
  250. return ContainerIpcMode(Container.from_id(self.client, container_name))
  251. except APIError:
  252. raise ConfigurationError(
  253. "Service '{name}' uses the IPC namespace of container '{dep}' which "
  254. "does not exist.".format(name=service_dict['name'], dep=container_name)
  255. )
  256. return IpcMode(ipc_mode)
  257. def get_service_scale(self, service_dict):
  258. # service.scale for v2 and deploy.replicas for v3
  259. scale = service_dict.get('scale', None)
  260. deploy_dict = service_dict.get('deploy', None)
  261. if not deploy_dict:
  262. return 1 if scale is None else scale
  263. if deploy_dict.get('mode', 'replicated') != 'replicated':
  264. return 1 if scale is None else scale
  265. replicas = deploy_dict.get('replicas', None)
  266. if scale and replicas:
  267. raise ConfigurationError(
  268. "Both service.scale and service.deploy.replicas are set."
  269. " Only one of them must be set."
  270. )
  271. if replicas:
  272. scale = replicas
  273. # deploy may contain placement constraints introduced in v3.8
  274. max_replicas = deploy_dict.get('placement', {}).get(
  275. 'max_replicas_per_node',
  276. scale)
  277. scale = min(scale, max_replicas)
  278. if max_replicas < scale:
  279. log.warning("Scale is limited to {} ('max_replicas_per_node' field).".format(
  280. max_replicas))
  281. return scale
  282. def start(self, service_names=None, **options):
  283. containers = []
  284. def start_service(service):
  285. service_containers = service.start(quiet=True, **options)
  286. containers.extend(service_containers)
  287. services = self.get_services(service_names)
  288. def get_deps(service):
  289. return {
  290. (self.get_service(dep), config)
  291. for dep, config in service.get_dependency_configs().items()
  292. }
  293. parallel.parallel_execute(
  294. services,
  295. start_service,
  296. operator.attrgetter('name'),
  297. 'Starting',
  298. get_deps,
  299. fail_check=lambda obj: not obj.containers(),
  300. )
  301. return containers
  302. def stop(self, service_names=None, one_off=OneOffFilter.exclude, **options):
  303. containers = self.containers(service_names, one_off=one_off)
  304. def get_deps(container):
  305. # actually returning inversed dependencies
  306. return {(other, None) for other in containers
  307. if container.service in
  308. self.get_service(other.service).get_dependency_names()}
  309. parallel.parallel_execute(
  310. containers,
  311. self.build_container_operation_with_timeout_func('stop', options),
  312. operator.attrgetter('name'),
  313. 'Stopping',
  314. get_deps,
  315. )
  316. def pause(self, service_names=None, **options):
  317. containers = self.containers(service_names)
  318. parallel.parallel_pause(reversed(containers), options)
  319. return containers
  320. def unpause(self, service_names=None, **options):
  321. containers = self.containers(service_names)
  322. parallel.parallel_unpause(containers, options)
  323. return containers
  324. def kill(self, service_names=None, **options):
  325. parallel.parallel_kill(self.containers(service_names), options)
  326. def remove_stopped(self, service_names=None, one_off=OneOffFilter.exclude, **options):
  327. parallel.parallel_remove(self.containers(
  328. service_names, stopped=True, one_off=one_off
  329. ), options)
  330. def down(
  331. self,
  332. remove_image_type,
  333. include_volumes,
  334. remove_orphans=False,
  335. timeout=None,
  336. ignore_orphans=False):
  337. self.stop(one_off=OneOffFilter.include, timeout=timeout)
  338. if not ignore_orphans:
  339. self.find_orphan_containers(remove_orphans)
  340. self.remove_stopped(v=include_volumes, one_off=OneOffFilter.include)
  341. self.networks.remove()
  342. if include_volumes:
  343. self.volumes.remove()
  344. self.remove_images(remove_image_type)
  345. def remove_images(self, remove_image_type):
  346. for service in self.get_services():
  347. service.remove_image(remove_image_type)
  348. def restart(self, service_names=None, **options):
  349. containers = self.containers(service_names, stopped=True)
  350. parallel.parallel_execute(
  351. containers,
  352. self.build_container_operation_with_timeout_func('restart', options),
  353. operator.attrgetter('name'),
  354. 'Restarting',
  355. )
  356. return containers
  357. def build(self, service_names=None, no_cache=False, pull=False, force_rm=False, memory=None,
  358. build_args=None, gzip=False, parallel_build=False, rm=True, silent=False, cli=False,
  359. progress=None):
  360. services = []
  361. for service in self.get_services(service_names):
  362. if service.can_be_built():
  363. services.append(service)
  364. elif not silent:
  365. log.info('%s uses an image, skipping' % service.name)
  366. if cli:
  367. log.warning("Native build is an experimental feature and could change at any time")
  368. if parallel_build:
  369. log.warning("Flag '--parallel' is ignored when building with "
  370. "COMPOSE_DOCKER_CLI_BUILD=1")
  371. if gzip:
  372. log.warning("Flag '--compress' is ignored when building with "
  373. "COMPOSE_DOCKER_CLI_BUILD=1")
  374. def build_service(service):
  375. service.build(no_cache, pull, force_rm, memory, build_args, gzip, rm, silent, cli, progress)
  376. if parallel_build:
  377. _, errors = parallel.parallel_execute(
  378. services,
  379. build_service,
  380. operator.attrgetter('name'),
  381. 'Building',
  382. limit=5,
  383. )
  384. if len(errors):
  385. combined_errors = '\n'.join([
  386. e.decode('utf-8') if isinstance(e, bytes) else e for e in errors.values()
  387. ])
  388. raise ProjectError(combined_errors)
  389. else:
  390. for service in services:
  391. build_service(service)
  392. def create(
  393. self,
  394. service_names=None,
  395. strategy=ConvergenceStrategy.changed,
  396. do_build=BuildAction.none,
  397. ):
  398. services = self.get_services_without_duplicate(service_names, include_deps=True)
  399. for svc in services:
  400. svc.ensure_image_exists(do_build=do_build)
  401. plans = self._get_convergence_plans(services, strategy)
  402. for service in services:
  403. service.execute_convergence_plan(
  404. plans[service.name],
  405. detached=True,
  406. start=False)
  407. def _legacy_event_processor(self, service_names):
  408. # Only for v1 files or when Compose is forced to use an older API version
  409. def build_container_event(event, container):
  410. time = datetime.datetime.fromtimestamp(event['time'])
  411. time = time.replace(
  412. microsecond=microseconds_from_time_nano(event['timeNano'])
  413. )
  414. return {
  415. 'time': time,
  416. 'type': 'container',
  417. 'action': event['status'],
  418. 'id': container.id,
  419. 'service': container.service,
  420. 'attributes': {
  421. 'name': container.name,
  422. 'image': event['from'],
  423. },
  424. 'container': container,
  425. }
  426. service_names = set(service_names or self.service_names)
  427. for event in self.client.events(
  428. filters={'label': self.labels()},
  429. decode=True
  430. ):
  431. # This is a guard against some events broadcasted by swarm that
  432. # don't have a status field.
  433. # See https://github.com/docker/compose/issues/3316
  434. if 'status' not in event:
  435. continue
  436. try:
  437. # this can fail if the container has been removed or if the event
  438. # refers to an image
  439. container = Container.from_id(self.client, event['id'])
  440. except APIError:
  441. continue
  442. if container.service not in service_names:
  443. continue
  444. yield build_container_event(event, container)
  445. def events(self, service_names=None):
  446. if version_lt(self.client.api_version, '1.22'):
  447. # New, better event API was introduced in 1.22.
  448. return self._legacy_event_processor(service_names)
  449. def build_container_event(event):
  450. container_attrs = event['Actor']['Attributes']
  451. time = datetime.datetime.fromtimestamp(event['time'])
  452. time = time.replace(
  453. microsecond=microseconds_from_time_nano(event['timeNano'])
  454. )
  455. container = None
  456. try:
  457. container = Container.from_id(self.client, event['id'])
  458. except APIError:
  459. # Container may have been removed (e.g. if this is a destroy event)
  460. pass
  461. return {
  462. 'time': time,
  463. 'type': 'container',
  464. 'action': event['status'],
  465. 'id': event['Actor']['ID'],
  466. 'service': container_attrs.get(LABEL_SERVICE),
  467. 'attributes': {
  468. k: v for k, v in container_attrs.items()
  469. if not k.startswith('com.docker.compose.')
  470. },
  471. 'container': container,
  472. }
  473. def yield_loop(service_names):
  474. for event in self.client.events(
  475. filters={'label': self.labels()},
  476. decode=True
  477. ):
  478. # TODO: support other event types
  479. if event.get('Type') != 'container':
  480. continue
  481. try:
  482. if event['Actor']['Attributes'][LABEL_SERVICE] not in service_names:
  483. continue
  484. except KeyError:
  485. continue
  486. yield build_container_event(event)
  487. return yield_loop(set(service_names) if service_names else self.service_names)
  488. def up(self,
  489. service_names=None,
  490. start_deps=True,
  491. strategy=ConvergenceStrategy.changed,
  492. do_build=BuildAction.none,
  493. timeout=None,
  494. detached=False,
  495. remove_orphans=False,
  496. ignore_orphans=False,
  497. scale_override=None,
  498. rescale=True,
  499. start=True,
  500. always_recreate_deps=False,
  501. reset_container_image=False,
  502. renew_anonymous_volumes=False,
  503. silent=False,
  504. cli=False,
  505. one_off=False,
  506. override_options=None,
  507. ):
  508. if cli:
  509. log.warning("Native build is an experimental feature and could change at any time")
  510. self.initialize()
  511. if not ignore_orphans:
  512. self.find_orphan_containers(remove_orphans)
  513. if scale_override is None:
  514. scale_override = {}
  515. services = self.get_services_without_duplicate(
  516. service_names,
  517. include_deps=start_deps)
  518. for svc in services:
  519. svc.ensure_image_exists(do_build=do_build, silent=silent, cli=cli)
  520. plans = self._get_convergence_plans(
  521. services,
  522. strategy,
  523. always_recreate_deps=always_recreate_deps,
  524. one_off=service_names if one_off else [],
  525. )
  526. def do(service):
  527. return service.execute_convergence_plan(
  528. plans[service.name],
  529. timeout=timeout,
  530. detached=detached,
  531. scale_override=scale_override.get(service.name),
  532. rescale=rescale,
  533. start=start,
  534. reset_container_image=reset_container_image,
  535. renew_anonymous_volumes=renew_anonymous_volumes,
  536. override_options=override_options,
  537. )
  538. def get_deps(service):
  539. return {
  540. (self.get_service(dep), config)
  541. for dep, config in service.get_dependency_configs().items()
  542. }
  543. results, errors = parallel.parallel_execute(
  544. services,
  545. do,
  546. operator.attrgetter('name'),
  547. None,
  548. get_deps,
  549. )
  550. if errors:
  551. raise ProjectError(
  552. 'Encountered errors while bringing up the project.'
  553. )
  554. return [
  555. container
  556. for svc_containers in results
  557. if svc_containers is not None
  558. for container in svc_containers
  559. ]
  560. def initialize(self):
  561. self.networks.initialize()
  562. self.volumes.initialize()
  563. def _get_convergence_plans(self, services, strategy, always_recreate_deps=False, one_off=None):
  564. plans = {}
  565. for service in services:
  566. updated_dependencies = [
  567. name
  568. for name in service.get_dependency_names()
  569. if name in plans and
  570. plans[name].action in ('recreate', 'create')
  571. ]
  572. is_one_off = one_off and service.name in one_off
  573. if updated_dependencies and strategy.allows_recreate:
  574. log.debug('%s has upstream changes (%s)',
  575. service.name,
  576. ", ".join(updated_dependencies))
  577. containers_stopped = any(
  578. service.containers(stopped=True, filters={'status': ['created', 'exited']}))
  579. service_has_links = any(service.get_link_names())
  580. container_has_links = any(c.get('HostConfig.Links') for c in service.containers())
  581. should_recreate_for_links = service_has_links ^ container_has_links
  582. if always_recreate_deps or containers_stopped or should_recreate_for_links:
  583. plan = service.convergence_plan(ConvergenceStrategy.always, is_one_off)
  584. else:
  585. plan = service.convergence_plan(strategy, is_one_off)
  586. else:
  587. plan = service.convergence_plan(strategy, is_one_off)
  588. plans[service.name] = plan
  589. return plans
  590. def pull(self, service_names=None, ignore_pull_failures=False, parallel_pull=False, silent=False,
  591. include_deps=False):
  592. services = self.get_services(service_names, include_deps)
  593. if parallel_pull:
  594. self.parallel_pull(services, silent=silent)
  595. else:
  596. must_build = []
  597. for service in services:
  598. try:
  599. service.pull(ignore_pull_failures, silent=silent)
  600. except (ImageNotFound, NotFound):
  601. if service.can_be_built():
  602. must_build.append(service.name)
  603. else:
  604. raise
  605. if len(must_build):
  606. log.warning('Some service image(s) must be built from source by running:\n'
  607. ' docker-compose build {}'
  608. .format(' '.join(must_build)))
  609. def parallel_pull(self, services, ignore_pull_failures=False, silent=False):
  610. msg = 'Pulling' if not silent else None
  611. must_build = []
  612. def pull_service(service):
  613. strm = service.pull(ignore_pull_failures, True, stream=True)
  614. if strm is None: # Attempting to pull service with no `image` key is a no-op
  615. return
  616. try:
  617. writer = parallel.get_stream_writer()
  618. for event in strm:
  619. if 'status' not in event:
  620. continue
  621. status = read_status(event)
  622. writer.write(
  623. msg, service.name, truncate_string(status), lambda s: s
  624. )
  625. except (ImageNotFound, NotFound):
  626. if service.can_be_built():
  627. must_build.append(service.name)
  628. else:
  629. raise
  630. _, errors = parallel.parallel_execute(
  631. services,
  632. pull_service,
  633. operator.attrgetter('name'),
  634. msg,
  635. limit=5,
  636. )
  637. if len(must_build):
  638. log.warning('Some service image(s) must be built from source by running:\n'
  639. ' docker-compose build {}'
  640. .format(' '.join(must_build)))
  641. if len(errors):
  642. combined_errors = '\n'.join([
  643. e.decode('utf-8') if isinstance(e, bytes) else e for e in errors.values()
  644. ])
  645. raise ProjectError(combined_errors)
  646. def push(self, service_names=None, ignore_push_failures=False):
  647. unique_images = set()
  648. for service in self.get_services(service_names, include_deps=False):
  649. # Considering <image> and <image:latest> as the same
  650. repo, tag, sep = parse_repository_tag(service.image_name)
  651. service_image_name = sep.join((repo, tag)) if tag else sep.join((repo, 'latest'))
  652. if service_image_name not in unique_images:
  653. service.push(ignore_push_failures)
  654. unique_images.add(service_image_name)
  655. def _labeled_containers(self, stopped=False, one_off=OneOffFilter.exclude):
  656. ctnrs = list(filter(None, [
  657. Container.from_ps(self.client, container)
  658. for container in self.client.containers(
  659. all=stopped,
  660. filters={'label': self.labels(one_off=one_off)})])
  661. )
  662. if ctnrs:
  663. return ctnrs
  664. return list(filter(lambda c: c.has_legacy_proj_name(self.name), filter(None, [
  665. Container.from_ps(self.client, container)
  666. for container in self.client.containers(
  667. all=stopped,
  668. filters={'label': self.labels(one_off=one_off, legacy=True)})])
  669. ))
  670. def containers(self, service_names=None, stopped=False, one_off=OneOffFilter.exclude):
  671. if service_names:
  672. self.validate_service_names(service_names)
  673. else:
  674. service_names = self.service_names
  675. containers = self._labeled_containers(stopped, one_off)
  676. def matches_service_names(container):
  677. return container.labels.get(LABEL_SERVICE) in service_names
  678. return [c for c in containers if matches_service_names(c)]
  679. def find_orphan_containers(self, remove_orphans):
  680. def _find():
  681. containers = set(self._labeled_containers() + self._labeled_containers(stopped=True))
  682. for ctnr in containers:
  683. service_name = ctnr.labels.get(LABEL_SERVICE)
  684. if service_name not in self.service_names:
  685. yield ctnr
  686. orphans = list(_find())
  687. if not orphans:
  688. return
  689. if remove_orphans:
  690. for ctnr in orphans:
  691. log.info('Removing orphan container "{}"'.format(ctnr.name))
  692. try:
  693. ctnr.kill()
  694. except APIError:
  695. pass
  696. ctnr.remove(force=True)
  697. else:
  698. log.warning(
  699. 'Found orphan containers ({}) for this project. If '
  700. 'you removed or renamed this service in your compose '
  701. 'file, you can run this command with the '
  702. '--remove-orphans flag to clean it up.'.format(
  703. ', '.join(["{}".format(ctnr.name) for ctnr in orphans])
  704. )
  705. )
  706. def _inject_deps(self, acc, service):
  707. dep_names = service.get_dependency_names()
  708. if len(dep_names) > 0:
  709. dep_services = self.get_services(
  710. service_names=list(set(dep_names)),
  711. include_deps=True
  712. )
  713. else:
  714. dep_services = []
  715. dep_services.append(service)
  716. return acc + dep_services
  717. def build_container_operation_with_timeout_func(self, operation, options):
  718. def container_operation_with_timeout(container):
  719. _options = options.copy()
  720. if _options.get('timeout') is None:
  721. service = self.get_service(container.service)
  722. _options['timeout'] = service.stop_timeout(None)
  723. return getattr(container, operation)(**_options)
  724. return container_operation_with_timeout
  725. def translate_credential_spec_to_security_opt(service_dict):
  726. result = []
  727. if 'credential_spec' in service_dict:
  728. spec = convert_credential_spec_to_security_opt(service_dict['credential_spec'])
  729. result.append('credentialspec={spec}'.format(spec=spec))
  730. if result:
  731. service_dict['security_opt'] = result
  732. return service_dict
  733. def translate_resource_keys_to_container_config(resources_dict, service_dict):
  734. if 'limits' in resources_dict:
  735. service_dict['mem_limit'] = resources_dict['limits'].get('memory')
  736. if 'cpus' in resources_dict['limits']:
  737. service_dict['cpus'] = float(resources_dict['limits']['cpus'])
  738. if 'reservations' in resources_dict:
  739. service_dict['mem_reservation'] = resources_dict['reservations'].get('memory')
  740. if 'cpus' in resources_dict['reservations']:
  741. return ['resources.reservations.cpus']
  742. return []
  743. def convert_restart_policy(name):
  744. try:
  745. return {
  746. 'any': 'always',
  747. 'none': 'no',
  748. 'on-failure': 'on-failure'
  749. }[name]
  750. except KeyError:
  751. raise ConfigurationError('Invalid restart policy "{}"'.format(name))
  752. def convert_credential_spec_to_security_opt(credential_spec):
  753. if 'file' in credential_spec:
  754. return 'file://{file}'.format(file=credential_spec['file'])
  755. return 'registry://{registry}'.format(registry=credential_spec['registry'])
  756. def translate_deploy_keys_to_container_config(service_dict):
  757. if 'credential_spec' in service_dict:
  758. del service_dict['credential_spec']
  759. if 'configs' in service_dict:
  760. del service_dict['configs']
  761. if 'deploy' not in service_dict:
  762. return service_dict, []
  763. deploy_dict = service_dict['deploy']
  764. ignored_keys = [
  765. k for k in ['endpoint_mode', 'labels', 'update_config', 'rollback_config']
  766. if k in deploy_dict
  767. ]
  768. if 'restart_policy' in deploy_dict:
  769. service_dict['restart'] = {
  770. 'Name': convert_restart_policy(deploy_dict['restart_policy'].get('condition', 'any')),
  771. 'MaximumRetryCount': deploy_dict['restart_policy'].get('max_attempts', 0)
  772. }
  773. for k in deploy_dict['restart_policy'].keys():
  774. if k != 'condition' and k != 'max_attempts':
  775. ignored_keys.append('restart_policy.{}'.format(k))
  776. ignored_keys.extend(
  777. translate_resource_keys_to_container_config(
  778. deploy_dict.get('resources', {}), service_dict
  779. )
  780. )
  781. del service_dict['deploy']
  782. return service_dict, ignored_keys
  783. def get_volumes_from(project, service_dict):
  784. volumes_from = service_dict.pop('volumes_from', None)
  785. if not volumes_from:
  786. return []
  787. def build_volume_from(spec):
  788. if spec.type == 'service':
  789. try:
  790. return spec._replace(source=project.get_service(spec.source))
  791. except NoSuchService:
  792. pass
  793. if spec.type == 'container':
  794. try:
  795. container = Container.from_id(project.client, spec.source)
  796. return spec._replace(source=container)
  797. except APIError:
  798. pass
  799. raise ConfigurationError(
  800. "Service \"{}\" mounts volumes from \"{}\", which is not the name "
  801. "of a service or container.".format(
  802. service_dict['name'],
  803. spec.source))
  804. return [build_volume_from(vf) for vf in volumes_from]
  805. def get_secrets(service, service_secrets, secret_defs):
  806. secrets = []
  807. for secret in service_secrets:
  808. secret_def = secret_defs.get(secret.source)
  809. if not secret_def:
  810. raise ConfigurationError(
  811. "Service \"{service}\" uses an undefined secret \"{secret}\" "
  812. .format(service=service, secret=secret.source))
  813. if secret_def.get('external'):
  814. log.warning('Service "{service}" uses secret "{secret}" which is external. '
  815. 'External secrets are not available to containers created by '
  816. 'docker-compose.'.format(service=service, secret=secret.source))
  817. continue
  818. if secret.uid or secret.gid or secret.mode:
  819. log.warning(
  820. 'Service "{service}" uses secret "{secret}" with uid, '
  821. 'gid, or mode. These fields are not supported by this '
  822. 'implementation of the Compose file'.format(
  823. service=service, secret=secret.source
  824. )
  825. )
  826. secret_file = secret_def.get('file')
  827. if not path.isfile(str(secret_file)):
  828. log.warning(
  829. 'Service "{service}" uses an undefined secret file "{secret_file}", '
  830. 'the following file should be created "{secret_file}"'.format(
  831. service=service, secret_file=secret_file
  832. )
  833. )
  834. secrets.append({'secret': secret, 'file': secret_file})
  835. return secrets
  836. def get_image_digests(project):
  837. digests = {}
  838. needs_push = set()
  839. needs_pull = set()
  840. for service in project.services:
  841. try:
  842. digests[service.name] = get_image_digest(service)
  843. except NeedsPush as e:
  844. needs_push.add(e.image_name)
  845. except NeedsPull as e:
  846. needs_pull.add(e.service_name)
  847. if needs_push or needs_pull:
  848. raise MissingDigests(needs_push, needs_pull)
  849. return digests
  850. def get_image_digest(service):
  851. if 'image' not in service.options:
  852. raise UserError(
  853. "Service '{s.name}' doesn't define an image tag. An image name is "
  854. "required to generate a proper image digest. Specify an image repo "
  855. "and tag with the 'image' option.".format(s=service))
  856. _, _, separator = parse_repository_tag(service.options['image'])
  857. # Compose file already uses a digest, no lookup required
  858. if separator == '@':
  859. return service.options['image']
  860. digest = get_digest(service)
  861. if digest:
  862. return digest
  863. if 'build' not in service.options:
  864. raise NeedsPull(service.image_name, service.name)
  865. raise NeedsPush(service.image_name)
  866. def get_digest(service):
  867. digest = None
  868. try:
  869. image = service.image()
  870. # TODO: pick a digest based on the image tag if there are multiple
  871. # digests
  872. if image['RepoDigests']:
  873. digest = image['RepoDigests'][0]
  874. except NoSuchImageError:
  875. try:
  876. # Fetch the image digest from the registry
  877. distribution = service.get_image_registry_data()
  878. if distribution['Descriptor']['digest']:
  879. digest = '{image_name}@{digest}'.format(
  880. image_name=service.image_name,
  881. digest=distribution['Descriptor']['digest']
  882. )
  883. except NoSuchImageError:
  884. raise UserError(
  885. "Digest not found for service '{service}'. "
  886. "Repository does not exist or may require 'docker login'"
  887. .format(service=service.name))
  888. return digest
  889. class MissingDigests(Exception):
  890. def __init__(self, needs_push, needs_pull):
  891. self.needs_push = needs_push
  892. self.needs_pull = needs_pull
  893. class NeedsPush(Exception):
  894. def __init__(self, image_name):
  895. self.image_name = image_name
  896. class NeedsPull(Exception):
  897. def __init__(self, image_name, service_name):
  898. self.image_name = image_name
  899. self.service_name = service_name
  900. class NoSuchService(Exception):
  901. def __init__(self, name):
  902. if isinstance(name, bytes):
  903. name = name.decode('utf-8')
  904. self.name = name
  905. self.msg = "No such service: %s" % self.name
  906. def __str__(self):
  907. return self.msg
  908. class ProjectError(Exception):
  909. def __init__(self, msg):
  910. self.msg = msg