juju/model.py

   1 import asyncio
   2 import collections
   3 import logging
   4 import re
   5 import weakref
   6 from concurrent.futures import CancelledError
   7 from functools import partial
   8
   9 import yaml
  10 from theblues import charmstore
  11
  12 from .client import client
  13 from .client import watcher
  14 from .client import connection
  15 from .delta import get_entity_delta
  16 from .delta import get_entity_class
  17 from .exceptions import DeadEntityException
  18 from .errors import JujuAPIError
  19
  20 log = logging.getLogger(__name__)
  21
  22
  23 class _Observer(object):
  24     """Wrapper around an observer callable.
  25
  26     This wrapper allows filter criteria to be associated with the
  27     callable so that it's only called for changes that meet the criteria.
  28
  29     """
  30     def __init__(self, callable_, entity_type, action, entity_id, predicate):
  31         self.callable_ = callable_
  32         self.entity_type = entity_type
  33         self.action = action
  34         self.entity_id = entity_id
  35         self.predicate = predicate
  36         if self.entity_id:
  37             self.entity_id = str(self.entity_id)
  38             if not self.entity_id.startswith('^'):
  39                 self.entity_id = '^' + self.entity_id
  40             if not self.entity_id.endswith('$'):
  41                 self.entity_id += '$'
  42
  43     async def __call__(self, delta, old, new, model):
  44         await self.callable_(delta, old, new, model)
  45
  46     def cares_about(self, delta):
  47         """Return True if this observer "cares about" (i.e. wants to be
  48         called) for a this delta.
  49
  50         """
  51         if (self.entity_id and delta.get_id() and
  52                 not re.match(self.entity_id, str(delta.get_id()))):
  53             return False
  54
  55         if self.entity_type and self.entity_type != delta.entity:
  56             return False
  57
  58         if self.action and self.action != delta.type:
  59             return False
  60
  61         if self.predicate and not self.predicate(delta):
  62             return False
  63
  64         return True
  65
  66
  67 class ModelObserver(object):
  68     async def __call__(self, delta, old, new, model):
  69         handler_name = 'on_{}_{}'.format(delta.entity, delta.type)
  70         method = getattr(self, handler_name, self.on_change)
  71         await method(delta, old, new, model)
  72
  73     async def on_change(self, delta, old, new, model):
  74         pass
  75
  76
  77 class ModelState(object):
  78     """Holds the state of the model, including the delta history of all
  79     entities in the model.
  80
  81     """
  82     def __init__(self, model):
  83         self.model = model
  84         self.state = dict()
  85
  86     def _live_entity_map(self, entity_type):
  87         """Return an id:Entity map of all the living entities of
  88         type ``entity_type``.
  89
  90         """
  91         return {
  92             entity_id: self.get_entity(entity_type, entity_id)
  93             for entity_id, history in self.state.get(entity_type, {}).items()
  94             if history[-1] is not None
  95         }
  96
  97     @property
  98     def applications(self):
  99         """Return a map of application-name:Application for all applications
 100         currently in the model.
 101
 102         """
 103         return self._live_entity_map('application')
 104
 105     @property
 106     def machines(self):
 107         """Return a map of machine-id:Machine for all machines currently in
 108         the model.
 109
 110         """
 111         return self._live_entity_map('machine')
 112
 113     @property
 114     def units(self):
 115         """Return a map of unit-id:Unit for all units currently in
 116         the model.
 117
 118         """
 119         return self._live_entity_map('unit')
 120
 121     def entity_history(self, entity_type, entity_id):
 122         """Return the history deque for an entity.
 123
 124         """
 125         return self.state[entity_type][entity_id]
 126
 127     def entity_data(self, entity_type, entity_id, history_index):
 128         """Return the data dict for an entity at a specific index of its
 129         history.
 130
 131         """
 132         return self.entity_history(entity_type, entity_id)[history_index]
 133
 134     def apply_delta(self, delta):
 135         """Apply delta to our state and return a copy of the
 136         affected object as it was before and after the update, e.g.:
 137
 138             old_obj, new_obj = self.apply_delta(delta)
 139
 140         old_obj may be None if the delta is for the creation of a new object,
 141         e.g. a new application or unit is deployed.
 142
 143         new_obj will never be None, but may be dead (new_obj.dead == True)
 144         if the object was deleted as a result of the delta being applied.
 145
 146         """
 147         history = (
 148             self.state
 149             .setdefault(delta.entity, {})
 150             .setdefault(delta.get_id(), collections.deque())
 151         )
 152
 153         history.append(delta.data)
 154         if delta.type == 'remove':
 155             history.append(None)
 156
 157         entity = self.get_entity(delta.entity, delta.get_id())
 158         return entity.previous(), entity
 159
 160     def get_entity(
 161             self, entity_type, entity_id, history_index=-1, connected=True):
 162         """Return an object instance for the given entity_type and id.
 163
 164         By default the object state matches the most recent state from
 165         Juju. To get an instance of the object in an older state, pass
 166         history_index, an index into the history deque for the entity.
 167
 168         """
 169
 170         if history_index < 0 and history_index != -1:
 171             history_index += len(self.entity_history(entity_type, entity_id))
 172             if history_index < 0:
 173                 return None
 174
 175         try:
 176             self.entity_data(entity_type, entity_id, history_index)
 177         except IndexError:
 178             return None
 179
 180         entity_class = get_entity_class(entity_type)
 181         return entity_class(
 182             entity_id, self.model, history_index=history_index,
 183             connected=connected)
 184
 185
 186 class ModelEntity(object):
 187     """An object in the Model tree"""
 188
 189     def __init__(self, entity_id, model, history_index=-1, connected=True):
 190         """Initialize a new entity
 191
 192         :param entity_id str: The unique id of the object in the model
 193         :param model: The model instance in whose object tree this
 194             entity resides
 195         :history_index int: The index of this object's state in the model's
 196             history deque for this entity
 197         :connected bool: Flag indicating whether this object gets live updates
 198             from the model.
 199
 200         """
 201         self.entity_id = entity_id
 202         self.model = model
 203         self._history_index = history_index
 204         self.connected = connected
 205         self.connection = model.connection
 206
 207     def __getattr__(self, name):
 208         """Fetch object attributes from the underlying data dict held in the
 209         model.
 210
 211         """
 212         if self.data is None:
 213             raise DeadEntityException(
 214                 "Entity {}:{} is dead - its attributes can no longer be "
 215                 "accessed. Use the .previous() method on this object to get "
 216                 "a copy of the object at its previous state.".format(
 217                     self.entity_type, self.entity_id))
 218         return self.data[name]
 219
 220     def __bool__(self):
 221         return bool(self.data)
 222
 223     def on_change(self, callable_):
 224         """Add a change observer to this entity.
 225
 226         """
 227         self.model.add_observer(
 228             callable_, self.entity_type, 'change', self.entity_id)
 229
 230     def on_remove(self, callable_):
 231         """Add a remove observer to this entity.
 232
 233         """
 234         self.model.add_observer(
 235             callable_, self.entity_type, 'remove', self.entity_id)
 236
 237     @property
 238     def entity_type(self):
 239         """A string identifying the entity type of this object, e.g.
 240         'application' or 'unit', etc.
 241
 242         """
 243         return self.__class__.__name__.lower()
 244
 245     @property
 246     def current(self):
 247         """Return True if this object represents the current state of the
 248         entity in the underlying model.
 249
 250         This will be True except when the object represents an entity at a
 251         non-latest state in history, e.g. if the object was obtained by calling
 252         .previous() on another object.
 253
 254         """
 255         return self._history_index == -1
 256
 257     @property
 258     def dead(self):
 259         """Returns True if this entity no longer exists in the underlying
 260         model.
 261
 262         """
 263         return (
 264             self.data is None or
 265             self.model.state.entity_data(
 266                 self.entity_type, self.entity_id, -1) is None
 267         )
 268
 269     @property
 270     def alive(self):
 271         """Returns True if this entity still exists in the underlying
 272         model.
 273
 274         """
 275         return not self.dead
 276
 277     @property
 278     def data(self):
 279         """The data dictionary for this entity.
 280
 281         """
 282         return self.model.state.entity_data(
 283             self.entity_type, self.entity_id, self._history_index)
 284
 285     def previous(self):
 286         """Return a copy of this object as was at its previous state in
 287         history.
 288
 289         Returns None if this object is new (and therefore has no history).
 290
 291         The returned object is always "disconnected", i.e. does not receive
 292         live updates.
 293
 294         """
 295         return self.model.state.get_entity(
 296             self.entity_type, self.entity_id, self._history_index - 1,
 297             connected=False)
 298
 299     def next(self):
 300         """Return a copy of this object at its next state in
 301         history.
 302
 303         Returns None if this object is already the latest.
 304
 305         The returned object is "disconnected", i.e. does not receive
 306         live updates, unless it is current (latest).
 307
 308         """
 309         if self._history_index == -1:
 310             return None
 311
 312         new_index = self._history_index + 1
 313         connected = (
 314             new_index == len(self.model.state.entity_history(
 315                 self.entity_type, self.entity_id)) - 1
 316         )
 317         return self.model.state.get_entity(
 318             self.entity_type, self.entity_id, self._history_index - 1,
 319             connected=connected)
 320
 321     def latest(self):
 322         """Return a copy of this object at its current state in the model.
 323
 324         Returns self if this object is already the latest.
 325
 326         The returned object is always "connected", i.e. receives
 327         live updates from the model.
 328
 329         """
 330         if self._history_index == -1:
 331             return self
 332
 333         return self.model.state.get_entity(self.entity_type, self.entity_id)
 334
 335
 336 class Model(object):
 337     def __init__(self, loop=None):
 338         """Instantiate a new connected Model.
 339
 340         :param loop: an asyncio event loop
 341
 342         """
 343         self.loop = loop or asyncio.get_event_loop()
 344         self.connection = None
 345         self.observers = weakref.WeakValueDictionary()
 346         self.state = ModelState(self)
 347         self._watcher_task = None
 348         self._watch_shutdown = asyncio.Event(loop=loop)
 349         self._watch_received = asyncio.Event(loop=loop)
 350         self._charmstore = CharmStore(self.loop)
 351
 352     async def connect_current(self):
 353         """Connect to the current Juju model.
 354
 355         """
 356         self.connection = await connection.Connection.connect_current()
 357         self._watch()
 358         await self._watch_received.wait()
 359
 360     async def disconnect(self):
 361         """Shut down the watcher task and close websockets.
 362
 363         """
 364         self._stop_watching()
 365         if self.connection and self.connection.is_open:
 366             await self._watch_shutdown.wait()
 367             log.debug('Closing model connection')
 368             await self.connection.close()
 369             self.connection = None
 370
 371     def all_units_idle(self):
 372         """Return True if all units are idle.
 373
 374         """
 375         for unit in self.units.values():
 376             unit_status = unit.data['agent-status']['current']
 377             if unit_status != 'idle':
 378                 return False
 379         return True
 380
 381     async def reset(self, force=False):
 382         """Reset the model to a clean state.
 383
 384         :param bool force: Force-terminate machines.
 385
 386         This returns only after the model has reached a clean state. "Clean"
 387         means no applications or machines exist in the model.
 388
 389         """
 390         log.debug('Resetting model')
 391         for app in self.applications.values():
 392             await app.destroy()
 393         for machine in self.machines.values():
 394             await machine.destroy(force=force)
 395         await self.block_until(
 396             lambda: len(self.machines) == 0
 397         )
 398
 399     async def block_until(self, *conditions, timeout=None):
 400         """Return only after all conditions are true.
 401
 402         """
 403         async def _block():
 404             while not all(c() for c in conditions):
 405                 await asyncio.sleep(0)
 406         await asyncio.wait_for(_block(), timeout)
 407
 408     @property
 409     def applications(self):
 410         """Return a map of application-name:Application for all applications
 411         currently in the model.
 412
 413         """
 414         return self.state.applications
 415
 416     @property
 417     def machines(self):
 418         """Return a map of machine-id:Machine for all machines currently in
 419         the model.
 420
 421         """
 422         return self.state.machines
 423
 424     @property
 425     def units(self):
 426         """Return a map of unit-id:Unit for all units currently in
 427         the model.
 428
 429         """
 430         return self.state.units
 431
 432     def add_observer(
 433             self, callable_, entity_type=None, action=None, entity_id=None,
 434             predicate=None):
 435         """Register an "on-model-change" callback
 436
 437         Once the model is connected, ``callable_``
 438         will be called each time the model changes. callable_ should
 439         be Awaitable and accept the following positional arguments:
 440
 441             delta - An instance of :class:`juju.delta.EntityDelta`
 442                 containing the raw delta data recv'd from the Juju
 443                 websocket.
 444
 445             old_obj - If the delta modifies an existing object in the model,
 446                 old_obj will be a copy of that object, as it was before the
 447                 delta was applied. Will be None if the delta creates a new
 448                 entity in the model.
 449
 450             new_obj - A copy of the new or updated object, after the delta
 451                 is applied. Will be None if the delta removes an entity
 452                 from the model.
 453
 454             model - The :class:`Model` itself.
 455
 456         Events for which ``callable_`` is called can be specified by passing
 457         entity_type, action, and/or id_ filter criteria, e.g.:
 458
 459             add_observer(
 460                 myfunc, entity_type='application', action='add', id_='ubuntu')
 461
 462         For more complex filtering conditions, pass a predicate function. It
 463         will be called with a delta as its only argument. If the predicate
 464         function returns True, the callable_ will be called.
 465
 466         """
 467         observer = _Observer(
 468             callable_, entity_type, action, entity_id, predicate)
 469         self.observers[observer] = callable_
 470
 471     def _watch(self):
 472         """Start an asynchronous watch against this model.
 473
 474         See :meth:`add_observer` to register an onchange callback.
 475
 476         """
 477         async def _start_watch():
 478             self._watch_shutdown.clear()
 479             try:
 480                 allwatcher = watcher.AllWatcher()
 481                 self._watch_conn = await self.connection.clone()
 482                 allwatcher.connect(self._watch_conn)
 483                 while True:
 484                     results = await allwatcher.Next()
 485                     for delta in results.deltas:
 486                         delta = get_entity_delta(delta)
 487                         old_obj, new_obj = self.state.apply_delta(delta)
 488                         # XXX: Might not want to shield at this level
 489                         # We are shielding because when the watcher is
 490                         # canceled (on disconnect()), we don't want all of
 491                         # its children (every observer callback) to be
 492                         # canceled with it. So we shield them. But this means
 493                         # they can *never* be canceled.
 494                         await asyncio.shield(
 495                             self._notify_observers(delta, old_obj, new_obj))
 496                     self._watch_received.set()
 497             except CancelledError:
 498                 log.debug('Closing watcher connection')
 499                 await self._watch_conn.close()
 500                 self._watch_shutdown.set()
 501                 self._watch_conn = None
 502
 503         log.debug('Starting watcher task')
 504         self._watcher_task = self.loop.create_task(_start_watch())
 505
 506     def _stop_watching(self):
 507         """Stop the asynchronous watch against this model.
 508
 509         """
 510         log.debug('Stopping watcher task')
 511         if self._watcher_task:
 512             self._watcher_task.cancel()
 513
 514     async def _notify_observers(self, delta, old_obj, new_obj):
 515         """Call observing callbacks, notifying them of a change in model state
 516
 517         :param delta: The raw change from the watcher
 518             (:class:`juju.client.overrides.Delta`)
 519         :param old_obj: The object in the model that this delta updates.
 520             May be None.
 521         :param new_obj: The object in the model that is created or updated
 522             by applying this delta.
 523
 524         """
 525         if new_obj and not old_obj:
 526             delta.type = 'add'
 527
 528         log.debug(
 529             'Model changed: %s %s %s',
 530             delta.entity, delta.type, delta.get_id())
 531
 532         for o in self.observers:
 533             if o.cares_about(delta):
 534                 asyncio.ensure_future(o(delta, old_obj, new_obj, self))
 535
 536     async def _wait(self, entity_type, entity_id, action, predicate=None):
 537         """
 538         Block the calling routine until a given action has happened to the
 539         given entity
 540
 541         :param entity_type: The entity's type.
 542         :param entity_id: The entity's id.
 543         :param action: the type of action (e.g., 'add' or 'change')
 544         :param predicate: optional callable that must take as an
 545             argument a delta, and must return a boolean, indicating
 546             whether the delta contains the specific action we're looking
 547             for. For example, you might check to see whether a 'change'
 548             has a 'completed' status. See the _Observer class for details.
 549
 550         """
 551         q = asyncio.Queue(loop=self.loop)
 552
 553         async def callback(delta, old, new, model):
 554             await q.put(delta.get_id())
 555
 556         self.add_observer(callback, entity_type, action, entity_id, predicate)
 557         entity_id = await q.get()
 558         return self.state._live_entity_map(entity_type)[entity_id]
 559
 560     async def _wait_for_new(self, entity_type, entity_id, predicate=None):
 561         """Wait for a new object to appear in the Model and return it.
 562
 563         Waits for an object of type ``entity_type`` with id ``entity_id``.
 564
 565         This coroutine blocks until the new object appears in the model.
 566
 567         """
 568         return await self._wait(entity_type, entity_id, 'add', predicate)
 569
 570     async def wait_for_action(self, action_id):
 571         """Given an action, wait for it to complete."""
 572
 573         if action_id.startswith("action-"):
 574             # if we've been passed action.tag, transform it into the
 575             # id that the api deltas will use.
 576             action_id = action_id[7:]
 577
 578         def predicate(delta):
 579             return delta.data['status'] in ('completed', 'failed')
 580
 581         return await self._wait('action', action_id, 'change', predicate)
 582
 583     def add_machine(
 584             self, spec=None, constraints=None, disks=None, series=None,
 585             count=1):
 586         """Start a new, empty machine and optionally a container, or add a
 587         container to a machine.
 588
 589         :param str spec: Machine specification
 590             Examples::
 591
 592                 (None) - starts a new machine
 593                 'lxc' - starts a new machine with on lxc container
 594                 'lxc:4' - starts a new lxc container on machine 4
 595                 'ssh:user@10.10.0.3' - manually provisions a machine with ssh
 596                 'zone=us-east-1a' - starts a machine in zone us-east-1s on AWS
 597                 'maas2.name' - acquire machine maas2.name on MAAS
 598         :param constraints: Machine constraints
 599         :type constraints: :class:`juju.Constraints`
 600         :param list disks: List of disk :class:`constraints <juju.Constraints>`
 601         :param str series: Series
 602         :param int count: Number of machines to deploy
 603
 604         Supported container types are: lxc, lxd, kvm
 605
 606         When deploying a container to an existing machine, constraints cannot
 607         be used.
 608
 609         """
 610         pass
 611     add_machines = add_machine
 612
 613     async def add_relation(self, relation1, relation2):
 614         """Add a relation between two applications.
 615
 616         :param str relation1: '<application>[:<relation_name>]'
 617         :param str relation2: '<application>[:<relation_name>]'
 618
 619         """
 620         app_facade = client.ApplicationFacade()
 621         app_facade.connect(self.connection)
 622
 623         log.debug(
 624             'Adding relation %s <-> %s', relation1, relation2)
 625
 626         try:
 627             result = await app_facade.AddRelation([relation1, relation2])
 628         except JujuAPIError as e:
 629             if 'relation already exists' not in e.message:
 630                 raise
 631             log.debug(
 632                 'Relation %s <-> %s already exists', relation1, relation2)
 633             # TODO: if relation already exists we should return the
 634             # Relation ModelEntity here
 635             return None
 636
 637         def predicate(delta):
 638             endpoints = {}
 639             for endpoint in delta.data['endpoints']:
 640                 endpoints[endpoint['application-name']] = endpoint['relation']
 641             return endpoints == result.endpoints
 642
 643         return await self._wait_for_new('relation', None, predicate)
 644
 645     def add_space(self, name, *cidrs):
 646         """Add a new network space.
 647
 648         Adds a new space with the given name and associates the given
 649         (optional) list of existing subnet CIDRs with it.
 650
 651         :param str name: Name of the space
 652         :param \*cidrs: Optional list of existing subnet CIDRs
 653
 654         """
 655         pass
 656
 657     def add_ssh_key(self, key):
 658         """Add a public SSH key to this model.
 659
 660         :param str key: The public ssh key
 661
 662         """
 663         pass
 664     add_ssh_keys = add_ssh_key
 665
 666     def add_subnet(self, cidr_or_id, space, *zones):
 667         """Add an existing subnet to this model.
 668
 669         :param str cidr_or_id: CIDR or provider ID of the existing subnet
 670         :param str space: Network space with which to associate
 671         :param str \*zones: Zone(s) in which the subnet resides
 672
 673         """
 674         pass
 675
 676     def get_backups(self):
 677         """Retrieve metadata for backups in this model.
 678
 679         """
 680         pass
 681
 682     def block(self, *commands):
 683         """Add a new block to this model.
 684
 685         :param str \*commands: The commands to block. Valid values are
 686             'all-changes', 'destroy-model', 'remove-object'
 687
 688         """
 689         pass
 690
 691     def get_blocks(self):
 692         """List blocks for this model.
 693
 694         """
 695         pass
 696
 697     def get_cached_images(self, arch=None, kind=None, series=None):
 698         """Return a list of cached OS images.
 699
 700         :param str arch: Filter by image architecture
 701         :param str kind: Filter by image kind, e.g. 'lxd'
 702         :param str series: Filter by image series, e.g. 'xenial'
 703
 704         """
 705         pass
 706
 707     def create_backup(self, note=None, no_download=False):
 708         """Create a backup of this model.
 709
 710         :param str note: A note to store with the backup
 711         :param bool no_download: Do not download the backup archive
 712         :return str: Path to downloaded archive
 713
 714         """
 715         pass
 716
 717     def create_storage_pool(self, name, provider_type, **pool_config):
 718         """Create or define a storage pool.
 719
 720         :param str name: Name to give the storage pool
 721         :param str provider_type: Pool provider type
 722         :param \*\*pool_config: key/value pool configuration pairs
 723
 724         """
 725         pass
 726
 727     def debug_log(
 728             self, no_tail=False, exclude_module=None, include_module=None,
 729             include=None, level=None, limit=0, lines=10, replay=False,
 730             exclude=None):
 731         """Get log messages for this model.
 732
 733         :param bool no_tail: Stop after returning existing log messages
 734         :param list exclude_module: Do not show log messages for these logging
 735             modules
 736         :param list include_module: Only show log messages for these logging
 737             modules
 738         :param list include: Only show log messages for these entities
 739         :param str level: Log level to show, valid options are 'TRACE',
 740             'DEBUG', 'INFO', 'WARNING', 'ERROR,
 741         :param int limit: Return this many of the most recent (possibly
 742             filtered) lines are shown
 743         :param int lines: Yield this many of the most recent lines, and keep
 744             yielding
 745         :param bool replay: Yield the entire log, and keep yielding
 746         :param list exclude: Do not show log messages for these entities
 747
 748         """
 749         pass
 750
 751     async def deploy(
 752             self, entity_url, service_name=None, bind=None, budget=None,
 753             channel=None, config=None, constraints=None, force=False,
 754             num_units=1, plan=None, resources=None, series=None, storage=None,
 755             to=None):
 756         """Deploy a new service or bundle.
 757
 758         :param str entity_url: Charm or bundle url
 759         :param str service_name: Name to give the service
 760         :param dict bind: <charm endpoint>:<network space> pairs
 761         :param dict budget: <budget name>:<limit> pairs
 762         :param str channel: Charm store channel from which to retrieve
 763             the charm or bundle, e.g. 'development'
 764         :param dict config: Charm configuration dictionary
 765         :param constraints: Service constraints
 766         :type constraints: :class:`juju.Constraints`
 767         :param bool force: Allow charm to be deployed to a machine running
 768             an unsupported series
 769         :param int num_units: Number of units to deploy
 770         :param str plan: Plan under which to deploy charm
 771         :param dict resources: <resource name>:<file path> pairs
 772         :param str series: Series on which to deploy
 773         :param dict storage: Storage constraints TODO how do these look?
 774         :param str to: Placement directive, e.g.::
 775
 776             '23' - machine 23
 777             'lxc:7' - new lxc container on machine 7
 778             '24/lxc/3' - lxc container 3 or machine 24
 779
 780             If None, a new machine is provisioned.
 781
 782
 783         TODO::
 784
 785             - service_name is required; fill this in automatically if not
 786               provided by caller
 787             - series is required; how do we pick a default?
 788
 789         """
 790         if to:
 791             placement = [
 792                 client.Placement(**p) for p in to
 793             ]
 794         else:
 795             placement = []
 796
 797         if storage:
 798             storage = {
 799                 k: client.Constraints(**v)
 800                 for k, v in storage.items()
 801             }
 802
 803         entity_id = await self.charmstore.entityId(entity_url)
 804
 805         app_facade = client.ApplicationFacade()
 806         client_facade = client.ClientFacade()
 807         app_facade.connect(self.connection)
 808         client_facade.connect(self.connection)
 809
 810         if 'bundle/' in entity_id:
 811             handler = BundleHandler(self)
 812             await handler.fetch_plan(entity_id)
 813             await handler.execute_plan()
 814             extant_apps = {app for app in self.applications}
 815             pending_apps = set(handler.applications) - extant_apps
 816             if pending_apps:
 817                 # new apps will usually be in the model by now, but if some
 818                 # haven't made it yet we'll need to wait on them to be added
 819                 await asyncio.gather(*[
 820                     asyncio.ensure_future(
 821                         self.model._wait_for_new('application', app_name))
 822                     for app_name in pending_apps
 823                 ])
 824             return [app for name, app in self.applications.items()
 825                     if name in handler.applications]
 826         else:
 827             log.debug(
 828                 'Deploying %s', entity_id)
 829
 830             await client_facade.AddCharm(channel, entity_id)
 831             app = client.ApplicationDeploy(
 832                 application=service_name,
 833                 channel=channel,
 834                 charm_url=entity_id,
 835                 config=config,
 836                 constraints=constraints,
 837                 endpoint_bindings=bind,
 838                 num_units=num_units,
 839                 placement=placement,
 840                 resources=resources,
 841                 series=series,
 842                 storage=storage,
 843             )
 844
 845             await app_facade.Deploy([app])
 846             return await self._wait_for_new('application', service_name)
 847
 848     def destroy(self):
 849         """Terminate all machines and resources for this model.
 850
 851         """
 852         pass
 853
 854     async def destroy_unit(self, *unit_names):
 855         """Destroy units by name.
 856
 857         """
 858         app_facade = client.ApplicationFacade()
 859         app_facade.connect(self.connection)
 860
 861         log.debug(
 862             'Destroying unit%s %s',
 863             's' if len(unit_names) == 1 else '',
 864             ' '.join(unit_names))
 865
 866         return await app_facade.Destroy(self.name)
 867     destroy_units = destroy_unit
 868
 869     def get_backup(self, archive_id):
 870         """Download a backup archive file.
 871
 872         :param str archive_id: The id of the archive to download
 873         :return str: Path to the archive file
 874
 875         """
 876         pass
 877
 878     def enable_ha(
 879             self, num_controllers=0, constraints=None, series=None, to=None):
 880         """Ensure sufficient controllers exist to provide redundancy.
 881
 882         :param int num_controllers: Number of controllers to make available
 883         :param constraints: Constraints to apply to the controller machines
 884         :type constraints: :class:`juju.Constraints`
 885         :param str series: Series of the controller machines
 886         :param list to: Placement directives for controller machines, e.g.::
 887
 888             '23' - machine 23
 889             'lxc:7' - new lxc container on machine 7
 890             '24/lxc/3' - lxc container 3 or machine 24
 891
 892             If None, a new machine is provisioned.
 893
 894         """
 895         pass
 896
 897     def get_config(self):
 898         """Return the configuration settings for this model.
 899
 900         """
 901         pass
 902
 903     def get_constraints(self):
 904         """Return the machine constraints for this model.
 905
 906         """
 907         pass
 908
 909     def grant(self, username, acl='read'):
 910         """Grant a user access to this model.
 911
 912         :param str username: Username
 913         :param str acl: Access control ('read' or 'write')
 914
 915         """
 916         pass
 917
 918     def import_ssh_key(self, identity):
 919         """Add a public SSH key from a trusted indentity source to this model.
 920
 921         :param str identity: User identity in the form <lp|gh>:<username>
 922
 923         """
 924         pass
 925     import_ssh_keys = import_ssh_key
 926
 927     def get_machines(self, machine, utc=False):
 928         """Return list of machines in this model.
 929
 930         :param str machine: Machine id, e.g. '0'
 931         :param bool utc: Display time as UTC in RFC3339 format
 932
 933         """
 934         pass
 935
 936     def get_shares(self):
 937         """Return list of all users with access to this model.
 938
 939         """
 940         pass
 941
 942     def get_spaces(self):
 943         """Return list of all known spaces, including associated subnets.
 944
 945         """
 946         pass
 947
 948     def get_ssh_key(self):
 949         """Return known SSH keys for this model.
 950
 951         """
 952         pass
 953     get_ssh_keys = get_ssh_key
 954
 955     def get_storage(self, filesystem=False, volume=False):
 956         """Return details of storage instances.
 957
 958         :param bool filesystem: Include filesystem storage
 959         :param bool volume: Include volume storage
 960
 961         """
 962         pass
 963
 964     def get_storage_pools(self, names=None, providers=None):
 965         """Return list of storage pools.
 966
 967         :param list names: Only include pools with these names
 968         :param list providers: Only include pools for these providers
 969
 970         """
 971         pass
 972
 973     def get_subnets(self, space=None, zone=None):
 974         """Return list of known subnets.
 975
 976         :param str space: Only include subnets in this space
 977         :param str zone: Only include subnets in this zone
 978
 979         """
 980         pass
 981
 982     def remove_blocks(self):
 983         """Remove all blocks from this model.
 984
 985         """
 986         pass
 987
 988     def remove_backup(self, backup_id):
 989         """Delete a backup.
 990
 991         :param str backup_id: The id of the backup to remove
 992
 993         """
 994         pass
 995
 996     def remove_cached_images(self, arch=None, kind=None, series=None):
 997         """Remove cached OS images.
 998
 999         :param str arch: Architecture of the images to remove
1000         :param str kind: Image kind to remove, e.g. 'lxd'
1001         :param str series: Image series to remove, e.g. 'xenial'
1002
1003         """
1004         pass
1005
1006     def remove_machine(self, *machine_ids):
1007         """Remove a machine from this model.
1008
1009         :param str \*machine_ids: Ids of the machines to remove
1010
1011         """
1012         pass
1013     remove_machines = remove_machine
1014
1015     def remove_ssh_key(self, *keys):
1016         """Remove a public SSH key(s) from this model.
1017
1018         :param str \*keys: Keys to remove
1019
1020         """
1021         pass
1022     remove_ssh_keys = remove_ssh_key
1023
1024     def restore_backup(
1025             self, bootstrap=False, constraints=None, archive=None,
1026             backup_id=None, upload_tools=False):
1027         """Restore a backup archive to a new controller.
1028
1029         :param bool bootstrap: Bootstrap a new state machine
1030         :param constraints: Model constraints
1031         :type constraints: :class:`juju.Constraints`
1032         :param str archive: Path to backup archive to restore
1033         :param str backup_id: Id of backup to restore
1034         :param bool upload_tools: Upload tools if bootstrapping a new machine
1035
1036         """
1037         pass
1038
1039     def retry_provisioning(self):
1040         """Retry provisioning for failed machines.
1041
1042         """
1043         pass
1044
1045     def revoke(self, username, acl='read'):
1046         """Revoke a user's access to this model.
1047
1048         :param str username: Username to revoke
1049         :param str acl: Access control ('read' or 'write')
1050
1051         """
1052         pass
1053
1054     def run(self, command, timeout=None):
1055         """Run command on all machines in this model.
1056
1057         :param str command: The command to run
1058         :param int timeout: Time to wait before command is considered failed
1059
1060         """
1061         pass
1062
1063     def set_config(self, **config):
1064         """Set configuration keys on this model.
1065
1066         :param \*\*config: Config key/values
1067
1068         """
1069         pass
1070
1071     def set_constraints(self, constraints):
1072         """Set machine constraints on this model.
1073
1074         :param :class:`juju.Constraints` constraints: Machine constraints
1075
1076         """
1077         pass
1078
1079     def get_action_output(self, action_uuid, wait=-1):
1080         """Get the results of an action by ID.
1081
1082         :param str action_uuid: Id of the action
1083         :param int wait: Time in seconds to wait for action to complete
1084
1085         """
1086         pass
1087
1088     def get_action_status(self, uuid_or_prefix=None, name=None):
1089         """Get the status of all actions, filtered by ID, ID prefix, or action name.
1090
1091         :param str uuid_or_prefix: Filter by action uuid or prefix
1092         :param str name: Filter by action name
1093
1094         """
1095         pass
1096
1097     def get_budget(self, budget_name):
1098         """Get budget usage info.
1099
1100         :param str budget_name: Name of budget
1101
1102         """
1103         pass
1104
1105     def get_status(self, filter_=None, utc=False):
1106         """Return the status of the model.
1107
1108         :param str filter_: Service or unit name or wildcard ('*')
1109         :param bool utc: Display time as UTC in RFC3339 format
1110
1111         """
1112         pass
1113     status = get_status
1114
1115     def sync_tools(
1116             self, all_=False, destination=None, dry_run=False, public=False,
1117             source=None, stream=None, version=None):
1118         """Copy Juju tools into this model.
1119
1120         :param bool all_: Copy all versions, not just the latest
1121         :param str destination: Path to local destination directory
1122         :param bool dry_run: Don't do the actual copy
1123         :param bool public: Tools are for a public cloud, so generate mirrors
1124             information
1125         :param str source: Path to local source directory
1126         :param str stream: Simplestreams stream for which to sync metadata
1127         :param str version: Copy a specific major.minor version
1128
1129         """
1130         pass
1131
1132     def unblock(self, *commands):
1133         """Unblock an operation that would alter this model.
1134
1135         :param str \*commands: The commands to unblock. Valid values are
1136             'all-changes', 'destroy-model', 'remove-object'
1137
1138         """
1139         pass
1140
1141     def unset_config(self, *keys):
1142         """Unset configuration on this model.
1143
1144         :param str \*keys: The keys to unset
1145
1146         """
1147         pass
1148
1149     def upgrade_gui(self):
1150         """Upgrade the Juju GUI for this model.
1151
1152         """
1153         pass
1154
1155     def upgrade_juju(
1156             self, dry_run=False, reset_previous_upgrade=False,
1157             upload_tools=False, version=None):
1158         """Upgrade Juju on all machines in a model.
1159
1160         :param bool dry_run: Don't do the actual upgrade
1161         :param bool reset_previous_upgrade: Clear the previous (incomplete)
1162             upgrade status
1163         :param bool upload_tools: Upload local version of tools
1164         :param str version: Upgrade to a specific version
1165
1166         """
1167         pass
1168
1169     def upload_backup(self, archive_path):
1170         """Store a backup archive remotely in Juju.
1171
1172         :param str archive_path: Path to local archive
1173
1174         """
1175         pass
1176
1177     @property
1178     def charmstore(self):
1179         return self._charmstore
1180
1181
1182 class BundleHandler(object):
1183     """
1184     Handle bundles by using the API to translate bundle YAML into a plan of
1185     steps and then dispatching each of those using the API.
1186     """
1187     def __init__(self, model):
1188         self.model = model
1189         self.charmstore = model.charmstore
1190         self.plan = []
1191         self.references = {}
1192         self._units_by_app = {}
1193         for unit_name, unit in model.units.items():
1194             app_units = self._units_by_app.setdefault(unit.application, [])
1195             app_units.append(unit_name)
1196         self.client_facade = client.ClientFacade()
1197         self.client_facade.connect(model.connection)
1198         self.app_facade = client.ApplicationFacade()
1199         self.app_facade.connect(model.connection)
1200         self.ann_facade = client.AnnotationsFacade()
1201         self.ann_facade.connect(model.connection)
1202
1203     async def fetch_plan(self, entity_id):
1204         bundle_yaml = await self.charmstore.files(entity_id,
1205                                                   filename='bundle.yaml',
1206                                                   read_file=True)
1207         self.bundle = yaml.safe_load(bundle_yaml)
1208         self.plan = await self.client_facade.GetBundleChanges(bundle_yaml)
1209
1210     async def execute_plan(self):
1211         for step in self.plan.changes:
1212             method = getattr(self, step.method)
1213             result = await method(*step.args)
1214             self.references[step.id_] = result
1215
1216     @property
1217     def applications(self):
1218         return list(self.bundle['services'].keys())
1219
1220     def resolve(self, reference):
1221         if reference and reference.startswith('$'):
1222             reference = self.references[reference[1:]]
1223         return reference
1224
1225     async def addCharm(self, charm, series):
1226         """
1227         :param charm string:
1228             Charm holds the URL of the charm to be added.
1229
1230         :param series string:
1231             Series holds the series of the charm to be added
1232             if the charm default is not sufficient.
1233         """
1234         entity_id = await self.charmstore.entityId(charm)
1235         log.debug('Adding %s', entity_id)
1236         await self.client_facade.AddCharm(None, entity_id)
1237         return entity_id
1238
1239     async def addMachines(self, series, constraints, container_type,
1240                           parent_id):
1241         """
1242         :param series string:
1243             Series holds the optional machine OS series.
1244
1245         :param constraints string:
1246             Constraints holds the optional machine constraints.
1247
1248         :param Container_type string:
1249             ContainerType optionally holds the type of the container (for
1250             instance ""lxc" or kvm"). It is not specified for top level
1251             machines.
1252
1253         :param parent_id string:
1254             ParentId optionally holds a placeholder pointing to another machine
1255             change or to a unit change. This value is only specified in the
1256             case this machine is a container, in which case also ContainerType
1257             is set.
1258         """
1259         params = client.AddMachineParams(
1260             series=series,
1261             constraints=constraints,
1262             container_type=container_type,
1263             parent_id=self.resolve(parent_id),
1264         )
1265         results = await self.client_facade.AddMachines(params)
1266         log.debug('Added new machine %s', results[0].machine)
1267         return results[0].machine
1268
1269     async def addRelation(self, endpoint1, endpoint2):
1270         """
1271         :param endpoint1 string:
1272         :param endpoint2 string:
1273             Endpoint1 and Endpoint2 hold relation endpoints in the
1274             "application:interface" form, where the application is always a
1275             placeholder pointing to an application change, and the interface is
1276             optional. Examples are "$deploy-42:web" or just "$deploy-42".
1277         """
1278         endpoints = [endpoint1, endpoint2]
1279         # resolve indirect references
1280         for i in range(len(endpoints)):
1281             parts = endpoints[i].split(':')
1282             parts[0] = self.resolve(parts[0])
1283             endpoints[i] = ':'.join(parts)
1284
1285         log.info('Relating %s <-> %s', *endpoints)
1286         return await self.model.add_relation(*endpoints)
1287
1288     async def deploy(self, charm, series, application, options, constraints,
1289                      storage, endpoint_bindings, resources):
1290         """
1291         :param charm string:
1292             Charm holds the URL of the charm to be used to deploy this
1293             application.
1294
1295         :param series string:
1296             Series holds the series of the application to be deployed
1297             if the charm default is not sufficient.
1298
1299         :param application string:
1300             Application holds the application name.
1301
1302         :param options map[string]interface{}:
1303             Options holds application options.
1304
1305         :param constraints string:
1306             Constraints holds the optional application constraints.
1307
1308         :param storage map[string]string:
1309             Storage holds the optional storage constraints.
1310
1311         :param endpoint_bindings map[string]string:
1312             EndpointBindings holds the optional endpoint bindings
1313
1314         :param resources map[string]int:
1315             Resources identifies the revision to use for each resource
1316             of the application's charm.
1317         """
1318         # resolve indirect references
1319         charm = self.resolve(charm)
1320         # stringify all config values for API
1321         options = {k: str(v) for k, v in options.items()}
1322         # build param object
1323         app = client.ApplicationDeploy(
1324             charm_url=charm,
1325             series=series,
1326             application=application,
1327             config=options,
1328             constraints=constraints,
1329             storage=storage,
1330             endpoint_bindings=endpoint_bindings,
1331             resources=resources,
1332         )
1333         # do the do
1334         log.info('Deploying %s', charm)
1335         await self.app_facade.Deploy([app])
1336         return application
1337
1338     async def addUnit(self, application, to):
1339         """
1340         :param application string:
1341             Application holds the application placeholder name for which a unit
1342             is added.
1343
1344         :param to string:
1345             To holds the optional location where to add the unit, as a
1346             placeholder pointing to another unit change or to a machine change.
1347         """
1348         application = self.resolve(application)
1349         placement = self.resolve(to)
1350         if self._units_by_app.get(application):
1351             # enough units for this application already exist;
1352             # claim one, and carry on
1353             # NB: this should probably honor placement, but the juju client
1354             # doesn't, so we're not bothering, either
1355             unit_name = self._units_by_app[application].pop()
1356             log.debug('Reusing unit %s for %s', unit_name, application)
1357             return self.model.units[unit_name]
1358
1359         log.debug('Adding new unit for %s%s', application,
1360                   ' to %s' % placement if placement else '')
1361         return await self.model.applications[application].add_unit(
1362             count=1,
1363             to=placement,
1364         )
1365
1366     async def expose(self, application):
1367         """
1368         :param application string:
1369             Application holds the placeholder name of the application that must
1370             be exposed.
1371         """
1372         application = self.resolve(application)
1373         log.info('Exposing %s', application)
1374         return await self.model.applications[application].expose()
1375
1376     async def setAnnotations(self, id_, entity_type, annotations):
1377         """
1378         :param id_ string:
1379             Id is the placeholder for the application or machine change
1380             corresponding to the entity to be annotated.
1381
1382         :param entity_type EntityType:
1383             EntityType holds the type of the entity, "application" or
1384             "machine".
1385
1386         :param annotations map[string]string:
1387             Annotations holds the annotations as key/value pairs.
1388         """
1389         entity_id = self.resolve(id_)
1390         try:
1391             entity = self.model.state.get_entity(entity_type, entity_id)
1392         except KeyError:
1393             entity = await self.model._wait_for_new(entity_type, entity_id)
1394         return await entity.set_annotations(annotations)
1395
1396
1397 class CharmStore(object):
1398     """
1399     Async wrapper around theblues.charmstore.CharmStore
1400     """
1401     def __init__(self, loop):
1402         self.loop = loop
1403         self._cs = charmstore.CharmStore()
1404
1405     def __getattr__(self, name):
1406         """
1407         Wrap method calls in coroutines that use run_in_executor to make them
1408         async.
1409         """
1410         attr = getattr(self._cs, name)
1411         if not callable(attr):
1412             wrapper = partial(getattr, self._cs, name)
1413             setattr(self, name, wrapper)
1414         else:
1415             async def coro(*args, **kwargs):
1416                 method = partial(attr, *args, **kwargs)
1417                 return await self.loop.run_in_executor(None, method)
1418             setattr(self, name, coro)
1419             wrapper = coro
1420         return wrapper