juju/model.py

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