from functools import partial
from pathlib import Path
+import websockets
import yaml
import theblues.charmstore
import theblues.errors
-from . import tag
+from . import tag, utils
from .client import client
-from .client import watcher
from .client import connection
+from .client.client import ConfigValue
from .constraints import parse as parse_constraints, normalize_key
from .delta import get_entity_delta
from .delta import get_entity_class
class ModelObserver(object):
+ """
+ Base class for creating observers that react to changes in a model.
+ """
async def __call__(self, delta, old, new, model):
handler_name = 'on_{}_{}'.format(delta.entity, delta.type)
method = getattr(self, handler_name, self.on_change)
async def on_change(self, delta, old, new, model):
"""Generic model-change handler.
+ This should be overridden in a subclass.
+
:param delta: :class:`juju.client.overrides.Delta`
:param old: :class:`juju.model.ModelEntity`
:param new: :class:`juju.model.ModelEntity`
class Model(object):
- def __init__(self, loop=None):
+ """
+ The main API for interacting with a Juju model.
+ """
+ def __init__(self, loop=None,
+ max_frame_size=connection.Connection.DEFAULT_FRAME_SIZE):
"""Instantiate a new connected Model.
:param loop: an asyncio event loop
+ :param max_frame_size: See
+ `juju.client.connection.Connection.MAX_FRAME_SIZE`
"""
self.loop = loop or asyncio.get_event_loop()
+ self.max_frame_size = max_frame_size
self.connection = None
self.observers = weakref.WeakValueDictionary()
self.state = ModelState(self)
self.info = None
- self._watcher_task = None
- self._watch_shutdown = asyncio.Event(loop=self.loop)
+ self._watch_stopping = asyncio.Event(loop=self.loop)
+ self._watch_stopped = asyncio.Event(loop=self.loop)
self._watch_received = asyncio.Event(loop=self.loop)
self._charmstore = CharmStore(self.loop)
+ async def __aenter__(self):
+ await self.connect_current()
+ return self
+
+ async def __aexit__(self, exc_type, exc, tb):
+ await self.disconnect()
+
+ if exc_type is not None:
+ return False
+
async def connect(self, *args, **kw):
"""Connect to an arbitrary Juju model.
"""
if 'loop' not in kw:
kw['loop'] = self.loop
+ if 'max_frame_size' not in kw:
+ kw['max_frame_size'] = self.max_frame_size
self.connection = await connection.Connection.connect(*args, **kw)
await self._after_connect()
"""
self.connection = await connection.Connection.connect_current(
- self.loop)
+ self.loop, max_frame_size=self.max_frame_size)
await self._after_connect()
async def connect_model(self, model_name):
:param model_name: Format [controller:][user/]model
"""
- self.connection = await connection.Connection.connect_model(model_name,
- self.loop)
+ self.connection = await connection.Connection.connect_model(
+ model_name, self.loop, self.max_frame_size)
await self._after_connect()
async def _after_connect(self):
"""Shut down the watcher task and close websockets.
"""
- self._stop_watching()
if self.connection and self.connection.is_open:
- await self._watch_shutdown.wait()
+ log.debug('Stopping watcher task')
+ self._watch_stopping.set()
+ await self._watch_stopped.wait()
log.debug('Closing model connection')
await self.connection.close()
self.connection = None
"""
async def _block():
while not all(c() for c in conditions):
+ if not (self.connection and self.connection.is_open):
+ raise websockets.ConnectionClosed(1006, 'no reason')
await asyncio.sleep(wait_period, loop=self.loop)
await asyncio.wait_for(_block(), timeout, loop=self.loop)
explicit call to this method.
"""
- facade = client.ClientFacade()
- facade.connect(self.connection)
+ facade = client.ClientFacade.from_connection(self.connection)
self.info = await facade.ModelInfo()
log.debug('Got ModelInfo: %s', vars(self.info))
See :meth:`add_observer` to register an onchange callback.
"""
- async def _start_watch():
- self._watch_shutdown.clear()
+ async def _all_watcher():
try:
- allwatcher = watcher.AllWatcher()
- self._watch_conn = await self.connection.clone()
- allwatcher.connect(self._watch_conn)
- while True:
- results = await allwatcher.Next()
+ allwatcher = client.AllWatcherFacade.from_connection(
+ self.connection)
+ while not self._watch_stopping.is_set():
+ try:
+ results = await utils.run_with_interrupt(
+ allwatcher.Next(),
+ self._watch_stopping,
+ self.loop)
+ except JujuAPIError as e:
+ if 'watcher was stopped' not in str(e):
+ raise
+ if self._watch_stopping.is_set():
+ # this shouldn't ever actually happen, because
+ # the event should trigger before the controller
+ # has a chance to tell us the watcher is stopped
+ # but handle it gracefully, just in case
+ break
+ # controller stopped our watcher for some reason
+ # but we're not actually stopping, so just restart it
+ log.warning(
+ 'Watcher: watcher stopped, restarting')
+ del allwatcher.Id
+ continue
+ except websockets.ConnectionClosed:
+ monitor = self.connection.monitor
+ if monitor.status == monitor.ERROR:
+ # closed unexpectedly, try to reopen
+ log.warning(
+ 'Watcher: connection closed, reopening')
+ await self.connection.reconnect()
+ del allwatcher.Id
+ continue
+ else:
+ # closed on request, go ahead and shutdown
+ break
+ if self._watch_stopping.is_set():
+ await allwatcher.Stop()
+ break
for delta in results.deltas:
delta = get_entity_delta(delta)
old_obj, new_obj = self.state.apply_delta(delta)
- # XXX: Might not want to shield at this level
- # We are shielding because when the watcher is
- # canceled (on disconnect()), we don't want all of
- # its children (every observer callback) to be
- # canceled with it. So we shield them. But this means
- # they can *never* be canceled.
- await asyncio.shield(
- self._notify_observers(delta, old_obj, new_obj),
- loop=self.loop)
+ await self._notify_observers(delta, old_obj, new_obj)
self._watch_received.set()
except CancelledError:
- log.debug('Closing watcher connection')
- await self._watch_conn.close()
- self._watch_shutdown.set()
- self._watch_conn = None
+ pass
+ except Exception:
+ log.exception('Error in watcher')
+ raise
+ finally:
+ self._watch_stopped.set()
log.debug('Starting watcher task')
- self._watcher_task = self.loop.create_task(_start_watch())
-
- def _stop_watching(self):
- """Stop the asynchronous watch against this model.
-
- """
- log.debug('Stopping watcher task')
- if self._watcher_task:
- self._watcher_task.cancel()
+ self._watch_received.clear()
+ self._watch_stopping.clear()
+ self._watch_stopped.clear()
+ self.loop.create_task(_all_watcher())
async def _notify_observers(self, delta, old_obj, new_obj):
"""Call observing callbacks, notifying them of a change in model state
'zone=us-east-1a' - starts a machine in zone us-east-1s on AWS
'maas2.name' - acquire machine maas2.name on MAAS
- :param dict constraints: Machine constraints
+ :param dict constraints: Machine constraints, which can contain the
+ the following keys::
+
+ arch : str
+ container : str
+ cores : int
+ cpu_power : int
+ instance_type : str
+ mem : int
+ root_disk : int
+ spaces : list(str)
+ tags : list(str)
+ virt_type : str
+
Example::
constraints={
'mem': 256 * MB,
+ 'tags': ['virtual'],
}
- :param list disks: List of disk constraint dictionaries
+ :param list disks: List of disk constraint dictionaries, which can
+ contain the following keys::
+
+ count : int
+ pool : str
+ size : int
+
Example::
disks=[{
params.series = series
# Submit the request.
- client_facade = client.ClientFacade()
- client_facade.connect(self.connection)
+ client_facade = client.ClientFacade.from_connection(self.connection)
results = await client_facade.AddMachines([params])
error = results.machines[0].error
if error:
:param str relation2: '<application>[:<relation_name>]'
"""
- app_facade = client.ApplicationFacade()
- app_facade.connect(self.connection)
+ app_facade = client.ApplicationFacade.from_connection(self.connection)
log.debug(
'Adding relation %s <-> %s', relation1, relation2)
:param str key: The public ssh key
"""
- key_facade = client.KeyManagerFacade()
- key_facade.connect(self.connection)
+ key_facade = client.KeyManagerFacade.from_connection(self.connection)
return await key_facade.AddKeys([key], user)
add_ssh_keys = add_ssh_key
:param dict bind: <charm endpoint>:<network space> pairs
:param dict budget: <budget name>:<limit> pairs
:param str channel: Charm store channel from which to retrieve
- the charm or bundle, e.g. 'development'
+ the charm or bundle, e.g. 'edge'
:param dict config: Charm configuration dictionary
:param constraints: Service constraints
:type constraints: :class:`juju.Constraints`
os.path.isdir(entity_url)
)
if is_local:
- entity_id = entity_url
+ entity_id = entity_url.replace('local:', '')
else:
- entity = await self.charmstore.entity(entity_url)
+ entity = await self.charmstore.entity(entity_url, channel=channel)
entity_id = entity['Id']
- client_facade = client.ClientFacade()
- client_facade.connect(self.connection)
+ client_facade = client.ClientFacade.from_connection(self.connection)
is_bundle = ((is_local and
(Path(entity_id) / 'bundle.yaml').exists()) or
application_name = entity['Meta']['charm-metadata']['Name']
if not series:
series = self._get_series(entity_url, entity)
- if not channel:
- channel = 'stable'
await client_facade.AddCharm(channel, entity_id)
# XXX: we're dropping local resources here, but we don't
# actually support them yet anyway
if not resources:
return None
- resources_facade = client.ResourcesFacade()
- resources_facade.connect(self.connection)
+ resources_facade = client.ResourcesFacade.from_connection(
+ self.connection)
response = await resources_facade.AddPendingResources(
tag.application(application),
entity_url,
config = yaml.dump({application: config},
default_flow_style=False)
- app_facade = client.ApplicationFacade()
- app_facade.connect(self.connection)
+ app_facade = client.ApplicationFacade.from_connection(
+ self.connection)
app = client.ApplicationDeploy(
charm_url=charm_url,
"""Destroy units by name.
"""
- app_facade = client.ApplicationFacade()
- app_facade.connect(self.connection)
+ app_facade = client.ApplicationFacade.from_connection(self.connection)
log.debug(
'Destroying unit%s %s',
"""
raise NotImplementedError()
- def get_config(self):
+ async def get_config(self):
"""Return the configuration settings for this model.
+ :returns: A ``dict`` mapping keys to `ConfigValue` instances,
+ which have `source` and `value` attributes.
"""
- raise NotImplementedError()
+ config_facade = client.ModelConfigFacade.from_connection(
+ self.connection
+ )
+ result = await config_facade.ModelGet()
+ config = result.config
+ for key, value in config.items():
+ config[key] = ConfigValue.from_json(value)
+ return config
def get_constraints(self):
"""Return the machine constraints for this model.
:param str acl: Access control ('read' or 'write')
"""
- model_facade = client.ModelManagerFacade()
controller_conn = await self.connection.controller()
- model_facade.connect(controller_conn)
+ model_facade = client.ModelManagerFacade.from_connection(
+ controller_conn)
user = tag.user(username)
model = tag.model(self.info.uuid)
changes = client.ModifyModelAccess(acl, 'grant', model, user)
else it's fingerprint
"""
- key_facade = client.KeyManagerFacade()
- key_facade.connect(self.connection)
+ key_facade = client.KeyManagerFacade.from_connection(self.connection)
entity = {'tag': tag.model(self.info.uuid)}
entities = client.Entities([entity])
return await key_facade.ListKeys(entities, raw_ssh)
:param str user: Juju user to which the key is registered
"""
- key_facade = client.KeyManagerFacade()
- key_facade.connect(self.connection)
+ key_facade = client.KeyManagerFacade.from_connection(self.connection)
key = base64.b64decode(bytes(key.strip().split()[1].encode('ascii')))
key = hashlib.md5(key).hexdigest()
key = ':'.join(a+b for a, b in zip(key[::2], key[1::2]))
:param str username: Username to revoke
"""
- model_facade = client.ModelManagerFacade()
controller_conn = await self.connection.controller()
- model_facade.connect(controller_conn)
+ model_facade = client.ModelManagerFacade.from_connection(
+ controller_conn)
user = tag.user(username)
model = tag.model(self.info.uuid)
changes = client.ModifyModelAccess('read', 'revoke', model, user)
"""
raise NotImplementedError()
- def set_config(self, **config):
+ async def set_config(self, config):
"""Set configuration keys on this model.
- :param \*\*config: Config key/values
-
+ :param dict config: Mapping of config keys to either string values or
+ `ConfigValue` instances, as returned by `get_config`.
"""
- raise NotImplementedError()
+ config_facade = client.ModelConfigFacade.from_connection(
+ self.connection
+ )
+ for key, value in config.items():
+ if isinstance(value, ConfigValue):
+ config[key] = value.value
+ await config_facade.ModelSet(config)
def set_constraints(self, constraints):
"""Set machine constraints on this model.
:param bool utc: Display time as UTC in RFC3339 format
"""
- client_facade = client.ClientFacade()
- client_facade.connect(self.connection)
+ client_facade = client.ClientFacade.from_connection(self.connection)
return await client_facade.FullStatus(filters)
def sync_tools(
log.debug("Retrieving metrics for %s",
', '.join(tags) if tags else "all units")
- metrics_facade = client.MetricsDebugFacade()
- metrics_facade.connect(self.connection)
+ metrics_facade = client.MetricsDebugFacade.from_connection(
+ self.connection)
entities = [client.Entity(tag) for tag in tags]
metrics_result = await metrics_facade.GetMetrics(entities)
for unit_name, unit in model.units.items():
app_units = self._units_by_app.setdefault(unit.application, [])
app_units.append(unit_name)
- self.client_facade = client.ClientFacade()
- self.client_facade.connect(model.connection)
- self.app_facade = client.ApplicationFacade()
- self.app_facade.connect(model.connection)
- self.ann_facade = client.AnnotationsFacade()
- self.ann_facade.connect(model.connection)
+ self.client_facade = client.ClientFacade.from_connection(
+ model.connection)
+ self.app_facade = client.ApplicationFacade.from_connection(
+ model.connection)
+ self.ann_facade = client.AnnotationsFacade.from_connection(
+ model.connection)
async def _handle_local_charms(self, bundle):
"""Search for references to local charms (i.e. filesystem paths)
class CharmArchiveGenerator(object):
+ """
+ Create a Zip archive of a local charm directory for upload to a controller.
+
+ This is used automatically by
+ `Model.add_local_charm_dir <#juju.model.Model.add_local_charm_dir>`_.
+ """
def __init__(self, path):
self.path = os.path.abspath(os.path.expanduser(path))
projects / osm / N2VC.git / blobdiff