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, 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
 584         This coroutine blocks until the new object appears in the model.
 585
 586         """
 587         return await self._wait(entity_type, entity_id, 'add', predicate)
 588
 589     async def wait_for_action(self, action_id):
 590         """Given an action, wait for it to complete."""
 591
 592         if action_id.startswith("action-"):
 593             # if we've been passed action.tag, transform it into the
 594             # id that the api deltas will use.
 595             action_id = action_id[7:]
 596
 597         def predicate(delta):
 598             return delta.data['status'] in ('completed', 'failed')
 599
 600         return await self._wait('action', action_id, 'change', predicate)
 601
 602     def add_machine(
 603             self, spec=None, constraints=None, disks=None, series=None,
 604             count=1):
 605         """Start a new, empty machine and optionally a container, or add a
 606         container to a machine.
 607
 608         :param str spec: Machine specification
 609             Examples::
 610
 611                 (None) - starts a new machine
 612                 'lxc' - starts a new machine with on lxc container
 613                 'lxc:4' - starts a new lxc container on machine 4
 614                 'ssh:user@10.10.0.3' - manually provisions a machine with ssh
 615                 'zone=us-east-1a' - starts a machine in zone us-east-1s on AWS
 616                 'maas2.name' - acquire machine maas2.name on MAAS
 617         :param constraints: Machine constraints
 618         :type constraints: :class:`juju.Constraints`
 619         :param list disks: List of disk :class:`constraints <juju.Constraints>`
 620         :param str series: Series
 621         :param int count: Number of machines to deploy
 622
 623         Supported container types are: lxc, lxd, kvm
 624
 625         When deploying a container to an existing machine, constraints cannot
 626         be used.
 627
 628         """
 629         pass
 630     add_machines = add_machine
 631
 632     async def add_relation(self, relation1, relation2):
 633         """Add a relation between two applications.
 634
 635         :param str relation1: '<application>[:<relation_name>]'
 636         :param str relation2: '<application>[:<relation_name>]'
 637
 638         """
 639         app_facade = client.ApplicationFacade()
 640         app_facade.connect(self.connection)
 641
 642         log.debug(
 643             'Adding relation %s <-> %s', relation1, relation2)
 644
 645         try:
 646             result = await app_facade.AddRelation([relation1, relation2])
 647         except JujuAPIError as e:
 648             if 'relation already exists' not in e.message:
 649                 raise
 650             log.debug(
 651                 'Relation %s <-> %s already exists', relation1, relation2)
 652             # TODO: if relation already exists we should return the
 653             # Relation ModelEntity here
 654             return None
 655
 656         def predicate(delta):
 657             endpoints = {}
 658             for endpoint in delta.data['endpoints']:
 659                 endpoints[endpoint['application-name']] = endpoint['relation']
 660             return endpoints == result.endpoints
 661
 662         return await self._wait_for_new('relation', None, predicate)
 663
 664     def add_space(self, name, *cidrs):
 665         """Add a new network space.
 666
 667         Adds a new space with the given name and associates the given
 668         (optional) list of existing subnet CIDRs with it.
 669
 670         :param str name: Name of the space
 671         :param \*cidrs: Optional list of existing subnet CIDRs
 672
 673         """
 674         pass
 675
 676     def add_ssh_key(self, key):
 677         """Add a public SSH key to this model.
 678
 679         :param str key: The public ssh key
 680
 681         """
 682         pass
 683     add_ssh_keys = add_ssh_key
 684
 685     def add_subnet(self, cidr_or_id, space, *zones):
 686         """Add an existing subnet to this model.
 687
 688         :param str cidr_or_id: CIDR or provider ID of the existing subnet
 689         :param str space: Network space with which to associate
 690         :param str \*zones: Zone(s) in which the subnet resides
 691
 692         """
 693         pass
 694
 695     def get_backups(self):
 696         """Retrieve metadata for backups in this model.
 697
 698         """
 699         pass
 700
 701     def block(self, *commands):
 702         """Add a new block to this model.
 703
 704         :param str \*commands: The commands to block. Valid values are
 705             'all-changes', 'destroy-model', 'remove-object'
 706
 707         """
 708         pass
 709
 710     def get_blocks(self):
 711         """List blocks for this model.
 712
 713         """
 714         pass
 715
 716     def get_cached_images(self, arch=None, kind=None, series=None):
 717         """Return a list of cached OS images.
 718
 719         :param str arch: Filter by image architecture
 720         :param str kind: Filter by image kind, e.g. 'lxd'
 721         :param str series: Filter by image series, e.g. 'xenial'
 722
 723         """
 724         pass
 725
 726     def create_backup(self, note=None, no_download=False):
 727         """Create a backup of this model.
 728
 729         :param str note: A note to store with the backup
 730         :param bool no_download: Do not download the backup archive
 731         :return str: Path to downloaded archive
 732
 733         """
 734         pass
 735
 736     def create_storage_pool(self, name, provider_type, **pool_config):
 737         """Create or define a storage pool.
 738
 739         :param str name: Name to give the storage pool
 740         :param str provider_type: Pool provider type
 741         :param \*\*pool_config: key/value pool configuration pairs
 742
 743         """
 744         pass
 745
 746     def debug_log(
 747             self, no_tail=False, exclude_module=None, include_module=None,
 748             include=None, level=None, limit=0, lines=10, replay=False,
 749             exclude=None):
 750         """Get log messages for this model.
 751
 752         :param bool no_tail: Stop after returning existing log messages
 753         :param list exclude_module: Do not show log messages for these logging
 754             modules
 755         :param list include_module: Only show log messages for these logging
 756             modules
 757         :param list include: Only show log messages for these entities
 758         :param str level: Log level to show, valid options are 'TRACE',
 759             'DEBUG', 'INFO', 'WARNING', 'ERROR,
 760         :param int limit: Return this many of the most recent (possibly
 761             filtered) lines are shown
 762         :param int lines: Yield this many of the most recent lines, and keep
 763             yielding
 764         :param bool replay: Yield the entire log, and keep yielding
 765         :param list exclude: Do not show log messages for these entities
 766
 767         """
 768         pass
 769
 770     async def deploy(
 771             self, entity_url, service_name=None, bind=None, budget=None,
 772             channel=None, config=None, constraints=None, force=False,
 773             num_units=1, plan=None, resources=None, series=None, storage=None,
 774             to=None):
 775         """Deploy a new service or bundle.
 776
 777         :param str entity_url: Charm or bundle url
 778         :param str service_name: Name to give the service
 779         :param dict bind: <charm endpoint>:<network space> pairs
 780         :param dict budget: <budget name>:<limit> pairs
 781         :param str channel: Charm store channel from which to retrieve
 782             the charm or bundle, e.g. 'development'
 783         :param dict config: Charm configuration dictionary
 784         :param constraints: Service constraints
 785         :type constraints: :class:`juju.Constraints`
 786         :param bool force: Allow charm to be deployed to a machine running
 787             an unsupported series
 788         :param int num_units: Number of units to deploy
 789         :param str plan: Plan under which to deploy charm
 790         :param dict resources: <resource name>:<file path> pairs
 791         :param str series: Series on which to deploy
 792         :param dict storage: Storage constraints TODO how do these look?
 793         :param str to: Placement directive, e.g.::
 794
 795             '23' - machine 23
 796             'lxc:7' - new lxc container on machine 7
 797             '24/lxc/3' - lxc container 3 or machine 24
 798
 799             If None, a new machine is provisioned.
 800
 801
 802         TODO::
 803
 804             - service_name is required; fill this in automatically if not
 805               provided by caller
 806             - series is required; how do we pick a default?
 807
 808         """
 809         if to:
 810             placement = [
 811                 client.Placement(**p) for p in to
 812             ]
 813         else:
 814             placement = []
 815
 816         if storage:
 817             storage = {
 818                 k: client.Constraints(**v)
 819                 for k, v in storage.items()
 820             }
 821
 822         entity_id = await self.charmstore.entityId(entity_url)
 823
 824         app_facade = client.ApplicationFacade()
 825         client_facade = client.ClientFacade()
 826         app_facade.connect(self.connection)
 827         client_facade.connect(self.connection)
 828
 829         if 'bundle/' in entity_id:
 830             handler = BundleHandler(self)
 831             await handler.fetch_plan(entity_id)
 832             await handler.execute_plan()
 833             extant_apps = {app for app in self.applications}
 834             pending_apps = set(handler.applications) - extant_apps
 835             if pending_apps:
 836                 # new apps will usually be in the model by now, but if some
 837                 # haven't made it yet we'll need to wait on them to be added
 838                 await asyncio.gather(*[
 839                     asyncio.ensure_future(
 840                         self.model._wait_for_new('application', app_name))
 841                     for app_name in pending_apps
 842                 ])
 843             return [app for name, app in self.applications.items()
 844                     if name in handler.applications]
 845         else:
 846             log.debug(
 847                 'Deploying %s', entity_id)
 848
 849             await client_facade.AddCharm(channel, entity_id)
 850             app = client.ApplicationDeploy(
 851                 application=service_name,
 852                 channel=channel,
 853                 charm_url=entity_id,
 854                 config=config,
 855                 constraints=constraints,
 856                 endpoint_bindings=bind,
 857                 num_units=num_units,
 858                 placement=placement,
 859                 resources=resources,
 860                 series=series,
 861                 storage=storage,
 862             )
 863
 864             await app_facade.Deploy([app])
 865             return await self._wait_for_new('application', service_name)
 866
 867     def destroy(self):
 868         """Terminate all machines and resources for this model.
 869
 870         """
 871         pass
 872
 873     async def destroy_unit(self, *unit_names):
 874         """Destroy units by name.
 875
 876         """
 877         app_facade = client.ApplicationFacade()
 878         app_facade.connect(self.connection)
 879
 880         log.debug(
 881             'Destroying unit%s %s',
 882             's' if len(unit_names) == 1 else '',
 883             ' '.join(unit_names))
 884
 885         return await app_facade.Destroy(self.name)
 886     destroy_units = destroy_unit
 887
 888     def get_backup(self, archive_id):
 889         """Download a backup archive file.
 890
 891         :param str archive_id: The id of the archive to download
 892         :return str: Path to the archive file
 893
 894         """
 895         pass
 896
 897     def enable_ha(
 898             self, num_controllers=0, constraints=None, series=None, to=None):
 899         """Ensure sufficient controllers exist to provide redundancy.
 900
 901         :param int num_controllers: Number of controllers to make available
 902         :param constraints: Constraints to apply to the controller machines
 903         :type constraints: :class:`juju.Constraints`
 904         :param str series: Series of the controller machines
 905         :param list to: Placement directives for controller machines, e.g.::
 906
 907             '23' - machine 23
 908             'lxc:7' - new lxc container on machine 7
 909             '24/lxc/3' - lxc container 3 or machine 24
 910
 911             If None, a new machine is provisioned.
 912
 913         """
 914         pass
 915
 916     def get_config(self):
 917         """Return the configuration settings for this model.
 918
 919         """
 920         pass
 921
 922     def get_constraints(self):
 923         """Return the machine constraints for this model.
 924
 925         """
 926         pass
 927
 928     def grant(self, username, acl='read'):
 929         """Grant a user access to this model.
 930
 931         :param str username: Username
 932         :param str acl: Access control ('read' or 'write')
 933
 934         """
 935         pass
 936
 937     def import_ssh_key(self, identity):
 938         """Add a public SSH key from a trusted indentity source to this model.
 939
 940         :param str identity: User identity in the form <lp|gh>:<username>
 941
 942         """
 943         pass
 944     import_ssh_keys = import_ssh_key
 945
 946     def get_machines(self, machine, utc=False):
 947         """Return list of machines in this model.
 948
 949         :param str machine: Machine id, e.g. '0'
 950         :param bool utc: Display time as UTC in RFC3339 format
 951
 952         """
 953         pass
 954
 955     def get_shares(self):
 956         """Return list of all users with access to this model.
 957
 958         """
 959         pass
 960
 961     def get_spaces(self):
 962         """Return list of all known spaces, including associated subnets.
 963
 964         """
 965         pass
 966
 967     def get_ssh_key(self):
 968         """Return known SSH keys for this model.
 969
 970         """
 971         pass
 972     get_ssh_keys = get_ssh_key
 973
 974     def get_storage(self, filesystem=False, volume=False):
 975         """Return details of storage instances.
 976
 977         :param bool filesystem: Include filesystem storage
 978         :param bool volume: Include volume storage
 979
 980         """
 981         pass
 982
 983     def get_storage_pools(self, names=None, providers=None):
 984         """Return list of storage pools.
 985
 986         :param list names: Only include pools with these names
 987         :param list providers: Only include pools for these providers
 988
 989         """
 990         pass
 991
 992     def get_subnets(self, space=None, zone=None):
 993         """Return list of known subnets.
 994
 995         :param str space: Only include subnets in this space
 996         :param str zone: Only include subnets in this zone
 997
 998         """
 999         pass
1000
1001     def remove_blocks(self):
1002         """Remove all blocks from this model.
1003
1004         """
1005         pass
1006
1007     def remove_backup(self, backup_id):
1008         """Delete a backup.
1009
1010         :param str backup_id: The id of the backup to remove
1011
1012         """
1013         pass
1014
1015     def remove_cached_images(self, arch=None, kind=None, series=None):
1016         """Remove cached OS images.
1017
1018         :param str arch: Architecture of the images to remove
1019         :param str kind: Image kind to remove, e.g. 'lxd'
1020         :param str series: Image series to remove, e.g. 'xenial'
1021
1022         """
1023         pass
1024
1025     def remove_machine(self, *machine_ids):
1026         """Remove a machine from this model.
1027
1028         :param str \*machine_ids: Ids of the machines to remove
1029
1030         """
1031         pass
1032     remove_machines = remove_machine
1033
1034     def remove_ssh_key(self, *keys):
1035         """Remove a public SSH key(s) from this model.
1036
1037         :param str \*keys: Keys to remove
1038
1039         """
1040         pass
1041     remove_ssh_keys = remove_ssh_key
1042
1043     def restore_backup(
1044             self, bootstrap=False, constraints=None, archive=None,
1045             backup_id=None, upload_tools=False):
1046         """Restore a backup archive to a new controller.
1047
1048         :param bool bootstrap: Bootstrap a new state machine
1049         :param constraints: Model constraints
1050         :type constraints: :class:`juju.Constraints`
1051         :param str archive: Path to backup archive to restore
1052         :param str backup_id: Id of backup to restore
1053         :param bool upload_tools: Upload tools if bootstrapping a new machine
1054
1055         """
1056         pass
1057
1058     def retry_provisioning(self):
1059         """Retry provisioning for failed machines.
1060
1061         """
1062         pass
1063
1064     def revoke(self, username, acl='read'):
1065         """Revoke a user's access to this model.
1066
1067         :param str username: Username to revoke
1068         :param str acl: Access control ('read' or 'write')
1069
1070         """
1071         pass
1072
1073     def run(self, command, timeout=None):
1074         """Run command on all machines in this model.
1075
1076         :param str command: The command to run
1077         :param int timeout: Time to wait before command is considered failed
1078
1079         """
1080         pass
1081
1082     def set_config(self, **config):
1083         """Set configuration keys on this model.
1084
1085         :param \*\*config: Config key/values
1086
1087         """
1088         pass
1089
1090     def set_constraints(self, constraints):
1091         """Set machine constraints on this model.
1092
1093         :param :class:`juju.Constraints` constraints: Machine constraints
1094
1095         """
1096         pass
1097
1098     def get_action_output(self, action_uuid, wait=-1):
1099         """Get the results of an action by ID.
1100
1101         :param str action_uuid: Id of the action
1102         :param int wait: Time in seconds to wait for action to complete
1103
1104         """
1105         pass
1106
1107     def get_action_status(self, uuid_or_prefix=None, name=None):
1108         """Get the status of all actions, filtered by ID, ID prefix, or action name.
1109
1110         :param str uuid_or_prefix: Filter by action uuid or prefix
1111         :param str name: Filter by action name
1112
1113         """
1114         pass
1115
1116     def get_budget(self, budget_name):
1117         """Get budget usage info.
1118
1119         :param str budget_name: Name of budget
1120
1121         """
1122         pass
1123
1124     def get_status(self, filter_=None, utc=False):
1125         """Return the status of the model.
1126
1127         :param str filter_: Service or unit name or wildcard ('*')
1128         :param bool utc: Display time as UTC in RFC3339 format
1129
1130         """
1131         pass
1132     status = get_status
1133
1134     def sync_tools(
1135             self, all_=False, destination=None, dry_run=False, public=False,
1136             source=None, stream=None, version=None):
1137         """Copy Juju tools into this model.
1138
1139         :param bool all_: Copy all versions, not just the latest
1140         :param str destination: Path to local destination directory
1141         :param bool dry_run: Don't do the actual copy
1142         :param bool public: Tools are for a public cloud, so generate mirrors
1143             information
1144         :param str source: Path to local source directory
1145         :param str stream: Simplestreams stream for which to sync metadata
1146         :param str version: Copy a specific major.minor version
1147
1148         """
1149         pass
1150
1151     def unblock(self, *commands):
1152         """Unblock an operation that would alter this model.
1153
1154         :param str \*commands: The commands to unblock. Valid values are
1155             'all-changes', 'destroy-model', 'remove-object'
1156
1157         """
1158         pass
1159
1160     def unset_config(self, *keys):
1161         """Unset configuration on this model.
1162
1163         :param str \*keys: The keys to unset
1164
1165         """
1166         pass
1167
1168     def upgrade_gui(self):
1169         """Upgrade the Juju GUI for this model.
1170
1171         """
1172         pass
1173
1174     def upgrade_juju(
1175             self, dry_run=False, reset_previous_upgrade=False,
1176             upload_tools=False, version=None):
1177         """Upgrade Juju on all machines in a model.
1178
1179         :param bool dry_run: Don't do the actual upgrade
1180         :param bool reset_previous_upgrade: Clear the previous (incomplete)
1181             upgrade status
1182         :param bool upload_tools: Upload local version of tools
1183         :param str version: Upgrade to a specific version
1184
1185         """
1186         pass
1187
1188     def upload_backup(self, archive_path):
1189         """Store a backup archive remotely in Juju.
1190
1191         :param str archive_path: Path to local archive
1192
1193         """
1194         pass
1195
1196     @property
1197     def charmstore(self):
1198         return self._charmstore
1199
1200
1201 class BundleHandler(object):
1202     """
1203     Handle bundles by using the API to translate bundle YAML into a plan of
1204     steps and then dispatching each of those using the API.
1205     """
1206     def __init__(self, model):
1207         self.model = model
1208         self.charmstore = model.charmstore
1209         self.plan = []
1210         self.references = {}
1211         self._units_by_app = {}
1212         for unit_name, unit in model.units.items():
1213             app_units = self._units_by_app.setdefault(unit.application, [])
1214             app_units.append(unit_name)
1215         self.client_facade = client.ClientFacade()
1216         self.client_facade.connect(model.connection)
1217         self.app_facade = client.ApplicationFacade()
1218         self.app_facade.connect(model.connection)
1219         self.ann_facade = client.AnnotationsFacade()
1220         self.ann_facade.connect(model.connection)
1221
1222     async def fetch_plan(self, entity_id):
1223         bundle_yaml = await self.charmstore.files(entity_id,
1224                                                   filename='bundle.yaml',
1225                                                   read_file=True)
1226         self.bundle = yaml.safe_load(bundle_yaml)
1227         self.plan = await self.client_facade.GetBundleChanges(bundle_yaml)
1228
1229     async def execute_plan(self):
1230         for step in self.plan.changes:
1231             method = getattr(self, step.method)
1232             result = await method(*step.args)
1233             self.references[step.id_] = result
1234
1235     @property
1236     def applications(self):
1237         return list(self.bundle['services'].keys())
1238
1239     def resolve(self, reference):
1240         if reference and reference.startswith('$'):
1241             reference = self.references[reference[1:]]
1242         return reference
1243
1244     async def addCharm(self, charm, series):
1245         """
1246         :param charm string:
1247             Charm holds the URL of the charm to be added.
1248
1249         :param series string:
1250             Series holds the series of the charm to be added
1251             if the charm default is not sufficient.
1252         """
1253         entity_id = await self.charmstore.entityId(charm)
1254         log.debug('Adding %s', entity_id)
1255         await self.client_facade.AddCharm(None, entity_id)
1256         return entity_id
1257
1258     async def addMachines(self, series, constraints, container_type,
1259                           parent_id):
1260         """
1261         :param series string:
1262             Series holds the optional machine OS series.
1263
1264         :param constraints string:
1265             Constraints holds the optional machine constraints.
1266
1267         :param Container_type string:
1268             ContainerType optionally holds the type of the container (for
1269             instance ""lxc" or kvm"). It is not specified for top level
1270             machines.
1271
1272         :param parent_id string:
1273             ParentId optionally holds a placeholder pointing to another machine
1274             change or to a unit change. This value is only specified in the
1275             case this machine is a container, in which case also ContainerType
1276             is set.
1277         """
1278         params = client.AddMachineParams(
1279             series=series,
1280             constraints=constraints,
1281             container_type=container_type,
1282             parent_id=self.resolve(parent_id),
1283         )
1284         results = await self.client_facade.AddMachines(params)
1285         log.debug('Added new machine %s', results[0].machine)
1286         return results[0].machine
1287
1288     async def addRelation(self, endpoint1, endpoint2):
1289         """
1290         :param endpoint1 string:
1291         :param endpoint2 string:
1292             Endpoint1 and Endpoint2 hold relation endpoints in the
1293             "application:interface" form, where the application is always a
1294             placeholder pointing to an application change, and the interface is
1295             optional. Examples are "$deploy-42:web" or just "$deploy-42".
1296         """
1297         endpoints = [endpoint1, endpoint2]
1298         # resolve indirect references
1299         for i in range(len(endpoints)):
1300             parts = endpoints[i].split(':')
1301             parts[0] = self.resolve(parts[0])
1302             endpoints[i] = ':'.join(parts)
1303
1304         log.info('Relating %s <-> %s', *endpoints)
1305         return await self.model.add_relation(*endpoints)
1306
1307     async def deploy(self, charm, series, application, options, constraints,
1308                      storage, endpoint_bindings, resources):
1309         """
1310         :param charm string:
1311             Charm holds the URL of the charm to be used to deploy this
1312             application.
1313
1314         :param series string:
1315             Series holds the series of the application to be deployed
1316             if the charm default is not sufficient.
1317
1318         :param application string:
1319             Application holds the application name.
1320
1321         :param options map[string]interface{}:
1322             Options holds application options.
1323
1324         :param constraints string:
1325             Constraints holds the optional application constraints.
1326
1327         :param storage map[string]string:
1328             Storage holds the optional storage constraints.
1329
1330         :param endpoint_bindings map[string]string:
1331             EndpointBindings holds the optional endpoint bindings
1332
1333         :param resources map[string]int:
1334             Resources identifies the revision to use for each resource
1335             of the application's charm.
1336         """
1337         # resolve indirect references
1338         charm = self.resolve(charm)
1339         # stringify all config values for API
1340         options = {k: str(v) for k, v in options.items()}
1341         # build param object
1342         app = client.ApplicationDeploy(
1343             charm_url=charm,
1344             series=series,
1345             application=application,
1346             config=options,
1347             constraints=constraints,
1348             storage=storage,
1349             endpoint_bindings=endpoint_bindings,
1350             resources=resources,
1351         )
1352         # do the do
1353         log.info('Deploying %s', charm)
1354         await self.app_facade.Deploy([app])
1355         return application
1356
1357     async def addUnit(self, application, to):
1358         """
1359         :param application string:
1360             Application holds the application placeholder name for which a unit
1361             is added.
1362
1363         :param to string:
1364             To holds the optional location where to add the unit, as a
1365             placeholder pointing to another unit change or to a machine change.
1366         """
1367         application = self.resolve(application)
1368         placement = self.resolve(to)
1369         if self._units_by_app.get(application):
1370             # enough units for this application already exist;
1371             # claim one, and carry on
1372             # NB: this should probably honor placement, but the juju client
1373             # doesn't, so we're not bothering, either
1374             unit_name = self._units_by_app[application].pop()
1375             log.debug('Reusing unit %s for %s', unit_name, application)
1376             return self.model.units[unit_name]
1377
1378         log.debug('Adding new unit for %s%s', application,
1379                   ' to %s' % placement if placement else '')
1380         return await self.model.applications[application].add_unit(
1381             count=1,
1382             to=placement,
1383         )
1384
1385     async def expose(self, application):
1386         """
1387         :param application string:
1388             Application holds the placeholder name of the application that must
1389             be exposed.
1390         """
1391         application = self.resolve(application)
1392         log.info('Exposing %s', application)
1393         return await self.model.applications[application].expose()
1394
1395     async def setAnnotations(self, id_, entity_type, annotations):
1396         """
1397         :param id_ string:
1398             Id is the placeholder for the application or machine change
1399             corresponding to the entity to be annotated.
1400
1401         :param entity_type EntityType:
1402             EntityType holds the type of the entity, "application" or
1403             "machine".
1404
1405         :param annotations map[string]string:
1406             Annotations holds the annotations as key/value pairs.
1407         """
1408         entity_id = self.resolve(id_)
1409         try:
1410             entity = self.model.state.get_entity(entity_type, entity_id)
1411         except KeyError:
1412             entity = await self.model._wait_for_new(entity_type, entity_id)
1413         return await entity.set_annotations(annotations)
1414
1415
1416 class CharmStore(object):
1417     """
1418     Async wrapper around theblues.charmstore.CharmStore
1419     """
1420     def __init__(self, loop):
1421         self.loop = loop
1422         self._cs = charmstore.CharmStore()
1423
1424     def __getattr__(self, name):
1425         """
1426         Wrap method calls in coroutines that use run_in_executor to make them
1427         async.
1428         """
1429         attr = getattr(self._cs, name)
1430         if not callable(attr):
1431             wrapper = partial(getattr, self._cs, name)
1432             setattr(self, name, wrapper)
1433         else:
1434             async def coro(*args, **kwargs):
1435                 method = partial(attr, *args, **kwargs)
1436                 return await self.loop.run_in_executor(None, method)
1437             setattr(self, name, coro)
1438             wrapper = coro
1439         return wrapper