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 .delta
import get_entity_delta
16 from .delta
import get_entity_class
17 from .exceptions
import DeadEntityException
18 from .errors
import JujuAPIError
20 log
= logging
.getLogger(__name__
)
23 class _Observer(object):
24 """Wrapper around an observer callable.
26 This wrapper allows filter criteria to be associated with the
27 callable so that it's only called for changes that meet the criteria.
30 def __init__(self
, callable_
, entity_type
, action
, entity_id
, predicate
):
31 self
.callable_
= callable_
32 self
.entity_type
= entity_type
34 self
.entity_id
= entity_id
35 self
.predicate
= predicate
37 self
.entity_id
= str(self
.entity_id
)
38 if not self
.entity_id
.startswith('^'):
39 self
.entity_id
= '^' + self
.entity_id
40 if not self
.entity_id
.endswith('$'):
43 async def __call__(self
, delta
, old
, new
, model
):
44 await self
.callable_(delta
, old
, new
, model
)
46 def cares_about(self
, delta
):
47 """Return True if this observer "cares about" (i.e. wants to be
48 called) for a this delta.
51 if (self
.entity_id
and delta
.get_id() and
52 not re
.match(self
.entity_id
, str(delta
.get_id()))):
55 if self
.entity_type
and self
.entity_type
!= delta
.entity
:
58 if self
.action
and self
.action
!= delta
.type:
61 if self
.predicate
and not self
.predicate(delta
):
67 class ModelObserver(object):
68 async def __call__(self
, delta
, old
, new
, model
):
69 handler_name
= 'on_{}_{}'.format(delta
.entity
, delta
.type)
70 method
= getattr(self
, handler_name
, self
.on_change
)
71 await method(delta
, old
, new
, model
)
73 async def on_change(self
, delta
, old
, new
, model
):
77 class ModelState(object):
78 """Holds the state of the model, including the delta history of all
79 entities in the model.
82 def __init__(self
, model
):
86 def _live_entity_map(self
, entity_type
):
87 """Return an id:Entity map of all the living entities of
92 entity_id
: self
.get_entity(entity_type
, entity_id
)
93 for entity_id
, history
in self
.state
.get(entity_type
, {}).items()
94 if history
[-1] is not None
98 def applications(self
):
99 """Return a map of application-name:Application for all applications
100 currently in the model.
103 return self
._live
_entity
_map
('application')
107 """Return a map of machine-id:Machine for all machines currently in
111 return self
._live
_entity
_map
('machine')
115 """Return a map of unit-id:Unit for all units currently in
119 return self
._live
_entity
_map
('unit')
121 def entity_history(self
, entity_type
, entity_id
):
122 """Return the history deque for an entity.
125 return self
.state
[entity_type
][entity_id
]
127 def entity_data(self
, entity_type
, entity_id
, history_index
):
128 """Return the data dict for an entity at a specific index of its
132 return self
.entity_history(entity_type
, entity_id
)[history_index
]
134 def apply_delta(self
, delta
):
135 """Apply delta to our state and return a copy of the
136 affected object as it was before and after the update, e.g.:
138 old_obj, new_obj = self.apply_delta(delta)
140 old_obj may be None if the delta is for the creation of a new object,
141 e.g. a new application or unit is deployed.
143 new_obj will never be None, but may be dead (new_obj.dead == True)
144 if the object was deleted as a result of the delta being applied.
149 .setdefault(delta
.entity
, {})
150 .setdefault(delta
.get_id(), collections
.deque())
153 history
.append(delta
.data
)
154 if delta
.type == 'remove':
157 entity
= self
.get_entity(delta
.entity
, delta
.get_id())
158 return entity
.previous(), entity
161 self
, entity_type
, entity_id
, history_index
=-1, connected
=True):
162 """Return an object instance for the given entity_type and id.
164 By default the object state matches the most recent state from
165 Juju. To get an instance of the object in an older state, pass
166 history_index, an index into the history deque for the entity.
170 if history_index
< 0 and history_index
!= -1:
171 history_index
+= len(self
.entity_history(entity_type
, entity_id
))
172 if history_index
< 0:
176 self
.entity_data(entity_type
, entity_id
, history_index
)
180 entity_class
= get_entity_class(entity_type
)
182 entity_id
, self
.model
, history_index
=history_index
,
186 class ModelEntity(object):
187 """An object in the Model tree"""
189 def __init__(self
, entity_id
, model
, history_index
=-1, connected
=True):
190 """Initialize a new entity
192 :param entity_id str: The unique id of the object in the model
193 :param model: The model instance in whose object tree this
195 :history_index int: The index of this object's state in the model's
196 history deque for this entity
197 :connected bool: Flag indicating whether this object gets live updates
201 self
.entity_id
= entity_id
203 self
._history
_index
= history_index
204 self
.connected
= connected
205 self
.connection
= model
.connection
207 def __getattr__(self
, name
):
208 """Fetch object attributes from the underlying data dict held in the
212 if self
.data
is None:
213 raise DeadEntityException(
214 "Entity {}:{} is dead - its attributes can no longer be "
215 "accessed. Use the .previous() method on this object to get "
216 "a copy of the object at its previous state.".format(
217 self
.entity_type
, self
.entity_id
))
218 return self
.data
[name
]
221 return bool(self
.data
)
223 def on_change(self
, callable_
):
224 """Add a change observer to this entity.
227 self
.model
.add_observer(
228 callable_
, self
.entity_type
, 'change', self
.entity_id
)
230 def on_remove(self
, callable_
):
231 """Add a remove observer to this entity.
234 self
.model
.add_observer(
235 callable_
, self
.entity_type
, 'remove', self
.entity_id
)
238 def entity_type(self
):
239 """A string identifying the entity type of this object, e.g.
240 'application' or 'unit', etc.
243 return self
.__class
__.__name
__.lower()
247 """Return True if this object represents the current state of the
248 entity in the underlying model.
250 This will be True except when the object represents an entity at a
251 non-latest state in history, e.g. if the object was obtained by calling
252 .previous() on another object.
255 return self
._history
_index
== -1
259 """Returns True if this entity no longer exists in the underlying
265 self
.model
.state
.entity_data(
266 self
.entity_type
, self
.entity_id
, -1) is None
271 """Returns True if this entity still exists in the underlying
279 """The data dictionary for this entity.
282 return self
.model
.state
.entity_data(
283 self
.entity_type
, self
.entity_id
, self
._history
_index
)
286 """Return a copy of this object as was at its previous state in
289 Returns None if this object is new (and therefore has no history).
291 The returned object is always "disconnected", i.e. does not receive
295 return self
.model
.state
.get_entity(
296 self
.entity_type
, self
.entity_id
, self
._history
_index
- 1,
300 """Return a copy of this object at its next state in
303 Returns None if this object is already the latest.
305 The returned object is "disconnected", i.e. does not receive
306 live updates, unless it is current (latest).
309 if self
._history
_index
== -1:
312 new_index
= self
._history
_index
+ 1
314 new_index
== len(self
.model
.state
.entity_history(
315 self
.entity_type
, self
.entity_id
)) - 1
317 return self
.model
.state
.get_entity(
318 self
.entity_type
, self
.entity_id
, self
._history
_index
- 1,
322 """Return a copy of this object at its current state in the model.
324 Returns self if this object is already the latest.
326 The returned object is always "connected", i.e. receives
327 live updates from the model.
330 if self
._history
_index
== -1:
333 return self
.model
.state
.get_entity(self
.entity_type
, self
.entity_id
)
337 def __init__(self
, loop
=None):
338 """Instantiate a new connected Model.
340 :param loop: an asyncio event loop
343 self
.loop
= loop
or asyncio
.get_event_loop()
344 self
.connection
= None
345 self
.observers
= weakref
.WeakValueDictionary()
346 self
.state
= ModelState(self
)
347 self
._watcher
_task
= None
348 self
._watch
_shutdown
= asyncio
.Event(loop
=loop
)
349 self
._watch
_received
= asyncio
.Event(loop
=loop
)
350 self
._charmstore
= CharmStore(self
.loop
)
352 async def connect(self
, *args
, **kw
):
353 """Connect to an arbitrary Juju model.
355 args and kw are passed through to Connection.connect()
358 self
.connection
= await connection
.Connection
.connect(*args
, **kw
)
360 await self
._watch
_received
.wait()
362 async def connect_current(self
):
363 """Connect to the current Juju model.
366 self
.connection
= await connection
.Connection
.connect_current()
368 await self
._watch
_received
.wait()
370 async def connect_model(self
, arg
):
371 """Connect to a specific Juju model.
372 :param arg: <controller>:<user/model>
375 self
.connection
= await connection
.Connection
.connect_model(arg
)
377 await self
._watch
_received
.wait()
379 async def disconnect(self
):
380 """Shut down the watcher task and close websockets.
383 self
._stop
_watching
()
384 if self
.connection
and self
.connection
.is_open
:
385 await self
._watch
_shutdown
.wait()
386 log
.debug('Closing model connection')
387 await self
.connection
.close()
388 self
.connection
= None
390 def all_units_idle(self
):
391 """Return True if all units are idle.
394 for unit
in self
.units
.values():
395 unit_status
= unit
.data
['agent-status']['current']
396 if unit_status
!= 'idle':
400 async def reset(self
, force
=False):
401 """Reset the model to a clean state.
403 :param bool force: Force-terminate machines.
405 This returns only after the model has reached a clean state. "Clean"
406 means no applications or machines exist in the model.
409 log
.debug('Resetting model')
410 for app
in self
.applications
.values():
412 for machine
in self
.machines
.values():
413 await machine
.destroy(force
=force
)
414 await self
.block_until(
415 lambda: len(self
.machines
) == 0
418 async def block_until(self
, *conditions
, timeout
=None):
419 """Return only after all conditions are true.
423 while not all(c() for c
in conditions
):
424 await asyncio
.sleep(0)
425 await asyncio
.wait_for(_block(), timeout
)
428 def applications(self
):
429 """Return a map of application-name:Application for all applications
430 currently in the model.
433 return self
.state
.applications
437 """Return a map of machine-id:Machine for all machines currently in
441 return self
.state
.machines
445 """Return a map of unit-id:Unit for all units currently in
449 return self
.state
.units
452 self
, callable_
, entity_type
=None, action
=None, entity_id
=None,
454 """Register an "on-model-change" callback
456 Once the model is connected, ``callable_``
457 will be called each time the model changes. callable_ should
458 be Awaitable and accept the following positional arguments:
460 delta - An instance of :class:`juju.delta.EntityDelta`
461 containing the raw delta data recv'd from the Juju
464 old_obj - If the delta modifies an existing object in the model,
465 old_obj will be a copy of that object, as it was before the
466 delta was applied. Will be None if the delta creates a new
469 new_obj - A copy of the new or updated object, after the delta
470 is applied. Will be None if the delta removes an entity
473 model - The :class:`Model` itself.
475 Events for which ``callable_`` is called can be specified by passing
476 entity_type, action, and/or id_ filter criteria, e.g.:
479 myfunc, entity_type='application', action='add', id_='ubuntu')
481 For more complex filtering conditions, pass a predicate function. It
482 will be called with a delta as its only argument. If the predicate
483 function returns True, the callable_ will be called.
486 observer
= _Observer(
487 callable_
, entity_type
, action
, entity_id
, predicate
)
488 self
.observers
[observer
] = callable_
491 """Start an asynchronous watch against this model.
493 See :meth:`add_observer` to register an onchange callback.
496 async def _start_watch():
497 self
._watch
_shutdown
.clear()
499 allwatcher
= watcher
.AllWatcher()
500 self
._watch
_conn
= await self
.connection
.clone()
501 allwatcher
.connect(self
._watch
_conn
)
503 results
= await allwatcher
.Next()
504 for delta
in results
.deltas
:
505 delta
= get_entity_delta(delta
)
506 old_obj
, new_obj
= self
.state
.apply_delta(delta
)
507 # XXX: Might not want to shield at this level
508 # We are shielding because when the watcher is
509 # canceled (on disconnect()), we don't want all of
510 # its children (every observer callback) to be
511 # canceled with it. So we shield them. But this means
512 # they can *never* be canceled.
513 await asyncio
.shield(
514 self
._notify
_observers
(delta
, old_obj
, new_obj
))
515 self
._watch
_received
.set()
516 except CancelledError
:
517 log
.debug('Closing watcher connection')
518 await self
._watch
_conn
.close()
519 self
._watch
_shutdown
.set()
520 self
._watch
_conn
= None
522 log
.debug('Starting watcher task')
523 self
._watcher
_task
= self
.loop
.create_task(_start_watch())
525 def _stop_watching(self
):
526 """Stop the asynchronous watch against this model.
529 log
.debug('Stopping watcher task')
530 if self
._watcher
_task
:
531 self
._watcher
_task
.cancel()
533 async def _notify_observers(self
, delta
, old_obj
, new_obj
):
534 """Call observing callbacks, notifying them of a change in model state
536 :param delta: The raw change from the watcher
537 (:class:`juju.client.overrides.Delta`)
538 :param old_obj: The object in the model that this delta updates.
540 :param new_obj: The object in the model that is created or updated
541 by applying this delta.
544 if new_obj
and not old_obj
:
548 'Model changed: %s %s %s',
549 delta
.entity
, delta
.type, delta
.get_id())
551 for o
in self
.observers
:
552 if o
.cares_about(delta
):
553 asyncio
.ensure_future(o(delta
, old_obj
, new_obj
, self
))
555 async def _wait(self
, entity_type
, entity_id
, action
, predicate
=None):
557 Block the calling routine until a given action has happened to the
560 :param entity_type: The entity's type.
561 :param entity_id: The entity's id.
562 :param action: the type of action (e.g., 'add' or 'change')
563 :param predicate: optional callable that must take as an
564 argument a delta, and must return a boolean, indicating
565 whether the delta contains the specific action we're looking
566 for. For example, you might check to see whether a 'change'
567 has a 'completed' status. See the _Observer class for details.
570 q
= asyncio
.Queue(loop
=self
.loop
)
572 async def callback(delta
, old
, new
, model
):
573 await q
.put(delta
.get_id())
575 self
.add_observer(callback
, entity_type
, action
, entity_id
, predicate
)
576 entity_id
= await q
.get()
577 return self
.state
._live
_entity
_map
(entity_type
)[entity_id
]
579 async def _wait_for_new(self
, entity_type
, entity_id
=None, predicate
=None):
580 """Wait for a new object to appear in the Model and return it.
582 Waits for an object of type ``entity_type`` with id ``entity_id``.
583 If ``entity_id`` is ``None``, it will wait for the first new entity
586 This coroutine blocks until the new object appears in the model.
589 # if the entity is already in the model, just return it
590 if entity_id
in self
.state
._live
_entity
_map
(entity_type
):
591 return self
.state
._live
_entity
_map
(entity_type
)[entity_id
]
592 # if we know the entity_id, we can trigger on any action that puts
593 # the enitty into the model; otherwise, we have to watch for the
594 # next "add" action on that entity_type
595 action
= 'add' if entity_id
is None else None
596 return await self
._wait
(entity_type
, entity_id
, action
, predicate
)
598 async def wait_for_action(self
, action_id
):
599 """Given an action, wait for it to complete."""
601 if action_id
.startswith("action-"):
602 # if we've been passed action.tag, transform it into the
603 # id that the api deltas will use.
604 action_id
= action_id
[7:]
606 def predicate(delta
):
607 return delta
.data
['status'] in ('completed', 'failed')
609 return await self
._wait
('action', action_id
, 'change', predicate
)
612 self
, spec
=None, constraints
=None, disks
=None, series
=None,
614 """Start a new, empty machine and optionally a container, or add a
615 container to a machine.
617 :param str spec: Machine specification
620 (None) - starts a new machine
621 'lxc' - starts a new machine with on lxc container
622 'lxc:4' - starts a new lxc container on machine 4
623 'ssh:user@10.10.0.3' - manually provisions a machine with ssh
624 'zone=us-east-1a' - starts a machine in zone us-east-1s on AWS
625 'maas2.name' - acquire machine maas2.name on MAAS
626 :param constraints: Machine constraints
627 :type constraints: :class:`juju.Constraints`
628 :param list disks: List of disk :class:`constraints <juju.Constraints>`
629 :param str series: Series
630 :param int count: Number of machines to deploy
632 Supported container types are: lxc, lxd, kvm
634 When deploying a container to an existing machine, constraints cannot
639 add_machines
= add_machine
641 async def add_relation(self
, relation1
, relation2
):
642 """Add a relation between two applications.
644 :param str relation1: '<application>[:<relation_name>]'
645 :param str relation2: '<application>[:<relation_name>]'
648 app_facade
= client
.ApplicationFacade()
649 app_facade
.connect(self
.connection
)
652 'Adding relation %s <-> %s', relation1
, relation2
)
655 result
= await app_facade
.AddRelation([relation1
, relation2
])
656 except JujuAPIError
as e
:
657 if 'relation already exists' not in e
.message
:
660 'Relation %s <-> %s already exists', relation1
, relation2
)
661 # TODO: if relation already exists we should return the
662 # Relation ModelEntity here
665 def predicate(delta
):
667 for endpoint
in delta
.data
['endpoints']:
668 endpoints
[endpoint
['application-name']] = endpoint
['relation']
669 return endpoints
== result
.endpoints
671 return await self
._wait
_for
_new
('relation', None, predicate
)
673 def add_space(self
, name
, *cidrs
):
674 """Add a new network space.
676 Adds a new space with the given name and associates the given
677 (optional) list of existing subnet CIDRs with it.
679 :param str name: Name of the space
680 :param \*cidrs: Optional list of existing subnet CIDRs
685 def add_ssh_key(self
, key
):
686 """Add a public SSH key to this model.
688 :param str key: The public ssh key
692 add_ssh_keys
= add_ssh_key
694 def add_subnet(self
, cidr_or_id
, space
, *zones
):
695 """Add an existing subnet to this model.
697 :param str cidr_or_id: CIDR or provider ID of the existing subnet
698 :param str space: Network space with which to associate
699 :param str \*zones: Zone(s) in which the subnet resides
704 def get_backups(self
):
705 """Retrieve metadata for backups in this model.
710 def block(self
, *commands
):
711 """Add a new block to this model.
713 :param str \*commands: The commands to block. Valid values are
714 'all-changes', 'destroy-model', 'remove-object'
719 def get_blocks(self
):
720 """List blocks for this model.
725 def get_cached_images(self
, arch
=None, kind
=None, series
=None):
726 """Return a list of cached OS images.
728 :param str arch: Filter by image architecture
729 :param str kind: Filter by image kind, e.g. 'lxd'
730 :param str series: Filter by image series, e.g. 'xenial'
735 def create_backup(self
, note
=None, no_download
=False):
736 """Create a backup of this model.
738 :param str note: A note to store with the backup
739 :param bool no_download: Do not download the backup archive
740 :return str: Path to downloaded archive
745 def create_storage_pool(self
, name
, provider_type
, **pool_config
):
746 """Create or define a storage pool.
748 :param str name: Name to give the storage pool
749 :param str provider_type: Pool provider type
750 :param \*\*pool_config: key/value pool configuration pairs
756 self
, no_tail
=False, exclude_module
=None, include_module
=None,
757 include
=None, level
=None, limit
=0, lines
=10, replay
=False,
759 """Get log messages for this model.
761 :param bool no_tail: Stop after returning existing log messages
762 :param list exclude_module: Do not show log messages for these logging
764 :param list include_module: Only show log messages for these logging
766 :param list include: Only show log messages for these entities
767 :param str level: Log level to show, valid options are 'TRACE',
768 'DEBUG', 'INFO', 'WARNING', 'ERROR,
769 :param int limit: Return this many of the most recent (possibly
770 filtered) lines are shown
771 :param int lines: Yield this many of the most recent lines, and keep
773 :param bool replay: Yield the entire log, and keep yielding
774 :param list exclude: Do not show log messages for these entities
780 self
, entity_url
, service_name
=None, bind
=None, budget
=None,
781 channel
=None, config
=None, constraints
=None, force
=False,
782 num_units
=1, plan
=None, resources
=None, series
=None, storage
=None,
784 """Deploy a new service or bundle.
786 :param str entity_url: Charm or bundle url
787 :param str service_name: Name to give the service
788 :param dict bind: <charm endpoint>:<network space> pairs
789 :param dict budget: <budget name>:<limit> pairs
790 :param str channel: Charm store channel from which to retrieve
791 the charm or bundle, e.g. 'development'
792 :param dict config: Charm configuration dictionary
793 :param constraints: Service constraints
794 :type constraints: :class:`juju.Constraints`
795 :param bool force: Allow charm to be deployed to a machine running
796 an unsupported series
797 :param int num_units: Number of units to deploy
798 :param str plan: Plan under which to deploy charm
799 :param dict resources: <resource name>:<file path> pairs
800 :param str series: Series on which to deploy
801 :param dict storage: Storage constraints TODO how do these look?
802 :param str to: Placement directive, e.g.::
805 'lxc:7' - new lxc container on machine 7
806 '24/lxc/3' - lxc container 3 or machine 24
808 If None, a new machine is provisioned.
813 - service_name is required; fill this in automatically if not
815 - series is required; how do we pick a default?
820 client
.Placement(**p
) for p
in to
827 k
: client
.Constraints(**v
)
828 for k
, v
in storage
.items()
831 entity_id
= await self
.charmstore
.entityId(entity_url
)
833 app_facade
= client
.ApplicationFacade()
834 client_facade
= client
.ClientFacade()
835 app_facade
.connect(self
.connection
)
836 client_facade
.connect(self
.connection
)
838 if 'bundle/' in entity_id
:
839 handler
= BundleHandler(self
)
840 await handler
.fetch_plan(entity_id
)
841 await handler
.execute_plan()
842 extant_apps
= {app
for app
in self
.applications
}
843 pending_apps
= set(handler
.applications
) - extant_apps
845 # new apps will usually be in the model by now, but if some
846 # haven't made it yet we'll need to wait on them to be added
847 await asyncio
.gather(*[
848 asyncio
.ensure_future(
849 self
.model
._wait
_for
_new
('application', app_name
))
850 for app_name
in pending_apps
852 return [app
for name
, app
in self
.applications
.items()
853 if name
in handler
.applications
]
856 'Deploying %s', entity_id
)
858 await client_facade
.AddCharm(channel
, entity_id
)
859 app
= client
.ApplicationDeploy(
860 application
=service_name
,
864 constraints
=constraints
,
865 endpoint_bindings
=bind
,
873 await app_facade
.Deploy([app
])
874 return await self
._wait
_for
_new
('application', service_name
)
877 """Terminate all machines and resources for this model.
882 async def destroy_unit(self
, *unit_names
):
883 """Destroy units by name.
886 app_facade
= client
.ApplicationFacade()
887 app_facade
.connect(self
.connection
)
890 'Destroying unit%s %s',
891 's' if len(unit_names
) == 1 else '',
892 ' '.join(unit_names
))
894 return await app_facade
.Destroy(self
.name
)
895 destroy_units
= destroy_unit
897 def get_backup(self
, archive_id
):
898 """Download a backup archive file.
900 :param str archive_id: The id of the archive to download
901 :return str: Path to the archive file
907 self
, num_controllers
=0, constraints
=None, series
=None, to
=None):
908 """Ensure sufficient controllers exist to provide redundancy.
910 :param int num_controllers: Number of controllers to make available
911 :param constraints: Constraints to apply to the controller machines
912 :type constraints: :class:`juju.Constraints`
913 :param str series: Series of the controller machines
914 :param list to: Placement directives for controller machines, e.g.::
917 'lxc:7' - new lxc container on machine 7
918 '24/lxc/3' - lxc container 3 or machine 24
920 If None, a new machine is provisioned.
925 def get_config(self
):
926 """Return the configuration settings for this model.
931 def get_constraints(self
):
932 """Return the machine constraints for this model.
937 def grant(self
, username
, acl
='read'):
938 """Grant a user access to this model.
940 :param str username: Username
941 :param str acl: Access control ('read' or 'write')
946 def import_ssh_key(self
, identity
):
947 """Add a public SSH key from a trusted indentity source to this model.
949 :param str identity: User identity in the form <lp|gh>:<username>
953 import_ssh_keys
= import_ssh_key
955 def get_machines(self
, machine
, utc
=False):
956 """Return list of machines in this model.
958 :param str machine: Machine id, e.g. '0'
959 :param bool utc: Display time as UTC in RFC3339 format
964 def get_shares(self
):
965 """Return list of all users with access to this model.
970 def get_spaces(self
):
971 """Return list of all known spaces, including associated subnets.
976 def get_ssh_key(self
):
977 """Return known SSH keys for this model.
981 get_ssh_keys
= get_ssh_key
983 def get_storage(self
, filesystem
=False, volume
=False):
984 """Return details of storage instances.
986 :param bool filesystem: Include filesystem storage
987 :param bool volume: Include volume storage
992 def get_storage_pools(self
, names
=None, providers
=None):
993 """Return list of storage pools.
995 :param list names: Only include pools with these names
996 :param list providers: Only include pools for these providers
1001 def get_subnets(self
, space
=None, zone
=None):
1002 """Return list of known subnets.
1004 :param str space: Only include subnets in this space
1005 :param str zone: Only include subnets in this zone
1010 def remove_blocks(self
):
1011 """Remove all blocks from this model.
1016 def remove_backup(self
, backup_id
):
1019 :param str backup_id: The id of the backup to remove
1024 def remove_cached_images(self
, arch
=None, kind
=None, series
=None):
1025 """Remove cached OS images.
1027 :param str arch: Architecture of the images to remove
1028 :param str kind: Image kind to remove, e.g. 'lxd'
1029 :param str series: Image series to remove, e.g. 'xenial'
1034 def remove_machine(self
, *machine_ids
):
1035 """Remove a machine from this model.
1037 :param str \*machine_ids: Ids of the machines to remove
1041 remove_machines
= remove_machine
1043 def remove_ssh_key(self
, *keys
):
1044 """Remove a public SSH key(s) from this model.
1046 :param str \*keys: Keys to remove
1050 remove_ssh_keys
= remove_ssh_key
1053 self
, bootstrap
=False, constraints
=None, archive
=None,
1054 backup_id
=None, upload_tools
=False):
1055 """Restore a backup archive to a new controller.
1057 :param bool bootstrap: Bootstrap a new state machine
1058 :param constraints: Model constraints
1059 :type constraints: :class:`juju.Constraints`
1060 :param str archive: Path to backup archive to restore
1061 :param str backup_id: Id of backup to restore
1062 :param bool upload_tools: Upload tools if bootstrapping a new machine
1067 def retry_provisioning(self
):
1068 """Retry provisioning for failed machines.
1073 def revoke(self
, username
, acl
='read'):
1074 """Revoke a user's access to this model.
1076 :param str username: Username to revoke
1077 :param str acl: Access control ('read' or 'write')
1082 def run(self
, command
, timeout
=None):
1083 """Run command on all machines in this model.
1085 :param str command: The command to run
1086 :param int timeout: Time to wait before command is considered failed
1091 def set_config(self
, **config
):
1092 """Set configuration keys on this model.
1094 :param \*\*config: Config key/values
1099 def set_constraints(self
, constraints
):
1100 """Set machine constraints on this model.
1102 :param :class:`juju.Constraints` constraints: Machine constraints
1107 def get_action_output(self
, action_uuid
, wait
=-1):
1108 """Get the results of an action by ID.
1110 :param str action_uuid: Id of the action
1111 :param int wait: Time in seconds to wait for action to complete
1116 def get_action_status(self
, uuid_or_prefix
=None, name
=None):
1117 """Get the status of all actions, filtered by ID, ID prefix, or action name.
1119 :param str uuid_or_prefix: Filter by action uuid or prefix
1120 :param str name: Filter by action name
1125 def get_budget(self
, budget_name
):
1126 """Get budget usage info.
1128 :param str budget_name: Name of budget
1133 def get_status(self
, filter_
=None, utc
=False):
1134 """Return the status of the model.
1136 :param str filter_: Service or unit name or wildcard ('*')
1137 :param bool utc: Display time as UTC in RFC3339 format
1144 self
, all_
=False, destination
=None, dry_run
=False, public
=False,
1145 source
=None, stream
=None, version
=None):
1146 """Copy Juju tools into this model.
1148 :param bool all_: Copy all versions, not just the latest
1149 :param str destination: Path to local destination directory
1150 :param bool dry_run: Don't do the actual copy
1151 :param bool public: Tools are for a public cloud, so generate mirrors
1153 :param str source: Path to local source directory
1154 :param str stream: Simplestreams stream for which to sync metadata
1155 :param str version: Copy a specific major.minor version
1160 def unblock(self
, *commands
):
1161 """Unblock an operation that would alter this model.
1163 :param str \*commands: The commands to unblock. Valid values are
1164 'all-changes', 'destroy-model', 'remove-object'
1169 def unset_config(self
, *keys
):
1170 """Unset configuration on this model.
1172 :param str \*keys: The keys to unset
1177 def upgrade_gui(self
):
1178 """Upgrade the Juju GUI for this model.
1184 self
, dry_run
=False, reset_previous_upgrade
=False,
1185 upload_tools
=False, version
=None):
1186 """Upgrade Juju on all machines in a model.
1188 :param bool dry_run: Don't do the actual upgrade
1189 :param bool reset_previous_upgrade: Clear the previous (incomplete)
1191 :param bool upload_tools: Upload local version of tools
1192 :param str version: Upgrade to a specific version
1197 def upload_backup(self
, archive_path
):
1198 """Store a backup archive remotely in Juju.
1200 :param str archive_path: Path to local archive
1206 def charmstore(self
):
1207 return self
._charmstore
1209 async def get_metrics(self
, *tags
):
1210 """Retrieve metrics.
1212 :param str \*tags: Tags of entities from which to retrieve metrics.
1213 No tags retrieves the metrics of all units in the model.
1215 log
.debug("Retrieving metrics for %s",
1216 ', '.join(tags
) if tags
else "all units")
1218 metrics_facade
= client
.MetricsDebugFacade()
1219 metrics_facade
.connect(self
.connection
)
1221 entities
= [client
.Entity(tag
) for tag
in tags
]
1222 metrics_result
= await metrics_facade
.GetMetrics(entities
)
1224 metrics
= collections
.defaultdict(list)
1226 for entity_metrics
in metrics_result
.results
:
1227 error
= entity_metrics
.error
1229 if "is not a valid tag" in error
:
1230 raise ValueError(error
.message
)
1232 raise Exception(error
.message
)
1234 for metric
in entity_metrics
.metrics
:
1235 metrics
[metric
.unit
].append(metric
.to_json())
1240 class BundleHandler(object):
1242 Handle bundles by using the API to translate bundle YAML into a plan of
1243 steps and then dispatching each of those using the API.
1245 def __init__(self
, model
):
1247 self
.charmstore
= model
.charmstore
1249 self
.references
= {}
1250 self
._units
_by
_app
= {}
1251 for unit_name
, unit
in model
.units
.items():
1252 app_units
= self
._units
_by
_app
.setdefault(unit
.application
, [])
1253 app_units
.append(unit_name
)
1254 self
.client_facade
= client
.ClientFacade()
1255 self
.client_facade
.connect(model
.connection
)
1256 self
.app_facade
= client
.ApplicationFacade()
1257 self
.app_facade
.connect(model
.connection
)
1258 self
.ann_facade
= client
.AnnotationsFacade()
1259 self
.ann_facade
.connect(model
.connection
)
1261 async def fetch_plan(self
, entity_id
):
1262 bundle_yaml
= await self
.charmstore
.files(entity_id
,
1263 filename
='bundle.yaml',
1265 self
.bundle
= yaml
.safe_load(bundle_yaml
)
1266 self
.plan
= await self
.client_facade
.GetBundleChanges(bundle_yaml
)
1268 async def execute_plan(self
):
1269 for step
in self
.plan
.changes
:
1270 method
= getattr(self
, step
.method
)
1271 result
= await method(*step
.args
)
1272 self
.references
[step
.id_
] = result
1275 def applications(self
):
1276 return list(self
.bundle
['services'].keys())
1278 def resolve(self
, reference
):
1279 if reference
and reference
.startswith('$'):
1280 reference
= self
.references
[reference
[1:]]
1283 async def addCharm(self
, charm
, series
):
1285 :param charm string:
1286 Charm holds the URL of the charm to be added.
1288 :param series string:
1289 Series holds the series of the charm to be added
1290 if the charm default is not sufficient.
1292 entity_id
= await self
.charmstore
.entityId(charm
)
1293 log
.debug('Adding %s', entity_id
)
1294 await self
.client_facade
.AddCharm(None, entity_id
)
1297 async def addMachines(self
, series
, constraints
, container_type
,
1300 :param series string:
1301 Series holds the optional machine OS series.
1303 :param constraints string:
1304 Constraints holds the optional machine constraints.
1306 :param Container_type string:
1307 ContainerType optionally holds the type of the container (for
1308 instance ""lxc" or kvm"). It is not specified for top level
1311 :param parent_id string:
1312 ParentId optionally holds a placeholder pointing to another machine
1313 change or to a unit change. This value is only specified in the
1314 case this machine is a container, in which case also ContainerType
1317 params
= client
.AddMachineParams(
1319 constraints
=constraints
,
1320 container_type
=container_type
,
1321 parent_id
=self
.resolve(parent_id
),
1323 results
= await self
.client_facade
.AddMachines(params
)
1324 log
.debug('Added new machine %s', results
[0].machine
)
1325 return results
[0].machine
1327 async def addRelation(self
, endpoint1
, endpoint2
):
1329 :param endpoint1 string:
1330 :param endpoint2 string:
1331 Endpoint1 and Endpoint2 hold relation endpoints in the
1332 "application:interface" form, where the application is always a
1333 placeholder pointing to an application change, and the interface is
1334 optional. Examples are "$deploy-42:web" or just "$deploy-42".
1336 endpoints
= [endpoint1
, endpoint2
]
1337 # resolve indirect references
1338 for i
in range(len(endpoints
)):
1339 parts
= endpoints
[i
].split(':')
1340 parts
[0] = self
.resolve(parts
[0])
1341 endpoints
[i
] = ':'.join(parts
)
1343 log
.info('Relating %s <-> %s', *endpoints
)
1344 return await self
.model
.add_relation(*endpoints
)
1346 async def deploy(self
, charm
, series
, application
, options
, constraints
,
1347 storage
, endpoint_bindings
, resources
):
1349 :param charm string:
1350 Charm holds the URL of the charm to be used to deploy this
1353 :param series string:
1354 Series holds the series of the application to be deployed
1355 if the charm default is not sufficient.
1357 :param application string:
1358 Application holds the application name.
1360 :param options map[string]interface{}:
1361 Options holds application options.
1363 :param constraints string:
1364 Constraints holds the optional application constraints.
1366 :param storage map[string]string:
1367 Storage holds the optional storage constraints.
1369 :param endpoint_bindings map[string]string:
1370 EndpointBindings holds the optional endpoint bindings
1372 :param resources map[string]int:
1373 Resources identifies the revision to use for each resource
1374 of the application's charm.
1376 # resolve indirect references
1377 charm
= self
.resolve(charm
)
1378 # stringify all config values for API
1379 options
= {k
: str(v
) for k
, v
in options
.items()}
1380 # build param object
1381 app
= client
.ApplicationDeploy(
1384 application
=application
,
1386 constraints
=constraints
,
1388 endpoint_bindings
=endpoint_bindings
,
1389 resources
=resources
,
1392 log
.info('Deploying %s', charm
)
1393 await self
.app_facade
.Deploy([app
])
1394 # ensure the app is in the model for future operations
1395 await self
.model
._wait
_for
_new
('application', application
)
1398 async def addUnit(self
, application
, to
):
1400 :param application string:
1401 Application holds the application placeholder name for which a unit
1405 To holds the optional location where to add the unit, as a
1406 placeholder pointing to another unit change or to a machine change.
1408 application
= self
.resolve(application
)
1409 placement
= self
.resolve(to
)
1410 if self
._units
_by
_app
.get(application
):
1411 # enough units for this application already exist;
1412 # claim one, and carry on
1413 # NB: this should probably honor placement, but the juju client
1414 # doesn't, so we're not bothering, either
1415 unit_name
= self
._units
_by
_app
[application
].pop()
1416 log
.debug('Reusing unit %s for %s', unit_name
, application
)
1417 return self
.model
.units
[unit_name
]
1419 log
.debug('Adding new unit for %s%s', application
,
1420 ' to %s' % placement
if placement
else '')
1421 return await self
.model
.applications
[application
].add_unit(
1426 async def expose(self
, application
):
1428 :param application string:
1429 Application holds the placeholder name of the application that must
1432 application
= self
.resolve(application
)
1433 log
.info('Exposing %s', application
)
1434 return await self
.model
.applications
[application
].expose()
1436 async def setAnnotations(self
, id_
, entity_type
, annotations
):
1439 Id is the placeholder for the application or machine change
1440 corresponding to the entity to be annotated.
1442 :param entity_type EntityType:
1443 EntityType holds the type of the entity, "application" or
1446 :param annotations map[string]string:
1447 Annotations holds the annotations as key/value pairs.
1449 entity_id
= self
.resolve(id_
)
1451 entity
= self
.model
.state
.get_entity(entity_type
, entity_id
)
1453 entity
= await self
.model
._wait
_for
_new
(entity_type
, entity_id
)
1454 return await entity
.set_annotations(annotations
)
1457 class CharmStore(object):
1459 Async wrapper around theblues.charmstore.CharmStore
1461 def __init__(self
, loop
):
1463 self
._cs
= charmstore
.CharmStore()
1465 def __getattr__(self
, name
):
1467 Wrap method calls in coroutines that use run_in_executor to make them
1470 attr
= getattr(self
._cs
, name
)
1471 if not callable(attr
):
1472 wrapper
= partial(getattr, self
._cs
, name
)
1473 setattr(self
, name
, wrapper
)
1475 async def coro(*args
, **kwargs
):
1476 method
= partial(attr
, *args
, **kwargs
)
1477 return await self
.loop
.run_in_executor(None, method
)
1478 setattr(self
, name
, coro
)