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