13 from concurrent
.futures
import CancelledError
14 from functools
import partial
15 from pathlib
import Path
17 import theblues
.charmstore
18 import theblues
.errors
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
35 log
= logging
.getLogger(__name__
)
39 """Wrapper around an observer callable.
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.
45 def __init__(self
, callable_
, entity_type
, action
, entity_id
, predicate
):
46 self
.callable_
= callable_
47 self
.entity_type
= entity_type
49 self
.entity_id
= entity_id
50 self
.predicate
= predicate
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('$'):
58 async def __call__(self
, delta
, old
, new
, model
):
59 await self
.callable_(delta
, old
, new
, model
)
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.
66 if (self
.entity_id
and delta
.get_id() and
67 not re
.match(self
.entity_id
, str(delta
.get_id()))):
70 if self
.entity_type
and self
.entity_type
!= delta
.entity
:
73 if self
.action
and self
.action
!= delta
.type:
76 if self
.predicate
and not self
.predicate(delta
):
84 Base class for creating observers that react to changes in a model.
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
)
91 async def on_change(self
, delta
, old
, new
, model
):
92 """Generic model-change handler.
94 This should be overridden in a subclass.
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`
106 """Holds the state of the model, including the delta history of all
107 entities in the model.
110 def __init__(self
, model
):
114 def _live_entity_map(self
, entity_type
):
115 """Return an id:Entity map of all the living entities of
116 type ``entity_type``.
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
126 def applications(self
):
127 """Return a map of application-name:Application for all applications
128 currently in the model.
131 return self
._live
_entity
_map
('application')
135 """Return a map of machine-id:Machine for all machines currently in
139 return self
._live
_entity
_map
('machine')
143 """Return a map of unit-id:Unit for all units currently in
147 return self
._live
_entity
_map
('unit')
151 """Return a map of relation-id:Relation for all relations currently in
155 return self
._live
_entity
_map
('relation')
157 def entity_history(self
, entity_type
, entity_id
):
158 """Return the history deque for an entity.
161 return self
.state
[entity_type
][entity_id
]
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
168 return self
.entity_history(entity_type
, entity_id
)[history_index
]
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.:
174 old_obj, new_obj = self.apply_delta(delta)
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.
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.
185 .setdefault(delta
.entity
, {})
186 .setdefault(delta
.get_id(), collections
.deque())
189 history
.append(delta
.data
)
190 if delta
.type == 'remove':
193 entity
= self
.get_entity(delta
.entity
, delta
.get_id())
194 return entity
.previous(), entity
197 self
, entity_type
, entity_id
, history_index
=-1, connected
=True):
198 """Return an object instance for the given entity_type and id.
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.
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:
212 self
.entity_data(entity_type
, entity_id
, history_index
)
216 entity_class
= get_entity_class(entity_type
)
218 entity_id
, self
.model
, history_index
=history_index
,
223 """An object in the Model tree"""
225 def __init__(self
, entity_id
, model
, history_index
=-1, connected
=True):
226 """Initialize a new entity
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
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
237 self
.entity_id
= entity_id
239 self
._history
_index
= history_index
240 self
.connected
= connected
241 self
.connection
= model
.connection()
244 return '<{} entity_id="{}">'.format(type(self
).__name
__,
247 def __getattr__(self
, name
):
248 """Fetch object attributes from the underlying data dict held in the
253 return self
.safe_data
[name
]
255 name
= name
.replace('_', '-')
256 if name
in self
.safe_data
:
257 return self
.safe_data
[name
]
262 return bool(self
.data
)
264 def on_change(self
, callable_
):
265 """Add a change observer to this entity.
268 self
.model
.add_observer(
269 callable_
, self
.entity_type
, 'change', self
.entity_id
)
271 def on_remove(self
, callable_
):
272 """Add a remove observer to this entity.
275 self
.model
.add_observer(
276 callable_
, self
.entity_type
, 'remove', self
.entity_id
)
279 def entity_type(self
):
280 """A string identifying the entity type of this object, e.g.
281 'application' or 'unit', etc.
284 return self
.__class
__.__name
__.lower()
288 """Return True if this object represents the current state of the
289 entity in the underlying model.
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.
296 return self
._history
_index
== -1
300 """Returns True if this entity no longer exists in the underlying
306 self
.model
.state
.entity_data(
307 self
.entity_type
, self
.entity_id
, -1) is None
312 """Returns True if this entity still exists in the underlying
320 """The data dictionary for this entity.
323 return self
.model
.state
.entity_data(
324 self
.entity_type
, self
.entity_id
, self
._history
_index
)
328 """The data dictionary for this entity.
330 If this `ModelEntity` points to the dead state, it will
331 raise `DeadEntityException`.
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
))
343 """Return a copy of this object as was at its previous state in
346 Returns None if this object is new (and therefore has no history).
348 The returned object is always "disconnected", i.e. does not receive
352 return self
.model
.state
.get_entity(
353 self
.entity_type
, self
.entity_id
, self
._history
_index
- 1,
357 """Return a copy of this object at its next state in
360 Returns None if this object is already the latest.
362 The returned object is "disconnected", i.e. does not receive
363 live updates, unless it is current (latest).
366 if self
._history
_index
== -1:
369 new_index
= self
._history
_index
+ 1
371 new_index
== len(self
.model
.state
.entity_history(
372 self
.entity_type
, self
.entity_id
)) - 1
374 return self
.model
.state
.get_entity(
375 self
.entity_type
, self
.entity_id
, self
._history
_index
- 1,
379 """Return a copy of this object at its current state in the model.
381 Returns self if this object is already the latest.
383 The returned object is always "connected", i.e. receives
384 live updates from the model.
387 if self
._history
_index
== -1:
390 return self
.model
.state
.get_entity(self
.entity_type
, self
.entity_id
)
395 The main API for interacting with a Juju model.
404 """Instantiate a new Model.
406 The connect method will need to be called before this
407 object can be used for anything interesting.
409 If jujudata is None, jujudata.FileJujuData will be used.
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
418 self
._connector
= connector
.Connector(
420 max_frame_size
=max_frame_size
,
421 bakery_client
=bakery_client
,
424 self
._observers
= weakref
.WeakValueDictionary()
425 self
.state
= ModelState(self
)
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
)
433 def is_connected(self
):
434 """Reports whether the Model is currently connected."""
435 return self
._connector
.is_connected()
439 return self
._connector
.loop
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()
446 async def get_controller(self
):
447 """Return a Controller instance for the currently connected model.
450 from juju
.controller
import Controller
451 controller
= Controller(jujudata
=self
._connector
.jujudata
)
452 kwargs
= self
.connection().connect_params()
454 await controller
._connect
_direct
(**kwargs
)
457 async def __aenter__(self
):
461 async def __aexit__(self
, exc_type
, exc
, tb
):
462 await self
.disconnect()
464 async def connect(self
, *args
, **kwargs
):
465 """Connect to a juju model.
467 This supports two calling conventions:
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.
474 Otherwise, all of the ``endpoint``, ``uuid``, and authentication
475 information (``username`` and ``password``, or ``bakery_client`` and/or
476 ``macaroons``) are required.
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``.
482 Available parameters are:
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
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
498 :param asyncio.BaseEventLoop loop: The event loop to use for async
500 :param int max_frame_size: The maximum websocket frame size to allow.
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')
509 model_name
= kwargs
.pop('model_name', None)
510 await self
._connector
.connect_model(model_name
, **kwargs
)
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')
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
()
548 async def connect_model(self
, model_name
):
550 .. deprecated:: 0.6.2
551 Use ``connect(model_name=model_name)`` instead.
553 return await self
.connect(model_name
=model_name
)
555 async def connect_current(self
):
557 .. deprecated:: 0.6.2
558 Use ``connect()`` instead.
560 return await self
.connect()
562 async def _connect_direct(self
, **kwargs
):
563 await self
.disconnect()
564 await self
._connector
.connect(**kwargs
)
565 await self
._after
_connect
()
567 async def _after_connect(self
):
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()
578 await self
.get_info()
580 async def disconnect(self
):
581 """Shut down the watcher task and close websockets.
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()
590 if self
.is_connected():
591 log
.debug('Closing model connection')
592 await self
._connector
.disconnect()
595 async def add_local_charm_dir(self
, charm_dir
, series
):
596 """Upload a local charm to the model.
598 This will automatically generate an archive from
601 :param charm_dir: Path to the charm directory
602 :param series: Charm series
605 fh
= tempfile
.NamedTemporaryFile()
606 CharmArchiveGenerator(charm_dir
).make_archive(fh
.name
)
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
)
612 log
.debug('Uploaded local charm: %s -> %s', charm_dir
, charm_url
)
615 def add_local_charm(self
, charm_file
, series
, size
=None):
616 """Upload a local charm archive to the model.
618 Returns the 'local:...' url that should be used to deploy the charm.
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
626 Uses an https endpoint at the same host:port as the wss.
627 Supports large file uploads.
631 This method will block. Consider using :meth:`add_local_charm_dir`
635 conn
, headers
, path_prefix
= self
.connection().https_connection()
636 path
= "%s/charms?series=%s" % (path_prefix
, series
)
637 headers
['Content-Type'] = 'application/zip'
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']
648 def all_units_idle(self
):
649 """Return True if all units are idle.
652 for unit
in self
.units
.values():
653 unit_status
= unit
.data
['agent-status']['current']
654 if unit_status
!= 'idle':
658 async def reset(self
, force
=False):
659 """Reset the model to a clean state.
661 :param bool force: Force-terminate machines.
663 This returns only after the model has reached a clean state. "Clean"
664 means no applications or machines exist in the model.
667 log
.debug('Resetting model')
668 for app
in self
.applications
.values():
670 for machine
in self
.machines
.values():
671 await machine
.destroy(force
=force
)
672 await self
.block_until(
673 lambda: len(self
.machines
) == 0
676 async def block_until(self
, *conditions
, timeout
=None, wait_period
=0.5):
677 """Return only after all conditions are true.
679 Raises `websockets.ConnectionClosed` if disconnected.
682 return not (self
.is_connected() and self
.connection().is_open
)
685 return _disconnected() or all(c() for c
in conditions
)
687 await utils
.block_until(done
,
689 wait_period
=wait_period
,
692 raise websockets
.ConnectionClosed(1006, 'no reason')
695 def applications(self
):
696 """Return a map of application-name:Application for all applications
697 currently in the model.
700 return self
.state
.applications
704 """Return a map of machine-id:Machine for all machines currently in
708 return self
.state
.machines
712 """Return a map of unit-id:Unit for all units currently in
716 return self
.state
.units
720 """Return a list of all Relations currently in the model.
723 return list(self
.state
.relations
.values())
725 async def get_info(self
):
726 """Return a client.ModelInfo object for this Model.
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
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.
738 facade
= client
.ClientFacade
.from_connection(self
.connection())
740 self
._info
= await facade
.ModelInfo()
741 log
.debug('Got ModelInfo: %s', vars(self
.info
))
747 """Return the cached client.ModelInfo object for this Model.
749 If Model.get_info() has not been called, this will return None.
754 self
, callable_
, entity_type
=None, action
=None, entity_id
=None,
756 """Register an "on-model-change" callback
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:
762 delta - An instance of :class:`juju.delta.EntityDelta`
763 containing the raw delta data recv'd from the Juju
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
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
775 model - The :class:`Model` itself.
777 Events for which ``callable_`` is called can be specified by passing
778 entity_type, action, and/or entitiy_id filter criteria, e.g.::
782 entity_type='application', action='add', entity_id='ubuntu')
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.
789 observer
= _Observer(
790 callable_
, entity_type
, action
, entity_id
, predicate
)
791 self
._observers
[observer
] = callable_
794 """Start an asynchronous watch against this model.
796 See :meth:`add_observer` to register an onchange callback.
799 async def _all_watcher():
801 allwatcher
= client
.AllWatcherFacade
.from_connection(
803 while not self
._watch
_stopping
.is_set():
805 results
= await utils
.run_with_interrupt(
807 self
._watch
_stopping
,
808 loop
=self
._connector
.loop
)
809 except JujuAPIError
as e
:
810 if 'watcher was stopped' not in str(e
):
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
818 # controller stopped our watcher for some reason
819 # but we're not actually stopping, so just restart it
821 'Watcher: watcher stopped, restarting')
824 except websockets
.ConnectionClosed
:
825 monitor
= self
.connection().monitor
826 if monitor
.status
== monitor
.ERROR
:
827 # closed unexpectedly, try to reopen
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')
839 # closed on request, go ahead and shutdown
841 if self
._watch
_stopping
.is_set():
843 await allwatcher
.Stop()
844 except websockets
.ConnectionClosed
:
845 pass # can't stop on a closed conn
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
:
855 log
.exception('Error in watcher')
858 self
._watch
_stopped
.set()
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())
866 async def _notify_observers(self
, delta
, old_obj
, new_obj
):
867 """Call observing callbacks, notifying them of a change in model state
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.
873 :param new_obj: The object in the model that is created or updated
874 by applying this delta.
877 if new_obj
and not old_obj
:
881 'Model changed: %s %s %s',
882 delta
.entity
, delta
.type, delta
.get_id())
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
)
889 async def _wait(self
, entity_type
, entity_id
, action
, predicate
=None):
891 Block the calling routine until a given action has happened to the
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.
904 q
= asyncio
.Queue(loop
=self
._connector
.loop
)
906 async def callback(delta
, old
, new
, model
):
907 await q
.put(delta
.get_id())
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
913 return self
.state
._live
_entity
_map
(entity_type
).get(entity_id
)
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.
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
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)
929 async def wait_for_action(self
, action_id
):
930 """Given an action, wait for it to complete."""
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:]
937 def predicate(delta
):
938 return delta
.data
['status'] in ('completed', 'failed')
940 return await self
._wait
('action', action_id
, None, predicate
)
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.
947 :param str spec: Machine specification
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
958 :param dict constraints: Machine constraints, which can contain the
979 :param list disks: List of disk constraint dictionaries, which can
980 contain the following keys::
994 :param str series: Series, e.g. 'xenial'
996 Supported container types are: lxd, kvm
998 When deploying a container to an existing machine, constraints cannot
1002 params
= client
.AddMachineParams()
1005 if spec
.startswith("ssh:"):
1006 placement
, target
, private_key_path
= spec
.split(":")
1007 user
, host
= target
.split("@")
1009 sshProvisioner
= provisioner
.SSHProvisioner(
1012 private_key_path
=private_key_path
,
1015 params
= sshProvisioner
.provision_machine()
1017 placement
= parse_placement(spec
)
1019 params
.placement
= placement
[0]
1021 params
.jobs
= ['JobHostUnits']
1024 params
.constraints
= client
.Value
.from_json(constraints
)
1028 client
.Constraints
.from_json(o
) for o
in disks
]
1031 params
.series
= series
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
1038 raise ValueError("Error adding machine: %s" % error
.message
)
1039 machine_id
= results
.machines
[0].machine
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(
1051 log
.debug('Added new machine %s', machine_id
)
1052 return await self
._wait
_for
_new
('machine', machine_id
)
1054 async def add_relation(self
, relation1
, relation2
):
1055 """Add a relation between two applications.
1057 :param str relation1: '<application>[:<relation_name>]'
1058 :param str relation2: '<application>[:<relation_name>]'
1061 connection
= self
.connection()
1062 app_facade
= client
.ApplicationFacade
.from_connection(connection
)
1065 'Adding relation %s <-> %s', relation1
, relation2
)
1067 def _find_relation(*specs
):
1068 for rel
in self
.relations
:
1069 if rel
.matches(*specs
):
1074 result
= await app_facade
.AddRelation([relation1
, relation2
])
1075 except JujuAPIError
as e
:
1076 if 'relation already exists' not in e
.message
:
1078 rel
= _find_relation(relation1
, relation2
)
1081 raise JujuError('Relation {} {} exists but not in model'.format(
1082 relation1
, relation2
))
1084 specs
= ['{}:{}'.format(app
, data
['name'])
1085 for app
, data
in result
.endpoints
.items()]
1087 await self
.block_until(lambda: _find_relation(*specs
) is not None)
1088 return _find_relation(*specs
)
1090 def add_space(self
, name
, *cidrs
):
1091 """Add a new network space.
1093 Adds a new space with the given name and associates the given
1094 (optional) list of existing subnet CIDRs with it.
1096 :param str name: Name of the space
1097 :param *cidrs: Optional list of existing subnet CIDRs
1100 raise NotImplementedError()
1102 async def add_ssh_key(self
, user
, key
):
1103 """Add a public SSH key to this model.
1105 :param str user: The username of the user
1106 :param str key: The public ssh key
1109 key_facade
= client
.KeyManagerFacade
.from_connection(self
.connection())
1110 return await key_facade
.AddKeys([key
], user
)
1111 add_ssh_keys
= add_ssh_key
1113 def add_subnet(self
, cidr_or_id
, space
, *zones
):
1114 """Add an existing subnet to this model.
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
1121 raise NotImplementedError()
1123 def get_backups(self
):
1124 """Retrieve metadata for backups in this model.
1127 raise NotImplementedError()
1129 def block(self
, *commands
):
1130 """Add a new block to this model.
1132 :param str *commands: The commands to block. Valid values are
1133 'all-changes', 'destroy-model', 'remove-object'
1136 raise NotImplementedError()
1138 def get_blocks(self
):
1139 """List blocks for this model.
1142 raise NotImplementedError()
1144 def get_cached_images(self
, arch
=None, kind
=None, series
=None):
1145 """Return a list of cached OS images.
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'
1152 raise NotImplementedError()
1154 def create_backup(self
, note
=None, no_download
=False):
1155 """Create a backup of this model.
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
1162 raise NotImplementedError()
1164 def create_storage_pool(self
, name
, provider_type
, **pool_config
):
1165 """Create or define a storage pool.
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
1172 raise NotImplementedError()
1175 self
, no_tail
=False, exclude_module
=None, include_module
=None,
1176 include
=None, level
=None, limit
=0, lines
=10, replay
=False,
1178 """Get log messages for this model.
1180 :param bool no_tail: Stop after returning existing log messages
1181 :param list exclude_module: Do not show log messages for these logging
1183 :param list include_module: Only show log messages for these logging
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
1192 :param bool replay: Yield the entire log, and keep yielding
1193 :param list exclude: Do not show log messages for these entities
1196 raise NotImplementedError()
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('/')
1203 parts
= entity_url
.split('/')
1204 if parts
[0].startswith('~'):
1207 # series was specified in the URL
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]
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, devices
=None):
1219 """Deploy a new service or bundle.
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:
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
1243 If None, a new machine is provisioned.
1248 - support local resources
1253 k
: client
.Constraints(**v
)
1254 for k
, v
in storage
.items()
1257 entity_path
= Path(entity_url
.replace('local:', ''))
1258 bundle_path
= entity_path
/ 'bundle.yaml'
1259 metadata_path
= entity_path
/ 'metadata.yaml'
1262 entity_url
.startswith('local:') or
1263 entity_path
.is_dir() or
1264 entity_path
.is_file()
1267 entity_id
= entity_url
.replace('local:', '')
1269 entity
= await self
.charmstore
.entity(entity_url
, channel
=channel
,
1270 include_stats
=False)
1271 entity_id
= entity
['Id']
1273 client_facade
= client
.ClientFacade
.from_connection(self
.connection())
1275 is_bundle
= ((is_local
and
1276 (entity_id
.endswith('.yaml') and entity_path
.exists()) or
1277 bundle_path
.exists()) or
1278 (not is_local
and 'bundle/' in entity_id
))
1281 handler
= BundleHandler(self
)
1282 await handler
.fetch_plan(entity_id
)
1283 await handler
.execute_plan()
1284 extant_apps
= {app
for app
in self
.applications
}
1285 pending_apps
= set(handler
.applications
) - extant_apps
1287 # new apps will usually be in the model by now, but if some
1288 # haven't made it yet we'll need to wait on them to be added
1289 await asyncio
.gather(*[
1290 asyncio
.ensure_future(
1291 self
._wait
_for
_new
('application', app_name
),
1292 loop
=self
._connector
.loop
)
1293 for app_name
in pending_apps
1294 ], loop
=self
._connector
.loop
)
1295 return [app
for name
, app
in self
.applications
.items()
1296 if name
in handler
.applications
]
1299 if not application_name
:
1300 application_name
= entity
['Meta']['charm-metadata']['Name']
1302 series
= self
._get
_series
(entity_url
, entity
)
1303 await client_facade
.AddCharm(channel
, entity_id
)
1304 # XXX: we're dropping local resources here, but we don't
1305 # actually support them yet anyway
1306 resources
= await self
._add
_store
_resources
(application_name
,
1310 if not application_name
:
1311 metadata
= yaml
.load(metadata_path
.read_text())
1312 application_name
= metadata
['name']
1313 # We have a local charm dir that needs to be uploaded
1314 charm_dir
= os
.path
.abspath(
1315 os
.path
.expanduser(entity_id
))
1316 series
= series
or get_charm_series(charm_dir
)
1319 "Couldn't determine series for charm at {}. "
1320 "Pass a 'series' kwarg to Model.deploy().".format(
1322 entity_id
= await self
.add_local_charm_dir(charm_dir
, series
)
1323 return await self
._deploy
(
1324 charm_url
=entity_id
,
1325 application
=application_name
,
1327 config
=config
or {},
1328 constraints
=constraints
,
1329 endpoint_bindings
=bind
,
1330 resources
=resources
,
1333 num_units
=num_units
,
1334 placement
=parse_placement(to
),
1338 async def _add_store_resources(self
, application
, entity_url
,
1339 overrides
=None, entity
=None):
1341 # avoid extra charm store call if one was already made
1342 entity
= await self
.charmstore
.entity(entity_url
,
1343 include_stats
=False)
1346 'description': resource
['Description'],
1347 'fingerprint': resource
['Fingerprint'],
1348 'name': resource
['Name'],
1349 'path': resource
['Path'],
1350 'revision': resource
['Revision'],
1351 'size': resource
['Size'],
1352 'type_': resource
['Type'],
1354 } for resource
in entity
['Meta']['resources']
1358 names
= {r
['name'] for r
in resources
}
1359 unknown
= overrides
.keys() - names
1361 raise JujuError('Unrecognized resource{}: {}'.format(
1362 's' if len(unknown
) > 1 else '',
1363 ', '.join(unknown
)))
1364 for resource
in resources
:
1365 if resource
['name'] in overrides
:
1366 resource
['revision'] = overrides
[resource
['name']]
1371 resources_facade
= client
.ResourcesFacade
.from_connection(
1373 response
= await resources_facade
.AddPendingResources(
1374 tag
.application(application
),
1376 [client
.CharmResource(**resource
) for resource
in resources
])
1377 resource_map
= {resource
['name']: pid
1379 in zip(resources
, response
.pending_ids
)}
1382 async def _deploy(self
, charm_url
, application
, series
, config
,
1383 constraints
, endpoint_bindings
, resources
, storage
,
1384 channel
=None, num_units
=None, placement
=None,
1386 """Logic shared between `Model.deploy` and `BundleHandler.deploy`.
1388 log
.info('Deploying %s', charm_url
)
1390 # stringify all config values for API, and convert to YAML
1391 config
= {k
: str(v
) for k
, v
in config
.items()}
1392 config
= yaml
.dump({application
: config
},
1393 default_flow_style
=False)
1395 app_facade
= client
.ApplicationFacade
.from_connection(
1398 app
= client
.ApplicationDeploy(
1399 charm_url
=charm_url
,
1400 application
=application
,
1404 constraints
=parse_constraints(constraints
),
1405 endpoint_bindings
=endpoint_bindings
,
1406 num_units
=num_units
,
1407 resources
=resources
,
1409 placement
=placement
,
1412 result
= await app_facade
.Deploy([app
])
1413 errors
= [r
.error
.message
for r
in result
.results
if r
.error
]
1415 raise JujuError('\n'.join(errors
))
1416 return await self
._wait
_for
_new
('application', application
)
1418 async def destroy(self
):
1419 """Terminate all machines and resources for this model.
1420 Is already implemented in controller.py.
1422 raise NotImplementedError()
1424 async def destroy_unit(self
, *unit_names
):
1425 """Destroy units by name.
1428 connection
= self
.connection()
1429 app_facade
= client
.ApplicationFacade
.from_connection(connection
)
1432 'Destroying unit%s %s',
1433 's' if len(unit_names
) == 1 else '',
1434 ' '.join(unit_names
))
1436 return await app_facade
.DestroyUnits(list(unit_names
))
1437 destroy_units
= destroy_unit
1439 def get_backup(self
, archive_id
):
1440 """Download a backup archive file.
1442 :param str archive_id: The id of the archive to download
1443 :return str: Path to the archive file
1446 raise NotImplementedError()
1449 self
, num_controllers
=0, constraints
=None, series
=None, to
=None):
1450 """Ensure sufficient controllers exist to provide redundancy.
1452 :param int num_controllers: Number of controllers to make available
1453 :param constraints: Constraints to apply to the controller machines
1454 :type constraints: :class:`juju.Constraints`
1455 :param str series: Series of the controller machines
1456 :param list to: Placement directives for controller machines, e.g.::
1459 'lxc:7' - new lxc container on machine 7
1460 '24/lxc/3' - lxc container 3 or machine 24
1462 If None, a new machine is provisioned.
1465 raise NotImplementedError()
1467 async def get_config(self
):
1468 """Return the configuration settings for this model.
1470 :returns: A ``dict`` mapping keys to `ConfigValue` instances,
1471 which have `source` and `value` attributes.
1473 config_facade
= client
.ModelConfigFacade
.from_connection(
1476 result
= await config_facade
.ModelGet()
1477 config
= result
.config
1478 for key
, value
in config
.items():
1479 config
[key
] = ConfigValue
.from_json(value
)
1482 async def get_constraints(self
):
1483 """Return the machine constraints for this model.
1485 :returns: A ``dict`` of constraints.
1488 client_facade
= client
.ClientFacade
.from_connection(self
.connection())
1489 result
= await client_facade
.GetModelConstraints()
1491 # GetModelConstraints returns GetConstraintsResults which has a
1492 # 'constraints' attribute. If no constraints have been set
1493 # GetConstraintsResults.constraints is None. Otherwise
1494 # GetConstraintsResults.constraints has an attribute for each possible
1495 # constraint, each of these in turn will be None if they have not been
1497 if result
.constraints
:
1498 constraint_types
= [a
for a
in dir(result
.constraints
)
1499 if a
in Value
._toSchema
.keys()]
1500 for constraint
in constraint_types
:
1501 value
= getattr(result
.constraints
, constraint
)
1502 if value
is not None:
1503 constraints
[constraint
] = getattr(result
.constraints
,
1507 def import_ssh_key(self
, identity
):
1508 """Add a public SSH key from a trusted indentity source to this model.
1510 :param str identity: User identity in the form <lp|gh>:<username>
1513 raise NotImplementedError()
1514 import_ssh_keys
= import_ssh_key
1516 async def get_machines(self
):
1517 """Return list of machines in this model.
1520 return list(self
.state
.machines
.keys())
1522 def get_shares(self
):
1523 """Return list of all users with access to this model.
1526 raise NotImplementedError()
1528 def get_spaces(self
):
1529 """Return list of all known spaces, including associated subnets.
1532 raise NotImplementedError()
1534 async def get_ssh_key(self
, raw_ssh
=False):
1535 """Return known SSH keys for this model.
1536 :param bool raw_ssh: if True, returns the raw ssh key,
1537 else it's fingerprint
1540 key_facade
= client
.KeyManagerFacade
.from_connection(self
.connection())
1541 entity
= {'tag': tag
.model(self
.info
.uuid
)}
1542 entities
= client
.Entities([entity
])
1543 return await key_facade
.ListKeys(entities
, raw_ssh
)
1544 get_ssh_keys
= get_ssh_key
1546 def get_storage(self
, filesystem
=False, volume
=False):
1547 """Return details of storage instances.
1549 :param bool filesystem: Include filesystem storage
1550 :param bool volume: Include volume storage
1553 raise NotImplementedError()
1555 def get_storage_pools(self
, names
=None, providers
=None):
1556 """Return list of storage pools.
1558 :param list names: Only include pools with these names
1559 :param list providers: Only include pools for these providers
1562 raise NotImplementedError()
1564 def get_subnets(self
, space
=None, zone
=None):
1565 """Return list of known subnets.
1567 :param str space: Only include subnets in this space
1568 :param str zone: Only include subnets in this zone
1571 raise NotImplementedError()
1573 def remove_blocks(self
):
1574 """Remove all blocks from this model.
1577 raise NotImplementedError()
1579 def remove_backup(self
, backup_id
):
1582 :param str backup_id: The id of the backup to remove
1585 raise NotImplementedError()
1587 def remove_cached_images(self
, arch
=None, kind
=None, series
=None):
1588 """Remove cached OS images.
1590 :param str arch: Architecture of the images to remove
1591 :param str kind: Image kind to remove, e.g. 'lxd'
1592 :param str series: Image series to remove, e.g. 'xenial'
1595 raise NotImplementedError()
1597 def remove_machine(self
, *machine_ids
):
1598 """Remove a machine from this model.
1600 :param str *machine_ids: Ids of the machines to remove
1603 raise NotImplementedError()
1604 remove_machines
= remove_machine
1606 async def remove_ssh_key(self
, user
, key
):
1607 """Remove a public SSH key(s) from this model.
1609 :param str key: Full ssh key
1610 :param str user: Juju user to which the key is registered
1613 key_facade
= client
.KeyManagerFacade
.from_connection(self
.connection())
1614 key
= base64
.b64decode(bytes(key
.strip().split()[1].encode('ascii')))
1615 key
= hashlib
.md5(key
).hexdigest()
1616 key
= ':'.join(a
+ b
for a
, b
in zip(key
[::2], key
[1::2]))
1617 await key_facade
.DeleteKeys([key
], user
)
1618 remove_ssh_keys
= remove_ssh_key
1621 self
, bootstrap
=False, constraints
=None, archive
=None,
1622 backup_id
=None, upload_tools
=False):
1623 """Restore a backup archive to a new controller.
1625 :param bool bootstrap: Bootstrap a new state machine
1626 :param constraints: Model constraints
1627 :type constraints: :class:`juju.Constraints`
1628 :param str archive: Path to backup archive to restore
1629 :param str backup_id: Id of backup to restore
1630 :param bool upload_tools: Upload tools if bootstrapping a new machine
1633 raise NotImplementedError()
1635 def retry_provisioning(self
):
1636 """Retry provisioning for failed machines.
1639 raise NotImplementedError()
1641 def run(self
, command
, timeout
=None):
1642 """Run command on all machines in this model.
1644 :param str command: The command to run
1645 :param int timeout: Time to wait before command is considered failed
1648 raise NotImplementedError()
1650 async def set_config(self
, config
):
1651 """Set configuration keys on this model.
1653 :param dict config: Mapping of config keys to either string values or
1654 `ConfigValue` instances, as returned by `get_config`.
1656 config_facade
= client
.ModelConfigFacade
.from_connection(
1659 for key
, value
in config
.items():
1660 if isinstance(value
, ConfigValue
):
1661 config
[key
] = value
.value
1662 await config_facade
.ModelSet(config
)
1664 async def set_constraints(self
, constraints
):
1665 """Set machine constraints on this model.
1667 :param dict config: Mapping of model constraints
1669 client_facade
= client
.ClientFacade
.from_connection(self
.connection())
1670 await client_facade
.SetModelConstraints(
1672 constraints
=constraints
)
1674 async def get_action_output(self
, action_uuid
, wait
=None):
1675 """Get the results of an action by ID.
1677 :param str action_uuid: Id of the action
1678 :param int wait: Time in seconds to wait for action to complete.
1679 :return dict: Output from action
1680 :raises: :class:`JujuError` if invalid action_uuid
1682 action_facade
= client
.ActionFacade
.from_connection(
1685 entity
= [{'tag': tag
.action(action_uuid
)}]
1686 # Cannot use self.wait_for_action as the action event has probably
1687 # already happened and self.wait_for_action works by processing
1688 # model deltas and checking if they match our type. If the action
1689 # has already occured then the delta has gone.
1691 async def _wait_for_action_status():
1693 action_output
= await action_facade
.Actions(entity
)
1694 if action_output
.results
[0].status
in ('completed', 'failed'):
1697 await asyncio
.sleep(1)
1698 await asyncio
.wait_for(
1699 _wait_for_action_status(),
1701 action_output
= await action_facade
.Actions(entity
)
1702 # ActionResult.output is None if the action produced no output
1703 if action_output
.results
[0].output
is None:
1706 output
= action_output
.results
[0].output
1709 async def get_action_status(self
, uuid_or_prefix
=None, name
=None):
1710 """Get the status of all actions, filtered by ID, ID prefix, or name.
1712 :param str uuid_or_prefix: Filter by action uuid or prefix
1713 :param str name: Filter by action name
1718 action_facade
= client
.ActionFacade
.from_connection(
1722 name_results
= await action_facade
.FindActionsByNames([name
])
1723 action_results
.extend(name_results
.actions
[0].actions
)
1725 # Collect list of actions matching uuid or prefix
1726 matching_actions
= await action_facade
.FindActionTagsByPrefix(
1729 for actions
in matching_actions
.matches
.values():
1730 entities
= [{'tag': a
.tag
} for a
in actions
]
1731 # Get action results matching action tags
1732 uuid_results
= await action_facade
.Actions(entities
)
1733 action_results
.extend(uuid_results
.results
)
1734 for a
in action_results
:
1735 results
[tag
.untag('action-', a
.action
.tag
)] = a
.status
1738 def get_budget(self
, budget_name
):
1739 """Get budget usage info.
1741 :param str budget_name: Name of budget
1744 raise NotImplementedError()
1746 async def get_status(self
, filters
=None, utc
=False):
1747 """Return the status of the model.
1749 :param str filters: Optional list of applications, units, or machines
1750 to include, which can use wildcards ('*').
1751 :param bool utc: Display time as UTC in RFC3339 format
1754 client_facade
= client
.ClientFacade
.from_connection(self
.connection())
1755 return await client_facade
.FullStatus(filters
)
1758 self
, all_
=False, destination
=None, dry_run
=False, public
=False,
1759 source
=None, stream
=None, version
=None):
1760 """Copy Juju tools into this model.
1762 :param bool all_: Copy all versions, not just the latest
1763 :param str destination: Path to local destination directory
1764 :param bool dry_run: Don't do the actual copy
1765 :param bool public: Tools are for a public cloud, so generate mirrors
1767 :param str source: Path to local source directory
1768 :param str stream: Simplestreams stream for which to sync metadata
1769 :param str version: Copy a specific major.minor version
1772 raise NotImplementedError()
1774 def unblock(self
, *commands
):
1775 """Unblock an operation that would alter this model.
1777 :param str *commands: The commands to unblock. Valid values are
1778 'all-changes', 'destroy-model', 'remove-object'
1781 raise NotImplementedError()
1783 def unset_config(self
, *keys
):
1784 """Unset configuration on this model.
1786 :param str *keys: The keys to unset
1789 raise NotImplementedError()
1791 def upgrade_gui(self
):
1792 """Upgrade the Juju GUI for this model.
1795 raise NotImplementedError()
1798 self
, dry_run
=False, reset_previous_upgrade
=False,
1799 upload_tools
=False, version
=None):
1800 """Upgrade Juju on all machines in a model.
1802 :param bool dry_run: Don't do the actual upgrade
1803 :param bool reset_previous_upgrade: Clear the previous (incomplete)
1805 :param bool upload_tools: Upload local version of tools
1806 :param str version: Upgrade to a specific version
1809 raise NotImplementedError()
1811 def upload_backup(self
, archive_path
):
1812 """Store a backup archive remotely in Juju.
1814 :param str archive_path: Path to local archive
1817 raise NotImplementedError()
1820 def charmstore(self
):
1821 return self
._charmstore
1823 async def get_metrics(self
, *tags
):
1824 """Retrieve metrics.
1826 :param str *tags: Tags of entities from which to retrieve metrics.
1827 No tags retrieves the metrics of all units in the model.
1828 :return: Dictionary of unit_name:metrics
1831 log
.debug("Retrieving metrics for %s",
1832 ', '.join(tags
) if tags
else "all units")
1834 metrics_facade
= client
.MetricsDebugFacade
.from_connection(
1837 entities
= [client
.Entity(tag
) for tag
in tags
]
1838 metrics_result
= await metrics_facade
.GetMetrics(entities
)
1840 metrics
= collections
.defaultdict(list)
1842 for entity_metrics
in metrics_result
.results
:
1843 error
= entity_metrics
.error
1845 if "is not a valid tag" in error
:
1846 raise ValueError(error
.message
)
1848 raise Exception(error
.message
)
1850 for metric
in entity_metrics
.metrics
:
1851 metrics
[metric
.unit
].append(vars(metric
))
1856 def get_charm_series(path
):
1857 """Inspects the charm directory at ``path`` and returns a default
1858 series from its metadata.yaml (the first item in the 'series' list).
1860 Returns None if no series can be determined.
1863 md
= Path(path
) / "metadata.yaml"
1866 data
= yaml
.load(md
.open())
1867 series
= data
.get('series')
1868 return series
[0] if series
else None
1871 class BundleHandler
:
1873 Handle bundles by using the API to translate bundle YAML into a plan of
1874 steps and then dispatching each of those using the API.
1876 def __init__(self
, model
):
1878 self
.charmstore
= model
.charmstore
1880 self
.references
= {}
1881 self
._units
_by
_app
= {}
1882 for unit_name
, unit
in model
.units
.items():
1883 app_units
= self
._units
_by
_app
.setdefault(unit
.application
, [])
1884 app_units
.append(unit_name
)
1885 self
.bundle_facade
= client
.BundleFacade
.from_connection(
1887 self
.client_facade
= client
.ClientFacade
.from_connection(
1889 self
.app_facade
= client
.ApplicationFacade
.from_connection(
1891 self
.ann_facade
= client
.AnnotationsFacade
.from_connection(
1894 async def _handle_local_charms(self
, bundle
):
1895 """Search for references to local charms (i.e. filesystem paths)
1896 in the bundle. Upload the local charms to the model, and replace
1897 the filesystem paths with appropriate 'local:' paths in the bundle.
1899 Return the modified bundle.
1901 :param dict bundle: Bundle dictionary
1902 :return: Modified bundle dictionary
1907 default_series
= bundle
.get('series')
1908 apps_dict
= bundle
.get('applications', bundle
.get('services', {}))
1909 for app_name
in self
.applications
:
1910 app_dict
= apps_dict
[app_name
]
1911 charm_dir
= os
.path
.abspath(os
.path
.expanduser(app_dict
['charm']))
1912 if not os
.path
.isdir(charm_dir
):
1915 app_dict
.get('series') or
1917 get_charm_series(charm_dir
)
1921 "Couldn't determine series for charm at {}. "
1922 "Add a 'series' key to the bundle.".format(charm_dir
))
1924 # Keep track of what we need to update. We keep a list of apps
1925 # that need to be updated, and a corresponding list of args
1926 # needed to update those apps.
1927 apps
.append(app_name
)
1928 args
.append((charm_dir
, series
))
1931 # If we have apps to update, spawn all the coroutines concurrently
1932 # and wait for them to finish.
1933 charm_urls
= await asyncio
.gather(*[
1934 self
.model
.add_local_charm_dir(*params
)
1936 ], loop
=self
.model
.loop
)
1937 # Update the 'charm:' entry for each app with the new 'local:' url.
1938 for app_name
, charm_url
in zip(apps
, charm_urls
):
1939 apps_dict
[app_name
]['charm'] = charm_url
1943 async def fetch_plan(self
, entity_id
):
1944 is_store_url
= entity_id
.startswith('cs:')
1946 if not is_store_url
and os
.path
.isfile(entity_id
):
1947 bundle_yaml
= Path(entity_id
).read_text()
1948 elif not is_store_url
and os
.path
.isdir(entity_id
):
1949 bundle_yaml
= (Path(entity_id
) / "bundle.yaml").read_text()
1951 bundle_yaml
= await self
.charmstore
.files(entity_id
,
1952 filename
='bundle.yaml',
1954 self
.bundle
= yaml
.safe_load(bundle_yaml
)
1955 self
.bundle
= await self
._handle
_local
_charms
(self
.bundle
)
1957 self
.plan
= await self
.bundle_facade
.GetChanges(
1958 yaml
.dump(self
.bundle
))
1960 if self
.plan
.errors
:
1961 raise JujuError(self
.plan
.errors
)
1963 async def execute_plan(self
):
1964 for step
in self
.plan
.changes
:
1965 method
= getattr(self
, step
.method
)
1966 result
= await method(*step
.args
)
1967 self
.references
[step
.id_
] = result
1970 def applications(self
):
1971 apps_dict
= self
.bundle
.get('applications',
1972 self
.bundle
.get('services', {}))
1973 return list(apps_dict
.keys())
1975 def resolve(self
, reference
):
1976 if reference
and reference
.startswith('$'):
1977 reference
= self
.references
[reference
[1:]]
1980 async def addCharm(self
, charm
, series
):
1982 :param charm string:
1983 Charm holds the URL of the charm to be added.
1985 :param series string:
1986 Series holds the series of the charm to be added
1987 if the charm default is not sufficient.
1989 # We don't add local charms because they've already been added
1990 # by self._handle_local_charms
1991 if charm
.startswith('local:'):
1994 entity_id
= await self
.charmstore
.entityId(charm
)
1995 log
.debug('Adding %s', entity_id
)
1996 await self
.client_facade
.AddCharm(None, entity_id
)
1999 async def addMachines(self
, params
=None):
2002 Dictionary specifying the machine to add. All keys are optional.
2005 series: string specifying the machine OS series.
2007 constraints: string holding machine constraints, if any. We'll
2008 parse this into the json friendly dict that the juju api
2011 container_type: string holding the type of the container (for
2012 instance ""lxd" or kvm"). It is not specified for top level
2015 parent_id: string holding a placeholder pointing to another
2016 machine change or to a unit change. This value is only
2017 specified in the case this machine is a container, in
2018 which case also ContainerType is set.
2021 params
= params
or {}
2024 params
= {normalize_key(k
): params
[k
] for k
in params
.keys()}
2026 # Fix up values, as necessary.
2027 if 'parent_id' in params
:
2028 if params
['parent_id'].startswith('$addUnit'):
2029 unit
= self
.resolve(params
['parent_id'])[0]
2030 params
['parent_id'] = unit
.machine
.entity_id
2032 params
['parent_id'] = self
.resolve(params
['parent_id'])
2034 params
['constraints'] = parse_constraints(
2035 params
.get('constraints'))
2036 params
['jobs'] = params
.get('jobs', ['JobHostUnits'])
2038 if params
.get('container_type') == 'lxc':
2039 log
.warning('Juju 2.0 does not support lxc containers. '
2040 'Converting containers to lxd.')
2041 params
['container_type'] = 'lxd'
2043 # Submit the request.
2044 params
= client
.AddMachineParams(**params
)
2045 results
= await self
.client_facade
.AddMachines([params
])
2046 error
= results
.machines
[0].error
2048 raise ValueError("Error adding machine: %s" % error
.message
)
2049 machine
= results
.machines
[0].machine
2050 log
.debug('Added new machine %s', machine
)
2053 async def addRelation(self
, endpoint1
, endpoint2
):
2055 :param endpoint1 string:
2056 :param endpoint2 string:
2057 Endpoint1 and Endpoint2 hold relation endpoints in the
2058 "application:interface" form, where the application is always a
2059 placeholder pointing to an application change, and the interface is
2060 optional. Examples are "$deploy-42:web" or just "$deploy-42".
2062 endpoints
= [endpoint1
, endpoint2
]
2063 # resolve indirect references
2064 for i
in range(len(endpoints
)):
2065 parts
= endpoints
[i
].split(':')
2066 parts
[0] = self
.resolve(parts
[0])
2067 endpoints
[i
] = ':'.join(parts
)
2069 log
.info('Relating %s <-> %s', *endpoints
)
2070 return await self
.model
.add_relation(*endpoints
)
2072 async def deploy(self
, charm
, series
, application
, options
, constraints
,
2073 storage
, endpoint_bindings
, *args
):
2075 :param charm string:
2076 Charm holds the URL of the charm to be used to deploy this
2079 :param series string:
2080 Series holds the series of the application to be deployed
2081 if the charm default is not sufficient.
2083 :param application string:
2084 Application holds the application name.
2086 :param options map[string]interface{}:
2087 Options holds application options.
2089 :param constraints string:
2090 Constraints holds the optional application constraints.
2092 :param storage map[string]string:
2093 Storage holds the optional storage constraints.
2095 :param endpoint_bindings map[string]string:
2096 EndpointBindings holds the optional endpoint bindings
2098 :param devices map[string]string:
2099 Devices holds the optional devices constraints.
2100 (Only given on Juju 2.5+)
2102 :param resources map[string]int:
2103 Resources identifies the revision to use for each resource
2104 of the application's charm.
2106 :param num_units int:
2107 NumUnits holds the number of units required. For IAAS models, this
2108 will be 0 and separate AddUnitChanges will be used. For Kubernetes
2109 models, this will be used to scale the application.
2110 (Only given on Juju 2.5+)
2112 # resolve indirect references
2113 charm
= self
.resolve(charm
)
2116 # Juju 2.4 and below only sends the resources
2118 devices
, num_units
= None, None
2120 # Juju 2.5+ sends devices before resources, as well as num_units
2121 # There might be placement but we need to ignore that.
2122 devices
, resources
, num_units
= args
[:3]
2124 if not charm
.startswith('local:'):
2125 resources
= await self
.model
._add
_store
_resources
(
2126 application
, charm
, overrides
=resources
)
2127 await self
.model
._deploy
(
2129 application
=application
,
2132 constraints
=constraints
,
2133 endpoint_bindings
=endpoint_bindings
,
2134 resources
=resources
,
2137 num_units
=num_units
,
2141 async def addUnit(self
, application
, to
):
2143 :param application string:
2144 Application holds the application placeholder name for which a unit
2148 To holds the optional location where to add the unit, as a
2149 placeholder pointing to another unit change or to a machine change.
2151 application
= self
.resolve(application
)
2152 placement
= self
.resolve(to
)
2153 if self
._units
_by
_app
.get(application
):
2154 # enough units for this application already exist;
2155 # claim one, and carry on
2156 # NB: this should probably honor placement, but the juju client
2157 # doesn't, so we're not bothering, either
2158 unit_name
= self
._units
_by
_app
[application
].pop()
2159 log
.debug('Reusing unit %s for %s', unit_name
, application
)
2160 return self
.model
.units
[unit_name
]
2162 log
.debug('Adding new unit for %s%s', application
,
2163 ' to %s' % placement
if placement
else '')
2164 return await self
.model
.applications
[application
].add_unit(
2169 async def scale(self
, application
, scale
):
2171 Handle a change of scale to a k8s application.
2173 :param string application:
2174 Application holds the application placeholder name for which a unit
2178 New scale value to use.
2180 application
= self
.resolve(application
)
2181 return await self
.model
.applications
[application
].scale(scale
=scale
)
2183 async def expose(self
, application
):
2185 :param application string:
2186 Application holds the placeholder name of the application that must
2189 application
= self
.resolve(application
)
2190 log
.info('Exposing %s', application
)
2191 return await self
.model
.applications
[application
].expose()
2193 async def setAnnotations(self
, id_
, entity_type
, annotations
):
2196 Id is the placeholder for the application or machine change
2197 corresponding to the entity to be annotated.
2199 :param entity_type EntityType:
2200 EntityType holds the type of the entity, "application" or
2203 :param annotations map[string]string:
2204 Annotations holds the annotations as key/value pairs.
2206 entity_id
= self
.resolve(id_
)
2208 entity
= self
.model
.state
.get_entity(entity_type
, entity_id
)
2210 entity
= await self
.model
._wait
_for
_new
(entity_type
, entity_id
)
2211 return await entity
.set_annotations(annotations
)
2216 Async wrapper around theblues.charmstore.CharmStore
2218 def __init__(self
, loop
, cs_timeout
=20):
2220 self
._cs
= theblues
.charmstore
.CharmStore(timeout
=cs_timeout
)
2222 def __getattr__(self
, name
):
2224 Wrap method calls in coroutines that use run_in_executor to make them
2227 attr
= getattr(self
._cs
, name
)
2228 if not callable(attr
):
2229 wrapper
= partial(getattr, self
._cs
, name
)
2230 setattr(self
, name
, wrapper
)
2232 async def coro(*args
, **kwargs
):
2233 method
= partial(attr
, *args
, **kwargs
)
2234 for attempt
in range(1, 4):
2236 return await self
.loop
.run_in_executor(None, method
)
2237 except theblues
.errors
.ServerError
:
2240 await asyncio
.sleep(1, loop
=self
.loop
)
2241 setattr(self
, name
, coro
)
2246 class CharmArchiveGenerator
:
2248 Create a Zip archive of a local charm directory for upload to a controller.
2250 This is used automatically by
2251 `Model.add_local_charm_dir <#juju.model.Model.add_local_charm_dir>`_.
2253 def __init__(self
, path
):
2254 self
.path
= os
.path
.abspath(os
.path
.expanduser(path
))
2256 def make_archive(self
, path
):
2257 """Create archive of directory and write to ``path``.
2259 :param path: Path to archive
2263 * build/* - This is used for packing the charm itself and any
2265 * */.* - Hidden files are all ignored for now. This will most
2266 likely be changed into a specific ignore list
2270 zf
= zipfile
.ZipFile(path
, 'w', zipfile
.ZIP_DEFLATED
)
2271 for dirpath
, dirnames
, filenames
in os
.walk(self
.path
):
2272 relative_path
= dirpath
[len(self
.path
) + 1:]
2273 if relative_path
and not self
._ignore
(relative_path
):
2274 zf
.write(dirpath
, relative_path
)
2275 for name
in filenames
:
2276 archive_name
= os
.path
.join(relative_path
, name
)
2277 if not self
._ignore
(archive_name
):
2278 real_path
= os
.path
.join(dirpath
, name
)
2279 self
._check
_type
(real_path
)
2280 if os
.path
.islink(real_path
):
2281 self
._check
_link
(real_path
)
2282 self
._write
_symlink
(
2283 zf
, os
.readlink(real_path
), archive_name
)
2285 zf
.write(real_path
, archive_name
)
2289 def _check_type(self
, path
):
2293 if stat
.S_ISDIR(s
.st_mode
) or stat
.S_ISREG(s
.st_mode
):
2295 raise ValueError("Invalid Charm at % %s" % (
2296 path
, "Invalid file type for a charm"))
2298 def _check_link(self
, path
):
2299 link_path
= os
.readlink(path
)
2300 if link_path
[0] == "/":
2302 "Invalid Charm at %s: %s" % (
2303 path
, "Absolute links are invalid"))
2304 path_dir
= os
.path
.dirname(path
)
2305 link_path
= os
.path
.join(path_dir
, link_path
)
2306 if not link_path
.startswith(os
.path
.abspath(self
.path
)):
2308 "Invalid charm at %s %s" % (
2309 path
, "Only internal symlinks are allowed"))
2311 def _write_symlink(self
, zf
, link_target
, link_path
):
2312 """Package symlinks with appropriate zipfile metadata."""
2313 info
= zipfile
.ZipInfo()
2314 info
.filename
= link_path
2315 info
.create_system
= 3
2316 # Magic code for symlinks / py2/3 compat
2317 # 27166663808 = (stat.S_IFLNK | 0755) << 16
2318 info
.external_attr
= 2716663808
2319 zf
.writestr(info
, link_target
)
2321 def _ignore(self
, path
):
2322 if path
== "build" or path
.startswith("build/"):
2324 if path
.startswith('.'):