juju/model.py

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