juju/model.py

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