5d436fd8209acc14cd89447ae42647d0d098e058
6 from concurrent
.futures
import CancelledError
7 from functools
import partial
9 from theblues
import charmstore
11 from .client
import client
12 from .client
import watcher
13 from .client
import connection
14 from .delta
import get_entity_delta
15 from .delta
import get_entity_class
16 from .exceptions
import DeadEntityException
17 from .errors
import JujuAPIError
19 log
= logging
.getLogger(__name__
)
22 class _Observer(object):
23 """Wrapper around an observer callable.
25 This wrapper allows filter criteria to be associated with the
26 callable so that it's only called for changes that meet the criteria.
29 def __init__(self
, callable_
, entity_type
, action
, entity_id
):
30 self
.callable_
= callable_
31 self
.entity_type
= entity_type
33 self
.entity_id
= entity_id
35 if not self
.entity_id
.startswith('^'):
36 self
.entity_id
= '^' + self
.entity_id
37 if not self
.entity_id
.endswith('$'):
40 async def __call__(self
, delta
, old
, new
, model
):
41 await self
.callable_(delta
, old
, new
, model
)
43 def cares_about(self
, entity_type
, action
, entity_id
):
44 """Return True if this observer "cares about" (i.e. wants to be
45 called) for a change matching the entity_type, action, and
49 if (self
.entity_id
and entity_id
and
50 not re
.match(self
.entity_id
, str(entity_id
))):
53 if self
.entity_type
and self
.entity_type
!= entity_type
:
56 if self
.action
and self
.action
!= action
:
62 class ModelObserver(object):
63 async def __call__(self
, delta
, old
, new
, model
):
64 handler_name
= 'on_{}_{}'.format(delta
.entity
, delta
.type)
65 method
= getattr(self
, handler_name
, self
.on_change
)
66 await method(delta
, old
, new
, model
)
68 async def on_change(self
, delta
, old
, new
, model
):
72 class ModelState(object):
73 """Holds the state of the model, including the delta history of all
74 entities in the model.
77 def __init__(self
, model
):
84 def _live_entity_map(self
, entity_type
):
85 """Return an id:Entity map of all the living entities of
90 entity_id
: self
.get_entity(entity_type
, entity_id
)
91 for entity_id
, history
in self
.state
.get(entity_type
, {}).items()
92 if history
[-1] is not None
96 def applications(self
):
97 """Return a map of application-name:Application for all applications
98 currently in the model.
101 return self
._live
_entity
_map
('application')
105 """Return a map of machine-id:Machine for all machines currently in
109 return self
._live
_entity
_map
('machine')
113 """Return a map of unit-id:Unit for all units currently in
117 return self
._live
_entity
_map
('unit')
119 def entity_history(self
, entity_type
, entity_id
):
120 """Return the history deque for an entity.
123 return self
.state
[entity_type
][entity_id
]
125 def entity_data(self
, entity_type
, entity_id
, history_index
):
126 """Return the data dict for an entity at a specific index of its
130 return self
.entity_history(entity_type
, entity_id
)[history_index
]
132 def apply_delta(self
, delta
):
133 """Apply delta to our state and return a copy of the
134 affected object as it was before and after the update, e.g.:
136 old_obj, new_obj = self.apply_delta(delta)
138 old_obj may be None if the delta is for the creation of a new object,
139 e.g. a new application or unit is deployed.
141 new_obj will never be None, but may be dead (new_obj.dead == True)
142 if the object was deleted as a result of the delta being applied.
147 .setdefault(delta
.entity
, {})
148 .setdefault(delta
.get_id(), collections
.deque())
151 history
.append(delta
.data
)
152 if delta
.type == 'remove':
155 entity
= self
.get_entity(delta
.entity
, delta
.get_id())
156 return entity
.previous(), entity
159 self
, entity_type
, entity_id
, history_index
=-1, connected
=True):
160 """Return an object instance representing the entity created or
166 'Getting %s:%s at index %s',
167 entity_type, entity_id, history_index)
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_current(self
):
353 """Connect to the current Juju model.
356 self
.connection
= await connection
.Connection
.connect_current()
358 await self
._watch
_received
.wait()
360 async def disconnect(self
):
361 """Shut down the watcher task and close websockets.
364 self
._stop
_watching
()
365 if self
.connection
and self
.connection
.is_open
:
366 await self
._watch
_shutdown
.wait()
367 log
.debug('Closing model connection')
368 await self
.connection
.close()
369 self
.connection
= None
371 def all_units_idle(self
):
372 """Return True if all units are idle.
375 for unit
in self
.units
.values():
376 unit_status
= unit
.data
['agent-status']['current']
377 if unit_status
!= 'idle':
381 async def reset(self
, force
=False):
382 """Reset the model to a clean state.
384 :param bool force: Force-terminate machines.
386 This returns only after the model has reached a clean state. "Clean"
387 means no applications or machines exist in the model.
390 log
.debug('Resetting model')
391 for app
in self
.applications
.values():
393 for machine
in self
.machines
.values():
394 await machine
.destroy(force
=force
)
395 await self
.block_until(
396 lambda: len(self
.machines
) == 0
400 async def block_until(self
, *conditions
, timeout
=None):
401 """Return only after all conditions are true.
405 while not all(c() for c
in conditions
):
406 await asyncio
.sleep(0)
407 await asyncio
.wait_for(_block(), timeout
)
410 def applications(self
):
411 """Return a map of application-name:Application for all applications
412 currently in the model.
415 return self
.state
.applications
419 """Return a map of machine-id:Machine for all machines currently in
423 return self
.state
.machines
427 """Return a map of unit-id:Unit for all units currently in
431 return self
.state
.units
434 self
, callable_
, entity_type
=None, action
=None, entity_id
=None):
435 """Register an "on-model-change" callback
437 Once the model is connected, ``callable_``
438 will be called each time the model changes. callable_ should
439 be Awaitable and accept the following positional arguments:
441 delta - An instance of :class:`juju.delta.EntityDelta`
442 containing the raw delta data recv'd from the Juju
445 old_obj - If the delta modifies an existing object in the model,
446 old_obj will be a copy of that object, as it was before the
447 delta was applied. Will be None if the delta creates a new
450 new_obj - A copy of the new or updated object, after the delta
451 is applied. Will be None if the delta removes an entity
454 model - The :class:`Model` itself.
456 Events for which ``callable_`` is called can be specified by passing
457 entity_type, action, and/or id_ filter criteria, e.g.:
460 myfunc, entity_type='application', action='add', id_='ubuntu')
463 observer
= _Observer(callable_
, entity_type
, action
, entity_id
)
464 self
.observers
[observer
] = callable_
467 """Start an asynchronous watch against this model.
469 See :meth:`add_observer` to register an onchange callback.
472 async def _start_watch():
473 self
._watch
_shutdown
.clear()
475 allwatcher
= watcher
.AllWatcher()
476 self
._watch
_conn
= await self
.connection
.clone()
477 allwatcher
.connect(self
._watch
_conn
)
479 results
= await allwatcher
.Next()
480 for delta
in results
.deltas
:
481 delta
= get_entity_delta(delta
)
482 old_obj
, new_obj
= self
.state
.apply_delta(delta
)
483 # XXX: Might not want to shield at this level
484 # We are shielding because when the watcher is
485 # canceled (on disconnect()), we don't want all of
486 # its children (every observer callback) to be
487 # canceled with it. So we shield them. But this means
488 # they can *never* be canceled.
489 await asyncio
.shield(
490 self
._notify
_observers
(delta
, old_obj
, new_obj
))
491 self
._watch
_received
.set()
492 except CancelledError
:
493 log
.debug('Closing watcher connection')
494 await self
._watch
_conn
.close()
495 self
._watch
_shutdown
.set()
496 self
._watch
_conn
= None
498 log
.debug('Starting watcher task')
499 self
._watcher
_task
= self
.loop
.create_task(_start_watch())
501 def _stop_watching(self
):
502 """Stop the asynchronous watch against this model.
505 log
.debug('Stopping watcher task')
506 if self
._watcher
_task
:
507 self
._watcher
_task
.cancel()
509 async def _notify_observers(self
, delta
, old_obj
, new_obj
):
510 """Call observing callbacks, notifying them of a change in model state
512 :param delta: The raw change from the watcher
513 (:class:`juju.client.overrides.Delta`)
514 :param old_obj: The object in the model that this delta updates.
516 :param new_obj: The object in the model that is created or updated
517 by applying this delta.
524 'Model changed: %s %s %s',
525 delta
.entity
, delta
.type, delta
.get_id())
527 for o
in self
.observers
:
528 if o
.cares_about(delta
.entity
, delta
.type, delta
.get_id()):
529 asyncio
.ensure_future(o(delta
, old_obj
, new_obj
, self
))
531 async def _wait_for_new(self
, entity_type
, entity_id
):
532 """Wait for a new object to appear in the Model and return it.
534 Waits for an object of type ``entity_type`` with id ``entity_id``.
536 This coroutine blocks until the new object appears in the model.
539 entity_added
= asyncio
.Event(loop
=self
.loop
)
541 async def callback(delta
, old
, new
, model
):
543 self
.add_observer(callback
, entity_type
, 'add', entity_id
)
544 await entity_added
.wait()
545 return self
.state
._live
_entity
_map
(entity_type
)[entity_id
]
548 self
, spec
=None, constraints
=None, disks
=None, series
=None,
550 """Start a new, empty machine and optionally a container, or add a
551 container to a machine.
553 :param str spec: Machine specification
556 (None) - starts a new machine
557 'lxc' - starts a new machine with on lxc container
558 'lxc:4' - starts a new lxc container on machine 4
559 'ssh:user@10.10.0.3' - manually provisions a machine with ssh
560 'zone=us-east-1a' - starts a machine in zone us-east-1s on AWS
561 'maas2.name' - acquire machine maas2.name on MAAS
562 :param constraints: Machine constraints
563 :type constraints: :class:`juju.Constraints`
564 :param list disks: List of disk :class:`constraints <juju.Constraints>`
565 :param str series: Series
566 :param int count: Number of machines to deploy
568 Supported container types are: lxc, lxd, kvm
570 When deploying a container to an existing machine, constraints cannot
575 add_machines
= add_machine
577 async def add_relation(self
, relation1
, relation2
):
578 """Add a relation between two applications.
580 :param str relation1: '<application>[:<relation_name>]'
581 :param str relation2: '<application>[:<relation_name>]'
584 app_facade
= client
.ApplicationFacade()
585 app_facade
.connect(self
.connection
)
588 'Adding relation %s <-> %s', relation1
, relation2
)
590 return await app_facade
.AddRelation([relation1
, relation2
])
592 def add_space(self
, name
, *cidrs
):
593 """Add a new network space.
595 Adds a new space with the given name and associates the given
596 (optional) list of existing subnet CIDRs with it.
598 :param str name: Name of the space
599 :param \*cidrs: Optional list of existing subnet CIDRs
604 def add_ssh_key(self
, key
):
605 """Add a public SSH key to this model.
607 :param str key: The public ssh key
611 add_ssh_keys
= add_ssh_key
613 def add_subnet(self
, cidr_or_id
, space
, *zones
):
614 """Add an existing subnet to this model.
616 :param str cidr_or_id: CIDR or provider ID of the existing subnet
617 :param str space: Network space with which to associate
618 :param str \*zones: Zone(s) in which the subnet resides
623 def get_backups(self
):
624 """Retrieve metadata for backups in this model.
629 def block(self
, *commands
):
630 """Add a new block to this model.
632 :param str \*commands: The commands to block. Valid values are
633 'all-changes', 'destroy-model', 'remove-object'
638 def get_blocks(self
):
639 """List blocks for this model.
644 def get_cached_images(self
, arch
=None, kind
=None, series
=None):
645 """Return a list of cached OS images.
647 :param str arch: Filter by image architecture
648 :param str kind: Filter by image kind, e.g. 'lxd'
649 :param str series: Filter by image series, e.g. 'xenial'
654 def create_backup(self
, note
=None, no_download
=False):
655 """Create a backup of this model.
657 :param str note: A note to store with the backup
658 :param bool no_download: Do not download the backup archive
659 :return str: Path to downloaded archive
664 def create_storage_pool(self
, name
, provider_type
, **pool_config
):
665 """Create or define a storage pool.
667 :param str name: Name to give the storage pool
668 :param str provider_type: Pool provider type
669 :param \*\*pool_config: key/value pool configuration pairs
675 self
, no_tail
=False, exclude_module
=None, include_module
=None,
676 include
=None, level
=None, limit
=0, lines
=10, replay
=False,
678 """Get log messages for this model.
680 :param bool no_tail: Stop after returning existing log messages
681 :param list exclude_module: Do not show log messages for these logging
683 :param list include_module: Only show log messages for these logging
685 :param list include: Only show log messages for these entities
686 :param str level: Log level to show, valid options are 'TRACE',
687 'DEBUG', 'INFO', 'WARNING', 'ERROR,
688 :param int limit: Return this many of the most recent (possibly
689 filtered) lines are shown
690 :param int lines: Yield this many of the most recent lines, and keep
692 :param bool replay: Yield the entire log, and keep yielding
693 :param list exclude: Do not show log messages for these entities
699 self
, entity_url
, service_name
=None, bind
=None, budget
=None,
700 channel
=None, config
=None, constraints
=None, force
=False,
701 num_units
=1, plan
=None, resources
=None, series
=None, storage
=None,
703 """Deploy a new service or bundle.
705 :param str entity_url: Charm or bundle url
706 :param str service_name: Name to give the service
707 :param dict bind: <charm endpoint>:<network space> pairs
708 :param dict budget: <budget name>:<limit> pairs
709 :param str channel: Charm store channel from which to retrieve
710 the charm or bundle, e.g. 'development'
711 :param dict config: Charm configuration dictionary
712 :param constraints: Service constraints
713 :type constraints: :class:`juju.Constraints`
714 :param bool force: Allow charm to be deployed to a machine running
715 an unsupported series
716 :param int num_units: Number of units to deploy
717 :param str plan: Plan under which to deploy charm
718 :param dict resources: <resource name>:<file path> pairs
719 :param str series: Series on which to deploy
720 :param dict storage: Storage constraints TODO how do these look?
721 :param str to: Placement directive, e.g.::
724 'lxc:7' - new lxc container on machine 7
725 '24/lxc/3' - lxc container 3 or machine 24
727 If None, a new machine is provisioned.
732 - entity_url must have a revision; look up latest automatically if
733 not provided by caller
734 - service_name is required; fill this in automatically if not
736 - series is required; how do we pick a default?
740 constraints
= client
.Value(**constraints
)
744 client
.Placement(**p
) for p
in to
751 k
: client
.Constraints(**v
)
752 for k
, v
in storage
.items()
755 entity_id
= await self
.charmstore
.entityId(entity_url
)
757 app_facade
= client
.ApplicationFacade()
758 client_facade
= client
.ClientFacade()
759 app_facade
.connect(self
.connection
)
760 client_facade
.connect(self
.connection
)
762 if 'bundle/' in entity_id
:
763 handler
= BundleHandler(self
)
764 await handler
.fetch_plan(entity_id
)
765 await handler
.execute_plan()
768 'Deploying %s', entity_id
)
770 await client_facade
.AddCharm(channel
, entity_id
)
771 app
= client
.ApplicationDeploy(
772 application
=service_name
,
776 constraints
=constraints
,
777 endpoint_bindings
=bind
,
785 await app_facade
.Deploy([app
])
786 return await self
._wait
_for
_new
('application', service_name
)
789 """Terminate all machines and resources for this model.
794 def get_backup(self
, archive_id
):
795 """Download a backup archive file.
797 :param str archive_id: The id of the archive to download
798 :return str: Path to the archive file
804 self
, num_controllers
=0, constraints
=None, series
=None, to
=None):
805 """Ensure sufficient controllers exist to provide redundancy.
807 :param int num_controllers: Number of controllers to make available
808 :param constraints: Constraints to apply to the controller machines
809 :type constraints: :class:`juju.Constraints`
810 :param str series: Series of the controller machines
811 :param list to: Placement directives for controller machines, e.g.::
814 'lxc:7' - new lxc container on machine 7
815 '24/lxc/3' - lxc container 3 or machine 24
817 If None, a new machine is provisioned.
822 def get_config(self
):
823 """Return the configuration settings for this model.
828 def get_constraints(self
):
829 """Return the machine constraints for this model.
834 def grant(self
, username
, acl
='read'):
835 """Grant a user access to this model.
837 :param str username: Username
838 :param str acl: Access control ('read' or 'write')
843 def import_ssh_key(self
, identity
):
844 """Add a public SSH key from a trusted indentity source to this model.
846 :param str identity: User identity in the form <lp|gh>:<username>
850 import_ssh_keys
= import_ssh_key
852 def get_machines(self
, machine
, utc
=False):
853 """Return list of machines in this model.
855 :param str machine: Machine id, e.g. '0'
856 :param bool utc: Display time as UTC in RFC3339 format
861 def get_shares(self
):
862 """Return list of all users with access to this model.
867 def get_spaces(self
):
868 """Return list of all known spaces, including associated subnets.
873 def get_ssh_key(self
):
874 """Return known SSH keys for this model.
878 get_ssh_keys
= get_ssh_key
880 def get_storage(self
, filesystem
=False, volume
=False):
881 """Return details of storage instances.
883 :param bool filesystem: Include filesystem storage
884 :param bool volume: Include volume storage
889 def get_storage_pools(self
, names
=None, providers
=None):
890 """Return list of storage pools.
892 :param list names: Only include pools with these names
893 :param list providers: Only include pools for these providers
898 def get_subnets(self
, space
=None, zone
=None):
899 """Return list of known subnets.
901 :param str space: Only include subnets in this space
902 :param str zone: Only include subnets in this zone
907 def remove_blocks(self
):
908 """Remove all blocks from this model.
913 def remove_backup(self
, backup_id
):
916 :param str backup_id: The id of the backup to remove
921 def remove_cached_images(self
, arch
=None, kind
=None, series
=None):
922 """Remove cached OS images.
924 :param str arch: Architecture of the images to remove
925 :param str kind: Image kind to remove, e.g. 'lxd'
926 :param str series: Image series to remove, e.g. 'xenial'
931 def remove_machine(self
, *machine_ids
):
932 """Remove a machine from this model.
934 :param str \*machine_ids: Ids of the machines to remove
938 remove_machines
= remove_machine
940 def remove_ssh_key(self
, *keys
):
941 """Remove a public SSH key(s) from this model.
943 :param str \*keys: Keys to remove
947 remove_ssh_keys
= remove_ssh_key
950 self
, bootstrap
=False, constraints
=None, archive
=None,
951 backup_id
=None, upload_tools
=False):
952 """Restore a backup archive to a new controller.
954 :param bool bootstrap: Bootstrap a new state machine
955 :param constraints: Model constraints
956 :type constraints: :class:`juju.Constraints`
957 :param str archive: Path to backup archive to restore
958 :param str backup_id: Id of backup to restore
959 :param bool upload_tools: Upload tools if bootstrapping a new machine
964 def retry_provisioning(self
):
965 """Retry provisioning for failed machines.
970 def revoke(self
, username
, acl
='read'):
971 """Revoke a user's access to this model.
973 :param str username: Username to revoke
974 :param str acl: Access control ('read' or 'write')
979 def run(self
, command
, timeout
=None):
980 """Run command on all machines in this model.
982 :param str command: The command to run
983 :param int timeout: Time to wait before command is considered failed
988 def set_config(self
, **config
):
989 """Set configuration keys on this model.
991 :param \*\*config: Config key/values
996 def set_constraints(self
, constraints
):
997 """Set machine constraints on this model.
999 :param :class:`juju.Constraints` constraints: Machine constraints
1004 def get_action_output(self
, action_uuid
, wait
=-1):
1005 """Get the results of an action by ID.
1007 :param str action_uuid: Id of the action
1008 :param int wait: Time in seconds to wait for action to complete
1013 def get_action_status(self
, uuid_or_prefix
=None, name
=None):
1014 """Get the status of all actions, filtered by ID, ID prefix, or action name.
1016 :param str uuid_or_prefix: Filter by action uuid or prefix
1017 :param str name: Filter by action name
1022 def get_budget(self
, budget_name
):
1023 """Get budget usage info.
1025 :param str budget_name: Name of budget
1030 def get_status(self
, filter_
=None, utc
=False):
1031 """Return the status of the model.
1033 :param str filter_: Service or unit name or wildcard ('*')
1034 :param bool utc: Display time as UTC in RFC3339 format
1041 self
, all_
=False, destination
=None, dry_run
=False, public
=False,
1042 source
=None, stream
=None, version
=None):
1043 """Copy Juju tools into this model.
1045 :param bool all_: Copy all versions, not just the latest
1046 :param str destination: Path to local destination directory
1047 :param bool dry_run: Don't do the actual copy
1048 :param bool public: Tools are for a public cloud, so generate mirrors
1050 :param str source: Path to local source directory
1051 :param str stream: Simplestreams stream for which to sync metadata
1052 :param str version: Copy a specific major.minor version
1057 def unblock(self
, *commands
):
1058 """Unblock an operation that would alter this model.
1060 :param str \*commands: The commands to unblock. Valid values are
1061 'all-changes', 'destroy-model', 'remove-object'
1066 def unset_config(self
, *keys
):
1067 """Unset configuration on this model.
1069 :param str \*keys: The keys to unset
1074 def upgrade_gui(self
):
1075 """Upgrade the Juju GUI for this model.
1081 self
, dry_run
=False, reset_previous_upgrade
=False,
1082 upload_tools
=False, version
=None):
1083 """Upgrade Juju on all machines in a model.
1085 :param bool dry_run: Don't do the actual upgrade
1086 :param bool reset_previous_upgrade: Clear the previous (incomplete)
1088 :param bool upload_tools: Upload local version of tools
1089 :param str version: Upgrade to a specific version
1094 def upload_backup(self
, archive_path
):
1095 """Store a backup archive remotely in Juju.
1097 :param str archive_path: Path to local archive
1103 def charmstore(self
):
1104 return self
._charmstore
1107 class BundleHandler(object):
1109 Handle bundles by using the API to translate bundle YAML into a plan of
1110 steps and then dispatching each of those using the API.
1112 def __init__(self
, model
):
1114 self
.charmstore
= model
.charmstore
1116 self
.references
= {}
1117 self
._units
_by
_app
= {}
1118 for unit_name
, unit
in model
.units
.items():
1119 app_units
= self
._units
_by
_app
.setdefault(unit
.application
, [])
1120 app_units
.append(unit_name
)
1121 self
.client_facade
= client
.ClientFacade()
1122 self
.client_facade
.connect(model
.connection
)
1123 self
.app_facade
= client
.ApplicationFacade()
1124 self
.app_facade
.connect(model
.connection
)
1125 self
.ann_facade
= client
.AnnotationsFacade()
1126 self
.ann_facade
.connect(model
.connection
)
1128 async def fetch_plan(self
, entity_id
):
1129 yaml
= await self
.charmstore
.files(entity_id
,
1130 filename
='bundle.yaml',
1132 self
.plan
= await self
.client_facade
.GetBundleChanges(yaml
)
1134 async def execute_plan(self
):
1135 for step
in self
.plan
.changes
:
1136 method
= getattr(self
, step
.method
)
1137 result
= await method(*step
.args
)
1138 self
.references
[step
.id_
] = result
1140 def resolve(self
, reference
):
1141 if reference
and reference
.startswith('$'):
1142 reference
= self
.references
[reference
[1:]]
1145 async def addCharm(self
, charm
, series
):
1147 :param charm string:
1148 Charm holds the URL of the charm to be added.
1150 :param series string:
1151 Series holds the series of the charm to be added
1152 if the charm default is not sufficient.
1154 entity_id
= await self
.charmstore
.entityId(charm
)
1155 log
.debug('Adding %s', entity_id
)
1156 await self
.client_facade
.AddCharm(None, entity_id
)
1159 async def addMachines(self
, series
, constraints
, container_type
,
1162 :param series string:
1163 Series holds the optional machine OS series.
1165 :param constraints string:
1166 Constraints holds the optional machine constraints.
1168 :param Container_type string:
1169 ContainerType optionally holds the type of the container (for
1170 instance ""lxc" or kvm"). It is not specified for top level
1173 :param parent_id string:
1174 ParentId optionally holds a placeholder pointing to another machine
1175 change or to a unit change. This value is only specified in the
1176 case this machine is a container, in which case also ContainerType
1179 params
= client
.AddMachineParams(
1181 constraints
=constraints
,
1182 container_type
=container_type
,
1183 parent_id
=self
.resolve(parent_id
),
1185 results
= await self
.client_facade
.AddMachines(params
)
1186 log
.debug('Added new machine %s', results
[0].machine
)
1187 return results
[0].machine
1189 async def addRelation(self
, endpoint1
, endpoint2
):
1191 :param endpoint1 string:
1192 :param endpoint2 string:
1193 Endpoint1 and Endpoint2 hold relation endpoints in the
1194 "application:interface" form, where the application is always a
1195 placeholder pointing to an application change, and the interface is
1196 optional. Examples are "$deploy-42:web" or just "$deploy-42".
1198 endpoints
= [endpoint1
, endpoint2
]
1199 # resolve indirect references
1200 for i
in range(len(endpoints
)):
1201 parts
= endpoints
[i
].split(':')
1202 parts
[0] = self
.resolve(parts
[0])
1203 endpoints
[i
] = ':'.join(parts
)
1205 await self
.app_facade
.AddRelation(endpoints
)
1206 log
.debug('Added relation %s <-> %s', *endpoints
)
1207 except JujuAPIError
as e
:
1208 if 'relation already exists' not in e
.message
:
1210 log
.debug('Relation %s <-> %s already exists', *endpoints
)
1213 async def deploy(self
, charm
, series
, application
, options
, constraints
,
1214 storage
, endpoint_bindings
, resources
):
1216 :param charm string:
1217 Charm holds the URL of the charm to be used to deploy this
1220 :param series string:
1221 Series holds the series of the application to be deployed
1222 if the charm default is not sufficient.
1224 :param application string:
1225 Application holds the application name.
1227 :param options map[string]interface{}:
1228 Options holds application options.
1230 :param constraints string:
1231 Constraints holds the optional application constraints.
1233 :param storage map[string]string:
1234 Storage holds the optional storage constraints.
1236 :param endpoint_bindings map[string]string:
1237 EndpointBindings holds the optional endpoint bindings
1239 :param resources map[string]int:
1240 Resources identifies the revision to use for each resource
1241 of the application's charm.
1243 # resolve indirect references
1244 charm
= self
.resolve(charm
)
1245 # stringify all config values for API
1246 options
= {k
: str(v
) for k
, v
in options
.items()}
1247 # build param object
1248 app
= client
.ApplicationDeploy(
1251 application
=application
,
1253 constraints
=constraints
,
1255 endpoint_bindings
=endpoint_bindings
,
1256 resources
=resources
,
1259 log
.debug('Deploying %s', charm
)
1260 await self
.app_facade
.Deploy([app
])
1263 async def addUnit(self
, application
, to
):
1265 :param application string:
1266 Application holds the application placeholder name for which a unit
1270 To holds the optional location where to add the unit, as a
1271 placeholder pointing to another unit change or to a machine change.
1273 application
= self
.resolve(application
)
1274 placement
= self
.resolve(to
)
1275 if self
._units
_by
_app
.get(application
):
1276 # enough units for this application already exist;
1277 # claim one, and carry on
1278 # NB: this should probably honor placement, but the juju client
1279 # doesn't, so we're not bothering, either
1280 unit_name
= self
._units
_by
_app
[application
].pop()
1281 log
.debug('Reusing unit %s for %s', unit_name
, application
)
1283 log
.debug('Adding unit of %s%s',
1285 (' to %s' % placement
) if placement
else '')
1286 result
= await self
.app_facade
.AddUnits(
1287 application
=application
,
1288 placement
=placement
,
1291 return result
.units
[0]
1293 async def expose(self
, application
):
1295 :param application string:
1296 Application holds the placeholder name of the application that must
1299 application
= self
.resolve(application
)
1300 log
.debug('Exposing %s', application
)
1301 await self
.app_facade
.Expose(application
)
1304 async def setAnnotations(self
, id_
, entity_type
, annotations
):
1307 Id is the placeholder for the application or machine change
1308 corresponding to the entity to be annotated.
1310 :param entity_type EntityType:
1311 EntityType holds the type of the entity, "application" or
1314 :param annotations map[string]string:
1315 Annotations holds the annotations as key/value pairs.
1317 entity_id
= self
.resolve(id_
)
1318 log
.debug('Updating annotations of %s', entity_id
)
1319 ann
= client
.EntityAnnotations(
1321 annotations
=annotations
,
1323 await self
.ann_facade
.Set([ann
])
1327 class CharmStore(object):
1329 Async wrapper around theblues.charmstore.CharmStore
1331 def __init__(self
, loop
):
1333 self
._cs
= charmstore
.CharmStore()
1335 def __getattr__(self
, name
):
1337 Wrap method calls in coroutines that use run_in_executor to make them
1340 attr
= getattr(self
._cs
, name
)
1341 if not callable(attr
):
1342 wrapper
= partial(getattr, self
._cs
, name
)
1343 setattr(self
, name
, wrapper
)
1345 async def coro(*args
, **kwargs
):
1346 method
= partial(attr
, *args
, **kwargs
)
1347 return await self
.loop
.run_in_executor(None, method
)
1348 setattr(self
, name
, coro
)