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