d63f6d0400c440d109dab041c9990cd765a592e9
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 representing the entity created or
168 'Getting %s:%s at index %s',
169 entity_type, entity_id, history_index)
172 if history_index
< 0 and history_index
!= -1:
173 history_index
+= len(self
.entity_history(entity_type
, entity_id
))
174 if history_index
< 0:
178 self
.entity_data(entity_type
, entity_id
, history_index
)
182 entity_class
= get_entity_class(entity_type
)
184 entity_id
, self
.model
, history_index
=history_index
,
188 class ModelEntity(object):
189 """An object in the Model tree"""
191 def __init__(self
, entity_id
, model
, history_index
=-1, connected
=True):
192 """Initialize a new entity
194 :param entity_id str: The unique id of the object in the model
195 :param model: The model instance in whose object tree this
197 :history_index int: The index of this object's state in the model's
198 history deque for this entity
199 :connected bool: Flag indicating whether this object gets live updates
203 self
.entity_id
= entity_id
205 self
._history
_index
= history_index
206 self
.connected
= connected
207 self
.connection
= model
.connection
209 def __getattr__(self
, name
):
210 """Fetch object attributes from the underlying data dict held in the
214 if self
.data
is None:
215 raise DeadEntityException(
216 "Entity {}:{} is dead - its attributes can no longer be "
217 "accessed. Use the .previous() method on this object to get "
218 "a copy of the object at its previous state.".format(
219 self
.entity_type
, self
.entity_id
))
220 return self
.data
[name
]
223 return bool(self
.data
)
225 def on_change(self
, callable_
):
226 """Add a change observer to this entity.
229 self
.model
.add_observer(
230 callable_
, self
.entity_type
, 'change', self
.entity_id
)
232 def on_remove(self
, callable_
):
233 """Add a remove observer to this entity.
236 self
.model
.add_observer(
237 callable_
, self
.entity_type
, 'remove', self
.entity_id
)
240 def entity_type(self
):
241 """A string identifying the entity type of this object, e.g.
242 'application' or 'unit', etc.
245 return self
.__class
__.__name
__.lower()
249 """Return True if this object represents the current state of the
250 entity in the underlying model.
252 This will be True except when the object represents an entity at a
253 non-latest state in history, e.g. if the object was obtained by calling
254 .previous() on another object.
257 return self
._history
_index
== -1
261 """Returns True if this entity no longer exists in the underlying
267 self
.model
.state
.entity_data(
268 self
.entity_type
, self
.entity_id
, -1) is None
273 """Returns True if this entity still exists in the underlying
281 """The data dictionary for this entity.
284 return self
.model
.state
.entity_data(
285 self
.entity_type
, self
.entity_id
, self
._history
_index
)
288 """Return a copy of this object as was at its previous state in
291 Returns None if this object is new (and therefore has no history).
293 The returned object is always "disconnected", i.e. does not receive
297 return self
.model
.state
.get_entity(
298 self
.entity_type
, self
.entity_id
, self
._history
_index
- 1,
302 """Return a copy of this object at its next state in
305 Returns None if this object is already the latest.
307 The returned object is "disconnected", i.e. does not receive
308 live updates, unless it is current (latest).
311 if self
._history
_index
== -1:
314 new_index
= self
._history
_index
+ 1
316 new_index
== len(self
.model
.state
.entity_history(
317 self
.entity_type
, self
.entity_id
)) - 1
319 return self
.model
.state
.get_entity(
320 self
.entity_type
, self
.entity_id
, self
._history
_index
- 1,
324 """Return a copy of this object at its current state in the model.
326 Returns self if this object is already the latest.
328 The returned object is always "connected", i.e. receives
329 live updates from the model.
332 if self
._history
_index
== -1:
335 return self
.model
.state
.get_entity(self
.entity_type
, self
.entity_id
)
339 def __init__(self
, loop
=None):
340 """Instantiate a new connected Model.
342 :param loop: an asyncio event loop
345 self
.loop
= loop
or asyncio
.get_event_loop()
346 self
.connection
= None
347 self
.observers
= weakref
.WeakValueDictionary()
348 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_current(self
):
355 """Connect to the current Juju model.
358 self
.connection
= await connection
.Connection
.connect_current()
360 await self
._watch
_received
.wait()
362 async def disconnect(self
):
363 """Shut down the watcher task and close websockets.
366 self
._stop
_watching
()
367 if self
.connection
and self
.connection
.is_open
:
368 await self
._watch
_shutdown
.wait()
369 log
.debug('Closing model connection')
370 await self
.connection
.close()
371 self
.connection
= None
373 def all_units_idle(self
):
374 """Return True if all units are idle.
377 for unit
in self
.units
.values():
378 unit_status
= unit
.data
['agent-status']['current']
379 if unit_status
!= 'idle':
383 async def reset(self
, force
=False):
384 """Reset the model to a clean state.
386 :param bool force: Force-terminate machines.
388 This returns only after the model has reached a clean state. "Clean"
389 means no applications or machines exist in the model.
392 log
.debug('Resetting model')
393 for app
in self
.applications
.values():
395 for machine
in self
.machines
.values():
396 await machine
.destroy(force
=force
)
397 await self
.block_until(
398 lambda: len(self
.machines
) == 0
401 async def block_until(self
, *conditions
, timeout
=None):
402 """Return only after all conditions are true.
406 while not all(c() for c
in conditions
):
407 await asyncio
.sleep(0)
408 await asyncio
.wait_for(_block(), timeout
)
411 def applications(self
):
412 """Return a map of application-name:Application for all applications
413 currently in the model.
416 return self
.state
.applications
420 """Return a map of machine-id:Machine for all machines currently in
424 return self
.state
.machines
428 """Return a map of unit-id:Unit for all units currently in
432 return self
.state
.units
435 self
, callable_
, entity_type
=None, action
=None, entity_id
=None,
437 """Register an "on-model-change" callback
439 Once the model is connected, ``callable_``
440 will be called each time the model changes. callable_ should
441 be Awaitable and accept the following positional arguments:
443 delta - An instance of :class:`juju.delta.EntityDelta`
444 containing the raw delta data recv'd from the Juju
447 old_obj - If the delta modifies an existing object in the model,
448 old_obj will be a copy of that object, as it was before the
449 delta was applied. Will be None if the delta creates a new
452 new_obj - A copy of the new or updated object, after the delta
453 is applied. Will be None if the delta removes an entity
456 model - The :class:`Model` itself.
458 Events for which ``callable_`` is called can be specified by passing
459 entity_type, action, and/or id_ filter criteria, e.g.:
462 myfunc, entity_type='application', action='add', id_='ubuntu')
464 For more complex filtering conditions, pass a predicate function. It
465 will be called with a delta as its only argument. If the predicate
466 function returns True, the callable_ will be called.
469 observer
= _Observer(
470 callable_
, entity_type
, action
, entity_id
, predicate
)
471 self
.observers
[observer
] = callable_
474 """Start an asynchronous watch against this model.
476 See :meth:`add_observer` to register an onchange callback.
479 async def _start_watch():
480 self
._watch
_shutdown
.clear()
482 allwatcher
= watcher
.AllWatcher()
483 self
._watch
_conn
= await self
.connection
.clone()
484 allwatcher
.connect(self
._watch
_conn
)
486 results
= await allwatcher
.Next()
487 for delta
in results
.deltas
:
488 delta
= get_entity_delta(delta
)
489 old_obj
, new_obj
= self
.state
.apply_delta(delta
)
490 # XXX: Might not want to shield at this level
491 # We are shielding because when the watcher is
492 # canceled (on disconnect()), we don't want all of
493 # its children (every observer callback) to be
494 # canceled with it. So we shield them. But this means
495 # they can *never* be canceled.
496 await asyncio
.shield(
497 self
._notify
_observers
(delta
, old_obj
, new_obj
))
498 self
._watch
_received
.set()
499 except CancelledError
:
500 log
.debug('Closing watcher connection')
501 await self
._watch
_conn
.close()
502 self
._watch
_shutdown
.set()
503 self
._watch
_conn
= None
505 log
.debug('Starting watcher task')
506 self
._watcher
_task
= self
.loop
.create_task(_start_watch())
508 def _stop_watching(self
):
509 """Stop the asynchronous watch against this model.
512 log
.debug('Stopping watcher task')
513 if self
._watcher
_task
:
514 self
._watcher
_task
.cancel()
516 async def _notify_observers(self
, delta
, old_obj
, new_obj
):
517 """Call observing callbacks, notifying them of a change in model state
519 :param delta: The raw change from the watcher
520 (:class:`juju.client.overrides.Delta`)
521 :param old_obj: The object in the model that this delta updates.
523 :param new_obj: The object in the model that is created or updated
524 by applying this delta.
527 if new_obj
and not old_obj
:
531 'Model changed: %s %s %s',
532 delta
.entity
, delta
.type, delta
.get_id())
534 for o
in self
.observers
:
535 if o
.cares_about(delta
):
536 asyncio
.ensure_future(o(delta
, old_obj
, new_obj
, self
))
538 async def _wait(self
, entity_type
, entity_id
, action
, predicate
=None):
540 Block the calling routine until a given action has happened to the
543 :param entity_type: The entity's type.
544 :param entity_id: The entity's id.
545 :param action: the type of action (e.g., 'add' or 'change')
546 :param predicate: optional callable that must take as an
547 argument a delta, and must return a boolean, indicating
548 whether the delta contains the specific action we're looking
549 for. For example, you might check to see whether a 'change'
550 has a 'completed' status. See the _Observer class for details.
553 q
= asyncio
.Queue(loop
=self
.loop
)
555 async def callback(delta
, old
, new
, model
):
556 await q
.put(delta
.get_id())
558 self
.add_observer(callback
, entity_type
, action
, entity_id
, predicate
)
559 entity_id
= await q
.get()
560 return self
.state
._live
_entity
_map
(entity_type
)[entity_id
]
562 async def _wait_for_new(self
, entity_type
, entity_id
, predicate
=None):
563 """Wait for a new object to appear in the Model and return it.
565 Waits for an object of type ``entity_type`` with id ``entity_id``.
567 This coroutine blocks until the new object appears in the model.
570 return await self
._wait
(entity_type
, entity_id
, 'add', predicate
)
572 async def wait_for_action(self
, action_id
):
573 """Given an action, wait for it to complete."""
575 if action_id
.startswith("action-"):
576 # if we've been passed action.tag, transform it into the
577 # id that the api deltas will use.
578 action_id
= action_id
[7:]
580 def predicate(delta
):
581 return delta
.data
['status'] in ('completed', 'error')
583 return await self
._wait
('action', action_id
, 'change', predicate
)
586 self
, spec
=None, constraints
=None, disks
=None, series
=None,
588 """Start a new, empty machine and optionally a container, or add a
589 container to a machine.
591 :param str spec: Machine specification
594 (None) - starts a new machine
595 'lxc' - starts a new machine with on lxc container
596 'lxc:4' - starts a new lxc container on machine 4
597 'ssh:user@10.10.0.3' - manually provisions a machine with ssh
598 'zone=us-east-1a' - starts a machine in zone us-east-1s on AWS
599 'maas2.name' - acquire machine maas2.name on MAAS
600 :param constraints: Machine constraints
601 :type constraints: :class:`juju.Constraints`
602 :param list disks: List of disk :class:`constraints <juju.Constraints>`
603 :param str series: Series
604 :param int count: Number of machines to deploy
606 Supported container types are: lxc, lxd, kvm
608 When deploying a container to an existing machine, constraints cannot
613 add_machines
= add_machine
615 async def add_relation(self
, relation1
, relation2
):
616 """Add a relation between two applications.
618 :param str relation1: '<application>[:<relation_name>]'
619 :param str relation2: '<application>[:<relation_name>]'
622 app_facade
= client
.ApplicationFacade()
623 app_facade
.connect(self
.connection
)
626 'Adding relation %s <-> %s', relation1
, relation2
)
629 result
= await app_facade
.AddRelation([relation1
, relation2
])
630 except JujuAPIError
as e
:
631 if 'relation already exists' not in e
.message
:
634 'Relation %s <-> %s already exists', relation1
, relation2
)
635 # TODO: if relation already exists we should return the
636 # Relation ModelEntity here
639 def predicate(delta
):
641 for endpoint
in delta
.data
['endpoints']:
642 endpoints
[endpoint
['application-name']] = endpoint
['relation']
643 return endpoints
== result
.endpoints
645 return await self
._wait
_for
_new
('relation', None, predicate
)
647 def add_space(self
, name
, *cidrs
):
648 """Add a new network space.
650 Adds a new space with the given name and associates the given
651 (optional) list of existing subnet CIDRs with it.
653 :param str name: Name of the space
654 :param \*cidrs: Optional list of existing subnet CIDRs
659 def add_ssh_key(self
, key
):
660 """Add a public SSH key to this model.
662 :param str key: The public ssh key
666 add_ssh_keys
= add_ssh_key
668 def add_subnet(self
, cidr_or_id
, space
, *zones
):
669 """Add an existing subnet to this model.
671 :param str cidr_or_id: CIDR or provider ID of the existing subnet
672 :param str space: Network space with which to associate
673 :param str \*zones: Zone(s) in which the subnet resides
678 def get_backups(self
):
679 """Retrieve metadata for backups in this model.
684 def block(self
, *commands
):
685 """Add a new block to this model.
687 :param str \*commands: The commands to block. Valid values are
688 'all-changes', 'destroy-model', 'remove-object'
693 def get_blocks(self
):
694 """List blocks for this model.
699 def get_cached_images(self
, arch
=None, kind
=None, series
=None):
700 """Return a list of cached OS images.
702 :param str arch: Filter by image architecture
703 :param str kind: Filter by image kind, e.g. 'lxd'
704 :param str series: Filter by image series, e.g. 'xenial'
709 def create_backup(self
, note
=None, no_download
=False):
710 """Create a backup of this model.
712 :param str note: A note to store with the backup
713 :param bool no_download: Do not download the backup archive
714 :return str: Path to downloaded archive
719 def create_storage_pool(self
, name
, provider_type
, **pool_config
):
720 """Create or define a storage pool.
722 :param str name: Name to give the storage pool
723 :param str provider_type: Pool provider type
724 :param \*\*pool_config: key/value pool configuration pairs
730 self
, no_tail
=False, exclude_module
=None, include_module
=None,
731 include
=None, level
=None, limit
=0, lines
=10, replay
=False,
733 """Get log messages for this model.
735 :param bool no_tail: Stop after returning existing log messages
736 :param list exclude_module: Do not show log messages for these logging
738 :param list include_module: Only show log messages for these logging
740 :param list include: Only show log messages for these entities
741 :param str level: Log level to show, valid options are 'TRACE',
742 'DEBUG', 'INFO', 'WARNING', 'ERROR,
743 :param int limit: Return this many of the most recent (possibly
744 filtered) lines are shown
745 :param int lines: Yield this many of the most recent lines, and keep
747 :param bool replay: Yield the entire log, and keep yielding
748 :param list exclude: Do not show log messages for these entities
754 self
, entity_url
, service_name
=None, bind
=None, budget
=None,
755 channel
=None, config
=None, constraints
=None, force
=False,
756 num_units
=1, plan
=None, resources
=None, series
=None, storage
=None,
758 """Deploy a new service or bundle.
760 :param str entity_url: Charm or bundle url
761 :param str service_name: Name to give the service
762 :param dict bind: <charm endpoint>:<network space> pairs
763 :param dict budget: <budget name>:<limit> pairs
764 :param str channel: Charm store channel from which to retrieve
765 the charm or bundle, e.g. 'development'
766 :param dict config: Charm configuration dictionary
767 :param constraints: Service constraints
768 :type constraints: :class:`juju.Constraints`
769 :param bool force: Allow charm to be deployed to a machine running
770 an unsupported series
771 :param int num_units: Number of units to deploy
772 :param str plan: Plan under which to deploy charm
773 :param dict resources: <resource name>:<file path> pairs
774 :param str series: Series on which to deploy
775 :param dict storage: Storage constraints TODO how do these look?
776 :param str to: Placement directive, e.g.::
779 'lxc:7' - new lxc container on machine 7
780 '24/lxc/3' - lxc container 3 or machine 24
782 If None, a new machine is provisioned.
787 - service_name is required; fill this in automatically if not
789 - series is required; how do we pick a default?
794 client
.Placement(**p
) for p
in to
801 k
: client
.Constraints(**v
)
802 for k
, v
in storage
.items()
805 entity_id
= await self
.charmstore
.entityId(entity_url
)
807 app_facade
= client
.ApplicationFacade()
808 client_facade
= client
.ClientFacade()
809 app_facade
.connect(self
.connection
)
810 client_facade
.connect(self
.connection
)
812 if 'bundle/' in entity_id
:
813 handler
= BundleHandler(self
)
814 await handler
.fetch_plan(entity_id
)
815 await handler
.execute_plan()
816 extant_apps
= {app
for app
in self
.applications
}
817 pending_apps
= set(handler
.applications
) - extant_apps
819 # new apps will usually be in the model by now, but if some
820 # haven't made it yet we'll need to wait on them to be added
821 await asyncio
.gather(*[
822 asyncio
.ensure_future(
823 self
.model
._wait
_for
_new
('application', app_name
))
824 for app_name
in pending_apps
826 return [app
for name
, app
in self
.applications
.items()
827 if name
in handler
.applications
]
830 'Deploying %s', entity_id
)
832 await client_facade
.AddCharm(channel
, entity_id
)
833 app
= client
.ApplicationDeploy(
834 application
=service_name
,
838 constraints
=constraints
,
839 endpoint_bindings
=bind
,
847 await app_facade
.Deploy([app
])
848 return await self
._wait
_for
_new
('application', service_name
)
851 """Terminate all machines and resources for this model.
856 async def destroy_unit(self
, *unit_names
):
857 """Destroy units by name.
860 app_facade
= client
.ApplicationFacade()
861 app_facade
.connect(self
.connection
)
864 'Destroying unit%s %s',
865 's' if len(unit_names
) == 1 else '',
866 ' '.join(unit_names
))
868 return await app_facade
.Destroy(self
.name
)
869 destroy_units
= destroy_unit
871 def get_backup(self
, archive_id
):
872 """Download a backup archive file.
874 :param str archive_id: The id of the archive to download
875 :return str: Path to the archive file
881 self
, num_controllers
=0, constraints
=None, series
=None, to
=None):
882 """Ensure sufficient controllers exist to provide redundancy.
884 :param int num_controllers: Number of controllers to make available
885 :param constraints: Constraints to apply to the controller machines
886 :type constraints: :class:`juju.Constraints`
887 :param str series: Series of the controller machines
888 :param list to: Placement directives for controller machines, e.g.::
891 'lxc:7' - new lxc container on machine 7
892 '24/lxc/3' - lxc container 3 or machine 24
894 If None, a new machine is provisioned.
899 def get_config(self
):
900 """Return the configuration settings for this model.
905 def get_constraints(self
):
906 """Return the machine constraints for this model.
911 def grant(self
, username
, acl
='read'):
912 """Grant a user access to this model.
914 :param str username: Username
915 :param str acl: Access control ('read' or 'write')
920 def import_ssh_key(self
, identity
):
921 """Add a public SSH key from a trusted indentity source to this model.
923 :param str identity: User identity in the form <lp|gh>:<username>
927 import_ssh_keys
= import_ssh_key
929 def get_machines(self
, machine
, utc
=False):
930 """Return list of machines in this model.
932 :param str machine: Machine id, e.g. '0'
933 :param bool utc: Display time as UTC in RFC3339 format
938 def get_shares(self
):
939 """Return list of all users with access to this model.
944 def get_spaces(self
):
945 """Return list of all known spaces, including associated subnets.
950 def get_ssh_key(self
):
951 """Return known SSH keys for this model.
955 get_ssh_keys
= get_ssh_key
957 def get_storage(self
, filesystem
=False, volume
=False):
958 """Return details of storage instances.
960 :param bool filesystem: Include filesystem storage
961 :param bool volume: Include volume storage
966 def get_storage_pools(self
, names
=None, providers
=None):
967 """Return list of storage pools.
969 :param list names: Only include pools with these names
970 :param list providers: Only include pools for these providers
975 def get_subnets(self
, space
=None, zone
=None):
976 """Return list of known subnets.
978 :param str space: Only include subnets in this space
979 :param str zone: Only include subnets in this zone
984 def remove_blocks(self
):
985 """Remove all blocks from this model.
990 def remove_backup(self
, backup_id
):
993 :param str backup_id: The id of the backup to remove
998 def remove_cached_images(self
, arch
=None, kind
=None, series
=None):
999 """Remove cached OS images.
1001 :param str arch: Architecture of the images to remove
1002 :param str kind: Image kind to remove, e.g. 'lxd'
1003 :param str series: Image series to remove, e.g. 'xenial'
1008 def remove_machine(self
, *machine_ids
):
1009 """Remove a machine from this model.
1011 :param str \*machine_ids: Ids of the machines to remove
1015 remove_machines
= remove_machine
1017 def remove_ssh_key(self
, *keys
):
1018 """Remove a public SSH key(s) from this model.
1020 :param str \*keys: Keys to remove
1024 remove_ssh_keys
= remove_ssh_key
1027 self
, bootstrap
=False, constraints
=None, archive
=None,
1028 backup_id
=None, upload_tools
=False):
1029 """Restore a backup archive to a new controller.
1031 :param bool bootstrap: Bootstrap a new state machine
1032 :param constraints: Model constraints
1033 :type constraints: :class:`juju.Constraints`
1034 :param str archive: Path to backup archive to restore
1035 :param str backup_id: Id of backup to restore
1036 :param bool upload_tools: Upload tools if bootstrapping a new machine
1041 def retry_provisioning(self
):
1042 """Retry provisioning for failed machines.
1047 def revoke(self
, username
, acl
='read'):
1048 """Revoke a user's access to this model.
1050 :param str username: Username to revoke
1051 :param str acl: Access control ('read' or 'write')
1056 def run(self
, command
, timeout
=None):
1057 """Run command on all machines in this model.
1059 :param str command: The command to run
1060 :param int timeout: Time to wait before command is considered failed
1065 def set_config(self
, **config
):
1066 """Set configuration keys on this model.
1068 :param \*\*config: Config key/values
1073 def set_constraints(self
, constraints
):
1074 """Set machine constraints on this model.
1076 :param :class:`juju.Constraints` constraints: Machine constraints
1081 def get_action_output(self
, action_uuid
, wait
=-1):
1082 """Get the results of an action by ID.
1084 :param str action_uuid: Id of the action
1085 :param int wait: Time in seconds to wait for action to complete
1090 def get_action_status(self
, uuid_or_prefix
=None, name
=None):
1091 """Get the status of all actions, filtered by ID, ID prefix, or action name.
1093 :param str uuid_or_prefix: Filter by action uuid or prefix
1094 :param str name: Filter by action name
1099 def get_budget(self
, budget_name
):
1100 """Get budget usage info.
1102 :param str budget_name: Name of budget
1107 def get_status(self
, filter_
=None, utc
=False):
1108 """Return the status of the model.
1110 :param str filter_: Service or unit name or wildcard ('*')
1111 :param bool utc: Display time as UTC in RFC3339 format
1118 self
, all_
=False, destination
=None, dry_run
=False, public
=False,
1119 source
=None, stream
=None, version
=None):
1120 """Copy Juju tools into this model.
1122 :param bool all_: Copy all versions, not just the latest
1123 :param str destination: Path to local destination directory
1124 :param bool dry_run: Don't do the actual copy
1125 :param bool public: Tools are for a public cloud, so generate mirrors
1127 :param str source: Path to local source directory
1128 :param str stream: Simplestreams stream for which to sync metadata
1129 :param str version: Copy a specific major.minor version
1134 def unblock(self
, *commands
):
1135 """Unblock an operation that would alter this model.
1137 :param str \*commands: The commands to unblock. Valid values are
1138 'all-changes', 'destroy-model', 'remove-object'
1143 def unset_config(self
, *keys
):
1144 """Unset configuration on this model.
1146 :param str \*keys: The keys to unset
1151 def upgrade_gui(self
):
1152 """Upgrade the Juju GUI for this model.
1158 self
, dry_run
=False, reset_previous_upgrade
=False,
1159 upload_tools
=False, version
=None):
1160 """Upgrade Juju on all machines in a model.
1162 :param bool dry_run: Don't do the actual upgrade
1163 :param bool reset_previous_upgrade: Clear the previous (incomplete)
1165 :param bool upload_tools: Upload local version of tools
1166 :param str version: Upgrade to a specific version
1171 def upload_backup(self
, archive_path
):
1172 """Store a backup archive remotely in Juju.
1174 :param str archive_path: Path to local archive
1180 def charmstore(self
):
1181 return self
._charmstore
1184 class BundleHandler(object):
1186 Handle bundles by using the API to translate bundle YAML into a plan of
1187 steps and then dispatching each of those using the API.
1189 def __init__(self
, model
):
1191 self
.charmstore
= model
.charmstore
1193 self
.references
= {}
1194 self
._units
_by
_app
= {}
1195 for unit_name
, unit
in model
.units
.items():
1196 app_units
= self
._units
_by
_app
.setdefault(unit
.application
, [])
1197 app_units
.append(unit_name
)
1198 self
.client_facade
= client
.ClientFacade()
1199 self
.client_facade
.connect(model
.connection
)
1200 self
.app_facade
= client
.ApplicationFacade()
1201 self
.app_facade
.connect(model
.connection
)
1202 self
.ann_facade
= client
.AnnotationsFacade()
1203 self
.ann_facade
.connect(model
.connection
)
1205 async def fetch_plan(self
, entity_id
):
1206 bundle_yaml
= await self
.charmstore
.files(entity_id
,
1207 filename
='bundle.yaml',
1209 self
.bundle
= yaml
.safe_load(bundle_yaml
)
1210 self
.plan
= await self
.client_facade
.GetBundleChanges(bundle_yaml
)
1212 async def execute_plan(self
):
1213 for step
in self
.plan
.changes
:
1214 method
= getattr(self
, step
.method
)
1215 result
= await method(*step
.args
)
1216 self
.references
[step
.id_
] = result
1219 def applications(self
):
1220 return list(self
.bundle
['services'].keys())
1222 def resolve(self
, reference
):
1223 if reference
and reference
.startswith('$'):
1224 reference
= self
.references
[reference
[1:]]
1227 async def addCharm(self
, charm
, series
):
1229 :param charm string:
1230 Charm holds the URL of the charm to be added.
1232 :param series string:
1233 Series holds the series of the charm to be added
1234 if the charm default is not sufficient.
1236 entity_id
= await self
.charmstore
.entityId(charm
)
1237 log
.debug('Adding %s', entity_id
)
1238 await self
.client_facade
.AddCharm(None, entity_id
)
1241 async def addMachines(self
, series
, constraints
, container_type
,
1244 :param series string:
1245 Series holds the optional machine OS series.
1247 :param constraints string:
1248 Constraints holds the optional machine constraints.
1250 :param Container_type string:
1251 ContainerType optionally holds the type of the container (for
1252 instance ""lxc" or kvm"). It is not specified for top level
1255 :param parent_id string:
1256 ParentId optionally holds a placeholder pointing to another machine
1257 change or to a unit change. This value is only specified in the
1258 case this machine is a container, in which case also ContainerType
1261 params
= client
.AddMachineParams(
1263 constraints
=constraints
,
1264 container_type
=container_type
,
1265 parent_id
=self
.resolve(parent_id
),
1267 results
= await self
.client_facade
.AddMachines(params
)
1268 log
.debug('Added new machine %s', results
[0].machine
)
1269 return results
[0].machine
1271 async def addRelation(self
, endpoint1
, endpoint2
):
1273 :param endpoint1 string:
1274 :param endpoint2 string:
1275 Endpoint1 and Endpoint2 hold relation endpoints in the
1276 "application:interface" form, where the application is always a
1277 placeholder pointing to an application change, and the interface is
1278 optional. Examples are "$deploy-42:web" or just "$deploy-42".
1280 endpoints
= [endpoint1
, endpoint2
]
1281 # resolve indirect references
1282 for i
in range(len(endpoints
)):
1283 parts
= endpoints
[i
].split(':')
1284 parts
[0] = self
.resolve(parts
[0])
1285 endpoints
[i
] = ':'.join(parts
)
1287 log
.info('Relating %s <-> %s', *endpoints
)
1288 return await self
.model
.add_relation(*endpoints
)
1290 async def deploy(self
, charm
, series
, application
, options
, constraints
,
1291 storage
, endpoint_bindings
, resources
):
1293 :param charm string:
1294 Charm holds the URL of the charm to be used to deploy this
1297 :param series string:
1298 Series holds the series of the application to be deployed
1299 if the charm default is not sufficient.
1301 :param application string:
1302 Application holds the application name.
1304 :param options map[string]interface{}:
1305 Options holds application options.
1307 :param constraints string:
1308 Constraints holds the optional application constraints.
1310 :param storage map[string]string:
1311 Storage holds the optional storage constraints.
1313 :param endpoint_bindings map[string]string:
1314 EndpointBindings holds the optional endpoint bindings
1316 :param resources map[string]int:
1317 Resources identifies the revision to use for each resource
1318 of the application's charm.
1320 # resolve indirect references
1321 charm
= self
.resolve(charm
)
1322 # stringify all config values for API
1323 options
= {k
: str(v
) for k
, v
in options
.items()}
1324 # build param object
1325 app
= client
.ApplicationDeploy(
1328 application
=application
,
1330 constraints
=constraints
,
1332 endpoint_bindings
=endpoint_bindings
,
1333 resources
=resources
,
1336 log
.info('Deploying %s', charm
)
1337 await self
.app_facade
.Deploy([app
])
1340 async def addUnit(self
, application
, to
):
1342 :param application string:
1343 Application holds the application placeholder name for which a unit
1347 To holds the optional location where to add the unit, as a
1348 placeholder pointing to another unit change or to a machine change.
1350 application
= self
.resolve(application
)
1351 placement
= self
.resolve(to
)
1352 if self
._units
_by
_app
.get(application
):
1353 # enough units for this application already exist;
1354 # claim one, and carry on
1355 # NB: this should probably honor placement, but the juju client
1356 # doesn't, so we're not bothering, either
1357 unit_name
= self
._units
_by
_app
[application
].pop()
1358 log
.debug('Reusing unit %s for %s', unit_name
, application
)
1359 return self
.model
.units
[unit_name
]
1361 log
.debug('Adding new unit for %s%s', application
,
1362 ' to %s' % placement
if placement
else '')
1363 return await self
.model
.applications
[application
].add_unit(
1368 async def expose(self
, application
):
1370 :param application string:
1371 Application holds the placeholder name of the application that must
1374 application
= self
.resolve(application
)
1375 log
.info('Exposing %s', application
)
1376 return await self
.model
.applications
[application
].expose()
1378 async def setAnnotations(self
, id_
, entity_type
, annotations
):
1381 Id is the placeholder for the application or machine change
1382 corresponding to the entity to be annotated.
1384 :param entity_type EntityType:
1385 EntityType holds the type of the entity, "application" or
1388 :param annotations map[string]string:
1389 Annotations holds the annotations as key/value pairs.
1391 entity_id
= self
.resolve(id_
)
1393 entity
= self
.model
.state
.get_entity(entity_type
, entity_id
)
1395 entity
= await self
.model
._wait
_for
_new
(entity_type
, entity_id
)
1396 return await entity
.set_annotations(annotations
)
1399 class CharmStore(object):
1401 Async wrapper around theblues.charmstore.CharmStore
1403 def __init__(self
, loop
):
1405 self
._cs
= charmstore
.CharmStore()
1407 def __getattr__(self
, name
):
1409 Wrap method calls in coroutines that use run_in_executor to make them
1412 attr
= getattr(self
._cs
, name
)
1413 if not callable(attr
):
1414 wrapper
= partial(getattr, self
._cs
, name
)
1415 setattr(self
, name
, wrapper
)
1417 async def coro(*args
, **kwargs
):
1418 method
= partial(attr
, *args
, **kwargs
)
1419 return await self
.loop
.run_in_executor(None, method
)
1420 setattr(self
, name
, coro
)