a0180a06b238b3c2e5bbe179c94610394aa14f05
7 from concurrent
.futures
import CancelledError
8 from functools
import partial
9 from pathlib
import Path
12 from theblues
import charmstore
14 from .client
import client
15 from .client
import watcher
16 from .client
import connection
17 from .constraints
import parse
as parse_constraints
, normalize_key
18 from .delta
import get_entity_delta
19 from .delta
import get_entity_class
20 from .exceptions
import DeadEntityException
21 from .errors
import JujuAPIError
23 log
= logging
.getLogger(__name__
)
26 class _Observer(object):
27 """Wrapper around an observer callable.
29 This wrapper allows filter criteria to be associated with the
30 callable so that it's only called for changes that meet the criteria.
33 def __init__(self
, callable_
, entity_type
, action
, entity_id
, predicate
):
34 self
.callable_
= callable_
35 self
.entity_type
= entity_type
37 self
.entity_id
= entity_id
38 self
.predicate
= predicate
40 self
.entity_id
= str(self
.entity_id
)
41 if not self
.entity_id
.startswith('^'):
42 self
.entity_id
= '^' + self
.entity_id
43 if not self
.entity_id
.endswith('$'):
46 async def __call__(self
, delta
, old
, new
, model
):
47 await self
.callable_(delta
, old
, new
, model
)
49 def cares_about(self
, delta
):
50 """Return True if this observer "cares about" (i.e. wants to be
51 called) for a this delta.
54 if (self
.entity_id
and delta
.get_id() and
55 not re
.match(self
.entity_id
, str(delta
.get_id()))):
58 if self
.entity_type
and self
.entity_type
!= delta
.entity
:
61 if self
.action
and self
.action
!= delta
.type:
64 if self
.predicate
and not self
.predicate(delta
):
70 class ModelObserver(object):
71 async def __call__(self
, delta
, old
, new
, model
):
72 handler_name
= 'on_{}_{}'.format(delta
.entity
, delta
.type)
73 method
= getattr(self
, handler_name
, self
.on_change
)
74 await method(delta
, old
, new
, model
)
76 async def on_change(self
, delta
, old
, new
, model
):
80 class ModelState(object):
81 """Holds the state of the model, including the delta history of all
82 entities in the model.
85 def __init__(self
, model
):
89 def _live_entity_map(self
, entity_type
):
90 """Return an id:Entity map of all the living entities of
95 entity_id
: self
.get_entity(entity_type
, entity_id
)
96 for entity_id
, history
in self
.state
.get(entity_type
, {}).items()
97 if history
[-1] is not None
101 def applications(self
):
102 """Return a map of application-name:Application for all applications
103 currently in the model.
106 return self
._live
_entity
_map
('application')
110 """Return a map of machine-id:Machine for all machines currently in
114 return self
._live
_entity
_map
('machine')
118 """Return a map of unit-id:Unit for all units currently in
122 return self
._live
_entity
_map
('unit')
124 def entity_history(self
, entity_type
, entity_id
):
125 """Return the history deque for an entity.
128 return self
.state
[entity_type
][entity_id
]
130 def entity_data(self
, entity_type
, entity_id
, history_index
):
131 """Return the data dict for an entity at a specific index of its
135 return self
.entity_history(entity_type
, entity_id
)[history_index
]
137 def apply_delta(self
, delta
):
138 """Apply delta to our state and return a copy of the
139 affected object as it was before and after the update, e.g.:
141 old_obj, new_obj = self.apply_delta(delta)
143 old_obj may be None if the delta is for the creation of a new object,
144 e.g. a new application or unit is deployed.
146 new_obj will never be None, but may be dead (new_obj.dead == True)
147 if the object was deleted as a result of the delta being applied.
152 .setdefault(delta
.entity
, {})
153 .setdefault(delta
.get_id(), collections
.deque())
156 history
.append(delta
.data
)
157 if delta
.type == 'remove':
160 entity
= self
.get_entity(delta
.entity
, delta
.get_id())
161 return entity
.previous(), entity
164 self
, entity_type
, entity_id
, history_index
=-1, connected
=True):
165 """Return an object instance for the given entity_type and id.
167 By default the object state matches the most recent state from
168 Juju. To get an instance of the object in an older state, pass
169 history_index, an index into the history deque for the entity.
173 if history_index
< 0 and history_index
!= -1:
174 history_index
+= len(self
.entity_history(entity_type
, entity_id
))
175 if history_index
< 0:
179 self
.entity_data(entity_type
, entity_id
, history_index
)
183 entity_class
= get_entity_class(entity_type
)
185 entity_id
, self
.model
, history_index
=history_index
,
189 class ModelEntity(object):
190 """An object in the Model tree"""
192 def __init__(self
, entity_id
, model
, history_index
=-1, connected
=True):
193 """Initialize a new entity
195 :param entity_id str: The unique id of the object in the model
196 :param model: The model instance in whose object tree this
198 :history_index int: The index of this object's state in the model's
199 history deque for this entity
200 :connected bool: Flag indicating whether this object gets live updates
204 self
.entity_id
= entity_id
206 self
._history
_index
= history_index
207 self
.connected
= connected
208 self
.connection
= model
.connection
211 return '<{} entity_id="{}">'.format(type(self
).__name
__,
214 def __getattr__(self
, name
):
215 """Fetch object attributes from the underlying data dict held in the
219 return self
.safe_data
[name
]
222 return bool(self
.data
)
224 def on_change(self
, callable_
):
225 """Add a change observer to this entity.
228 self
.model
.add_observer(
229 callable_
, self
.entity_type
, 'change', self
.entity_id
)
231 def on_remove(self
, callable_
):
232 """Add a remove observer to this entity.
235 self
.model
.add_observer(
236 callable_
, self
.entity_type
, 'remove', self
.entity_id
)
239 def entity_type(self
):
240 """A string identifying the entity type of this object, e.g.
241 'application' or 'unit', etc.
244 return self
.__class
__.__name
__.lower()
248 """Return True if this object represents the current state of the
249 entity in the underlying model.
251 This will be True except when the object represents an entity at a
252 non-latest state in history, e.g. if the object was obtained by calling
253 .previous() on another object.
256 return self
._history
_index
== -1
260 """Returns True if this entity no longer exists in the underlying
266 self
.model
.state
.entity_data(
267 self
.entity_type
, self
.entity_id
, -1) is None
272 """Returns True if this entity still exists in the underlying
280 """The data dictionary for this entity.
283 return self
.model
.state
.entity_data(
284 self
.entity_type
, self
.entity_id
, self
._history
_index
)
288 """The data dictionary for this entity.
290 If this `ModelEntity` points to the dead state, it will
291 raise `DeadEntityException`.
294 if self
.data
is None:
295 raise DeadEntityException(
296 "Entity {}:{} is dead - its attributes can no longer be "
297 "accessed. Use the .previous() method on this object to get "
298 "a copy of the object at its previous state.".format(
299 self
.entity_type
, self
.entity_id
))
303 """Return a copy of this object as was at its previous state in
306 Returns None if this object is new (and therefore has no history).
308 The returned object is always "disconnected", i.e. does not receive
312 return self
.model
.state
.get_entity(
313 self
.entity_type
, self
.entity_id
, self
._history
_index
- 1,
317 """Return a copy of this object at its next state in
320 Returns None if this object is already the latest.
322 The returned object is "disconnected", i.e. does not receive
323 live updates, unless it is current (latest).
326 if self
._history
_index
== -1:
329 new_index
= self
._history
_index
+ 1
331 new_index
== len(self
.model
.state
.entity_history(
332 self
.entity_type
, self
.entity_id
)) - 1
334 return self
.model
.state
.get_entity(
335 self
.entity_type
, self
.entity_id
, self
._history
_index
- 1,
339 """Return a copy of this object at its current state in the model.
341 Returns self if this object is already the latest.
343 The returned object is always "connected", i.e. receives
344 live updates from the model.
347 if self
._history
_index
== -1:
350 return self
.model
.state
.get_entity(self
.entity_type
, self
.entity_id
)
354 def __init__(self
, loop
=None):
355 """Instantiate a new connected Model.
357 :param loop: an asyncio event loop
360 self
.loop
= loop
or asyncio
.get_event_loop()
361 self
.connection
= None
362 self
.observers
= weakref
.WeakValueDictionary()
363 self
.state
= ModelState(self
)
365 self
._watcher
_task
= None
366 self
._watch
_shutdown
= asyncio
.Event(loop
=loop
)
367 self
._watch
_received
= asyncio
.Event(loop
=loop
)
368 self
._charmstore
= CharmStore(self
.loop
)
370 async def connect(self
, *args
, **kw
):
371 """Connect to an arbitrary Juju model.
373 args and kw are passed through to Connection.connect()
376 self
.connection
= await connection
.Connection
.connect(*args
, **kw
)
377 await self
._after
_connect
()
379 async def connect_current(self
):
380 """Connect to the current Juju model.
383 self
.connection
= await connection
.Connection
.connect_current()
384 await self
._after
_connect
()
386 async def connect_model(self
, model_name
):
387 """Connect to a specific Juju model by name.
389 :param model_name: Format [controller:][user/]model
392 self
.connection
= await connection
.Connection
.connect_model(model_name
)
393 await self
._after
_connect
()
395 async def _after_connect(self
):
396 """Run initialization steps after connecting to websocket.
400 await self
._watch
_received
.wait()
401 await self
.get_info()
403 async def disconnect(self
):
404 """Shut down the watcher task and close websockets.
407 self
._stop
_watching
()
408 if self
.connection
and self
.connection
.is_open
:
409 await self
._watch
_shutdown
.wait()
410 log
.debug('Closing model connection')
411 await self
.connection
.close()
412 self
.connection
= None
414 def all_units_idle(self
):
415 """Return True if all units are idle.
418 for unit
in self
.units
.values():
419 unit_status
= unit
.data
['agent-status']['current']
420 if unit_status
!= 'idle':
424 async def reset(self
, force
=False):
425 """Reset the model to a clean state.
427 :param bool force: Force-terminate machines.
429 This returns only after the model has reached a clean state. "Clean"
430 means no applications or machines exist in the model.
433 log
.debug('Resetting model')
434 for app
in self
.applications
.values():
436 for machine
in self
.machines
.values():
437 await machine
.destroy(force
=force
)
438 await self
.block_until(
439 lambda: len(self
.machines
) == 0
442 async def block_until(self
, *conditions
, timeout
=None, wait_period
=0.5):
443 """Return only after all conditions are true.
447 while not all(c() for c
in conditions
):
448 await asyncio
.sleep(wait_period
)
449 await asyncio
.wait_for(_block(), timeout
)
452 def applications(self
):
453 """Return a map of application-name:Application for all applications
454 currently in the model.
457 return self
.state
.applications
461 """Return a map of machine-id:Machine for all machines currently in
465 return self
.state
.machines
469 """Return a map of unit-id:Unit for all units currently in
473 return self
.state
.units
475 async def get_info(self
):
476 """Return a client.ModelInfo object for this Model.
478 Retrieves latest info for this Model from the api server. The
479 return value is cached on the Model.info attribute so that the
480 valued may be accessed again without another api call, if
483 This method is called automatically when the Model is connected,
484 resulting in Model.info being initialized without requiring an
485 explicit call to this method.
488 facade
= client
.ClientFacade()
489 facade
.connect(self
.connection
)
491 self
.info
= await facade
.ModelInfo()
492 log
.debug('Got ModelInfo: %s', vars(self
.info
))
497 self
, callable_
, entity_type
=None, action
=None, entity_id
=None,
499 """Register an "on-model-change" callback
501 Once the model is connected, ``callable_``
502 will be called each time the model changes. ``callable_`` should
503 be Awaitable and accept the following positional arguments:
505 delta - An instance of :class:`juju.delta.EntityDelta`
506 containing the raw delta data recv'd from the Juju
509 old_obj - If the delta modifies an existing object in the model,
510 old_obj will be a copy of that object, as it was before the
511 delta was applied. Will be None if the delta creates a new
514 new_obj - A copy of the new or updated object, after the delta
515 is applied. Will be None if the delta removes an entity
518 model - The :class:`Model` itself.
520 Events for which ``callable_`` is called can be specified by passing
521 entity_type, action, and/or entitiy_id filter criteria, e.g.::
525 entity_type='application', action='add', entity_id='ubuntu')
527 For more complex filtering conditions, pass a predicate function. It
528 will be called with a delta as its only argument. If the predicate
529 function returns True, the ``callable_`` will be called.
532 observer
= _Observer(
533 callable_
, entity_type
, action
, entity_id
, predicate
)
534 self
.observers
[observer
] = callable_
537 """Start an asynchronous watch against this model.
539 See :meth:`add_observer` to register an onchange callback.
542 async def _start_watch():
543 self
._watch
_shutdown
.clear()
545 allwatcher
= watcher
.AllWatcher()
546 self
._watch
_conn
= await self
.connection
.clone()
547 allwatcher
.connect(self
._watch
_conn
)
549 results
= await allwatcher
.Next()
550 for delta
in results
.deltas
:
551 delta
= get_entity_delta(delta
)
552 old_obj
, new_obj
= self
.state
.apply_delta(delta
)
553 # XXX: Might not want to shield at this level
554 # We are shielding because when the watcher is
555 # canceled (on disconnect()), we don't want all of
556 # its children (every observer callback) to be
557 # canceled with it. So we shield them. But this means
558 # they can *never* be canceled.
559 await asyncio
.shield(
560 self
._notify
_observers
(delta
, old_obj
, new_obj
))
561 self
._watch
_received
.set()
562 except CancelledError
:
563 log
.debug('Closing watcher connection')
564 await self
._watch
_conn
.close()
565 self
._watch
_shutdown
.set()
566 self
._watch
_conn
= None
568 log
.debug('Starting watcher task')
569 self
._watcher
_task
= self
.loop
.create_task(_start_watch())
571 def _stop_watching(self
):
572 """Stop the asynchronous watch against this model.
575 log
.debug('Stopping watcher task')
576 if self
._watcher
_task
:
577 self
._watcher
_task
.cancel()
579 async def _notify_observers(self
, delta
, old_obj
, new_obj
):
580 """Call observing callbacks, notifying them of a change in model state
582 :param delta: The raw change from the watcher
583 (:class:`juju.client.overrides.Delta`)
584 :param old_obj: The object in the model that this delta updates.
586 :param new_obj: The object in the model that is created or updated
587 by applying this delta.
590 if new_obj
and not old_obj
:
594 'Model changed: %s %s %s',
595 delta
.entity
, delta
.type, delta
.get_id())
597 for o
in self
.observers
:
598 if o
.cares_about(delta
):
599 asyncio
.ensure_future(o(delta
, old_obj
, new_obj
, self
))
601 async def _wait(self
, entity_type
, entity_id
, action
, predicate
=None):
603 Block the calling routine until a given action has happened to the
606 :param entity_type: The entity's type.
607 :param entity_id: The entity's id.
608 :param action: the type of action (e.g., 'add' or 'change')
609 :param predicate: optional callable that must take as an
610 argument a delta, and must return a boolean, indicating
611 whether the delta contains the specific action we're looking
612 for. For example, you might check to see whether a 'change'
613 has a 'completed' status. See the _Observer class for details.
616 q
= asyncio
.Queue(loop
=self
.loop
)
618 async def callback(delta
, old
, new
, model
):
619 await q
.put(delta
.get_id())
621 self
.add_observer(callback
, entity_type
, action
, entity_id
, predicate
)
622 entity_id
= await q
.get()
623 return self
.state
._live
_entity
_map
(entity_type
)[entity_id
]
625 async def _wait_for_new(self
, entity_type
, entity_id
=None, predicate
=None):
626 """Wait for a new object to appear in the Model and return it.
628 Waits for an object of type ``entity_type`` with id ``entity_id``.
629 If ``entity_id`` is ``None``, it will wait for the first new entity
632 This coroutine blocks until the new object appears in the model.
635 # if the entity is already in the model, just return it
636 if entity_id
in self
.state
._live
_entity
_map
(entity_type
):
637 return self
.state
._live
_entity
_map
(entity_type
)[entity_id
]
638 # if we know the entity_id, we can trigger on any action that puts
639 # the enitty into the model; otherwise, we have to watch for the
640 # next "add" action on that entity_type
641 action
= 'add' if entity_id
is None else None
642 return await self
._wait
(entity_type
, entity_id
, action
, predicate
)
644 async def wait_for_action(self
, action_id
):
645 """Given an action, wait for it to complete."""
647 if action_id
.startswith("action-"):
648 # if we've been passed action.tag, transform it into the
649 # id that the api deltas will use.
650 action_id
= action_id
[7:]
652 def predicate(delta
):
653 return delta
.data
['status'] in ('completed', 'failed')
655 return await self
._wait
('action', action_id
, 'change', predicate
)
658 self
, spec
=None, constraints
=None, disks
=None, series
=None,
660 """Start a new, empty machine and optionally a container, or add a
661 container to a machine.
663 :param str spec: Machine specification
666 (None) - starts a new machine
667 'lxc' - starts a new machine with on lxc container
668 'lxc:4' - starts a new lxc container on machine 4
669 'ssh:user@10.10.0.3' - manually provisions a machine with ssh
670 'zone=us-east-1a' - starts a machine in zone us-east-1s on AWS
671 'maas2.name' - acquire machine maas2.name on MAAS
672 :param constraints: Machine constraints
673 :type constraints: :class:`juju.Constraints`
674 :param list disks: List of disk :class:`constraints <juju.Constraints>`
675 :param str series: Series
676 :param int count: Number of machines to deploy
678 Supported container types are: lxc, lxd, kvm
680 When deploying a container to an existing machine, constraints cannot
685 add_machines
= add_machine
687 async def add_relation(self
, relation1
, relation2
):
688 """Add a relation between two applications.
690 :param str relation1: '<application>[:<relation_name>]'
691 :param str relation2: '<application>[:<relation_name>]'
694 app_facade
= client
.ApplicationFacade()
695 app_facade
.connect(self
.connection
)
698 'Adding relation %s <-> %s', relation1
, relation2
)
701 result
= await app_facade
.AddRelation([relation1
, relation2
])
702 except JujuAPIError
as e
:
703 if 'relation already exists' not in e
.message
:
706 'Relation %s <-> %s already exists', relation1
, relation2
)
707 # TODO: if relation already exists we should return the
708 # Relation ModelEntity here
711 def predicate(delta
):
713 for endpoint
in delta
.data
['endpoints']:
714 endpoints
[endpoint
['application-name']] = endpoint
['relation']
715 return endpoints
== result
.endpoints
717 return await self
._wait
_for
_new
('relation', None, predicate
)
719 def add_space(self
, name
, *cidrs
):
720 """Add a new network space.
722 Adds a new space with the given name and associates the given
723 (optional) list of existing subnet CIDRs with it.
725 :param str name: Name of the space
726 :param \*cidrs: Optional list of existing subnet CIDRs
731 def add_ssh_key(self
, key
):
732 """Add a public SSH key to this model.
734 :param str key: The public ssh key
738 add_ssh_keys
= add_ssh_key
740 def add_subnet(self
, cidr_or_id
, space
, *zones
):
741 """Add an existing subnet to this model.
743 :param str cidr_or_id: CIDR or provider ID of the existing subnet
744 :param str space: Network space with which to associate
745 :param str \*zones: Zone(s) in which the subnet resides
750 def get_backups(self
):
751 """Retrieve metadata for backups in this model.
756 def block(self
, *commands
):
757 """Add a new block to this model.
759 :param str \*commands: The commands to block. Valid values are
760 'all-changes', 'destroy-model', 'remove-object'
765 def get_blocks(self
):
766 """List blocks for this model.
771 def get_cached_images(self
, arch
=None, kind
=None, series
=None):
772 """Return a list of cached OS images.
774 :param str arch: Filter by image architecture
775 :param str kind: Filter by image kind, e.g. 'lxd'
776 :param str series: Filter by image series, e.g. 'xenial'
781 def create_backup(self
, note
=None, no_download
=False):
782 """Create a backup of this model.
784 :param str note: A note to store with the backup
785 :param bool no_download: Do not download the backup archive
786 :return str: Path to downloaded archive
791 def create_storage_pool(self
, name
, provider_type
, **pool_config
):
792 """Create or define a storage pool.
794 :param str name: Name to give the storage pool
795 :param str provider_type: Pool provider type
796 :param \*\*pool_config: key/value pool configuration pairs
802 self
, no_tail
=False, exclude_module
=None, include_module
=None,
803 include
=None, level
=None, limit
=0, lines
=10, replay
=False,
805 """Get log messages for this model.
807 :param bool no_tail: Stop after returning existing log messages
808 :param list exclude_module: Do not show log messages for these logging
810 :param list include_module: Only show log messages for these logging
812 :param list include: Only show log messages for these entities
813 :param str level: Log level to show, valid options are 'TRACE',
814 'DEBUG', 'INFO', 'WARNING', 'ERROR,
815 :param int limit: Return this many of the most recent (possibly
816 filtered) lines are shown
817 :param int lines: Yield this many of the most recent lines, and keep
819 :param bool replay: Yield the entire log, and keep yielding
820 :param list exclude: Do not show log messages for these entities
826 self
, entity_url
, application_name
=None, bind
=None, budget
=None,
827 channel
=None, config
=None, constraints
=None, force
=False,
828 num_units
=1, plan
=None, resources
=None, series
=None, storage
=None,
830 """Deploy a new service or bundle.
832 :param str entity_url: Charm or bundle url
833 :param str application_name: Name to give the service
834 :param dict bind: <charm endpoint>:<network space> pairs
835 :param dict budget: <budget name>:<limit> pairs
836 :param str channel: Charm store channel from which to retrieve
837 the charm or bundle, e.g. 'development'
838 :param dict config: Charm configuration dictionary
839 :param constraints: Service constraints
840 :type constraints: :class:`juju.Constraints`
841 :param bool force: Allow charm to be deployed to a machine running
842 an unsupported series
843 :param int num_units: Number of units to deploy
844 :param str plan: Plan under which to deploy charm
845 :param dict resources: <resource name>:<file path> pairs
846 :param str series: Series on which to deploy
847 :param dict storage: Storage constraints TODO how do these look?
848 :param str to: Placement directive, e.g.::
851 'lxc:7' - new lxc container on machine 7
852 '24/lxc/3' - lxc container 3 or machine 24
854 If None, a new machine is provisioned.
859 - application_name is required; fill this in automatically if not
861 - series is required; how do we pick a default?
866 client
.Placement(**p
) for p
in to
873 k
: client
.Constraints(**v
)
874 for k
, v
in storage
.items()
877 is_local
= not entity_url
.startswith('cs:') and \
878 os
.path
.isdir(entity_url
)
879 entity_id
= await self
.charmstore
.entityId(entity_url
) \
880 if not is_local
else entity_url
882 app_facade
= client
.ApplicationFacade()
883 client_facade
= client
.ClientFacade()
884 app_facade
.connect(self
.connection
)
885 client_facade
.connect(self
.connection
)
887 is_bundle
= ((is_local
and
888 (Path(entity_id
) / 'bundle.yaml').exists()) or
889 (not is_local
and 'bundle/' in entity_id
))
892 handler
= BundleHandler(self
)
893 await handler
.fetch_plan(entity_id
)
894 await handler
.execute_plan()
895 extant_apps
= {app
for app
in self
.applications
}
896 pending_apps
= set(handler
.applications
) - extant_apps
898 # new apps will usually be in the model by now, but if some
899 # haven't made it yet we'll need to wait on them to be added
900 await asyncio
.gather(*[
901 asyncio
.ensure_future(
902 self
._wait
_for
_new
('application', app_name
))
903 for app_name
in pending_apps
905 return [app
for name
, app
in self
.applications
.items()
906 if name
in handler
.applications
]
909 'Deploying %s', entity_id
)
911 await client_facade
.AddCharm(channel
, entity_id
)
912 app
= client
.ApplicationDeploy(
913 application
=application_name
,
917 constraints
=parse_constraints(constraints
),
918 endpoint_bindings
=bind
,
926 await app_facade
.Deploy([app
])
927 return await self
._wait
_for
_new
('application', application_name
)
930 """Terminate all machines and resources for this model.
935 async def destroy_unit(self
, *unit_names
):
936 """Destroy units by name.
939 app_facade
= client
.ApplicationFacade()
940 app_facade
.connect(self
.connection
)
943 'Destroying unit%s %s',
944 's' if len(unit_names
) == 1 else '',
945 ' '.join(unit_names
))
947 return await app_facade
.Destroy(self
.name
)
948 destroy_units
= destroy_unit
950 def get_backup(self
, archive_id
):
951 """Download a backup archive file.
953 :param str archive_id: The id of the archive to download
954 :return str: Path to the archive file
960 self
, num_controllers
=0, constraints
=None, series
=None, to
=None):
961 """Ensure sufficient controllers exist to provide redundancy.
963 :param int num_controllers: Number of controllers to make available
964 :param constraints: Constraints to apply to the controller machines
965 :type constraints: :class:`juju.Constraints`
966 :param str series: Series of the controller machines
967 :param list to: Placement directives for controller machines, e.g.::
970 'lxc:7' - new lxc container on machine 7
971 '24/lxc/3' - lxc container 3 or machine 24
973 If None, a new machine is provisioned.
978 def get_config(self
):
979 """Return the configuration settings for this model.
984 def get_constraints(self
):
985 """Return the machine constraints for this model.
990 def grant(self
, username
, acl
='read'):
991 """Grant a user access to this model.
993 :param str username: Username
994 :param str acl: Access control ('read' or 'write')
999 def import_ssh_key(self
, identity
):
1000 """Add a public SSH key from a trusted indentity source to this model.
1002 :param str identity: User identity in the form <lp|gh>:<username>
1006 import_ssh_keys
= import_ssh_key
1008 def get_machines(self
, machine
, utc
=False):
1009 """Return list of machines in this model.
1011 :param str machine: Machine id, e.g. '0'
1012 :param bool utc: Display time as UTC in RFC3339 format
1017 def get_shares(self
):
1018 """Return list of all users with access to this model.
1023 def get_spaces(self
):
1024 """Return list of all known spaces, including associated subnets.
1029 def get_ssh_key(self
):
1030 """Return known SSH keys for this model.
1034 get_ssh_keys
= get_ssh_key
1036 def get_storage(self
, filesystem
=False, volume
=False):
1037 """Return details of storage instances.
1039 :param bool filesystem: Include filesystem storage
1040 :param bool volume: Include volume storage
1045 def get_storage_pools(self
, names
=None, providers
=None):
1046 """Return list of storage pools.
1048 :param list names: Only include pools with these names
1049 :param list providers: Only include pools for these providers
1054 def get_subnets(self
, space
=None, zone
=None):
1055 """Return list of known subnets.
1057 :param str space: Only include subnets in this space
1058 :param str zone: Only include subnets in this zone
1063 def remove_blocks(self
):
1064 """Remove all blocks from this model.
1069 def remove_backup(self
, backup_id
):
1072 :param str backup_id: The id of the backup to remove
1077 def remove_cached_images(self
, arch
=None, kind
=None, series
=None):
1078 """Remove cached OS images.
1080 :param str arch: Architecture of the images to remove
1081 :param str kind: Image kind to remove, e.g. 'lxd'
1082 :param str series: Image series to remove, e.g. 'xenial'
1087 def remove_machine(self
, *machine_ids
):
1088 """Remove a machine from this model.
1090 :param str \*machine_ids: Ids of the machines to remove
1094 remove_machines
= remove_machine
1096 def remove_ssh_key(self
, *keys
):
1097 """Remove a public SSH key(s) from this model.
1099 :param str \*keys: Keys to remove
1103 remove_ssh_keys
= remove_ssh_key
1106 self
, bootstrap
=False, constraints
=None, archive
=None,
1107 backup_id
=None, upload_tools
=False):
1108 """Restore a backup archive to a new controller.
1110 :param bool bootstrap: Bootstrap a new state machine
1111 :param constraints: Model constraints
1112 :type constraints: :class:`juju.Constraints`
1113 :param str archive: Path to backup archive to restore
1114 :param str backup_id: Id of backup to restore
1115 :param bool upload_tools: Upload tools if bootstrapping a new machine
1120 def retry_provisioning(self
):
1121 """Retry provisioning for failed machines.
1126 def revoke(self
, username
, acl
='read'):
1127 """Revoke a user's access to this model.
1129 :param str username: Username to revoke
1130 :param str acl: Access control ('read' or 'write')
1135 def run(self
, command
, timeout
=None):
1136 """Run command on all machines in this model.
1138 :param str command: The command to run
1139 :param int timeout: Time to wait before command is considered failed
1144 def set_config(self
, **config
):
1145 """Set configuration keys on this model.
1147 :param \*\*config: Config key/values
1152 def set_constraints(self
, constraints
):
1153 """Set machine constraints on this model.
1155 :param :class:`juju.Constraints` constraints: Machine constraints
1160 def get_action_output(self
, action_uuid
, wait
=-1):
1161 """Get the results of an action by ID.
1163 :param str action_uuid: Id of the action
1164 :param int wait: Time in seconds to wait for action to complete
1169 def get_action_status(self
, uuid_or_prefix
=None, name
=None):
1170 """Get the status of all actions, filtered by ID, ID prefix, or action name.
1172 :param str uuid_or_prefix: Filter by action uuid or prefix
1173 :param str name: Filter by action name
1178 def get_budget(self
, budget_name
):
1179 """Get budget usage info.
1181 :param str budget_name: Name of budget
1186 def get_status(self
, filter_
=None, utc
=False):
1187 """Return the status of the model.
1189 :param str filter_: Service or unit name or wildcard ('*')
1190 :param bool utc: Display time as UTC in RFC3339 format
1197 self
, all_
=False, destination
=None, dry_run
=False, public
=False,
1198 source
=None, stream
=None, version
=None):
1199 """Copy Juju tools into this model.
1201 :param bool all_: Copy all versions, not just the latest
1202 :param str destination: Path to local destination directory
1203 :param bool dry_run: Don't do the actual copy
1204 :param bool public: Tools are for a public cloud, so generate mirrors
1206 :param str source: Path to local source directory
1207 :param str stream: Simplestreams stream for which to sync metadata
1208 :param str version: Copy a specific major.minor version
1213 def unblock(self
, *commands
):
1214 """Unblock an operation that would alter this model.
1216 :param str \*commands: The commands to unblock. Valid values are
1217 'all-changes', 'destroy-model', 'remove-object'
1222 def unset_config(self
, *keys
):
1223 """Unset configuration on this model.
1225 :param str \*keys: The keys to unset
1230 def upgrade_gui(self
):
1231 """Upgrade the Juju GUI for this model.
1237 self
, dry_run
=False, reset_previous_upgrade
=False,
1238 upload_tools
=False, version
=None):
1239 """Upgrade Juju on all machines in a model.
1241 :param bool dry_run: Don't do the actual upgrade
1242 :param bool reset_previous_upgrade: Clear the previous (incomplete)
1244 :param bool upload_tools: Upload local version of tools
1245 :param str version: Upgrade to a specific version
1250 def upload_backup(self
, archive_path
):
1251 """Store a backup archive remotely in Juju.
1253 :param str archive_path: Path to local archive
1259 def charmstore(self
):
1260 return self
._charmstore
1262 async def get_metrics(self
, *tags
):
1263 """Retrieve metrics.
1265 :param str \*tags: Tags of entities from which to retrieve metrics.
1266 No tags retrieves the metrics of all units in the model.
1267 :return: Dictionary of unit_name:metrics
1270 log
.debug("Retrieving metrics for %s",
1271 ', '.join(tags
) if tags
else "all units")
1273 metrics_facade
= client
.MetricsDebugFacade()
1274 metrics_facade
.connect(self
.connection
)
1276 entities
= [client
.Entity(tag
) for tag
in tags
]
1277 metrics_result
= await metrics_facade
.GetMetrics(entities
)
1279 metrics
= collections
.defaultdict(list)
1281 for entity_metrics
in metrics_result
.results
:
1282 error
= entity_metrics
.error
1284 if "is not a valid tag" in error
:
1285 raise ValueError(error
.message
)
1287 raise Exception(error
.message
)
1289 for metric
in entity_metrics
.metrics
:
1290 metrics
[metric
.unit
].append(vars(metric
))
1295 class BundleHandler(object):
1297 Handle bundles by using the API to translate bundle YAML into a plan of
1298 steps and then dispatching each of those using the API.
1300 def __init__(self
, model
):
1302 self
.charmstore
= model
.charmstore
1304 self
.references
= {}
1305 self
._units
_by
_app
= {}
1306 for unit_name
, unit
in model
.units
.items():
1307 app_units
= self
._units
_by
_app
.setdefault(unit
.application
, [])
1308 app_units
.append(unit_name
)
1309 self
.client_facade
= client
.ClientFacade()
1310 self
.client_facade
.connect(model
.connection
)
1311 self
.app_facade
= client
.ApplicationFacade()
1312 self
.app_facade
.connect(model
.connection
)
1313 self
.ann_facade
= client
.AnnotationsFacade()
1314 self
.ann_facade
.connect(model
.connection
)
1316 async def fetch_plan(self
, entity_id
):
1317 is_local
= not entity_id
.startswith('cs:') and os
.path
.isdir(entity_id
)
1319 bundle_yaml
= (Path(entity_id
) / "bundle.yaml").read_text()
1321 bundle_yaml
= await self
.charmstore
.files(entity_id
,
1322 filename
='bundle.yaml',
1324 self
.bundle
= yaml
.safe_load(bundle_yaml
)
1325 self
.plan
= await self
.client_facade
.GetBundleChanges(bundle_yaml
)
1327 async def execute_plan(self
):
1328 for step
in self
.plan
.changes
:
1329 method
= getattr(self
, step
.method
)
1330 result
= await method(*step
.args
)
1331 self
.references
[step
.id_
] = result
1334 def applications(self
):
1335 return list(self
.bundle
['services'].keys())
1337 def resolve(self
, reference
):
1338 if reference
and reference
.startswith('$'):
1339 reference
= self
.references
[reference
[1:]]
1342 async def addCharm(self
, charm
, series
):
1344 :param charm string:
1345 Charm holds the URL of the charm to be added.
1347 :param series string:
1348 Series holds the series of the charm to be added
1349 if the charm default is not sufficient.
1351 entity_id
= await self
.charmstore
.entityId(charm
)
1352 log
.debug('Adding %s', entity_id
)
1353 await self
.client_facade
.AddCharm(None, entity_id
)
1356 async def addMachines(self
, params
=None):
1359 Dictionary specifying the machine to add. All keys are optional.
1362 series: string specifying the machine OS series.
1364 constraints: string holding machine constraints, if any. We'll
1365 parse this into the json friendly dict that the juju api
1368 container_type: string holding the type of the container (for
1369 instance ""lxd" or kvm"). It is not specified for top level
1372 parent_id: string holding a placeholder pointing to another
1373 machine change or to a unit change. This value is only
1374 specified in the case this machine is a container, in
1375 which case also ContainerType is set.
1378 params
= params
or {}
1381 params
= {normalize_key(k
): params
[k
] for k
in params
.keys()}
1383 # Fix up values, as necessary.
1384 if 'parent_id' in params
:
1385 params
['parent_id'] = self
.resolve(params
['parent_id'])
1387 params
['constraints'] = parse_constraints(
1388 params
.get('constraints'))
1389 params
['jobs'] = params
.get('jobs', ['JobHostUnits'])
1391 if params
.get('container_type') == 'lxc':
1392 log
.warning('Juju 2.0 does not support lxc containers. '
1393 'Converting containers to lxd.')
1394 params
['container_type'] = 'lxd'
1396 # Submit the request.
1397 params
= client
.AddMachineParams(**params
)
1398 results
= await self
.client_facade
.AddMachines([params
])
1399 error
= results
.machines
[0].error
1401 raise ValueError("Error adding machine: %s", error
.message
)
1402 machine
= results
.machines
[0].machine
1403 log
.debug('Added new machine %s', machine
)
1406 async def addRelation(self
, endpoint1
, endpoint2
):
1408 :param endpoint1 string:
1409 :param endpoint2 string:
1410 Endpoint1 and Endpoint2 hold relation endpoints in the
1411 "application:interface" form, where the application is always a
1412 placeholder pointing to an application change, and the interface is
1413 optional. Examples are "$deploy-42:web" or just "$deploy-42".
1415 endpoints
= [endpoint1
, endpoint2
]
1416 # resolve indirect references
1417 for i
in range(len(endpoints
)):
1418 parts
= endpoints
[i
].split(':')
1419 parts
[0] = self
.resolve(parts
[0])
1420 endpoints
[i
] = ':'.join(parts
)
1422 log
.info('Relating %s <-> %s', *endpoints
)
1423 return await self
.model
.add_relation(*endpoints
)
1425 async def deploy(self
, charm
, series
, application
, options
, constraints
,
1426 storage
, endpoint_bindings
, resources
):
1428 :param charm string:
1429 Charm holds the URL of the charm to be used to deploy this
1432 :param series string:
1433 Series holds the series of the application to be deployed
1434 if the charm default is not sufficient.
1436 :param application string:
1437 Application holds the application name.
1439 :param options map[string]interface{}:
1440 Options holds application options.
1442 :param constraints string:
1443 Constraints holds the optional application constraints.
1445 :param storage map[string]string:
1446 Storage holds the optional storage constraints.
1448 :param endpoint_bindings map[string]string:
1449 EndpointBindings holds the optional endpoint bindings
1451 :param resources map[string]int:
1452 Resources identifies the revision to use for each resource
1453 of the application's charm.
1455 # resolve indirect references
1456 charm
= self
.resolve(charm
)
1457 # stringify all config values for API
1458 options
= {k
: str(v
) for k
, v
in options
.items()}
1459 # build param object
1460 app
= client
.ApplicationDeploy(
1463 application
=application
,
1465 constraints
=parse_constraints(constraints
),
1467 endpoint_bindings
=endpoint_bindings
,
1468 resources
=resources
,
1471 log
.info('Deploying %s', charm
)
1472 await self
.app_facade
.Deploy([app
])
1473 # ensure the app is in the model for future operations
1474 await self
.model
._wait
_for
_new
('application', application
)
1477 async def addUnit(self
, application
, to
):
1479 :param application string:
1480 Application holds the application placeholder name for which a unit
1484 To holds the optional location where to add the unit, as a
1485 placeholder pointing to another unit change or to a machine change.
1487 application
= self
.resolve(application
)
1488 placement
= self
.resolve(to
)
1489 if self
._units
_by
_app
.get(application
):
1490 # enough units for this application already exist;
1491 # claim one, and carry on
1492 # NB: this should probably honor placement, but the juju client
1493 # doesn't, so we're not bothering, either
1494 unit_name
= self
._units
_by
_app
[application
].pop()
1495 log
.debug('Reusing unit %s for %s', unit_name
, application
)
1496 return self
.model
.units
[unit_name
]
1498 log
.debug('Adding new unit for %s%s', application
,
1499 ' to %s' % placement
if placement
else '')
1500 return await self
.model
.applications
[application
].add_unit(
1505 async def expose(self
, application
):
1507 :param application string:
1508 Application holds the placeholder name of the application that must
1511 application
= self
.resolve(application
)
1512 log
.info('Exposing %s', application
)
1513 return await self
.model
.applications
[application
].expose()
1515 async def setAnnotations(self
, id_
, entity_type
, annotations
):
1518 Id is the placeholder for the application or machine change
1519 corresponding to the entity to be annotated.
1521 :param entity_type EntityType:
1522 EntityType holds the type of the entity, "application" or
1525 :param annotations map[string]string:
1526 Annotations holds the annotations as key/value pairs.
1528 entity_id
= self
.resolve(id_
)
1530 entity
= self
.model
.state
.get_entity(entity_type
, entity_id
)
1532 entity
= await self
.model
._wait
_for
_new
(entity_type
, entity_id
)
1533 return await entity
.set_annotations(annotations
)
1536 class CharmStore(object):
1538 Async wrapper around theblues.charmstore.CharmStore
1540 def __init__(self
, loop
):
1542 self
._cs
= charmstore
.CharmStore()
1544 def __getattr__(self
, name
):
1546 Wrap method calls in coroutines that use run_in_executor to make them
1549 attr
= getattr(self
._cs
, name
)
1550 if not callable(attr
):
1551 wrapper
= partial(getattr, self
._cs
, name
)
1552 setattr(self
, name
, wrapper
)
1554 async def coro(*args
, **kwargs
):
1555 method
= partial(attr
, *args
, **kwargs
)
1556 return await self
.loop
.run_in_executor(None, method
)
1557 setattr(self
, name
, coro
)