bug 563: Return initial-config-primitive uuids
[osm/N2VC.git] / modules / libjuju / juju / model.py
1 import asyncio
2 import base64
3 import collections
4 import hashlib
5 import json
6 import logging
7 import os
8 import re
9 import stat
10 import tempfile
11 import weakref
12 import zipfile
13 from concurrent.futures import CancelledError
14 from functools import partial
15 from pathlib import Path
16
17 import theblues.charmstore
18 import theblues.errors
19 import websockets
20 import yaml
21
22 from . import tag, utils
23 from .client import client, connector
24 from .client.client import ConfigValue
25 from .client.client import Value
26 from .constraints import parse as parse_constraints
27 from .constraints import normalize_key
28 from .delta import get_entity_class, get_entity_delta
29 from .errors import JujuAPIError, JujuError
30 from .exceptions import DeadEntityException
31 from .placement import parse as parse_placement
32 from . import provisioner
33
34
35 log = logging.getLogger(__name__)
36
37
38 class _Observer:
39 """Wrapper around an observer callable.
40
41 This wrapper allows filter criteria to be associated with the
42 callable so that it's only called for changes that meet the criteria.
43
44 """
45 def __init__(self, callable_, entity_type, action, entity_id, predicate):
46 self.callable_ = callable_
47 self.entity_type = entity_type
48 self.action = action
49 self.entity_id = entity_id
50 self.predicate = predicate
51 if self.entity_id:
52 self.entity_id = str(self.entity_id)
53 if not self.entity_id.startswith('^'):
54 self.entity_id = '^' + self.entity_id
55 if not self.entity_id.endswith('$'):
56 self.entity_id += '$'
57
58 async def __call__(self, delta, old, new, model):
59 await self.callable_(delta, old, new, model)
60
61 def cares_about(self, delta):
62 """Return True if this observer "cares about" (i.e. wants to be
63 called) for a this delta.
64
65 """
66 if (self.entity_id and delta.get_id() and
67 not re.match(self.entity_id, str(delta.get_id()))):
68 return False
69
70 if self.entity_type and self.entity_type != delta.entity:
71 return False
72
73 if self.action and self.action != delta.type:
74 return False
75
76 if self.predicate and not self.predicate(delta):
77 return False
78
79 return True
80
81
82 class ModelObserver:
83 """
84 Base class for creating observers that react to changes in a model.
85 """
86 async def __call__(self, delta, old, new, model):
87 handler_name = 'on_{}_{}'.format(delta.entity, delta.type)
88 method = getattr(self, handler_name, self.on_change)
89 await method(delta, old, new, model)
90
91 async def on_change(self, delta, old, new, model):
92 """Generic model-change handler.
93
94 This should be overridden in a subclass.
95
96 :param delta: :class:`juju.client.overrides.Delta`
97 :param old: :class:`juju.model.ModelEntity`
98 :param new: :class:`juju.model.ModelEntity`
99 :param model: :class:`juju.model.Model`
100
101 """
102 pass
103
104
105 class ModelState:
106 """Holds the state of the model, including the delta history of all
107 entities in the model.
108
109 """
110 def __init__(self, model):
111 self.model = model
112 self.state = dict()
113
114 def _live_entity_map(self, entity_type):
115 """Return an id:Entity map of all the living entities of
116 type ``entity_type``.
117
118 """
119 return {
120 entity_id: self.get_entity(entity_type, entity_id)
121 for entity_id, history in self.state.get(entity_type, {}).items()
122 if history[-1] is not None
123 }
124
125 @property
126 def applications(self):
127 """Return a map of application-name:Application for all applications
128 currently in the model.
129
130 """
131 return self._live_entity_map('application')
132
133 @property
134 def machines(self):
135 """Return a map of machine-id:Machine for all machines currently in
136 the model.
137
138 """
139 return self._live_entity_map('machine')
140
141 @property
142 def units(self):
143 """Return a map of unit-id:Unit for all units currently in
144 the model.
145
146 """
147 return self._live_entity_map('unit')
148
149 @property
150 def relations(self):
151 """Return a map of relation-id:Relation for all relations currently in
152 the model.
153
154 """
155 return self._live_entity_map('relation')
156
157 def entity_history(self, entity_type, entity_id):
158 """Return the history deque for an entity.
159
160 """
161 return self.state[entity_type][entity_id]
162
163 def entity_data(self, entity_type, entity_id, history_index):
164 """Return the data dict for an entity at a specific index of its
165 history.
166
167 """
168 return self.entity_history(entity_type, entity_id)[history_index]
169
170 def apply_delta(self, delta):
171 """Apply delta to our state and return a copy of the
172 affected object as it was before and after the update, e.g.:
173
174 old_obj, new_obj = self.apply_delta(delta)
175
176 old_obj may be None if the delta is for the creation of a new object,
177 e.g. a new application or unit is deployed.
178
179 new_obj will never be None, but may be dead (new_obj.dead == True)
180 if the object was deleted as a result of the delta being applied.
181
182 """
183 history = (
184 self.state
185 .setdefault(delta.entity, {})
186 .setdefault(delta.get_id(), collections.deque())
187 )
188
189 history.append(delta.data)
190 if delta.type == 'remove':
191 history.append(None)
192
193 entity = self.get_entity(delta.entity, delta.get_id())
194 return entity.previous(), entity
195
196 def get_entity(
197 self, entity_type, entity_id, history_index=-1, connected=True):
198 """Return an object instance for the given entity_type and id.
199
200 By default the object state matches the most recent state from
201 Juju. To get an instance of the object in an older state, pass
202 history_index, an index into the history deque for the entity.
203
204 """
205
206 if history_index < 0 and history_index != -1:
207 history_index += len(self.entity_history(entity_type, entity_id))
208 if history_index < 0:
209 return None
210
211 try:
212 self.entity_data(entity_type, entity_id, history_index)
213 except IndexError:
214 return None
215
216 entity_class = get_entity_class(entity_type)
217 return entity_class(
218 entity_id, self.model, history_index=history_index,
219 connected=connected)
220
221
222 class ModelEntity:
223 """An object in the Model tree"""
224
225 def __init__(self, entity_id, model, history_index=-1, connected=True):
226 """Initialize a new entity
227
228 :param entity_id str: The unique id of the object in the model
229 :param model: The model instance in whose object tree this
230 entity resides
231 :history_index int: The index of this object's state in the model's
232 history deque for this entity
233 :connected bool: Flag indicating whether this object gets live updates
234 from the model.
235
236 """
237 self.entity_id = entity_id
238 self.model = model
239 self._history_index = history_index
240 self.connected = connected
241 self.connection = model.connection()
242
243 def __repr__(self):
244 return '<{} entity_id="{}">'.format(type(self).__name__,
245 self.entity_id)
246
247 def __getattr__(self, name):
248 """Fetch object attributes from the underlying data dict held in the
249 model.
250
251 """
252 try:
253 return self.safe_data[name]
254 except KeyError:
255 name = name.replace('_', '-')
256 if name in self.safe_data:
257 return self.safe_data[name]
258 else:
259 raise
260
261 def __bool__(self):
262 return bool(self.data)
263
264 def on_change(self, callable_):
265 """Add a change observer to this entity.
266
267 """
268 self.model.add_observer(
269 callable_, self.entity_type, 'change', self.entity_id)
270
271 def on_remove(self, callable_):
272 """Add a remove observer to this entity.
273
274 """
275 self.model.add_observer(
276 callable_, self.entity_type, 'remove', self.entity_id)
277
278 @property
279 def entity_type(self):
280 """A string identifying the entity type of this object, e.g.
281 'application' or 'unit', etc.
282
283 """
284 return self.__class__.__name__.lower()
285
286 @property
287 def current(self):
288 """Return True if this object represents the current state of the
289 entity in the underlying model.
290
291 This will be True except when the object represents an entity at a
292 non-latest state in history, e.g. if the object was obtained by calling
293 .previous() on another object.
294
295 """
296 return self._history_index == -1
297
298 @property
299 def dead(self):
300 """Returns True if this entity no longer exists in the underlying
301 model.
302
303 """
304 return (
305 self.data is None or
306 self.model.state.entity_data(
307 self.entity_type, self.entity_id, -1) is None
308 )
309
310 @property
311 def alive(self):
312 """Returns True if this entity still exists in the underlying
313 model.
314
315 """
316 return not self.dead
317
318 @property
319 def data(self):
320 """The data dictionary for this entity.
321
322 """
323 return self.model.state.entity_data(
324 self.entity_type, self.entity_id, self._history_index)
325
326 @property
327 def safe_data(self):
328 """The data dictionary for this entity.
329
330 If this `ModelEntity` points to the dead state, it will
331 raise `DeadEntityException`.
332
333 """
334 if self.data is None:
335 raise DeadEntityException(
336 "Entity {}:{} is dead - its attributes can no longer be "
337 "accessed. Use the .previous() method on this object to get "
338 "a copy of the object at its previous state.".format(
339 self.entity_type, self.entity_id))
340 return self.data
341
342 def previous(self):
343 """Return a copy of this object as was at its previous state in
344 history.
345
346 Returns None if this object is new (and therefore has no history).
347
348 The returned object is always "disconnected", i.e. does not receive
349 live updates.
350
351 """
352 return self.model.state.get_entity(
353 self.entity_type, self.entity_id, self._history_index - 1,
354 connected=False)
355
356 def next(self):
357 """Return a copy of this object at its next state in
358 history.
359
360 Returns None if this object is already the latest.
361
362 The returned object is "disconnected", i.e. does not receive
363 live updates, unless it is current (latest).
364
365 """
366 if self._history_index == -1:
367 return None
368
369 new_index = self._history_index + 1
370 connected = (
371 new_index == len(self.model.state.entity_history(
372 self.entity_type, self.entity_id)) - 1
373 )
374 return self.model.state.get_entity(
375 self.entity_type, self.entity_id, self._history_index - 1,
376 connected=connected)
377
378 def latest(self):
379 """Return a copy of this object at its current state in the model.
380
381 Returns self if this object is already the latest.
382
383 The returned object is always "connected", i.e. receives
384 live updates from the model.
385
386 """
387 if self._history_index == -1:
388 return self
389
390 return self.model.state.get_entity(self.entity_type, self.entity_id)
391
392
393 class Model:
394 """
395 The main API for interacting with a Juju model.
396 """
397 def __init__(
398 self,
399 loop=None,
400 max_frame_size=None,
401 bakery_client=None,
402 jujudata=None,
403 ):
404 """Instantiate a new Model.
405
406 The connect method will need to be called before this
407 object can be used for anything interesting.
408
409 If jujudata is None, jujudata.FileJujuData will be used.
410
411 :param loop: an asyncio event loop
412 :param max_frame_size: See
413 `juju.client.connection.Connection.MAX_FRAME_SIZE`
414 :param bakery_client httpbakery.Client: The bakery client to use
415 for macaroon authorization.
416 :param jujudata JujuData: The source for current controller information
417 """
418 self._connector = connector.Connector(
419 loop=loop,
420 max_frame_size=max_frame_size,
421 bakery_client=bakery_client,
422 jujudata=jujudata,
423 )
424 self._observers = weakref.WeakValueDictionary()
425 self.state = ModelState(self)
426 self._info = None
427 self._watch_stopping = asyncio.Event(loop=self._connector.loop)
428 self._watch_stopped = asyncio.Event(loop=self._connector.loop)
429 self._watch_received = asyncio.Event(loop=self._connector.loop)
430 self._watch_stopped.set()
431 self._charmstore = CharmStore(self._connector.loop)
432
433 def is_connected(self):
434 """Reports whether the Model is currently connected."""
435 return self._connector.is_connected()
436
437 @property
438 def loop(self):
439 return self._connector.loop
440
441 def connection(self):
442 """Return the current Connection object. It raises an exception
443 if the Model is disconnected"""
444 return self._connector.connection()
445
446 async def get_controller(self):
447 """Return a Controller instance for the currently connected model.
448 :return Controller:
449 """
450 from juju.controller import Controller
451 controller = Controller(jujudata=self._connector.jujudata)
452 kwargs = self.connection().connect_params()
453 kwargs.pop('uuid')
454 await controller._connect_direct(**kwargs)
455 return controller
456
457 async def __aenter__(self):
458 await self.connect()
459 return self
460
461 async def __aexit__(self, exc_type, exc, tb):
462 await self.disconnect()
463
464 async def connect(self, *args, **kwargs):
465 """Connect to a juju model.
466
467 This supports two calling conventions:
468
469 The model and (optionally) authentication information can be taken
470 from the data files created by the Juju CLI. This convention will
471 be used if a ``model_name`` is specified, or if the ``endpoint``
472 and ``uuid`` are not.
473
474 Otherwise, all of the ``endpoint``, ``uuid``, and authentication
475 information (``username`` and ``password``, or ``bakery_client`` and/or
476 ``macaroons``) are required.
477
478 If a single positional argument is given, it will be assumed to be
479 the ``model_name``. Otherwise, the first positional argument, if any,
480 must be the ``endpoint``.
481
482 Available parameters are:
483
484 :param model_name: Format [controller:][user/]model
485 :param str endpoint: The hostname:port of the controller to connect to.
486 :param str uuid: The model UUID to connect to.
487 :param str username: The username for controller-local users (or None
488 to use macaroon-based login.)
489 :param str password: The password for controller-local users.
490 :param str cacert: The CA certificate of the controller
491 (PEM formatted).
492 :param httpbakery.Client bakery_client: The macaroon bakery client to
493 to use when performing macaroon-based login. Macaroon tokens
494 acquired when logging will be saved to bakery_client.cookies.
495 If this is None, a default bakery_client will be used.
496 :param list macaroons: List of macaroons to load into the
497 ``bakery_client``.
498 :param asyncio.BaseEventLoop loop: The event loop to use for async
499 operations.
500 :param int max_frame_size: The maximum websocket frame size to allow.
501 """
502 await self.disconnect()
503 if 'endpoint' not in kwargs and len(args) < 2:
504 if args and 'model_name' in kwargs:
505 raise TypeError('connect() got multiple values for model_name')
506 elif args:
507 model_name = args[0]
508 else:
509 model_name = kwargs.pop('model_name', None)
510 await self._connector.connect_model(model_name, **kwargs)
511 else:
512 if 'model_name' in kwargs:
513 raise TypeError('connect() got values for both '
514 'model_name and endpoint')
515 if args and 'endpoint' in kwargs:
516 raise TypeError('connect() got multiple values for endpoint')
517 if len(args) < 2 and 'uuid' not in kwargs:
518 raise TypeError('connect() missing value for uuid')
519 has_userpass = (len(args) >= 4 or
520 {'username', 'password'}.issubset(kwargs))
521 has_macaroons = (len(args) >= 6 or not
522 {'bakery_client', 'macaroons'}.isdisjoint(kwargs))
523 if not (has_userpass or has_macaroons):
524 raise TypeError('connect() missing auth params')
525 arg_names = [
526 'endpoint',
527 'uuid',
528 'username',
529 'password',
530 'cacert',
531 'bakery_client',
532 'macaroons',
533 'loop',
534 'max_frame_size',
535 ]
536 for i, arg in enumerate(args):
537 kwargs[arg_names[i]] = arg
538 if not {'endpoint', 'uuid'}.issubset(kwargs):
539 raise ValueError('endpoint and uuid are required '
540 'if model_name not given')
541 if not ({'username', 'password'}.issubset(kwargs) or
542 {'bakery_client', 'macaroons'}.intersection(kwargs)):
543 raise ValueError('Authentication parameters are required '
544 'if model_name not given')
545 await self._connector.connect(**kwargs)
546 await self._after_connect()
547
548 async def connect_model(self, model_name):
549 """
550 .. deprecated:: 0.6.2
551 Use ``connect(model_name=model_name)`` instead.
552 """
553 return await self.connect(model_name=model_name)
554
555 async def connect_current(self):
556 """
557 .. deprecated:: 0.6.2
558 Use ``connect()`` instead.
559 """
560 return await self.connect()
561
562 async def _connect_direct(self, **kwargs):
563 await self.disconnect()
564 await self._connector.connect(**kwargs)
565 await self._after_connect()
566
567 async def _after_connect(self):
568 self._watch()
569
570 # Wait for the first packet of data from the AllWatcher,
571 # which contains all information on the model.
572 # TODO this means that we can't do anything until
573 # we've received all the model data, which might be
574 # a whole load of unneeded data if all the client wants
575 # to do is make one RPC call.
576 await self._watch_received.wait()
577
578 await self.get_info()
579
580 async def disconnect(self):
581 """Shut down the watcher task and close websockets.
582
583 """
584 if not self._watch_stopped.is_set():
585 log.debug('Stopping watcher task')
586 self._watch_stopping.set()
587 await self._watch_stopped.wait()
588 self._watch_stopping.clear()
589
590 if self.is_connected():
591 log.debug('Closing model connection')
592 await self._connector.disconnect()
593 self._info = None
594
595 async def add_local_charm_dir(self, charm_dir, series):
596 """Upload a local charm to the model.
597
598 This will automatically generate an archive from
599 the charm dir.
600
601 :param charm_dir: Path to the charm directory
602 :param series: Charm series
603
604 """
605 fh = tempfile.NamedTemporaryFile()
606 CharmArchiveGenerator(charm_dir).make_archive(fh.name)
607 with fh:
608 func = partial(
609 self.add_local_charm, fh, series, os.stat(fh.name).st_size)
610 charm_url = await self._connector.loop.run_in_executor(None, func)
611
612 log.debug('Uploaded local charm: %s -> %s', charm_dir, charm_url)
613 return charm_url
614
615 def add_local_charm(self, charm_file, series, size=None):
616 """Upload a local charm archive to the model.
617
618 Returns the 'local:...' url that should be used to deploy the charm.
619
620 :param charm_file: Path to charm zip archive
621 :param series: Charm series
622 :param size: Size of the archive, in bytes
623 :return str: 'local:...' url for deploying the charm
624 :raises: :class:`JujuError` if the upload fails
625
626 Uses an https endpoint at the same host:port as the wss.
627 Supports large file uploads.
628
629 .. warning::
630
631 This method will block. Consider using :meth:`add_local_charm_dir`
632 instead.
633
634 """
635 conn, headers, path_prefix = self.connection().https_connection()
636 path = "%s/charms?series=%s" % (path_prefix, series)
637 headers['Content-Type'] = 'application/zip'
638 if size:
639 headers['Content-Length'] = size
640 conn.request("POST", path, charm_file, headers)
641 response = conn.getresponse()
642 result = response.read().decode()
643 if not response.status == 200:
644 raise JujuError(result)
645 result = json.loads(result)
646 return result['charm-url']
647
648 def all_units_idle(self):
649 """Return True if all units are idle.
650
651 """
652 for unit in self.units.values():
653 unit_status = unit.data['agent-status']['current']
654 if unit_status != 'idle':
655 return False
656 return True
657
658 async def reset(self, force=False):
659 """Reset the model to a clean state.
660
661 :param bool force: Force-terminate machines.
662
663 This returns only after the model has reached a clean state. "Clean"
664 means no applications or machines exist in the model.
665
666 """
667 log.debug('Resetting model')
668 for app in self.applications.values():
669 await app.destroy()
670 for machine in self.machines.values():
671 await machine.destroy(force=force)
672 await self.block_until(
673 lambda: len(self.machines) == 0
674 )
675
676 async def block_until(self, *conditions, timeout=None, wait_period=0.5):
677 """Return only after all conditions are true.
678
679 Raises `websockets.ConnectionClosed` if disconnected.
680 """
681 def _disconnected():
682 return not (self.is_connected() and self.connection().is_open)
683
684 def done():
685 return _disconnected() or all(c() for c in conditions)
686
687 await utils.block_until(done,
688 timeout=timeout,
689 wait_period=wait_period,
690 loop=self.loop)
691 if _disconnected():
692 raise websockets.ConnectionClosed(1006, 'no reason')
693
694 @property
695 def applications(self):
696 """Return a map of application-name:Application for all applications
697 currently in the model.
698
699 """
700 return self.state.applications
701
702 @property
703 def machines(self):
704 """Return a map of machine-id:Machine for all machines currently in
705 the model.
706
707 """
708 return self.state.machines
709
710 @property
711 def units(self):
712 """Return a map of unit-id:Unit for all units currently in
713 the model.
714
715 """
716 return self.state.units
717
718 @property
719 def relations(self):
720 """Return a list of all Relations currently in the model.
721
722 """
723 return list(self.state.relations.values())
724
725 async def get_info(self):
726 """Return a client.ModelInfo object for this Model.
727
728 Retrieves latest info for this Model from the api server. The
729 return value is cached on the Model.info attribute so that the
730 valued may be accessed again without another api call, if
731 desired.
732
733 This method is called automatically when the Model is connected,
734 resulting in Model.info being initialized without requiring an
735 explicit call to this method.
736
737 """
738 facade = client.ClientFacade.from_connection(self.connection())
739
740 self._info = await facade.ModelInfo()
741 log.debug('Got ModelInfo: %s', vars(self.info))
742
743 return self.info
744
745 @property
746 def info(self):
747 """Return the cached client.ModelInfo object for this Model.
748
749 If Model.get_info() has not been called, this will return None.
750 """
751 return self._info
752
753 def add_observer(
754 self, callable_, entity_type=None, action=None, entity_id=None,
755 predicate=None):
756 """Register an "on-model-change" callback
757
758 Once the model is connected, ``callable_``
759 will be called each time the model changes. ``callable_`` should
760 be Awaitable and accept the following positional arguments:
761
762 delta - An instance of :class:`juju.delta.EntityDelta`
763 containing the raw delta data recv'd from the Juju
764 websocket.
765
766 old_obj - If the delta modifies an existing object in the model,
767 old_obj will be a copy of that object, as it was before the
768 delta was applied. Will be None if the delta creates a new
769 entity in the model.
770
771 new_obj - A copy of the new or updated object, after the delta
772 is applied. Will be None if the delta removes an entity
773 from the model.
774
775 model - The :class:`Model` itself.
776
777 Events for which ``callable_`` is called can be specified by passing
778 entity_type, action, and/or entitiy_id filter criteria, e.g.::
779
780 add_observer(
781 myfunc,
782 entity_type='application', action='add', entity_id='ubuntu')
783
784 For more complex filtering conditions, pass a predicate function. It
785 will be called with a delta as its only argument. If the predicate
786 function returns True, the ``callable_`` will be called.
787
788 """
789 observer = _Observer(
790 callable_, entity_type, action, entity_id, predicate)
791 self._observers[observer] = callable_
792
793 def _watch(self):
794 """Start an asynchronous watch against this model.
795
796 See :meth:`add_observer` to register an onchange callback.
797
798 """
799 async def _all_watcher():
800 try:
801 allwatcher = client.AllWatcherFacade.from_connection(
802 self.connection())
803 while not self._watch_stopping.is_set():
804 try:
805 results = await utils.run_with_interrupt(
806 allwatcher.Next(),
807 self._watch_stopping,
808 self._connector.loop)
809 except JujuAPIError as e:
810 if 'watcher was stopped' not in str(e):
811 raise
812 if self._watch_stopping.is_set():
813 # this shouldn't ever actually happen, because
814 # the event should trigger before the controller
815 # has a chance to tell us the watcher is stopped
816 # but handle it gracefully, just in case
817 break
818 # controller stopped our watcher for some reason
819 # but we're not actually stopping, so just restart it
820 log.warning(
821 'Watcher: watcher stopped, restarting')
822 del allwatcher.Id
823 continue
824 except websockets.ConnectionClosed:
825 monitor = self.connection().monitor
826 if monitor.status == monitor.ERROR:
827 # closed unexpectedly, try to reopen
828 log.warning(
829 'Watcher: connection closed, reopening')
830 await self.connection().reconnect()
831 if monitor.status != monitor.CONNECTED:
832 # reconnect failed; abort and shutdown
833 log.error('Watcher: automatic reconnect '
834 'failed; stopping watcher')
835 break
836 del allwatcher.Id
837 continue
838 else:
839 # closed on request, go ahead and shutdown
840 break
841 if self._watch_stopping.is_set():
842 try:
843 await allwatcher.Stop()
844 except websockets.ConnectionClosed:
845 pass # can't stop on a closed conn
846 break
847 for delta in results.deltas:
848 delta = get_entity_delta(delta)
849 old_obj, new_obj = self.state.apply_delta(delta)
850 await self._notify_observers(delta, old_obj, new_obj)
851 self._watch_received.set()
852 except CancelledError:
853 pass
854 except Exception:
855 log.exception('Error in watcher')
856 raise
857 finally:
858 self._watch_stopped.set()
859
860 log.debug('Starting watcher task')
861 self._watch_received.clear()
862 self._watch_stopping.clear()
863 self._watch_stopped.clear()
864 self._connector.loop.create_task(_all_watcher())
865
866 async def _notify_observers(self, delta, old_obj, new_obj):
867 """Call observing callbacks, notifying them of a change in model state
868
869 :param delta: The raw change from the watcher
870 (:class:`juju.client.overrides.Delta`)
871 :param old_obj: The object in the model that this delta updates.
872 May be None.
873 :param new_obj: The object in the model that is created or updated
874 by applying this delta.
875
876 """
877 if new_obj and not old_obj:
878 delta.type = 'add'
879
880 log.debug(
881 'Model changed: %s %s %s',
882 delta.entity, delta.type, delta.get_id())
883
884 for o in self._observers:
885 if o.cares_about(delta):
886 asyncio.ensure_future(o(delta, old_obj, new_obj, self),
887 loop=self._connector.loop)
888
889 async def _wait(self, entity_type, entity_id, action, predicate=None):
890 """
891 Block the calling routine until a given action has happened to the
892 given entity
893
894 :param entity_type: The entity's type.
895 :param entity_id: The entity's id.
896 :param action: the type of action (e.g., 'add', 'change', or 'remove')
897 :param predicate: optional callable that must take as an
898 argument a delta, and must return a boolean, indicating
899 whether the delta contains the specific action we're looking
900 for. For example, you might check to see whether a 'change'
901 has a 'completed' status. See the _Observer class for details.
902
903 """
904 q = asyncio.Queue(loop=self._connector.loop)
905
906 async def callback(delta, old, new, model):
907 await q.put(delta.get_id())
908
909 self.add_observer(callback, entity_type, action, entity_id, predicate)
910 entity_id = await q.get()
911 # object might not be in the entity_map if we were waiting for a
912 # 'remove' action
913 return self.state._live_entity_map(entity_type).get(entity_id)
914
915 async def _wait_for_new(self, entity_type, entity_id):
916 """Wait for a new object to appear in the Model and return it.
917
918 Waits for an object of type ``entity_type`` with id ``entity_id``
919 to appear in the model. This is similar to watching for the
920 object using ``block_until``, but uses the watcher rather than
921 polling.
922
923 """
924 # if the entity is already in the model, just return it
925 if entity_id in self.state._live_entity_map(entity_type):
926 return self.state._live_entity_map(entity_type)[entity_id]
927 return await self._wait(entity_type, entity_id, None)
928
929 async def wait_for_action(self, action_id):
930 """Given an action, wait for it to complete."""
931
932 if action_id.startswith("action-"):
933 # if we've been passed action.tag, transform it into the
934 # id that the api deltas will use.
935 action_id = action_id[7:]
936
937 def predicate(delta):
938 return delta.data['status'] in ('completed', 'failed')
939
940 return await self._wait('action', action_id, None, predicate)
941
942 async def add_machine(
943 self, spec=None, constraints=None, disks=None, series=None):
944 """Start a new, empty machine and optionally a container, or add a
945 container to a machine.
946
947 :param str spec: Machine specification
948 Examples::
949
950 (None) - starts a new machine
951 'lxd' - starts a new machine with one lxd container
952 'lxd:4' - starts a new lxd container on machine 4
953 'ssh:user@10.10.0.3:/path/to/private/key' - manually provision
954 a machine with ssh and the private key used for authentication
955 'zone=us-east-1a' - starts a machine in zone us-east-1s on AWS
956 'maas2.name' - acquire machine maas2.name on MAAS
957
958 :param dict constraints: Machine constraints, which can contain the
959 the following keys::
960
961 arch : str
962 container : str
963 cores : int
964 cpu_power : int
965 instance_type : str
966 mem : int
967 root_disk : int
968 spaces : list(str)
969 tags : list(str)
970 virt_type : str
971
972 Example::
973
974 constraints={
975 'mem': 256 * MB,
976 'tags': ['virtual'],
977 }
978
979 :param list disks: List of disk constraint dictionaries, which can
980 contain the following keys::
981
982 count : int
983 pool : str
984 size : int
985
986 Example::
987
988 disks=[{
989 'pool': 'rootfs',
990 'size': 10 * GB,
991 'count': 1,
992 }]
993
994 :param str series: Series, e.g. 'xenial'
995
996 Supported container types are: lxd, kvm
997
998 When deploying a container to an existing machine, constraints cannot
999 be used.
1000
1001 """
1002 params = client.AddMachineParams()
1003
1004 if spec:
1005 if spec.startswith("ssh:"):
1006 placement, target, private_key_path = spec.split(":")
1007 user, host = target.split("@")
1008
1009 sshProvisioner = provisioner.SSHProvisioner(
1010 host=host,
1011 user=user,
1012 private_key_path=private_key_path,
1013 )
1014
1015 params = sshProvisioner.provision_machine()
1016 else:
1017 placement = parse_placement(spec)
1018 if placement:
1019 params.placement = placement[0]
1020
1021 params.jobs = ['JobHostUnits']
1022
1023 if constraints:
1024 params.constraints = client.Value.from_json(constraints)
1025
1026 if disks:
1027 params.disks = [
1028 client.Constraints.from_json(o) for o in disks]
1029
1030 if series:
1031 params.series = series
1032
1033 # Submit the request.
1034 client_facade = client.ClientFacade.from_connection(self.connection())
1035 results = await client_facade.AddMachines([params])
1036 error = results.machines[0].error
1037 if error:
1038 raise ValueError("Error adding machine: %s" % error.message)
1039 machine_id = results.machines[0].machine
1040
1041 if spec:
1042 if spec.startswith("ssh:"):
1043 # Need to run this after AddMachines has been called,
1044 # as we need the machine_id
1045 await sshProvisioner.install_agent(
1046 self.connection(),
1047 params.nonce,
1048 machine_id,
1049 )
1050
1051 log.debug('Added new machine %s', machine_id)
1052 return await self._wait_for_new('machine', machine_id)
1053
1054 async def add_relation(self, relation1, relation2):
1055 """Add a relation between two applications.
1056
1057 :param str relation1: '<application>[:<relation_name>]'
1058 :param str relation2: '<application>[:<relation_name>]'
1059
1060 """
1061 connection = self.connection()
1062 app_facade = client.ApplicationFacade.from_connection(connection)
1063
1064 log.debug(
1065 'Adding relation %s <-> %s', relation1, relation2)
1066
1067 def _find_relation(*specs):
1068 for rel in self.relations:
1069 if rel.matches(*specs):
1070 return rel
1071 return None
1072
1073 try:
1074 result = await app_facade.AddRelation([relation1, relation2])
1075 except JujuAPIError as e:
1076 if 'relation already exists' not in e.message:
1077 raise
1078 rel = _find_relation(relation1, relation2)
1079 if rel:
1080 return rel
1081 raise JujuError('Relation {} {} exists but not in model'.format(
1082 relation1, relation2))
1083
1084 specs = ['{}:{}'.format(app, data['name'])
1085 for app, data in result.endpoints.items()]
1086
1087 await self.block_until(lambda: _find_relation(*specs) is not None)
1088 return _find_relation(*specs)
1089
1090 def add_space(self, name, *cidrs):
1091 """Add a new network space.
1092
1093 Adds a new space with the given name and associates the given
1094 (optional) list of existing subnet CIDRs with it.
1095
1096 :param str name: Name of the space
1097 :param \*cidrs: Optional list of existing subnet CIDRs
1098
1099 """
1100 raise NotImplementedError()
1101
1102 async def add_ssh_key(self, user, key):
1103 """Add a public SSH key to this model.
1104
1105 :param str user: The username of the user
1106 :param str key: The public ssh key
1107
1108 """
1109 key_facade = client.KeyManagerFacade.from_connection(self.connection())
1110 return await key_facade.AddKeys([key], user)
1111 add_ssh_keys = add_ssh_key
1112
1113 def add_subnet(self, cidr_or_id, space, *zones):
1114 """Add an existing subnet to this model.
1115
1116 :param str cidr_or_id: CIDR or provider ID of the existing subnet
1117 :param str space: Network space with which to associate
1118 :param str \*zones: Zone(s) in which the subnet resides
1119
1120 """
1121 raise NotImplementedError()
1122
1123 def get_backups(self):
1124 """Retrieve metadata for backups in this model.
1125
1126 """
1127 raise NotImplementedError()
1128
1129 def block(self, *commands):
1130 """Add a new block to this model.
1131
1132 :param str \*commands: The commands to block. Valid values are
1133 'all-changes', 'destroy-model', 'remove-object'
1134
1135 """
1136 raise NotImplementedError()
1137
1138 def get_blocks(self):
1139 """List blocks for this model.
1140
1141 """
1142 raise NotImplementedError()
1143
1144 def get_cached_images(self, arch=None, kind=None, series=None):
1145 """Return a list of cached OS images.
1146
1147 :param str arch: Filter by image architecture
1148 :param str kind: Filter by image kind, e.g. 'lxd'
1149 :param str series: Filter by image series, e.g. 'xenial'
1150
1151 """
1152 raise NotImplementedError()
1153
1154 def create_backup(self, note=None, no_download=False):
1155 """Create a backup of this model.
1156
1157 :param str note: A note to store with the backup
1158 :param bool no_download: Do not download the backup archive
1159 :return str: Path to downloaded archive
1160
1161 """
1162 raise NotImplementedError()
1163
1164 def create_storage_pool(self, name, provider_type, **pool_config):
1165 """Create or define a storage pool.
1166
1167 :param str name: Name to give the storage pool
1168 :param str provider_type: Pool provider type
1169 :param \*\*pool_config: key/value pool configuration pairs
1170
1171 """
1172 raise NotImplementedError()
1173
1174 def debug_log(
1175 self, no_tail=False, exclude_module=None, include_module=None,
1176 include=None, level=None, limit=0, lines=10, replay=False,
1177 exclude=None):
1178 """Get log messages for this model.
1179
1180 :param bool no_tail: Stop after returning existing log messages
1181 :param list exclude_module: Do not show log messages for these logging
1182 modules
1183 :param list include_module: Only show log messages for these logging
1184 modules
1185 :param list include: Only show log messages for these entities
1186 :param str level: Log level to show, valid options are 'TRACE',
1187 'DEBUG', 'INFO', 'WARNING', 'ERROR,
1188 :param int limit: Return this many of the most recent (possibly
1189 filtered) lines are shown
1190 :param int lines: Yield this many of the most recent lines, and keep
1191 yielding
1192 :param bool replay: Yield the entire log, and keep yielding
1193 :param list exclude: Do not show log messages for these entities
1194
1195 """
1196 raise NotImplementedError()
1197
1198 def _get_series(self, entity_url, entity):
1199 # try to get the series from the provided charm URL
1200 if entity_url.startswith('cs:'):
1201 parts = entity_url[3:].split('/')
1202 else:
1203 parts = entity_url.split('/')
1204 if parts[0].startswith('~'):
1205 parts.pop(0)
1206 if len(parts) > 1:
1207 # series was specified in the URL
1208 return parts[0]
1209 # series was not supplied at all, so use the newest
1210 # supported series according to the charm store
1211 ss = entity['Meta']['supported-series']
1212 return ss['SupportedSeries'][0]
1213
1214 async def deploy(
1215 self, entity_url, application_name=None, bind=None, budget=None,
1216 channel=None, config=None, constraints=None, force=False,
1217 num_units=1, plan=None, resources=None, series=None, storage=None,
1218 to=None):
1219 """Deploy a new service or bundle.
1220
1221 :param str entity_url: Charm or bundle url
1222 :param str application_name: Name to give the service
1223 :param dict bind: <charm endpoint>:<network space> pairs
1224 :param dict budget: <budget name>:<limit> pairs
1225 :param str channel: Charm store channel from which to retrieve
1226 the charm or bundle, e.g. 'edge'
1227 :param dict config: Charm configuration dictionary
1228 :param constraints: Service constraints
1229 :type constraints: :class:`juju.Constraints`
1230 :param bool force: Allow charm to be deployed to a machine running
1231 an unsupported series
1232 :param int num_units: Number of units to deploy
1233 :param str plan: Plan under which to deploy charm
1234 :param dict resources: <resource name>:<file path> pairs
1235 :param str series: Series on which to deploy
1236 :param dict storage: Storage constraints TODO how do these look?
1237 :param to: Placement directive as a string. For example:
1238
1239 '23' - place on machine 23
1240 'lxd:7' - place in new lxd container on machine 7
1241 '24/lxd/3' - place in container 3 on machine 24
1242
1243 If None, a new machine is provisioned.
1244
1245
1246 TODO::
1247
1248 - support local resources
1249
1250 """
1251 if storage:
1252 storage = {
1253 k: client.Constraints(**v)
1254 for k, v in storage.items()
1255 }
1256
1257 entity_path = Path(entity_url.replace('local:', ''))
1258 bundle_path = entity_path / 'bundle.yaml'
1259 metadata_path = entity_path / 'metadata.yaml'
1260
1261 is_local = (
1262 entity_url.startswith('local:') or
1263 entity_path.is_dir() or
1264 entity_path.is_file()
1265 )
1266 if is_local:
1267 entity_id = entity_url.replace('local:', '')
1268 else:
1269 entity = await self.charmstore.entity(entity_url, channel=channel)
1270 entity_id = entity['Id']
1271
1272 client_facade = client.ClientFacade.from_connection(self.connection())
1273
1274 is_bundle = ((is_local and
1275 (entity_id.endswith('.yaml') and entity_path.exists()) or
1276 bundle_path.exists()) or
1277 (not is_local and 'bundle/' in entity_id))
1278
1279 if is_bundle:
1280 handler = BundleHandler(self)
1281 await handler.fetch_plan(entity_id)
1282 await handler.execute_plan()
1283 extant_apps = {app for app in self.applications}
1284 pending_apps = set(handler.applications) - extant_apps
1285 if pending_apps:
1286 # new apps will usually be in the model by now, but if some
1287 # haven't made it yet we'll need to wait on them to be added
1288 await asyncio.gather(*[
1289 asyncio.ensure_future(
1290 self._wait_for_new('application', app_name),
1291 loop=self._connector.loop)
1292 for app_name in pending_apps
1293 ], loop=self._connector.loop)
1294 return [app for name, app in self.applications.items()
1295 if name in handler.applications]
1296 else:
1297 if not is_local:
1298 if not application_name:
1299 application_name = entity['Meta']['charm-metadata']['Name']
1300 if not series:
1301 series = self._get_series(entity_url, entity)
1302 await client_facade.AddCharm(channel, entity_id)
1303 # XXX: we're dropping local resources here, but we don't
1304 # actually support them yet anyway
1305 resources = await self._add_store_resources(application_name,
1306 entity_id,
1307 entity)
1308 else:
1309 if not application_name:
1310 metadata = yaml.load(metadata_path.read_text())
1311 application_name = metadata['name']
1312 # We have a local charm dir that needs to be uploaded
1313 charm_dir = os.path.abspath(
1314 os.path.expanduser(entity_id))
1315 series = series or get_charm_series(charm_dir)
1316 if not series:
1317 raise JujuError(
1318 "Couldn't determine series for charm at {}. "
1319 "Pass a 'series' kwarg to Model.deploy().".format(
1320 charm_dir))
1321 entity_id = await self.add_local_charm_dir(charm_dir, series)
1322 return await self._deploy(
1323 charm_url=entity_id,
1324 application=application_name,
1325 series=series,
1326 config=config or {},
1327 constraints=constraints,
1328 endpoint_bindings=bind,
1329 resources=resources,
1330 storage=storage,
1331 channel=channel,
1332 num_units=num_units,
1333 placement=parse_placement(to)
1334 )
1335
1336 async def _add_store_resources(self, application, entity_url, entity=None):
1337 if not entity:
1338 # avoid extra charm store call if one was already made
1339 entity = await self.charmstore.entity(entity_url)
1340 resources = [
1341 {
1342 'description': resource['Description'],
1343 'fingerprint': resource['Fingerprint'],
1344 'name': resource['Name'],
1345 'path': resource['Path'],
1346 'revision': resource['Revision'],
1347 'size': resource['Size'],
1348 'type_': resource['Type'],
1349 'origin': 'store',
1350 } for resource in entity['Meta']['resources']
1351 ]
1352
1353 if not resources:
1354 return None
1355
1356 resources_facade = client.ResourcesFacade.from_connection(
1357 self.connection())
1358 response = await resources_facade.AddPendingResources(
1359 tag.application(application),
1360 entity_url,
1361 [client.CharmResource(**resource) for resource in resources])
1362 resource_map = {resource['name']: pid
1363 for resource, pid
1364 in zip(resources, response.pending_ids)}
1365 return resource_map
1366
1367 async def _deploy(self, charm_url, application, series, config,
1368 constraints, endpoint_bindings, resources, storage,
1369 channel=None, num_units=None, placement=None):
1370 """Logic shared between `Model.deploy` and `BundleHandler.deploy`.
1371 """
1372 log.info('Deploying %s', charm_url)
1373
1374 # stringify all config values for API, and convert to YAML
1375 config = {k: str(v) for k, v in config.items()}
1376 config = yaml.dump({application: config},
1377 default_flow_style=False)
1378
1379 app_facade = client.ApplicationFacade.from_connection(
1380 self.connection())
1381
1382 app = client.ApplicationDeploy(
1383 charm_url=charm_url,
1384 application=application,
1385 series=series,
1386 channel=channel,
1387 config_yaml=config,
1388 constraints=parse_constraints(constraints),
1389 endpoint_bindings=endpoint_bindings,
1390 num_units=num_units,
1391 resources=resources,
1392 storage=storage,
1393 placement=placement
1394 )
1395 result = await app_facade.Deploy([app])
1396 errors = [r.error.message for r in result.results if r.error]
1397 if errors:
1398 raise JujuError('\n'.join(errors))
1399 return await self._wait_for_new('application', application)
1400
1401 async def destroy(self):
1402 """Terminate all machines and resources for this model.
1403 Is already implemented in controller.py.
1404 """
1405 raise NotImplementedError()
1406
1407 async def destroy_unit(self, *unit_names):
1408 """Destroy units by name.
1409
1410 """
1411 connection = self.connection()
1412 app_facade = client.ApplicationFacade.from_connection(connection)
1413
1414 log.debug(
1415 'Destroying unit%s %s',
1416 's' if len(unit_names) == 1 else '',
1417 ' '.join(unit_names))
1418
1419 return await app_facade.DestroyUnits(list(unit_names))
1420 destroy_units = destroy_unit
1421
1422 def get_backup(self, archive_id):
1423 """Download a backup archive file.
1424
1425 :param str archive_id: The id of the archive to download
1426 :return str: Path to the archive file
1427
1428 """
1429 raise NotImplementedError()
1430
1431 def enable_ha(
1432 self, num_controllers=0, constraints=None, series=None, to=None):
1433 """Ensure sufficient controllers exist to provide redundancy.
1434
1435 :param int num_controllers: Number of controllers to make available
1436 :param constraints: Constraints to apply to the controller machines
1437 :type constraints: :class:`juju.Constraints`
1438 :param str series: Series of the controller machines
1439 :param list to: Placement directives for controller machines, e.g.::
1440
1441 '23' - machine 23
1442 'lxc:7' - new lxc container on machine 7
1443 '24/lxc/3' - lxc container 3 or machine 24
1444
1445 If None, a new machine is provisioned.
1446
1447 """
1448 raise NotImplementedError()
1449
1450 async def get_config(self):
1451 """Return the configuration settings for this model.
1452
1453 :returns: A ``dict`` mapping keys to `ConfigValue` instances,
1454 which have `source` and `value` attributes.
1455 """
1456 config_facade = client.ModelConfigFacade.from_connection(
1457 self.connection()
1458 )
1459 result = await config_facade.ModelGet()
1460 config = result.config
1461 for key, value in config.items():
1462 config[key] = ConfigValue.from_json(value)
1463 return config
1464
1465 async def get_constraints(self):
1466 """Return the machine constraints for this model.
1467
1468 :returns: A ``dict`` of constraints.
1469 """
1470 constraints = {}
1471 client_facade = client.ClientFacade.from_connection(self.connection())
1472 result = await client_facade.GetModelConstraints()
1473
1474 # GetModelConstraints returns GetConstraintsResults which has a 'constraints'
1475 # attribute. If no constraints have been set GetConstraintsResults.constraints
1476 # is None. Otherwise GetConstraintsResults.constraints has an attribute for each
1477 # possible constraint, each of these in turn will be None if they have not been
1478 # set.
1479 if result.constraints:
1480 constraint_types = [a for a in dir(result.constraints)
1481 if a in Value._toSchema.keys()]
1482 for constraint in constraint_types:
1483 value = getattr(result.constraints, constraint)
1484 if value is not None:
1485 constraints[constraint] = getattr(result.constraints, constraint)
1486 return constraints
1487
1488 def import_ssh_key(self, identity):
1489 """Add a public SSH key from a trusted indentity source to this model.
1490
1491 :param str identity: User identity in the form <lp|gh>:<username>
1492
1493 """
1494 raise NotImplementedError()
1495 import_ssh_keys = import_ssh_key
1496
1497 async def get_machines(self):
1498 """Return list of machines in this model.
1499
1500 """
1501 return list(self.state.machines.keys())
1502
1503 def get_shares(self):
1504 """Return list of all users with access to this model.
1505
1506 """
1507 raise NotImplementedError()
1508
1509 def get_spaces(self):
1510 """Return list of all known spaces, including associated subnets.
1511
1512 """
1513 raise NotImplementedError()
1514
1515 async def get_ssh_key(self, raw_ssh=False):
1516 """Return known SSH keys for this model.
1517 :param bool raw_ssh: if True, returns the raw ssh key,
1518 else it's fingerprint
1519
1520 """
1521 key_facade = client.KeyManagerFacade.from_connection(self.connection())
1522 entity = {'tag': tag.model(self.info.uuid)}
1523 entities = client.Entities([entity])
1524 return await key_facade.ListKeys(entities, raw_ssh)
1525 get_ssh_keys = get_ssh_key
1526
1527 def get_storage(self, filesystem=False, volume=False):
1528 """Return details of storage instances.
1529
1530 :param bool filesystem: Include filesystem storage
1531 :param bool volume: Include volume storage
1532
1533 """
1534 raise NotImplementedError()
1535
1536 def get_storage_pools(self, names=None, providers=None):
1537 """Return list of storage pools.
1538
1539 :param list names: Only include pools with these names
1540 :param list providers: Only include pools for these providers
1541
1542 """
1543 raise NotImplementedError()
1544
1545 def get_subnets(self, space=None, zone=None):
1546 """Return list of known subnets.
1547
1548 :param str space: Only include subnets in this space
1549 :param str zone: Only include subnets in this zone
1550
1551 """
1552 raise NotImplementedError()
1553
1554 def remove_blocks(self):
1555 """Remove all blocks from this model.
1556
1557 """
1558 raise NotImplementedError()
1559
1560 def remove_backup(self, backup_id):
1561 """Delete a backup.
1562
1563 :param str backup_id: The id of the backup to remove
1564
1565 """
1566 raise NotImplementedError()
1567
1568 def remove_cached_images(self, arch=None, kind=None, series=None):
1569 """Remove cached OS images.
1570
1571 :param str arch: Architecture of the images to remove
1572 :param str kind: Image kind to remove, e.g. 'lxd'
1573 :param str series: Image series to remove, e.g. 'xenial'
1574
1575 """
1576 raise NotImplementedError()
1577
1578 def remove_machine(self, *machine_ids):
1579 """Remove a machine from this model.
1580
1581 :param str \*machine_ids: Ids of the machines to remove
1582
1583 """
1584 raise NotImplementedError()
1585 remove_machines = remove_machine
1586
1587 async def remove_ssh_key(self, user, key):
1588 """Remove a public SSH key(s) from this model.
1589
1590 :param str key: Full ssh key
1591 :param str user: Juju user to which the key is registered
1592
1593 """
1594 key_facade = client.KeyManagerFacade.from_connection(self.connection())
1595 key = base64.b64decode(bytes(key.strip().split()[1].encode('ascii')))
1596 key = hashlib.md5(key).hexdigest()
1597 key = ':'.join(a + b for a, b in zip(key[::2], key[1::2]))
1598 await key_facade.DeleteKeys([key], user)
1599 remove_ssh_keys = remove_ssh_key
1600
1601 def restore_backup(
1602 self, bootstrap=False, constraints=None, archive=None,
1603 backup_id=None, upload_tools=False):
1604 """Restore a backup archive to a new controller.
1605
1606 :param bool bootstrap: Bootstrap a new state machine
1607 :param constraints: Model constraints
1608 :type constraints: :class:`juju.Constraints`
1609 :param str archive: Path to backup archive to restore
1610 :param str backup_id: Id of backup to restore
1611 :param bool upload_tools: Upload tools if bootstrapping a new machine
1612
1613 """
1614 raise NotImplementedError()
1615
1616 def retry_provisioning(self):
1617 """Retry provisioning for failed machines.
1618
1619 """
1620 raise NotImplementedError()
1621
1622 def run(self, command, timeout=None):
1623 """Run command on all machines in this model.
1624
1625 :param str command: The command to run
1626 :param int timeout: Time to wait before command is considered failed
1627
1628 """
1629 raise NotImplementedError()
1630
1631 async def set_config(self, config):
1632 """Set configuration keys on this model.
1633
1634 :param dict config: Mapping of config keys to either string values or
1635 `ConfigValue` instances, as returned by `get_config`.
1636 """
1637 config_facade = client.ModelConfigFacade.from_connection(
1638 self.connection()
1639 )
1640 for key, value in config.items():
1641 if isinstance(value, ConfigValue):
1642 config[key] = value.value
1643 await config_facade.ModelSet(config)
1644
1645 async def set_constraints(self, constraints):
1646 """Set machine constraints on this model.
1647
1648 :param dict config: Mapping of model constraints
1649 """
1650 client_facade = client.ClientFacade.from_connection(self.connection())
1651 await client_facade.SetModelConstraints(
1652 application='',
1653 constraints=constraints)
1654
1655 async def get_action_output(self, action_uuid, wait=None):
1656 """Get the results of an action by ID.
1657
1658 :param str action_uuid: Id of the action
1659 :param int wait: Time in seconds to wait for action to complete.
1660 :return dict: Output from action
1661 :raises: :class:`JujuError` if invalid action_uuid
1662 """
1663 action_facade = client.ActionFacade.from_connection(
1664 self.connection()
1665 )
1666 entity = [{'tag': tag.action(action_uuid)}]
1667 # Cannot use self.wait_for_action as the action event has probably
1668 # already happened and self.wait_for_action works by processing
1669 # model deltas and checking if they match our type. If the action
1670 # has already occured then the delta has gone.
1671
1672 async def _wait_for_action_status():
1673 while True:
1674 action_output = await action_facade.Actions(entity)
1675 if action_output.results[0].status in ('completed', 'failed'):
1676 return
1677 else:
1678 await asyncio.sleep(1)
1679 await asyncio.wait_for(
1680 _wait_for_action_status(),
1681 timeout=wait)
1682 action_output = await action_facade.Actions(entity)
1683 # ActionResult.output is None if the action produced no output
1684 if action_output.results[0].output is None:
1685 output = {}
1686 else:
1687 output = action_output.results[0].output
1688 return output
1689
1690 async def get_action_status(self, uuid_or_prefix=None, name=None):
1691 """Get the status of all actions, filtered by ID, ID prefix, or name.
1692
1693 :param str uuid_or_prefix: Filter by action uuid or prefix
1694 :param str name: Filter by action name
1695
1696 """
1697 results = {}
1698 action_results = []
1699 action_facade = client.ActionFacade.from_connection(
1700 self.connection()
1701 )
1702 if name:
1703 name_results = await action_facade.FindActionsByNames([name])
1704 action_results.extend(name_results.actions[0].actions)
1705 if uuid_or_prefix:
1706 # Collect list of actions matching uuid or prefix
1707 matching_actions = await action_facade.FindActionTagsByPrefix(
1708 [uuid_or_prefix])
1709 entities = []
1710 for actions in matching_actions.matches.values():
1711 entities = [{'tag': a.tag} for a in actions]
1712 # Get action results matching action tags
1713 uuid_results = await action_facade.Actions(entities)
1714 action_results.extend(uuid_results.results)
1715 for a in action_results:
1716 results[tag.untag('action-', a.action.tag)] = a.status
1717 return results
1718
1719 def get_budget(self, budget_name):
1720 """Get budget usage info.
1721
1722 :param str budget_name: Name of budget
1723
1724 """
1725 raise NotImplementedError()
1726
1727 async def get_status(self, filters=None, utc=False):
1728 """Return the status of the model.
1729
1730 :param str filters: Optional list of applications, units, or machines
1731 to include, which can use wildcards ('*').
1732 :param bool utc: Display time as UTC in RFC3339 format
1733
1734 """
1735 client_facade = client.ClientFacade.from_connection(self.connection())
1736 return await client_facade.FullStatus(filters)
1737
1738 def sync_tools(
1739 self, all_=False, destination=None, dry_run=False, public=False,
1740 source=None, stream=None, version=None):
1741 """Copy Juju tools into this model.
1742
1743 :param bool all_: Copy all versions, not just the latest
1744 :param str destination: Path to local destination directory
1745 :param bool dry_run: Don't do the actual copy
1746 :param bool public: Tools are for a public cloud, so generate mirrors
1747 information
1748 :param str source: Path to local source directory
1749 :param str stream: Simplestreams stream for which to sync metadata
1750 :param str version: Copy a specific major.minor version
1751
1752 """
1753 raise NotImplementedError()
1754
1755 def unblock(self, *commands):
1756 """Unblock an operation that would alter this model.
1757
1758 :param str \*commands: The commands to unblock. Valid values are
1759 'all-changes', 'destroy-model', 'remove-object'
1760
1761 """
1762 raise NotImplementedError()
1763
1764 def unset_config(self, *keys):
1765 """Unset configuration on this model.
1766
1767 :param str \*keys: The keys to unset
1768
1769 """
1770 raise NotImplementedError()
1771
1772 def upgrade_gui(self):
1773 """Upgrade the Juju GUI for this model.
1774
1775 """
1776 raise NotImplementedError()
1777
1778 def upgrade_juju(
1779 self, dry_run=False, reset_previous_upgrade=False,
1780 upload_tools=False, version=None):
1781 """Upgrade Juju on all machines in a model.
1782
1783 :param bool dry_run: Don't do the actual upgrade
1784 :param bool reset_previous_upgrade: Clear the previous (incomplete)
1785 upgrade status
1786 :param bool upload_tools: Upload local version of tools
1787 :param str version: Upgrade to a specific version
1788
1789 """
1790 raise NotImplementedError()
1791
1792 def upload_backup(self, archive_path):
1793 """Store a backup archive remotely in Juju.
1794
1795 :param str archive_path: Path to local archive
1796
1797 """
1798 raise NotImplementedError()
1799
1800 @property
1801 def charmstore(self):
1802 return self._charmstore
1803
1804 async def get_metrics(self, *tags):
1805 """Retrieve metrics.
1806
1807 :param str \*tags: Tags of entities from which to retrieve metrics.
1808 No tags retrieves the metrics of all units in the model.
1809 :return: Dictionary of unit_name:metrics
1810
1811 """
1812 log.debug("Retrieving metrics for %s",
1813 ', '.join(tags) if tags else "all units")
1814
1815 metrics_facade = client.MetricsDebugFacade.from_connection(
1816 self.connection())
1817
1818 entities = [client.Entity(tag) for tag in tags]
1819 metrics_result = await metrics_facade.GetMetrics(entities)
1820
1821 metrics = collections.defaultdict(list)
1822
1823 for entity_metrics in metrics_result.results:
1824 error = entity_metrics.error
1825 if error:
1826 if "is not a valid tag" in error:
1827 raise ValueError(error.message)
1828 else:
1829 raise Exception(error.message)
1830
1831 for metric in entity_metrics.metrics:
1832 metrics[metric.unit].append(vars(metric))
1833
1834 return metrics
1835
1836
1837 def get_charm_series(path):
1838 """Inspects the charm directory at ``path`` and returns a default
1839 series from its metadata.yaml (the first item in the 'series' list).
1840
1841 Returns None if no series can be determined.
1842
1843 """
1844 md = Path(path) / "metadata.yaml"
1845 if not md.exists():
1846 return None
1847 data = yaml.load(md.open())
1848 series = data.get('series')
1849 return series[0] if series else None
1850
1851
1852 class BundleHandler:
1853 """
1854 Handle bundles by using the API to translate bundle YAML into a plan of
1855 steps and then dispatching each of those using the API.
1856 """
1857 def __init__(self, model):
1858 self.model = model
1859 self.charmstore = model.charmstore
1860 self.plan = []
1861 self.references = {}
1862 self._units_by_app = {}
1863 for unit_name, unit in model.units.items():
1864 app_units = self._units_by_app.setdefault(unit.application, [])
1865 app_units.append(unit_name)
1866 self.client_facade = client.ClientFacade.from_connection(
1867 model.connection())
1868 self.app_facade = client.ApplicationFacade.from_connection(
1869 model.connection())
1870 self.ann_facade = client.AnnotationsFacade.from_connection(
1871 model.connection())
1872
1873 async def _handle_local_charms(self, bundle):
1874 """Search for references to local charms (i.e. filesystem paths)
1875 in the bundle. Upload the local charms to the model, and replace
1876 the filesystem paths with appropriate 'local:' paths in the bundle.
1877
1878 Return the modified bundle.
1879
1880 :param dict bundle: Bundle dictionary
1881 :return: Modified bundle dictionary
1882
1883 """
1884 apps, args = [], []
1885
1886 default_series = bundle.get('series')
1887 apps_dict = bundle.get('applications', bundle.get('services', {}))
1888 for app_name in self.applications:
1889 app_dict = apps_dict[app_name]
1890 charm_dir = os.path.abspath(os.path.expanduser(app_dict['charm']))
1891 if not os.path.isdir(charm_dir):
1892 continue
1893 series = (
1894 app_dict.get('series') or
1895 default_series or
1896 get_charm_series(charm_dir)
1897 )
1898 if not series:
1899 raise JujuError(
1900 "Couldn't determine series for charm at {}. "
1901 "Add a 'series' key to the bundle.".format(charm_dir))
1902
1903 # Keep track of what we need to update. We keep a list of apps
1904 # that need to be updated, and a corresponding list of args
1905 # needed to update those apps.
1906 apps.append(app_name)
1907 args.append((charm_dir, series))
1908
1909 if apps:
1910 # If we have apps to update, spawn all the coroutines concurrently
1911 # and wait for them to finish.
1912 charm_urls = await asyncio.gather(*[
1913 self.model.add_local_charm_dir(*params)
1914 for params in args
1915 ], loop=self.model.loop)
1916 # Update the 'charm:' entry for each app with the new 'local:' url.
1917 for app_name, charm_url in zip(apps, charm_urls):
1918 apps_dict[app_name]['charm'] = charm_url
1919
1920 return bundle
1921
1922 async def fetch_plan(self, entity_id):
1923 is_local = not entity_id.startswith('cs:')
1924
1925 if is_local and os.path.isfile(entity_id):
1926 bundle_yaml = Path(entity_id).read_text()
1927 elif is_local and os.path.isdir(entity_id):
1928 bundle_yaml = (Path(entity_id) / "bundle.yaml").read_text()
1929 else:
1930 bundle_yaml = await self.charmstore.files(entity_id,
1931 filename='bundle.yaml',
1932 read_file=True)
1933 self.bundle = yaml.safe_load(bundle_yaml)
1934 self.bundle = await self._handle_local_charms(self.bundle)
1935
1936 self.plan = await self.client_facade.GetBundleChanges(
1937 yaml.dump(self.bundle))
1938
1939 if self.plan.errors:
1940 raise JujuError(self.plan.errors)
1941
1942 async def execute_plan(self):
1943 for step in self.plan.changes:
1944 method = getattr(self, step.method)
1945 result = await method(*step.args)
1946 self.references[step.id_] = result
1947
1948 @property
1949 def applications(self):
1950 apps_dict = self.bundle.get('applications',
1951 self.bundle.get('services', {}))
1952 return list(apps_dict.keys())
1953
1954 def resolve(self, reference):
1955 if reference and reference.startswith('$'):
1956 reference = self.references[reference[1:]]
1957 return reference
1958
1959 async def addCharm(self, charm, series):
1960 """
1961 :param charm string:
1962 Charm holds the URL of the charm to be added.
1963
1964 :param series string:
1965 Series holds the series of the charm to be added
1966 if the charm default is not sufficient.
1967 """
1968 # We don't add local charms because they've already been added
1969 # by self._handle_local_charms
1970 if charm.startswith('local:'):
1971 return charm
1972
1973 entity_id = await self.charmstore.entityId(charm)
1974 log.debug('Adding %s', entity_id)
1975 await self.client_facade.AddCharm(None, entity_id)
1976 return entity_id
1977
1978 async def addMachines(self, params=None):
1979 """
1980 :param params dict:
1981 Dictionary specifying the machine to add. All keys are optional.
1982 Keys include:
1983
1984 series: string specifying the machine OS series.
1985
1986 constraints: string holding machine constraints, if any. We'll
1987 parse this into the json friendly dict that the juju api
1988 expects.
1989
1990 container_type: string holding the type of the container (for
1991 instance ""lxd" or kvm"). It is not specified for top level
1992 machines.
1993
1994 parent_id: string holding a placeholder pointing to another
1995 machine change or to a unit change. This value is only
1996 specified in the case this machine is a container, in
1997 which case also ContainerType is set.
1998
1999 """
2000 params = params or {}
2001
2002 # Normalize keys
2003 params = {normalize_key(k): params[k] for k in params.keys()}
2004
2005 # Fix up values, as necessary.
2006 if 'parent_id' in params:
2007 if params['parent_id'].startswith('$addUnit'):
2008 unit = self.resolve(params['parent_id'])[0]
2009 params['parent_id'] = unit.machine.entity_id
2010 else:
2011 params['parent_id'] = self.resolve(params['parent_id'])
2012
2013 params['constraints'] = parse_constraints(
2014 params.get('constraints'))
2015 params['jobs'] = params.get('jobs', ['JobHostUnits'])
2016
2017 if params.get('container_type') == 'lxc':
2018 log.warning('Juju 2.0 does not support lxc containers. '
2019 'Converting containers to lxd.')
2020 params['container_type'] = 'lxd'
2021
2022 # Submit the request.
2023 params = client.AddMachineParams(**params)
2024 results = await self.client_facade.AddMachines([params])
2025 error = results.machines[0].error
2026 if error:
2027 raise ValueError("Error adding machine: %s" % error.message)
2028 machine = results.machines[0].machine
2029 log.debug('Added new machine %s', machine)
2030 return machine
2031
2032 async def addRelation(self, endpoint1, endpoint2):
2033 """
2034 :param endpoint1 string:
2035 :param endpoint2 string:
2036 Endpoint1 and Endpoint2 hold relation endpoints in the
2037 "application:interface" form, where the application is always a
2038 placeholder pointing to an application change, and the interface is
2039 optional. Examples are "$deploy-42:web" or just "$deploy-42".
2040 """
2041 endpoints = [endpoint1, endpoint2]
2042 # resolve indirect references
2043 for i in range(len(endpoints)):
2044 parts = endpoints[i].split(':')
2045 parts[0] = self.resolve(parts[0])
2046 endpoints[i] = ':'.join(parts)
2047
2048 log.info('Relating %s <-> %s', *endpoints)
2049 return await self.model.add_relation(*endpoints)
2050
2051 async def deploy(self, charm, series, application, options, constraints,
2052 storage, endpoint_bindings, resources):
2053 """
2054 :param charm string:
2055 Charm holds the URL of the charm to be used to deploy this
2056 application.
2057
2058 :param series string:
2059 Series holds the series of the application to be deployed
2060 if the charm default is not sufficient.
2061
2062 :param application string:
2063 Application holds the application name.
2064
2065 :param options map[string]interface{}:
2066 Options holds application options.
2067
2068 :param constraints string:
2069 Constraints holds the optional application constraints.
2070
2071 :param storage map[string]string:
2072 Storage holds the optional storage constraints.
2073
2074 :param endpoint_bindings map[string]string:
2075 EndpointBindings holds the optional endpoint bindings
2076
2077 :param resources map[string]int:
2078 Resources identifies the revision to use for each resource
2079 of the application's charm.
2080 """
2081 # resolve indirect references
2082 charm = self.resolve(charm)
2083 # the bundle plan doesn't actually do anything with resources, even
2084 # though it ostensibly gives us something (None) for that param
2085 if not charm.startswith('local:'):
2086 resources = await self.model._add_store_resources(application,
2087 charm)
2088 await self.model._deploy(
2089 charm_url=charm,
2090 application=application,
2091 series=series,
2092 config=options,
2093 constraints=constraints,
2094 endpoint_bindings=endpoint_bindings,
2095 resources=resources,
2096 storage=storage,
2097 )
2098 return application
2099
2100 async def addUnit(self, application, to):
2101 """
2102 :param application string:
2103 Application holds the application placeholder name for which a unit
2104 is added.
2105
2106 :param to string:
2107 To holds the optional location where to add the unit, as a
2108 placeholder pointing to another unit change or to a machine change.
2109 """
2110 application = self.resolve(application)
2111 placement = self.resolve(to)
2112 if self._units_by_app.get(application):
2113 # enough units for this application already exist;
2114 # claim one, and carry on
2115 # NB: this should probably honor placement, but the juju client
2116 # doesn't, so we're not bothering, either
2117 unit_name = self._units_by_app[application].pop()
2118 log.debug('Reusing unit %s for %s', unit_name, application)
2119 return self.model.units[unit_name]
2120
2121 log.debug('Adding new unit for %s%s', application,
2122 ' to %s' % placement if placement else '')
2123 return await self.model.applications[application].add_unit(
2124 count=1,
2125 to=placement,
2126 )
2127
2128 async def expose(self, application):
2129 """
2130 :param application string:
2131 Application holds the placeholder name of the application that must
2132 be exposed.
2133 """
2134 application = self.resolve(application)
2135 log.info('Exposing %s', application)
2136 return await self.model.applications[application].expose()
2137
2138 async def setAnnotations(self, id_, entity_type, annotations):
2139 """
2140 :param id_ string:
2141 Id is the placeholder for the application or machine change
2142 corresponding to the entity to be annotated.
2143
2144 :param entity_type EntityType:
2145 EntityType holds the type of the entity, "application" or
2146 "machine".
2147
2148 :param annotations map[string]string:
2149 Annotations holds the annotations as key/value pairs.
2150 """
2151 entity_id = self.resolve(id_)
2152 try:
2153 entity = self.model.state.get_entity(entity_type, entity_id)
2154 except KeyError:
2155 entity = await self.model._wait_for_new(entity_type, entity_id)
2156 return await entity.set_annotations(annotations)
2157
2158
2159 class CharmStore:
2160 """
2161 Async wrapper around theblues.charmstore.CharmStore
2162 """
2163 def __init__(self, loop):
2164 self.loop = loop
2165 self._cs = theblues.charmstore.CharmStore(timeout=5)
2166
2167 def __getattr__(self, name):
2168 """
2169 Wrap method calls in coroutines that use run_in_executor to make them
2170 async.
2171 """
2172 attr = getattr(self._cs, name)
2173 if not callable(attr):
2174 wrapper = partial(getattr, self._cs, name)
2175 setattr(self, name, wrapper)
2176 else:
2177 async def coro(*args, **kwargs):
2178 method = partial(attr, *args, **kwargs)
2179 for attempt in range(1, 4):
2180 try:
2181 return await self.loop.run_in_executor(None, method)
2182 except theblues.errors.ServerError:
2183 if attempt == 3:
2184 raise
2185 await asyncio.sleep(1, loop=self.loop)
2186 setattr(self, name, coro)
2187 wrapper = coro
2188 return wrapper
2189
2190
2191 class CharmArchiveGenerator:
2192 """
2193 Create a Zip archive of a local charm directory for upload to a controller.
2194
2195 This is used automatically by
2196 `Model.add_local_charm_dir <#juju.model.Model.add_local_charm_dir>`_.
2197 """
2198 def __init__(self, path):
2199 self.path = os.path.abspath(os.path.expanduser(path))
2200
2201 def make_archive(self, path):
2202 """Create archive of directory and write to ``path``.
2203
2204 :param path: Path to archive
2205
2206 Ignored::
2207
2208 * build/\* - This is used for packing the charm itself and any
2209 similar tasks.
2210 * \*/.\* - Hidden files are all ignored for now. This will most
2211 likely be changed into a specific ignore list
2212 (.bzr, etc)
2213
2214 """
2215 zf = zipfile.ZipFile(path, 'w', zipfile.ZIP_DEFLATED)
2216 for dirpath, dirnames, filenames in os.walk(self.path):
2217 relative_path = dirpath[len(self.path) + 1:]
2218 if relative_path and not self._ignore(relative_path):
2219 zf.write(dirpath, relative_path)
2220 for name in filenames:
2221 archive_name = os.path.join(relative_path, name)
2222 if not self._ignore(archive_name):
2223 real_path = os.path.join(dirpath, name)
2224 self._check_type(real_path)
2225 if os.path.islink(real_path):
2226 self._check_link(real_path)
2227 self._write_symlink(
2228 zf, os.readlink(real_path), archive_name)
2229 else:
2230 zf.write(real_path, archive_name)
2231 zf.close()
2232 return path
2233
2234 def _check_type(self, path):
2235 """Check the path
2236 """
2237 s = os.stat(path)
2238 if stat.S_ISDIR(s.st_mode) or stat.S_ISREG(s.st_mode):
2239 return path
2240 raise ValueError("Invalid Charm at % %s" % (
2241 path, "Invalid file type for a charm"))
2242
2243 def _check_link(self, path):
2244 link_path = os.readlink(path)
2245 if link_path[0] == "/":
2246 raise ValueError(
2247 "Invalid Charm at %s: %s" % (
2248 path, "Absolute links are invalid"))
2249 path_dir = os.path.dirname(path)
2250 link_path = os.path.join(path_dir, link_path)
2251 if not link_path.startswith(os.path.abspath(self.path)):
2252 raise ValueError(
2253 "Invalid charm at %s %s" % (
2254 path, "Only internal symlinks are allowed"))
2255
2256 def _write_symlink(self, zf, link_target, link_path):
2257 """Package symlinks with appropriate zipfile metadata."""
2258 info = zipfile.ZipInfo()
2259 info.filename = link_path
2260 info.create_system = 3
2261 # Magic code for symlinks / py2/3 compat
2262 # 27166663808 = (stat.S_IFLNK | 0755) << 16
2263 info.external_attr = 2716663808
2264 zf.writestr(info, link_target)
2265
2266 def _ignore(self, path):
2267 if path == "build" or path.startswith("build/"):
2268 return True
2269 if path.startswith('.'):
2270 return True