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