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             await app_facade.Deploy([app])
 935             return await self._wait_for_new('application', application_name)
 936
 937     def destroy(self):
 938         """Terminate all machines and resources for this model.
 939
 940         """
 941         pass
 942
 943     async def destroy_unit(self, *unit_names):
 944         """Destroy units by name.
 945
 946         """
 947         app_facade = client.ApplicationFacade()
 948         app_facade.connect(self.connection)
 949
 950         log.debug(
 951             'Destroying unit%s %s',
 952             's' if len(unit_names) == 1 else '',
 953             ' '.join(unit_names))
 954
 955         return await app_facade.Destroy(self.name)
 956     destroy_units = destroy_unit
 957
 958     def get_backup(self, archive_id):
 959         """Download a backup archive file.
 960
 961         :param str archive_id: The id of the archive to download
 962         :return str: Path to the archive file
 963
 964         """
 965         pass
 966
 967     def enable_ha(
 968             self, num_controllers=0, constraints=None, series=None, to=None):
 969         """Ensure sufficient controllers exist to provide redundancy.
 970
 971         :param int num_controllers: Number of controllers to make available
 972         :param constraints: Constraints to apply to the controller machines
 973         :type constraints: :class:`juju.Constraints`
 974         :param str series: Series of the controller machines
 975         :param list to: Placement directives for controller machines, e.g.::
 976
 977             '23' - machine 23
 978             'lxc:7' - new lxc container on machine 7
 979             '24/lxc/3' - lxc container 3 or machine 24
 980
 981             If None, a new machine is provisioned.
 982
 983         """
 984         pass
 985
 986     def get_config(self):
 987         """Return the configuration settings for this model.
 988
 989         """
 990         pass
 991
 992     def get_constraints(self):
 993         """Return the machine constraints for this model.
 994
 995         """
 996         pass
 997
 998     def grant(self, username, acl='read'):
 999         """Grant a user access to this model.
1000
1001         :param str username: Username
1002         :param str acl: Access control ('read' or 'write')
1003
1004         """
1005         pass
1006
1007     def import_ssh_key(self, identity):
1008         """Add a public SSH key from a trusted indentity source to this model.
1009
1010         :param str identity: User identity in the form <lp|gh>:<username>
1011
1012         """
1013         pass
1014     import_ssh_keys = import_ssh_key
1015
1016     def get_machines(self, machine, utc=False):
1017         """Return list of machines in this model.
1018
1019         :param str machine: Machine id, e.g. '0'
1020         :param bool utc: Display time as UTC in RFC3339 format
1021
1022         """
1023         pass
1024
1025     def get_shares(self):
1026         """Return list of all users with access to this model.
1027
1028         """
1029         pass
1030
1031     def get_spaces(self):
1032         """Return list of all known spaces, including associated subnets.
1033
1034         """
1035         pass
1036
1037     def get_ssh_key(self):
1038         """Return known SSH keys for this model.
1039
1040         """
1041         pass
1042     get_ssh_keys = get_ssh_key
1043
1044     def get_storage(self, filesystem=False, volume=False):
1045         """Return details of storage instances.
1046
1047         :param bool filesystem: Include filesystem storage
1048         :param bool volume: Include volume storage
1049
1050         """
1051         pass
1052
1053     def get_storage_pools(self, names=None, providers=None):
1054         """Return list of storage pools.
1055
1056         :param list names: Only include pools with these names
1057         :param list providers: Only include pools for these providers
1058
1059         """
1060         pass
1061
1062     def get_subnets(self, space=None, zone=None):
1063         """Return list of known subnets.
1064
1065         :param str space: Only include subnets in this space
1066         :param str zone: Only include subnets in this zone
1067
1068         """
1069         pass
1070
1071     def remove_blocks(self):
1072         """Remove all blocks from this model.
1073
1074         """
1075         pass
1076
1077     def remove_backup(self, backup_id):
1078         """Delete a backup.
1079
1080         :param str backup_id: The id of the backup to remove
1081
1082         """
1083         pass
1084
1085     def remove_cached_images(self, arch=None, kind=None, series=None):
1086         """Remove cached OS images.
1087
1088         :param str arch: Architecture of the images to remove
1089         :param str kind: Image kind to remove, e.g. 'lxd'
1090         :param str series: Image series to remove, e.g. 'xenial'
1091
1092         """
1093         pass
1094
1095     def remove_machine(self, *machine_ids):
1096         """Remove a machine from this model.
1097
1098         :param str \*machine_ids: Ids of the machines to remove
1099
1100         """
1101         pass
1102     remove_machines = remove_machine
1103
1104     def remove_ssh_key(self, *keys):
1105         """Remove a public SSH key(s) from this model.
1106
1107         :param str \*keys: Keys to remove
1108
1109         """
1110         pass
1111     remove_ssh_keys = remove_ssh_key
1112
1113     def restore_backup(
1114             self, bootstrap=False, constraints=None, archive=None,
1115             backup_id=None, upload_tools=False):
1116         """Restore a backup archive to a new controller.
1117
1118         :param bool bootstrap: Bootstrap a new state machine
1119         :param constraints: Model constraints
1120         :type constraints: :class:`juju.Constraints`
1121         :param str archive: Path to backup archive to restore
1122         :param str backup_id: Id of backup to restore
1123         :param bool upload_tools: Upload tools if bootstrapping a new machine
1124
1125         """
1126         pass
1127
1128     def retry_provisioning(self):
1129         """Retry provisioning for failed machines.
1130
1131         """
1132         pass
1133
1134     def revoke(self, username, acl='read'):
1135         """Revoke a user's access to this model.
1136
1137         :param str username: Username to revoke
1138         :param str acl: Access control ('read' or 'write')
1139
1140         """
1141         pass
1142
1143     def run(self, command, timeout=None):
1144         """Run command on all machines in this model.
1145
1146         :param str command: The command to run
1147         :param int timeout: Time to wait before command is considered failed
1148
1149         """
1150         pass
1151
1152     def set_config(self, **config):
1153         """Set configuration keys on this model.
1154
1155         :param \*\*config: Config key/values
1156
1157         """
1158         pass
1159
1160     def set_constraints(self, constraints):
1161         """Set machine constraints on this model.
1162
1163         :param :class:`juju.Constraints` constraints: Machine constraints
1164
1165         """
1166         pass
1167
1168     def get_action_output(self, action_uuid, wait=-1):
1169         """Get the results of an action by ID.
1170
1171         :param str action_uuid: Id of the action
1172         :param int wait: Time in seconds to wait for action to complete
1173
1174         """
1175         pass
1176
1177     def get_action_status(self, uuid_or_prefix=None, name=None):
1178         """Get the status of all actions, filtered by ID, ID prefix, or action name.
1179
1180         :param str uuid_or_prefix: Filter by action uuid or prefix
1181         :param str name: Filter by action name
1182
1183         """
1184         pass
1185
1186     def get_budget(self, budget_name):
1187         """Get budget usage info.
1188
1189         :param str budget_name: Name of budget
1190
1191         """
1192         pass
1193
1194     def get_status(self, filter_=None, utc=False):
1195         """Return the status of the model.
1196
1197         :param str filter_: Service or unit name or wildcard ('*')
1198         :param bool utc: Display time as UTC in RFC3339 format
1199
1200         """
1201         pass
1202     status = get_status
1203
1204     def sync_tools(
1205             self, all_=False, destination=None, dry_run=False, public=False,
1206             source=None, stream=None, version=None):
1207         """Copy Juju tools into this model.
1208
1209         :param bool all_: Copy all versions, not just the latest
1210         :param str destination: Path to local destination directory
1211         :param bool dry_run: Don't do the actual copy
1212         :param bool public: Tools are for a public cloud, so generate mirrors
1213             information
1214         :param str source: Path to local source directory
1215         :param str stream: Simplestreams stream for which to sync metadata
1216         :param str version: Copy a specific major.minor version
1217
1218         """
1219         pass
1220
1221     def unblock(self, *commands):
1222         """Unblock an operation that would alter this model.
1223
1224         :param str \*commands: The commands to unblock. Valid values are
1225             'all-changes', 'destroy-model', 'remove-object'
1226
1227         """
1228         pass
1229
1230     def unset_config(self, *keys):
1231         """Unset configuration on this model.
1232
1233         :param str \*keys: The keys to unset
1234
1235         """
1236         pass
1237
1238     def upgrade_gui(self):
1239         """Upgrade the Juju GUI for this model.
1240
1241         """
1242         pass
1243
1244     def upgrade_juju(
1245             self, dry_run=False, reset_previous_upgrade=False,
1246             upload_tools=False, version=None):
1247         """Upgrade Juju on all machines in a model.
1248
1249         :param bool dry_run: Don't do the actual upgrade
1250         :param bool reset_previous_upgrade: Clear the previous (incomplete)
1251             upgrade status
1252         :param bool upload_tools: Upload local version of tools
1253         :param str version: Upgrade to a specific version
1254
1255         """
1256         pass
1257
1258     def upload_backup(self, archive_path):
1259         """Store a backup archive remotely in Juju.
1260
1261         :param str archive_path: Path to local archive
1262
1263         """
1264         pass
1265
1266     @property
1267     def charmstore(self):
1268         return self._charmstore
1269
1270     async def get_metrics(self, *tags):
1271         """Retrieve metrics.
1272
1273         :param str \*tags: Tags of entities from which to retrieve metrics.
1274             No tags retrieves the metrics of all units in the model.
1275         :return: Dictionary of unit_name:metrics
1276
1277         """
1278         log.debug("Retrieving metrics for %s",
1279                   ', '.join(tags) if tags else "all units")
1280
1281         metrics_facade = client.MetricsDebugFacade()
1282         metrics_facade.connect(self.connection)
1283
1284         entities = [client.Entity(tag) for tag in tags]
1285         metrics_result = await metrics_facade.GetMetrics(entities)
1286
1287         metrics = collections.defaultdict(list)
1288
1289         for entity_metrics in metrics_result.results:
1290             error = entity_metrics.error
1291             if error:
1292                 if "is not a valid tag" in error:
1293                     raise ValueError(error.message)
1294                 else:
1295                     raise Exception(error.message)
1296
1297             for metric in entity_metrics.metrics:
1298                 metrics[metric.unit].append(vars(metric))
1299
1300         return metrics
1301
1302
1303 class BundleHandler(object):
1304     """
1305     Handle bundles by using the API to translate bundle YAML into a plan of
1306     steps and then dispatching each of those using the API.
1307     """
1308     def __init__(self, model):
1309         self.model = model
1310         self.charmstore = model.charmstore
1311         self.plan = []
1312         self.references = {}
1313         self._units_by_app = {}
1314         for unit_name, unit in model.units.items():
1315             app_units = self._units_by_app.setdefault(unit.application, [])
1316             app_units.append(unit_name)
1317         self.client_facade = client.ClientFacade()
1318         self.client_facade.connect(model.connection)
1319         self.app_facade = client.ApplicationFacade()
1320         self.app_facade.connect(model.connection)
1321         self.ann_facade = client.AnnotationsFacade()
1322         self.ann_facade.connect(model.connection)
1323
1324     async def fetch_plan(self, entity_id):
1325         is_local = not entity_id.startswith('cs:') and os.path.isdir(entity_id)
1326         if is_local:
1327             bundle_yaml = (Path(entity_id) / "bundle.yaml").read_text()
1328         else:
1329             bundle_yaml = await self.charmstore.files(entity_id,
1330                                                       filename='bundle.yaml',
1331                                                       read_file=True)
1332         self.bundle = yaml.safe_load(bundle_yaml)
1333         self.plan = await self.client_facade.GetBundleChanges(bundle_yaml)
1334
1335         if self.plan.errors:
1336             raise JujuError('\n'.join(self.plan.errors))
1337
1338     async def execute_plan(self):
1339         for step in self.plan.changes:
1340             method = getattr(self, step.method)
1341             result = await method(*step.args)
1342             self.references[step.id_] = result
1343
1344     @property
1345     def applications(self):
1346         return list(self.bundle['services'].keys())
1347
1348     def resolve(self, reference):
1349         if reference and reference.startswith('$'):
1350             reference = self.references[reference[1:]]
1351         return reference
1352
1353     async def addCharm(self, charm, series):
1354         """
1355         :param charm string:
1356             Charm holds the URL of the charm to be added.
1357
1358         :param series string:
1359             Series holds the series of the charm to be added
1360             if the charm default is not sufficient.
1361         """
1362         entity_id = await self.charmstore.entityId(charm)
1363         log.debug('Adding %s', entity_id)
1364         await self.client_facade.AddCharm(None, entity_id)
1365         return entity_id
1366
1367     async def addMachines(self, params=None):
1368         """
1369         :param params dict:
1370             Dictionary specifying the machine to add. All keys are optional.
1371             Keys include:
1372
1373             series: string specifying the machine OS series.
1374
1375             constraints: string holding machine constraints, if any. We'll
1376                 parse this into the json friendly dict that the juju api
1377                 expects.
1378
1379             container_type: string holding the type of the container (for
1380                 instance ""lxd" or kvm"). It is not specified for top level
1381                 machines.
1382
1383             parent_id: string holding a placeholder pointing to another
1384                 machine change or to a unit change. This value is only
1385                 specified in the case this machine is a container, in
1386                 which case also ContainerType is set.
1387
1388         """
1389         params = params or {}
1390
1391         # Normalize keys
1392         params = {normalize_key(k): params[k] for k in params.keys()}
1393
1394         # Fix up values, as necessary.
1395         if 'parent_id' in params:
1396             params['parent_id'] = self.resolve(params['parent_id'])
1397
1398         params['constraints'] = parse_constraints(
1399             params.get('constraints'))
1400         params['jobs'] = params.get('jobs', ['JobHostUnits'])
1401
1402         if params.get('container_type') == 'lxc':
1403             log.warning('Juju 2.0 does not support lxc containers. '
1404                         'Converting containers to lxd.')
1405             params['container_type'] = 'lxd'
1406
1407         # Submit the request.
1408         params = client.AddMachineParams(**params)
1409         results = await self.client_facade.AddMachines([params])
1410         error = results.machines[0].error
1411         if error:
1412             raise ValueError("Error adding machine: %s", error.message)
1413         machine = results.machines[0].machine
1414         log.debug('Added new machine %s', machine)
1415         return machine
1416
1417     async def addRelation(self, endpoint1, endpoint2):
1418         """
1419         :param endpoint1 string:
1420         :param endpoint2 string:
1421             Endpoint1 and Endpoint2 hold relation endpoints in the
1422             "application:interface" form, where the application is always a
1423             placeholder pointing to an application change, and the interface is
1424             optional. Examples are "$deploy-42:web" or just "$deploy-42".
1425         """
1426         endpoints = [endpoint1, endpoint2]
1427         # resolve indirect references
1428         for i in range(len(endpoints)):
1429             parts = endpoints[i].split(':')
1430             parts[0] = self.resolve(parts[0])
1431             endpoints[i] = ':'.join(parts)
1432
1433         log.info('Relating %s <-> %s', *endpoints)
1434         return await self.model.add_relation(*endpoints)
1435
1436     async def deploy(self, charm, series, application, options, constraints,
1437                      storage, endpoint_bindings, resources):
1438         """
1439         :param charm string:
1440             Charm holds the URL of the charm to be used to deploy this
1441             application.
1442
1443         :param series string:
1444             Series holds the series of the application to be deployed
1445             if the charm default is not sufficient.
1446
1447         :param application string:
1448             Application holds the application name.
1449
1450         :param options map[string]interface{}:
1451             Options holds application options.
1452
1453         :param constraints string:
1454             Constraints holds the optional application constraints.
1455
1456         :param storage map[string]string:
1457             Storage holds the optional storage constraints.
1458
1459         :param endpoint_bindings map[string]string:
1460             EndpointBindings holds the optional endpoint bindings
1461
1462         :param resources map[string]int:
1463             Resources identifies the revision to use for each resource
1464             of the application's charm.
1465         """
1466         # resolve indirect references
1467         charm = self.resolve(charm)
1468         # stringify all config values for API
1469         options = {k: str(v) for k, v in options.items()}
1470         # build param object
1471         app = client.ApplicationDeploy(
1472             charm_url=charm,
1473             series=series,
1474             application=application,
1475             config=options,
1476             constraints=parse_constraints(constraints),
1477             storage=storage,
1478             endpoint_bindings=endpoint_bindings,
1479             resources=resources,
1480         )
1481         # do the do
1482         log.info('Deploying %s', charm)
1483         await self.app_facade.Deploy([app])
1484         # ensure the app is in the model for future operations
1485         await self.model._wait_for_new('application', application)
1486         return application
1487
1488     async def addUnit(self, application, to):
1489         """
1490         :param application string:
1491             Application holds the application placeholder name for which a unit
1492             is added.
1493
1494         :param to string:
1495             To holds the optional location where to add the unit, as a
1496             placeholder pointing to another unit change or to a machine change.
1497         """
1498         application = self.resolve(application)
1499         placement = self.resolve(to)
1500         if self._units_by_app.get(application):
1501             # enough units for this application already exist;
1502             # claim one, and carry on
1503             # NB: this should probably honor placement, but the juju client
1504             # doesn't, so we're not bothering, either
1505             unit_name = self._units_by_app[application].pop()
1506             log.debug('Reusing unit %s for %s', unit_name, application)
1507             return self.model.units[unit_name]
1508
1509         log.debug('Adding new unit for %s%s', application,
1510                   ' to %s' % placement if placement else '')
1511         return await self.model.applications[application].add_unit(
1512             count=1,
1513             to=placement,
1514         )
1515
1516     async def expose(self, application):
1517         """
1518         :param application string:
1519             Application holds the placeholder name of the application that must
1520             be exposed.
1521         """
1522         application = self.resolve(application)
1523         log.info('Exposing %s', application)
1524         return await self.model.applications[application].expose()
1525
1526     async def setAnnotations(self, id_, entity_type, annotations):
1527         """
1528         :param id_ string:
1529             Id is the placeholder for the application or machine change
1530             corresponding to the entity to be annotated.
1531
1532         :param entity_type EntityType:
1533             EntityType holds the type of the entity, "application" or
1534             "machine".
1535
1536         :param annotations map[string]string:
1537             Annotations holds the annotations as key/value pairs.
1538         """
1539         entity_id = self.resolve(id_)
1540         try:
1541             entity = self.model.state.get_entity(entity_type, entity_id)
1542         except KeyError:
1543             entity = await self.model._wait_for_new(entity_type, entity_id)
1544         return await entity.set_annotations(annotations)
1545
1546
1547 class CharmStore(object):
1548     """
1549     Async wrapper around theblues.charmstore.CharmStore
1550     """
1551     def __init__(self, loop):
1552         self.loop = loop
1553         self._cs = charmstore.CharmStore()
1554
1555     def __getattr__(self, name):
1556         """
1557         Wrap method calls in coroutines that use run_in_executor to make them
1558         async.
1559         """
1560         attr = getattr(self._cs, name)
1561         if not callable(attr):
1562             wrapper = partial(getattr, self._cs, name)
1563             setattr(self, name, wrapper)
1564         else:
1565             async def coro(*args, **kwargs):
1566                 method = partial(attr, *args, **kwargs)
1567                 return await self.loop.run_in_executor(None, method)
1568             setattr(self, name, coro)
1569             wrapper = coro
1570         return wrapper