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
:
849 delta
= get_entity_delta(delta
)
850 old_obj
, new_obj
= self
.state
.apply_delta(delta
)
851 await self
._notify
_observers
(delta
, old_obj
, new_obj
)
852 except KeyError as e
:
853 log
.debug("unknown delta type: %s", e
.args
[0])
854 self
._watch
_received
.set()
855 except CancelledError
:
858 log
.exception('Error in watcher')
861 self
._watch
_stopped
.set()
863 log
.debug('Starting watcher task')
864 self
._watch
_received
.clear()
865 self
._watch
_stopping
.clear()
866 self
._watch
_stopped
.clear()
867 self
._connector
.loop
.create_task(_all_watcher())
869 async def _notify_observers(self
, delta
, old_obj
, new_obj
):
870 """Call observing callbacks, notifying them of a change in model state
872 :param delta: The raw change from the watcher
873 (:class:`juju.client.overrides.Delta`)
874 :param old_obj: The object in the model that this delta updates.
876 :param new_obj: The object in the model that is created or updated
877 by applying this delta.
880 if new_obj
and not old_obj
:
884 'Model changed: %s %s %s',
885 delta
.entity
, delta
.type, delta
.get_id())
887 for o
in self
._observers
:
888 if o
.cares_about(delta
):
889 asyncio
.ensure_future(o(delta
, old_obj
, new_obj
, self
),
890 loop
=self
._connector
.loop
)
892 async def _wait(self
, entity_type
, entity_id
, action
, predicate
=None):
894 Block the calling routine until a given action has happened to the
897 :param entity_type: The entity's type.
898 :param entity_id: The entity's id.
899 :param action: the type of action (e.g., 'add', 'change', or 'remove')
900 :param predicate: optional callable that must take as an
901 argument a delta, and must return a boolean, indicating
902 whether the delta contains the specific action we're looking
903 for. For example, you might check to see whether a 'change'
904 has a 'completed' status. See the _Observer class for details.
907 q
= asyncio
.Queue(loop
=self
._connector
.loop
)
909 async def callback(delta
, old
, new
, model
):
910 await q
.put(delta
.get_id())
912 self
.add_observer(callback
, entity_type
, action
, entity_id
, predicate
)
913 entity_id
= await q
.get()
914 # object might not be in the entity_map if we were waiting for a
916 return self
.state
._live
_entity
_map
(entity_type
).get(entity_id
)
918 async def _wait_for_new(self
, entity_type
, entity_id
):
919 """Wait for a new object to appear in the Model and return it.
921 Waits for an object of type ``entity_type`` with id ``entity_id``
922 to appear in the model. This is similar to watching for the
923 object using ``block_until``, but uses the watcher rather than
927 # if the entity is already in the model, just return it
928 if entity_id
in self
.state
._live
_entity
_map
(entity_type
):
929 return self
.state
._live
_entity
_map
(entity_type
)[entity_id
]
930 return await self
._wait
(entity_type
, entity_id
, None)
932 async def wait_for_action(self
, action_id
):
933 """Given an action, wait for it to complete."""
935 if action_id
.startswith("action-"):
936 # if we've been passed action.tag, transform it into the
937 # id that the api deltas will use.
938 action_id
= action_id
[7:]
940 def predicate(delta
):
941 return delta
.data
['status'] in ('completed', 'failed')
943 return await self
._wait
('action', action_id
, None, predicate
)
945 async def add_machine(
946 self
, spec
=None, constraints
=None, disks
=None, series
=None):
947 """Start a new, empty machine and optionally a container, or add a
948 container to a machine.
950 :param str spec: Machine specification
953 (None) - starts a new machine
954 'lxd' - starts a new machine with one lxd container
955 'lxd:4' - starts a new lxd container on machine 4
956 'ssh:user@10.10.0.3:/path/to/private/key' - manually provision
957 a machine with ssh and the private key used for authentication
958 'zone=us-east-1a' - starts a machine in zone us-east-1s on AWS
959 'maas2.name' - acquire machine maas2.name on MAAS
961 :param dict constraints: Machine constraints, which can contain the
982 :param list disks: List of disk constraint dictionaries, which can
983 contain the following keys::
997 :param str series: Series, e.g. 'xenial'
999 Supported container types are: lxd, kvm
1001 When deploying a container to an existing machine, constraints cannot
1005 params
= client
.AddMachineParams()
1008 if spec
.startswith("ssh:"):
1009 placement
, target
, private_key_path
= spec
.split(":")
1010 user
, host
= target
.split("@")
1012 sshProvisioner
= provisioner
.SSHProvisioner(
1015 private_key_path
=private_key_path
,
1018 params
= sshProvisioner
.provision_machine()
1020 placement
= parse_placement(spec
)
1022 params
.placement
= placement
[0]
1024 params
.jobs
= ['JobHostUnits']
1027 params
.constraints
= client
.Value
.from_json(constraints
)
1031 client
.Constraints
.from_json(o
) for o
in disks
]
1034 params
.series
= series
1036 # Submit the request.
1037 client_facade
= client
.ClientFacade
.from_connection(self
.connection())
1038 results
= await client_facade
.AddMachines([params
])
1039 error
= results
.machines
[0].error
1041 raise ValueError("Error adding machine: %s" % error
.message
)
1042 machine_id
= results
.machines
[0].machine
1045 if spec
.startswith("ssh:"):
1046 # Need to run this after AddMachines has been called,
1047 # as we need the machine_id
1048 await sshProvisioner
.install_agent(
1054 log
.debug('Added new machine %s', machine_id
)
1055 return await self
._wait
_for
_new
('machine', machine_id
)
1057 async def add_relation(self
, relation1
, relation2
):
1058 """Add a relation between two applications.
1060 :param str relation1: '<application>[:<relation_name>]'
1061 :param str relation2: '<application>[:<relation_name>]'
1064 connection
= self
.connection()
1065 app_facade
= client
.ApplicationFacade
.from_connection(connection
)
1068 'Adding relation %s <-> %s', relation1
, relation2
)
1070 def _find_relation(*specs
):
1071 for rel
in self
.relations
:
1072 if rel
.matches(*specs
):
1077 result
= await app_facade
.AddRelation([relation1
, relation2
])
1078 except JujuAPIError
as e
:
1079 if 'relation already exists' not in e
.message
:
1081 rel
= _find_relation(relation1
, relation2
)
1084 raise JujuError('Relation {} {} exists but not in model'.format(
1085 relation1
, relation2
))
1087 specs
= ['{}:{}'.format(app
, data
['name'])
1088 for app
, data
in result
.endpoints
.items()]
1090 await self
.block_until(lambda: _find_relation(*specs
) is not None)
1091 return _find_relation(*specs
)
1093 def add_space(self
, name
, *cidrs
):
1094 """Add a new network space.
1096 Adds a new space with the given name and associates the given
1097 (optional) list of existing subnet CIDRs with it.
1099 :param str name: Name of the space
1100 :param *cidrs: Optional list of existing subnet CIDRs
1103 raise NotImplementedError()
1105 async def add_ssh_key(self
, user
, key
):
1106 """Add a public SSH key to this model.
1108 :param str user: The username of the user
1109 :param str key: The public ssh key
1112 key_facade
= client
.KeyManagerFacade
.from_connection(self
.connection())
1113 return await key_facade
.AddKeys([key
], user
)
1114 add_ssh_keys
= add_ssh_key
1116 def add_subnet(self
, cidr_or_id
, space
, *zones
):
1117 """Add an existing subnet to this model.
1119 :param str cidr_or_id: CIDR or provider ID of the existing subnet
1120 :param str space: Network space with which to associate
1121 :param str *zones: Zone(s) in which the subnet resides
1124 raise NotImplementedError()
1126 def get_backups(self
):
1127 """Retrieve metadata for backups in this model.
1130 raise NotImplementedError()
1132 def block(self
, *commands
):
1133 """Add a new block to this model.
1135 :param str *commands: The commands to block. Valid values are
1136 'all-changes', 'destroy-model', 'remove-object'
1139 raise NotImplementedError()
1141 def get_blocks(self
):
1142 """List blocks for this model.
1145 raise NotImplementedError()
1147 def get_cached_images(self
, arch
=None, kind
=None, series
=None):
1148 """Return a list of cached OS images.
1150 :param str arch: Filter by image architecture
1151 :param str kind: Filter by image kind, e.g. 'lxd'
1152 :param str series: Filter by image series, e.g. 'xenial'
1155 raise NotImplementedError()
1157 def create_backup(self
, note
=None, no_download
=False):
1158 """Create a backup of this model.
1160 :param str note: A note to store with the backup
1161 :param bool no_download: Do not download the backup archive
1162 :return str: Path to downloaded archive
1165 raise NotImplementedError()
1167 def create_storage_pool(self
, name
, provider_type
, **pool_config
):
1168 """Create or define a storage pool.
1170 :param str name: Name to give the storage pool
1171 :param str provider_type: Pool provider type
1172 :param **pool_config: key/value pool configuration pairs
1175 raise NotImplementedError()
1178 self
, no_tail
=False, exclude_module
=None, include_module
=None,
1179 include
=None, level
=None, limit
=0, lines
=10, replay
=False,
1181 """Get log messages for this model.
1183 :param bool no_tail: Stop after returning existing log messages
1184 :param list exclude_module: Do not show log messages for these logging
1186 :param list include_module: Only show log messages for these logging
1188 :param list include: Only show log messages for these entities
1189 :param str level: Log level to show, valid options are 'TRACE',
1190 'DEBUG', 'INFO', 'WARNING', 'ERROR,
1191 :param int limit: Return this many of the most recent (possibly
1192 filtered) lines are shown
1193 :param int lines: Yield this many of the most recent lines, and keep
1195 :param bool replay: Yield the entire log, and keep yielding
1196 :param list exclude: Do not show log messages for these entities
1199 raise NotImplementedError()
1201 def _get_series(self
, entity_url
, entity
):
1202 # try to get the series from the provided charm URL
1203 if entity_url
.startswith('cs:'):
1204 parts
= entity_url
[3:].split('/')
1206 parts
= entity_url
.split('/')
1207 if parts
[0].startswith('~'):
1210 # series was specified in the URL
1212 # series was not supplied at all, so use the newest
1213 # supported series according to the charm store
1214 ss
= entity
['Meta']['supported-series']
1215 return ss
['SupportedSeries'][0]
1218 self
, entity_url
, application_name
=None, bind
=None, budget
=None,
1219 channel
=None, config
=None, constraints
=None, force
=False,
1220 num_units
=1, plan
=None, resources
=None, series
=None, storage
=None,
1221 to
=None, devices
=None):
1222 """Deploy a new service or bundle.
1224 :param str entity_url: Charm or bundle url
1225 :param str application_name: Name to give the service
1226 :param dict bind: <charm endpoint>:<network space> pairs
1227 :param dict budget: <budget name>:<limit> pairs
1228 :param str channel: Charm store channel from which to retrieve
1229 the charm or bundle, e.g. 'edge'
1230 :param dict config: Charm configuration dictionary
1231 :param constraints: Service constraints
1232 :type constraints: :class:`juju.Constraints`
1233 :param bool force: Allow charm to be deployed to a machine running
1234 an unsupported series
1235 :param int num_units: Number of units to deploy
1236 :param str plan: Plan under which to deploy charm
1237 :param dict resources: <resource name>:<file path> pairs
1238 :param str series: Series on which to deploy
1239 :param dict storage: Storage constraints TODO how do these look?
1240 :param to: Placement directive as a string. For example:
1242 '23' - place on machine 23
1243 'lxd:7' - place in new lxd container on machine 7
1244 '24/lxd/3' - place in container 3 on machine 24
1246 If None, a new machine is provisioned.
1251 - support local resources
1256 k
: client
.Constraints(**v
)
1257 for k
, v
in storage
.items()
1260 entity_path
= Path(entity_url
.replace('local:', ''))
1261 bundle_path
= entity_path
/ 'bundle.yaml'
1262 metadata_path
= entity_path
/ 'metadata.yaml'
1265 entity_url
.startswith('local:') or
1266 entity_path
.is_dir() or
1267 entity_path
.is_file()
1270 entity_id
= entity_url
.replace('local:', '')
1272 entity
= await self
.charmstore
.entity(entity_url
, channel
=channel
,
1273 include_stats
=False)
1274 entity_id
= entity
['Id']
1276 client_facade
= client
.ClientFacade
.from_connection(self
.connection())
1278 is_bundle
= ((is_local
and
1279 (entity_id
.endswith('.yaml') and entity_path
.exists()) or
1280 bundle_path
.exists()) or
1281 (not is_local
and 'bundle/' in entity_id
))
1284 handler
= BundleHandler(self
)
1285 await handler
.fetch_plan(entity_id
)
1286 await handler
.execute_plan()
1287 extant_apps
= {app
for app
in self
.applications
}
1288 pending_apps
= set(handler
.applications
) - extant_apps
1290 # new apps will usually be in the model by now, but if some
1291 # haven't made it yet we'll need to wait on them to be added
1292 await asyncio
.gather(*[
1293 asyncio
.ensure_future(
1294 self
._wait
_for
_new
('application', app_name
),
1295 loop
=self
._connector
.loop
)
1296 for app_name
in pending_apps
1297 ], loop
=self
._connector
.loop
)
1298 return [app
for name
, app
in self
.applications
.items()
1299 if name
in handler
.applications
]
1302 if not application_name
:
1303 application_name
= entity
['Meta']['charm-metadata']['Name']
1305 series
= self
._get
_series
(entity_url
, entity
)
1306 await client_facade
.AddCharm(channel
, entity_id
)
1307 # XXX: we're dropping local resources here, but we don't
1308 # actually support them yet anyway
1309 resources
= await self
._add
_store
_resources
(application_name
,
1313 if not application_name
:
1314 metadata
= yaml
.load(metadata_path
.read_text())
1315 application_name
= metadata
['name']
1316 # We have a local charm dir that needs to be uploaded
1317 charm_dir
= os
.path
.abspath(
1318 os
.path
.expanduser(entity_id
))
1319 series
= series
or get_charm_series(charm_dir
)
1322 "Couldn't determine series for charm at {}. "
1323 "Pass a 'series' kwarg to Model.deploy().".format(
1325 entity_id
= await self
.add_local_charm_dir(charm_dir
, series
)
1326 return await self
._deploy
(
1327 charm_url
=entity_id
,
1328 application
=application_name
,
1330 config
=config
or {},
1331 constraints
=constraints
,
1332 endpoint_bindings
=bind
,
1333 resources
=resources
,
1336 num_units
=num_units
,
1337 placement
=parse_placement(to
),
1341 async def _add_store_resources(self
, application
, entity_url
,
1342 overrides
=None, entity
=None):
1344 # avoid extra charm store call if one was already made
1345 entity
= await self
.charmstore
.entity(entity_url
,
1346 include_stats
=False)
1349 'description': resource
['Description'],
1350 'fingerprint': resource
['Fingerprint'],
1351 'name': resource
['Name'],
1352 'path': resource
['Path'],
1353 'revision': resource
['Revision'],
1354 'size': resource
['Size'],
1355 'type_': resource
['Type'],
1357 } for resource
in entity
['Meta']['resources']
1361 names
= {r
['name'] for r
in resources
}
1362 unknown
= overrides
.keys() - names
1364 raise JujuError('Unrecognized resource{}: {}'.format(
1365 's' if len(unknown
) > 1 else '',
1366 ', '.join(unknown
)))
1367 for resource
in resources
:
1368 if resource
['name'] in overrides
:
1369 resource
['revision'] = overrides
[resource
['name']]
1374 resources_facade
= client
.ResourcesFacade
.from_connection(
1376 response
= await resources_facade
.AddPendingResources(
1377 tag
.application(application
),
1379 [client
.CharmResource(**resource
) for resource
in resources
])
1380 resource_map
= {resource
['name']: pid
1382 in zip(resources
, response
.pending_ids
)}
1385 async def _deploy(self
, charm_url
, application
, series
, config
,
1386 constraints
, endpoint_bindings
, resources
, storage
,
1387 channel
=None, num_units
=None, placement
=None,
1389 """Logic shared between `Model.deploy` and `BundleHandler.deploy`.
1391 log
.info('Deploying %s', charm_url
)
1393 # stringify all config values for API, and convert to YAML
1394 config
= {k
: str(v
) for k
, v
in config
.items()}
1395 config
= yaml
.dump({application
: config
},
1396 default_flow_style
=False)
1398 app_facade
= client
.ApplicationFacade
.from_connection(
1401 app
= client
.ApplicationDeploy(
1402 charm_url
=charm_url
,
1403 application
=application
,
1407 constraints
=parse_constraints(constraints
),
1408 endpoint_bindings
=endpoint_bindings
,
1409 num_units
=num_units
,
1410 resources
=resources
,
1412 placement
=placement
,
1415 result
= await app_facade
.Deploy([app
])
1416 errors
= [r
.error
.message
for r
in result
.results
if r
.error
]
1418 raise JujuError('\n'.join(errors
))
1419 return await self
._wait
_for
_new
('application', application
)
1421 async def destroy(self
):
1422 """Terminate all machines and resources for this model.
1423 Is already implemented in controller.py.
1425 raise NotImplementedError()
1427 async def destroy_unit(self
, *unit_names
):
1428 """Destroy units by name.
1431 connection
= self
.connection()
1432 app_facade
= client
.ApplicationFacade
.from_connection(connection
)
1435 'Destroying unit%s %s',
1436 's' if len(unit_names
) == 1 else '',
1437 ' '.join(unit_names
))
1439 return await app_facade
.DestroyUnits(list(unit_names
))
1440 destroy_units
= destroy_unit
1442 def get_backup(self
, archive_id
):
1443 """Download a backup archive file.
1445 :param str archive_id: The id of the archive to download
1446 :return str: Path to the archive file
1449 raise NotImplementedError()
1452 self
, num_controllers
=0, constraints
=None, series
=None, to
=None):
1453 """Ensure sufficient controllers exist to provide redundancy.
1455 :param int num_controllers: Number of controllers to make available
1456 :param constraints: Constraints to apply to the controller machines
1457 :type constraints: :class:`juju.Constraints`
1458 :param str series: Series of the controller machines
1459 :param list to: Placement directives for controller machines, e.g.::
1462 'lxc:7' - new lxc container on machine 7
1463 '24/lxc/3' - lxc container 3 or machine 24
1465 If None, a new machine is provisioned.
1468 raise NotImplementedError()
1470 async def get_config(self
):
1471 """Return the configuration settings for this model.
1473 :returns: A ``dict`` mapping keys to `ConfigValue` instances,
1474 which have `source` and `value` attributes.
1476 config_facade
= client
.ModelConfigFacade
.from_connection(
1479 result
= await config_facade
.ModelGet()
1480 config
= result
.config
1481 for key
, value
in config
.items():
1482 config
[key
] = ConfigValue
.from_json(value
)
1485 async def get_constraints(self
):
1486 """Return the machine constraints for this model.
1488 :returns: A ``dict`` of constraints.
1491 client_facade
= client
.ClientFacade
.from_connection(self
.connection())
1492 result
= await client_facade
.GetModelConstraints()
1494 # GetModelConstraints returns GetConstraintsResults which has a
1495 # 'constraints' attribute. If no constraints have been set
1496 # GetConstraintsResults.constraints is None. Otherwise
1497 # GetConstraintsResults.constraints has an attribute for each possible
1498 # constraint, each of these in turn will be None if they have not been
1500 if result
.constraints
:
1501 constraint_types
= [a
for a
in dir(result
.constraints
)
1502 if a
in Value
._toSchema
.keys()]
1503 for constraint
in constraint_types
:
1504 value
= getattr(result
.constraints
, constraint
)
1505 if value
is not None:
1506 constraints
[constraint
] = getattr(result
.constraints
,
1510 def import_ssh_key(self
, identity
):
1511 """Add a public SSH key from a trusted indentity source to this model.
1513 :param str identity: User identity in the form <lp|gh>:<username>
1516 raise NotImplementedError()
1517 import_ssh_keys
= import_ssh_key
1519 async def get_machines(self
):
1520 """Return list of machines in this model.
1523 return list(self
.state
.machines
.keys())
1525 def get_shares(self
):
1526 """Return list of all users with access to this model.
1529 raise NotImplementedError()
1531 def get_spaces(self
):
1532 """Return list of all known spaces, including associated subnets.
1535 raise NotImplementedError()
1537 async def get_ssh_key(self
, raw_ssh
=False):
1538 """Return known SSH keys for this model.
1539 :param bool raw_ssh: if True, returns the raw ssh key,
1540 else it's fingerprint
1543 key_facade
= client
.KeyManagerFacade
.from_connection(self
.connection())
1544 entity
= {'tag': tag
.model(self
.info
.uuid
)}
1545 entities
= client
.Entities([entity
])
1546 return await key_facade
.ListKeys(entities
, raw_ssh
)
1547 get_ssh_keys
= get_ssh_key
1549 def get_storage(self
, filesystem
=False, volume
=False):
1550 """Return details of storage instances.
1552 :param bool filesystem: Include filesystem storage
1553 :param bool volume: Include volume storage
1556 raise NotImplementedError()
1558 def get_storage_pools(self
, names
=None, providers
=None):
1559 """Return list of storage pools.
1561 :param list names: Only include pools with these names
1562 :param list providers: Only include pools for these providers
1565 raise NotImplementedError()
1567 def get_subnets(self
, space
=None, zone
=None):
1568 """Return list of known subnets.
1570 :param str space: Only include subnets in this space
1571 :param str zone: Only include subnets in this zone
1574 raise NotImplementedError()
1576 def remove_blocks(self
):
1577 """Remove all blocks from this model.
1580 raise NotImplementedError()
1582 def remove_backup(self
, backup_id
):
1585 :param str backup_id: The id of the backup to remove
1588 raise NotImplementedError()
1590 def remove_cached_images(self
, arch
=None, kind
=None, series
=None):
1591 """Remove cached OS images.
1593 :param str arch: Architecture of the images to remove
1594 :param str kind: Image kind to remove, e.g. 'lxd'
1595 :param str series: Image series to remove, e.g. 'xenial'
1598 raise NotImplementedError()
1600 def remove_machine(self
, *machine_ids
):
1601 """Remove a machine from this model.
1603 :param str *machine_ids: Ids of the machines to remove
1606 raise NotImplementedError()
1607 remove_machines
= remove_machine
1609 async def remove_ssh_key(self
, user
, key
):
1610 """Remove a public SSH key(s) from this model.
1612 :param str key: Full ssh key
1613 :param str user: Juju user to which the key is registered
1616 key_facade
= client
.KeyManagerFacade
.from_connection(self
.connection())
1617 key
= base64
.b64decode(bytes(key
.strip().split()[1].encode('ascii')))
1618 key
= hashlib
.md5(key
).hexdigest()
1619 key
= ':'.join(a
+ b
for a
, b
in zip(key
[::2], key
[1::2]))
1620 await key_facade
.DeleteKeys([key
], user
)
1621 remove_ssh_keys
= remove_ssh_key
1624 self
, bootstrap
=False, constraints
=None, archive
=None,
1625 backup_id
=None, upload_tools
=False):
1626 """Restore a backup archive to a new controller.
1628 :param bool bootstrap: Bootstrap a new state machine
1629 :param constraints: Model constraints
1630 :type constraints: :class:`juju.Constraints`
1631 :param str archive: Path to backup archive to restore
1632 :param str backup_id: Id of backup to restore
1633 :param bool upload_tools: Upload tools if bootstrapping a new machine
1636 raise NotImplementedError()
1638 def retry_provisioning(self
):
1639 """Retry provisioning for failed machines.
1642 raise NotImplementedError()
1644 def run(self
, command
, timeout
=None):
1645 """Run command on all machines in this model.
1647 :param str command: The command to run
1648 :param int timeout: Time to wait before command is considered failed
1651 raise NotImplementedError()
1653 async def set_config(self
, config
):
1654 """Set configuration keys on this model.
1656 :param dict config: Mapping of config keys to either string values or
1657 `ConfigValue` instances, as returned by `get_config`.
1659 config_facade
= client
.ModelConfigFacade
.from_connection(
1662 for key
, value
in config
.items():
1663 if isinstance(value
, ConfigValue
):
1664 config
[key
] = value
.value
1665 await config_facade
.ModelSet(config
)
1667 async def set_constraints(self
, constraints
):
1668 """Set machine constraints on this model.
1670 :param dict config: Mapping of model constraints
1672 client_facade
= client
.ClientFacade
.from_connection(self
.connection())
1673 await client_facade
.SetModelConstraints(
1675 constraints
=constraints
)
1677 async def get_action_output(self
, action_uuid
, wait
=None):
1678 """Get the results of an action by ID.
1680 :param str action_uuid: Id of the action
1681 :param int wait: Time in seconds to wait for action to complete.
1682 :return dict: Output from action
1683 :raises: :class:`JujuError` if invalid action_uuid
1685 action_facade
= client
.ActionFacade
.from_connection(
1688 entity
= [{'tag': tag
.action(action_uuid
)}]
1689 # Cannot use self.wait_for_action as the action event has probably
1690 # already happened and self.wait_for_action works by processing
1691 # model deltas and checking if they match our type. If the action
1692 # has already occured then the delta has gone.
1694 async def _wait_for_action_status():
1696 action_output
= await action_facade
.Actions(entity
)
1697 if action_output
.results
[0].status
in ('completed', 'failed'):
1700 await asyncio
.sleep(1)
1701 await asyncio
.wait_for(
1702 _wait_for_action_status(),
1704 action_output
= await action_facade
.Actions(entity
)
1705 # ActionResult.output is None if the action produced no output
1706 if action_output
.results
[0].output
is None:
1709 output
= action_output
.results
[0].output
1712 async def get_action_status(self
, uuid_or_prefix
=None, name
=None):
1713 """Get the status of all actions, filtered by ID, ID prefix, or name.
1715 :param str uuid_or_prefix: Filter by action uuid or prefix
1716 :param str name: Filter by action name
1721 action_facade
= client
.ActionFacade
.from_connection(
1725 name_results
= await action_facade
.FindActionsByNames([name
])
1726 action_results
.extend(name_results
.actions
[0].actions
)
1728 # Collect list of actions matching uuid or prefix
1729 matching_actions
= await action_facade
.FindActionTagsByPrefix(
1732 for actions
in matching_actions
.matches
.values():
1733 entities
= [{'tag': a
.tag
} for a
in actions
]
1734 # Get action results matching action tags
1735 uuid_results
= await action_facade
.Actions(entities
)
1736 action_results
.extend(uuid_results
.results
)
1737 for a
in action_results
:
1738 results
[tag
.untag('action-', a
.action
.tag
)] = a
.status
1741 def get_budget(self
, budget_name
):
1742 """Get budget usage info.
1744 :param str budget_name: Name of budget
1747 raise NotImplementedError()
1749 async def get_status(self
, filters
=None, utc
=False):
1750 """Return the status of the model.
1752 :param str filters: Optional list of applications, units, or machines
1753 to include, which can use wildcards ('*').
1754 :param bool utc: Display time as UTC in RFC3339 format
1757 client_facade
= client
.ClientFacade
.from_connection(self
.connection())
1758 return await client_facade
.FullStatus(filters
)
1761 self
, all_
=False, destination
=None, dry_run
=False, public
=False,
1762 source
=None, stream
=None, version
=None):
1763 """Copy Juju tools into this model.
1765 :param bool all_: Copy all versions, not just the latest
1766 :param str destination: Path to local destination directory
1767 :param bool dry_run: Don't do the actual copy
1768 :param bool public: Tools are for a public cloud, so generate mirrors
1770 :param str source: Path to local source directory
1771 :param str stream: Simplestreams stream for which to sync metadata
1772 :param str version: Copy a specific major.minor version
1775 raise NotImplementedError()
1777 def unblock(self
, *commands
):
1778 """Unblock an operation that would alter this model.
1780 :param str *commands: The commands to unblock. Valid values are
1781 'all-changes', 'destroy-model', 'remove-object'
1784 raise NotImplementedError()
1786 def unset_config(self
, *keys
):
1787 """Unset configuration on this model.
1789 :param str *keys: The keys to unset
1792 raise NotImplementedError()
1794 def upgrade_gui(self
):
1795 """Upgrade the Juju GUI for this model.
1798 raise NotImplementedError()
1801 self
, dry_run
=False, reset_previous_upgrade
=False,
1802 upload_tools
=False, version
=None):
1803 """Upgrade Juju on all machines in a model.
1805 :param bool dry_run: Don't do the actual upgrade
1806 :param bool reset_previous_upgrade: Clear the previous (incomplete)
1808 :param bool upload_tools: Upload local version of tools
1809 :param str version: Upgrade to a specific version
1812 raise NotImplementedError()
1814 def upload_backup(self
, archive_path
):
1815 """Store a backup archive remotely in Juju.
1817 :param str archive_path: Path to local archive
1820 raise NotImplementedError()
1823 def charmstore(self
):
1824 return self
._charmstore
1826 async def get_metrics(self
, *tags
):
1827 """Retrieve metrics.
1829 :param str *tags: Tags of entities from which to retrieve metrics.
1830 No tags retrieves the metrics of all units in the model.
1831 :return: Dictionary of unit_name:metrics
1834 log
.debug("Retrieving metrics for %s",
1835 ', '.join(tags
) if tags
else "all units")
1837 metrics_facade
= client
.MetricsDebugFacade
.from_connection(
1840 entities
= [client
.Entity(tag
) for tag
in tags
]
1841 metrics_result
= await metrics_facade
.GetMetrics(entities
)
1843 metrics
= collections
.defaultdict(list)
1845 for entity_metrics
in metrics_result
.results
:
1846 error
= entity_metrics
.error
1848 if "is not a valid tag" in error
:
1849 raise ValueError(error
.message
)
1851 raise Exception(error
.message
)
1853 for metric
in entity_metrics
.metrics
:
1854 metrics
[metric
.unit
].append(vars(metric
))
1859 def get_charm_series(path
):
1860 """Inspects the charm directory at ``path`` and returns a default
1861 series from its metadata.yaml (the first item in the 'series' list).
1863 Returns None if no series can be determined.
1866 md
= Path(path
) / "metadata.yaml"
1869 data
= yaml
.load(md
.open())
1870 series
= data
.get('series')
1871 return series
[0] if series
else None
1874 class BundleHandler
:
1876 Handle bundles by using the API to translate bundle YAML into a plan of
1877 steps and then dispatching each of those using the API.
1879 def __init__(self
, model
):
1881 self
.charmstore
= model
.charmstore
1883 self
.references
= {}
1884 self
._units
_by
_app
= {}
1885 for unit_name
, unit
in model
.units
.items():
1886 app_units
= self
._units
_by
_app
.setdefault(unit
.application
, [])
1887 app_units
.append(unit_name
)
1888 self
.bundle_facade
= client
.BundleFacade
.from_connection(
1890 self
.client_facade
= client
.ClientFacade
.from_connection(
1892 self
.app_facade
= client
.ApplicationFacade
.from_connection(
1894 self
.ann_facade
= client
.AnnotationsFacade
.from_connection(
1897 async def _handle_local_charms(self
, bundle
):
1898 """Search for references to local charms (i.e. filesystem paths)
1899 in the bundle. Upload the local charms to the model, and replace
1900 the filesystem paths with appropriate 'local:' paths in the bundle.
1902 Return the modified bundle.
1904 :param dict bundle: Bundle dictionary
1905 :return: Modified bundle dictionary
1910 default_series
= bundle
.get('series')
1911 apps_dict
= bundle
.get('applications', bundle
.get('services', {}))
1912 for app_name
in self
.applications
:
1913 app_dict
= apps_dict
[app_name
]
1914 charm_dir
= os
.path
.abspath(os
.path
.expanduser(app_dict
['charm']))
1915 if not os
.path
.isdir(charm_dir
):
1918 app_dict
.get('series') or
1920 get_charm_series(charm_dir
)
1924 "Couldn't determine series for charm at {}. "
1925 "Add a 'series' key to the bundle.".format(charm_dir
))
1927 # Keep track of what we need to update. We keep a list of apps
1928 # that need to be updated, and a corresponding list of args
1929 # needed to update those apps.
1930 apps
.append(app_name
)
1931 args
.append((charm_dir
, series
))
1934 # If we have apps to update, spawn all the coroutines concurrently
1935 # and wait for them to finish.
1936 charm_urls
= await asyncio
.gather(*[
1937 self
.model
.add_local_charm_dir(*params
)
1939 ], loop
=self
.model
.loop
)
1940 # Update the 'charm:' entry for each app with the new 'local:' url.
1941 for app_name
, charm_url
in zip(apps
, charm_urls
):
1942 apps_dict
[app_name
]['charm'] = charm_url
1946 async def fetch_plan(self
, entity_id
):
1947 is_store_url
= entity_id
.startswith('cs:')
1949 if not is_store_url
and os
.path
.isfile(entity_id
):
1950 bundle_yaml
= Path(entity_id
).read_text()
1951 elif not is_store_url
and os
.path
.isdir(entity_id
):
1952 bundle_yaml
= (Path(entity_id
) / "bundle.yaml").read_text()
1954 bundle_yaml
= await self
.charmstore
.files(entity_id
,
1955 filename
='bundle.yaml',
1957 self
.bundle
= yaml
.safe_load(bundle_yaml
)
1958 self
.bundle
= await self
._handle
_local
_charms
(self
.bundle
)
1960 self
.plan
= await self
.bundle_facade
.GetChanges(
1961 yaml
.dump(self
.bundle
))
1963 if self
.plan
.errors
:
1964 raise JujuError(self
.plan
.errors
)
1966 async def execute_plan(self
):
1967 for step
in self
.plan
.changes
:
1968 method
= getattr(self
, step
.method
)
1969 result
= await method(*step
.args
)
1970 self
.references
[step
.id_
] = result
1973 def applications(self
):
1974 apps_dict
= self
.bundle
.get('applications',
1975 self
.bundle
.get('services', {}))
1976 return list(apps_dict
.keys())
1978 def resolve(self
, reference
):
1979 if reference
and reference
.startswith('$'):
1980 reference
= self
.references
[reference
[1:]]
1983 async def addCharm(self
, charm
, series
):
1985 :param charm string:
1986 Charm holds the URL of the charm to be added.
1988 :param series string:
1989 Series holds the series of the charm to be added
1990 if the charm default is not sufficient.
1992 # We don't add local charms because they've already been added
1993 # by self._handle_local_charms
1994 if charm
.startswith('local:'):
1997 entity_id
= await self
.charmstore
.entityId(charm
)
1998 log
.debug('Adding %s', entity_id
)
1999 await self
.client_facade
.AddCharm(None, entity_id
)
2002 async def addMachines(self
, params
=None):
2005 Dictionary specifying the machine to add. All keys are optional.
2008 series: string specifying the machine OS series.
2010 constraints: string holding machine constraints, if any. We'll
2011 parse this into the json friendly dict that the juju api
2014 container_type: string holding the type of the container (for
2015 instance ""lxd" or kvm"). It is not specified for top level
2018 parent_id: string holding a placeholder pointing to another
2019 machine change or to a unit change. This value is only
2020 specified in the case this machine is a container, in
2021 which case also ContainerType is set.
2024 params
= params
or {}
2027 params
= {normalize_key(k
): params
[k
] for k
in params
.keys()}
2029 # Fix up values, as necessary.
2030 if 'parent_id' in params
:
2031 if params
['parent_id'].startswith('$addUnit'):
2032 unit
= self
.resolve(params
['parent_id'])[0]
2033 params
['parent_id'] = unit
.machine
.entity_id
2035 params
['parent_id'] = self
.resolve(params
['parent_id'])
2037 params
['constraints'] = parse_constraints(
2038 params
.get('constraints'))
2039 params
['jobs'] = params
.get('jobs', ['JobHostUnits'])
2041 if params
.get('container_type') == 'lxc':
2042 log
.warning('Juju 2.0 does not support lxc containers. '
2043 'Converting containers to lxd.')
2044 params
['container_type'] = 'lxd'
2046 # Submit the request.
2047 params
= client
.AddMachineParams(**params
)
2048 results
= await self
.client_facade
.AddMachines([params
])
2049 error
= results
.machines
[0].error
2051 raise ValueError("Error adding machine: %s" % error
.message
)
2052 machine
= results
.machines
[0].machine
2053 log
.debug('Added new machine %s', machine
)
2056 async def addRelation(self
, endpoint1
, endpoint2
):
2058 :param endpoint1 string:
2059 :param endpoint2 string:
2060 Endpoint1 and Endpoint2 hold relation endpoints in the
2061 "application:interface" form, where the application is always a
2062 placeholder pointing to an application change, and the interface is
2063 optional. Examples are "$deploy-42:web" or just "$deploy-42".
2065 endpoints
= [endpoint1
, endpoint2
]
2066 # resolve indirect references
2067 for i
in range(len(endpoints
)):
2068 parts
= endpoints
[i
].split(':')
2069 parts
[0] = self
.resolve(parts
[0])
2070 endpoints
[i
] = ':'.join(parts
)
2072 log
.info('Relating %s <-> %s', *endpoints
)
2073 return await self
.model
.add_relation(*endpoints
)
2075 async def deploy(self
, charm
, series
, application
, options
, constraints
,
2076 storage
, endpoint_bindings
, *args
):
2078 :param charm string:
2079 Charm holds the URL of the charm to be used to deploy this
2082 :param series string:
2083 Series holds the series of the application to be deployed
2084 if the charm default is not sufficient.
2086 :param application string:
2087 Application holds the application name.
2089 :param options map[string]interface{}:
2090 Options holds application options.
2092 :param constraints string:
2093 Constraints holds the optional application constraints.
2095 :param storage map[string]string:
2096 Storage holds the optional storage constraints.
2098 :param endpoint_bindings map[string]string:
2099 EndpointBindings holds the optional endpoint bindings
2101 :param devices map[string]string:
2102 Devices holds the optional devices constraints.
2103 (Only given on Juju 2.5+)
2105 :param resources map[string]int:
2106 Resources identifies the revision to use for each resource
2107 of the application's charm.
2109 :param num_units int:
2110 NumUnits holds the number of units required. For IAAS models, this
2111 will be 0 and separate AddUnitChanges will be used. For Kubernetes
2112 models, this will be used to scale the application.
2113 (Only given on Juju 2.5+)
2115 # resolve indirect references
2116 charm
= self
.resolve(charm
)
2119 # Juju 2.4 and below only sends the resources
2121 devices
, num_units
= None, None
2123 # Juju 2.5+ sends devices before resources, as well as num_units
2124 # There might be placement but we need to ignore that.
2125 devices
, resources
, num_units
= args
[:3]
2127 if not charm
.startswith('local:'):
2128 resources
= await self
.model
._add
_store
_resources
(
2129 application
, charm
, overrides
=resources
)
2130 await self
.model
._deploy
(
2132 application
=application
,
2135 constraints
=constraints
,
2136 endpoint_bindings
=endpoint_bindings
,
2137 resources
=resources
,
2140 num_units
=num_units
,
2144 async def addUnit(self
, application
, to
):
2146 :param application string:
2147 Application holds the application placeholder name for which a unit
2151 To holds the optional location where to add the unit, as a
2152 placeholder pointing to another unit change or to a machine change.
2154 application
= self
.resolve(application
)
2155 placement
= self
.resolve(to
)
2156 if self
._units
_by
_app
.get(application
):
2157 # enough units for this application already exist;
2158 # claim one, and carry on
2159 # NB: this should probably honor placement, but the juju client
2160 # doesn't, so we're not bothering, either
2161 unit_name
= self
._units
_by
_app
[application
].pop()
2162 log
.debug('Reusing unit %s for %s', unit_name
, application
)
2163 return self
.model
.units
[unit_name
]
2165 log
.debug('Adding new unit for %s%s', application
,
2166 ' to %s' % placement
if placement
else '')
2167 return await self
.model
.applications
[application
].add_unit(
2172 async def scale(self
, application
, scale
):
2174 Handle a change of scale to a k8s application.
2176 :param string application:
2177 Application holds the application placeholder name for which a unit
2181 New scale value to use.
2183 application
= self
.resolve(application
)
2184 return await self
.model
.applications
[application
].scale(scale
=scale
)
2186 async def expose(self
, application
):
2188 :param application string:
2189 Application holds the placeholder name of the application that must
2192 application
= self
.resolve(application
)
2193 log
.info('Exposing %s', application
)
2194 return await self
.model
.applications
[application
].expose()
2196 async def setAnnotations(self
, id_
, entity_type
, annotations
):
2199 Id is the placeholder for the application or machine change
2200 corresponding to the entity to be annotated.
2202 :param entity_type EntityType:
2203 EntityType holds the type of the entity, "application" or
2206 :param annotations map[string]string:
2207 Annotations holds the annotations as key/value pairs.
2209 entity_id
= self
.resolve(id_
)
2211 entity
= self
.model
.state
.get_entity(entity_type
, entity_id
)
2213 entity
= await self
.model
._wait
_for
_new
(entity_type
, entity_id
)
2214 return await entity
.set_annotations(annotations
)
2219 Async wrapper around theblues.charmstore.CharmStore
2221 def __init__(self
, loop
, cs_timeout
=20):
2223 self
._cs
= theblues
.charmstore
.CharmStore(timeout
=cs_timeout
)
2225 def __getattr__(self
, name
):
2227 Wrap method calls in coroutines that use run_in_executor to make them
2230 attr
= getattr(self
._cs
, name
)
2231 if not callable(attr
):
2232 wrapper
= partial(getattr, self
._cs
, name
)
2233 setattr(self
, name
, wrapper
)
2235 async def coro(*args
, **kwargs
):
2236 method
= partial(attr
, *args
, **kwargs
)
2237 for attempt
in range(1, 4):
2239 return await self
.loop
.run_in_executor(None, method
)
2240 except theblues
.errors
.ServerError
:
2243 await asyncio
.sleep(1, loop
=self
.loop
)
2244 setattr(self
, name
, coro
)
2249 class CharmArchiveGenerator
:
2251 Create a Zip archive of a local charm directory for upload to a controller.
2253 This is used automatically by
2254 `Model.add_local_charm_dir <#juju.model.Model.add_local_charm_dir>`_.
2256 def __init__(self
, path
):
2257 self
.path
= os
.path
.abspath(os
.path
.expanduser(path
))
2259 def make_archive(self
, path
):
2260 """Create archive of directory and write to ``path``.
2262 :param path: Path to archive
2266 * build/* - This is used for packing the charm itself and any
2268 * */.* - Hidden files are all ignored for now. This will most
2269 likely be changed into a specific ignore list
2273 zf
= zipfile
.ZipFile(path
, 'w', zipfile
.ZIP_DEFLATED
)
2274 for dirpath
, dirnames
, filenames
in os
.walk(self
.path
):
2275 relative_path
= dirpath
[len(self
.path
) + 1:]
2276 if relative_path
and not self
._ignore
(relative_path
):
2277 zf
.write(dirpath
, relative_path
)
2278 for name
in filenames
:
2279 archive_name
= os
.path
.join(relative_path
, name
)
2280 if not self
._ignore
(archive_name
):
2281 real_path
= os
.path
.join(dirpath
, name
)
2282 self
._check
_type
(real_path
)
2283 if os
.path
.islink(real_path
):
2284 self
._check
_link
(real_path
)
2285 self
._write
_symlink
(
2286 zf
, os
.readlink(real_path
), archive_name
)
2288 zf
.write(real_path
, archive_name
)
2292 def _check_type(self
, path
):
2296 if stat
.S_ISDIR(s
.st_mode
) or stat
.S_ISREG(s
.st_mode
):
2298 raise ValueError("Invalid Charm at % %s" % (
2299 path
, "Invalid file type for a charm"))
2301 def _check_link(self
, path
):
2302 link_path
= os
.readlink(path
)
2303 if link_path
[0] == "/":
2305 "Invalid Charm at %s: %s" % (
2306 path
, "Absolute links are invalid"))
2307 path_dir
= os
.path
.dirname(path
)
2308 link_path
= os
.path
.join(path_dir
, link_path
)
2309 if not link_path
.startswith(os
.path
.abspath(self
.path
)):
2311 "Invalid charm at %s %s" % (
2312 path
, "Only internal symlinks are allowed"))
2314 def _write_symlink(self
, zf
, link_target
, link_path
):
2315 """Package symlinks with appropriate zipfile metadata."""
2316 info
= zipfile
.ZipInfo()
2317 info
.filename
= link_path
2318 info
.create_system
= 3
2319 # Magic code for symlinks / py2/3 compat
2320 # 27166663808 = (stat.S_IFLNK | 0755) << 16
2321 info
.external_attr
= 2716663808
2322 zf
.writestr(info
, link_target
)
2324 def _ignore(self
, path
):
2325 if path
== "build" or path
.startswith("build/"):
2327 if path
.startswith('.'):