import asyncio
import collections
import logging
+import os
import re
import weakref
from concurrent.futures import CancelledError
from functools import partial
+from pathlib import Path
import yaml
from theblues import charmstore
self.connection = None
self.observers = weakref.WeakValueDictionary()
self.state = ModelState(self)
+ self.info = None
self._watcher_task = None
self._watch_shutdown = asyncio.Event(loop=loop)
self._watch_received = asyncio.Event(loop=loop)
"""
self.connection = await connection.Connection.connect(*args, **kw)
- self._watch()
- await self._watch_received.wait()
+ await self._after_connect()
async def connect_current(self):
"""Connect to the current Juju model.
"""
self.connection = await connection.Connection.connect_current()
- self._watch()
- await self._watch_received.wait()
+ await self._after_connect()
- async def connect_model(self, arg):
- """Connect to a specific Juju model.
- :param arg: <controller>:<user/model>
+ async def connect_model(self, model_name):
+ """Connect to a specific Juju model by name.
+
+ :param model_name: Format [controller:][user/]model
+
+ """
+ self.connection = await connection.Connection.connect_model(model_name)
+ await self._after_connect()
+
+ async def _after_connect(self):
+ """Run initialization steps after connecting to websocket.
"""
- self.connection = await connection.Connection.connect_model(arg)
self._watch()
await self._watch_received.wait()
+ await self.get_info()
async def disconnect(self):
"""Shut down the watcher task and close websockets.
"""
return self.state.units
+ async def get_info(self):
+ """Return a client.ModelInfo object for this Model.
+
+ Retrieves latest info for this Model from the api server. The
+ return value is cached on the Model.info attribute so that the
+ valued may be accessed again without another api call, if
+ desired.
+
+ This method is called automatically when the Model is connected,
+ resulting in Model.info being initialized without requiring an
+ explicit call to this method.
+
+ """
+ facade = client.ClientFacade()
+ facade.connect(self.connection)
+
+ self.info = await facade.ModelInfo()
+ log.debug('Got ModelInfo: %s', vars(self.info))
+
+ return self.info
+
def add_observer(
self, callable_, entity_type=None, action=None, entity_id=None,
predicate=None):
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."""
for k, v in storage.items()
}
- entity_id = await self.charmstore.entityId(entity_url)
+ is_local = not entity_url.startswith('cs:') and \
+ os.path.isdir(entity_url)
+ entity_id = await self.charmstore.entityId(entity_url) \
+ if not is_local else entity_url
app_facade = client.ApplicationFacade()
client_facade = client.ClientFacade()
app_facade.connect(self.connection)
client_facade.connect(self.connection)
- if 'bundle/' in entity_id:
+ is_bundle = ((is_local and
+ (Path(entity_id) / 'bundle.yaml').exists()) or
+ (not is_local and 'bundle/' in entity_id))
+
+ if is_bundle:
handler = BundleHandler(self)
await handler.fetch_plan(entity_id)
await handler.execute_plan()
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(vars(metric))
+
+ return metrics
+
class BundleHandler(object):
"""
self.ann_facade.connect(model.connection)
async def fetch_plan(self, entity_id):
- bundle_yaml = await self.charmstore.files(entity_id,
- filename='bundle.yaml',
- read_file=True)
+ is_local = not entity_id.startswith('cs:') and os.path.isdir(entity_id)
+ if is_local:
+ bundle_yaml = (Path(entity_id) / "bundle.yaml").read_text()
+ else:
+ 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)
# 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