from concurrent.futures import CancelledError
from functools import partial
+import yaml
from theblues import charmstore
from .client import client
def get_entity(
self, entity_type, entity_id, history_index=-1, connected=True):
- """Return an object instance representing the entity created or
- updated by ``delta``
+ """Return an object instance for the given entity_type and id.
+
+ By default the object state matches the most recent state from
+ Juju. To get an instance of the object in an older state, pass
+ history_index, an index into the history deque for the entity.
- """
- """
- log.debug(
- 'Getting %s:%s at index %s',
- entity_type, entity_id, history_index)
"""
if history_index < 0 and history_index != -1:
self._watch_received = asyncio.Event(loop=loop)
self._charmstore = CharmStore(self.loop)
+ async def connect(self, *args, **kw):
+ """Connect to an arbitrary Juju model.
+
+ args and kw are passed through to Connection.connect()
+
+ """
+ self.connection = await connection.Connection.connect(*args, **kw)
+ self._watch()
+ await self._watch_received.wait()
+
async def connect_current(self):
"""Connect to the current Juju model.
self._watch()
await self._watch_received.wait()
+ async def connect_model(self, arg):
+ """Connect to a specific Juju model.
+ :param arg: <controller>:<user/model>
+
+ """
+ self.connection = await connection.Connection.connect_model(arg)
+ self._watch()
+ await self._watch_received.wait()
+
async def disconnect(self):
"""Shut down the watcher task and close websockets.
if o.cares_about(delta):
asyncio.ensure_future(o(delta, old_obj, new_obj, self))
- async def _wait_for_new(self, entity_type, entity_id, predicate=None):
+ async def _wait(self, entity_type, entity_id, action, predicate=None):
+ """
+ Block the calling routine until a given action has happened to the
+ given entity
+
+ :param entity_type: The entity's type.
+ :param entity_id: The entity's id.
+ :param action: the type of action (e.g., 'add' or 'change')
+ :param predicate: optional callable that must take as an
+ argument a delta, and must return a boolean, indicating
+ whether the delta contains the specific action we're looking
+ for. For example, you might check to see whether a 'change'
+ has a 'completed' status. See the _Observer class for details.
+
+ """
+ q = asyncio.Queue(loop=self.loop)
+
+ async def callback(delta, old, new, model):
+ await q.put(delta.get_id())
+
+ self.add_observer(callback, entity_type, action, entity_id, predicate)
+ entity_id = await q.get()
+ return self.state._live_entity_map(entity_type)[entity_id]
+
+ async def _wait_for_new(self, entity_type, entity_id=None, predicate=None):
"""Wait for a new object to appear in the Model and return it.
Waits for an object of type ``entity_type`` with id ``entity_id``.
+ If ``entity_id`` is ``None``, it will wait for the first new entity
+ of the correct type.
This coroutine blocks until the new object appears in the model.
"""
- entity_added = asyncio.Queue(loop=self.loop)
+ # if the entity is already in the model, just return it
+ if entity_id in self.state._live_entity_map(entity_type):
+ return self.state._live_entity_map(entity_type)[entity_id]
+ # if we know the entity_id, we can trigger on any action that puts
+ # the enitty into the model; otherwise, we have to watch for the
+ # next "add" action on that entity_type
+ action = 'add' if entity_id is None else None
+ return await self._wait(entity_type, entity_id, action, predicate)
- async def callback(delta, old, new, model):
- await entity_added.put(delta.get_id())
+ async def wait_for_action(self, action_id):
+ """Given an action, wait for it to complete."""
- self.add_observer(callback, entity_type, 'add', entity_id, predicate)
- entity_id = await entity_added.get()
- return self.state._live_entity_map(entity_type)[entity_id]
+ if action_id.startswith("action-"):
+ # if we've been passed action.tag, transform it into the
+ # id that the api deltas will use.
+ action_id = action_id[7:]
+
+ def predicate(delta):
+ return delta.data['status'] in ('completed', 'failed')
+
+ return await self._wait('action', action_id, 'change', predicate)
def add_machine(
self, spec=None, constraints=None, disks=None, series=None,
- series is required; how do we pick a default?
"""
- if constraints:
- constraints = client.Value(**constraints)
-
if to:
placement = [
client.Placement(**p) for p in to
handler = BundleHandler(self)
await handler.fetch_plan(entity_id)
await handler.execute_plan()
+ extant_apps = {app for app in self.applications}
+ pending_apps = set(handler.applications) - extant_apps
+ if pending_apps:
+ # new apps will usually be in the model by now, but if some
+ # haven't made it yet we'll need to wait on them to be added
+ await asyncio.gather(*[
+ asyncio.ensure_future(
+ self.model._wait_for_new('application', app_name))
+ for app_name in pending_apps
+ ])
+ return [app for name, app in self.applications.items()
+ if name in handler.applications]
else:
log.debug(
'Deploying %s', entity_id)
"""
pass
+ async def destroy_unit(self, *unit_names):
+ """Destroy units by name.
+
+ """
+ app_facade = client.ApplicationFacade()
+ app_facade.connect(self.connection)
+
+ log.debug(
+ 'Destroying unit%s %s',
+ 's' if len(unit_names) == 1 else '',
+ ' '.join(unit_names))
+
+ return await app_facade.Destroy(self.name)
+ destroy_units = destroy_unit
+
def get_backup(self, archive_id):
"""Download a backup archive file.
self.ann_facade.connect(model.connection)
async def fetch_plan(self, entity_id):
- yaml = await self.charmstore.files(entity_id,
- filename='bundle.yaml',
- read_file=True)
- self.plan = await self.client_facade.GetBundleChanges(yaml)
+ bundle_yaml = await self.charmstore.files(entity_id,
+ filename='bundle.yaml',
+ read_file=True)
+ self.bundle = yaml.safe_load(bundle_yaml)
+ self.plan = await self.client_facade.GetBundleChanges(bundle_yaml)
async def execute_plan(self):
for step in self.plan.changes:
result = await method(*step.args)
self.references[step.id_] = result
+ @property
+ def applications(self):
+ return list(self.bundle['services'].keys())
+
def resolve(self, reference):
if reference and reference.startswith('$'):
reference = self.references[reference[1:]]
parts[0] = self.resolve(parts[0])
endpoints[i] = ':'.join(parts)
+ log.info('Relating %s <-> %s', *endpoints)
return await self.model.add_relation(*endpoints)
async def deploy(self, charm, series, application, options, constraints,
resources=resources,
)
# do the do
- log.debug('Deploying %s', charm)
+ log.info('Deploying %s', charm)
await self.app_facade.Deploy([app])
+ # ensure the app is in the model for future operations
+ await self.model._wait_for_new('application', application)
return application
async def addUnit(self, application, to):
# doesn't, so we're not bothering, either
unit_name = self._units_by_app[application].pop()
log.debug('Reusing unit %s for %s', unit_name, application)
- return unit_name
- log.debug('Adding unit of %s%s',
- application,
- (' to %s' % placement) if placement else '')
- result = await self.app_facade.AddUnits(
- application=application,
- placement=placement,
- num_units=1,
+ return self.model.units[unit_name]
+
+ log.debug('Adding new unit for %s%s', application,
+ ' to %s' % placement if placement else '')
+ return await self.model.applications[application].add_unit(
+ count=1,
+ to=placement,
)
- return result.units[0]
async def expose(self, application):
"""
be exposed.
"""
application = self.resolve(application)
+ log.info('Exposing %s', application)
return await self.model.applications[application].expose()
async def setAnnotations(self, id_, entity_type, annotations):
Annotations holds the annotations as key/value pairs.
"""
entity_id = self.resolve(id_)
- log.debug('Updating annotations of %s', entity_id)
- ann = client.EntityAnnotations(
- entity=entity_id,
- annotations=annotations,
- )
- await self.ann_facade.Set([ann])
- return None
+ try:
+ entity = self.model.state.get_entity(entity_type, entity_id)
+ except KeyError:
+ entity = await self.model._wait_for_new(entity_type, entity_id)
+ return await entity.set_annotations(annotations)
class CharmStore(object):