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