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