project.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410
  1. from __future__ import absolute_import
  2. from __future__ import unicode_literals
  3. import logging
  4. from functools import reduce
  5. from docker.errors import APIError
  6. from docker.errors import NotFound
  7. from . import parallel
  8. from .config import ConfigurationError
  9. from .config import get_service_name_from_net
  10. from .const import DEFAULT_TIMEOUT
  11. from .const import LABEL_ONE_OFF
  12. from .const import LABEL_PROJECT
  13. from .const import LABEL_SERVICE
  14. from .container import Container
  15. from .legacy import check_for_legacy_containers
  16. from .service import ContainerNet
  17. from .service import ConvergenceStrategy
  18. from .service import Net
  19. from .service import Service
  20. from .service import ServiceNet
  21. log = logging.getLogger(__name__)
  22. def sort_service_dicts(services):
  23. # Topological sort (Cormen/Tarjan algorithm).
  24. unmarked = services[:]
  25. temporary_marked = set()
  26. sorted_services = []
  27. def get_service_names(links):
  28. return [link.split(':')[0] for link in links]
  29. def get_service_names_from_volumes_from(volumes_from):
  30. return [volume_from.source for volume_from in volumes_from]
  31. def get_service_dependents(service_dict, services):
  32. name = service_dict['name']
  33. return [
  34. service for service in services
  35. if (name in get_service_names(service.get('links', [])) or
  36. name in get_service_names_from_volumes_from(service.get('volumes_from', [])) or
  37. name == get_service_name_from_net(service.get('net')))
  38. ]
  39. def visit(n):
  40. if n['name'] in temporary_marked:
  41. if n['name'] in get_service_names(n.get('links', [])):
  42. raise DependencyError('A service can not link to itself: %s' % n['name'])
  43. if n['name'] in n.get('volumes_from', []):
  44. raise DependencyError('A service can not mount itself as volume: %s' % n['name'])
  45. else:
  46. raise DependencyError('Circular import between %s' % ' and '.join(temporary_marked))
  47. if n in unmarked:
  48. temporary_marked.add(n['name'])
  49. for m in get_service_dependents(n, services):
  50. visit(m)
  51. temporary_marked.remove(n['name'])
  52. unmarked.remove(n)
  53. sorted_services.insert(0, n)
  54. while unmarked:
  55. visit(unmarked[-1])
  56. return sorted_services
  57. class Project(object):
  58. """
  59. A collection of services.
  60. """
  61. def __init__(self, name, services, client, use_networking=False, network_driver=None):
  62. self.name = name
  63. self.services = services
  64. self.client = client
  65. self.use_networking = use_networking
  66. self.network_driver = network_driver
  67. def labels(self, one_off=False):
  68. return [
  69. '{0}={1}'.format(LABEL_PROJECT, self.name),
  70. '{0}={1}'.format(LABEL_ONE_OFF, "True" if one_off else "False"),
  71. ]
  72. @classmethod
  73. def from_dicts(cls, name, service_dicts, client, use_networking=False, network_driver=None):
  74. """
  75. Construct a ServiceCollection from a list of dicts representing services.
  76. """
  77. project = cls(name, [], client, use_networking=use_networking, network_driver=network_driver)
  78. if use_networking:
  79. remove_links(service_dicts)
  80. for service_dict in sort_service_dicts(service_dicts):
  81. links = project.get_links(service_dict)
  82. volumes_from = project.get_volumes_from(service_dict)
  83. net = project.get_net(service_dict)
  84. project.services.append(
  85. Service(
  86. client=client,
  87. project=name,
  88. use_networking=use_networking,
  89. links=links,
  90. net=net,
  91. volumes_from=volumes_from,
  92. **service_dict))
  93. return project
  94. @property
  95. def service_names(self):
  96. return [service.name for service in self.services]
  97. def get_service(self, name):
  98. """
  99. Retrieve a service by name. Raises NoSuchService
  100. if the named service does not exist.
  101. """
  102. for service in self.services:
  103. if service.name == name:
  104. return service
  105. raise NoSuchService(name)
  106. def validate_service_names(self, service_names):
  107. """
  108. Validate that the given list of service names only contains valid
  109. services. Raises NoSuchService if one of the names is invalid.
  110. """
  111. valid_names = self.service_names
  112. for name in service_names:
  113. if name not in valid_names:
  114. raise NoSuchService(name)
  115. def get_services(self, service_names=None, include_deps=False):
  116. """
  117. Returns a list of this project's services filtered
  118. by the provided list of names, or all services if service_names is None
  119. or [].
  120. If include_deps is specified, returns a list including the dependencies for
  121. service_names, in order of dependency.
  122. Preserves the original order of self.services where possible,
  123. reordering as needed to resolve dependencies.
  124. Raises NoSuchService if any of the named services do not exist.
  125. """
  126. if service_names is None or len(service_names) == 0:
  127. return self.get_services(
  128. service_names=self.service_names,
  129. include_deps=include_deps
  130. )
  131. else:
  132. unsorted = [self.get_service(name) for name in service_names]
  133. services = [s for s in self.services if s in unsorted]
  134. if include_deps:
  135. services = reduce(self._inject_deps, services, [])
  136. uniques = []
  137. [uniques.append(s) for s in services if s not in uniques]
  138. return uniques
  139. def get_links(self, service_dict):
  140. links = []
  141. if 'links' in service_dict:
  142. for link in service_dict.get('links', []):
  143. if ':' in link:
  144. service_name, link_name = link.split(':', 1)
  145. else:
  146. service_name, link_name = link, None
  147. try:
  148. links.append((self.get_service(service_name), link_name))
  149. except NoSuchService:
  150. raise ConfigurationError(
  151. 'Service "%s" has a link to service "%s" which does not '
  152. 'exist.' % (service_dict['name'], service_name))
  153. del service_dict['links']
  154. return links
  155. def get_volumes_from(self, service_dict):
  156. volumes_from = []
  157. if 'volumes_from' in service_dict:
  158. for volume_from_spec in service_dict.get('volumes_from', []):
  159. # Get service
  160. try:
  161. service = self.get_service(volume_from_spec.source)
  162. volume_from_spec = volume_from_spec._replace(source=service)
  163. except NoSuchService:
  164. try:
  165. container = Container.from_id(self.client, volume_from_spec.source)
  166. volume_from_spec = volume_from_spec._replace(source=container)
  167. except APIError:
  168. raise ConfigurationError(
  169. 'Service "%s" mounts volumes from "%s", which is '
  170. 'not the name of a service or container.' % (
  171. service_dict['name'],
  172. volume_from_spec.source))
  173. volumes_from.append(volume_from_spec)
  174. del service_dict['volumes_from']
  175. return volumes_from
  176. def get_net(self, service_dict):
  177. net = service_dict.pop('net', None)
  178. if not net:
  179. if self.use_networking:
  180. return Net(self.name)
  181. return Net(None)
  182. net_name = get_service_name_from_net(net)
  183. if not net_name:
  184. return Net(net)
  185. try:
  186. return ServiceNet(self.get_service(net_name))
  187. except NoSuchService:
  188. pass
  189. try:
  190. return ContainerNet(Container.from_id(self.client, net_name))
  191. except APIError:
  192. raise ConfigurationError(
  193. 'Service "%s" is trying to use the network of "%s", '
  194. 'which is not the name of a service or container.' % (
  195. service_dict['name'],
  196. net_name))
  197. def start(self, service_names=None, **options):
  198. for service in self.get_services(service_names):
  199. service.start(**options)
  200. def stop(self, service_names=None, **options):
  201. parallel.parallel_stop(self.containers(service_names), options)
  202. def pause(self, service_names=None, **options):
  203. parallel.parallel_pause(reversed(self.containers(service_names)), options)
  204. def unpause(self, service_names=None, **options):
  205. parallel.parallel_unpause(self.containers(service_names), options)
  206. def kill(self, service_names=None, **options):
  207. parallel.parallel_kill(self.containers(service_names), options)
  208. def remove_stopped(self, service_names=None, **options):
  209. parallel.parallel_remove(self.containers(service_names, stopped=True), options)
  210. def restart(self, service_names=None, **options):
  211. parallel.parallel_restart(self.containers(service_names, stopped=True), options)
  212. def build(self, service_names=None, no_cache=False, pull=False, force_rm=False):
  213. for service in self.get_services(service_names):
  214. if service.can_be_built():
  215. service.build(no_cache, pull, force_rm)
  216. else:
  217. log.info('%s uses an image, skipping' % service.name)
  218. def up(self,
  219. service_names=None,
  220. start_deps=True,
  221. strategy=ConvergenceStrategy.changed,
  222. do_build=True,
  223. timeout=DEFAULT_TIMEOUT,
  224. detached=False):
  225. services = self.get_services(service_names, include_deps=start_deps)
  226. for service in services:
  227. service.remove_duplicate_containers()
  228. plans = self._get_convergence_plans(services, strategy)
  229. if self.use_networking and self.uses_default_network():
  230. self.ensure_network_exists()
  231. return [
  232. container
  233. for service in services
  234. for container in service.execute_convergence_plan(
  235. plans[service.name],
  236. do_build=do_build,
  237. timeout=timeout,
  238. detached=detached
  239. )
  240. ]
  241. def _get_convergence_plans(self, services, strategy):
  242. plans = {}
  243. for service in services:
  244. updated_dependencies = [
  245. name
  246. for name in service.get_dependency_names()
  247. if name in plans
  248. and plans[name].action in ('recreate', 'create')
  249. ]
  250. if updated_dependencies and strategy.allows_recreate:
  251. log.debug('%s has upstream changes (%s)',
  252. service.name,
  253. ", ".join(updated_dependencies))
  254. plan = service.convergence_plan(ConvergenceStrategy.always)
  255. else:
  256. plan = service.convergence_plan(strategy)
  257. plans[service.name] = plan
  258. return plans
  259. def pull(self, service_names=None, ignore_pull_failures=False):
  260. for service in self.get_services(service_names, include_deps=False):
  261. service.pull(ignore_pull_failures)
  262. def containers(self, service_names=None, stopped=False, one_off=False):
  263. if service_names:
  264. self.validate_service_names(service_names)
  265. else:
  266. service_names = self.service_names
  267. containers = list(filter(None, [
  268. Container.from_ps(self.client, container)
  269. for container in self.client.containers(
  270. all=stopped,
  271. filters={'label': self.labels(one_off=one_off)})]))
  272. def matches_service_names(container):
  273. return container.labels.get(LABEL_SERVICE) in service_names
  274. if not containers:
  275. check_for_legacy_containers(
  276. self.client,
  277. self.name,
  278. self.service_names,
  279. )
  280. return [c for c in containers if matches_service_names(c)]
  281. def get_network(self):
  282. try:
  283. return self.client.inspect_network(self.name)
  284. except NotFound:
  285. return None
  286. def ensure_network_exists(self):
  287. # TODO: recreate network if driver has changed?
  288. if self.get_network() is None:
  289. log.info(
  290. 'Creating network "{}" with driver "{}"'
  291. .format(self.name, self.network_driver)
  292. )
  293. self.client.create_network(self.name, driver=self.network_driver)
  294. def remove_network(self):
  295. network = self.get_network()
  296. if network:
  297. self.client.remove_network(network['Id'])
  298. def uses_default_network(self):
  299. return any(service.net.mode == self.name for service in self.services)
  300. def _inject_deps(self, acc, service):
  301. dep_names = service.get_dependency_names()
  302. if len(dep_names) > 0:
  303. dep_services = self.get_services(
  304. service_names=list(set(dep_names)),
  305. include_deps=True
  306. )
  307. else:
  308. dep_services = []
  309. dep_services.append(service)
  310. return acc + dep_services
  311. def remove_links(service_dicts):
  312. services_with_links = [s for s in service_dicts if 'links' in s]
  313. if not services_with_links:
  314. return
  315. if len(services_with_links) == 1:
  316. prefix = '"{}" defines'.format(services_with_links[0]['name'])
  317. else:
  318. prefix = 'Some services ({}) define'.format(
  319. ", ".join('"{}"'.format(s['name']) for s in services_with_links))
  320. log.warn(
  321. '\n{} links, which are not compatible with Docker networking and will be ignored.\n'
  322. 'Future versions of Docker will not support links - you should remove them for '
  323. 'forwards-compatibility.\n'.format(prefix))
  324. for s in services_with_links:
  325. del s['links']
  326. class NoSuchService(Exception):
  327. def __init__(self, name):
  328. self.name = name
  329. self.msg = "No such service: %s" % self.name
  330. def __str__(self):
  331. return self.msg
  332. class DependencyError(ConfigurationError):
  333. pass