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