juju/model.py

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