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