e2d3d5296e39df51fd55c6c490d8223687aeed29
7 from concurrent
.futures
import CancelledError
8 from functools
import partial
9 from pathlib
import Path
12 from theblues
import charmstore
14 from .client
import client
15 from .client
import watcher
16 from .client
import connection
17 from .constraints
import parse
as parse_constraints
18 from .delta
import get_entity_delta
19 from .delta
import get_entity_class
20 from .exceptions
import DeadEntityException
21 from .errors
import JujuAPIError
23 log
= logging
.getLogger(__name__
)
26 class _Observer(object):
27 """Wrapper around an observer callable.
29 This wrapper allows filter criteria to be associated with the
30 callable so that it's only called for changes that meet the criteria.
33 def __init__(self
, callable_
, entity_type
, action
, entity_id
, predicate
):
34 self
.callable_
= callable_
35 self
.entity_type
= entity_type
37 self
.entity_id
= entity_id
38 self
.predicate
= predicate
40 self
.entity_id
= str(self
.entity_id
)
41 if not self
.entity_id
.startswith('^'):
42 self
.entity_id
= '^' + self
.entity_id
43 if not self
.entity_id
.endswith('$'):
46 async def __call__(self
, delta
, old
, new
, model
):
47 await self
.callable_(delta
, old
, new
, model
)
49 def cares_about(self
, delta
):
50 """Return True if this observer "cares about" (i.e. wants to be
51 called) for a this delta.
54 if (self
.entity_id
and delta
.get_id() and
55 not re
.match(self
.entity_id
, str(delta
.get_id()))):
58 if self
.entity_type
and self
.entity_type
!= delta
.entity
:
61 if self
.action
and self
.action
!= delta
.type:
64 if self
.predicate
and not self
.predicate(delta
):
70 class ModelObserver(object):
71 async def __call__(self
, delta
, old
, new
, model
):
72 handler_name
= 'on_{}_{}'.format(delta
.entity
, delta
.type)
73 method
= getattr(self
, handler_name
, self
.on_change
)
74 await method(delta
, old
, new
, model
)
76 async def on_change(self
, delta
, old
, new
, model
):
80 class ModelState(object):
81 """Holds the state of the model, including the delta history of all
82 entities in the model.
85 def __init__(self
, model
):
89 def _live_entity_map(self
, entity_type
):
90 """Return an id:Entity map of all the living entities of
95 entity_id
: self
.get_entity(entity_type
, entity_id
)
96 for entity_id
, history
in self
.state
.get(entity_type
, {}).items()
97 if history
[-1] is not None
101 def applications(self
):
102 """Return a map of application-name:Application for all applications
103 currently in the model.
106 return self
._live
_entity
_map
('application')
110 """Return a map of machine-id:Machine for all machines currently in
114 return self
._live
_entity
_map
('machine')
118 """Return a map of unit-id:Unit for all units currently in
122 return self
._live
_entity
_map
('unit')
124 def entity_history(self
, entity_type
, entity_id
):
125 """Return the history deque for an entity.
128 return self
.state
[entity_type
][entity_id
]
130 def entity_data(self
, entity_type
, entity_id
, history_index
):
131 """Return the data dict for an entity at a specific index of its
135 return self
.entity_history(entity_type
, entity_id
)[history_index
]
137 def apply_delta(self
, delta
):
138 """Apply delta to our state and return a copy of the
139 affected object as it was before and after the update, e.g.:
141 old_obj, new_obj = self.apply_delta(delta)
143 old_obj may be None if the delta is for the creation of a new object,
144 e.g. a new application or unit is deployed.
146 new_obj will never be None, but may be dead (new_obj.dead == True)
147 if the object was deleted as a result of the delta being applied.
152 .setdefault(delta
.entity
, {})
153 .setdefault(delta
.get_id(), collections
.deque())
156 history
.append(delta
.data
)
157 if delta
.type == 'remove':
160 entity
= self
.get_entity(delta
.entity
, delta
.get_id())
161 return entity
.previous(), entity
164 self
, entity_type
, entity_id
, history_index
=-1, connected
=True):
165 """Return an object instance for the given entity_type and id.
167 By default the object state matches the most recent state from
168 Juju. To get an instance of the object in an older state, pass
169 history_index, an index into the history deque for the entity.
173 if history_index
< 0 and history_index
!= -1:
174 history_index
+= len(self
.entity_history(entity_type
, entity_id
))
175 if history_index
< 0:
179 self
.entity_data(entity_type
, entity_id
, history_index
)
183 entity_class
= get_entity_class(entity_type
)
185 entity_id
, self
.model
, history_index
=history_index
,
189 class ModelEntity(object):
190 """An object in the Model tree"""
192 def __init__(self
, entity_id
, model
, history_index
=-1, connected
=True):
193 """Initialize a new entity
195 :param entity_id str: The unique id of the object in the model
196 :param model: The model instance in whose object tree this
198 :history_index int: The index of this object's state in the model's
199 history deque for this entity
200 :connected bool: Flag indicating whether this object gets live updates
204 self
.entity_id
= entity_id
206 self
._history
_index
= history_index
207 self
.connected
= connected
208 self
.connection
= model
.connection
210 def __getattr__(self
, name
):
211 """Fetch object attributes from the underlying data dict held in the
215 if self
.data
is None:
216 raise DeadEntityException(
217 "Entity {}:{} is dead - its attributes can no longer be "
218 "accessed. Use the .previous() method on this object to get "
219 "a copy of the object at its previous state.".format(
220 self
.entity_type
, self
.entity_id
))
221 return self
.data
[name
]
224 return bool(self
.data
)
226 def on_change(self
, callable_
):
227 """Add a change observer to this entity.
230 self
.model
.add_observer(
231 callable_
, self
.entity_type
, 'change', self
.entity_id
)
233 def on_remove(self
, callable_
):
234 """Add a remove observer to this entity.
237 self
.model
.add_observer(
238 callable_
, self
.entity_type
, 'remove', self
.entity_id
)
241 def entity_type(self
):
242 """A string identifying the entity type of this object, e.g.
243 'application' or 'unit', etc.
246 return self
.__class
__.__name
__.lower()
250 """Return True if this object represents the current state of the
251 entity in the underlying model.
253 This will be True except when the object represents an entity at a
254 non-latest state in history, e.g. if the object was obtained by calling
255 .previous() on another object.
258 return self
._history
_index
== -1
262 """Returns True if this entity no longer exists in the underlying
268 self
.model
.state
.entity_data(
269 self
.entity_type
, self
.entity_id
, -1) is None
274 """Returns True if this entity still exists in the underlying
282 """The data dictionary for this entity.
285 return self
.model
.state
.entity_data(
286 self
.entity_type
, self
.entity_id
, self
._history
_index
)
289 """Return a copy of this object as was at its previous state in
292 Returns None if this object is new (and therefore has no history).
294 The returned object is always "disconnected", i.e. does not receive
298 return self
.model
.state
.get_entity(
299 self
.entity_type
, self
.entity_id
, self
._history
_index
- 1,
303 """Return a copy of this object at its next state in
306 Returns None if this object is already the latest.
308 The returned object is "disconnected", i.e. does not receive
309 live updates, unless it is current (latest).
312 if self
._history
_index
== -1:
315 new_index
= self
._history
_index
+ 1
317 new_index
== len(self
.model
.state
.entity_history(
318 self
.entity_type
, self
.entity_id
)) - 1
320 return self
.model
.state
.get_entity(
321 self
.entity_type
, self
.entity_id
, self
._history
_index
- 1,
325 """Return a copy of this object at its current state in the model.
327 Returns self if this object is already the latest.
329 The returned object is always "connected", i.e. receives
330 live updates from the model.
333 if self
._history
_index
== -1:
336 return self
.model
.state
.get_entity(self
.entity_type
, self
.entity_id
)
340 def __init__(self
, loop
=None):
341 """Instantiate a new connected Model.
343 :param loop: an asyncio event loop
346 self
.loop
= loop
or asyncio
.get_event_loop()
347 self
.connection
= None
348 self
.observers
= weakref
.WeakValueDictionary()
349 self
.state
= ModelState(self
)
351 self
._watcher
_task
= None
352 self
._watch
_shutdown
= asyncio
.Event(loop
=loop
)
353 self
._watch
_received
= asyncio
.Event(loop
=loop
)
354 self
._charmstore
= CharmStore(self
.loop
)
356 async def connect(self
, *args
, **kw
):
357 """Connect to an arbitrary Juju model.
359 args and kw are passed through to Connection.connect()
362 self
.connection
= await connection
.Connection
.connect(*args
, **kw
)
363 await self
._after
_connect
()
365 async def connect_current(self
):
366 """Connect to the current Juju model.
369 self
.connection
= await connection
.Connection
.connect_current()
370 await self
._after
_connect
()
372 async def connect_model(self
, model_name
):
373 """Connect to a specific Juju model by name.
375 :param model_name: Format [controller:][user/]model
378 self
.connection
= await connection
.Connection
.connect_model(model_name
)
379 await self
._after
_connect
()
381 async def _after_connect(self
):
382 """Run initialization steps after connecting to websocket.
386 await self
._watch
_received
.wait()
387 await self
.get_info()
389 async def disconnect(self
):
390 """Shut down the watcher task and close websockets.
393 self
._stop
_watching
()
394 if self
.connection
and self
.connection
.is_open
:
395 await self
._watch
_shutdown
.wait()
396 log
.debug('Closing model connection')
397 await self
.connection
.close()
398 self
.connection
= None
400 def all_units_idle(self
):
401 """Return True if all units are idle.
404 for unit
in self
.units
.values():
405 unit_status
= unit
.data
['agent-status']['current']
406 if unit_status
!= 'idle':
410 async def reset(self
, force
=False):
411 """Reset the model to a clean state.
413 :param bool force: Force-terminate machines.
415 This returns only after the model has reached a clean state. "Clean"
416 means no applications or machines exist in the model.
419 log
.debug('Resetting model')
420 for app
in self
.applications
.values():
422 for machine
in self
.machines
.values():
423 await machine
.destroy(force
=force
)
424 await self
.block_until(
425 lambda: len(self
.machines
) == 0
428 async def block_until(self
, *conditions
, timeout
=None):
429 """Return only after all conditions are true.
433 while not all(c() for c
in conditions
):
434 await asyncio
.sleep(0)
435 await asyncio
.wait_for(_block(), timeout
)
438 def applications(self
):
439 """Return a map of application-name:Application for all applications
440 currently in the model.
443 return self
.state
.applications
447 """Return a map of machine-id:Machine for all machines currently in
451 return self
.state
.machines
455 """Return a map of unit-id:Unit for all units currently in
459 return self
.state
.units
461 async def get_info(self
):
462 """Return a client.ModelInfo object for this Model.
464 Retrieves latest info for this Model from the api server. The
465 return value is cached on the Model.info attribute so that the
466 valued may be accessed again without another api call, if
469 This method is called automatically when the Model is connected,
470 resulting in Model.info being initialized without requiring an
471 explicit call to this method.
474 facade
= client
.ClientFacade()
475 facade
.connect(self
.connection
)
477 self
.info
= await facade
.ModelInfo()
478 log
.debug('Got ModelInfo: %s', vars(self
.info
))
483 self
, callable_
, entity_type
=None, action
=None, entity_id
=None,
485 """Register an "on-model-change" callback
487 Once the model is connected, ``callable_``
488 will be called each time the model changes. ``callable_`` should
489 be Awaitable and accept the following positional arguments:
491 delta - An instance of :class:`juju.delta.EntityDelta`
492 containing the raw delta data recv'd from the Juju
495 old_obj - If the delta modifies an existing object in the model,
496 old_obj will be a copy of that object, as it was before the
497 delta was applied. Will be None if the delta creates a new
500 new_obj - A copy of the new or updated object, after the delta
501 is applied. Will be None if the delta removes an entity
504 model - The :class:`Model` itself.
506 Events for which ``callable_`` is called can be specified by passing
507 entity_type, action, and/or entitiy_id filter criteria, e.g.::
511 entity_type='application', action='add', entity_id='ubuntu')
513 For more complex filtering conditions, pass a predicate function. It
514 will be called with a delta as its only argument. If the predicate
515 function returns True, the ``callable_`` will be called.
518 observer
= _Observer(
519 callable_
, entity_type
, action
, entity_id
, predicate
)
520 self
.observers
[observer
] = callable_
523 """Start an asynchronous watch against this model.
525 See :meth:`add_observer` to register an onchange callback.
528 async def _start_watch():
529 self
._watch
_shutdown
.clear()
531 allwatcher
= watcher
.AllWatcher()
532 self
._watch
_conn
= await self
.connection
.clone()
533 allwatcher
.connect(self
._watch
_conn
)
535 results
= await allwatcher
.Next()
536 for delta
in results
.deltas
:
537 delta
= get_entity_delta(delta
)
538 old_obj
, new_obj
= self
.state
.apply_delta(delta
)
539 # XXX: Might not want to shield at this level
540 # We are shielding because when the watcher is
541 # canceled (on disconnect()), we don't want all of
542 # its children (every observer callback) to be
543 # canceled with it. So we shield them. But this means
544 # they can *never* be canceled.
545 await asyncio
.shield(
546 self
._notify
_observers
(delta
, old_obj
, new_obj
))
547 self
._watch
_received
.set()
548 except CancelledError
:
549 log
.debug('Closing watcher connection')
550 await self
._watch
_conn
.close()
551 self
._watch
_shutdown
.set()
552 self
._watch
_conn
= None
554 log
.debug('Starting watcher task')
555 self
._watcher
_task
= self
.loop
.create_task(_start_watch())
557 def _stop_watching(self
):
558 """Stop the asynchronous watch against this model.
561 log
.debug('Stopping watcher task')
562 if self
._watcher
_task
:
563 self
._watcher
_task
.cancel()
565 async def _notify_observers(self
, delta
, old_obj
, new_obj
):
566 """Call observing callbacks, notifying them of a change in model state
568 :param delta: The raw change from the watcher
569 (:class:`juju.client.overrides.Delta`)
570 :param old_obj: The object in the model that this delta updates.
572 :param new_obj: The object in the model that is created or updated
573 by applying this delta.
576 if new_obj
and not old_obj
:
580 'Model changed: %s %s %s',
581 delta
.entity
, delta
.type, delta
.get_id())
583 for o
in self
.observers
:
584 if o
.cares_about(delta
):
585 asyncio
.ensure_future(o(delta
, old_obj
, new_obj
, self
))
587 async def _wait(self
, entity_type
, entity_id
, action
, predicate
=None):
589 Block the calling routine until a given action has happened to the
592 :param entity_type: The entity's type.
593 :param entity_id: The entity's id.
594 :param action: the type of action (e.g., 'add' or 'change')
595 :param predicate: optional callable that must take as an
596 argument a delta, and must return a boolean, indicating
597 whether the delta contains the specific action we're looking
598 for. For example, you might check to see whether a 'change'
599 has a 'completed' status. See the _Observer class for details.
602 q
= asyncio
.Queue(loop
=self
.loop
)
604 async def callback(delta
, old
, new
, model
):
605 await q
.put(delta
.get_id())
607 self
.add_observer(callback
, entity_type
, action
, entity_id
, predicate
)
608 entity_id
= await q
.get()
609 return self
.state
._live
_entity
_map
(entity_type
)[entity_id
]
611 async def _wait_for_new(self
, entity_type
, entity_id
=None, predicate
=None):
612 """Wait for a new object to appear in the Model and return it.
614 Waits for an object of type ``entity_type`` with id ``entity_id``.
615 If ``entity_id`` is ``None``, it will wait for the first new entity
618 This coroutine blocks until the new object appears in the model.
621 # if the entity is already in the model, just return it
622 if entity_id
in self
.state
._live
_entity
_map
(entity_type
):
623 return self
.state
._live
_entity
_map
(entity_type
)[entity_id
]
624 # if we know the entity_id, we can trigger on any action that puts
625 # the enitty into the model; otherwise, we have to watch for the
626 # next "add" action on that entity_type
627 action
= 'add' if entity_id
is None else None
628 return await self
._wait
(entity_type
, entity_id
, action
, predicate
)
630 async def wait_for_action(self
, action_id
):
631 """Given an action, wait for it to complete."""
633 if action_id
.startswith("action-"):
634 # if we've been passed action.tag, transform it into the
635 # id that the api deltas will use.
636 action_id
= action_id
[7:]
638 def predicate(delta
):
639 return delta
.data
['status'] in ('completed', 'failed')
641 return await self
._wait
('action', action_id
, 'change', predicate
)
644 self
, spec
=None, constraints
=None, disks
=None, series
=None,
646 """Start a new, empty machine and optionally a container, or add a
647 container to a machine.
649 :param str spec: Machine specification
652 (None) - starts a new machine
653 'lxc' - starts a new machine with on lxc container
654 'lxc:4' - starts a new lxc container on machine 4
655 'ssh:user@10.10.0.3' - manually provisions a machine with ssh
656 'zone=us-east-1a' - starts a machine in zone us-east-1s on AWS
657 'maas2.name' - acquire machine maas2.name on MAAS
658 :param constraints: Machine constraints
659 :type constraints: :class:`juju.Constraints`
660 :param list disks: List of disk :class:`constraints <juju.Constraints>`
661 :param str series: Series
662 :param int count: Number of machines to deploy
664 Supported container types are: lxc, lxd, kvm
666 When deploying a container to an existing machine, constraints cannot
671 add_machines
= add_machine
673 async def add_relation(self
, relation1
, relation2
):
674 """Add a relation between two applications.
676 :param str relation1: '<application>[:<relation_name>]'
677 :param str relation2: '<application>[:<relation_name>]'
680 app_facade
= client
.ApplicationFacade()
681 app_facade
.connect(self
.connection
)
684 'Adding relation %s <-> %s', relation1
, relation2
)
687 result
= await app_facade
.AddRelation([relation1
, relation2
])
688 except JujuAPIError
as e
:
689 if 'relation already exists' not in e
.message
:
692 'Relation %s <-> %s already exists', relation1
, relation2
)
693 # TODO: if relation already exists we should return the
694 # Relation ModelEntity here
697 def predicate(delta
):
699 for endpoint
in delta
.data
['endpoints']:
700 endpoints
[endpoint
['application-name']] = endpoint
['relation']
701 return endpoints
== result
.endpoints
703 return await self
._wait
_for
_new
('relation', None, predicate
)
705 def add_space(self
, name
, *cidrs
):
706 """Add a new network space.
708 Adds a new space with the given name and associates the given
709 (optional) list of existing subnet CIDRs with it.
711 :param str name: Name of the space
712 :param \*cidrs: Optional list of existing subnet CIDRs
717 def add_ssh_key(self
, key
):
718 """Add a public SSH key to this model.
720 :param str key: The public ssh key
724 add_ssh_keys
= add_ssh_key
726 def add_subnet(self
, cidr_or_id
, space
, *zones
):
727 """Add an existing subnet to this model.
729 :param str cidr_or_id: CIDR or provider ID of the existing subnet
730 :param str space: Network space with which to associate
731 :param str \*zones: Zone(s) in which the subnet resides
736 def get_backups(self
):
737 """Retrieve metadata for backups in this model.
742 def block(self
, *commands
):
743 """Add a new block to this model.
745 :param str \*commands: The commands to block. Valid values are
746 'all-changes', 'destroy-model', 'remove-object'
751 def get_blocks(self
):
752 """List blocks for this model.
757 def get_cached_images(self
, arch
=None, kind
=None, series
=None):
758 """Return a list of cached OS images.
760 :param str arch: Filter by image architecture
761 :param str kind: Filter by image kind, e.g. 'lxd'
762 :param str series: Filter by image series, e.g. 'xenial'
767 def create_backup(self
, note
=None, no_download
=False):
768 """Create a backup of this model.
770 :param str note: A note to store with the backup
771 :param bool no_download: Do not download the backup archive
772 :return str: Path to downloaded archive
777 def create_storage_pool(self
, name
, provider_type
, **pool_config
):
778 """Create or define a storage pool.
780 :param str name: Name to give the storage pool
781 :param str provider_type: Pool provider type
782 :param \*\*pool_config: key/value pool configuration pairs
788 self
, no_tail
=False, exclude_module
=None, include_module
=None,
789 include
=None, level
=None, limit
=0, lines
=10, replay
=False,
791 """Get log messages for this model.
793 :param bool no_tail: Stop after returning existing log messages
794 :param list exclude_module: Do not show log messages for these logging
796 :param list include_module: Only show log messages for these logging
798 :param list include: Only show log messages for these entities
799 :param str level: Log level to show, valid options are 'TRACE',
800 'DEBUG', 'INFO', 'WARNING', 'ERROR,
801 :param int limit: Return this many of the most recent (possibly
802 filtered) lines are shown
803 :param int lines: Yield this many of the most recent lines, and keep
805 :param bool replay: Yield the entire log, and keep yielding
806 :param list exclude: Do not show log messages for these entities
812 self
, entity_url
, application_name
=None, bind
=None, budget
=None,
813 channel
=None, config
=None, constraints
=None, force
=False,
814 num_units
=1, plan
=None, resources
=None, series
=None, storage
=None,
816 """Deploy a new service or bundle.
818 :param str entity_url: Charm or bundle url
819 :param str application_name: Name to give the service
820 :param dict bind: <charm endpoint>:<network space> pairs
821 :param dict budget: <budget name>:<limit> pairs
822 :param str channel: Charm store channel from which to retrieve
823 the charm or bundle, e.g. 'development'
824 :param dict config: Charm configuration dictionary
825 :param constraints: Service constraints
826 :type constraints: :class:`juju.Constraints`
827 :param bool force: Allow charm to be deployed to a machine running
828 an unsupported series
829 :param int num_units: Number of units to deploy
830 :param str plan: Plan under which to deploy charm
831 :param dict resources: <resource name>:<file path> pairs
832 :param str series: Series on which to deploy
833 :param dict storage: Storage constraints TODO how do these look?
834 :param str to: Placement directive, e.g.::
837 'lxc:7' - new lxc container on machine 7
838 '24/lxc/3' - lxc container 3 or machine 24
840 If None, a new machine is provisioned.
845 - application_name is required; fill this in automatically if not
847 - series is required; how do we pick a default?
852 client
.Placement(**p
) for p
in to
859 k
: client
.Constraints(**v
)
860 for k
, v
in storage
.items()
863 is_local
= not entity_url
.startswith('cs:') and \
864 os
.path
.isdir(entity_url
)
865 entity_id
= await self
.charmstore
.entityId(entity_url
) \
866 if not is_local
else entity_url
868 app_facade
= client
.ApplicationFacade()
869 client_facade
= client
.ClientFacade()
870 app_facade
.connect(self
.connection
)
871 client_facade
.connect(self
.connection
)
873 is_bundle
= ((is_local
and
874 (Path(entity_id
) / 'bundle.yaml').exists()) or
875 (not is_local
and 'bundle/' in entity_id
))
878 handler
= BundleHandler(self
)
879 await handler
.fetch_plan(entity_id
)
880 await handler
.execute_plan()
881 extant_apps
= {app
for app
in self
.applications
}
882 pending_apps
= set(handler
.applications
) - extant_apps
884 # new apps will usually be in the model by now, but if some
885 # haven't made it yet we'll need to wait on them to be added
886 await asyncio
.gather(*[
887 asyncio
.ensure_future(
888 self
.model
._wait
_for
_new
('application', app_name
))
889 for app_name
in pending_apps
891 return [app
for name
, app
in self
.applications
.items()
892 if name
in handler
.applications
]
895 'Deploying %s', entity_id
)
897 await client_facade
.AddCharm(channel
, entity_id
)
898 app
= client
.ApplicationDeploy(
899 application
=application_name
,
903 constraints
=constraints
,
904 endpoint_bindings
=bind
,
912 await app_facade
.Deploy([app
])
913 return await self
._wait
_for
_new
('application', application_name
)
916 """Terminate all machines and resources for this model.
921 async def destroy_unit(self
, *unit_names
):
922 """Destroy units by name.
925 app_facade
= client
.ApplicationFacade()
926 app_facade
.connect(self
.connection
)
929 'Destroying unit%s %s',
930 's' if len(unit_names
) == 1 else '',
931 ' '.join(unit_names
))
933 return await app_facade
.Destroy(self
.name
)
934 destroy_units
= destroy_unit
936 def get_backup(self
, archive_id
):
937 """Download a backup archive file.
939 :param str archive_id: The id of the archive to download
940 :return str: Path to the archive file
946 self
, num_controllers
=0, constraints
=None, series
=None, to
=None):
947 """Ensure sufficient controllers exist to provide redundancy.
949 :param int num_controllers: Number of controllers to make available
950 :param constraints: Constraints to apply to the controller machines
951 :type constraints: :class:`juju.Constraints`
952 :param str series: Series of the controller machines
953 :param list to: Placement directives for controller machines, e.g.::
956 'lxc:7' - new lxc container on machine 7
957 '24/lxc/3' - lxc container 3 or machine 24
959 If None, a new machine is provisioned.
964 def get_config(self
):
965 """Return the configuration settings for this model.
970 def get_constraints(self
):
971 """Return the machine constraints for this model.
976 def grant(self
, username
, acl
='read'):
977 """Grant a user access to this model.
979 :param str username: Username
980 :param str acl: Access control ('read' or 'write')
985 def import_ssh_key(self
, identity
):
986 """Add a public SSH key from a trusted indentity source to this model.
988 :param str identity: User identity in the form <lp|gh>:<username>
992 import_ssh_keys
= import_ssh_key
994 def get_machines(self
, machine
, utc
=False):
995 """Return list of machines in this model.
997 :param str machine: Machine id, e.g. '0'
998 :param bool utc: Display time as UTC in RFC3339 format
1003 def get_shares(self
):
1004 """Return list of all users with access to this model.
1009 def get_spaces(self
):
1010 """Return list of all known spaces, including associated subnets.
1015 def get_ssh_key(self
):
1016 """Return known SSH keys for this model.
1020 get_ssh_keys
= get_ssh_key
1022 def get_storage(self
, filesystem
=False, volume
=False):
1023 """Return details of storage instances.
1025 :param bool filesystem: Include filesystem storage
1026 :param bool volume: Include volume storage
1031 def get_storage_pools(self
, names
=None, providers
=None):
1032 """Return list of storage pools.
1034 :param list names: Only include pools with these names
1035 :param list providers: Only include pools for these providers
1040 def get_subnets(self
, space
=None, zone
=None):
1041 """Return list of known subnets.
1043 :param str space: Only include subnets in this space
1044 :param str zone: Only include subnets in this zone
1049 def remove_blocks(self
):
1050 """Remove all blocks from this model.
1055 def remove_backup(self
, backup_id
):
1058 :param str backup_id: The id of the backup to remove
1063 def remove_cached_images(self
, arch
=None, kind
=None, series
=None):
1064 """Remove cached OS images.
1066 :param str arch: Architecture of the images to remove
1067 :param str kind: Image kind to remove, e.g. 'lxd'
1068 :param str series: Image series to remove, e.g. 'xenial'
1073 def remove_machine(self
, *machine_ids
):
1074 """Remove a machine from this model.
1076 :param str \*machine_ids: Ids of the machines to remove
1080 remove_machines
= remove_machine
1082 def remove_ssh_key(self
, *keys
):
1083 """Remove a public SSH key(s) from this model.
1085 :param str \*keys: Keys to remove
1089 remove_ssh_keys
= remove_ssh_key
1092 self
, bootstrap
=False, constraints
=None, archive
=None,
1093 backup_id
=None, upload_tools
=False):
1094 """Restore a backup archive to a new controller.
1096 :param bool bootstrap: Bootstrap a new state machine
1097 :param constraints: Model constraints
1098 :type constraints: :class:`juju.Constraints`
1099 :param str archive: Path to backup archive to restore
1100 :param str backup_id: Id of backup to restore
1101 :param bool upload_tools: Upload tools if bootstrapping a new machine
1106 def retry_provisioning(self
):
1107 """Retry provisioning for failed machines.
1112 def revoke(self
, username
, acl
='read'):
1113 """Revoke a user's access to this model.
1115 :param str username: Username to revoke
1116 :param str acl: Access control ('read' or 'write')
1121 def run(self
, command
, timeout
=None):
1122 """Run command on all machines in this model.
1124 :param str command: The command to run
1125 :param int timeout: Time to wait before command is considered failed
1130 def set_config(self
, **config
):
1131 """Set configuration keys on this model.
1133 :param \*\*config: Config key/values
1138 def set_constraints(self
, constraints
):
1139 """Set machine constraints on this model.
1141 :param :class:`juju.Constraints` constraints: Machine constraints
1146 def get_action_output(self
, action_uuid
, wait
=-1):
1147 """Get the results of an action by ID.
1149 :param str action_uuid: Id of the action
1150 :param int wait: Time in seconds to wait for action to complete
1155 def get_action_status(self
, uuid_or_prefix
=None, name
=None):
1156 """Get the status of all actions, filtered by ID, ID prefix, or action name.
1158 :param str uuid_or_prefix: Filter by action uuid or prefix
1159 :param str name: Filter by action name
1164 def get_budget(self
, budget_name
):
1165 """Get budget usage info.
1167 :param str budget_name: Name of budget
1172 def get_status(self
, filter_
=None, utc
=False):
1173 """Return the status of the model.
1175 :param str filter_: Service or unit name or wildcard ('*')
1176 :param bool utc: Display time as UTC in RFC3339 format
1183 self
, all_
=False, destination
=None, dry_run
=False, public
=False,
1184 source
=None, stream
=None, version
=None):
1185 """Copy Juju tools into this model.
1187 :param bool all_: Copy all versions, not just the latest
1188 :param str destination: Path to local destination directory
1189 :param bool dry_run: Don't do the actual copy
1190 :param bool public: Tools are for a public cloud, so generate mirrors
1192 :param str source: Path to local source directory
1193 :param str stream: Simplestreams stream for which to sync metadata
1194 :param str version: Copy a specific major.minor version
1199 def unblock(self
, *commands
):
1200 """Unblock an operation that would alter this model.
1202 :param str \*commands: The commands to unblock. Valid values are
1203 'all-changes', 'destroy-model', 'remove-object'
1208 def unset_config(self
, *keys
):
1209 """Unset configuration on this model.
1211 :param str \*keys: The keys to unset
1216 def upgrade_gui(self
):
1217 """Upgrade the Juju GUI for this model.
1223 self
, dry_run
=False, reset_previous_upgrade
=False,
1224 upload_tools
=False, version
=None):
1225 """Upgrade Juju on all machines in a model.
1227 :param bool dry_run: Don't do the actual upgrade
1228 :param bool reset_previous_upgrade: Clear the previous (incomplete)
1230 :param bool upload_tools: Upload local version of tools
1231 :param str version: Upgrade to a specific version
1236 def upload_backup(self
, archive_path
):
1237 """Store a backup archive remotely in Juju.
1239 :param str archive_path: Path to local archive
1245 def charmstore(self
):
1246 return self
._charmstore
1248 async def get_metrics(self
, *tags
):
1249 """Retrieve metrics.
1251 :param str \*tags: Tags of entities from which to retrieve metrics.
1252 No tags retrieves the metrics of all units in the model.
1254 log
.debug("Retrieving metrics for %s",
1255 ', '.join(tags
) if tags
else "all units")
1257 metrics_facade
= client
.MetricsDebugFacade()
1258 metrics_facade
.connect(self
.connection
)
1260 entities
= [client
.Entity(tag
) for tag
in tags
]
1261 metrics_result
= await metrics_facade
.GetMetrics(entities
)
1263 metrics
= collections
.defaultdict(list)
1265 for entity_metrics
in metrics_result
.results
:
1266 error
= entity_metrics
.error
1268 if "is not a valid tag" in error
:
1269 raise ValueError(error
.message
)
1271 raise Exception(error
.message
)
1273 for metric
in entity_metrics
.metrics
:
1274 metrics
[metric
.unit
].append(vars(metric
))
1279 class BundleHandler(object):
1281 Handle bundles by using the API to translate bundle YAML into a plan of
1282 steps and then dispatching each of those using the API.
1284 def __init__(self
, model
):
1286 self
.charmstore
= model
.charmstore
1288 self
.references
= {}
1289 self
._units
_by
_app
= {}
1290 for unit_name
, unit
in model
.units
.items():
1291 app_units
= self
._units
_by
_app
.setdefault(unit
.application
, [])
1292 app_units
.append(unit_name
)
1293 self
.client_facade
= client
.ClientFacade()
1294 self
.client_facade
.connect(model
.connection
)
1295 self
.app_facade
= client
.ApplicationFacade()
1296 self
.app_facade
.connect(model
.connection
)
1297 self
.ann_facade
= client
.AnnotationsFacade()
1298 self
.ann_facade
.connect(model
.connection
)
1300 async def fetch_plan(self
, entity_id
):
1301 is_local
= not entity_id
.startswith('cs:') and os
.path
.isdir(entity_id
)
1303 bundle_yaml
= (Path(entity_id
) / "bundle.yaml").read_text()
1305 bundle_yaml
= await self
.charmstore
.files(entity_id
,
1306 filename
='bundle.yaml',
1308 self
.bundle
= yaml
.safe_load(bundle_yaml
)
1309 self
.plan
= await self
.client_facade
.GetBundleChanges(bundle_yaml
)
1311 async def execute_plan(self
):
1312 for step
in self
.plan
.changes
:
1313 method
= getattr(self
, step
.method
)
1314 result
= await method(*step
.args
)
1315 self
.references
[step
.id_
] = result
1318 def applications(self
):
1319 return list(self
.bundle
['services'].keys())
1321 def resolve(self
, reference
):
1322 if reference
and reference
.startswith('$'):
1323 reference
= self
.references
[reference
[1:]]
1326 async def addCharm(self
, charm
, series
):
1328 :param charm string:
1329 Charm holds the URL of the charm to be added.
1331 :param series string:
1332 Series holds the series of the charm to be added
1333 if the charm default is not sufficient.
1335 entity_id
= await self
.charmstore
.entityId(charm
)
1336 log
.debug('Adding %s', entity_id
)
1337 await self
.client_facade
.AddCharm(None, entity_id
)
1340 async def addMachines(self
, params
=None):
1343 Dictionary specifying the machine to add. All keys are optional.
1346 series: string specifying the machine OS series.
1348 constraints: string holding machine constraints, if any. We'll
1349 parse this into the json friendly dict that the juju api
1352 container_type: string holding the type of the container (for
1353 instance ""lxc" or kvm"). It is not specified for top level
1356 parent_id: string holding a placeholder pointing to another
1357 machine change or to a unit change. This value is only
1358 specified in the case this machine is a container, in
1359 which case also ContainerType is set.
1362 params
= params
or {}
1364 if 'parent_id' in params
:
1365 params
['parent_id'] = self
.resolve(params
['parent_id'])
1367 params
['constraints'] = parse_constraints(
1368 params
.get('constraints'))
1369 params
['jobs'] = params
.get('jobs', ['JobHostUnits'])
1371 params
= client
.AddMachineParams(**params
)
1372 results
= await self
.client_facade
.AddMachines([params
])
1373 error
= results
.machines
[0].error
1375 raise ValueError("Error adding machine: %s", error
.message
)
1376 machine
= results
.machines
[0].machine
1377 log
.debug('Added new machine %s', machine
)
1380 async def addRelation(self
, endpoint1
, endpoint2
):
1382 :param endpoint1 string:
1383 :param endpoint2 string:
1384 Endpoint1 and Endpoint2 hold relation endpoints in the
1385 "application:interface" form, where the application is always a
1386 placeholder pointing to an application change, and the interface is
1387 optional. Examples are "$deploy-42:web" or just "$deploy-42".
1389 endpoints
= [endpoint1
, endpoint2
]
1390 # resolve indirect references
1391 for i
in range(len(endpoints
)):
1392 parts
= endpoints
[i
].split(':')
1393 parts
[0] = self
.resolve(parts
[0])
1394 endpoints
[i
] = ':'.join(parts
)
1396 log
.info('Relating %s <-> %s', *endpoints
)
1397 return await self
.model
.add_relation(*endpoints
)
1399 async def deploy(self
, charm
, series
, application
, options
, constraints
,
1400 storage
, endpoint_bindings
, resources
):
1402 :param charm string:
1403 Charm holds the URL of the charm to be used to deploy this
1406 :param series string:
1407 Series holds the series of the application to be deployed
1408 if the charm default is not sufficient.
1410 :param application string:
1411 Application holds the application name.
1413 :param options map[string]interface{}:
1414 Options holds application options.
1416 :param constraints string:
1417 Constraints holds the optional application constraints.
1419 :param storage map[string]string:
1420 Storage holds the optional storage constraints.
1422 :param endpoint_bindings map[string]string:
1423 EndpointBindings holds the optional endpoint bindings
1425 :param resources map[string]int:
1426 Resources identifies the revision to use for each resource
1427 of the application's charm.
1429 # resolve indirect references
1430 charm
= self
.resolve(charm
)
1431 # stringify all config values for API
1432 options
= {k
: str(v
) for k
, v
in options
.items()}
1433 # build param object
1434 app
= client
.ApplicationDeploy(
1437 application
=application
,
1439 constraints
=constraints
,
1441 endpoint_bindings
=endpoint_bindings
,
1442 resources
=resources
,
1445 log
.info('Deploying %s', charm
)
1446 await self
.app_facade
.Deploy([app
])
1447 # ensure the app is in the model for future operations
1448 await self
.model
._wait
_for
_new
('application', application
)
1451 async def addUnit(self
, application
, to
):
1453 :param application string:
1454 Application holds the application placeholder name for which a unit
1458 To holds the optional location where to add the unit, as a
1459 placeholder pointing to another unit change or to a machine change.
1461 application
= self
.resolve(application
)
1462 placement
= self
.resolve(to
)
1463 if self
._units
_by
_app
.get(application
):
1464 # enough units for this application already exist;
1465 # claim one, and carry on
1466 # NB: this should probably honor placement, but the juju client
1467 # doesn't, so we're not bothering, either
1468 unit_name
= self
._units
_by
_app
[application
].pop()
1469 log
.debug('Reusing unit %s for %s', unit_name
, application
)
1470 return self
.model
.units
[unit_name
]
1472 log
.debug('Adding new unit for %s%s', application
,
1473 ' to %s' % placement
if placement
else '')
1474 return await self
.model
.applications
[application
].add_unit(
1479 async def expose(self
, application
):
1481 :param application string:
1482 Application holds the placeholder name of the application that must
1485 application
= self
.resolve(application
)
1486 log
.info('Exposing %s', application
)
1487 return await self
.model
.applications
[application
].expose()
1489 async def setAnnotations(self
, id_
, entity_type
, annotations
):
1492 Id is the placeholder for the application or machine change
1493 corresponding to the entity to be annotated.
1495 :param entity_type EntityType:
1496 EntityType holds the type of the entity, "application" or
1499 :param annotations map[string]string:
1500 Annotations holds the annotations as key/value pairs.
1502 entity_id
= self
.resolve(id_
)
1504 entity
= self
.model
.state
.get_entity(entity_type
, entity_id
)
1506 entity
= await self
.model
._wait
_for
_new
(entity_type
, entity_id
)
1507 return await entity
.set_annotations(annotations
)
1510 class CharmStore(object):
1512 Async wrapper around theblues.charmstore.CharmStore
1514 def __init__(self
, loop
):
1516 self
._cs
= charmstore
.CharmStore()
1518 def __getattr__(self
, name
):
1520 Wrap method calls in coroutines that use run_in_executor to make them
1523 attr
= getattr(self
._cs
, name
)
1524 if not callable(attr
):
1525 wrapper
= partial(getattr, self
._cs
, name
)
1526 setattr(self
, name
, wrapper
)
1528 async def coro(*args
, **kwargs
):
1529 method
= partial(attr
, *args
, **kwargs
)
1530 return await self
.loop
.run_in_executor(None, method
)
1531 setattr(self
, name
, coro
)