n2vc/vnf.py

   1
   2 import logging
   3 import os
   4 import os.path
   5 import re
   6 import ssl
   7 import sys
   8 import time
   9
  10 # FIXME: this should load the juju inside or modules without having to
  11 # explicitly install it. Check why it's not working.
  12 # Load our subtree of the juju library
  13 path = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
  14 path = os.path.join(path, "modules/libjuju/")
  15 if path not in sys.path:
  16     sys.path.insert(1, path)
  17
  18 from juju.controller import Controller
  19 from juju.model import Model, ModelObserver
  20
  21
  22 # We might need this to connect to the websocket securely, but test and verify.
  23 try:
  24     ssl._create_default_https_context = ssl._create_unverified_context
  25 except AttributeError:
  26     # Legacy Python doesn't verify by default (see pep-0476)
  27     #   https://www.python.org/dev/peps/pep-0476/
  28     pass
  29
  30
  31 # Custom exceptions
  32 class JujuCharmNotFound(Exception):
  33     """The Charm can't be found or is not readable."""
  34
  35
  36 class JujuApplicationExists(Exception):
  37     """The Application already exists."""
  38
  39 class N2VCPrimitiveExecutionFailed(Exception):
  40     """Something failed while attempting to execute a primitive."""
  41
  42
  43 # Quiet the debug logging
  44 logging.getLogger('websockets.protocol').setLevel(logging.INFO)
  45 logging.getLogger('juju.client.connection').setLevel(logging.WARN)
  46 logging.getLogger('juju.model').setLevel(logging.WARN)
  47 logging.getLogger('juju.machine').setLevel(logging.WARN)
  48
  49 class VCAMonitor(ModelObserver):
  50     """Monitor state changes within the Juju Model."""
  51     callback = None
  52     callback_args = None
  53     log = None
  54     ns_name = None
  55     application_name = None
  56
  57     def __init__(self, ns_name, application_name, callback, *args):
  58         self.log = logging.getLogger(__name__)
  59
  60         self.ns_name = ns_name
  61         self.application_name = application_name
  62         self.callback = callback
  63         self.callback_args = args
  64
  65     async def on_change(self, delta, old, new, model):
  66         """React to changes in the Juju model."""
  67
  68         if delta.entity == "unit":
  69             try:
  70                 if old and new:
  71                     old_status = old.workload_status
  72                     new_status = new.workload_status
  73
  74                     if old_status == new_status:
  75
  76                         """The workload status may fluctuate around certain events,
  77                         so wait until the status has stabilized before triggering
  78                         the callback."""
  79                         if self.callback:
  80                             self.callback(
  81                                 self.ns_name,
  82                                 self.application_name,
  83                                 new_status,
  84                                 *self.callback_args)
  85             except Exception as e:
  86                 self.log.debug("[1] notify_callback exception {}".format(e))
  87         elif delta.entity == "action":
  88             # TODO: Decide how we want to notify the user of actions
  89
  90             # uuid = delta.data['id']     # The Action's unique id
  91             # msg = delta.data['message'] # The output of the action
  92             #
  93             # if delta.data['status'] == "pending":
  94             #     # The action is queued
  95             #     pass
  96             # elif delta.data['status'] == "completed""
  97             #     # The action was successful
  98             #     pass
  99             # elif delta.data['status'] == "failed":
 100             #     # The action failed.
 101             #     pass
 102
 103             pass
 104
 105 ########
 106 # TODO
 107 #
 108 # Create unique models per network service
 109 # Document all public functions
 110
 111 class N2VC:
 112
 113     # Juju API
 114     api = None
 115     log = None
 116     controller = None
 117     connecting = False
 118     authenticated = False
 119
 120     models = {}
 121     default_model = None
 122
 123     # Model Observers
 124     monitors = {}
 125
 126     # VCA config
 127     hostname = ""
 128     port = 17070
 129     username = ""
 130     secret = ""
 131
 132     def __init__(self,
 133                  log=None,
 134                  server='127.0.0.1',
 135                  port=17070,
 136                  user='admin',
 137                  secret=None,
 138                  artifacts=None
 139                  ):
 140         """Initialize N2VC
 141
 142         :param vcaconfig dict A dictionary containing the VCA configuration
 143
 144         :param artifacts str The directory where charms required by a vnfd are
 145             stored.
 146
 147         :Example:
 148         n2vc = N2VC(vcaconfig={
 149             'secret': 'MzI3MDJhOTYxYmM0YzRjNTJiYmY1Yzdm',
 150             'user': 'admin',
 151             'ip-address': '10.44.127.137',
 152             'port': 17070,
 153             'artifacts': '/path/to/charms'
 154         })
 155
 156         """
 157
 158         if log:
 159             self.log = log
 160         else:
 161             self.log = logging.getLogger(__name__)
 162
 163         # Quiet websocket traffic
 164         logging.getLogger('websockets.protocol').setLevel(logging.INFO)
 165         logging.getLogger('juju.client.connection').setLevel(logging.WARN)
 166         logging.getLogger('model').setLevel(logging.WARN)
 167         # logging.getLogger('websockets.protocol').setLevel(logging.DEBUG)
 168
 169         self.log.debug('JujuApi: instantiated')
 170
 171         self.server = server
 172         self.port = port
 173
 174         self.secret = secret
 175         if user.startswith('user-'):
 176             self.user = user
 177         else:
 178             self.user = 'user-{}'.format(user)
 179
 180         self.endpoint = '%s:%d' % (server, int(port))
 181
 182         self.artifacts = artifacts
 183
 184     def __del__(self):
 185         """Close any open connections."""
 186         yield self.logout()
 187
 188     def notify_callback(self, model_name, application_name, status, callback=None, *callback_args):
 189         try:
 190             if callback:
 191                 callback(model_name, application_name, status, *callback_args)
 192         except Exception as e:
 193             self.log.error("[0] notify_callback exception {}".format(e))
 194             raise e
 195         return True
 196
 197     # Public methods
 198     async def CreateNetworkService(self, nsd):
 199         """Create a new model to encapsulate this network service.
 200
 201         Create a new model in the Juju controller to encapsulate the
 202         charms associated with a network service.
 203
 204         You can pass either the nsd record or the id of the network
 205         service, but this method will fail without one of them.
 206         """
 207         if not self.authenticated:
 208             await self.login()
 209
 210         # Ideally, we will create a unique model per network service.
 211         # This change will require all components, i.e., LCM and SO, to use
 212         # N2VC for 100% compatibility. If we adopt unique models for the LCM,
 213         # services deployed via LCM would't be manageable via SO and vice versa
 214
 215         return self.default_model
 216
 217     async def DeployCharms(self, model_name, application_name, vnfd, charm_path, params={}, machine_spec={}, callback=None, *callback_args):
 218         """Deploy one or more charms associated with a VNF.
 219
 220         Deploy the charm(s) referenced in a VNF Descriptor.
 221
 222         You can pass either the nsd record or the id of the network
 223         service, but this method will fail without one of them.
 224
 225         :param str ns_name: The name of the network service
 226         :param str application_name: The name of the application
 227         :param dict vnfd: The name of the application
 228         :param str charm_path: The path to the Juju charm
 229         :param dict params: A dictionary of runtime parameters
 230           Examples::
 231           {
 232             'rw_mgmt_ip': '1.2.3.4',
 233             # Pass the initial-config-primitives section of the vnf or vdu
 234             'initial-config-primitives': {...}
 235           }
 236         :param dict machine_spec: A dictionary describing the machine to install to
 237           Examples::
 238           {
 239             'hostname': '1.2.3.4',
 240             'username': 'ubuntu',
 241           }
 242         :param obj callback: A callback function to receive status changes.
 243         :param tuple callback_args: A list of arguments to be passed to the callback
 244         """
 245
 246         ########################################################
 247         # Verify the path to the charm exists and is readable. #
 248         ########################################################
 249         if not os.path.exists(charm_path):
 250             self.log.debug("Charm path doesn't exist: {}".format(charm_path))
 251             self.notify_callback(model_name, application_name, "failed", callback, *callback_args)
 252             raise JujuCharmNotFound("No artifacts configured.")
 253
 254         ################################
 255         # Login to the Juju controller #
 256         ################################
 257         if not self.authenticated:
 258             self.log.debug("Authenticating with Juju")
 259             await self.login()
 260
 261         ##########################################
 262         # Get the model for this network service #
 263         ##########################################
 264         # TODO: In a point release, we will use a model per deployed network
 265         # service. In the meantime, we will always use the 'default' model.
 266         model_name = 'default'
 267         model = await self.get_model(model_name)
 268
 269         ########################################
 270         # Verify the application doesn't exist #
 271         ########################################
 272         app = await self.get_application(model, application_name)
 273         if app:
 274             raise JujuApplicationExists("Can't deploy application \"{}\" to model \"{}\" because it already exists.".format(application_name, model))
 275
 276         ############################################################
 277         # Create a monitor to watch for application status changes #
 278         ############################################################
 279         if callback:
 280             self.log.debug("Setting monitor<->callback")
 281             self.monitors[application_name] = VCAMonitor(model_name, application_name, callback, *callback_args)
 282             model.add_observer(self.monitors[application_name])
 283
 284         ########################################################
 285         # Check for specific machine placement (native charms) #
 286         ########################################################
 287         to = ""
 288         if machine_spec.keys():
 289             # TODO: This needs to be tested.
 290             # if all(k in machine_spec for k in ['hostname', 'username']):
 291             #     # Enlist the existing machine in Juju
 292             #     machine = await self.model.add_machine(spec='ssh:%@%'.format(
 293             #         specs['host'],
 294             #         specs['user'],
 295             #     ))
 296             #     to = machine.id
 297             pass
 298
 299         #######################################
 300         # Get the initial charm configuration #
 301         #######################################
 302
 303         rw_mgmt_ip = None
 304         if 'rw_mgmt_ip' in params:
 305             rw_mgmt_ip = params['rw_mgmt_ip']
 306
 307         initial_config = self._get_config_from_dict(
 308             params['initial-config-primitive'],
 309             {'<rw_mgmt_ip>': rw_mgmt_ip}
 310         )
 311
 312         self.log.debug("JujuApi: Deploying charm ({}) from {}".format(
 313             application_name,
 314             charm_path,
 315             to=to,
 316         ))
 317
 318         ########################################################
 319         # Deploy the charm and apply the initial configuration #
 320         ########################################################
 321         app = await model.deploy(
 322             # We expect charm_path to be either the path to the charm on disk
 323             # or in the format of cs:series/name
 324             charm_path,
 325             # This is the formatted, unique name for this charm
 326             application_name=application_name,
 327             # Proxy charms should use the current LTS. This will need to be
 328             # changed for native charms.
 329             series='xenial',
 330             # Apply the initial 'config' primitive during deployment
 331             config=initial_config,
 332             # TBD: Where to deploy the charm to.
 333             to=None,
 334         )
 335
 336         # #######################################
 337         # # Execute initial config primitive(s) #
 338         # #######################################
 339         primitives = {}
 340
 341         # Build a sequential list of the primitives to execute
 342         for primitive in params['initial-config-primitive']:
 343             try:
 344                 if primitive['name'] == 'config':
 345                     # This is applied when the Application is deployed
 346                     pass
 347                 else:
 348                     # TODO: We need to sort by seq, and queue the actions in order.
 349
 350                     seq = primitive['seq']
 351
 352                     primitives[seq] = {
 353                         'name': primitive['name'],
 354                         'parameters': self._map_primitive_parameters(
 355                             primitive['parameter'],
 356                             {'<rw_mgmt_ip>': rw_mgmt_ip}
 357                         ),
 358                     }
 359
 360                     for primitive in sorted(primitives):
 361                         await self.ExecutePrimitive(
 362                             model_name,
 363                             application_name,
 364                             primitives[primitive]['name'],
 365                             callback,
 366                             callback_args,
 367                             **primitives[primitive]['parameters'],
 368                         )
 369             except N2VCPrimitiveExecutionFailed as e:
 370                 self.debug.log(
 371                     "[N2VC] Exception executing primitive: {}".format(e)
 372                 )
 373                 raise
 374
 375     async def ExecutePrimitive(self, model_name, application_name, primitive, callback, *callback_args, **params):
 376         try:
 377             if not self.authenticated:
 378                 await self.login()
 379
 380             # FIXME: This is hard-coded until model-per-ns is added
 381             model_name = 'default'
 382
 383             if primitive == 'config':
 384                 # config is special, and expecting params to be a dictionary
 385                 await self.set_config(application_name, params['params'])
 386             else:
 387                 model = await self.controller.get_model(model_name)
 388                 app = await self.get_application(model, application_name)
 389                 if app:
 390                     # Run against the first (and probably only) unit in the app
 391                     unit = app.units[0]
 392                     if unit:
 393                         self.log.debug("Executing primitive {}".format(primitive))
 394                         action = await unit.run_action(primitive, **params)
 395                         # action = await action.wait()
 396                 await model.disconnect()
 397         except Exception as e:
 398             self.log.debug("Caught exception while executing primitive: {}".format(e))
 399             raise e
 400
 401     async def RemoveCharms(self, model_name, application_name, callback=None, *callback_args):
 402         try:
 403             if not self.authenticated:
 404                 await self.login()
 405
 406             model = await self.get_model(model_name)
 407             app = await self.get_application(model, application_name)
 408             if app:
 409                 self.notify_callback(model_name, application_name, "removing", callback, *callback_args)
 410                 await app.remove()
 411                 self.notify_callback(model_name, application_name, "removed", callback, *callback_args)
 412         except Exception as e:
 413             print("Caught exception: {}".format(e))
 414             self.log.debug(e)
 415             raise e
 416
 417     async def DestroyNetworkService(self, nsd):
 418         raise NotImplementedError()
 419
 420     async def GetMetrics(self, nsd, vnfd):
 421         """Get the metrics collected by the VCA."""
 422         raise NotImplementedError()
 423
 424     # Non-public methods
 425     async def add_relation(self, a, b, via=None):
 426         """
 427         Add a relation between two application endpoints.
 428
 429         :param a An application endpoint
 430         :param b An application endpoint
 431         :param via The egress subnet(s) for outbound traffic, e.g.,
 432             (192.168.0.0/16,10.0.0.0/8)
 433         """
 434         if not self.authenticated:
 435             await self.login()
 436
 437         m = await self.get_model()
 438         try:
 439             m.add_relation(a, b, via)
 440         finally:
 441             await m.disconnect()
 442
 443     async def apply_config(self, config, application):
 444         """Apply a configuration to the application."""
 445         print("JujuApi: Applying configuration to {}.".format(
 446             application
 447         ))
 448         return await self.set_config(application=application, config=config)
 449
 450     def _get_config_from_dict(self, config_primitive, values):
 451         """Transform the yang config primitive to dict.
 452
 453         Expected result:
 454
 455             config = {
 456                 'config':
 457             }
 458         """
 459         config = {}
 460         for primitive in config_primitive:
 461             if primitive['name'] == 'config':
 462                 # config = self._map_primitive_parameters()
 463                 for parameter in primitive['parameter']:
 464                     param = str(parameter['name'])
 465                     if parameter['value'] == "<rw_mgmt_ip>":
 466                         config[param] = str(values[parameter['value']])
 467                     else:
 468                         config[param] = str(parameter['value'])
 469
 470         return config
 471
 472     def _map_primitive_parameters(self, parameters, values):
 473         params = {}
 474         for parameter in parameters:
 475             param = str(parameter['name'])
 476             if parameter['value'] == "<rw_mgmt_ip>":
 477                 params[param] = str(values[parameter['value']])
 478             else:
 479                 params[param] = str(parameter['value'])
 480         return params
 481
 482     def _get_config_from_yang(self, config_primitive, values):
 483         """Transform the yang config primitive to dict."""
 484         config = {}
 485         for primitive in config_primitive.values():
 486             if primitive['name'] == 'config':
 487                 for parameter in primitive['parameter'].values():
 488                     param = str(parameter['name'])
 489                     if parameter['value'] == "<rw_mgmt_ip>":
 490                         config[param] = str(values[parameter['value']])
 491                     else:
 492                         config[param] = str(parameter['value'])
 493
 494         return config
 495
 496     def FormatApplicationName(self, *args):
 497         """
 498         Generate a Juju-compatible Application name
 499
 500         :param args tuple: Positional arguments to be used to construct the
 501         application name.
 502
 503         Limitations::
 504         - Only accepts characters a-z and non-consequitive dashes (-)
 505         - Application name should not exceed 50 characters
 506
 507         Examples::
 508
 509             FormatApplicationName("ping_pong_ns", "ping_vnf", "a")
 510         """
 511
 512         appname = ""
 513         for c in "-".join(list(args)):
 514             if c.isdigit():
 515                 c = chr(97 + int(c))
 516             elif not c.isalpha():
 517                 c = "-"
 518             appname += c
 519         return re.sub('\-+', '-', appname.lower())
 520
 521
 522     # def format_application_name(self, nsd_name, vnfr_name, member_vnf_index=0):
 523     #     """Format the name of the application
 524     #
 525     #     Limitations:
 526     #     - Only accepts characters a-z and non-consequitive dashes (-)
 527     #     - Application name should not exceed 50 characters
 528     #     """
 529     #     name = "{}-{}-{}".format(nsd_name, vnfr_name, member_vnf_index)
 530     #     new_name = ''
 531     #     for c in name:
 532     #         if c.isdigit():
 533     #             c = chr(97 + int(c))
 534     #         elif not c.isalpha():
 535     #             c = "-"
 536     #         new_name += c
 537     #     return re.sub('\-+', '-', new_name.lower())
 538
 539     def format_model_name(self, name):
 540         """Format the name of model.
 541
 542         Model names may only contain lowercase letters, digits and hyphens
 543         """
 544
 545         return name.replace('_', '-').lower()
 546
 547     async def get_application(self, model, application):
 548         """Get the deployed application."""
 549         if not self.authenticated:
 550             await self.login()
 551
 552         app = None
 553         if application and model:
 554             if model.applications:
 555                 if application in model.applications:
 556                     app = model.applications[application]
 557
 558         return app
 559
 560     async def get_model(self, model_name='default'):
 561         """Get a model from the Juju Controller.
 562
 563         Note: Model objects returned must call disconnected() before it goes
 564         out of scope."""
 565         if not self.authenticated:
 566             await self.login()
 567
 568         if model_name not in self.models:
 569             print("connecting to model {}".format(model_name))
 570             self.models[model_name] = await self.controller.get_model(model_name)
 571
 572         return self.models[model_name]
 573
 574     async def login(self):
 575         """Login to the Juju controller."""
 576
 577         if self.authenticated:
 578             return
 579
 580         self.connecting = True
 581
 582         self.log.debug("JujuApi: Logging into controller")
 583
 584         cacert = None
 585         self.controller = Controller()
 586
 587         if self.secret:
 588             self.log.debug("Connecting to controller... ws://{}:{} as {}/{}".format(self.endpoint, self.port, self.user, self.secret))
 589             await self.controller.connect(
 590                 endpoint=self.endpoint,
 591                 username=self.user,
 592                 password=self.secret,
 593                 cacert=cacert,
 594             )
 595         else:
 596             # current_controller no longer exists
 597             # self.log.debug("Connecting to current controller...")
 598             # await self.controller.connect_current()
 599             # await self.controller.connect(
 600             #     endpoint=self.endpoint,
 601             #     username=self.user,
 602             #     cacert=cacert,
 603             # )
 604             self.log.fatal("VCA credentials not configured.")
 605
 606         self.authenticated = True
 607         self.log.debug("JujuApi: Logged into controller")
 608
 609         # self.default_model = await self.controller.get_model("default")
 610
 611     async def logout(self):
 612         """Logout of the Juju controller."""
 613         if not self.authenticated:
 614             return
 615
 616         try:
 617             if self.default_model:
 618                 self.log.debug("Disconnecting model {}".format(self.default_model))
 619                 await self.default_model.disconnect()
 620                 self.default_model = None
 621
 622             for model in self.models:
 623                 await self.models[model].disconnect()
 624
 625             if self.controller:
 626                 self.log.debug("Disconnecting controller {}".format(self.controller))
 627                 await self.controller.disconnect()
 628                 # self.controller = None
 629
 630             self.authenticated = False
 631         except Exception as e:
 632             self.log.fail("Fatal error logging out of Juju Controller: {}".format(e))
 633             raise e
 634
 635
 636     # async def remove_application(self, name):
 637     #     """Remove the application."""
 638     #     if not self.authenticated:
 639     #         await self.login()
 640     #
 641     #     app = await self.get_application(name)
 642     #     if app:
 643     #         self.log.debug("JujuApi: Destroying application {}".format(
 644     #             name,
 645     #         ))
 646     #
 647     #         await app.destroy()
 648
 649     async def remove_relation(self, a, b):
 650         """
 651         Remove a relation between two application endpoints
 652
 653         :param a An application endpoint
 654         :param b An application endpoint
 655         """
 656         if not self.authenticated:
 657             await self.login()
 658
 659         m = await self.get_model()
 660         try:
 661             m.remove_relation(a, b)
 662         finally:
 663             await m.disconnect()
 664
 665     async def resolve_error(self, application=None):
 666         """Resolve units in error state."""
 667         if not self.authenticated:
 668             await self.login()
 669
 670         app = await self.get_application(self.default_model, application)
 671         if app:
 672             self.log.debug("JujuApi: Resolving errors for application {}".format(
 673                 application,
 674             ))
 675
 676             for unit in app.units:
 677                 app.resolved(retry=True)
 678
 679     async def run_action(self, application, action_name, **params):
 680         """Execute an action and return an Action object."""
 681         if not self.authenticated:
 682             await self.login()
 683         result = {
 684             'status': '',
 685             'action': {
 686                 'tag': None,
 687                 'results': None,
 688             }
 689         }
 690         app = await self.get_application(self.default_model, application)
 691         if app:
 692             # We currently only have one unit per application
 693             # so use the first unit available.
 694             unit = app.units[0]
 695
 696             self.log.debug("JujuApi: Running Action {} against Application {}".format(
 697                 action_name,
 698                 application,
 699             ))
 700
 701             action = await unit.run_action(action_name, **params)
 702
 703             # Wait for the action to complete
 704             await action.wait()
 705
 706             result['status'] = action.status
 707             result['action']['tag'] = action.data['id']
 708             result['action']['results'] = action.results
 709
 710         return result
 711
 712     async def set_config(self, application, config):
 713         """Apply a configuration to the application."""
 714         if not self.authenticated:
 715             await self.login()
 716
 717         app = await self.get_application(self.default_model, application)
 718         if app:
 719             self.log.debug("JujuApi: Setting config for Application {}".format(
 720                 application,
 721             ))
 722             await app.set_config(config)
 723
 724             # Verify the config is set
 725             newconf = await app.get_config()
 726             for key in config:
 727                 if config[key] != newconf[key]['value']:
 728                     self.log.debug("JujuApi: Config not set! Key {} Value {} doesn't match {}".format(key, config[key], newconf[key]))
 729
 730     async def set_parameter(self, parameter, value, application=None):
 731         """Set a config parameter for a service."""
 732         if not self.authenticated:
 733             await self.login()
 734
 735         self.log.debug("JujuApi: Setting {}={} for Application {}".format(
 736             parameter,
 737             value,
 738             application,
 739         ))
 740         return await self.apply_config(
 741             {parameter: value},
 742             application=application,
 743         )
 744
 745     async def wait_for_application(self, name, timeout=300):
 746         """Wait for an application to become active."""
 747         if not self.authenticated:
 748             await self.login()
 749
 750         app = await self.get_application(self.default_model, name)
 751         if app:
 752             self.log.debug(
 753                 "JujuApi: Waiting {} seconds for Application {}".format(
 754                     timeout,
 755                     name,
 756                 )
 757             )
 758
 759             await self.default_model.block_until(
 760                 lambda: all(
 761                     unit.agent_status == 'idle'
 762                     and unit.workload_status
 763                     in ['active', 'unknown'] for unit in app.units
 764                 ),
 765                 timeout=timeout
 766             )