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