juju/model.py

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