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