juju/model.py

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