modules/libjuju/juju/application.py

   1 # Copyright 2016 Canonical Ltd.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 #     Unless required by applicable law or agreed to in writing, software
  10 #     distributed under the License is distributed on an "AS IS" BASIS,
  11 #     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 #     See the License for the specific language governing permissions and
  13 #     limitations under the License.
  14
  15 import asyncio
  16 import logging
  17
  18 from . import model
  19 from .client import client
  20 from .errors import JujuError
  21 from .placement import parse as parse_placement
  22
  23 log = logging.getLogger(__name__)
  24
  25
  26 class Application(model.ModelEntity):
  27     @property
  28     def _unit_match_pattern(self):
  29         return r'^{}.*$'.format(self.entity_id)
  30
  31     def on_unit_add(self, callable_):
  32         """Add a "unit added" observer to this entity, which will be called
  33         whenever a unit is added to this application.
  34
  35         """
  36         self.model.add_observer(
  37             callable_, 'unit', 'add', self._unit_match_pattern)
  38
  39     def on_unit_remove(self, callable_):
  40         """Add a "unit removed" observer to this entity, which will be called
  41         whenever a unit is removed from this application.
  42
  43         """
  44         self.model.add_observer(
  45             callable_, 'unit', 'remove', self._unit_match_pattern)
  46
  47     @property
  48     def units(self):
  49         return [
  50             unit for unit in self.model.units.values()
  51             if unit.application == self.name
  52         ]
  53
  54     @property
  55     def relations(self):
  56         return [rel for rel in self.model.relations if rel.matches(self.name)]
  57
  58     def related_applications(self, endpoint_name=None):
  59         apps = {}
  60         for rel in self.relations:
  61             if rel.is_peer:
  62                 local_ep, remote_ep = rel.endpoints[0]
  63             else:
  64                 def is_us(ep):
  65                     return ep.application.name == self.name
  66                 local_ep, remote_ep = sorted(rel.endpoints, key=is_us)
  67             if endpoint_name is not None and endpoint_name != local_ep.name:
  68                 continue
  69             apps[remote_ep.application.name] = remote_ep.application
  70         return apps
  71
  72     @property
  73     def status(self):
  74         """Get the application status, as set by the charm's leader.
  75
  76         """
  77         return self.safe_data['status']['current']
  78
  79     @property
  80     def status_message(self):
  81         """Get the application status message, as set by the charm's leader.
  82
  83         """
  84         return self.safe_data['status']['message']
  85
  86     @property
  87     def tag(self):
  88         return 'application-%s' % self.name
  89
  90     async def add_relation(self, local_relation, remote_relation):
  91         """Add a relation to another application.
  92
  93         :param str local_relation: Name of relation on this application
  94         :param str remote_relation: Name of relation on the other
  95             application in the form '<application>[:<relation_name>]'
  96
  97         """
  98         if ':' not in local_relation:
  99             local_relation = '{}:{}'.format(self.name, local_relation)
 100
 101         return await self.model.add_relation(local_relation, remote_relation)
 102
 103     async def add_unit(self, count=1, to=None):
 104         """Add one or more units to this application.
 105
 106         :param int count: Number of units to add
 107         :param str to: Placement directive, e.g.::
 108             '23' - machine 23
 109             'lxc:7' - new lxc container on machine 7
 110             '24/lxc/3' - lxc container 3 or machine 24
 111
 112             If None, a new machine is provisioned.
 113
 114         """
 115         app_facade = client.ApplicationFacade.from_connection(self.connection)
 116
 117         log.debug(
 118             'Adding %s unit%s to %s',
 119             count, '' if count == 1 else 's', self.name)
 120
 121         result = await app_facade.AddUnits(
 122             application=self.name,
 123             placement=parse_placement(to) if to else None,
 124             num_units=count,
 125         )
 126
 127         return await asyncio.gather(*[
 128             asyncio.ensure_future(self.model._wait_for_new('unit', unit_id))
 129             for unit_id in result.units
 130         ])
 131
 132     add_units = add_unit
 133
 134     async def scale(self, scale=None, scale_change=None):
 135         """
 136         Set or adjust the scale of this (K8s) application.
 137
 138         One or the other of scale or scale_change must be provided.
 139
 140         :param int scale: Scale to which to set this application.
 141         :param int scale_change: Amount by which to adjust the scale of this
 142             application (can be positive or negative).
 143         """
 144         app_facade = client.ApplicationFacade.from_connection(self.connection)
 145
 146         if (scale, scale_change) == (None, None):
 147             raise ValueError('Must provide either scale or scale_change')
 148
 149         log.debug(
 150             'Scaling application %s %s %s',
 151             self.name, 'to' if scale else 'by', scale or scale_change)
 152
 153         await app_facade.ScaleApplications([
 154             client.ScaleApplicationParam(application_tag=self.tag,
 155                                          scale=scale,
 156                                          scale_change=scale_change)
 157         ])
 158
 159     def allocate(self, budget, value):
 160         """Allocate budget to this application.
 161
 162         :param str budget: Name of budget
 163         :param int value: Budget limit
 164
 165         """
 166         raise NotImplementedError()
 167
 168     def attach(self, resource_name, file_path):
 169         """Upload a file as a resource for this application.
 170
 171         :param str resource: Name of the resource
 172         :param str file_path: Path to the file to upload
 173
 174         """
 175         raise NotImplementedError()
 176
 177     def collect_metrics(self):
 178         """Collect metrics on this application.
 179
 180         """
 181         raise NotImplementedError()
 182
 183     async def destroy_relation(self, local_relation, remote_relation):
 184         """Remove a relation to another application.
 185
 186         :param str local_relation: Name of relation on this application
 187         :param str remote_relation: Name of relation on the other
 188             application in the form '<application>[:<relation_name>]'
 189
 190         """
 191         if ':' not in local_relation:
 192             local_relation = '{}:{}'.format(self.name, local_relation)
 193
 194         app_facade = client.ApplicationFacade.from_connection(self.connection)
 195
 196         log.debug(
 197             'Destroying relation %s <-> %s', local_relation, remote_relation)
 198
 199         return await app_facade.DestroyRelation([
 200             local_relation, remote_relation])
 201     remove_relation = destroy_relation
 202
 203     async def destroy_unit(self, *unit_names):
 204         """Destroy units by name.
 205
 206         """
 207         return await self.model.destroy_units(*unit_names)
 208     destroy_units = destroy_unit
 209
 210     async def destroy(self):
 211         """Remove this application from the model.
 212
 213         """
 214         app_facade = client.ApplicationFacade.from_connection(self.connection)
 215
 216         log.debug(
 217             'Destroying %s', self.name)
 218
 219         return await app_facade.Destroy(self.name)
 220     remove = destroy
 221
 222     async def expose(self):
 223         """Make this application publicly available over the network.
 224
 225         """
 226         app_facade = client.ApplicationFacade.from_connection(self.connection)
 227
 228         log.debug(
 229             'Exposing %s', self.name)
 230
 231         return await app_facade.Expose(self.name)
 232
 233     async def get_config(self):
 234         """Return the configuration settings dict for this application.
 235
 236         """
 237         app_facade = client.ApplicationFacade.from_connection(self.connection)
 238
 239         log.debug(
 240             'Getting config for %s', self.name)
 241
 242         return (await app_facade.Get(self.name)).config
 243
 244     async def get_constraints(self):
 245         """Return the machine constraints dict for this application.
 246
 247         """
 248         app_facade = client.ApplicationFacade.from_connection(self.connection)
 249
 250         log.debug(
 251             'Getting constraints for %s', self.name)
 252
 253         result = (await app_facade.Get(self.name)).constraints
 254         return vars(result) if result else result
 255
 256     async def get_actions(self, schema=False):
 257         """Get actions defined for this application.
 258
 259         :param bool schema: Return the full action schema
 260         :return dict: The charms actions, empty dict if none are defined.
 261         """
 262         actions = {}
 263         entity = [{"tag": self.tag}]
 264         action_facade = client.ActionFacade.from_connection(self.connection)
 265         results = (
 266             await action_facade.ApplicationsCharmsActions(entity)).results
 267         for result in results:
 268             if result.application_tag == self.tag and result.actions:
 269                 actions = result.actions
 270                 break
 271         if not schema:
 272             actions = {k: v['description'] for k, v in actions.items()}
 273         return actions
 274
 275     async def get_resources(self):
 276         """Return resources for this application.
 277
 278         Returns a dict mapping resource name to
 279         :class:`~juju._definitions.CharmResource` instances.
 280         """
 281         facade = client.ResourcesFacade.from_connection(self.connection)
 282         response = await facade.ListResources([client.Entity(self.tag)])
 283
 284         resources = dict()
 285         for result in response.results:
 286             for resource in result.charm_store_resources or []:
 287                 resources[resource.name] = resource
 288             for resource in result.resources or []:
 289                 if resource.charmresource:
 290                     resource = resource.charmresource
 291                     resources[resource.name] = resource
 292         return resources
 293
 294     async def run(self, command, timeout=None):
 295         """Run command on all units for this application.
 296
 297         :param str command: The command to run
 298         :param int timeout: Time to wait before command is considered failed
 299
 300         """
 301         action = client.ActionFacade.from_connection(self.connection)
 302
 303         log.debug(
 304             'Running `%s` on all units of %s', command, self.name)
 305
 306         # TODO this should return a list of Actions
 307         return await action.Run(
 308             [self.name],
 309             command,
 310             [],
 311             timeout,
 312             [],
 313         )
 314
 315     async def set_annotations(self, annotations):
 316         """Set annotations on this application.
 317
 318         :param annotations map[string]string: the annotations as key/value
 319             pairs.
 320
 321         """
 322         log.debug('Updating annotations on application %s', self.name)
 323
 324         self.ann_facade = client.AnnotationsFacade.from_connection(
 325             self.connection)
 326
 327         ann = client.EntityAnnotations(
 328             entity=self.tag,
 329             annotations=annotations,
 330         )
 331         return await self.ann_facade.Set([ann])
 332
 333     async def set_config(self, config):
 334         """Set configuration options for this application.
 335
 336         :param config: Dict of configuration to set
 337         """
 338         app_facade = client.ApplicationFacade.from_connection(self.connection)
 339
 340         log.debug(
 341             'Setting config for %s: %s', self.name, config)
 342
 343         return await app_facade.Set(self.name, config)
 344
 345     async def reset_config(self, to_default):
 346         """
 347         Restore application config to default values.
 348
 349         :param list to_default: A list of config options to be reset to their
 350         default value.
 351         """
 352         app_facade = client.ApplicationFacade.from_connection(self.connection)
 353
 354         log.debug(
 355             'Restoring default config for %s: %s', self.name, to_default)
 356
 357         return await app_facade.Unset(self.name, to_default)
 358
 359     async def set_constraints(self, constraints):
 360         """Set machine constraints for this application.
 361
 362         :param dict constraints: Dict of machine constraints
 363
 364         """
 365         app_facade = client.ApplicationFacade.from_connection(self.connection)
 366
 367         log.debug(
 368             'Setting constraints for %s: %s', self.name, constraints)
 369
 370         return await app_facade.SetConstraints(self.name, constraints)
 371
 372     def set_meter_status(self, status, info=None):
 373         """Set the meter status on this status.
 374
 375         :param str status: Meter status, e.g. 'RED', 'AMBER'
 376         :param str info: Extra info message
 377
 378         """
 379         raise NotImplementedError()
 380
 381     def set_plan(self, plan_name):
 382         """Set the plan for this application, effective immediately.
 383
 384         :param str plan_name: Name of plan
 385
 386         """
 387         raise NotImplementedError()
 388
 389     async def unexpose(self):
 390         """Remove public availability over the network for this application.
 391
 392         """
 393         app_facade = client.ApplicationFacade.from_connection(self.connection)
 394
 395         log.debug(
 396             'Unexposing %s', self.name)
 397
 398         return await app_facade.Unexpose(self.name)
 399
 400     def update_allocation(self, allocation):
 401         """Update existing allocation for this application.
 402
 403         :param int allocation: The allocation to set
 404
 405         """
 406         raise NotImplementedError()
 407
 408     async def upgrade_charm(
 409             self, channel=None, force_series=False, force_units=False,
 410             path=None, resources=None, revision=None, switch=None):
 411         """Upgrade the charm for this application.
 412
 413         :param str channel: Channel to use when getting the charm from the
 414             charm store, e.g. 'development'
 415         :param bool force_series: Upgrade even if series of deployed
 416             application is not supported by the new charm
 417         :param bool force_units: Upgrade all units immediately, even if in
 418             error state
 419         :param str path: Uprade to a charm located at path
 420         :param dict resources: Dictionary of resource name/filepath pairs
 421         :param int revision: Explicit upgrade revision
 422         :param str switch: Crossgrade charm url
 423
 424         """
 425         # TODO: Support local upgrades
 426         if path is not None:
 427             raise NotImplementedError("path option is not implemented")
 428         if resources is not None:
 429             raise NotImplementedError("resources option is not implemented")
 430
 431         if switch is not None and revision is not None:
 432             raise ValueError("switch and revision are mutually exclusive")
 433
 434         client_facade = client.ClientFacade.from_connection(self.connection)
 435         resources_facade = client.ResourcesFacade.from_connection(
 436             self.connection)
 437         app_facade = client.ApplicationFacade.from_connection(self.connection)
 438
 439         charmstore = self.model.charmstore
 440         charmstore_entity = None
 441
 442         if switch is not None:
 443             charm_url = switch
 444             if not charm_url.startswith('cs:'):
 445                 charm_url = 'cs:' + charm_url
 446         else:
 447             charm_url = self.data['charm-url']
 448             charm_url = charm_url.rpartition('-')[0]
 449             if revision is not None:
 450                 charm_url = "%s-%d" % (charm_url, revision)
 451             else:
 452                 charmstore_entity = await charmstore.entity(charm_url,
 453                                                             channel=channel)
 454                 charm_url = charmstore_entity['Id']
 455
 456         if charm_url == self.data['charm-url']:
 457             raise JujuError('already running charm "%s"' % charm_url)
 458
 459         # Update charm
 460         await client_facade.AddCharm(
 461             url=charm_url,
 462             channel=channel
 463         )
 464
 465         # Update resources
 466         if not charmstore_entity:
 467             charmstore_entity = await charmstore.entity(charm_url,
 468                                                         channel=channel)
 469         store_resources = charmstore_entity['Meta']['resources']
 470
 471         request_data = [client.Entity(self.tag)]
 472         response = await resources_facade.ListResources(request_data)
 473         existing_resources = {
 474             resource.name: resource
 475             for resource in response.results[0].resources
 476         }
 477
 478         resources_to_update = [
 479             resource for resource in store_resources
 480             if resource['Name'] not in existing_resources or
 481             existing_resources[resource['Name']].origin != 'upload'
 482         ]
 483
 484         if resources_to_update:
 485             request_data = [
 486                 client.CharmResource(
 487                     description=resource.get('Description'),
 488                     fingerprint=resource['Fingerprint'],
 489                     name=resource['Name'],
 490                     path=resource['Path'],
 491                     revision=resource['Revision'],
 492                     size=resource['Size'],
 493                     type_=resource['Type'],
 494                     origin='store',
 495                 ) for resource in resources_to_update
 496             ]
 497             response = await resources_facade.AddPendingResources(
 498                 self.tag,
 499                 charm_url,
 500                 request_data
 501             )
 502             pending_ids = response.pending_ids
 503             resource_ids = {
 504                 resource['Name']: id
 505                 for resource, id in zip(resources_to_update, pending_ids)
 506             }
 507         else:
 508             resource_ids = None
 509
 510         # Update application
 511         await app_facade.SetCharm(
 512             application=self.entity_id,
 513             channel=channel,
 514             charm_url=charm_url,
 515             config_settings=None,
 516             config_settings_yaml=None,
 517             force_series=force_series,
 518             force_units=force_units,
 519             resource_ids=resource_ids,
 520             storage_constraints=None
 521         )
 522
 523         await self.model.block_until(
 524             lambda: self.data['charm-url'] == charm_url
 525         )
 526
 527     async def get_metrics(self):
 528         """Get metrics for this application's units.
 529
 530         :return: Dictionary of unit_name:metrics
 531
 532         """
 533         return await self.model.get_metrics(self.tag)