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