--- /dev/null
+import asyncio
+import base64
+import collections
+import hashlib
+import json
+import logging
+import os
+import re
+import stat
+import tempfile
+import weakref
+import zipfile
+from concurrent.futures import CancelledError
+from functools import partial
+from pathlib import Path
+
+import theblues.charmstore
+import theblues.errors
+import websockets
+import yaml
+
+from . import tag, utils
+from .client import client, connector
+from .client.client import ConfigValue
+from .client.client import Value
+from .constraints import parse as parse_constraints
+from .constraints import normalize_key
+from .delta import get_entity_class, get_entity_delta
+from .errors import JujuAPIError, JujuError
+from .exceptions import DeadEntityException
+from .placement import parse as parse_placement
+from . import provisioner
+
+
+log = logging.getLogger(__name__)
+
+
+class _Observer:
+ """Wrapper around an observer callable.
+
+ This wrapper allows filter criteria to be associated with the
+ callable so that it's only called for changes that meet the criteria.
+
+ """
+ def __init__(self, callable_, entity_type, action, entity_id, predicate):
+ self.callable_ = callable_
+ self.entity_type = entity_type
+ self.action = action
+ self.entity_id = entity_id
+ self.predicate = predicate
+ if self.entity_id:
+ self.entity_id = str(self.entity_id)
+ if not self.entity_id.startswith('^'):
+ self.entity_id = '^' + self.entity_id
+ if not self.entity_id.endswith('$'):
+ self.entity_id += '$'
+
+ async def __call__(self, delta, old, new, model):
+ await self.callable_(delta, old, new, model)
+
+ def cares_about(self, delta):
+ """Return True if this observer "cares about" (i.e. wants to be
+ called) for a this delta.
+
+ """
+ if (self.entity_id and delta.get_id() and
+ not re.match(self.entity_id, str(delta.get_id()))):
+ return False
+
+ if self.entity_type and self.entity_type != delta.entity:
+ return False
+
+ if self.action and self.action != delta.type:
+ return False
+
+ if self.predicate and not self.predicate(delta):
+ return False
+
+ return True
+
+
+class ModelObserver:
+ """
+ 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)
+ await method(delta, old, new, model)
+
+ 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`
+ :param model: :class:`juju.model.Model`
+
+ """
+ pass
+
+
+class ModelState:
+ """Holds the state of the model, including the delta history of all
+ entities in the model.
+
+ """
+ def __init__(self, model):
+ self.model = model
+ self.state = dict()
+
+ def _live_entity_map(self, entity_type):
+ """Return an id:Entity map of all the living entities of
+ type ``entity_type``.
+
+ """
+ return {
+ entity_id: self.get_entity(entity_type, entity_id)
+ for entity_id, history in self.state.get(entity_type, {}).items()
+ if history[-1] is not None
+ }
+
+ @property
+ def applications(self):
+ """Return a map of application-name:Application for all applications
+ currently in the model.
+
+ """
+ return self._live_entity_map('application')
+
+ @property
+ def machines(self):
+ """Return a map of machine-id:Machine for all machines currently in
+ the model.
+
+ """
+ return self._live_entity_map('machine')
+
+ @property
+ def units(self):
+ """Return a map of unit-id:Unit for all units currently in
+ the model.
+
+ """
+ return self._live_entity_map('unit')
+
+ @property
+ def relations(self):
+ """Return a map of relation-id:Relation for all relations currently in
+ the model.
+
+ """
+ return self._live_entity_map('relation')
+
+ def entity_history(self, entity_type, entity_id):
+ """Return the history deque for an entity.
+
+ """
+ return self.state[entity_type][entity_id]
+
+ def entity_data(self, entity_type, entity_id, history_index):
+ """Return the data dict for an entity at a specific index of its
+ history.
+
+ """
+ return self.entity_history(entity_type, entity_id)[history_index]
+
+ def apply_delta(self, delta):
+ """Apply delta to our state and return a copy of the
+ affected object as it was before and after the update, e.g.:
+
+ old_obj, new_obj = self.apply_delta(delta)
+
+ old_obj may be None if the delta is for the creation of a new object,
+ e.g. a new application or unit is deployed.
+
+ new_obj will never be None, but may be dead (new_obj.dead == True)
+ if the object was deleted as a result of the delta being applied.
+
+ """
+ history = (
+ self.state
+ .setdefault(delta.entity, {})
+ .setdefault(delta.get_id(), collections.deque())
+ )
+
+ history.append(delta.data)
+ if delta.type == 'remove':
+ history.append(None)
+
+ entity = self.get_entity(delta.entity, delta.get_id())
+ return entity.previous(), entity
+
+ def get_entity(
+ self, entity_type, entity_id, history_index=-1, connected=True):
+ """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.
+
+ """
+
+ if history_index < 0 and history_index != -1:
+ history_index += len(self.entity_history(entity_type, entity_id))
+ if history_index < 0:
+ return None
+
+ try:
+ self.entity_data(entity_type, entity_id, history_index)
+ except IndexError:
+ return None
+
+ entity_class = get_entity_class(entity_type)
+ return entity_class(
+ entity_id, self.model, history_index=history_index,
+ connected=connected)
+
+
+class ModelEntity:
+ """An object in the Model tree"""
+
+ def __init__(self, entity_id, model, history_index=-1, connected=True):
+ """Initialize a new entity
+
+ :param entity_id str: The unique id of the object in the model
+ :param model: The model instance in whose object tree this
+ entity resides
+ :history_index int: The index of this object's state in the model's
+ history deque for this entity
+ :connected bool: Flag indicating whether this object gets live updates
+ from the model.
+
+ """
+ self.entity_id = entity_id
+ self.model = model
+ self._history_index = history_index
+ self.connected = connected
+ self.connection = model.connection()
+
+ def __repr__(self):
+ return '<{} entity_id="{}">'.format(type(self).__name__,
+ self.entity_id)
+
+ def __getattr__(self, name):
+ """Fetch object attributes from the underlying data dict held in the
+ model.
+
+ """
+ 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)
+
+ def on_change(self, callable_):
+ """Add a change observer to this entity.
+
+ """
+ self.model.add_observer(
+ callable_, self.entity_type, 'change', self.entity_id)
+
+ def on_remove(self, callable_):
+ """Add a remove observer to this entity.
+
+ """
+ self.model.add_observer(
+ callable_, self.entity_type, 'remove', self.entity_id)
+
+ @property
+ def entity_type(self):
+ """A string identifying the entity type of this object, e.g.
+ 'application' or 'unit', etc.
+
+ """
+ return self.__class__.__name__.lower()
+
+ @property
+ def current(self):
+ """Return True if this object represents the current state of the
+ entity in the underlying model.
+
+ This will be True except when the object represents an entity at a
+ non-latest state in history, e.g. if the object was obtained by calling
+ .previous() on another object.
+
+ """
+ return self._history_index == -1
+
+ @property
+ def dead(self):
+ """Returns True if this entity no longer exists in the underlying
+ model.
+
+ """
+ return (
+ self.data is None or
+ self.model.state.entity_data(
+ self.entity_type, self.entity_id, -1) is None
+ )
+
+ @property
+ def alive(self):
+ """Returns True if this entity still exists in the underlying
+ model.
+
+ """
+ return not self.dead
+
+ @property
+ def data(self):
+ """The data dictionary for this entity.
+
+ """
+ return self.model.state.entity_data(
+ self.entity_type, self.entity_id, self._history_index)
+
+ @property
+ def safe_data(self):
+ """The data dictionary for this entity.
+
+ If this `ModelEntity` points to the dead state, it will
+ raise `DeadEntityException`.
+
+ """
+ if self.data is None:
+ raise DeadEntityException(
+ "Entity {}:{} is dead - its attributes can no longer be "
+ "accessed. Use the .previous() method on this object to get "
+ "a copy of the object at its previous state.".format(
+ self.entity_type, self.entity_id))
+ return self.data
+
+ def previous(self):
+ """Return a copy of this object as was at its previous state in
+ history.
+
+ Returns None if this object is new (and therefore has no history).
+
+ The returned object is always "disconnected", i.e. does not receive
+ live updates.
+
+ """
+ return self.model.state.get_entity(
+ self.entity_type, self.entity_id, self._history_index - 1,
+ connected=False)
+
+ def next(self):
+ """Return a copy of this object at its next state in
+ history.
+
+ Returns None if this object is already the latest.
+
+ The returned object is "disconnected", i.e. does not receive
+ live updates, unless it is current (latest).
+
+ """
+ if self._history_index == -1:
+ return None
+
+ new_index = self._history_index + 1
+ connected = (
+ new_index == len(self.model.state.entity_history(
+ self.entity_type, self.entity_id)) - 1
+ )
+ return self.model.state.get_entity(
+ self.entity_type, self.entity_id, self._history_index - 1,
+ connected=connected)
+
+ def latest(self):
+ """Return a copy of this object at its current state in the model.
+
+ Returns self if this object is already the latest.
+
+ The returned object is always "connected", i.e. receives
+ live updates from the model.
+
+ """
+ if self._history_index == -1:
+ return self
+
+ return self.model.state.get_entity(self.entity_type, self.entity_id)
+
+
+class Model:
+ """
+ The main API for interacting with a Juju model.
+ """
+ def __init__(
+ self,
+ loop=None,
+ max_frame_size=None,
+ bakery_client=None,
+ jujudata=None,
+ ):
+ """Instantiate a new Model.
+
+ The connect method will need to be called before this
+ object can be used for anything interesting.
+
+ If jujudata is None, jujudata.FileJujuData will be used.
+
+ :param loop: an asyncio event loop
+ :param max_frame_size: See
+ `juju.client.connection.Connection.MAX_FRAME_SIZE`
+ :param bakery_client httpbakery.Client: The bakery client to use
+ for macaroon authorization.
+ :param jujudata JujuData: The source for current controller information
+ """
+ self._connector = connector.Connector(
+ loop=loop,
+ max_frame_size=max_frame_size,
+ bakery_client=bakery_client,
+ jujudata=jujudata,
+ )
+ self._observers = weakref.WeakValueDictionary()
+ self.state = ModelState(self)
+ self._info = None
+ self._watch_stopping = asyncio.Event(loop=self._connector.loop)
+ self._watch_stopped = asyncio.Event(loop=self._connector.loop)
+ self._watch_received = asyncio.Event(loop=self._connector.loop)
+ self._watch_stopped.set()
+ self._charmstore = CharmStore(self._connector.loop)
+
+ def is_connected(self):
+ """Reports whether the Model is currently connected."""
+ return self._connector.is_connected()
+
+ @property
+ def loop(self):
+ return self._connector.loop
+
+ def connection(self):
+ """Return the current Connection object. It raises an exception
+ if the Model is disconnected"""
+ return self._connector.connection()
+
+ async def get_controller(self):
+ """Return a Controller instance for the currently connected model.
+ :return Controller:
+ """
+ from juju.controller import Controller
+ controller = Controller(jujudata=self._connector.jujudata)
+ kwargs = self.connection().connect_params()
+ kwargs.pop('uuid')
+ await controller._connect_direct(**kwargs)
+ return controller
+
+ async def __aenter__(self):
+ await self.connect()
+ return self
+
+ async def __aexit__(self, exc_type, exc, tb):
+ await self.disconnect()
+
+ async def connect(self, *args, **kwargs):
+ """Connect to a juju model.
+
+ This supports two calling conventions:
+
+ The model and (optionally) authentication information can be taken
+ from the data files created by the Juju CLI. This convention will
+ be used if a ``model_name`` is specified, or if the ``endpoint``
+ and ``uuid`` are not.
+
+ Otherwise, all of the ``endpoint``, ``uuid``, and authentication
+ information (``username`` and ``password``, or ``bakery_client`` and/or
+ ``macaroons``) are required.
+
+ If a single positional argument is given, it will be assumed to be
+ the ``model_name``. Otherwise, the first positional argument, if any,
+ must be the ``endpoint``.
+
+ Available parameters are:
+
+ :param model_name: Format [controller:][user/]model
+ :param str endpoint: The hostname:port of the controller to connect to.
+ :param str uuid: The model UUID to connect to.
+ :param str username: The username for controller-local users (or None
+ to use macaroon-based login.)
+ :param str password: The password for controller-local users.
+ :param str cacert: The CA certificate of the controller
+ (PEM formatted).
+ :param httpbakery.Client bakery_client: The macaroon bakery client to
+ to use when performing macaroon-based login. Macaroon tokens
+ acquired when logging will be saved to bakery_client.cookies.
+ If this is None, a default bakery_client will be used.
+ :param list macaroons: List of macaroons to load into the
+ ``bakery_client``.
+ :param asyncio.BaseEventLoop loop: The event loop to use for async
+ operations.
+ :param int max_frame_size: The maximum websocket frame size to allow.
+ """
+ await self.disconnect()
+ if 'endpoint' not in kwargs and len(args) < 2:
+ if args and 'model_name' in kwargs:
+ raise TypeError('connect() got multiple values for model_name')
+ elif args:
+ model_name = args[0]
+ else:
+ model_name = kwargs.pop('model_name', None)
+ await self._connector.connect_model(model_name, **kwargs)
+ else:
+ if 'model_name' in kwargs:
+ raise TypeError('connect() got values for both '
+ 'model_name and endpoint')
+ if args and 'endpoint' in kwargs:
+ raise TypeError('connect() got multiple values for endpoint')
+ if len(args) < 2 and 'uuid' not in kwargs:
+ raise TypeError('connect() missing value for uuid')
+ has_userpass = (len(args) >= 4 or
+ {'username', 'password'}.issubset(kwargs))
+ has_macaroons = (len(args) >= 6 or not
+ {'bakery_client', 'macaroons'}.isdisjoint(kwargs))
+ if not (has_userpass or has_macaroons):
+ raise TypeError('connect() missing auth params')
+ arg_names = [
+ 'endpoint',
+ 'uuid',
+ 'username',
+ 'password',
+ 'cacert',
+ 'bakery_client',
+ 'macaroons',
+ 'loop',
+ 'max_frame_size',
+ ]
+ for i, arg in enumerate(args):
+ kwargs[arg_names[i]] = arg
+ if not {'endpoint', 'uuid'}.issubset(kwargs):
+ raise ValueError('endpoint and uuid are required '
+ 'if model_name not given')
+ if not ({'username', 'password'}.issubset(kwargs) or
+ {'bakery_client', 'macaroons'}.intersection(kwargs)):
+ raise ValueError('Authentication parameters are required '
+ 'if model_name not given')
+ await self._connector.connect(**kwargs)
+ await self._after_connect()
+
+ async def connect_model(self, model_name):
+ """
+ .. deprecated:: 0.6.2
+ Use ``connect(model_name=model_name)`` instead.
+ """
+ return await self.connect(model_name=model_name)
+
+ async def connect_current(self):
+ """
+ .. deprecated:: 0.6.2
+ Use ``connect()`` instead.
+ """
+ return await self.connect()
+
+ async def _connect_direct(self, **kwargs):
+ await self.disconnect()
+ await self._connector.connect(**kwargs)
+ await self._after_connect()
+
+ async def _after_connect(self):
+ self._watch()
+
+ # Wait for the first packet of data from the AllWatcher,
+ # which contains all information on the model.
+ # TODO this means that we can't do anything until
+ # we've received all the model data, which might be
+ # a whole load of unneeded data if all the client wants
+ # to do is make one RPC call.
+ await self._watch_received.wait()
+
+ await self.get_info()
+
+ async def disconnect(self):
+ """Shut down the watcher task and close websockets.
+
+ """
+ if not self._watch_stopped.is_set():
+ log.debug('Stopping watcher task')
+ self._watch_stopping.set()
+ await self._watch_stopped.wait()
+ self._watch_stopping.clear()
+
+ if self.is_connected():
+ log.debug('Closing model connection')
+ await self._connector.disconnect()
+ self._info = None
+
+ async def add_local_charm_dir(self, charm_dir, series):
+ """Upload a local charm to the model.
+
+ This will automatically generate an archive from
+ the charm dir.
+
+ :param charm_dir: Path to the charm directory
+ :param series: Charm series
+
+ """
+ fh = tempfile.NamedTemporaryFile()
+ CharmArchiveGenerator(charm_dir).make_archive(fh.name)
+ with fh:
+ func = partial(
+ self.add_local_charm, fh, series, os.stat(fh.name).st_size)
+ charm_url = await self._connector.loop.run_in_executor(None, func)
+
+ log.debug('Uploaded local charm: %s -> %s', charm_dir, charm_url)
+ return charm_url
+
+ def add_local_charm(self, charm_file, series, size=None):
+ """Upload a local charm archive to the model.
+
+ Returns the 'local:...' url that should be used to deploy the charm.
+
+ :param charm_file: Path to charm zip archive
+ :param series: Charm series
+ :param size: Size of the archive, in bytes
+ :return str: 'local:...' url for deploying the charm
+ :raises: :class:`JujuError` if the upload fails
+
+ Uses an https endpoint at the same host:port as the wss.
+ Supports large file uploads.
+
+ .. warning::
+
+ This method will block. Consider using :meth:`add_local_charm_dir`
+ instead.
+
+ """
+ conn, headers, path_prefix = self.connection().https_connection()
+ path = "%s/charms?series=%s" % (path_prefix, series)
+ headers['Content-Type'] = 'application/zip'
+ if size:
+ headers['Content-Length'] = size
+ conn.request("POST", path, charm_file, headers)
+ response = conn.getresponse()
+ result = response.read().decode()
+ if not response.status == 200:
+ raise JujuError(result)
+ result = json.loads(result)
+ return result['charm-url']
+
+ def all_units_idle(self):
+ """Return True if all units are idle.
+
+ """
+ for unit in self.units.values():
+ unit_status = unit.data['agent-status']['current']
+ if unit_status != 'idle':
+ return False
+ return True
+
+ async def reset(self, force=False):
+ """Reset the model to a clean state.
+
+ :param bool force: Force-terminate machines.
+
+ This returns only after the model has reached a clean state. "Clean"
+ means no applications or machines exist in the model.
+
+ """
+ log.debug('Resetting model')
+ for app in self.applications.values():
+ await app.destroy()
+ for machine in self.machines.values():
+ await machine.destroy(force=force)
+ await self.block_until(
+ lambda: len(self.machines) == 0
+ )
+
+ async def block_until(self, *conditions, timeout=None, wait_period=0.5):
+ """Return only after all conditions are true.
+
+ Raises `websockets.ConnectionClosed` if disconnected.
+ """
+ def _disconnected():
+ return not (self.is_connected() and self.connection().is_open)
+
+ def done():
+ return _disconnected() or all(c() for c in conditions)
+
+ await utils.block_until(done,
+ timeout=timeout,
+ wait_period=wait_period,
+ loop=self.loop)
+ if _disconnected():
+ raise websockets.ConnectionClosed(1006, 'no reason')
+
+ @property
+ def applications(self):
+ """Return a map of application-name:Application for all applications
+ currently in the model.
+
+ """
+ return self.state.applications
+
+ @property
+ def machines(self):
+ """Return a map of machine-id:Machine for all machines currently in
+ the model.
+
+ """
+ return self.state.machines
+
+ @property
+ def units(self):
+ """Return a map of unit-id:Unit for all units currently in
+ the model.
+
+ """
+ return self.state.units
+
+ @property
+ def relations(self):
+ """Return a list of all Relations currently in the model.
+
+ """
+ return list(self.state.relations.values())
+
+ 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.from_connection(self.connection())
+
+ self._info = await facade.ModelInfo()
+ log.debug('Got ModelInfo: %s', vars(self.info))
+
+ return self.info
+
+ @property
+ def info(self):
+ """Return the cached client.ModelInfo object for this Model.
+
+ If Model.get_info() has not been called, this will return None.
+ """
+ return self._info
+
+ def add_observer(
+ self, callable_, entity_type=None, action=None, entity_id=None,
+ predicate=None):
+ """Register an "on-model-change" callback
+
+ Once the model is connected, ``callable_``
+ will be called each time the model changes. ``callable_`` should
+ be Awaitable and accept the following positional arguments:
+
+ delta - An instance of :class:`juju.delta.EntityDelta`
+ containing the raw delta data recv'd from the Juju
+ websocket.
+
+ old_obj - If the delta modifies an existing object in the model,
+ old_obj will be a copy of that object, as it was before the
+ delta was applied. Will be None if the delta creates a new
+ entity in the model.
+
+ new_obj - A copy of the new or updated object, after the delta
+ is applied. Will be None if the delta removes an entity
+ from the model.
+
+ model - The :class:`Model` itself.
+
+ Events for which ``callable_`` is called can be specified by passing
+ entity_type, action, and/or entitiy_id filter criteria, e.g.::
+
+ add_observer(
+ myfunc,
+ entity_type='application', action='add', entity_id='ubuntu')
+
+ For more complex filtering conditions, pass a predicate function. It
+ will be called with a delta as its only argument. If the predicate
+ function returns True, the ``callable_`` will be called.
+
+ """
+ observer = _Observer(
+ callable_, entity_type, action, entity_id, predicate)
+ self._observers[observer] = callable_
+
+ def _watch(self):
+ """Start an asynchronous watch against this model.
+
+ See :meth:`add_observer` to register an onchange callback.
+
+ """
+ async def _all_watcher():
+ try:
+ 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,
+ loop=self._connector.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()
+ if monitor.status != monitor.CONNECTED:
+ # reconnect failed; abort and shutdown
+ log.error('Watcher: automatic reconnect '
+ 'failed; stopping watcher')
+ break
+ del allwatcher.Id
+ continue
+ else:
+ # closed on request, go ahead and shutdown
+ break
+ if self._watch_stopping.is_set():
+ try:
+ await allwatcher.Stop()
+ except websockets.ConnectionClosed:
+ pass # can't stop on a closed conn
+ break
+ for delta in results.deltas:
+ try:
+ delta = get_entity_delta(delta)
+ old_obj, new_obj = self.state.apply_delta(delta)
+ await self._notify_observers(delta, old_obj, new_obj)
+ except KeyError as e:
+ log.debug("unknown delta type: %s", e.args[0])
+ self._watch_received.set()
+ except CancelledError:
+ pass
+ except Exception:
+ log.exception('Error in watcher')
+ raise
+ finally:
+ self._watch_stopped.set()
+
+ log.debug('Starting watcher task')
+ self._watch_received.clear()
+ self._watch_stopping.clear()
+ self._watch_stopped.clear()
+ self._connector.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
+
+ :param delta: The raw change from the watcher
+ (:class:`juju.client.overrides.Delta`)
+ :param old_obj: The object in the model that this delta updates.
+ May be None.
+ :param new_obj: The object in the model that is created or updated
+ by applying this delta.
+
+ """
+ if new_obj and not old_obj:
+ delta.type = 'add'
+
+ log.debug(
+ 'Model changed: %s %s %s',
+ delta.entity, delta.type, delta.get_id())
+
+ for o in self._observers:
+ if o.cares_about(delta):
+ asyncio.ensure_future(o(delta, old_obj, new_obj, self),
+ loop=self._connector.loop)
+
+ async def _wait(self, entity_type, entity_id, action, predicate=None):
+ """
+ Block the calling routine until a given action has happened to the
+ given entity
+
+ :param entity_type: The entity's type.
+ :param entity_id: The entity's id.
+ :param action: the type of action (e.g., 'add', 'change', or 'remove')
+ :param predicate: optional callable that must take as an
+ argument a delta, and must return a boolean, indicating
+ whether the delta contains the specific action we're looking
+ for. For example, you might check to see whether a 'change'
+ has a 'completed' status. See the _Observer class for details.
+
+ """
+ q = asyncio.Queue(loop=self._connector.loop)
+
+ async def callback(delta, old, new, model):
+ await q.put(delta.get_id())
+
+ self.add_observer(callback, entity_type, action, entity_id, predicate)
+ entity_id = await q.get()
+ # object might not be in the entity_map if we were waiting for a
+ # 'remove' action
+ return self.state._live_entity_map(entity_type).get(entity_id)
+
+ async def _wait_for_new(self, entity_type, entity_id):
+ """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``
+ to appear in the model. This is similar to watching for the
+ object using ``block_until``, but uses the watcher rather than
+ polling.
+
+ """
+ # 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]
+ return await self._wait(entity_type, entity_id, None)
+
+ async def wait_for_action(self, action_id):
+ """Given an action, wait for it to complete."""
+
+ if action_id.startswith("action-"):
+ # if we've been passed action.tag, transform it into the
+ # id that the api deltas will use.
+ action_id = action_id[7:]
+
+ def predicate(delta):
+ return delta.data['status'] in ('completed', 'failed')
+
+ return await self._wait('action', action_id, None, predicate)
+
+ async def add_machine(
+ self, spec=None, constraints=None, disks=None, series=None):
+ """Start a new, empty machine and optionally a container, or add a
+ container to a machine.
+
+ :param str spec: Machine specification
+ Examples::
+
+ (None) - starts a new machine
+ 'lxd' - starts a new machine with one lxd container
+ 'lxd:4' - starts a new lxd container on machine 4
+ 'ssh:user@10.10.0.3:/path/to/private/key' - manually provision
+ a machine with ssh and the private key used for authentication
+ '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, 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, which can
+ contain the following keys::
+
+ count : int
+ pool : str
+ size : int
+
+ Example::
+
+ disks=[{
+ 'pool': 'rootfs',
+ 'size': 10 * GB,
+ 'count': 1,
+ }]
+
+ :param str series: Series, e.g. 'xenial'
+
+ Supported container types are: lxd, kvm
+
+ When deploying a container to an existing machine, constraints cannot
+ be used.
+
+ """
+ params = client.AddMachineParams()
+
+ if spec:
+ if spec.startswith("ssh:"):
+ placement, target, private_key_path = spec.split(":")
+ user, host = target.split("@")
+
+ sshProvisioner = provisioner.SSHProvisioner(
+ host=host,
+ user=user,
+ private_key_path=private_key_path,
+ )
+
+ params = sshProvisioner.provision_machine()
+ else:
+ placement = parse_placement(spec)
+ if placement:
+ params.placement = placement[0]
+
+ params.jobs = ['JobHostUnits']
+
+ if constraints:
+ params.constraints = client.Value.from_json(constraints)
+
+ if disks:
+ params.disks = [
+ client.Constraints.from_json(o) for o in disks]
+
+ if series:
+ params.series = series
+
+ # Submit the request.
+ 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)
+ machine_id = results.machines[0].machine
+
+ if spec:
+ if spec.startswith("ssh:"):
+ # Need to run this after AddMachines has been called,
+ # as we need the machine_id
+ await sshProvisioner.install_agent(
+ self.connection(),
+ params.nonce,
+ machine_id,
+ )
+
+ log.debug('Added new machine %s', machine_id)
+ return await self._wait_for_new('machine', machine_id)
+
+ async def add_relation(self, relation1, relation2):
+ """Add a relation between two applications.
+
+ :param str relation1: '<application>[:<relation_name>]'
+ :param str relation2: '<application>[:<relation_name>]'
+
+ """
+ connection = self.connection()
+ app_facade = client.ApplicationFacade.from_connection(connection)
+
+ log.debug(
+ 'Adding relation %s <-> %s', relation1, relation2)
+
+ def _find_relation(*specs):
+ for rel in self.relations:
+ if rel.matches(*specs):
+ return rel
+ return None
+
+ try:
+ result = await app_facade.AddRelation([relation1, relation2])
+ except JujuAPIError as e:
+ if 'relation already exists' not in e.message:
+ raise
+ rel = _find_relation(relation1, relation2)
+ if rel:
+ return rel
+ raise JujuError('Relation {} {} exists but not in model'.format(
+ relation1, relation2))
+
+ specs = ['{}:{}'.format(app, data['name'])
+ for app, data in result.endpoints.items()]
+
+ await self.block_until(lambda: _find_relation(*specs) is not None)
+ return _find_relation(*specs)
+
+ def add_space(self, name, *cidrs):
+ """Add a new network space.
+
+ Adds a new space with the given name and associates the given
+ (optional) list of existing subnet CIDRs with it.
+
+ :param str name: Name of the space
+ :param *cidrs: Optional list of existing subnet CIDRs
+
+ """
+ raise NotImplementedError()
+
+ 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
+
+ """
+ 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):
+ """Add an existing subnet to this model.
+
+ :param str cidr_or_id: CIDR or provider ID of the existing subnet
+ :param str space: Network space with which to associate
+ :param str *zones: Zone(s) in which the subnet resides
+
+ """
+ raise NotImplementedError()
+
+ def get_backups(self):
+ """Retrieve metadata for backups in this model.
+
+ """
+ raise NotImplementedError()
+
+ def block(self, *commands):
+ """Add a new block to this model.
+
+ :param str *commands: The commands to block. Valid values are
+ 'all-changes', 'destroy-model', 'remove-object'
+
+ """
+ raise NotImplementedError()
+
+ def get_blocks(self):
+ """List blocks for this model.
+
+ """
+ raise NotImplementedError()
+
+ def get_cached_images(self, arch=None, kind=None, series=None):
+ """Return a list of cached OS images.
+
+ :param str arch: Filter by image architecture
+ :param str kind: Filter by image kind, e.g. 'lxd'
+ :param str series: Filter by image series, e.g. 'xenial'
+
+ """
+ raise NotImplementedError()
+
+ def create_backup(self, note=None, no_download=False):
+ """Create a backup of this model.
+
+ :param str note: A note to store with the backup
+ :param bool no_download: Do not download the backup archive
+ :return str: Path to downloaded archive
+
+ """
+ raise NotImplementedError()
+
+ def create_storage_pool(self, name, provider_type, **pool_config):
+ """Create or define a storage pool.
+
+ :param str name: Name to give the storage pool
+ :param str provider_type: Pool provider type
+ :param **pool_config: key/value pool configuration pairs
+
+ """
+ raise NotImplementedError()
+
+ def debug_log(
+ self, no_tail=False, exclude_module=None, include_module=None,
+ include=None, level=None, limit=0, lines=10, replay=False,
+ exclude=None):
+ """Get log messages for this model.
+
+ :param bool no_tail: Stop after returning existing log messages
+ :param list exclude_module: Do not show log messages for these logging
+ modules
+ :param list include_module: Only show log messages for these logging
+ modules
+ :param list include: Only show log messages for these entities
+ :param str level: Log level to show, valid options are 'TRACE',
+ 'DEBUG', 'INFO', 'WARNING', 'ERROR,
+ :param int limit: Return this many of the most recent (possibly
+ filtered) lines are shown
+ :param int lines: Yield this many of the most recent lines, and keep
+ yielding
+ :param bool replay: Yield the entire log, and keep yielding
+ :param list exclude: Do not show log messages for these entities
+
+ """
+ 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,
+ channel=None, config=None, constraints=None, force=False,
+ num_units=1, plan=None, resources=None, series=None, storage=None,
+ to=None, devices=None):
+ """Deploy a new service or bundle.
+
+ :param str entity_url: Charm or bundle url
+ :param str application_name: Name to give the service
+ :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. 'edge'
+ :param dict config: Charm configuration dictionary
+ :param constraints: Service constraints
+ :type constraints: :class:`juju.Constraints`
+ :param bool force: Allow charm to be deployed to a machine running
+ an unsupported series
+ :param int num_units: Number of units to deploy
+ :param str plan: Plan under which to deploy charm
+ :param dict resources: <resource name>:<file path> pairs
+ :param str series: Series on which to deploy
+ :param dict storage: Storage constraints TODO how do these look?
+ :param to: Placement directive as a string. For example:
+
+ '23' - place on machine 23
+ 'lxd:7' - place in new lxd container on machine 7
+ '24/lxd/3' - place in container 3 on machine 24
+
+ If None, a new machine is provisioned.
+
+
+ TODO::
+
+ - support local resources
+
+ """
+ if storage:
+ storage = {
+ k: client.Constraints(**v)
+ for k, v in storage.items()
+ }
+
+ entity_path = Path(entity_url.replace('local:', ''))
+ bundle_path = entity_path / 'bundle.yaml'
+ metadata_path = entity_path / 'metadata.yaml'
+
+ is_local = (
+ entity_url.startswith('local:') or
+ entity_path.is_dir() or
+ entity_path.is_file()
+ )
+ if is_local:
+ entity_id = entity_url.replace('local:', '')
+ else:
+ entity = await self.charmstore.entity(entity_url, channel=channel,
+ include_stats=False)
+ entity_id = entity['Id']
+
+ client_facade = client.ClientFacade.from_connection(self.connection())
+
+ is_bundle = ((is_local and
+ (entity_id.endswith('.yaml') and entity_path.exists()) or
+ bundle_path.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()
+ extant_apps = {app for app in self.applications}
+ pending_apps = set(handler.applications) - extant_apps
+ if pending_apps:
+ # new apps will usually be in the model by now, but if some
+ # 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),
+ loop=self._connector.loop)
+ for app_name in pending_apps
+ ], loop=self._connector.loop)
+ return [app for name, app in self.applications.items()
+ if name in handler.applications]
+ else:
+ 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)
+ await client_facade.AddCharm(channel, entity_id)
+ # 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=entity)
+ else:
+ if not application_name:
+ metadata = yaml.load(metadata_path.read_text())
+ application_name = metadata['name']
+ # We have a local charm dir that needs to be uploaded
+ charm_dir = os.path.abspath(
+ os.path.expanduser(entity_id))
+ series = series or get_charm_series(charm_dir)
+ if not series:
+ raise JujuError(
+ "Couldn't determine series for charm at {}. "
+ "Pass a 'series' kwarg to Model.deploy().".format(
+ charm_dir))
+ entity_id = await self.add_local_charm_dir(charm_dir, series)
+ return await self._deploy(
+ charm_url=entity_id,
+ application=application_name,
+ series=series,
+ config=config or {},
+ constraints=constraints,
+ endpoint_bindings=bind,
+ resources=resources,
+ storage=storage,
+ channel=channel,
+ num_units=num_units,
+ placement=parse_placement(to),
+ devices=devices,
+ )
+
+ async def _add_store_resources(self, application, entity_url,
+ overrides=None, entity=None):
+ if not entity:
+ # avoid extra charm store call if one was already made
+ entity = await self.charmstore.entity(entity_url,
+ include_stats=False)
+ 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 overrides:
+ names = {r['name'] for r in resources}
+ unknown = overrides.keys() - names
+ if unknown:
+ raise JujuError('Unrecognized resource{}: {}'.format(
+ 's' if len(unknown) > 1 else '',
+ ', '.join(unknown)))
+ for resource in resources:
+ if resource['name'] in overrides:
+ resource['revision'] = overrides[resource['name']]
+
+ if not resources:
+ return None
+
+ 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,
+ devices=None):
+ """Logic shared between `Model.deploy` and `BundleHandler.deploy`.
+ """
+ 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,
+ devices=devices,
+ )
+ 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.
+
+ """
+ connection = self.connection()
+ app_facade = client.ApplicationFacade.from_connection(connection)
+
+ log.debug(
+ 'Destroying unit%s %s',
+ 's' if len(unit_names) == 1 else '',
+ ' '.join(unit_names))
+
+ return await app_facade.DestroyUnits(list(unit_names))
+ destroy_units = destroy_unit
+
+ def get_backup(self, archive_id):
+ """Download a backup archive file.
+
+ :param str archive_id: The id of the archive to download
+ :return str: Path to the archive file
+
+ """
+ raise NotImplementedError()
+
+ def enable_ha(
+ self, num_controllers=0, constraints=None, series=None, to=None):
+ """Ensure sufficient controllers exist to provide redundancy.
+
+ :param int num_controllers: Number of controllers to make available
+ :param constraints: Constraints to apply to the controller machines
+ :type constraints: :class:`juju.Constraints`
+ :param str series: Series of the controller machines
+ :param list to: Placement directives for controller machines, e.g.::
+
+ '23' - machine 23
+ 'lxc:7' - new lxc container on machine 7
+ '24/lxc/3' - lxc container 3 or machine 24
+
+ If None, a new machine is provisioned.
+
+ """
+ raise NotImplementedError()
+
+ 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.
+ """
+ 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
+
+ async def get_constraints(self):
+ """Return the machine constraints for this model.
+
+ :returns: A ``dict`` of constraints.
+ """
+ constraints = {}
+ client_facade = client.ClientFacade.from_connection(self.connection())
+ result = await client_facade.GetModelConstraints()
+
+ # GetModelConstraints returns GetConstraintsResults which has a
+ # 'constraints' attribute. If no constraints have been set
+ # GetConstraintsResults.constraints is None. Otherwise
+ # GetConstraintsResults.constraints has an attribute for each possible
+ # constraint, each of these in turn will be None if they have not been
+ # set.
+ if result.constraints:
+ constraint_types = [a for a in dir(result.constraints)
+ if a in Value._toSchema.keys()]
+ for constraint in constraint_types:
+ value = getattr(result.constraints, constraint)
+ if value is not None:
+ constraints[constraint] = getattr(result.constraints,
+ constraint)
+ return constraints
+
+ 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>
+
+ """
+ raise NotImplementedError()
+ import_ssh_keys = import_ssh_key
+
+ async def get_machines(self):
+ """Return list of machines in this model.
+
+ """
+ return list(self.state.machines.keys())
+
+ def get_shares(self):
+ """Return list of all users with access to this model.
+
+ """
+ raise NotImplementedError()
+
+ def get_spaces(self):
+ """Return list of all known spaces, including associated subnets.
+
+ """
+ raise NotImplementedError()
+
+ 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
+
+ """
+ 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):
+ """Return details of storage instances.
+
+ :param bool filesystem: Include filesystem storage
+ :param bool volume: Include volume storage
+
+ """
+ raise NotImplementedError()
+
+ def get_storage_pools(self, names=None, providers=None):
+ """Return list of storage pools.
+
+ :param list names: Only include pools with these names
+ :param list providers: Only include pools for these providers
+
+ """
+ raise NotImplementedError()
+
+ def get_subnets(self, space=None, zone=None):
+ """Return list of known subnets.
+
+ :param str space: Only include subnets in this space
+ :param str zone: Only include subnets in this zone
+
+ """
+ raise NotImplementedError()
+
+ def remove_blocks(self):
+ """Remove all blocks from this model.
+
+ """
+ raise NotImplementedError()
+
+ def remove_backup(self, backup_id):
+ """Delete a backup.
+
+ :param str backup_id: The id of the backup to remove
+
+ """
+ raise NotImplementedError()
+
+ def remove_cached_images(self, arch=None, kind=None, series=None):
+ """Remove cached OS images.
+
+ :param str arch: Architecture of the images to remove
+ :param str kind: Image kind to remove, e.g. 'lxd'
+ :param str series: Image series to remove, e.g. 'xenial'
+
+ """
+ raise NotImplementedError()
+
+ def remove_machine(self, *machine_ids):
+ """Remove a machine from this model.
+
+ :param str *machine_ids: Ids of the machines to remove
+
+ """
+ raise NotImplementedError()
+ remove_machines = remove_machine
+
+ async def remove_ssh_key(self, user, key):
+ """Remove a public SSH key(s) from this model.
+
+ :param str key: Full ssh key
+ :param str user: Juju user to which the key is registered
+
+ """
+ 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(
+ self, bootstrap=False, constraints=None, archive=None,
+ backup_id=None, upload_tools=False):
+ """Restore a backup archive to a new controller.
+
+ :param bool bootstrap: Bootstrap a new state machine
+ :param constraints: Model constraints
+ :type constraints: :class:`juju.Constraints`
+ :param str archive: Path to backup archive to restore
+ :param str backup_id: Id of backup to restore
+ :param bool upload_tools: Upload tools if bootstrapping a new machine
+
+ """
+ raise NotImplementedError()
+
+ def retry_provisioning(self):
+ """Retry provisioning for failed machines.
+
+ """
+ raise NotImplementedError()
+
+ def run(self, command, timeout=None):
+ """Run command on all machines in this model.
+
+ :param str command: The command to run
+ :param int timeout: Time to wait before command is considered failed
+
+ """
+ raise NotImplementedError()
+
+ async def set_config(self, config):
+ """Set configuration keys on this model.
+
+ :param dict config: Mapping of config keys to either string values or
+ `ConfigValue` instances, as returned by `get_config`.
+ """
+ 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)
+
+ async def set_constraints(self, constraints):
+ """Set machine constraints on this model.
+
+ :param dict config: Mapping of model constraints
+ """
+ client_facade = client.ClientFacade.from_connection(self.connection())
+ await client_facade.SetModelConstraints(
+ application='',
+ constraints=constraints)
+
+ async def get_action_output(self, action_uuid, wait=None):
+ """Get the results of an action by ID.
+
+ :param str action_uuid: Id of the action
+ :param int wait: Time in seconds to wait for action to complete.
+ :return dict: Output from action
+ :raises: :class:`JujuError` if invalid action_uuid
+ """
+ action_facade = client.ActionFacade.from_connection(
+ self.connection()
+ )
+ entity = [{'tag': tag.action(action_uuid)}]
+ # Cannot use self.wait_for_action as the action event has probably
+ # already happened and self.wait_for_action works by processing
+ # model deltas and checking if they match our type. If the action
+ # has already occured then the delta has gone.
+
+ async def _wait_for_action_status():
+ while True:
+ action_output = await action_facade.Actions(entity)
+ if action_output.results[0].status in ('completed', 'failed'):
+ return
+ else:
+ await asyncio.sleep(1)
+ await asyncio.wait_for(
+ _wait_for_action_status(),
+ timeout=wait)
+ action_output = await action_facade.Actions(entity)
+ # ActionResult.output is None if the action produced no output
+ if action_output.results[0].output is None:
+ output = {}
+ else:
+ output = action_output.results[0].output
+ return output
+
+ async def get_action_status(self, uuid_or_prefix=None, name=None):
+ """Get the status of all actions, filtered by ID, ID prefix, or name.
+
+ :param str uuid_or_prefix: Filter by action uuid or prefix
+ :param str name: Filter by action name
+
+ """
+ results = {}
+ action_results = []
+ action_facade = client.ActionFacade.from_connection(
+ self.connection()
+ )
+ if name:
+ name_results = await action_facade.FindActionsByNames([name])
+ action_results.extend(name_results.actions[0].actions)
+ if uuid_or_prefix:
+ # Collect list of actions matching uuid or prefix
+ matching_actions = await action_facade.FindActionTagsByPrefix(
+ [uuid_or_prefix])
+ entities = []
+ for actions in matching_actions.matches.values():
+ entities = [{'tag': a.tag} for a in actions]
+ # Get action results matching action tags
+ uuid_results = await action_facade.Actions(entities)
+ action_results.extend(uuid_results.results)
+ for a in action_results:
+ results[tag.untag('action-', a.action.tag)] = a.status
+ return results
+
+ def get_budget(self, budget_name):
+ """Get budget usage info.
+
+ :param str budget_name: Name of budget
+
+ """
+ raise NotImplementedError()
+
+ async def get_status(self, filters=None, utc=False):
+ """Return the status of the model.
+
+ :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
+
+ """
+ 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,
+ source=None, stream=None, version=None):
+ """Copy Juju tools into this model.
+
+ :param bool all_: Copy all versions, not just the latest
+ :param str destination: Path to local destination directory
+ :param bool dry_run: Don't do the actual copy
+ :param bool public: Tools are for a public cloud, so generate mirrors
+ information
+ :param str source: Path to local source directory
+ :param str stream: Simplestreams stream for which to sync metadata
+ :param str version: Copy a specific major.minor version
+
+ """
+ raise NotImplementedError()
+
+ def unblock(self, *commands):
+ """Unblock an operation that would alter this model.
+
+ :param str *commands: The commands to unblock. Valid values are
+ 'all-changes', 'destroy-model', 'remove-object'
+
+ """
+ raise NotImplementedError()
+
+ def unset_config(self, *keys):
+ """Unset configuration on this model.
+
+ :param str *keys: The keys to unset
+
+ """
+ raise NotImplementedError()
+
+ def upgrade_gui(self):
+ """Upgrade the Juju GUI for this model.
+
+ """
+ raise NotImplementedError()
+
+ def upgrade_juju(
+ self, dry_run=False, reset_previous_upgrade=False,
+ upload_tools=False, version=None):
+ """Upgrade Juju on all machines in a model.
+
+ :param bool dry_run: Don't do the actual upgrade
+ :param bool reset_previous_upgrade: Clear the previous (incomplete)
+ upgrade status
+ :param bool upload_tools: Upload local version of tools
+ :param str version: Upgrade to a specific version
+
+ """
+ raise NotImplementedError()
+
+ def upload_backup(self, archive_path):
+ """Store a backup archive remotely in Juju.
+
+ :param str archive_path: Path to local archive
+
+ """
+ raise NotImplementedError()
+
+ @property
+ 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.
+ :return: Dictionary of unit_name:metrics
+
+ """
+ log.debug("Retrieving metrics for %s",
+ ', '.join(tags) if tags else "all units")
+
+ metrics_facade = client.MetricsDebugFacade.from_connection(
+ 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
+
+
+def get_charm_series(path):
+ """Inspects the charm directory at ``path`` and returns a default
+ series from its metadata.yaml (the first item in the 'series' list).
+
+ Returns None if no series can be determined.
+
+ """
+ md = Path(path) / "metadata.yaml"
+ if not md.exists():
+ return None
+ data = yaml.load(md.open())
+ series = data.get('series')
+ return series[0] if series else None
+
+
+class BundleHandler:
+ """
+ Handle bundles by using the API to translate bundle YAML into a plan of
+ steps and then dispatching each of those using the API.
+ """
+ def __init__(self, model):
+ self.model = model
+ self.charmstore = model.charmstore
+ self.plan = []
+ self.references = {}
+ self._units_by_app = {}
+ for unit_name, unit in model.units.items():
+ app_units = self._units_by_app.setdefault(unit.application, [])
+ app_units.append(unit_name)
+ self.bundle_facade = client.BundleFacade.from_connection(
+ 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)
+ in the bundle. Upload the local charms to the model, and replace
+ the filesystem paths with appropriate 'local:' paths in the bundle.
+
+ Return the modified bundle.
+
+ :param dict bundle: Bundle dictionary
+ :return: Modified bundle dictionary
+
+ """
+ apps, args = [], []
+
+ default_series = bundle.get('series')
+ apps_dict = bundle.get('applications', bundle.get('services', {}))
+ for app_name in self.applications:
+ app_dict = apps_dict[app_name]
+ charm_dir = os.path.abspath(os.path.expanduser(app_dict['charm']))
+ if not os.path.isdir(charm_dir):
+ continue
+ series = (
+ app_dict.get('series') or
+ default_series or
+ get_charm_series(charm_dir)
+ )
+ if not series:
+ raise JujuError(
+ "Couldn't determine series for charm at {}. "
+ "Add a 'series' key to the bundle.".format(charm_dir))
+
+ # Keep track of what we need to update. We keep a list of apps
+ # that need to be updated, and a corresponding list of args
+ # needed to update those apps.
+ apps.append(app_name)
+ args.append((charm_dir, series))
+
+ if apps:
+ # If we have apps to update, spawn all the coroutines concurrently
+ # and wait for them to finish.
+ 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):
+ apps_dict[app_name]['charm'] = charm_url
+
+ return bundle
+
+ async def fetch_plan(self, entity_id):
+ is_store_url = entity_id.startswith('cs:')
+
+ if not is_store_url and os.path.isfile(entity_id):
+ bundle_yaml = Path(entity_id).read_text()
+ elif not is_store_url and os.path.isdir(entity_id):
+ 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.bundle = await self._handle_local_charms(self.bundle)
+
+ self.plan = await self.bundle_facade.GetChanges(
+ yaml.dump(self.bundle))
+
+ if self.plan.errors:
+ raise JujuError(self.plan.errors)
+
+ async def execute_plan(self):
+ for step in self.plan.changes:
+ method = getattr(self, step.method)
+ result = await method(*step.args)
+ self.references[step.id_] = result
+
+ @property
+ def applications(self):
+ apps_dict = self.bundle.get('applications',
+ self.bundle.get('services', {}))
+ return list(apps_dict.keys())
+
+ def resolve(self, reference):
+ if reference and reference.startswith('$'):
+ reference = self.references[reference[1:]]
+ return reference
+
+ async def addCharm(self, charm, series):
+ """
+ :param charm string:
+ Charm holds the URL of the charm to be added.
+
+ :param series string:
+ Series holds the series of the charm to be added
+ if the charm default is not sufficient.
+ """
+ # We don't add local charms because they've already been added
+ # by self._handle_local_charms
+ if charm.startswith('local:'):
+ return charm
+
+ entity_id = await self.charmstore.entityId(charm)
+ log.debug('Adding %s', entity_id)
+ await self.client_facade.AddCharm(None, entity_id)
+ return entity_id
+
+ 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 ""lxd" 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 {}
+
+ # Normalize keys
+ params = {normalize_key(k): params[k] for k in params.keys()}
+
+ # Fix up values, as necessary.
+ if 'parent_id' in params:
+ if params['parent_id'].startswith('$addUnit'):
+ unit = self.resolve(params['parent_id'])[0]
+ params['parent_id'] = unit.machine.entity_id
+ else:
+ params['parent_id'] = self.resolve(params['parent_id'])
+
+ params['constraints'] = parse_constraints(
+ params.get('constraints'))
+ params['jobs'] = params.get('jobs', ['JobHostUnits'])
+
+ if params.get('container_type') == 'lxc':
+ log.warning('Juju 2.0 does not support lxc containers. '
+ 'Converting containers to lxd.')
+ params['container_type'] = 'lxd'
+
+ # Submit the request.
+ 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):
+ """
+ :param endpoint1 string:
+ :param endpoint2 string:
+ Endpoint1 and Endpoint2 hold relation endpoints in the
+ "application:interface" form, where the application is always a
+ placeholder pointing to an application change, and the interface is
+ optional. Examples are "$deploy-42:web" or just "$deploy-42".
+ """
+ endpoints = [endpoint1, endpoint2]
+ # resolve indirect references
+ for i in range(len(endpoints)):
+ parts = endpoints[i].split(':')
+ parts[0] = self.resolve(parts[0])
+ endpoints[i] = ':'.join(parts)
+
+ log.info('Relating %s <-> %s', *endpoints)
+ return await self.model.add_relation(*endpoints)
+
+ async def deploy(self, charm, series, application, options, constraints,
+ storage, endpoint_bindings, *args):
+ """
+ :param charm string:
+ Charm holds the URL of the charm to be used to deploy this
+ application.
+
+ :param series string:
+ Series holds the series of the application to be deployed
+ if the charm default is not sufficient.
+
+ :param application string:
+ Application holds the application name.
+
+ :param options map[string]interface{}:
+ Options holds application options.
+
+ :param constraints string:
+ Constraints holds the optional application constraints.
+
+ :param storage map[string]string:
+ Storage holds the optional storage constraints.
+
+ :param endpoint_bindings map[string]string:
+ EndpointBindings holds the optional endpoint bindings
+
+ :param devices map[string]string:
+ Devices holds the optional devices constraints.
+ (Only given on Juju 2.5+)
+
+ :param resources map[string]int:
+ Resources identifies the revision to use for each resource
+ of the application's charm.
+
+ :param num_units int:
+ NumUnits holds the number of units required. For IAAS models, this
+ will be 0 and separate AddUnitChanges will be used. For Kubernetes
+ models, this will be used to scale the application.
+ (Only given on Juju 2.5+)
+ """
+ # resolve indirect references
+ charm = self.resolve(charm)
+
+ if len(args) == 1:
+ # Juju 2.4 and below only sends the resources
+ resources = args[0]
+ devices, num_units = None, None
+ else:
+ # Juju 2.5+ sends devices before resources, as well as num_units
+ # There might be placement but we need to ignore that.
+ devices, resources, num_units = args[:3]
+
+ if not charm.startswith('local:'):
+ resources = await self.model._add_store_resources(
+ application, charm, overrides=resources)
+ await self.model._deploy(
+ charm_url=charm,
+ application=application,
+ series=series,
+ config=options,
+ constraints=constraints,
+ endpoint_bindings=endpoint_bindings,
+ resources=resources,
+ storage=storage,
+ devices=devices,
+ num_units=num_units,
+ )
+ return application
+
+ async def addUnit(self, application, to):
+ """
+ :param application string:
+ Application holds the application placeholder name for which a unit
+ is added.
+
+ :param to string:
+ To holds the optional location where to add the unit, as a
+ placeholder pointing to another unit change or to a machine change.
+ """
+ application = self.resolve(application)
+ placement = self.resolve(to)
+ if self._units_by_app.get(application):
+ # enough units for this application already exist;
+ # claim one, and carry on
+ # NB: this should probably honor placement, but the juju client
+ # doesn't, so we're not bothering, either
+ unit_name = self._units_by_app[application].pop()
+ log.debug('Reusing unit %s for %s', unit_name, application)
+ return self.model.units[unit_name]
+
+ log.debug('Adding new unit for %s%s', application,
+ ' to %s' % placement if placement else '')
+ return await self.model.applications[application].add_unit(
+ count=1,
+ to=placement,
+ )
+
+ async def scale(self, application, scale):
+ """
+ Handle a change of scale to a k8s application.
+
+ :param string application:
+ Application holds the application placeholder name for which a unit
+ is added.
+
+ :param int scale:
+ New scale value to use.
+ """
+ application = self.resolve(application)
+ return await self.model.applications[application].scale(scale=scale)
+
+ async def expose(self, application):
+ """
+ :param application string:
+ Application holds the placeholder name of the application that must
+ be exposed.
+ """
+ application = self.resolve(application)
+ log.info('Exposing %s', application)
+ return await self.model.applications[application].expose()
+
+ async def setAnnotations(self, id_, entity_type, annotations):
+ """
+ :param id_ string:
+ Id is the placeholder for the application or machine change
+ corresponding to the entity to be annotated.
+
+ :param entity_type EntityType:
+ EntityType holds the type of the entity, "application" or
+ "machine".
+
+ :param annotations map[string]string:
+ Annotations holds the annotations as key/value pairs.
+ """
+ entity_id = self.resolve(id_)
+ try:
+ entity = self.model.state.get_entity(entity_type, entity_id)
+ except KeyError:
+ entity = await self.model._wait_for_new(entity_type, entity_id)
+ return await entity.set_annotations(annotations)
+
+
+class CharmStore:
+ """
+ Async wrapper around theblues.charmstore.CharmStore
+ """
+ def __init__(self, loop, cs_timeout=20):
+ self.loop = loop
+ self._cs = theblues.charmstore.CharmStore(timeout=cs_timeout)
+
+ def __getattr__(self, name):
+ """
+ Wrap method calls in coroutines that use run_in_executor to make them
+ async.
+ """
+ attr = getattr(self._cs, name)
+ if not callable(attr):
+ wrapper = partial(getattr, self._cs, name)
+ setattr(self, name, wrapper)
+ else:
+ async def coro(*args, **kwargs):
+ method = partial(attr, *args, **kwargs)
+ 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
+
+
+class CharmArchiveGenerator:
+ """
+ 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))
+
+ def make_archive(self, path):
+ """Create archive of directory and write to ``path``.
+
+ :param path: Path to archive
+
+ Ignored::
+
+ * build/* - This is used for packing the charm itself and any
+ similar tasks.
+ * */.* - Hidden files are all ignored for now. This will most
+ likely be changed into a specific ignore list
+ (.bzr, etc)
+
+ """
+ zf = zipfile.ZipFile(path, 'w', zipfile.ZIP_DEFLATED)
+ for dirpath, dirnames, filenames in os.walk(self.path):
+ relative_path = dirpath[len(self.path) + 1:]
+ if relative_path and not self._ignore(relative_path):
+ zf.write(dirpath, relative_path)
+ for name in filenames:
+ archive_name = os.path.join(relative_path, name)
+ if not self._ignore(archive_name):
+ real_path = os.path.join(dirpath, name)
+ self._check_type(real_path)
+ if os.path.islink(real_path):
+ self._check_link(real_path)
+ self._write_symlink(
+ zf, os.readlink(real_path), archive_name)
+ else:
+ zf.write(real_path, archive_name)
+ zf.close()
+ return path
+
+ def _check_type(self, path):
+ """Check the path
+ """
+ s = os.stat(path)
+ if stat.S_ISDIR(s.st_mode) or stat.S_ISREG(s.st_mode):
+ return path
+ raise ValueError("Invalid Charm at % %s" % (
+ path, "Invalid file type for a charm"))
+
+ def _check_link(self, path):
+ link_path = os.readlink(path)
+ if link_path[0] == "/":
+ raise ValueError(
+ "Invalid Charm at %s: %s" % (
+ path, "Absolute links are invalid"))
+ path_dir = os.path.dirname(path)
+ link_path = os.path.join(path_dir, link_path)
+ if not link_path.startswith(os.path.abspath(self.path)):
+ raise ValueError(
+ "Invalid charm at %s %s" % (
+ path, "Only internal symlinks are allowed"))
+
+ def _write_symlink(self, zf, link_target, link_path):
+ """Package symlinks with appropriate zipfile metadata."""
+ info = zipfile.ZipInfo()
+ info.filename = link_path
+ info.create_system = 3
+ # Magic code for symlinks / py2/3 compat
+ # 27166663808 = (stat.S_IFLNK | 0755) << 16
+ info.external_attr = 2716663808
+ zf.writestr(info, link_target)
+
+ def _ignore(self, path):
+ if path == "build" or path.startswith("build/"):
+ return True
+ if path.startswith('.'):
+ return True
projects / osm / N2VC.git / blobdiff