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
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
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, 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 connect_model(self, arg):
- """Connect to a specific Juju model.
- :param arg: <controller>:<user/model>
+ 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):
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()
raise Exception(error.message)
for metric in entity_metrics.metrics:
- metrics[metric.unit].append(metric.to_json())
+ metrics[metric.unit].append(vars(metric))
return metrics
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)
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):
"""