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