juju/model.py

   1 import asyncio
   2 import collections
   3 import json
   4 import logging
   5 import os
   6 import re
   7 import stat
   8 import tempfile
   9 import weakref
  10 import zipfile
  11 from concurrent.futures import CancelledError
  12 from functools import partial
  13 from pathlib import Path
  14
  15 import yaml
  16 from theblues import charmstore
  17
  18 from .client import client
  19 from .client import watcher
  20 from .client import connection
  21 from .constraints import parse as parse_constraints, normalize_key
  22 from .delta import get_entity_delta
  23 from .delta import get_entity_class
  24 from .exceptions import DeadEntityException
  25 from .errors import JujuError, JujuAPIError
  26 from .placement import parse as parse_placement
  27
  28 log = logging.getLogger(__name__)
  29
  30
  31 class _Observer(object):
  32     """Wrapper around an observer callable.
  33
  34     This wrapper allows filter criteria to be associated with the
  35     callable so that it's only called for changes that meet the criteria.
  36
  37     """
  38     def __init__(self, callable_, entity_type, action, entity_id, predicate):
  39         self.callable_ = callable_
  40         self.entity_type = entity_type
  41         self.action = action
  42         self.entity_id = entity_id
  43         self.predicate = predicate
  44         if self.entity_id:
  45             self.entity_id = str(self.entity_id)
  46             if not self.entity_id.startswith('^'):
  47                 self.entity_id = '^' + self.entity_id
  48             if not self.entity_id.endswith('$'):
  49                 self.entity_id += '$'
  50
  51     async def __call__(self, delta, old, new, model):
  52         await self.callable_(delta, old, new, model)
  53
  54     def cares_about(self, delta):
  55         """Return True if this observer "cares about" (i.e. wants to be
  56         called) for a this delta.
  57
  58         """
  59         if (self.entity_id and delta.get_id() and
  60                 not re.match(self.entity_id, str(delta.get_id()))):
  61             return False
  62
  63         if self.entity_type and self.entity_type != delta.entity:
  64             return False
  65
  66         if self.action and self.action != delta.type:
  67             return False
  68
  69         if self.predicate and not self.predicate(delta):
  70             return False
  71
  72         return True
  73
  74
  75 class ModelObserver(object):
  76     async def __call__(self, delta, old, new, model):
  77         handler_name = 'on_{}_{}'.format(delta.entity, delta.type)
  78         method = getattr(self, handler_name, self.on_change)
  79         await method(delta, old, new, model)
  80
  81     async def on_change(self, delta, old, new, model):
  82         """Generic model-change handler.
  83
  84         :param delta: :class:`juju.client.overrides.Delta`
  85         :param old: :class:`juju.model.ModelEntity`
  86         :param new: :class:`juju.model.ModelEntity`
  87         :param model: :class:`juju.model.Model`
  88
  89         """
  90         pass
  91
  92
  93 class ModelState(object):
  94     """Holds the state of the model, including the delta history of all
  95     entities in the model.
  96
  97     """
  98     def __init__(self, model):
  99         self.model = model
 100         self.state = dict()
 101
 102     def _live_entity_map(self, entity_type):
 103         """Return an id:Entity map of all the living entities of
 104         type ``entity_type``.
 105
 106         """
 107         return {
 108             entity_id: self.get_entity(entity_type, entity_id)
 109             for entity_id, history in self.state.get(entity_type, {}).items()
 110             if history[-1] is not None
 111         }
 112
 113     @property
 114     def applications(self):
 115         """Return a map of application-name:Application for all applications
 116         currently in the model.
 117
 118         """
 119         return self._live_entity_map('application')
 120
 121     @property
 122     def machines(self):
 123         """Return a map of machine-id:Machine for all machines currently in
 124         the model.
 125
 126         """
 127         return self._live_entity_map('machine')
 128
 129     @property
 130     def units(self):
 131         """Return a map of unit-id:Unit for all units currently in
 132         the model.
 133
 134         """
 135         return self._live_entity_map('unit')
 136
 137     def entity_history(self, entity_type, entity_id):
 138         """Return the history deque for an entity.
 139
 140         """
 141         return self.state[entity_type][entity_id]
 142
 143     def entity_data(self, entity_type, entity_id, history_index):
 144         """Return the data dict for an entity at a specific index of its
 145         history.
 146
 147         """
 148         return self.entity_history(entity_type, entity_id)[history_index]
 149
 150     def apply_delta(self, delta):
 151         """Apply delta to our state and return a copy of the
 152         affected object as it was before and after the update, e.g.:
 153
 154             old_obj, new_obj = self.apply_delta(delta)
 155
 156         old_obj may be None if the delta is for the creation of a new object,
 157         e.g. a new application or unit is deployed.
 158
 159         new_obj will never be None, but may be dead (new_obj.dead == True)
 160         if the object was deleted as a result of the delta being applied.
 161
 162         """
 163         history = (
 164             self.state
 165             .setdefault(delta.entity, {})
 166             .setdefault(delta.get_id(), collections.deque())
 167         )
 168
 169         history.append(delta.data)
 170         if delta.type == 'remove':
 171             history.append(None)
 172
 173         entity = self.get_entity(delta.entity, delta.get_id())
 174         return entity.previous(), entity
 175
 176     def get_entity(
 177             self, entity_type, entity_id, history_index=-1, connected=True):
 178         """Return an object instance for the given entity_type and id.
 179
 180         By default the object state matches the most recent state from
 181         Juju. To get an instance of the object in an older state, pass
 182         history_index, an index into the history deque for the entity.
 183
 184         """
 185
 186         if history_index < 0 and history_index != -1:
 187             history_index += len(self.entity_history(entity_type, entity_id))
 188             if history_index < 0:
 189                 return None
 190
 191         try:
 192             self.entity_data(entity_type, entity_id, history_index)
 193         except IndexError:
 194             return None
 195
 196         entity_class = get_entity_class(entity_type)
 197         return entity_class(
 198             entity_id, self.model, history_index=history_index,
 199             connected=connected)
 200
 201
 202 class ModelEntity(object):
 203     """An object in the Model tree"""
 204
 205     def __init__(self, entity_id, model, history_index=-1, connected=True):
 206         """Initialize a new entity
 207
 208         :param entity_id str: The unique id of the object in the model
 209         :param model: The model instance in whose object tree this
 210             entity resides
 211         :history_index int: The index of this object's state in the model's
 212             history deque for this entity
 213         :connected bool: Flag indicating whether this object gets live updates
 214             from the model.
 215
 216         """
 217         self.entity_id = entity_id
 218         self.model = model
 219         self._history_index = history_index
 220         self.connected = connected
 221         self.connection = model.connection
 222
 223     def __repr__(self):
 224         return '<{} entity_id="{}">'.format(type(self).__name__,
 225                                             self.entity_id)
 226
 227     def __getattr__(self, name):
 228         """Fetch object attributes from the underlying data dict held in the
 229         model.
 230
 231         """
 232         return self.safe_data[name]
 233
 234     def __bool__(self):
 235         return bool(self.data)
 236
 237     def on_change(self, callable_):
 238         """Add a change observer to this entity.
 239
 240         """
 241         self.model.add_observer(
 242             callable_, self.entity_type, 'change', self.entity_id)
 243
 244     def on_remove(self, callable_):
 245         """Add a remove observer to this entity.
 246
 247         """
 248         self.model.add_observer(
 249             callable_, self.entity_type, 'remove', self.entity_id)
 250
 251     @property
 252     def entity_type(self):
 253         """A string identifying the entity type of this object, e.g.
 254         'application' or 'unit', etc.
 255
 256         """
 257         return self.__class__.__name__.lower()
 258
 259     @property
 260     def current(self):
 261         """Return True if this object represents the current state of the
 262         entity in the underlying model.
 263
 264         This will be True except when the object represents an entity at a
 265         non-latest state in history, e.g. if the object was obtained by calling
 266         .previous() on another object.
 267
 268         """
 269         return self._history_index == -1
 270
 271     @property
 272     def dead(self):
 273         """Returns True if this entity no longer exists in the underlying
 274         model.
 275
 276         """
 277         return (
 278             self.data is None or
 279             self.model.state.entity_data(
 280                 self.entity_type, self.entity_id, -1) is None
 281         )
 282
 283     @property
 284     def alive(self):
 285         """Returns True if this entity still exists in the underlying
 286         model.
 287
 288         """
 289         return not self.dead
 290
 291     @property
 292     def data(self):
 293         """The data dictionary for this entity.
 294
 295         """
 296         return self.model.state.entity_data(
 297             self.entity_type, self.entity_id, self._history_index)
 298
 299     @property
 300     def safe_data(self):
 301         """The data dictionary for this entity.
 302
 303         If this `ModelEntity` points to the dead state, it will
 304         raise `DeadEntityException`.
 305
 306         """
 307         if self.data is None:
 308             raise DeadEntityException(
 309                 "Entity {}:{} is dead - its attributes can no longer be "
 310                 "accessed. Use the .previous() method on this object to get "
 311                 "a copy of the object at its previous state.".format(
 312                     self.entity_type, self.entity_id))
 313         return self.data
 314
 315     def previous(self):
 316         """Return a copy of this object as was at its previous state in
 317         history.
 318
 319         Returns None if this object is new (and therefore has no history).
 320
 321         The returned object is always "disconnected", i.e. does not receive
 322         live updates.
 323
 324         """
 325         return self.model.state.get_entity(
 326             self.entity_type, self.entity_id, self._history_index - 1,
 327             connected=False)
 328
 329     def next(self):
 330         """Return a copy of this object at its next state in
 331         history.
 332
 333         Returns None if this object is already the latest.
 334
 335         The returned object is "disconnected", i.e. does not receive
 336         live updates, unless it is current (latest).
 337
 338         """
 339         if self._history_index == -1:
 340             return None
 341
 342         new_index = self._history_index + 1
 343         connected = (
 344             new_index == len(self.model.state.entity_history(
 345                 self.entity_type, self.entity_id)) - 1
 346         )
 347         return self.model.state.get_entity(
 348             self.entity_type, self.entity_id, self._history_index - 1,
 349             connected=connected)
 350
 351     def latest(self):
 352         """Return a copy of this object at its current state in the model.
 353
 354         Returns self if this object is already the latest.
 355
 356         The returned object is always "connected", i.e. receives
 357         live updates from the model.
 358
 359         """
 360         if self._history_index == -1:
 361             return self
 362
 363         return self.model.state.get_entity(self.entity_type, self.entity_id)
 364
 365
 366 class Model(object):
 367     def __init__(self, loop=None):
 368         """Instantiate a new connected Model.
 369
 370         :param loop: an asyncio event loop
 371
 372         """
 373         self.loop = loop or asyncio.get_event_loop()
 374         self.connection = None
 375         self.observers = weakref.WeakValueDictionary()
 376         self.state = ModelState(self)
 377         self.info = None
 378         self._watcher_task = None
 379         self._watch_shutdown = asyncio.Event(loop=loop)
 380         self._watch_received = asyncio.Event(loop=loop)
 381         self._charmstore = CharmStore(self.loop)
 382
 383     async def connect(self, *args, **kw):
 384         """Connect to an arbitrary Juju model.
 385
 386         args and kw are passed through to Connection.connect()
 387
 388         """
 389         self.connection = await connection.Connection.connect(*args, **kw)
 390         await self._after_connect()
 391
 392     async def connect_current(self):
 393         """Connect to the current Juju model.
 394
 395         """
 396         self.connection = await connection.Connection.connect_current()
 397         await self._after_connect()
 398
 399     async def connect_model(self, model_name):
 400         """Connect to a specific Juju model by name.
 401
 402         :param model_name:  Format [controller:][user/]model
 403
 404         """
 405         self.connection = await connection.Connection.connect_model(model_name)
 406         await self._after_connect()
 407
 408     async def _after_connect(self):
 409         """Run initialization steps after connecting to websocket.
 410
 411         """
 412         self._watch()
 413         await self._watch_received.wait()
 414         await self.get_info()
 415
 416     async def disconnect(self):
 417         """Shut down the watcher task and close websockets.
 418
 419         """
 420         self._stop_watching()
 421         if self.connection and self.connection.is_open:
 422             await self._watch_shutdown.wait()
 423             log.debug('Closing model connection')
 424             await self.connection.close()
 425             self.connection = None
 426
 427     async def add_local_charm_dir(self, charm_dir, series):
 428         """Upload a local charm to the model.
 429
 430         This will automatically generate an archive from
 431         the charm dir.
 432
 433         :param charm_dir: Path to the charm directory
 434         :param series: Charm series
 435
 436         """
 437         fh = tempfile.NamedTemporaryFile()
 438         CharmArchiveGenerator(charm_dir).make_archive(fh.name)
 439         with fh:
 440             func = partial(
 441                 self.add_local_charm, fh, series, os.stat(fh.name).st_size)
 442             charm_url = await self.loop.run_in_executor(None, func)
 443
 444         log.debug('Uploaded local charm: %s -> %s', charm_dir, charm_url)
 445         return charm_url
 446
 447     def add_local_charm(self, charm_file, series, size=None):
 448         """Upload a local charm archive to the model.
 449
 450         Returns the 'local:...' url that should be used to deploy the charm.
 451
 452         :param charm_file: Path to charm zip archive
 453         :param series: Charm series
 454         :param size: Size of the archive, in bytes
 455         :return str: 'local:...' url for deploying the charm
 456         :raises: :class:`JujuError` if the upload fails
 457
 458         Uses an https endpoint at the same host:port as the wss.
 459         Supports large file uploads.
 460
 461         .. warning::
 462
 463            This method will block. Consider using :meth:`add_local_charm_dir`
 464            instead.
 465
 466         """
 467         conn, headers, path_prefix = self.connection.https_connection()
 468         path = "%s/charms?series=%s" % (path_prefix, series)
 469         headers['Content-Type'] = 'application/zip'
 470         if size:
 471             headers['Content-Length'] = size
 472         conn.request("POST", path, charm_file, headers)
 473         response = conn.getresponse()
 474         result = response.read().decode()
 475         if not response.status == 200:
 476             raise JujuError(result)
 477         result = json.loads(result)
 478         return result['charm-url']
 479
 480     def all_units_idle(self):
 481         """Return True if all units are idle.
 482
 483         """
 484         for unit in self.units.values():
 485             unit_status = unit.data['agent-status']['current']
 486             if unit_status != 'idle':
 487                 return False
 488         return True
 489
 490     async def reset(self, force=False):
 491         """Reset the model to a clean state.
 492
 493         :param bool force: Force-terminate machines.
 494
 495         This returns only after the model has reached a clean state. "Clean"
 496         means no applications or machines exist in the model.
 497
 498         """
 499         log.debug('Resetting model')
 500         for app in self.applications.values():
 501             await app.destroy()
 502         for machine in self.machines.values():
 503             await machine.destroy(force=force)
 504         await self.block_until(
 505             lambda: len(self.machines) == 0
 506         )
 507
 508     async def block_until(self, *conditions, timeout=None, wait_period=0.5):
 509         """Return only after all conditions are true.
 510
 511         """
 512         async def _block():
 513             while not all(c() for c in conditions):
 514                 await asyncio.sleep(wait_period)
 515         await asyncio.wait_for(_block(), timeout)
 516
 517     @property
 518     def applications(self):
 519         """Return a map of application-name:Application for all applications
 520         currently in the model.
 521
 522         """
 523         return self.state.applications
 524
 525     @property
 526     def machines(self):
 527         """Return a map of machine-id:Machine for all machines currently in
 528         the model.
 529
 530         """
 531         return self.state.machines
 532
 533     @property
 534     def units(self):
 535         """Return a map of unit-id:Unit for all units currently in
 536         the model.
 537
 538         """
 539         return self.state.units
 540
 541     async def get_info(self):
 542         """Return a client.ModelInfo object for this Model.
 543
 544         Retrieves latest info for this Model from the api server. The
 545         return value is cached on the Model.info attribute so that the
 546         valued may be accessed again without another api call, if
 547         desired.
 548
 549         This method is called automatically when the Model is connected,
 550         resulting in Model.info being initialized without requiring an
 551         explicit call to this method.
 552
 553         """
 554         facade = client.ClientFacade()
 555         facade.connect(self.connection)
 556
 557         self.info = await facade.ModelInfo()
 558         log.debug('Got ModelInfo: %s', vars(self.info))
 559
 560         return self.info
 561
 562     def add_observer(
 563             self, callable_, entity_type=None, action=None, entity_id=None,
 564             predicate=None):
 565         """Register an "on-model-change" callback
 566
 567         Once the model is connected, ``callable_``
 568         will be called each time the model changes. ``callable_`` should
 569         be Awaitable and accept the following positional arguments:
 570
 571             delta - An instance of :class:`juju.delta.EntityDelta`
 572                 containing the raw delta data recv'd from the Juju
 573                 websocket.
 574
 575             old_obj - If the delta modifies an existing object in the model,
 576                 old_obj will be a copy of that object, as it was before the
 577                 delta was applied. Will be None if the delta creates a new
 578                 entity in the model.
 579
 580             new_obj - A copy of the new or updated object, after the delta
 581                 is applied. Will be None if the delta removes an entity
 582                 from the model.
 583
 584             model - The :class:`Model` itself.
 585
 586         Events for which ``callable_`` is called can be specified by passing
 587         entity_type, action, and/or entitiy_id filter criteria, e.g.::
 588
 589             add_observer(
 590                 myfunc,
 591                 entity_type='application', action='add', entity_id='ubuntu')
 592
 593         For more complex filtering conditions, pass a predicate function. It
 594         will be called with a delta as its only argument. If the predicate
 595         function returns True, the ``callable_`` will be called.
 596
 597         """
 598         observer = _Observer(
 599             callable_, entity_type, action, entity_id, predicate)
 600         self.observers[observer] = callable_
 601
 602     def _watch(self):
 603         """Start an asynchronous watch against this model.
 604
 605         See :meth:`add_observer` to register an onchange callback.
 606
 607         """
 608         async def _start_watch():
 609             self._watch_shutdown.clear()
 610             try:
 611                 allwatcher = watcher.AllWatcher()
 612                 self._watch_conn = await self.connection.clone()
 613                 allwatcher.connect(self._watch_conn)
 614                 while True:
 615                     results = await allwatcher.Next()
 616                     for delta in results.deltas:
 617                         delta = get_entity_delta(delta)
 618                         old_obj, new_obj = self.state.apply_delta(delta)
 619                         # XXX: Might not want to shield at this level
 620                         # We are shielding because when the watcher is
 621                         # canceled (on disconnect()), we don't want all of
 622                         # its children (every observer callback) to be
 623                         # canceled with it. So we shield them. But this means
 624                         # they can *never* be canceled.
 625                         await asyncio.shield(
 626                             self._notify_observers(delta, old_obj, new_obj))
 627                     self._watch_received.set()
 628             except CancelledError:
 629                 log.debug('Closing watcher connection')
 630                 await self._watch_conn.close()
 631                 self._watch_shutdown.set()
 632                 self._watch_conn = None
 633
 634         log.debug('Starting watcher task')
 635         self._watcher_task = self.loop.create_task(_start_watch())
 636
 637     def _stop_watching(self):
 638         """Stop the asynchronous watch against this model.
 639
 640         """
 641         log.debug('Stopping watcher task')
 642         if self._watcher_task:
 643             self._watcher_task.cancel()
 644
 645     async def _notify_observers(self, delta, old_obj, new_obj):
 646         """Call observing callbacks, notifying them of a change in model state
 647
 648         :param delta: The raw change from the watcher
 649             (:class:`juju.client.overrides.Delta`)
 650         :param old_obj: The object in the model that this delta updates.
 651             May be None.
 652         :param new_obj: The object in the model that is created or updated
 653             by applying this delta.
 654
 655         """
 656         if new_obj and not old_obj:
 657             delta.type = 'add'
 658
 659         log.debug(
 660             'Model changed: %s %s %s',
 661             delta.entity, delta.type, delta.get_id())
 662
 663         for o in self.observers:
 664             if o.cares_about(delta):
 665                 asyncio.ensure_future(o(delta, old_obj, new_obj, self))
 666
 667     async def _wait(self, entity_type, entity_id, action, predicate=None):
 668         """
 669         Block the calling routine until a given action has happened to the
 670         given entity
 671
 672         :param entity_type: The entity's type.
 673         :param entity_id: The entity's id.
 674         :param action: the type of action (e.g., 'add', 'change', or 'remove')
 675         :param predicate: optional callable that must take as an
 676             argument a delta, and must return a boolean, indicating
 677             whether the delta contains the specific action we're looking
 678             for. For example, you might check to see whether a 'change'
 679             has a 'completed' status. See the _Observer class for details.
 680
 681         """
 682         q = asyncio.Queue(loop=self.loop)
 683
 684         async def callback(delta, old, new, model):
 685             await q.put(delta.get_id())
 686
 687         self.add_observer(callback, entity_type, action, entity_id, predicate)
 688         entity_id = await q.get()
 689         # object might not be in the entity_map if we were waiting for a
 690         # 'remove' action
 691         return self.state._live_entity_map(entity_type).get(entity_id)
 692
 693     async def _wait_for_new(self, entity_type, entity_id=None, predicate=None):
 694         """Wait for a new object to appear in the Model and return it.
 695
 696         Waits for an object of type ``entity_type`` with id ``entity_id``.
 697         If ``entity_id`` is ``None``, it will wait for the first new entity
 698         of the correct type.
 699
 700         This coroutine blocks until the new object appears in the model.
 701
 702         """
 703         # if the entity is already in the model, just return it
 704         if entity_id in self.state._live_entity_map(entity_type):
 705             return self.state._live_entity_map(entity_type)[entity_id]
 706         # if we know the entity_id, we can trigger on any action that puts
 707         # the enitty into the model; otherwise, we have to watch for the
 708         # next "add" action on that entity_type
 709         action = 'add' if entity_id is None else None
 710         return await self._wait(entity_type, entity_id, action, predicate)
 711
 712     async def wait_for_action(self, action_id):
 713         """Given an action, wait for it to complete."""
 714
 715         if action_id.startswith("action-"):
 716             # if we've been passed action.tag, transform it into the
 717             # id that the api deltas will use.
 718             action_id = action_id[7:]
 719
 720         def predicate(delta):
 721             return delta.data['status'] in ('completed', 'failed')
 722
 723         return await self._wait('action', action_id, 'change', predicate)
 724
 725     async def add_machine(
 726             self, spec=None, constraints=None, disks=None, series=None):
 727         """Start a new, empty machine and optionally a container, or add a
 728         container to a machine.
 729
 730         :param str spec: Machine specification
 731             Examples::
 732
 733                 (None) - starts a new machine
 734                 'lxd' - starts a new machine with one lxd container
 735                 'lxd:4' - starts a new lxd container on machine 4
 736                 'ssh:user@10.10.0.3' - manually provisions a machine with ssh
 737                 'zone=us-east-1a' - starts a machine in zone us-east-1s on AWS
 738                 'maas2.name' - acquire machine maas2.name on MAAS
 739
 740         :param dict constraints: Machine constraints
 741             Example::
 742
 743                 constraints={
 744                     'mem': 256 * MB,
 745                 }
 746
 747         :param list disks: List of disk constraint dictionaries
 748             Example::
 749
 750                 disks=[{
 751                     'pool': 'rootfs',
 752                     'size': 10 * GB,
 753                     'count': 1,
 754                 }]
 755
 756         :param str series: Series, e.g. 'xenial'
 757
 758         Supported container types are: lxd, kvm
 759
 760         When deploying a container to an existing machine, constraints cannot
 761         be used.
 762
 763         """
 764         params = client.AddMachineParams()
 765         params.jobs = ['JobHostUnits']
 766
 767         if spec:
 768             placement = parse_placement(spec)
 769             if placement:
 770                 params.placement = placement[0]
 771
 772         if constraints:
 773             params.constraints = client.Value.from_json(constraints)
 774
 775         if disks:
 776             params.disks = [
 777                 client.Constraints.from_json(o) for o in disks]
 778
 779         if series:
 780             params.series = series
 781
 782         # Submit the request.
 783         client_facade = client.ClientFacade()
 784         client_facade.connect(self.connection)
 785         results = await client_facade.AddMachines([params])
 786         error = results.machines[0].error
 787         if error:
 788             raise ValueError("Error adding machine: %s", error.message)
 789         machine_id = results.machines[0].machine
 790         log.debug('Added new machine %s', machine_id)
 791         return await self._wait_for_new('machine', machine_id)
 792
 793     async def add_relation(self, relation1, relation2):
 794         """Add a relation between two applications.
 795
 796         :param str relation1: '<application>[:<relation_name>]'
 797         :param str relation2: '<application>[:<relation_name>]'
 798
 799         """
 800         app_facade = client.ApplicationFacade()
 801         app_facade.connect(self.connection)
 802
 803         log.debug(
 804             'Adding relation %s <-> %s', relation1, relation2)
 805
 806         try:
 807             result = await app_facade.AddRelation([relation1, relation2])
 808         except JujuAPIError as e:
 809             if 'relation already exists' not in e.message:
 810                 raise
 811             log.debug(
 812                 'Relation %s <-> %s already exists', relation1, relation2)
 813             # TODO: if relation already exists we should return the
 814             # Relation ModelEntity here
 815             return None
 816
 817         def predicate(delta):
 818             endpoints = {}
 819             for endpoint in delta.data['endpoints']:
 820                 endpoints[endpoint['application-name']] = endpoint['relation']
 821             return endpoints == result.endpoints
 822
 823         return await self._wait_for_new('relation', None, predicate)
 824
 825     def add_space(self, name, *cidrs):
 826         """Add a new network space.
 827
 828         Adds a new space with the given name and associates the given
 829         (optional) list of existing subnet CIDRs with it.
 830
 831         :param str name: Name of the space
 832         :param \*cidrs: Optional list of existing subnet CIDRs
 833
 834         """
 835         raise NotImplementedError()
 836
 837     def add_ssh_key(self, key):
 838         """Add a public SSH key to this model.
 839
 840         :param str key: The public ssh key
 841
 842         """
 843         raise NotImplementedError()
 844     add_ssh_keys = add_ssh_key
 845
 846     def add_subnet(self, cidr_or_id, space, *zones):
 847         """Add an existing subnet to this model.
 848
 849         :param str cidr_or_id: CIDR or provider ID of the existing subnet
 850         :param str space: Network space with which to associate
 851         :param str \*zones: Zone(s) in which the subnet resides
 852
 853         """
 854         raise NotImplementedError()
 855
 856     def get_backups(self):
 857         """Retrieve metadata for backups in this model.
 858
 859         """
 860         raise NotImplementedError()
 861
 862     def block(self, *commands):
 863         """Add a new block to this model.
 864
 865         :param str \*commands: The commands to block. Valid values are
 866             'all-changes', 'destroy-model', 'remove-object'
 867
 868         """
 869         raise NotImplementedError()
 870
 871     def get_blocks(self):
 872         """List blocks for this model.
 873
 874         """
 875         raise NotImplementedError()
 876
 877     def get_cached_images(self, arch=None, kind=None, series=None):
 878         """Return a list of cached OS images.
 879
 880         :param str arch: Filter by image architecture
 881         :param str kind: Filter by image kind, e.g. 'lxd'
 882         :param str series: Filter by image series, e.g. 'xenial'
 883
 884         """
 885         raise NotImplementedError()
 886
 887     def create_backup(self, note=None, no_download=False):
 888         """Create a backup of this model.
 889
 890         :param str note: A note to store with the backup
 891         :param bool no_download: Do not download the backup archive
 892         :return str: Path to downloaded archive
 893
 894         """
 895         raise NotImplementedError()
 896
 897     def create_storage_pool(self, name, provider_type, **pool_config):
 898         """Create or define a storage pool.
 899
 900         :param str name: Name to give the storage pool
 901         :param str provider_type: Pool provider type
 902         :param \*\*pool_config: key/value pool configuration pairs
 903
 904         """
 905         raise NotImplementedError()
 906
 907     def debug_log(
 908             self, no_tail=False, exclude_module=None, include_module=None,
 909             include=None, level=None, limit=0, lines=10, replay=False,
 910             exclude=None):
 911         """Get log messages for this model.
 912
 913         :param bool no_tail: Stop after returning existing log messages
 914         :param list exclude_module: Do not show log messages for these logging
 915             modules
 916         :param list include_module: Only show log messages for these logging
 917             modules
 918         :param list include: Only show log messages for these entities
 919         :param str level: Log level to show, valid options are 'TRACE',
 920             'DEBUG', 'INFO', 'WARNING', 'ERROR,
 921         :param int limit: Return this many of the most recent (possibly
 922             filtered) lines are shown
 923         :param int lines: Yield this many of the most recent lines, and keep
 924             yielding
 925         :param bool replay: Yield the entire log, and keep yielding
 926         :param list exclude: Do not show log messages for these entities
 927
 928         """
 929         raise NotImplementedError()
 930
 931     async def deploy(
 932             self, entity_url, application_name=None, bind=None, budget=None,
 933             channel=None, config=None, constraints=None, force=False,
 934             num_units=1, plan=None, resources=None, series=None, storage=None,
 935             to=None):
 936         """Deploy a new service or bundle.
 937
 938         :param str entity_url: Charm or bundle url
 939         :param str application_name: Name to give the service
 940         :param dict bind: <charm endpoint>:<network space> pairs
 941         :param dict budget: <budget name>:<limit> pairs
 942         :param str channel: Charm store channel from which to retrieve
 943             the charm or bundle, e.g. 'development'
 944         :param dict config: Charm configuration dictionary
 945         :param constraints: Service constraints
 946         :type constraints: :class:`juju.Constraints`
 947         :param bool force: Allow charm to be deployed to a machine running
 948             an unsupported series
 949         :param int num_units: Number of units to deploy
 950         :param str plan: Plan under which to deploy charm
 951         :param dict resources: <resource name>:<file path> pairs
 952         :param str series: Series on which to deploy
 953         :param dict storage: Storage constraints TODO how do these look?
 954         :param to: Placement directive as a string. For example:
 955
 956             '23' - place on machine 23
 957             'lxd:7' - place in new lxd container on machine 7
 958             '24/lxd/3' - place in container 3 on machine 24
 959
 960             If None, a new machine is provisioned.
 961
 962
 963         TODO::
 964
 965             - application_name is required; fill this in automatically if not
 966               provided by caller
 967             - series is required; how do we pick a default?
 968
 969         """
 970         if to:
 971             placement = parse_placement(to)
 972         else:
 973             placement = []
 974
 975         if storage:
 976             storage = {
 977                 k: client.Constraints(**v)
 978                 for k, v in storage.items()
 979             }
 980
 981         is_local = (
 982             entity_url.startswith('local:') or
 983             os.path.isdir(entity_url)
 984         )
 985         entity_id = await self.charmstore.entityId(entity_url) \
 986             if not is_local else entity_url
 987
 988         app_facade = client.ApplicationFacade()
 989         client_facade = client.ClientFacade()
 990         app_facade.connect(self.connection)
 991         client_facade.connect(self.connection)
 992
 993         is_bundle = ((is_local and
 994                       (Path(entity_id) / 'bundle.yaml').exists()) or
 995                      (not is_local and 'bundle/' in entity_id))
 996
 997         if is_bundle:
 998             handler = BundleHandler(self)
 999             await handler.fetch_plan(entity_id)
1000             await handler.execute_plan()
1001             extant_apps = {app for app in self.applications}
1002             pending_apps = set(handler.applications) - extant_apps
1003             if pending_apps:
1004                 # new apps will usually be in the model by now, but if some
1005                 # haven't made it yet we'll need to wait on them to be added
1006                 await asyncio.gather(*[
1007                     asyncio.ensure_future(
1008                         self._wait_for_new('application', app_name))
1009                     for app_name in pending_apps
1010                 ])
1011             return [app for name, app in self.applications.items()
1012                     if name in handler.applications]
1013         else:
1014             log.debug(
1015                 'Deploying %s', entity_id)
1016
1017             if not is_local:
1018                 await client_facade.AddCharm(channel, entity_id)
1019             elif not entity_id.startswith('local:'):
1020                 # We have a local charm dir that needs to be uploaded
1021                 charm_dir = os.path.abspath(
1022                     os.path.expanduser(entity_id))
1023                 series = series or get_charm_series(charm_dir)
1024                 if not series:
1025                     raise JujuError(
1026                         "Couldn't determine series for charm at {}. "
1027                         "Pass a 'series' kwarg to Model.deploy().".format(
1028                             charm_dir))
1029                 entity_id = await self.add_local_charm_dir(charm_dir, series)
1030
1031             app = client.ApplicationDeploy(
1032                 application=application_name,
1033                 channel=channel,
1034                 charm_url=entity_id,
1035                 config=config,
1036                 constraints=parse_constraints(constraints),
1037                 endpoint_bindings=bind,
1038                 num_units=num_units,
1039                 resources=resources,
1040                 series=series,
1041                 storage=storage,
1042             )
1043             app.placement = placement
1044
1045             result = await app_facade.Deploy([app])
1046             errors = [r.error.message for r in result.results if r.error]
1047             if errors:
1048                 raise JujuError('\n'.join(errors))
1049             return await self._wait_for_new('application', application_name)
1050
1051     def destroy(self):
1052         """Terminate all machines and resources for this model.
1053
1054         """
1055         raise NotImplementedError()
1056
1057     async def destroy_unit(self, *unit_names):
1058         """Destroy units by name.
1059
1060         """
1061         app_facade = client.ApplicationFacade()
1062         app_facade.connect(self.connection)
1063
1064         log.debug(
1065             'Destroying unit%s %s',
1066             's' if len(unit_names) == 1 else '',
1067             ' '.join(unit_names))
1068
1069         return await app_facade.DestroyUnits(list(unit_names))
1070     destroy_units = destroy_unit
1071
1072     def get_backup(self, archive_id):
1073         """Download a backup archive file.
1074
1075         :param str archive_id: The id of the archive to download
1076         :return str: Path to the archive file
1077
1078         """
1079         raise NotImplementedError()
1080
1081     def enable_ha(
1082             self, num_controllers=0, constraints=None, series=None, to=None):
1083         """Ensure sufficient controllers exist to provide redundancy.
1084
1085         :param int num_controllers: Number of controllers to make available
1086         :param constraints: Constraints to apply to the controller machines
1087         :type constraints: :class:`juju.Constraints`
1088         :param str series: Series of the controller machines
1089         :param list to: Placement directives for controller machines, e.g.::
1090
1091             '23' - machine 23
1092             'lxc:7' - new lxc container on machine 7
1093             '24/lxc/3' - lxc container 3 or machine 24
1094
1095             If None, a new machine is provisioned.
1096
1097         """
1098         raise NotImplementedError()
1099
1100     def get_config(self):
1101         """Return the configuration settings for this model.
1102
1103         """
1104         raise NotImplementedError()
1105
1106     def get_constraints(self):
1107         """Return the machine constraints for this model.
1108
1109         """
1110         raise NotImplementedError()
1111
1112     def grant(self, username, acl='read'):
1113         """Grant a user access to this model.
1114
1115         :param str username: Username
1116         :param str acl: Access control ('read' or 'write')
1117
1118         """
1119         raise NotImplementedError()
1120
1121     def import_ssh_key(self, identity):
1122         """Add a public SSH key from a trusted indentity source to this model.
1123
1124         :param str identity: User identity in the form <lp|gh>:<username>
1125
1126         """
1127         raise NotImplementedError()
1128     import_ssh_keys = import_ssh_key
1129
1130     def get_machines(self, machine, utc=False):
1131         """Return list of machines in this model.
1132
1133         :param str machine: Machine id, e.g. '0'
1134         :param bool utc: Display time as UTC in RFC3339 format
1135
1136         """
1137         raise NotImplementedError()
1138
1139     def get_shares(self):
1140         """Return list of all users with access to this model.
1141
1142         """
1143         raise NotImplementedError()
1144
1145     def get_spaces(self):
1146         """Return list of all known spaces, including associated subnets.
1147
1148         """
1149         raise NotImplementedError()
1150
1151     def get_ssh_key(self):
1152         """Return known SSH keys for this model.
1153
1154         """
1155         raise NotImplementedError()
1156     get_ssh_keys = get_ssh_key
1157
1158     def get_storage(self, filesystem=False, volume=False):
1159         """Return details of storage instances.
1160
1161         :param bool filesystem: Include filesystem storage
1162         :param bool volume: Include volume storage
1163
1164         """
1165         raise NotImplementedError()
1166
1167     def get_storage_pools(self, names=None, providers=None):
1168         """Return list of storage pools.
1169
1170         :param list names: Only include pools with these names
1171         :param list providers: Only include pools for these providers
1172
1173         """
1174         raise NotImplementedError()
1175
1176     def get_subnets(self, space=None, zone=None):
1177         """Return list of known subnets.
1178
1179         :param str space: Only include subnets in this space
1180         :param str zone: Only include subnets in this zone
1181
1182         """
1183         raise NotImplementedError()
1184
1185     def remove_blocks(self):
1186         """Remove all blocks from this model.
1187
1188         """
1189         raise NotImplementedError()
1190
1191     def remove_backup(self, backup_id):
1192         """Delete a backup.
1193
1194         :param str backup_id: The id of the backup to remove
1195
1196         """
1197         raise NotImplementedError()
1198
1199     def remove_cached_images(self, arch=None, kind=None, series=None):
1200         """Remove cached OS images.
1201
1202         :param str arch: Architecture of the images to remove
1203         :param str kind: Image kind to remove, e.g. 'lxd'
1204         :param str series: Image series to remove, e.g. 'xenial'
1205
1206         """
1207         raise NotImplementedError()
1208
1209     def remove_machine(self, *machine_ids):
1210         """Remove a machine from this model.
1211
1212         :param str \*machine_ids: Ids of the machines to remove
1213
1214         """
1215         raise NotImplementedError()
1216     remove_machines = remove_machine
1217
1218     def remove_ssh_key(self, *keys):
1219         """Remove a public SSH key(s) from this model.
1220
1221         :param str \*keys: Keys to remove
1222
1223         """
1224         raise NotImplementedError()
1225     remove_ssh_keys = remove_ssh_key
1226
1227     def restore_backup(
1228             self, bootstrap=False, constraints=None, archive=None,
1229             backup_id=None, upload_tools=False):
1230         """Restore a backup archive to a new controller.
1231
1232         :param bool bootstrap: Bootstrap a new state machine
1233         :param constraints: Model constraints
1234         :type constraints: :class:`juju.Constraints`
1235         :param str archive: Path to backup archive to restore
1236         :param str backup_id: Id of backup to restore
1237         :param bool upload_tools: Upload tools if bootstrapping a new machine
1238
1239         """
1240         raise NotImplementedError()
1241
1242     def retry_provisioning(self):
1243         """Retry provisioning for failed machines.
1244
1245         """
1246         raise NotImplementedError()
1247
1248     def revoke(self, username, acl='read'):
1249         """Revoke a user's access to this model.
1250
1251         :param str username: Username to revoke
1252         :param str acl: Access control ('read' or 'write')
1253
1254         """
1255         raise NotImplementedError()
1256
1257     def run(self, command, timeout=None):
1258         """Run command on all machines in this model.
1259
1260         :param str command: The command to run
1261         :param int timeout: Time to wait before command is considered failed
1262
1263         """
1264         raise NotImplementedError()
1265
1266     def set_config(self, **config):
1267         """Set configuration keys on this model.
1268
1269         :param \*\*config: Config key/values
1270
1271         """
1272         raise NotImplementedError()
1273
1274     def set_constraints(self, constraints):
1275         """Set machine constraints on this model.
1276
1277         :param :class:`juju.Constraints` constraints: Machine constraints
1278
1279         """
1280         raise NotImplementedError()
1281
1282     def get_action_output(self, action_uuid, wait=-1):
1283         """Get the results of an action by ID.
1284
1285         :param str action_uuid: Id of the action
1286         :param int wait: Time in seconds to wait for action to complete
1287
1288         """
1289         raise NotImplementedError()
1290
1291     def get_action_status(self, uuid_or_prefix=None, name=None):
1292         """Get the status of all actions, filtered by ID, ID prefix, or action name.
1293
1294         :param str uuid_or_prefix: Filter by action uuid or prefix
1295         :param str name: Filter by action name
1296
1297         """
1298         raise NotImplementedError()
1299
1300     def get_budget(self, budget_name):
1301         """Get budget usage info.
1302
1303         :param str budget_name: Name of budget
1304
1305         """
1306         raise NotImplementedError()
1307
1308     async def get_status(self, filters=None, utc=False):
1309         """Return the status of the model.
1310
1311         :param str filters: Optional list of applications, units, or machines
1312             to include, which can use wildcards ('*').
1313         :param bool utc: Display time as UTC in RFC3339 format
1314
1315         """
1316         client_facade = client.ClientFacade()
1317         client_facade.connect(self.connection)
1318         return await client_facade.FullStatus(filters)
1319
1320     def sync_tools(
1321             self, all_=False, destination=None, dry_run=False, public=False,
1322             source=None, stream=None, version=None):
1323         """Copy Juju tools into this model.
1324
1325         :param bool all_: Copy all versions, not just the latest
1326         :param str destination: Path to local destination directory
1327         :param bool dry_run: Don't do the actual copy
1328         :param bool public: Tools are for a public cloud, so generate mirrors
1329             information
1330         :param str source: Path to local source directory
1331         :param str stream: Simplestreams stream for which to sync metadata
1332         :param str version: Copy a specific major.minor version
1333
1334         """
1335         raise NotImplementedError()
1336
1337     def unblock(self, *commands):
1338         """Unblock an operation that would alter this model.
1339
1340         :param str \*commands: The commands to unblock. Valid values are
1341             'all-changes', 'destroy-model', 'remove-object'
1342
1343         """
1344         raise NotImplementedError()
1345
1346     def unset_config(self, *keys):
1347         """Unset configuration on this model.
1348
1349         :param str \*keys: The keys to unset
1350
1351         """
1352         raise NotImplementedError()
1353
1354     def upgrade_gui(self):
1355         """Upgrade the Juju GUI for this model.
1356
1357         """
1358         raise NotImplementedError()
1359
1360     def upgrade_juju(
1361             self, dry_run=False, reset_previous_upgrade=False,
1362             upload_tools=False, version=None):
1363         """Upgrade Juju on all machines in a model.
1364
1365         :param bool dry_run: Don't do the actual upgrade
1366         :param bool reset_previous_upgrade: Clear the previous (incomplete)
1367             upgrade status
1368         :param bool upload_tools: Upload local version of tools
1369         :param str version: Upgrade to a specific version
1370
1371         """
1372         raise NotImplementedError()
1373
1374     def upload_backup(self, archive_path):
1375         """Store a backup archive remotely in Juju.
1376
1377         :param str archive_path: Path to local archive
1378
1379         """
1380         raise NotImplementedError()
1381
1382     @property
1383     def charmstore(self):
1384         return self._charmstore
1385
1386     async def get_metrics(self, *tags):
1387         """Retrieve metrics.
1388
1389         :param str \*tags: Tags of entities from which to retrieve metrics.
1390             No tags retrieves the metrics of all units in the model.
1391         :return: Dictionary of unit_name:metrics
1392
1393         """
1394         log.debug("Retrieving metrics for %s",
1395                   ', '.join(tags) if tags else "all units")
1396
1397         metrics_facade = client.MetricsDebugFacade()
1398         metrics_facade.connect(self.connection)
1399
1400         entities = [client.Entity(tag) for tag in tags]
1401         metrics_result = await metrics_facade.GetMetrics(entities)
1402
1403         metrics = collections.defaultdict(list)
1404
1405         for entity_metrics in metrics_result.results:
1406             error = entity_metrics.error
1407             if error:
1408                 if "is not a valid tag" in error:
1409                     raise ValueError(error.message)
1410                 else:
1411                     raise Exception(error.message)
1412
1413             for metric in entity_metrics.metrics:
1414                 metrics[metric.unit].append(vars(metric))
1415
1416         return metrics
1417
1418
1419 def get_charm_series(path):
1420     """Inspects the charm directory at ``path`` and returns a default
1421     series from its metadata.yaml (the first item in the 'series' list).
1422
1423     Returns None if no series can be determined.
1424
1425     """
1426     md = Path(path) / "metadata.yaml"
1427     if not md.exists():
1428         return None
1429     data = yaml.load(md.open())
1430     series = data.get('series')
1431     return series[0] if series else None
1432
1433
1434 class BundleHandler(object):
1435     """
1436     Handle bundles by using the API to translate bundle YAML into a plan of
1437     steps and then dispatching each of those using the API.
1438     """
1439     def __init__(self, model):
1440         self.model = model
1441         self.charmstore = model.charmstore
1442         self.plan = []
1443         self.references = {}
1444         self._units_by_app = {}
1445         for unit_name, unit in model.units.items():
1446             app_units = self._units_by_app.setdefault(unit.application, [])
1447             app_units.append(unit_name)
1448         self.client_facade = client.ClientFacade()
1449         self.client_facade.connect(model.connection)
1450         self.app_facade = client.ApplicationFacade()
1451         self.app_facade.connect(model.connection)
1452         self.ann_facade = client.AnnotationsFacade()
1453         self.ann_facade.connect(model.connection)
1454
1455     async def _handle_local_charms(self, bundle):
1456         """Search for references to local charms (i.e. filesystem paths)
1457         in the bundle. Upload the local charms to the model, and replace
1458         the filesystem paths with appropriate 'local:' paths in the bundle.
1459
1460         Return the modified bundle.
1461
1462         :param dict bundle: Bundle dictionary
1463         :return: Modified bundle dictionary
1464
1465         """
1466         apps, args = [], []
1467
1468         default_series = bundle.get('series')
1469         for app_name in self.applications:
1470             app_dict = bundle['services'][app_name]
1471             charm_dir = os.path.abspath(os.path.expanduser(app_dict['charm']))
1472             if not os.path.isdir(charm_dir):
1473                 continue
1474             series = (
1475                 app_dict.get('series') or
1476                 default_series or
1477                 get_charm_series(charm_dir)
1478             )
1479             if not series:
1480                 raise JujuError(
1481                     "Couldn't determine series for charm at {}. "
1482                     "Add a 'series' key to the bundle.".format(charm_dir))
1483
1484             # Keep track of what we need to update. We keep a list of apps
1485             # that need to be updated, and a corresponding list of args
1486             # needed to update those apps.
1487             apps.append(app_name)
1488             args.append((charm_dir, series))
1489
1490         if apps:
1491             # If we have apps to update, spawn all the coroutines concurrently
1492             # and wait for them to finish.
1493             charm_urls = await asyncio.gather(*[
1494                 self.model.add_local_charm_dir(*params)
1495                 for params in args
1496             ])
1497             # Update the 'charm:' entry for each app with the new 'local:' url.
1498             for app_name, charm_url in zip(apps, charm_urls):
1499                 bundle['services'][app_name]['charm'] = charm_url
1500
1501         return bundle
1502
1503     async def fetch_plan(self, entity_id):
1504         is_local = not entity_id.startswith('cs:') and os.path.isdir(entity_id)
1505         if is_local:
1506             bundle_yaml = (Path(entity_id) / "bundle.yaml").read_text()
1507         else:
1508             bundle_yaml = await self.charmstore.files(entity_id,
1509                                                       filename='bundle.yaml',
1510                                                       read_file=True)
1511         self.bundle = yaml.safe_load(bundle_yaml)
1512         self.bundle = await self._handle_local_charms(self.bundle)
1513
1514         self.plan = await self.client_facade.GetBundleChanges(
1515             yaml.dump(self.bundle))
1516
1517         if self.plan.errors:
1518             raise JujuError('\n'.join(self.plan.errors))
1519
1520     async def execute_plan(self):
1521         for step in self.plan.changes:
1522             method = getattr(self, step.method)
1523             result = await method(*step.args)
1524             self.references[step.id_] = result
1525
1526     @property
1527     def applications(self):
1528         return list(self.bundle['services'].keys())
1529
1530     def resolve(self, reference):
1531         if reference and reference.startswith('$'):
1532             reference = self.references[reference[1:]]
1533         return reference
1534
1535     async def addCharm(self, charm, series):
1536         """
1537         :param charm string:
1538             Charm holds the URL of the charm to be added.
1539
1540         :param series string:
1541             Series holds the series of the charm to be added
1542             if the charm default is not sufficient.
1543         """
1544         # We don't add local charms because they've already been added
1545         # by self._handle_local_charms
1546         if charm.startswith('local:'):
1547             return charm
1548
1549         entity_id = await self.charmstore.entityId(charm)
1550         log.debug('Adding %s', entity_id)
1551         await self.client_facade.AddCharm(None, entity_id)
1552         return entity_id
1553
1554     async def addMachines(self, params=None):
1555         """
1556         :param params dict:
1557             Dictionary specifying the machine to add. All keys are optional.
1558             Keys include:
1559
1560             series: string specifying the machine OS series.
1561
1562             constraints: string holding machine constraints, if any. We'll
1563                 parse this into the json friendly dict that the juju api
1564                 expects.
1565
1566             container_type: string holding the type of the container (for
1567                 instance ""lxd" or kvm"). It is not specified for top level
1568                 machines.
1569
1570             parent_id: string holding a placeholder pointing to another
1571                 machine change or to a unit change. This value is only
1572                 specified in the case this machine is a container, in
1573                 which case also ContainerType is set.
1574
1575         """
1576         params = params or {}
1577
1578         # Normalize keys
1579         params = {normalize_key(k): params[k] for k in params.keys()}
1580
1581         # Fix up values, as necessary.
1582         if 'parent_id' in params:
1583             params['parent_id'] = self.resolve(params['parent_id'])
1584
1585         params['constraints'] = parse_constraints(
1586             params.get('constraints'))
1587         params['jobs'] = params.get('jobs', ['JobHostUnits'])
1588
1589         if params.get('container_type') == 'lxc':
1590             log.warning('Juju 2.0 does not support lxc containers. '
1591                         'Converting containers to lxd.')
1592             params['container_type'] = 'lxd'
1593
1594         # Submit the request.
1595         params = client.AddMachineParams(**params)
1596         results = await self.client_facade.AddMachines([params])
1597         error = results.machines[0].error
1598         if error:
1599             raise ValueError("Error adding machine: %s", error.message)
1600         machine = results.machines[0].machine
1601         log.debug('Added new machine %s', machine)
1602         return machine
1603
1604     async def addRelation(self, endpoint1, endpoint2):
1605         """
1606         :param endpoint1 string:
1607         :param endpoint2 string:
1608             Endpoint1 and Endpoint2 hold relation endpoints in the
1609             "application:interface" form, where the application is always a
1610             placeholder pointing to an application change, and the interface is
1611             optional. Examples are "$deploy-42:web" or just "$deploy-42".
1612         """
1613         endpoints = [endpoint1, endpoint2]
1614         # resolve indirect references
1615         for i in range(len(endpoints)):
1616             parts = endpoints[i].split(':')
1617             parts[0] = self.resolve(parts[0])
1618             endpoints[i] = ':'.join(parts)
1619
1620         log.info('Relating %s <-> %s', *endpoints)
1621         return await self.model.add_relation(*endpoints)
1622
1623     async def deploy(self, charm, series, application, options, constraints,
1624                      storage, endpoint_bindings, resources):
1625         """
1626         :param charm string:
1627             Charm holds the URL of the charm to be used to deploy this
1628             application.
1629
1630         :param series string:
1631             Series holds the series of the application to be deployed
1632             if the charm default is not sufficient.
1633
1634         :param application string:
1635             Application holds the application name.
1636
1637         :param options map[string]interface{}:
1638             Options holds application options.
1639
1640         :param constraints string:
1641             Constraints holds the optional application constraints.
1642
1643         :param storage map[string]string:
1644             Storage holds the optional storage constraints.
1645
1646         :param endpoint_bindings map[string]string:
1647             EndpointBindings holds the optional endpoint bindings
1648
1649         :param resources map[string]int:
1650             Resources identifies the revision to use for each resource
1651             of the application's charm.
1652         """
1653         # resolve indirect references
1654         charm = self.resolve(charm)
1655         # stringify all config values for API, and convert to YAML
1656         options = {k: str(v) for k, v in options.items()}
1657         options = yaml.dump({application: options}, default_flow_style=False)
1658         # build param object
1659         app = client.ApplicationDeploy(
1660             charm_url=charm,
1661             series=series,
1662             application=application,
1663             # Pass options to config-yaml rather than config, as
1664             # config-yaml invokes a newer codepath that better handles
1665             # empty strings in the options values.
1666             config_yaml=options,
1667             constraints=parse_constraints(constraints),
1668             storage=storage,
1669             endpoint_bindings=endpoint_bindings,
1670             resources=resources,
1671         )
1672         # do the do
1673         log.info('Deploying %s', charm)
1674         await self.app_facade.Deploy([app])
1675         # ensure the app is in the model for future operations
1676         await self.model._wait_for_new('application', application)
1677         return application
1678
1679     async def addUnit(self, application, to):
1680         """
1681         :param application string:
1682             Application holds the application placeholder name for which a unit
1683             is added.
1684
1685         :param to string:
1686             To holds the optional location where to add the unit, as a
1687             placeholder pointing to another unit change or to a machine change.
1688         """
1689         application = self.resolve(application)
1690         placement = self.resolve(to)
1691         if self._units_by_app.get(application):
1692             # enough units for this application already exist;
1693             # claim one, and carry on
1694             # NB: this should probably honor placement, but the juju client
1695             # doesn't, so we're not bothering, either
1696             unit_name = self._units_by_app[application].pop()
1697             log.debug('Reusing unit %s for %s', unit_name, application)
1698             return self.model.units[unit_name]
1699
1700         log.debug('Adding new unit for %s%s', application,
1701                   ' to %s' % placement if placement else '')
1702         return await self.model.applications[application].add_unit(
1703             count=1,
1704             to=placement,
1705         )
1706
1707     async def expose(self, application):
1708         """
1709         :param application string:
1710             Application holds the placeholder name of the application that must
1711             be exposed.
1712         """
1713         application = self.resolve(application)
1714         log.info('Exposing %s', application)
1715         return await self.model.applications[application].expose()
1716
1717     async def setAnnotations(self, id_, entity_type, annotations):
1718         """
1719         :param id_ string:
1720             Id is the placeholder for the application or machine change
1721             corresponding to the entity to be annotated.
1722
1723         :param entity_type EntityType:
1724             EntityType holds the type of the entity, "application" or
1725             "machine".
1726
1727         :param annotations map[string]string:
1728             Annotations holds the annotations as key/value pairs.
1729         """
1730         entity_id = self.resolve(id_)
1731         try:
1732             entity = self.model.state.get_entity(entity_type, entity_id)
1733         except KeyError:
1734             entity = await self.model._wait_for_new(entity_type, entity_id)
1735         return await entity.set_annotations(annotations)
1736
1737
1738 class CharmStore(object):
1739     """
1740     Async wrapper around theblues.charmstore.CharmStore
1741     """
1742     def __init__(self, loop):
1743         self.loop = loop
1744         self._cs = charmstore.CharmStore()
1745
1746     def __getattr__(self, name):
1747         """
1748         Wrap method calls in coroutines that use run_in_executor to make them
1749         async.
1750         """
1751         attr = getattr(self._cs, name)
1752         if not callable(attr):
1753             wrapper = partial(getattr, self._cs, name)
1754             setattr(self, name, wrapper)
1755         else:
1756             async def coro(*args, **kwargs):
1757                 method = partial(attr, *args, **kwargs)
1758                 return await self.loop.run_in_executor(None, method)
1759             setattr(self, name, coro)
1760             wrapper = coro
1761         return wrapper
1762
1763
1764 class CharmArchiveGenerator(object):
1765     def __init__(self, path):
1766         self.path = os.path.abspath(os.path.expanduser(path))
1767
1768     def make_archive(self, path):
1769         """Create archive of directory and write to ``path``.
1770
1771         :param path: Path to archive
1772
1773         Ignored::
1774
1775             * build/\* - This is used for packing the charm itself and any
1776                           similar tasks.
1777             * \*/.\*    - Hidden files are all ignored for now.  This will most
1778                           likely be changed into a specific ignore list
1779                           (.bzr, etc)
1780
1781         """
1782         zf = zipfile.ZipFile(path, 'w', zipfile.ZIP_DEFLATED)
1783         for dirpath, dirnames, filenames in os.walk(self.path):
1784             relative_path = dirpath[len(self.path) + 1:]
1785             if relative_path and not self._ignore(relative_path):
1786                 zf.write(dirpath, relative_path)
1787             for name in filenames:
1788                 archive_name = os.path.join(relative_path, name)
1789                 if not self._ignore(archive_name):
1790                     real_path = os.path.join(dirpath, name)
1791                     self._check_type(real_path)
1792                     if os.path.islink(real_path):
1793                         self._check_link(real_path)
1794                         self._write_symlink(
1795                             zf, os.readlink(real_path), archive_name)
1796                     else:
1797                         zf.write(real_path, archive_name)
1798         zf.close()
1799         return path
1800
1801     def _check_type(self, path):
1802         """Check the path
1803         """
1804         s = os.stat(path)
1805         if stat.S_ISDIR(s.st_mode) or stat.S_ISREG(s.st_mode):
1806             return path
1807         raise ValueError("Invalid Charm at % %s" % (
1808             path, "Invalid file type for a charm"))
1809
1810     def _check_link(self, path):
1811         link_path = os.readlink(path)
1812         if link_path[0] == "/":
1813             raise ValueError(
1814                 "Invalid Charm at %s: %s" % (
1815                     path, "Absolute links are invalid"))
1816         path_dir = os.path.dirname(path)
1817         link_path = os.path.join(path_dir, link_path)
1818         if not link_path.startswith(os.path.abspath(self.path)):
1819             raise ValueError(
1820                 "Invalid charm at %s %s" % (
1821                     path, "Only internal symlinks are allowed"))
1822
1823     def _write_symlink(self, zf, link_target, link_path):
1824         """Package symlinks with appropriate zipfile metadata."""
1825         info = zipfile.ZipInfo()
1826         info.filename = link_path
1827         info.create_system = 3
1828         # Magic code for symlinks / py2/3 compat
1829         # 27166663808 = (stat.S_IFLNK | 0755) << 16
1830         info.external_attr = 2716663808
1831         zf.writestr(info, link_target)
1832
1833     def _ignore(self, path):
1834         if path == "build" or path.startswith("build/"):
1835             return True
1836         if path.startswith('.'):
1837             return True