import asyncio
+import base64
import collections
+import hashlib
import json
import logging
import os
from pathlib import Path
import yaml
-from theblues import charmstore
+import theblues.charmstore
+import theblues.errors
+from . import tag
from .client import client
-from .client import watcher
from .client import connection
from .constraints import parse as parse_constraints, normalize_key
from .delta import get_entity_delta
model.
"""
- return self.safe_data[name]
+ try:
+ return self.safe_data[name]
+ except KeyError:
+ name = name.replace('_', '-')
+ if name in self.safe_data:
+ return self.safe_data[name]
+ else:
+ raise
def __bool__(self):
return bool(self.data)
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._watch_shutdown = asyncio.Event(loop=self.loop)
+ self._watch_received = asyncio.Event(loop=self.loop)
self._charmstore = CharmStore(self.loop)
async def connect(self, *args, **kw):
args and kw are passed through to Connection.connect()
"""
+ if 'loop' not in kw:
+ kw['loop'] = self.loop
self.connection = await connection.Connection.connect(*args, **kw)
await self._after_connect()
"""Connect to the current Juju model.
"""
- self.connection = await connection.Connection.connect_current()
+ self.connection = await connection.Connection.connect_current(
+ self.loop)
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.connection = await connection.Connection.connect_model(model_name,
+ self.loop)
await self._after_connect()
async def _after_connect(self):
"""
async def _block():
while not all(c() for c in conditions):
- await asyncio.sleep(wait_period)
- await asyncio.wait_for(_block(), timeout)
+ await asyncio.sleep(wait_period, loop=self.loop)
+ await asyncio.wait_for(_block(), timeout, loop=self.loop)
@property
def applications(self):
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))
async def _start_watch():
self._watch_shutdown.clear()
try:
- allwatcher = watcher.AllWatcher()
- self._watch_conn = await self.connection.clone()
- allwatcher.connect(self._watch_conn)
+ allwatcher = client.AllWatcherFacade.from_connection(
+ self.connection)
while True:
results = await allwatcher.Next()
for delta in results.deltas:
# 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))
+ self._notify_observers(delta, old_obj, new_obj),
+ loop=self.loop)
self._watch_received.set()
except CancelledError:
- log.debug('Closing watcher connection')
- await self._watch_conn.close()
self._watch_shutdown.set()
- self._watch_conn = None
+ except Exception:
+ log.exception('Error in watcher')
+ raise
log.debug('Starting watcher task')
self._watcher_task = self.loop.create_task(_start_watch())
for o in self.observers:
if o.cares_about(delta):
- asyncio.ensure_future(o(delta, old_obj, new_obj, self))
+ asyncio.ensure_future(o(delta, old_obj, new_obj, self),
+ loop=self.loop)
async def _wait(self, entity_type, entity_id, action, predicate=None):
"""
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:
- raise ValueError("Error adding machine: %s", error.message)
+ raise ValueError("Error adding machine: %s" % error.message)
machine_id = results.machines[0].machine
log.debug('Added new machine %s', machine_id)
return await self._wait_for_new('machine', machine_id)
: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 \*cidrs: Optional list of existing subnet CIDRs
"""
- pass
+ raise NotImplementedError()
- def add_ssh_key(self, key):
+ async def add_ssh_key(self, user, key):
"""Add a public SSH key to this model.
+ :param str user: The username of the user
:param str key: The public ssh key
"""
- pass
+ key_facade = client.KeyManagerFacade.from_connection(self.connection)
+ return await key_facade.AddKeys([key], user)
add_ssh_keys = add_ssh_key
def add_subnet(self, cidr_or_id, space, *zones):
:param str \*zones: Zone(s) in which the subnet resides
"""
- pass
+ raise NotImplementedError()
def get_backups(self):
"""Retrieve metadata for backups in this model.
"""
- pass
+ raise NotImplementedError()
def block(self, *commands):
"""Add a new block to this model.
'all-changes', 'destroy-model', 'remove-object'
"""
- pass
+ raise NotImplementedError()
def get_blocks(self):
"""List blocks for this model.
"""
- pass
+ raise NotImplementedError()
def get_cached_images(self, arch=None, kind=None, series=None):
"""Return a list of cached OS images.
:param str series: Filter by image series, e.g. 'xenial'
"""
- pass
+ raise NotImplementedError()
def create_backup(self, note=None, no_download=False):
"""Create a backup of this model.
:return str: Path to downloaded archive
"""
- pass
+ raise NotImplementedError()
def create_storage_pool(self, name, provider_type, **pool_config):
"""Create or define a storage pool.
:param \*\*pool_config: key/value pool configuration pairs
"""
- pass
+ raise NotImplementedError()
def debug_log(
self, no_tail=False, exclude_module=None, include_module=None,
:param list exclude: Do not show log messages for these entities
"""
- pass
+ raise NotImplementedError()
+
+ def _get_series(self, entity_url, entity):
+ # try to get the series from the provided charm URL
+ if entity_url.startswith('cs:'):
+ parts = entity_url[3:].split('/')
+ else:
+ parts = entity_url.split('/')
+ if parts[0].startswith('~'):
+ parts.pop(0)
+ if len(parts) > 1:
+ # series was specified in the URL
+ return parts[0]
+ # series was not supplied at all, so use the newest
+ # supported series according to the charm store
+ ss = entity['Meta']['supported-series']
+ return ss['SupportedSeries'][0]
async def deploy(
self, entity_url, application_name=None, bind=None, budget=None,
TODO::
- - application_name is required; fill this in automatically if not
- provided by caller
- - series is required; how do we pick a default?
+ - support local resources
"""
- if to:
- placement = parse_placement(to)
- else:
- placement = []
-
if storage:
storage = {
k: client.Constraints(**v)
entity_url.startswith('local:') or
os.path.isdir(entity_url)
)
- entity_id = await self.charmstore.entityId(entity_url) \
- if not is_local else entity_url
+ if is_local:
+ entity_id = entity_url
+ else:
+ entity = await self.charmstore.entity(entity_url)
+ entity_id = entity['Id']
- app_facade = client.ApplicationFacade()
- client_facade = client.ClientFacade()
- app_facade.connect(self.connection)
- 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
# haven't made it yet we'll need to wait on them to be added
await asyncio.gather(*[
asyncio.ensure_future(
- self._wait_for_new('application', app_name))
+ self._wait_for_new('application', app_name),
+ loop=self.loop)
for app_name in pending_apps
- ])
+ ], loop=self.loop)
return [app for name, app in self.applications.items()
if name in handler.applications]
else:
- log.debug(
- 'Deploying %s', entity_id)
-
if not is_local:
+ if not application_name:
+ 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)
- elif not entity_id.startswith('local:'):
+ # XXX: we're dropping local resources here, but we don't
+ # actually support them yet anyway
+ resources = await self._add_store_resources(application_name,
+ entity_id,
+ entity)
+ else:
# We have a local charm dir that needs to be uploaded
charm_dir = os.path.abspath(
os.path.expanduser(entity_id))
"Pass a 'series' kwarg to Model.deploy().".format(
charm_dir))
entity_id = await self.add_local_charm_dir(charm_dir, series)
-
- app = client.ApplicationDeploy(
- application=application_name,
- channel=channel,
+ return await self._deploy(
charm_url=entity_id,
- config=config,
- constraints=parse_constraints(constraints),
+ application=application_name,
+ series=series,
+ config=config or {},
+ constraints=constraints,
endpoint_bindings=bind,
- num_units=num_units,
resources=resources,
- series=series,
storage=storage,
+ channel=channel,
+ num_units=num_units,
+ placement=parse_placement(to)
)
- app.placement = placement
- result = await app_facade.Deploy([app])
- errors = [r.error.message for r in result.results if r.error]
- if errors:
- raise JujuError('\n'.join(errors))
- return await self._wait_for_new('application', application_name)
+ async def _add_store_resources(self, application, entity_url, entity=None):
+ if not entity:
+ # avoid extra charm store call if one was already made
+ entity = await self.charmstore.entity(entity_url)
+ resources = [
+ {
+ 'description': resource['Description'],
+ 'fingerprint': resource['Fingerprint'],
+ 'name': resource['Name'],
+ 'path': resource['Path'],
+ 'revision': resource['Revision'],
+ 'size': resource['Size'],
+ 'type_': resource['Type'],
+ 'origin': 'store',
+ } for resource in entity['Meta']['resources']
+ ]
+
+ if not resources:
+ return None
- def destroy(self):
- """Terminate all machines and resources for this model.
+ resources_facade = client.ResourcesFacade.from_connection(
+ self.connection)
+ response = await resources_facade.AddPendingResources(
+ tag.application(application),
+ entity_url,
+ [client.CharmResource(**resource) for resource in resources])
+ resource_map = {resource['name']: pid
+ for resource, pid
+ in zip(resources, response.pending_ids)}
+ return resource_map
+ async def _deploy(self, charm_url, application, series, config,
+ constraints, endpoint_bindings, resources, storage,
+ channel=None, num_units=None, placement=None):
+ """Logic shared between `Model.deploy` and `BundleHandler.deploy`.
"""
- pass
+ log.info('Deploying %s', charm_url)
+
+ # stringify all config values for API, and convert to YAML
+ config = {k: str(v) for k, v in config.items()}
+ config = yaml.dump({application: config},
+ default_flow_style=False)
+
+ app_facade = client.ApplicationFacade.from_connection(
+ self.connection)
+
+ app = client.ApplicationDeploy(
+ charm_url=charm_url,
+ application=application,
+ series=series,
+ channel=channel,
+ config_yaml=config,
+ constraints=parse_constraints(constraints),
+ endpoint_bindings=endpoint_bindings,
+ num_units=num_units,
+ resources=resources,
+ storage=storage,
+ placement=placement
+ )
+
+ result = await app_facade.Deploy([app])
+ errors = [r.error.message for r in result.results if r.error]
+ if errors:
+ raise JujuError('\n'.join(errors))
+ return await self._wait_for_new('application', application)
+
+ async def destroy(self):
+ """Terminate all machines and resources for this model.
+ Is already implemented in controller.py.
+ """
+ raise NotImplementedError()
async def destroy_unit(self, *unit_names):
"""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',
:return str: Path to the archive file
"""
- pass
+ raise NotImplementedError()
def enable_ha(
self, num_controllers=0, constraints=None, series=None, to=None):
If None, a new machine is provisioned.
"""
- pass
+ raise NotImplementedError()
def get_config(self):
"""Return the configuration settings for this model.
"""
- pass
+ raise NotImplementedError()
def get_constraints(self):
"""Return the machine constraints for this model.
"""
- pass
+ raise NotImplementedError()
- def grant(self, username, acl='read'):
+ async def grant(self, username, acl='read'):
"""Grant a user access to this model.
:param str username: Username
:param str acl: Access control ('read' or 'write')
"""
- pass
+ controller_conn = await self.connection.controller()
+ 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)
+ await self.revoke(username)
+ return await model_facade.ModifyModelAccess([changes])
def import_ssh_key(self, identity):
"""Add a public SSH key from a trusted indentity source to this model.
:param str identity: User identity in the form <lp|gh>:<username>
"""
- pass
+ raise NotImplementedError()
import_ssh_keys = import_ssh_key
- def get_machines(self, machine, utc=False):
+ async def get_machines(self):
"""Return list of machines in this model.
- :param str machine: Machine id, e.g. '0'
- :param bool utc: Display time as UTC in RFC3339 format
-
"""
- pass
+ return list(self.state.machines.keys())
def get_shares(self):
"""Return list of all users with access to this model.
"""
- pass
+ raise NotImplementedError()
def get_spaces(self):
"""Return list of all known spaces, including associated subnets.
"""
- pass
+ raise NotImplementedError()
- def get_ssh_key(self):
+ async def get_ssh_key(self, raw_ssh=False):
"""Return known SSH keys for this model.
+ :param bool raw_ssh: if True, returns the raw ssh key,
+ else it's fingerprint
"""
- pass
+ 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)
get_ssh_keys = get_ssh_key
def get_storage(self, filesystem=False, volume=False):
:param bool volume: Include volume storage
"""
- pass
+ raise NotImplementedError()
def get_storage_pools(self, names=None, providers=None):
"""Return list of storage pools.
:param list providers: Only include pools for these providers
"""
- pass
+ raise NotImplementedError()
def get_subnets(self, space=None, zone=None):
"""Return list of known subnets.
:param str zone: Only include subnets in this zone
"""
- pass
+ raise NotImplementedError()
def remove_blocks(self):
"""Remove all blocks from this model.
"""
- pass
+ raise NotImplementedError()
def remove_backup(self, backup_id):
"""Delete a backup.
:param str backup_id: The id of the backup to remove
"""
- pass
+ raise NotImplementedError()
def remove_cached_images(self, arch=None, kind=None, series=None):
"""Remove cached OS images.
:param str series: Image series to remove, e.g. 'xenial'
"""
- pass
+ raise NotImplementedError()
def remove_machine(self, *machine_ids):
"""Remove a machine from this model.
:param str \*machine_ids: Ids of the machines to remove
"""
- pass
+ raise NotImplementedError()
remove_machines = remove_machine
- def remove_ssh_key(self, *keys):
+ async def remove_ssh_key(self, user, key):
"""Remove a public SSH key(s) from this model.
- :param str \*keys: Keys to remove
+ :param str key: Full ssh key
+ :param str user: Juju user to which the key is registered
"""
- pass
+ 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]))
+ await key_facade.DeleteKeys([key], user)
remove_ssh_keys = remove_ssh_key
def restore_backup(
:param bool upload_tools: Upload tools if bootstrapping a new machine
"""
- pass
+ raise NotImplementedError()
def retry_provisioning(self):
"""Retry provisioning for failed machines.
"""
- pass
+ raise NotImplementedError()
- def revoke(self, username, acl='read'):
+ async def revoke(self, username):
"""Revoke a user's access to this model.
:param str username: Username to revoke
- :param str acl: Access control ('read' or 'write')
"""
- pass
+ controller_conn = await self.connection.controller()
+ 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)
+ return await model_facade.ModifyModelAccess([changes])
def run(self, command, timeout=None):
"""Run command on all machines in this model.
:param int timeout: Time to wait before command is considered failed
"""
- pass
+ raise NotImplementedError()
def set_config(self, **config):
"""Set configuration keys on this model.
:param \*\*config: Config key/values
"""
- pass
+ raise NotImplementedError()
def set_constraints(self, constraints):
"""Set machine constraints on this model.
:param :class:`juju.Constraints` constraints: Machine constraints
"""
- pass
+ raise NotImplementedError()
def get_action_output(self, action_uuid, wait=-1):
"""Get the results of an action by ID.
:param int wait: Time in seconds to wait for action to complete
"""
- pass
+ raise NotImplementedError()
def get_action_status(self, uuid_or_prefix=None, name=None):
"""Get the status of all actions, filtered by ID, ID prefix, or action name.
:param str name: Filter by action name
"""
- pass
+ raise NotImplementedError()
def get_budget(self, budget_name):
"""Get budget usage info.
:param str budget_name: Name of budget
"""
- pass
+ raise NotImplementedError()
- def get_status(self, filter_=None, utc=False):
+ async def get_status(self, filters=None, utc=False):
"""Return the status of the model.
- :param str filter_: Service or unit name or wildcard ('*')
+ :param str filters: Optional list of applications, units, or machines
+ to include, which can use wildcards ('*').
:param bool utc: Display time as UTC in RFC3339 format
"""
- pass
- status = get_status
+ client_facade = client.ClientFacade.from_connection(self.connection)
+ return await client_facade.FullStatus(filters)
def sync_tools(
self, all_=False, destination=None, dry_run=False, public=False,
:param str version: Copy a specific major.minor version
"""
- pass
+ raise NotImplementedError()
def unblock(self, *commands):
"""Unblock an operation that would alter this model.
'all-changes', 'destroy-model', 'remove-object'
"""
- pass
+ raise NotImplementedError()
def unset_config(self, *keys):
"""Unset configuration on this model.
:param str \*keys: The keys to unset
"""
- pass
+ raise NotImplementedError()
def upgrade_gui(self):
"""Upgrade the Juju GUI for this model.
"""
- pass
+ raise NotImplementedError()
def upgrade_juju(
self, dry_run=False, reset_previous_upgrade=False,
:param str version: Upgrade to a specific version
"""
- pass
+ raise NotImplementedError()
def upload_backup(self, archive_path):
"""Store a backup archive remotely in Juju.
:param str archive_path: Path to local archive
"""
- pass
+ raise NotImplementedError()
@property
def charmstore(self):
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)
charm_urls = await asyncio.gather(*[
self.model.add_local_charm_dir(*params)
for params in args
- ])
+ ], loop=self.model.loop)
# Update the 'charm:' entry for each app with the new 'local:' url.
for app_name, charm_url in zip(apps, charm_urls):
bundle['services'][app_name]['charm'] = charm_url
self.plan = await self.client_facade.GetBundleChanges(
yaml.dump(self.bundle))
- if self.plan.errors:
- raise JujuError('\n'.join(self.plan.errors))
-
async def execute_plan(self):
for step in self.plan.changes:
method = getattr(self, step.method)
results = await self.client_facade.AddMachines([params])
error = results.machines[0].error
if error:
- raise ValueError("Error adding machine: %s", error.message)
+ raise ValueError("Error adding machine: %s" % error.message)
machine = results.machines[0].machine
log.debug('Added new machine %s', machine)
return machine
"""
# resolve indirect references
charm = self.resolve(charm)
- # stringify all config values for API
- options = {k: str(v) for k, v in options.items()}
- # build param object
- app = client.ApplicationDeploy(
+ # the bundle plan doesn't actually do anything with resources, even
+ # though it ostensibly gives us something (None) for that param
+ if not charm.startswith('local:'):
+ resources = await self.model._add_store_resources(application,
+ charm)
+ await self.model._deploy(
charm_url=charm,
- series=series,
application=application,
+ series=series,
config=options,
- constraints=parse_constraints(constraints),
- storage=storage,
+ constraints=constraints,
endpoint_bindings=endpoint_bindings,
resources=resources,
+ storage=storage,
)
- # 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):
"""
def __init__(self, loop):
self.loop = loop
- self._cs = charmstore.CharmStore()
+ self._cs = theblues.charmstore.CharmStore(timeout=5)
def __getattr__(self, name):
"""
else:
async def coro(*args, **kwargs):
method = partial(attr, *args, **kwargs)
- return await self.loop.run_in_executor(None, method)
+ for attempt in range(1, 4):
+ try:
+ return await self.loop.run_in_executor(None, method)
+ except theblues.errors.ServerError:
+ if attempt == 3:
+ raise
+ await asyncio.sleep(1, loop=self.loop)
setattr(self, name, coro)
wrapper = coro
return wrapper
projects / osm / N2VC.git / blobdiff