50b3ad0df431c789148a59291872866cc880aebe
6 from concurrent
.futures
import CancelledError
7 from functools
import partial
10 from theblues
import charmstore
12 from .client
import client
13 from .client
import watcher
14 from .client
import connection
15 from .constraints
import parse
as parse_constraints
16 from .delta
import get_entity_delta
17 from .delta
import get_entity_class
18 from .exceptions
import DeadEntityException
19 from .errors
import JujuAPIError
21 log
= logging
.getLogger(__name__
)
24 class _Observer(object):
25 """Wrapper around an observer callable.
27 This wrapper allows filter criteria to be associated with the
28 callable so that it's only called for changes that meet the criteria.
31 def __init__(self
, callable_
, entity_type
, action
, entity_id
, predicate
):
32 self
.callable_
= callable_
33 self
.entity_type
= entity_type
35 self
.entity_id
= entity_id
36 self
.predicate
= predicate
38 self
.entity_id
= str(self
.entity_id
)
39 if not self
.entity_id
.startswith('^'):
40 self
.entity_id
= '^' + self
.entity_id
41 if not self
.entity_id
.endswith('$'):
44 async def __call__(self
, delta
, old
, new
, model
):
45 await self
.callable_(delta
, old
, new
, model
)
47 def cares_about(self
, delta
):
48 """Return True if this observer "cares about" (i.e. wants to be
49 called) for a this delta.
52 if (self
.entity_id
and delta
.get_id() and
53 not re
.match(self
.entity_id
, str(delta
.get_id()))):
56 if self
.entity_type
and self
.entity_type
!= delta
.entity
:
59 if self
.action
and self
.action
!= delta
.type:
62 if self
.predicate
and not self
.predicate(delta
):
68 class ModelObserver(object):
69 async def __call__(self
, delta
, old
, new
, model
):
70 handler_name
= 'on_{}_{}'.format(delta
.entity
, delta
.type)
71 method
= getattr(self
, handler_name
, self
.on_change
)
72 await method(delta
, old
, new
, model
)
74 async def on_change(self
, delta
, old
, new
, model
):
78 class ModelState(object):
79 """Holds the state of the model, including the delta history of all
80 entities in the model.
83 def __init__(self
, model
):
87 def _live_entity_map(self
, entity_type
):
88 """Return an id:Entity map of all the living entities of
93 entity_id
: self
.get_entity(entity_type
, entity_id
)
94 for entity_id
, history
in self
.state
.get(entity_type
, {}).items()
95 if history
[-1] is not None
99 def applications(self
):
100 """Return a map of application-name:Application for all applications
101 currently in the model.
104 return self
._live
_entity
_map
('application')
108 """Return a map of machine-id:Machine for all machines currently in
112 return self
._live
_entity
_map
('machine')
116 """Return a map of unit-id:Unit for all units currently in
120 return self
._live
_entity
_map
('unit')
122 def entity_history(self
, entity_type
, entity_id
):
123 """Return the history deque for an entity.
126 return self
.state
[entity_type
][entity_id
]
128 def entity_data(self
, entity_type
, entity_id
, history_index
):
129 """Return the data dict for an entity at a specific index of its
133 return self
.entity_history(entity_type
, entity_id
)[history_index
]
135 def apply_delta(self
, delta
):
136 """Apply delta to our state and return a copy of the
137 affected object as it was before and after the update, e.g.:
139 old_obj, new_obj = self.apply_delta(delta)
141 old_obj may be None if the delta is for the creation of a new object,
142 e.g. a new application or unit is deployed.
144 new_obj will never be None, but may be dead (new_obj.dead == True)
145 if the object was deleted as a result of the delta being applied.
150 .setdefault(delta
.entity
, {})
151 .setdefault(delta
.get_id(), collections
.deque())
154 history
.append(delta
.data
)
155 if delta
.type == 'remove':
158 entity
= self
.get_entity(delta
.entity
, delta
.get_id())
159 return entity
.previous(), entity
162 self
, entity_type
, entity_id
, history_index
=-1, connected
=True):
163 """Return an object instance for the given entity_type and id.
165 By default the object state matches the most recent state from
166 Juju. To get an instance of the object in an older state, pass
167 history_index, an index into the history deque for the entity.
171 if history_index
< 0 and history_index
!= -1:
172 history_index
+= len(self
.entity_history(entity_type
, entity_id
))
173 if history_index
< 0:
177 self
.entity_data(entity_type
, entity_id
, history_index
)
181 entity_class
= get_entity_class(entity_type
)
183 entity_id
, self
.model
, history_index
=history_index
,
187 class ModelEntity(object):
188 """An object in the Model tree"""
190 def __init__(self
, entity_id
, model
, history_index
=-1, connected
=True):
191 """Initialize a new entity
193 :param entity_id str: The unique id of the object in the model
194 :param model: The model instance in whose object tree this
196 :history_index int: The index of this object's state in the model's
197 history deque for this entity
198 :connected bool: Flag indicating whether this object gets live updates
202 self
.entity_id
= entity_id
204 self
._history
_index
= history_index
205 self
.connected
= connected
206 self
.connection
= model
.connection
208 def __getattr__(self
, name
):
209 """Fetch object attributes from the underlying data dict held in the
213 if self
.data
is None:
214 raise DeadEntityException(
215 "Entity {}:{} is dead - its attributes can no longer be "
216 "accessed. Use the .previous() method on this object to get "
217 "a copy of the object at its previous state.".format(
218 self
.entity_type
, self
.entity_id
))
219 return self
.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
)
287 """Return a copy of this object as was at its previous state in
290 Returns None if this object is new (and therefore has no history).
292 The returned object is always "disconnected", i.e. does not receive
296 return self
.model
.state
.get_entity(
297 self
.entity_type
, self
.entity_id
, self
._history
_index
- 1,
301 """Return a copy of this object at its next state in
304 Returns None if this object is already the latest.
306 The returned object is "disconnected", i.e. does not receive
307 live updates, unless it is current (latest).
310 if self
._history
_index
== -1:
313 new_index
= self
._history
_index
+ 1
315 new_index
== len(self
.model
.state
.entity_history(
316 self
.entity_type
, self
.entity_id
)) - 1
318 return self
.model
.state
.get_entity(
319 self
.entity_type
, self
.entity_id
, self
._history
_index
- 1,
323 """Return a copy of this object at its current state in the model.
325 Returns self if this object is already the latest.
327 The returned object is always "connected", i.e. receives
328 live updates from the model.
331 if self
._history
_index
== -1:
334 return self
.model
.state
.get_entity(self
.entity_type
, self
.entity_id
)
338 def __init__(self
, loop
=None):
339 """Instantiate a new connected Model.
341 :param loop: an asyncio event loop
344 self
.loop
= loop
or asyncio
.get_event_loop()
345 self
.connection
= None
346 self
.observers
= weakref
.WeakValueDictionary()
347 self
.state
= ModelState(self
)
349 self
._watcher
_task
= None
350 self
._watch
_shutdown
= asyncio
.Event(loop
=loop
)
351 self
._watch
_received
= asyncio
.Event(loop
=loop
)
352 self
._charmstore
= CharmStore(self
.loop
)
354 async def connect(self
, *args
, **kw
):
355 """Connect to an arbitrary Juju model.
357 args and kw are passed through to Connection.connect()
360 self
.connection
= await connection
.Connection
.connect(*args
, **kw
)
361 await self
._after
_connect
()
363 async def connect_current(self
):
364 """Connect to the current Juju model.
367 self
.connection
= await connection
.Connection
.connect_current()
368 await self
._after
_connect
()
370 async def connect_model(self
, model_name
):
371 """Connect to a specific Juju model by name.
373 :param model_name: Format [controller:][user/]model
376 self
.connection
= await connection
.Connection
.connect_model(model_name
)
377 await self
._after
_connect
()
379 async def _after_connect(self
):
380 """Run initialization steps after connecting to websocket.
384 await self
._watch
_received
.wait()
385 await self
.get_info()
387 async def disconnect(self
):
388 """Shut down the watcher task and close websockets.
391 self
._stop
_watching
()
392 if self
.connection
and self
.connection
.is_open
:
393 await self
._watch
_shutdown
.wait()
394 log
.debug('Closing model connection')
395 await self
.connection
.close()
396 self
.connection
= None
398 def all_units_idle(self
):
399 """Return True if all units are idle.
402 for unit
in self
.units
.values():
403 unit_status
= unit
.data
['agent-status']['current']
404 if unit_status
!= 'idle':
408 async def reset(self
, force
=False):
409 """Reset the model to a clean state.
411 :param bool force: Force-terminate machines.
413 This returns only after the model has reached a clean state. "Clean"
414 means no applications or machines exist in the model.
417 log
.debug('Resetting model')
418 for app
in self
.applications
.values():
420 for machine
in self
.machines
.values():
421 await machine
.destroy(force
=force
)
422 await self
.block_until(
423 lambda: len(self
.machines
) == 0
426 async def block_until(self
, *conditions
, timeout
=None):
427 """Return only after all conditions are true.
431 while not all(c() for c
in conditions
):
432 await asyncio
.sleep(0)
433 await asyncio
.wait_for(_block(), timeout
)
436 def applications(self
):
437 """Return a map of application-name:Application for all applications
438 currently in the model.
441 return self
.state
.applications
445 """Return a map of machine-id:Machine for all machines currently in
449 return self
.state
.machines
453 """Return a map of unit-id:Unit for all units currently in
457 return self
.state
.units
459 async def get_info(self
):
460 """Return a client.ModelInfo object for this Model.
462 Retrieves latest info for this Model from the api server. The
463 return value is cached on the Model.info attribute so that the
464 valued may be accessed again without another api call, if
467 This method is called automatically when the Model is connected,
468 resulting in Model.info being initialized without requiring an
469 explicit call to this method.
472 facade
= client
.ClientFacade()
473 facade
.connect(self
.connection
)
475 self
.info
= await facade
.ModelInfo()
476 log
.debug('Got ModelInfo: %s', vars(self
.info
))
481 self
, callable_
, entity_type
=None, action
=None, entity_id
=None,
483 """Register an "on-model-change" callback
485 Once the model is connected, ``callable_``
486 will be called each time the model changes. callable_ should
487 be Awaitable and accept the following positional arguments:
489 delta - An instance of :class:`juju.delta.EntityDelta`
490 containing the raw delta data recv'd from the Juju
493 old_obj - If the delta modifies an existing object in the model,
494 old_obj will be a copy of that object, as it was before the
495 delta was applied. Will be None if the delta creates a new
498 new_obj - A copy of the new or updated object, after the delta
499 is applied. Will be None if the delta removes an entity
502 model - The :class:`Model` itself.
504 Events for which ``callable_`` is called can be specified by passing
505 entity_type, action, and/or id_ filter criteria, e.g.:
508 myfunc, entity_type='application', action='add', id_='ubuntu')
510 For more complex filtering conditions, pass a predicate function. It
511 will be called with a delta as its only argument. If the predicate
512 function returns True, the callable_ will be called.
515 observer
= _Observer(
516 callable_
, entity_type
, action
, entity_id
, predicate
)
517 self
.observers
[observer
] = callable_
520 """Start an asynchronous watch against this model.
522 See :meth:`add_observer` to register an onchange callback.
525 async def _start_watch():
526 self
._watch
_shutdown
.clear()
528 allwatcher
= watcher
.AllWatcher()
529 self
._watch
_conn
= await self
.connection
.clone()
530 allwatcher
.connect(self
._watch
_conn
)
532 results
= await allwatcher
.Next()
533 for delta
in results
.deltas
:
534 delta
= get_entity_delta(delta
)
535 old_obj
, new_obj
= self
.state
.apply_delta(delta
)
536 # XXX: Might not want to shield at this level
537 # We are shielding because when the watcher is
538 # canceled (on disconnect()), we don't want all of
539 # its children (every observer callback) to be
540 # canceled with it. So we shield them. But this means
541 # they can *never* be canceled.
542 await asyncio
.shield(
543 self
._notify
_observers
(delta
, old_obj
, new_obj
))
544 self
._watch
_received
.set()
545 except CancelledError
:
546 log
.debug('Closing watcher connection')
547 await self
._watch
_conn
.close()
548 self
._watch
_shutdown
.set()
549 self
._watch
_conn
= None
551 log
.debug('Starting watcher task')
552 self
._watcher
_task
= self
.loop
.create_task(_start_watch())
554 def _stop_watching(self
):
555 """Stop the asynchronous watch against this model.
558 log
.debug('Stopping watcher task')
559 if self
._watcher
_task
:
560 self
._watcher
_task
.cancel()
562 async def _notify_observers(self
, delta
, old_obj
, new_obj
):
563 """Call observing callbacks, notifying them of a change in model state
565 :param delta: The raw change from the watcher
566 (:class:`juju.client.overrides.Delta`)
567 :param old_obj: The object in the model that this delta updates.
569 :param new_obj: The object in the model that is created or updated
570 by applying this delta.
573 if new_obj
and not old_obj
:
577 'Model changed: %s %s %s',
578 delta
.entity
, delta
.type, delta
.get_id())
580 for o
in self
.observers
:
581 if o
.cares_about(delta
):
582 asyncio
.ensure_future(o(delta
, old_obj
, new_obj
, self
))
584 async def _wait(self
, entity_type
, entity_id
, action
, predicate
=None):
586 Block the calling routine until a given action has happened to the
589 :param entity_type: The entity's type.
590 :param entity_id: The entity's id.
591 :param action: the type of action (e.g., 'add' or 'change')
592 :param predicate: optional callable that must take as an
593 argument a delta, and must return a boolean, indicating
594 whether the delta contains the specific action we're looking
595 for. For example, you might check to see whether a 'change'
596 has a 'completed' status. See the _Observer class for details.
599 q
= asyncio
.Queue(loop
=self
.loop
)
601 async def callback(delta
, old
, new
, model
):
602 await q
.put(delta
.get_id())
604 self
.add_observer(callback
, entity_type
, action
, entity_id
, predicate
)
605 entity_id
= await q
.get()
606 return self
.state
._live
_entity
_map
(entity_type
)[entity_id
]
608 async def _wait_for_new(self
, entity_type
, entity_id
=None, predicate
=None):
609 """Wait for a new object to appear in the Model and return it.
611 Waits for an object of type ``entity_type`` with id ``entity_id``.
612 If ``entity_id`` is ``None``, it will wait for the first new entity
615 This coroutine blocks until the new object appears in the model.
618 # if the entity is already in the model, just return it
619 if entity_id
in self
.state
._live
_entity
_map
(entity_type
):
620 return self
.state
._live
_entity
_map
(entity_type
)[entity_id
]
621 # if we know the entity_id, we can trigger on any action that puts
622 # the enitty into the model; otherwise, we have to watch for the
623 # next "add" action on that entity_type
624 action
= 'add' if entity_id
is None else None
625 return await self
._wait
(entity_type
, entity_id
, action
, predicate
)
627 async def wait_for_action(self
, action_id
):
628 """Given an action, wait for it to complete."""
630 if action_id
.startswith("action-"):
631 # if we've been passed action.tag, transform it into the
632 # id that the api deltas will use.
633 action_id
= action_id
[7:]
635 def predicate(delta
):
636 return delta
.data
['status'] in ('completed', 'failed')
638 return await self
._wait
('action', action_id
, 'change', predicate
)
641 self
, spec
=None, constraints
=None, disks
=None, series
=None,
643 """Start a new, empty machine and optionally a container, or add a
644 container to a machine.
646 :param str spec: Machine specification
649 (None) - starts a new machine
650 'lxc' - starts a new machine with on lxc container
651 'lxc:4' - starts a new lxc container on machine 4
652 'ssh:user@10.10.0.3' - manually provisions a machine with ssh
653 'zone=us-east-1a' - starts a machine in zone us-east-1s on AWS
654 'maas2.name' - acquire machine maas2.name on MAAS
655 :param constraints: Machine constraints
656 :type constraints: :class:`juju.Constraints`
657 :param list disks: List of disk :class:`constraints <juju.Constraints>`
658 :param str series: Series
659 :param int count: Number of machines to deploy
661 Supported container types are: lxc, lxd, kvm
663 When deploying a container to an existing machine, constraints cannot
668 add_machines
= add_machine
670 async def add_relation(self
, relation1
, relation2
):
671 """Add a relation between two applications.
673 :param str relation1: '<application>[:<relation_name>]'
674 :param str relation2: '<application>[:<relation_name>]'
677 app_facade
= client
.ApplicationFacade()
678 app_facade
.connect(self
.connection
)
681 'Adding relation %s <-> %s', relation1
, relation2
)
684 result
= await app_facade
.AddRelation([relation1
, relation2
])
685 except JujuAPIError
as e
:
686 if 'relation already exists' not in e
.message
:
689 'Relation %s <-> %s already exists', relation1
, relation2
)
690 # TODO: if relation already exists we should return the
691 # Relation ModelEntity here
694 def predicate(delta
):
696 for endpoint
in delta
.data
['endpoints']:
697 endpoints
[endpoint
['application-name']] = endpoint
['relation']
698 return endpoints
== result
.endpoints
700 return await self
._wait
_for
_new
('relation', None, predicate
)
702 def add_space(self
, name
, *cidrs
):
703 """Add a new network space.
705 Adds a new space with the given name and associates the given
706 (optional) list of existing subnet CIDRs with it.
708 :param str name: Name of the space
709 :param \*cidrs: Optional list of existing subnet CIDRs
714 def add_ssh_key(self
, key
):
715 """Add a public SSH key to this model.
717 :param str key: The public ssh key
721 add_ssh_keys
= add_ssh_key
723 def add_subnet(self
, cidr_or_id
, space
, *zones
):
724 """Add an existing subnet to this model.
726 :param str cidr_or_id: CIDR or provider ID of the existing subnet
727 :param str space: Network space with which to associate
728 :param str \*zones: Zone(s) in which the subnet resides
733 def get_backups(self
):
734 """Retrieve metadata for backups in this model.
739 def block(self
, *commands
):
740 """Add a new block to this model.
742 :param str \*commands: The commands to block. Valid values are
743 'all-changes', 'destroy-model', 'remove-object'
748 def get_blocks(self
):
749 """List blocks for this model.
754 def get_cached_images(self
, arch
=None, kind
=None, series
=None):
755 """Return a list of cached OS images.
757 :param str arch: Filter by image architecture
758 :param str kind: Filter by image kind, e.g. 'lxd'
759 :param str series: Filter by image series, e.g. 'xenial'
764 def create_backup(self
, note
=None, no_download
=False):
765 """Create a backup of this model.
767 :param str note: A note to store with the backup
768 :param bool no_download: Do not download the backup archive
769 :return str: Path to downloaded archive
774 def create_storage_pool(self
, name
, provider_type
, **pool_config
):
775 """Create or define a storage pool.
777 :param str name: Name to give the storage pool
778 :param str provider_type: Pool provider type
779 :param \*\*pool_config: key/value pool configuration pairs
785 self
, no_tail
=False, exclude_module
=None, include_module
=None,
786 include
=None, level
=None, limit
=0, lines
=10, replay
=False,
788 """Get log messages for this model.
790 :param bool no_tail: Stop after returning existing log messages
791 :param list exclude_module: Do not show log messages for these logging
793 :param list include_module: Only show log messages for these logging
795 :param list include: Only show log messages for these entities
796 :param str level: Log level to show, valid options are 'TRACE',
797 'DEBUG', 'INFO', 'WARNING', 'ERROR,
798 :param int limit: Return this many of the most recent (possibly
799 filtered) lines are shown
800 :param int lines: Yield this many of the most recent lines, and keep
802 :param bool replay: Yield the entire log, and keep yielding
803 :param list exclude: Do not show log messages for these entities
809 self
, entity_url
, service_name
=None, bind
=None, budget
=None,
810 channel
=None, config
=None, constraints
=None, force
=False,
811 num_units
=1, plan
=None, resources
=None, series
=None, storage
=None,
813 """Deploy a new service or bundle.
815 :param str entity_url: Charm or bundle url
816 :param str service_name: Name to give the service
817 :param dict bind: <charm endpoint>:<network space> pairs
818 :param dict budget: <budget name>:<limit> pairs
819 :param str channel: Charm store channel from which to retrieve
820 the charm or bundle, e.g. 'development'
821 :param dict config: Charm configuration dictionary
822 :param constraints: Service constraints
823 :type constraints: :class:`juju.Constraints`
824 :param bool force: Allow charm to be deployed to a machine running
825 an unsupported series
826 :param int num_units: Number of units to deploy
827 :param str plan: Plan under which to deploy charm
828 :param dict resources: <resource name>:<file path> pairs
829 :param str series: Series on which to deploy
830 :param dict storage: Storage constraints TODO how do these look?
831 :param str to: Placement directive, e.g.::
834 'lxc:7' - new lxc container on machine 7
835 '24/lxc/3' - lxc container 3 or machine 24
837 If None, a new machine is provisioned.
842 - service_name is required; fill this in automatically if not
844 - series is required; how do we pick a default?
849 client
.Placement(**p
) for p
in to
856 k
: client
.Constraints(**v
)
857 for k
, v
in storage
.items()
860 entity_id
= await self
.charmstore
.entityId(entity_url
)
862 app_facade
= client
.ApplicationFacade()
863 client_facade
= client
.ClientFacade()
864 app_facade
.connect(self
.connection
)
865 client_facade
.connect(self
.connection
)
867 if 'bundle/' in entity_id
:
868 handler
= BundleHandler(self
)
869 await handler
.fetch_plan(entity_id
)
870 await handler
.execute_plan()
871 extant_apps
= {app
for app
in self
.applications
}
872 pending_apps
= set(handler
.applications
) - extant_apps
874 # new apps will usually be in the model by now, but if some
875 # haven't made it yet we'll need to wait on them to be added
876 await asyncio
.gather(*[
877 asyncio
.ensure_future(
878 self
.model
._wait
_for
_new
('application', app_name
))
879 for app_name
in pending_apps
881 return [app
for name
, app
in self
.applications
.items()
882 if name
in handler
.applications
]
885 'Deploying %s', entity_id
)
887 await client_facade
.AddCharm(channel
, entity_id
)
888 app
= client
.ApplicationDeploy(
889 application
=service_name
,
893 constraints
=constraints
,
894 endpoint_bindings
=bind
,
902 await app_facade
.Deploy([app
])
903 return await self
._wait
_for
_new
('application', service_name
)
906 """Terminate all machines and resources for this model.
911 async def destroy_unit(self
, *unit_names
):
912 """Destroy units by name.
915 app_facade
= client
.ApplicationFacade()
916 app_facade
.connect(self
.connection
)
919 'Destroying unit%s %s',
920 's' if len(unit_names
) == 1 else '',
921 ' '.join(unit_names
))
923 return await app_facade
.Destroy(self
.name
)
924 destroy_units
= destroy_unit
926 def get_backup(self
, archive_id
):
927 """Download a backup archive file.
929 :param str archive_id: The id of the archive to download
930 :return str: Path to the archive file
936 self
, num_controllers
=0, constraints
=None, series
=None, to
=None):
937 """Ensure sufficient controllers exist to provide redundancy.
939 :param int num_controllers: Number of controllers to make available
940 :param constraints: Constraints to apply to the controller machines
941 :type constraints: :class:`juju.Constraints`
942 :param str series: Series of the controller machines
943 :param list to: Placement directives for controller machines, e.g.::
946 'lxc:7' - new lxc container on machine 7
947 '24/lxc/3' - lxc container 3 or machine 24
949 If None, a new machine is provisioned.
954 def get_config(self
):
955 """Return the configuration settings for this model.
960 def get_constraints(self
):
961 """Return the machine constraints for this model.
966 def grant(self
, username
, acl
='read'):
967 """Grant a user access to this model.
969 :param str username: Username
970 :param str acl: Access control ('read' or 'write')
975 def import_ssh_key(self
, identity
):
976 """Add a public SSH key from a trusted indentity source to this model.
978 :param str identity: User identity in the form <lp|gh>:<username>
982 import_ssh_keys
= import_ssh_key
984 def get_machines(self
, machine
, utc
=False):
985 """Return list of machines in this model.
987 :param str machine: Machine id, e.g. '0'
988 :param bool utc: Display time as UTC in RFC3339 format
993 def get_shares(self
):
994 """Return list of all users with access to this model.
999 def get_spaces(self
):
1000 """Return list of all known spaces, including associated subnets.
1005 def get_ssh_key(self
):
1006 """Return known SSH keys for this model.
1010 get_ssh_keys
= get_ssh_key
1012 def get_storage(self
, filesystem
=False, volume
=False):
1013 """Return details of storage instances.
1015 :param bool filesystem: Include filesystem storage
1016 :param bool volume: Include volume storage
1021 def get_storage_pools(self
, names
=None, providers
=None):
1022 """Return list of storage pools.
1024 :param list names: Only include pools with these names
1025 :param list providers: Only include pools for these providers
1030 def get_subnets(self
, space
=None, zone
=None):
1031 """Return list of known subnets.
1033 :param str space: Only include subnets in this space
1034 :param str zone: Only include subnets in this zone
1039 def remove_blocks(self
):
1040 """Remove all blocks from this model.
1045 def remove_backup(self
, backup_id
):
1048 :param str backup_id: The id of the backup to remove
1053 def remove_cached_images(self
, arch
=None, kind
=None, series
=None):
1054 """Remove cached OS images.
1056 :param str arch: Architecture of the images to remove
1057 :param str kind: Image kind to remove, e.g. 'lxd'
1058 :param str series: Image series to remove, e.g. 'xenial'
1063 def remove_machine(self
, *machine_ids
):
1064 """Remove a machine from this model.
1066 :param str \*machine_ids: Ids of the machines to remove
1070 remove_machines
= remove_machine
1072 def remove_ssh_key(self
, *keys
):
1073 """Remove a public SSH key(s) from this model.
1075 :param str \*keys: Keys to remove
1079 remove_ssh_keys
= remove_ssh_key
1082 self
, bootstrap
=False, constraints
=None, archive
=None,
1083 backup_id
=None, upload_tools
=False):
1084 """Restore a backup archive to a new controller.
1086 :param bool bootstrap: Bootstrap a new state machine
1087 :param constraints: Model constraints
1088 :type constraints: :class:`juju.Constraints`
1089 :param str archive: Path to backup archive to restore
1090 :param str backup_id: Id of backup to restore
1091 :param bool upload_tools: Upload tools if bootstrapping a new machine
1096 def retry_provisioning(self
):
1097 """Retry provisioning for failed machines.
1102 def revoke(self
, username
, acl
='read'):
1103 """Revoke a user's access to this model.
1105 :param str username: Username to revoke
1106 :param str acl: Access control ('read' or 'write')
1111 def run(self
, command
, timeout
=None):
1112 """Run command on all machines in this model.
1114 :param str command: The command to run
1115 :param int timeout: Time to wait before command is considered failed
1120 def set_config(self
, **config
):
1121 """Set configuration keys on this model.
1123 :param \*\*config: Config key/values
1128 def set_constraints(self
, constraints
):
1129 """Set machine constraints on this model.
1131 :param :class:`juju.Constraints` constraints: Machine constraints
1136 def get_action_output(self
, action_uuid
, wait
=-1):
1137 """Get the results of an action by ID.
1139 :param str action_uuid: Id of the action
1140 :param int wait: Time in seconds to wait for action to complete
1145 def get_action_status(self
, uuid_or_prefix
=None, name
=None):
1146 """Get the status of all actions, filtered by ID, ID prefix, or action name.
1148 :param str uuid_or_prefix: Filter by action uuid or prefix
1149 :param str name: Filter by action name
1154 def get_budget(self
, budget_name
):
1155 """Get budget usage info.
1157 :param str budget_name: Name of budget
1162 def get_status(self
, filter_
=None, utc
=False):
1163 """Return the status of the model.
1165 :param str filter_: Service or unit name or wildcard ('*')
1166 :param bool utc: Display time as UTC in RFC3339 format
1173 self
, all_
=False, destination
=None, dry_run
=False, public
=False,
1174 source
=None, stream
=None, version
=None):
1175 """Copy Juju tools into this model.
1177 :param bool all_: Copy all versions, not just the latest
1178 :param str destination: Path to local destination directory
1179 :param bool dry_run: Don't do the actual copy
1180 :param bool public: Tools are for a public cloud, so generate mirrors
1182 :param str source: Path to local source directory
1183 :param str stream: Simplestreams stream for which to sync metadata
1184 :param str version: Copy a specific major.minor version
1189 def unblock(self
, *commands
):
1190 """Unblock an operation that would alter this model.
1192 :param str \*commands: The commands to unblock. Valid values are
1193 'all-changes', 'destroy-model', 'remove-object'
1198 def unset_config(self
, *keys
):
1199 """Unset configuration on this model.
1201 :param str \*keys: The keys to unset
1206 def upgrade_gui(self
):
1207 """Upgrade the Juju GUI for this model.
1213 self
, dry_run
=False, reset_previous_upgrade
=False,
1214 upload_tools
=False, version
=None):
1215 """Upgrade Juju on all machines in a model.
1217 :param bool dry_run: Don't do the actual upgrade
1218 :param bool reset_previous_upgrade: Clear the previous (incomplete)
1220 :param bool upload_tools: Upload local version of tools
1221 :param str version: Upgrade to a specific version
1226 def upload_backup(self
, archive_path
):
1227 """Store a backup archive remotely in Juju.
1229 :param str archive_path: Path to local archive
1235 def charmstore(self
):
1236 return self
._charmstore
1238 async def get_metrics(self
, *tags
):
1239 """Retrieve metrics.
1241 :param str \*tags: Tags of entities from which to retrieve metrics.
1242 No tags retrieves the metrics of all units in the model.
1244 log
.debug("Retrieving metrics for %s",
1245 ', '.join(tags
) if tags
else "all units")
1247 metrics_facade
= client
.MetricsDebugFacade()
1248 metrics_facade
.connect(self
.connection
)
1250 entities
= [client
.Entity(tag
) for tag
in tags
]
1251 metrics_result
= await metrics_facade
.GetMetrics(entities
)
1253 metrics
= collections
.defaultdict(list)
1255 for entity_metrics
in metrics_result
.results
:
1256 error
= entity_metrics
.error
1258 if "is not a valid tag" in error
:
1259 raise ValueError(error
.message
)
1261 raise Exception(error
.message
)
1263 for metric
in entity_metrics
.metrics
:
1264 metrics
[metric
.unit
].append(metric
.to_json())
1269 class BundleHandler(object):
1271 Handle bundles by using the API to translate bundle YAML into a plan of
1272 steps and then dispatching each of those using the API.
1274 def __init__(self
, model
):
1276 self
.charmstore
= model
.charmstore
1278 self
.references
= {}
1279 self
._units
_by
_app
= {}
1280 for unit_name
, unit
in model
.units
.items():
1281 app_units
= self
._units
_by
_app
.setdefault(unit
.application
, [])
1282 app_units
.append(unit_name
)
1283 self
.client_facade
= client
.ClientFacade()
1284 self
.client_facade
.connect(model
.connection
)
1285 self
.app_facade
= client
.ApplicationFacade()
1286 self
.app_facade
.connect(model
.connection
)
1287 self
.ann_facade
= client
.AnnotationsFacade()
1288 self
.ann_facade
.connect(model
.connection
)
1290 async def fetch_plan(self
, entity_id
):
1291 bundle_yaml
= await self
.charmstore
.files(entity_id
,
1292 filename
='bundle.yaml',
1294 self
.bundle
= yaml
.safe_load(bundle_yaml
)
1295 self
.plan
= await self
.client_facade
.GetBundleChanges(bundle_yaml
)
1297 async def execute_plan(self
):
1298 for step
in self
.plan
.changes
:
1299 method
= getattr(self
, step
.method
)
1300 result
= await method(*step
.args
)
1301 self
.references
[step
.id_
] = result
1304 def applications(self
):
1305 return list(self
.bundle
['services'].keys())
1307 def resolve(self
, reference
):
1308 if reference
and reference
.startswith('$'):
1309 reference
= self
.references
[reference
[1:]]
1312 async def addCharm(self
, charm
, series
):
1314 :param charm string:
1315 Charm holds the URL of the charm to be added.
1317 :param series string:
1318 Series holds the series of the charm to be added
1319 if the charm default is not sufficient.
1321 entity_id
= await self
.charmstore
.entityId(charm
)
1322 log
.debug('Adding %s', entity_id
)
1323 await self
.client_facade
.AddCharm(None, entity_id
)
1326 async def addMachines(self
, params
=None):
1329 Dictionary specifying the machine to add. All keys are optional.
1332 series: string specifying the machine OS series.
1333 constraints: string holding machine constraints, if any. We'll
1334 parse this into the json friendly dict that the juju api
1336 container_type: string holding the type of the container (for
1337 instance ""lxc" or kvm"). It is not specified for top level
1339 parent_id: string holding a placeholder pointing to another
1340 machine change or to a unit change. This value is only
1341 specified in the case this machine is a container, in
1342 which case also ContainerType is set.
1344 params
= params
or {}
1346 if 'parent_id' in params
:
1347 params
['parent_id'] = self
.resolve(params
['parent_id'])
1349 params
['constraints'] = parse_constraints(
1350 params
.get('constraints'))
1351 params
['jobs'] = params
.get('jobs', ['JobHostUnits'])
1353 params
= client
.AddMachineParams(**params
)
1354 results
= await self
.client_facade
.AddMachines([params
])
1355 error
= results
.machines
[0].error
1357 raise ValueError("Error adding machine: %s", error
.message
)
1358 machine
= results
.machines
[0].machine
1359 log
.debug('Added new machine %s', machine
)
1362 async def addRelation(self
, endpoint1
, endpoint2
):
1364 :param endpoint1 string:
1365 :param endpoint2 string:
1366 Endpoint1 and Endpoint2 hold relation endpoints in the
1367 "application:interface" form, where the application is always a
1368 placeholder pointing to an application change, and the interface is
1369 optional. Examples are "$deploy-42:web" or just "$deploy-42".
1371 endpoints
= [endpoint1
, endpoint2
]
1372 # resolve indirect references
1373 for i
in range(len(endpoints
)):
1374 parts
= endpoints
[i
].split(':')
1375 parts
[0] = self
.resolve(parts
[0])
1376 endpoints
[i
] = ':'.join(parts
)
1378 log
.info('Relating %s <-> %s', *endpoints
)
1379 return await self
.model
.add_relation(*endpoints
)
1381 async def deploy(self
, charm
, series
, application
, options
, constraints
,
1382 storage
, endpoint_bindings
, resources
):
1384 :param charm string:
1385 Charm holds the URL of the charm to be used to deploy this
1388 :param series string:
1389 Series holds the series of the application to be deployed
1390 if the charm default is not sufficient.
1392 :param application string:
1393 Application holds the application name.
1395 :param options map[string]interface{}:
1396 Options holds application options.
1398 :param constraints string:
1399 Constraints holds the optional application constraints.
1401 :param storage map[string]string:
1402 Storage holds the optional storage constraints.
1404 :param endpoint_bindings map[string]string:
1405 EndpointBindings holds the optional endpoint bindings
1407 :param resources map[string]int:
1408 Resources identifies the revision to use for each resource
1409 of the application's charm.
1411 # resolve indirect references
1412 charm
= self
.resolve(charm
)
1413 # stringify all config values for API
1414 options
= {k
: str(v
) for k
, v
in options
.items()}
1415 # build param object
1416 app
= client
.ApplicationDeploy(
1419 application
=application
,
1421 constraints
=constraints
,
1423 endpoint_bindings
=endpoint_bindings
,
1424 resources
=resources
,
1427 log
.info('Deploying %s', charm
)
1428 await self
.app_facade
.Deploy([app
])
1429 # ensure the app is in the model for future operations
1430 await self
.model
._wait
_for
_new
('application', application
)
1433 async def addUnit(self
, application
, to
):
1435 :param application string:
1436 Application holds the application placeholder name for which a unit
1440 To holds the optional location where to add the unit, as a
1441 placeholder pointing to another unit change or to a machine change.
1443 application
= self
.resolve(application
)
1444 placement
= self
.resolve(to
)
1445 if self
._units
_by
_app
.get(application
):
1446 # enough units for this application already exist;
1447 # claim one, and carry on
1448 # NB: this should probably honor placement, but the juju client
1449 # doesn't, so we're not bothering, either
1450 unit_name
= self
._units
_by
_app
[application
].pop()
1451 log
.debug('Reusing unit %s for %s', unit_name
, application
)
1452 return self
.model
.units
[unit_name
]
1454 log
.debug('Adding new unit for %s%s', application
,
1455 ' to %s' % placement
if placement
else '')
1456 return await self
.model
.applications
[application
].add_unit(
1461 async def expose(self
, application
):
1463 :param application string:
1464 Application holds the placeholder name of the application that must
1467 application
= self
.resolve(application
)
1468 log
.info('Exposing %s', application
)
1469 return await self
.model
.applications
[application
].expose()
1471 async def setAnnotations(self
, id_
, entity_type
, annotations
):
1474 Id is the placeholder for the application or machine change
1475 corresponding to the entity to be annotated.
1477 :param entity_type EntityType:
1478 EntityType holds the type of the entity, "application" or
1481 :param annotations map[string]string:
1482 Annotations holds the annotations as key/value pairs.
1484 entity_id
= self
.resolve(id_
)
1486 entity
= self
.model
.state
.get_entity(entity_type
, entity_id
)
1488 entity
= await self
.model
._wait
_for
_new
(entity_type
, entity_id
)
1489 return await entity
.set_annotations(annotations
)
1492 class CharmStore(object):
1494 Async wrapper around theblues.charmstore.CharmStore
1496 def __init__(self
, loop
):
1498 self
._cs
= charmstore
.CharmStore()
1500 def __getattr__(self
, name
):
1502 Wrap method calls in coroutines that use run_in_executor to make them
1505 attr
= getattr(self
._cs
, name
)
1506 if not callable(attr
):
1507 wrapper
= partial(getattr, self
._cs
, name
)
1508 setattr(self
, name
, wrapper
)
1510 async def coro(*args
, **kwargs
):
1511 method
= partial(attr
, *args
, **kwargs
)
1512 return await self
.loop
.run_in_executor(None, method
)
1513 setattr(self
, name
, coro
)