project.py 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530
  1. from __future__ import absolute_import
  2. from __future__ import unicode_literals
  3. import datetime
  4. import logging
  5. import operator
  6. from functools import reduce
  7. import enum
  8. from docker.errors import APIError
  9. from . import parallel
  10. from .config import ConfigurationError
  11. from .config.config import V1
  12. from .config.sort_services import get_container_name_from_network_mode
  13. from .config.sort_services import get_service_name_from_network_mode
  14. from .const import DEFAULT_TIMEOUT
  15. from .const import IMAGE_EVENTS
  16. from .const import LABEL_ONE_OFF
  17. from .const import LABEL_PROJECT
  18. from .const import LABEL_SERVICE
  19. from .container import Container
  20. from .network import build_networks
  21. from .network import get_networks
  22. from .network import ProjectNetworks
  23. from .service import BuildAction
  24. from .service import ContainerNetworkMode
  25. from .service import ConvergenceStrategy
  26. from .service import NetworkMode
  27. from .service import Service
  28. from .service import ServiceNetworkMode
  29. from .utils import microseconds_from_time_nano
  30. from .volume import ProjectVolumes
  31. log = logging.getLogger(__name__)
  32. @enum.unique
  33. class OneOffFilter(enum.Enum):
  34. include = 0
  35. exclude = 1
  36. only = 2
  37. @classmethod
  38. def update_labels(cls, value, labels):
  39. if value == cls.only:
  40. labels.append('{0}={1}'.format(LABEL_ONE_OFF, "True"))
  41. elif value == cls.exclude:
  42. labels.append('{0}={1}'.format(LABEL_ONE_OFF, "False"))
  43. elif value == cls.include:
  44. pass
  45. else:
  46. raise ValueError("Invalid value for one_off: {}".format(repr(value)))
  47. class Project(object):
  48. """
  49. A collection of services.
  50. """
  51. def __init__(self, name, services, client, networks=None, volumes=None):
  52. self.name = name
  53. self.services = services
  54. self.client = client
  55. self.volumes = volumes or ProjectVolumes({})
  56. self.networks = networks or ProjectNetworks({}, False)
  57. def labels(self, one_off=OneOffFilter.exclude):
  58. labels = ['{0}={1}'.format(LABEL_PROJECT, self.name)]
  59. OneOffFilter.update_labels(one_off, labels)
  60. return labels
  61. @classmethod
  62. def from_config(cls, name, config_data, client):
  63. """
  64. Construct a Project from a config.Config object.
  65. """
  66. use_networking = (config_data.version and config_data.version != V1)
  67. networks = build_networks(name, config_data, client)
  68. project_networks = ProjectNetworks.from_services(
  69. config_data.services,
  70. networks,
  71. use_networking)
  72. volumes = ProjectVolumes.from_config(name, config_data, client)
  73. project = cls(name, [], client, project_networks, volumes)
  74. for service_dict in config_data.services:
  75. service_dict = dict(service_dict)
  76. if use_networking:
  77. service_networks = get_networks(service_dict, networks)
  78. else:
  79. service_networks = {}
  80. service_dict.pop('networks', None)
  81. links = project.get_links(service_dict)
  82. network_mode = project.get_network_mode(
  83. service_dict, list(service_networks.keys())
  84. )
  85. volumes_from = get_volumes_from(project, service_dict)
  86. if config_data.version != V1:
  87. service_dict['volumes'] = [
  88. volumes.namespace_spec(volume_spec)
  89. for volume_spec in service_dict.get('volumes', [])
  90. ]
  91. project.services.append(
  92. Service(
  93. service_dict.pop('name'),
  94. client=client,
  95. project=name,
  96. use_networking=use_networking,
  97. networks=service_networks,
  98. links=links,
  99. network_mode=network_mode,
  100. volumes_from=volumes_from,
  101. **service_dict)
  102. )
  103. return project
  104. @property
  105. def service_names(self):
  106. return [service.name for service in self.services]
  107. def get_service(self, name):
  108. """
  109. Retrieve a service by name. Raises NoSuchService
  110. if the named service does not exist.
  111. """
  112. for service in self.services:
  113. if service.name == name:
  114. return service
  115. raise NoSuchService(name)
  116. def validate_service_names(self, service_names):
  117. """
  118. Validate that the given list of service names only contains valid
  119. services. Raises NoSuchService if one of the names is invalid.
  120. """
  121. valid_names = self.service_names
  122. for name in service_names:
  123. if name not in valid_names:
  124. raise NoSuchService(name)
  125. def get_services(self, service_names=None, include_deps=False):
  126. """
  127. Returns a list of this project's services filtered
  128. by the provided list of names, or all services if service_names is None
  129. or [].
  130. If include_deps is specified, returns a list including the dependencies for
  131. service_names, in order of dependency.
  132. Preserves the original order of self.services where possible,
  133. reordering as needed to resolve dependencies.
  134. Raises NoSuchService if any of the named services do not exist.
  135. """
  136. if service_names is None or len(service_names) == 0:
  137. service_names = self.service_names
  138. unsorted = [self.get_service(name) for name in service_names]
  139. services = [s for s in self.services if s in unsorted]
  140. if include_deps:
  141. services = reduce(self._inject_deps, services, [])
  142. uniques = []
  143. [uniques.append(s) for s in services if s not in uniques]
  144. return uniques
  145. def get_services_without_duplicate(self, service_names=None, include_deps=False):
  146. services = self.get_services(service_names, include_deps)
  147. for service in services:
  148. service.remove_duplicate_containers()
  149. return services
  150. def get_links(self, service_dict):
  151. links = []
  152. if 'links' in service_dict:
  153. for link in service_dict.get('links', []):
  154. if ':' in link:
  155. service_name, link_name = link.split(':', 1)
  156. else:
  157. service_name, link_name = link, None
  158. try:
  159. links.append((self.get_service(service_name), link_name))
  160. except NoSuchService:
  161. raise ConfigurationError(
  162. 'Service "%s" has a link to service "%s" which does not '
  163. 'exist.' % (service_dict['name'], service_name))
  164. del service_dict['links']
  165. return links
  166. def get_network_mode(self, service_dict, networks):
  167. network_mode = service_dict.pop('network_mode', None)
  168. if not network_mode:
  169. if self.networks.use_networking:
  170. return NetworkMode(networks[0]) if networks else NetworkMode('none')
  171. return NetworkMode(None)
  172. service_name = get_service_name_from_network_mode(network_mode)
  173. if service_name:
  174. return ServiceNetworkMode(self.get_service(service_name))
  175. container_name = get_container_name_from_network_mode(network_mode)
  176. if container_name:
  177. try:
  178. return ContainerNetworkMode(Container.from_id(self.client, container_name))
  179. except APIError:
  180. raise ConfigurationError(
  181. "Service '{name}' uses the network stack of container '{dep}' which "
  182. "does not exist.".format(name=service_dict['name'], dep=container_name))
  183. return NetworkMode(network_mode)
  184. def start(self, service_names=None, **options):
  185. containers = []
  186. def start_service(service):
  187. service_containers = service.start(quiet=True, **options)
  188. containers.extend(service_containers)
  189. services = self.get_services(service_names)
  190. def get_deps(service):
  191. return {self.get_service(dep) for dep in service.get_dependency_names()}
  192. parallel.parallel_execute(
  193. services,
  194. start_service,
  195. operator.attrgetter('name'),
  196. 'Starting',
  197. get_deps)
  198. return containers
  199. def stop(self, service_names=None, one_off=OneOffFilter.exclude, **options):
  200. containers = self.containers(service_names, one_off=one_off)
  201. def get_deps(container):
  202. # actually returning inversed dependencies
  203. return {other for other in containers
  204. if container.service in
  205. self.get_service(other.service).get_dependency_names()}
  206. parallel.parallel_execute(
  207. containers,
  208. operator.methodcaller('stop', **options),
  209. operator.attrgetter('name'),
  210. 'Stopping',
  211. get_deps)
  212. def pause(self, service_names=None, **options):
  213. containers = self.containers(service_names)
  214. parallel.parallel_pause(reversed(containers), options)
  215. return containers
  216. def unpause(self, service_names=None, **options):
  217. containers = self.containers(service_names)
  218. parallel.parallel_unpause(containers, options)
  219. return containers
  220. def kill(self, service_names=None, **options):
  221. parallel.parallel_kill(self.containers(service_names), options)
  222. def remove_stopped(self, service_names=None, one_off=OneOffFilter.exclude, **options):
  223. parallel.parallel_remove(self.containers(
  224. service_names, stopped=True, one_off=one_off
  225. ), options)
  226. def down(self, remove_image_type, include_volumes, remove_orphans=False):
  227. self.stop(one_off=OneOffFilter.include)
  228. self.find_orphan_containers(remove_orphans)
  229. self.remove_stopped(v=include_volumes, one_off=OneOffFilter.include)
  230. self.networks.remove()
  231. if include_volumes:
  232. self.volumes.remove()
  233. self.remove_images(remove_image_type)
  234. def remove_images(self, remove_image_type):
  235. for service in self.get_services():
  236. service.remove_image(remove_image_type)
  237. def restart(self, service_names=None, **options):
  238. containers = self.containers(service_names, stopped=True)
  239. parallel.parallel_restart(containers, options)
  240. return containers
  241. def build(self, service_names=None, no_cache=False, pull=False, force_rm=False):
  242. for service in self.get_services(service_names):
  243. if service.can_be_built():
  244. service.build(no_cache, pull, force_rm)
  245. else:
  246. log.info('%s uses an image, skipping' % service.name)
  247. def create(
  248. self,
  249. service_names=None,
  250. strategy=ConvergenceStrategy.changed,
  251. do_build=BuildAction.none,
  252. ):
  253. services = self.get_services_without_duplicate(service_names, include_deps=True)
  254. for svc in services:
  255. svc.ensure_image_exists(do_build=do_build)
  256. plans = self._get_convergence_plans(services, strategy)
  257. for service in services:
  258. service.execute_convergence_plan(
  259. plans[service.name],
  260. detached=True,
  261. start=False)
  262. def events(self, service_names=None):
  263. def build_container_event(event, container):
  264. time = datetime.datetime.fromtimestamp(event['time'])
  265. time = time.replace(
  266. microsecond=microseconds_from_time_nano(event['timeNano']))
  267. return {
  268. 'time': time,
  269. 'type': 'container',
  270. 'action': event['status'],
  271. 'id': container.id,
  272. 'service': container.service,
  273. 'attributes': {
  274. 'name': container.name,
  275. 'image': event['from'],
  276. },
  277. 'container': container,
  278. }
  279. service_names = set(service_names or self.service_names)
  280. for event in self.client.events(
  281. filters={'label': self.labels()},
  282. decode=True
  283. ):
  284. if event['status'] in IMAGE_EVENTS:
  285. # We don't receive any image events because labels aren't applied
  286. # to images
  287. continue
  288. # TODO: get labels from the API v1.22 , see github issue 2618
  289. try:
  290. # this can fail if the conatiner has been removed
  291. container = Container.from_id(self.client, event['id'])
  292. except APIError:
  293. continue
  294. if container.service not in service_names:
  295. continue
  296. yield build_container_event(event, container)
  297. def up(self,
  298. service_names=None,
  299. start_deps=True,
  300. strategy=ConvergenceStrategy.changed,
  301. do_build=BuildAction.none,
  302. timeout=DEFAULT_TIMEOUT,
  303. detached=False,
  304. remove_orphans=False):
  305. self.initialize()
  306. self.find_orphan_containers(remove_orphans)
  307. services = self.get_services_without_duplicate(
  308. service_names,
  309. include_deps=start_deps)
  310. for svc in services:
  311. svc.ensure_image_exists(do_build=do_build)
  312. plans = self._get_convergence_plans(services, strategy)
  313. def do(service):
  314. return service.execute_convergence_plan(
  315. plans[service.name],
  316. timeout=timeout,
  317. detached=detached
  318. )
  319. def get_deps(service):
  320. return {self.get_service(dep) for dep in service.get_dependency_names()}
  321. results = parallel.parallel_execute(
  322. services,
  323. do,
  324. operator.attrgetter('name'),
  325. None,
  326. get_deps
  327. )
  328. return [
  329. container
  330. for svc_containers in results
  331. if svc_containers is not None
  332. for container in svc_containers
  333. ]
  334. def initialize(self):
  335. self.networks.initialize()
  336. self.volumes.initialize()
  337. def _get_convergence_plans(self, services, strategy):
  338. plans = {}
  339. for service in services:
  340. updated_dependencies = [
  341. name
  342. for name in service.get_dependency_names()
  343. if name in plans and
  344. plans[name].action in ('recreate', 'create')
  345. ]
  346. if updated_dependencies and strategy.allows_recreate:
  347. log.debug('%s has upstream changes (%s)',
  348. service.name,
  349. ", ".join(updated_dependencies))
  350. plan = service.convergence_plan(ConvergenceStrategy.always)
  351. else:
  352. plan = service.convergence_plan(strategy)
  353. plans[service.name] = plan
  354. return plans
  355. def pull(self, service_names=None, ignore_pull_failures=False):
  356. for service in self.get_services(service_names, include_deps=False):
  357. service.pull(ignore_pull_failures)
  358. def _labeled_containers(self, stopped=False, one_off=OneOffFilter.exclude):
  359. return list(filter(None, [
  360. Container.from_ps(self.client, container)
  361. for container in self.client.containers(
  362. all=stopped,
  363. filters={'label': self.labels(one_off=one_off)})])
  364. )
  365. def containers(self, service_names=None, stopped=False, one_off=OneOffFilter.exclude):
  366. if service_names:
  367. self.validate_service_names(service_names)
  368. else:
  369. service_names = self.service_names
  370. containers = self._labeled_containers(stopped, one_off)
  371. def matches_service_names(container):
  372. return container.labels.get(LABEL_SERVICE) in service_names
  373. return [c for c in containers if matches_service_names(c)]
  374. def find_orphan_containers(self, remove_orphans):
  375. def _find():
  376. containers = self._labeled_containers()
  377. for ctnr in containers:
  378. service_name = ctnr.labels.get(LABEL_SERVICE)
  379. if service_name not in self.service_names:
  380. yield ctnr
  381. orphans = list(_find())
  382. if not orphans:
  383. return
  384. if remove_orphans:
  385. for ctnr in orphans:
  386. log.info('Removing orphan container "{0}"'.format(ctnr.name))
  387. ctnr.kill()
  388. ctnr.remove(force=True)
  389. else:
  390. log.warning(
  391. 'Found orphan containers ({0}) for this project. If '
  392. 'you removed or renamed this service in your compose '
  393. 'file, you can run this command with the '
  394. '--remove-orphans flag to clean it up.'.format(
  395. ', '.join(["{}".format(ctnr.name) for ctnr in orphans])
  396. )
  397. )
  398. def _inject_deps(self, acc, service):
  399. dep_names = service.get_dependency_names()
  400. if len(dep_names) > 0:
  401. dep_services = self.get_services(
  402. service_names=list(set(dep_names)),
  403. include_deps=True
  404. )
  405. else:
  406. dep_services = []
  407. dep_services.append(service)
  408. return acc + dep_services
  409. def get_volumes_from(project, service_dict):
  410. volumes_from = service_dict.pop('volumes_from', None)
  411. if not volumes_from:
  412. return []
  413. def build_volume_from(spec):
  414. if spec.type == 'service':
  415. try:
  416. return spec._replace(source=project.get_service(spec.source))
  417. except NoSuchService:
  418. pass
  419. if spec.type == 'container':
  420. try:
  421. container = Container.from_id(project.client, spec.source)
  422. return spec._replace(source=container)
  423. except APIError:
  424. pass
  425. raise ConfigurationError(
  426. "Service \"{}\" mounts volumes from \"{}\", which is not the name "
  427. "of a service or container.".format(
  428. service_dict['name'],
  429. spec.source))
  430. return [build_volume_from(vf) for vf in volumes_from]
  431. class NoSuchService(Exception):
  432. def __init__(self, name):
  433. self.name = name
  434. self.msg = "No such service: %s" % self.name
  435. def __str__(self):
  436. return self.msg