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