from .client import client
from .client import watcher
from .client import connection
+from .constraints import parse as parse_constraints
from .delta import get_entity_delta
from .delta import get_entity_class
from .exceptions import DeadEntityException
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.
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, predicate=None):
+ 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.
"""
- return await self._wait(entity_type, entity_id, 'add', predicate)
+ # 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 wait_for_action(self, action_id):
"""Given an action, wait for it to complete."""
action_id = action_id[7:]
def predicate(delta):
- return delta.data['status'] in ('completed', 'error')
+ return delta.data['status'] in ('completed', 'failed')
return await self._wait('action', action_id, 'change', predicate)
def charmstore(self):
return self._charmstore
+ async def get_metrics(self, *tags):
+ """Retrieve metrics.
+
+ :param str \*tags: Tags of entities from which to retrieve metrics.
+ No tags retrieves the metrics of all units in the model.
+ """
+ log.debug("Retrieving metrics for %s",
+ ', '.join(tags) if tags else "all units")
+
+ metrics_facade = client.MetricsDebugFacade()
+ metrics_facade.connect(self.connection)
+
+ entities = [client.Entity(tag) for tag in tags]
+ metrics_result = await metrics_facade.GetMetrics(entities)
+
+ metrics = collections.defaultdict(list)
+
+ for entity_metrics in metrics_result.results:
+ error = entity_metrics.error
+ if error:
+ if "is not a valid tag" in error:
+ raise ValueError(error.message)
+ else:
+ raise Exception(error.message)
+
+ for metric in entity_metrics.metrics:
+ metrics[metric.unit].append(metric.to_json())
+
+ return metrics
+
class BundleHandler(object):
"""
await self.client_facade.AddCharm(None, entity_id)
return entity_id
- async def addMachines(self, series, constraints, container_type,
- parent_id):
- """
- :param series string:
- Series holds the optional machine OS series.
-
- :param constraints string:
- Constraints holds the optional machine constraints.
-
- :param Container_type string:
- ContainerType optionally holds the type of the container (for
- instance ""lxc" or kvm"). It is not specified for top level
- machines.
-
- :param parent_id string:
- ParentId optionally holds a placeholder pointing to another machine
- change or to a unit change. This value is only specified in the
- case this machine is a container, in which case also ContainerType
- is set.
- """
- params = client.AddMachineParams(
- series=series,
- constraints=constraints,
- container_type=container_type,
- parent_id=self.resolve(parent_id),
- )
- results = await self.client_facade.AddMachines(params)
- log.debug('Added new machine %s', results[0].machine)
- return results[0].machine
+ async def addMachines(self, params=None):
+ """
+ :param params dict:
+ Dictionary specifying the machine to add. All keys are optional.
+ Keys include:
+
+ series: string specifying the machine OS series.
+ constraints: string holding machine constraints, if any. We'll
+ parse this into the json friendly dict that the juju api
+ expects.
+ container_type: string holding the type of the container (for
+ instance ""lxc" or kvm"). It is not specified for top level
+ machines.
+ parent_id: string holding a placeholder pointing to another
+ machine change or to a unit change. This value is only
+ specified in the case this machine is a container, in
+ which case also ContainerType is set.
+ """
+ params = params or {}
+
+ if 'parent_id' in params:
+ params['parent_id'] = self.resolve(params['parent_id'])
+
+ params['constraints'] = parse_constraints(
+ params.get('constraints'))
+ params['jobs'] = params.get('jobs', ['JobHostUnits'])
+
+ params = client.AddMachineParams(**params)
+ results = await self.client_facade.AddMachines([params])
+ error = results.machines[0].error
+ if error:
+ raise ValueError("Error adding machine: %s", error.message)
+ machine = results.machines[0].machine
+ log.debug('Added new machine %s', machine)
+ return machine
async def addRelation(self, endpoint1, endpoint2):
"""
# do the do
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):
projects / osm / N2VC.git / blobdiff