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. 'edge'
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, channel=channel)
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                 await client_facade.AddCharm(channel, entity_id)
1076                 # XXX: we're dropping local resources here, but we don't
1077                 # actually support them yet anyway
1078                 resources = await self._add_store_resources(application_name,
1079                                                             entity_id,
1080                                                             entity)
1081             else:
1082                 # We have a local charm dir that needs to be uploaded
1083                 charm_dir = os.path.abspath(
1084                     os.path.expanduser(entity_id))
1085                 series = series or get_charm_series(charm_dir)
1086                 if not series:
1087                     raise JujuError(
1088                         "Couldn't determine series for charm at {}. "
1089                         "Pass a 'series' kwarg to Model.deploy().".format(
1090                             charm_dir))
1091                 entity_id = await self.add_local_charm_dir(charm_dir, series)
1092             return await self._deploy(
1093                 charm_url=entity_id,
1094                 application=application_name,
1095                 series=series,
1096                 config=config or {},
1097                 constraints=constraints,
1098                 endpoint_bindings=bind,
1099                 resources=resources,
1100                 storage=storage,
1101                 channel=channel,
1102                 num_units=num_units,
1103                 placement=parse_placement(to)
1104             )
1105
1106     async def _add_store_resources(self, application, entity_url, entity=None):
1107         if not entity:
1108             # avoid extra charm store call if one was already made
1109             entity = await self.charmstore.entity(entity_url)
1110         resources = [
1111             {
1112                 'description': resource['Description'],
1113                 'fingerprint': resource['Fingerprint'],
1114                 'name': resource['Name'],
1115                 'path': resource['Path'],
1116                 'revision': resource['Revision'],
1117                 'size': resource['Size'],
1118                 'type_': resource['Type'],
1119                 'origin': 'store',
1120             } for resource in entity['Meta']['resources']
1121         ]
1122
1123         if not resources:
1124             return None
1125
1126         resources_facade = client.ResourcesFacade.from_connection(
1127             self.connection)
1128         response = await resources_facade.AddPendingResources(
1129             tag.application(application),
1130             entity_url,
1131             [client.CharmResource(**resource) for resource in resources])
1132         resource_map = {resource['name']: pid
1133                         for resource, pid
1134                         in zip(resources, response.pending_ids)}
1135         return resource_map
1136
1137     async def _deploy(self, charm_url, application, series, config,
1138                       constraints, endpoint_bindings, resources, storage,
1139                       channel=None, num_units=None, placement=None):
1140         """Logic shared between `Model.deploy` and `BundleHandler.deploy`.
1141         """
1142         log.info('Deploying %s', charm_url)
1143
1144         # stringify all config values for API, and convert to YAML
1145         config = {k: str(v) for k, v in config.items()}
1146         config = yaml.dump({application: config},
1147                            default_flow_style=False)
1148
1149         app_facade = client.ApplicationFacade.from_connection(
1150             self.connection)
1151
1152         app = client.ApplicationDeploy(
1153             charm_url=charm_url,
1154             application=application,
1155             series=series,
1156             channel=channel,
1157             config_yaml=config,
1158             constraints=parse_constraints(constraints),
1159             endpoint_bindings=endpoint_bindings,
1160             num_units=num_units,
1161             resources=resources,
1162             storage=storage,
1163             placement=placement
1164         )
1165
1166         result = await app_facade.Deploy([app])
1167         errors = [r.error.message for r in result.results if r.error]
1168         if errors:
1169             raise JujuError('\n'.join(errors))
1170         return await self._wait_for_new('application', application)
1171
1172     async def destroy(self):
1173         """Terminate all machines and resources for this model.
1174             Is already implemented in controller.py.
1175         """
1176         raise NotImplementedError()
1177
1178     async def destroy_unit(self, *unit_names):
1179         """Destroy units by name.
1180
1181         """
1182         app_facade = client.ApplicationFacade.from_connection(self.connection)
1183
1184         log.debug(
1185             'Destroying unit%s %s',
1186             's' if len(unit_names) == 1 else '',
1187             ' '.join(unit_names))
1188
1189         return await app_facade.DestroyUnits(list(unit_names))
1190     destroy_units = destroy_unit
1191
1192     def get_backup(self, archive_id):
1193         """Download a backup archive file.
1194
1195         :param str archive_id: The id of the archive to download
1196         :return str: Path to the archive file
1197
1198         """
1199         raise NotImplementedError()
1200
1201     def enable_ha(
1202             self, num_controllers=0, constraints=None, series=None, to=None):
1203         """Ensure sufficient controllers exist to provide redundancy.
1204
1205         :param int num_controllers: Number of controllers to make available
1206         :param constraints: Constraints to apply to the controller machines
1207         :type constraints: :class:`juju.Constraints`
1208         :param str series: Series of the controller machines
1209         :param list to: Placement directives for controller machines, e.g.::
1210
1211             '23' - machine 23
1212             'lxc:7' - new lxc container on machine 7
1213             '24/lxc/3' - lxc container 3 or machine 24
1214
1215             If None, a new machine is provisioned.
1216
1217         """
1218         raise NotImplementedError()
1219
1220     def get_config(self):
1221         """Return the configuration settings for this model.
1222
1223         """
1224         raise NotImplementedError()
1225
1226     def get_constraints(self):
1227         """Return the machine constraints for this model.
1228
1229         """
1230         raise NotImplementedError()
1231
1232     async def grant(self, username, acl='read'):
1233         """Grant a user access to this model.
1234
1235         :param str username: Username
1236         :param str acl: Access control ('read' or 'write')
1237
1238         """
1239         controller_conn = await self.connection.controller()
1240         model_facade = client.ModelManagerFacade.from_connection(
1241             controller_conn)
1242         user = tag.user(username)
1243         model = tag.model(self.info.uuid)
1244         changes = client.ModifyModelAccess(acl, 'grant', model, user)
1245         await self.revoke(username)
1246         return await model_facade.ModifyModelAccess([changes])
1247
1248     def import_ssh_key(self, identity):
1249         """Add a public SSH key from a trusted indentity source to this model.
1250
1251         :param str identity: User identity in the form <lp|gh>:<username>
1252
1253         """
1254         raise NotImplementedError()
1255     import_ssh_keys = import_ssh_key
1256
1257     async def get_machines(self):
1258         """Return list of machines in this model.
1259
1260         """
1261         return list(self.state.machines.keys())
1262
1263     def get_shares(self):
1264         """Return list of all users with access to this model.
1265
1266         """
1267         raise NotImplementedError()
1268
1269     def get_spaces(self):
1270         """Return list of all known spaces, including associated subnets.
1271
1272         """
1273         raise NotImplementedError()
1274
1275     async def get_ssh_key(self, raw_ssh=False):
1276         """Return known SSH keys for this model.
1277         :param bool raw_ssh: if True, returns the raw ssh key,
1278             else it's fingerprint
1279
1280         """
1281         key_facade = client.KeyManagerFacade.from_connection(self.connection)
1282         entity = {'tag': tag.model(self.info.uuid)}
1283         entities = client.Entities([entity])
1284         return await key_facade.ListKeys(entities, raw_ssh)
1285     get_ssh_keys = get_ssh_key
1286
1287     def get_storage(self, filesystem=False, volume=False):
1288         """Return details of storage instances.
1289
1290         :param bool filesystem: Include filesystem storage
1291         :param bool volume: Include volume storage
1292
1293         """
1294         raise NotImplementedError()
1295
1296     def get_storage_pools(self, names=None, providers=None):
1297         """Return list of storage pools.
1298
1299         :param list names: Only include pools with these names
1300         :param list providers: Only include pools for these providers
1301
1302         """
1303         raise NotImplementedError()
1304
1305     def get_subnets(self, space=None, zone=None):
1306         """Return list of known subnets.
1307
1308         :param str space: Only include subnets in this space
1309         :param str zone: Only include subnets in this zone
1310
1311         """
1312         raise NotImplementedError()
1313
1314     def remove_blocks(self):
1315         """Remove all blocks from this model.
1316
1317         """
1318         raise NotImplementedError()
1319
1320     def remove_backup(self, backup_id):
1321         """Delete a backup.
1322
1323         :param str backup_id: The id of the backup to remove
1324
1325         """
1326         raise NotImplementedError()
1327
1328     def remove_cached_images(self, arch=None, kind=None, series=None):
1329         """Remove cached OS images.
1330
1331         :param str arch: Architecture of the images to remove
1332         :param str kind: Image kind to remove, e.g. 'lxd'
1333         :param str series: Image series to remove, e.g. 'xenial'
1334
1335         """
1336         raise NotImplementedError()
1337
1338     def remove_machine(self, *machine_ids):
1339         """Remove a machine from this model.
1340
1341         :param str \*machine_ids: Ids of the machines to remove
1342
1343         """
1344         raise NotImplementedError()
1345     remove_machines = remove_machine
1346
1347     async def remove_ssh_key(self, user, key):
1348         """Remove a public SSH key(s) from this model.
1349
1350         :param str key: Full ssh key
1351         :param str user: Juju user to which the key is registered
1352
1353         """
1354         key_facade = client.KeyManagerFacade.from_connection(self.connection)
1355         key = base64.b64decode(bytes(key.strip().split()[1].encode('ascii')))
1356         key = hashlib.md5(key).hexdigest()
1357         key = ':'.join(a+b for a, b in zip(key[::2], key[1::2]))
1358         await key_facade.DeleteKeys([key], user)
1359     remove_ssh_keys = remove_ssh_key
1360
1361     def restore_backup(
1362             self, bootstrap=False, constraints=None, archive=None,
1363             backup_id=None, upload_tools=False):
1364         """Restore a backup archive to a new controller.
1365
1366         :param bool bootstrap: Bootstrap a new state machine
1367         :param constraints: Model constraints
1368         :type constraints: :class:`juju.Constraints`
1369         :param str archive: Path to backup archive to restore
1370         :param str backup_id: Id of backup to restore
1371         :param bool upload_tools: Upload tools if bootstrapping a new machine
1372
1373         """
1374         raise NotImplementedError()
1375
1376     def retry_provisioning(self):
1377         """Retry provisioning for failed machines.
1378
1379         """
1380         raise NotImplementedError()
1381
1382     async def revoke(self, username):
1383         """Revoke a user's access to this model.
1384
1385         :param str username: Username to revoke
1386
1387         """
1388         controller_conn = await self.connection.controller()
1389         model_facade = client.ModelManagerFacade.from_connection(
1390             controller_conn)
1391         user = tag.user(username)
1392         model = tag.model(self.info.uuid)
1393         changes = client.ModifyModelAccess('read', 'revoke', model, user)
1394         return await model_facade.ModifyModelAccess([changes])
1395
1396     def run(self, command, timeout=None):
1397         """Run command on all machines in this model.
1398
1399         :param str command: The command to run
1400         :param int timeout: Time to wait before command is considered failed
1401
1402         """
1403         raise NotImplementedError()
1404
1405     def set_config(self, **config):
1406         """Set configuration keys on this model.
1407
1408         :param \*\*config: Config key/values
1409
1410         """
1411         raise NotImplementedError()
1412
1413     def set_constraints(self, constraints):
1414         """Set machine constraints on this model.
1415
1416         :param :class:`juju.Constraints` constraints: Machine constraints
1417
1418         """
1419         raise NotImplementedError()
1420
1421     def get_action_output(self, action_uuid, wait=-1):
1422         """Get the results of an action by ID.
1423
1424         :param str action_uuid: Id of the action
1425         :param int wait: Time in seconds to wait for action to complete
1426
1427         """
1428         raise NotImplementedError()
1429
1430     def get_action_status(self, uuid_or_prefix=None, name=None):
1431         """Get the status of all actions, filtered by ID, ID prefix, or action name.
1432
1433         :param str uuid_or_prefix: Filter by action uuid or prefix
1434         :param str name: Filter by action name
1435
1436         """
1437         raise NotImplementedError()
1438
1439     def get_budget(self, budget_name):
1440         """Get budget usage info.
1441
1442         :param str budget_name: Name of budget
1443
1444         """
1445         raise NotImplementedError()
1446
1447     async def get_status(self, filters=None, utc=False):
1448         """Return the status of the model.
1449
1450         :param str filters: Optional list of applications, units, or machines
1451             to include, which can use wildcards ('*').
1452         :param bool utc: Display time as UTC in RFC3339 format
1453
1454         """
1455         client_facade = client.ClientFacade.from_connection(self.connection)
1456         return await client_facade.FullStatus(filters)
1457
1458     def sync_tools(
1459             self, all_=False, destination=None, dry_run=False, public=False,
1460             source=None, stream=None, version=None):
1461         """Copy Juju tools into this model.
1462
1463         :param bool all_: Copy all versions, not just the latest
1464         :param str destination: Path to local destination directory
1465         :param bool dry_run: Don't do the actual copy
1466         :param bool public: Tools are for a public cloud, so generate mirrors
1467             information
1468         :param str source: Path to local source directory
1469         :param str stream: Simplestreams stream for which to sync metadata
1470         :param str version: Copy a specific major.minor version
1471
1472         """
1473         raise NotImplementedError()
1474
1475     def unblock(self, *commands):
1476         """Unblock an operation that would alter this model.
1477
1478         :param str \*commands: The commands to unblock. Valid values are
1479             'all-changes', 'destroy-model', 'remove-object'
1480
1481         """
1482         raise NotImplementedError()
1483
1484     def unset_config(self, *keys):
1485         """Unset configuration on this model.
1486
1487         :param str \*keys: The keys to unset
1488
1489         """
1490         raise NotImplementedError()
1491
1492     def upgrade_gui(self):
1493         """Upgrade the Juju GUI for this model.
1494
1495         """
1496         raise NotImplementedError()
1497
1498     def upgrade_juju(
1499             self, dry_run=False, reset_previous_upgrade=False,
1500             upload_tools=False, version=None):
1501         """Upgrade Juju on all machines in a model.
1502
1503         :param bool dry_run: Don't do the actual upgrade
1504         :param bool reset_previous_upgrade: Clear the previous (incomplete)
1505             upgrade status
1506         :param bool upload_tools: Upload local version of tools
1507         :param str version: Upgrade to a specific version
1508
1509         """
1510         raise NotImplementedError()
1511
1512     def upload_backup(self, archive_path):
1513         """Store a backup archive remotely in Juju.
1514
1515         :param str archive_path: Path to local archive
1516
1517         """
1518         raise NotImplementedError()
1519
1520     @property
1521     def charmstore(self):
1522         return self._charmstore
1523
1524     async def get_metrics(self, *tags):
1525         """Retrieve metrics.
1526
1527         :param str \*tags: Tags of entities from which to retrieve metrics.
1528             No tags retrieves the metrics of all units in the model.
1529         :return: Dictionary of unit_name:metrics
1530
1531         """
1532         log.debug("Retrieving metrics for %s",
1533                   ', '.join(tags) if tags else "all units")
1534
1535         metrics_facade = client.MetricsDebugFacade.from_connection(
1536             self.connection)
1537
1538         entities = [client.Entity(tag) for tag in tags]
1539         metrics_result = await metrics_facade.GetMetrics(entities)
1540
1541         metrics = collections.defaultdict(list)
1542
1543         for entity_metrics in metrics_result.results:
1544             error = entity_metrics.error
1545             if error:
1546                 if "is not a valid tag" in error:
1547                     raise ValueError(error.message)
1548                 else:
1549                     raise Exception(error.message)
1550
1551             for metric in entity_metrics.metrics:
1552                 metrics[metric.unit].append(vars(metric))
1553
1554         return metrics
1555
1556
1557 def get_charm_series(path):
1558     """Inspects the charm directory at ``path`` and returns a default
1559     series from its metadata.yaml (the first item in the 'series' list).
1560
1561     Returns None if no series can be determined.
1562
1563     """
1564     md = Path(path) / "metadata.yaml"
1565     if not md.exists():
1566         return None
1567     data = yaml.load(md.open())
1568     series = data.get('series')
1569     return series[0] if series else None
1570
1571
1572 class BundleHandler(object):
1573     """
1574     Handle bundles by using the API to translate bundle YAML into a plan of
1575     steps and then dispatching each of those using the API.
1576     """
1577     def __init__(self, model):
1578         self.model = model
1579         self.charmstore = model.charmstore
1580         self.plan = []
1581         self.references = {}
1582         self._units_by_app = {}
1583         for unit_name, unit in model.units.items():
1584             app_units = self._units_by_app.setdefault(unit.application, [])
1585             app_units.append(unit_name)
1586         self.client_facade = client.ClientFacade.from_connection(
1587             model.connection)
1588         self.app_facade = client.ApplicationFacade.from_connection(
1589             model.connection)
1590         self.ann_facade = client.AnnotationsFacade.from_connection(
1591             model.connection)
1592
1593     async def _handle_local_charms(self, bundle):
1594         """Search for references to local charms (i.e. filesystem paths)
1595         in the bundle. Upload the local charms to the model, and replace
1596         the filesystem paths with appropriate 'local:' paths in the bundle.
1597
1598         Return the modified bundle.
1599
1600         :param dict bundle: Bundle dictionary
1601         :return: Modified bundle dictionary
1602
1603         """
1604         apps, args = [], []
1605
1606         default_series = bundle.get('series')
1607         for app_name in self.applications:
1608             app_dict = bundle['services'][app_name]
1609             charm_dir = os.path.abspath(os.path.expanduser(app_dict['charm']))
1610             if not os.path.isdir(charm_dir):
1611                 continue
1612             series = (
1613                 app_dict.get('series') or
1614                 default_series or
1615                 get_charm_series(charm_dir)
1616             )
1617             if not series:
1618                 raise JujuError(
1619                     "Couldn't determine series for charm at {}. "
1620                     "Add a 'series' key to the bundle.".format(charm_dir))
1621
1622             # Keep track of what we need to update. We keep a list of apps
1623             # that need to be updated, and a corresponding list of args
1624             # needed to update those apps.
1625             apps.append(app_name)
1626             args.append((charm_dir, series))
1627
1628         if apps:
1629             # If we have apps to update, spawn all the coroutines concurrently
1630             # and wait for them to finish.
1631             charm_urls = await asyncio.gather(*[
1632                 self.model.add_local_charm_dir(*params)
1633                 for params in args
1634             ], loop=self.model.loop)
1635             # Update the 'charm:' entry for each app with the new 'local:' url.
1636             for app_name, charm_url in zip(apps, charm_urls):
1637                 bundle['services'][app_name]['charm'] = charm_url
1638
1639         return bundle
1640
1641     async def fetch_plan(self, entity_id):
1642         is_local = not entity_id.startswith('cs:') and os.path.isdir(entity_id)
1643         if is_local:
1644             bundle_yaml = (Path(entity_id) / "bundle.yaml").read_text()
1645         else:
1646             bundle_yaml = await self.charmstore.files(entity_id,
1647                                                       filename='bundle.yaml',
1648                                                       read_file=True)
1649         self.bundle = yaml.safe_load(bundle_yaml)
1650         self.bundle = await self._handle_local_charms(self.bundle)
1651
1652         self.plan = await self.client_facade.GetBundleChanges(
1653             yaml.dump(self.bundle))
1654
1655     async def execute_plan(self):
1656         for step in self.plan.changes:
1657             method = getattr(self, step.method)
1658             result = await method(*step.args)
1659             self.references[step.id_] = result
1660
1661     @property
1662     def applications(self):
1663         return list(self.bundle['services'].keys())
1664
1665     def resolve(self, reference):
1666         if reference and reference.startswith('$'):
1667             reference = self.references[reference[1:]]
1668         return reference
1669
1670     async def addCharm(self, charm, series):
1671         """
1672         :param charm string:
1673             Charm holds the URL of the charm to be added.
1674
1675         :param series string:
1676             Series holds the series of the charm to be added
1677             if the charm default is not sufficient.
1678         """
1679         # We don't add local charms because they've already been added
1680         # by self._handle_local_charms
1681         if charm.startswith('local:'):
1682             return charm
1683
1684         entity_id = await self.charmstore.entityId(charm)
1685         log.debug('Adding %s', entity_id)
1686         await self.client_facade.AddCharm(None, entity_id)
1687         return entity_id
1688
1689     async def addMachines(self, params=None):
1690         """
1691         :param params dict:
1692             Dictionary specifying the machine to add. All keys are optional.
1693             Keys include:
1694
1695             series: string specifying the machine OS series.
1696
1697             constraints: string holding machine constraints, if any. We'll
1698                 parse this into the json friendly dict that the juju api
1699                 expects.
1700
1701             container_type: string holding the type of the container (for
1702                 instance ""lxd" or kvm"). It is not specified for top level
1703                 machines.
1704
1705             parent_id: string holding a placeholder pointing to another
1706                 machine change or to a unit change. This value is only
1707                 specified in the case this machine is a container, in
1708                 which case also ContainerType is set.
1709
1710         """
1711         params = params or {}
1712
1713         # Normalize keys
1714         params = {normalize_key(k): params[k] for k in params.keys()}
1715
1716         # Fix up values, as necessary.
1717         if 'parent_id' in params:
1718             params['parent_id'] = self.resolve(params['parent_id'])
1719
1720         params['constraints'] = parse_constraints(
1721             params.get('constraints'))
1722         params['jobs'] = params.get('jobs', ['JobHostUnits'])
1723
1724         if params.get('container_type') == 'lxc':
1725             log.warning('Juju 2.0 does not support lxc containers. '
1726                         'Converting containers to lxd.')
1727             params['container_type'] = 'lxd'
1728
1729         # Submit the request.
1730         params = client.AddMachineParams(**params)
1731         results = await self.client_facade.AddMachines([params])
1732         error = results.machines[0].error
1733         if error:
1734             raise ValueError("Error adding machine: %s" % error.message)
1735         machine = results.machines[0].machine
1736         log.debug('Added new machine %s', machine)
1737         return machine
1738
1739     async def addRelation(self, endpoint1, endpoint2):
1740         """
1741         :param endpoint1 string:
1742         :param endpoint2 string:
1743             Endpoint1 and Endpoint2 hold relation endpoints in the
1744             "application:interface" form, where the application is always a
1745             placeholder pointing to an application change, and the interface is
1746             optional. Examples are "$deploy-42:web" or just "$deploy-42".
1747         """
1748         endpoints = [endpoint1, endpoint2]
1749         # resolve indirect references
1750         for i in range(len(endpoints)):
1751             parts = endpoints[i].split(':')
1752             parts[0] = self.resolve(parts[0])
1753             endpoints[i] = ':'.join(parts)
1754
1755         log.info('Relating %s <-> %s', *endpoints)
1756         return await self.model.add_relation(*endpoints)
1757
1758     async def deploy(self, charm, series, application, options, constraints,
1759                      storage, endpoint_bindings, resources):
1760         """
1761         :param charm string:
1762             Charm holds the URL of the charm to be used to deploy this
1763             application.
1764
1765         :param series string:
1766             Series holds the series of the application to be deployed
1767             if the charm default is not sufficient.
1768
1769         :param application string:
1770             Application holds the application name.
1771
1772         :param options map[string]interface{}:
1773             Options holds application options.
1774
1775         :param constraints string:
1776             Constraints holds the optional application constraints.
1777
1778         :param storage map[string]string:
1779             Storage holds the optional storage constraints.
1780
1781         :param endpoint_bindings map[string]string:
1782             EndpointBindings holds the optional endpoint bindings
1783
1784         :param resources map[string]int:
1785             Resources identifies the revision to use for each resource
1786             of the application's charm.
1787         """
1788         # resolve indirect references
1789         charm = self.resolve(charm)
1790         # the bundle plan doesn't actually do anything with resources, even
1791         # though it ostensibly gives us something (None) for that param
1792         if not charm.startswith('local:'):
1793             resources = await self.model._add_store_resources(application,
1794                                                               charm)
1795         await self.model._deploy(
1796             charm_url=charm,
1797             application=application,
1798             series=series,
1799             config=options,
1800             constraints=constraints,
1801             endpoint_bindings=endpoint_bindings,
1802             resources=resources,
1803             storage=storage,
1804         )
1805         return application
1806
1807     async def addUnit(self, application, to):
1808         """
1809         :param application string:
1810             Application holds the application placeholder name for which a unit
1811             is added.
1812
1813         :param to string:
1814             To holds the optional location where to add the unit, as a
1815             placeholder pointing to another unit change or to a machine change.
1816         """
1817         application = self.resolve(application)
1818         placement = self.resolve(to)
1819         if self._units_by_app.get(application):
1820             # enough units for this application already exist;
1821             # claim one, and carry on
1822             # NB: this should probably honor placement, but the juju client
1823             # doesn't, so we're not bothering, either
1824             unit_name = self._units_by_app[application].pop()
1825             log.debug('Reusing unit %s for %s', unit_name, application)
1826             return self.model.units[unit_name]
1827
1828         log.debug('Adding new unit for %s%s', application,
1829                   ' to %s' % placement if placement else '')
1830         return await self.model.applications[application].add_unit(
1831             count=1,
1832             to=placement,
1833         )
1834
1835     async def expose(self, application):
1836         """
1837         :param application string:
1838             Application holds the placeholder name of the application that must
1839             be exposed.
1840         """
1841         application = self.resolve(application)
1842         log.info('Exposing %s', application)
1843         return await self.model.applications[application].expose()
1844
1845     async def setAnnotations(self, id_, entity_type, annotations):
1846         """
1847         :param id_ string:
1848             Id is the placeholder for the application or machine change
1849             corresponding to the entity to be annotated.
1850
1851         :param entity_type EntityType:
1852             EntityType holds the type of the entity, "application" or
1853             "machine".
1854
1855         :param annotations map[string]string:
1856             Annotations holds the annotations as key/value pairs.
1857         """
1858         entity_id = self.resolve(id_)
1859         try:
1860             entity = self.model.state.get_entity(entity_type, entity_id)
1861         except KeyError:
1862             entity = await self.model._wait_for_new(entity_type, entity_id)
1863         return await entity.set_annotations(annotations)
1864
1865
1866 class CharmStore(object):
1867     """
1868     Async wrapper around theblues.charmstore.CharmStore
1869     """
1870     def __init__(self, loop):
1871         self.loop = loop
1872         self._cs = theblues.charmstore.CharmStore(timeout=5)
1873
1874     def __getattr__(self, name):
1875         """
1876         Wrap method calls in coroutines that use run_in_executor to make them
1877         async.
1878         """
1879         attr = getattr(self._cs, name)
1880         if not callable(attr):
1881             wrapper = partial(getattr, self._cs, name)
1882             setattr(self, name, wrapper)
1883         else:
1884             async def coro(*args, **kwargs):
1885                 method = partial(attr, *args, **kwargs)
1886                 for attempt in range(1, 4):
1887                     try:
1888                         return await self.loop.run_in_executor(None, method)
1889                     except theblues.errors.ServerError:
1890                         if attempt == 3:
1891                             raise
1892                         await asyncio.sleep(1, loop=self.loop)
1893             setattr(self, name, coro)
1894             wrapper = coro
1895         return wrapper
1896
1897
1898 class CharmArchiveGenerator(object):
1899     """
1900     Create a Zip archive of a local charm directory for upload to a controller.
1901
1902     This is used automatically by
1903     `Model.add_local_charm_dir <#juju.model.Model.add_local_charm_dir>`_.
1904     """
1905     def __init__(self, path):
1906         self.path = os.path.abspath(os.path.expanduser(path))
1907
1908     def make_archive(self, path):
1909         """Create archive of directory and write to ``path``.
1910
1911         :param path: Path to archive
1912
1913         Ignored::
1914
1915             * build/\* - This is used for packing the charm itself and any
1916                           similar tasks.
1917             * \*/.\*    - Hidden files are all ignored for now.  This will most
1918                           likely be changed into a specific ignore list
1919                           (.bzr, etc)
1920
1921         """
1922         zf = zipfile.ZipFile(path, 'w', zipfile.ZIP_DEFLATED)
1923         for dirpath, dirnames, filenames in os.walk(self.path):
1924             relative_path = dirpath[len(self.path) + 1:]
1925             if relative_path and not self._ignore(relative_path):
1926                 zf.write(dirpath, relative_path)
1927             for name in filenames:
1928                 archive_name = os.path.join(relative_path, name)
1929                 if not self._ignore(archive_name):
1930                     real_path = os.path.join(dirpath, name)
1931                     self._check_type(real_path)
1932                     if os.path.islink(real_path):
1933                         self._check_link(real_path)
1934                         self._write_symlink(
1935                             zf, os.readlink(real_path), archive_name)
1936                     else:
1937                         zf.write(real_path, archive_name)
1938         zf.close()
1939         return path
1940
1941     def _check_type(self, path):
1942         """Check the path
1943         """
1944         s = os.stat(path)
1945         if stat.S_ISDIR(s.st_mode) or stat.S_ISREG(s.st_mode):
1946             return path
1947         raise ValueError("Invalid Charm at % %s" % (
1948             path, "Invalid file type for a charm"))
1949
1950     def _check_link(self, path):
1951         link_path = os.readlink(path)
1952         if link_path[0] == "/":
1953             raise ValueError(
1954                 "Invalid Charm at %s: %s" % (
1955                     path, "Absolute links are invalid"))
1956         path_dir = os.path.dirname(path)
1957         link_path = os.path.join(path_dir, link_path)
1958         if not link_path.startswith(os.path.abspath(self.path)):
1959             raise ValueError(
1960                 "Invalid charm at %s %s" % (
1961                     path, "Only internal symlinks are allowed"))
1962
1963     def _write_symlink(self, zf, link_target, link_path):
1964         """Package symlinks with appropriate zipfile metadata."""
1965         info = zipfile.ZipInfo()
1966         info.filename = link_path
1967         info.create_system = 3
1968         # Magic code for symlinks / py2/3 compat
1969         # 27166663808 = (stat.S_IFLNK | 0755) << 16
1970         info.external_attr = 2716663808
1971         zf.writestr(info, link_target)
1972
1973     def _ignore(self, path):
1974         if path == "build" or path.startswith("build/"):
1975             return True
1976         if path.startswith('.'):
1977             return True