rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/riftcm_config_plugin.py

   1 #
   2 #   Copyright 2016 RIFT.IO Inc
   3 #
   4 #   Licensed under the Apache License, Version 2.0 (the "License");
   5 #   you may not use this file except in compliance with the License.
   6 #   You may obtain a copy of the License at
   7 #
   8 #       http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #   Unless required by applicable law or agreed to in writing, software
  11 #   distributed under the License is distributed on an "AS IS" BASIS,
  12 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 #   See the License for the specific language governing permissions and
  14 #   limitations under the License.
  15 #
  16
  17 import abc
  18 import asyncio
  19 import gi
  20 import os
  21 import os
  22 import stat
  23 import tempfile
  24 import yaml
  25
  26 from urllib.parse import urlparse
  27
  28 gi.require_version('RwDts', '1.0')
  29 from gi.repository import (
  30     RwDts as rwdts,
  31 )
  32 gi.require_version('RwKeyspec', '1.0')
  33 from gi.repository.RwKeyspec import quoted_key
  34
  35 # Default config agent plugin type
  36 DEFAULT_CAP_TYPE = "riftca"
  37
  38
  39 class XPaths(object):
  40     @staticmethod
  41     def nsr_opdata(k=None):
  42         return ("D,/nsr:ns-instance-opdata/nsr:nsr" +
  43                 ("[nsr:ns-instance-config-ref={}]".format(quoted_key(k)) if k is not None else ""))
  44
  45     @staticmethod
  46     def nsd_msg(k=None):
  47         return ("C,/nsd:nsd-catalog/nsd:nsd" +
  48                 "[nsd:id={}]".format(quoted_key(k)) if k is not None else "")
  49
  50     @staticmethod
  51     def vnfr_opdata(k=None):
  52         return ("D,/vnfr:vnfr-catalog/vnfr:vnfr" +
  53                 ("[vnfr:id={}]".format(quoted_key(k)) if k is not None else ""))
  54
  55     @staticmethod
  56     def nsr_config(k=None):
  57         return ("C,/nsr:ns-instance-config/nsr:nsr[nsr:id={}]".format(quoted_key(k)) if k is not None else "")
  58
  59
  60 class RiftCMnsr(object):
  61     '''
  62     Agent class for NSR
  63     created for Agents to use objects from NSR
  64     '''
  65     def __init__(self, nsr_dict, cfg, project):
  66         self._nsr = nsr_dict
  67         self._cfg = cfg
  68         self._project = project
  69         self._vnfrs = []
  70         self._vnfrs_msg = []
  71         self._vnfr_ids = {}
  72         self._job_id = 0
  73
  74     @property
  75     def name(self):
  76         return self._nsr['name_ref']
  77
  78     @property
  79     def nsd_name(self):
  80         return self._nsr['nsd_name_ref']
  81
  82     @property
  83     def nsd_id(self):
  84         return self._nsr['nsd_ref']
  85
  86     @property
  87     def id(self):
  88         return self._nsr['ns_instance_config_ref']
  89
  90     @property
  91     def nsr_dict(self):
  92         return self._nsr
  93
  94     @property
  95     def nsr_cfg_msg(self):
  96         return self._cfg
  97
  98     @property
  99     def nsd(self):
 100         return self._cfg.nsd
 101
 102     @property
 103     def job_id(self):
 104         ''' Get a new job id for config primitive'''
 105         self._job_id += 1
 106         return self._job_id
 107
 108     @property
 109     def vnfrs(self):
 110         return self._vnfrs
 111
 112     @property
 113     def member_vnf_index(self):
 114         return self._vnfr['member_vnf_index_ref']
 115
 116     def add_vnfr(self, vnfr, vnfr_msg):
 117         if vnfr['id'] in self._vnfr_ids.keys():
 118             agent_vnfr = self._vnfr_ids[vnfr['id']]
 119         else:
 120             agent_vnfr = RiftCMvnfr(self.name, vnfr, vnfr_msg, self._project)
 121             self._vnfrs.append(agent_vnfr)
 122             self._vnfrs_msg.append(vnfr_msg)
 123             self._vnfr_ids[agent_vnfr.id] = agent_vnfr
 124         return agent_vnfr
 125
 126     @property
 127     def vnfr_ids(self):
 128         return self._vnfr_ids
 129
 130     def get_member_vnfr(self, member_index):
 131         for vnfr in self._vnfrs:
 132             if vnfr.member_vnf_index == member_index:
 133                 return vnfr
 134
 135
 136 class RiftCMvnfr(object):
 137     '''
 138     Agent base class for VNFR processing
 139     '''
 140     def __init__(self, nsr_name, vnfr_dict, vnfr_msg, project):
 141         self._vnfr = vnfr_dict
 142         self._vnfr_msg = vnfr_msg
 143         self._vnfd_msg = vnfr_msg.vnfd
 144         self._nsr_name = nsr_name
 145         self._configurable = False
 146         self._project = project
 147         self._error = False
 148
 149     @property
 150     def nsr_name(self):
 151         return self._nsr_name
 152
 153     @property
 154     def vnfr(self):
 155         return self._vnfr
 156
 157     @property
 158     def vnfr_msg(self):
 159         return self._vnfr_msg
 160
 161     @property
 162     def vnfd(self):
 163         return self._vnfd_msg
 164
 165     @property
 166     def name(self):
 167         return self._vnfr['name']
 168
 169     @property
 170     def tags(self):
 171         try:
 172             return self._vnfr['tags']
 173         except KeyError:
 174             return None
 175
 176     @property
 177     def id(self):
 178         return self._vnfr['id']
 179
 180     @property
 181     def member_vnf_index(self):
 182         return self._vnfr['member_vnf_index_ref']
 183
 184     @property
 185     def vnf_configuration(self):
 186         return self._vnfr['vnf_configuration']
 187
 188     @property
 189     def xpath(self):
 190         """ VNFR xpath """
 191         return self._project.add_project("D,/vnfr:vnfr-catalog/vnfr:vnfr[vnfr:id={}]".
 192                                          format(quoted_key(self.id)))
 193
 194     def set_to_configurable(self):
 195         self._configurable = True
 196
 197     @property
 198     def is_configurable(self):
 199         return self._configurable
 200
 201     @property
 202     def vnf_cfg(self):
 203         return self._vnfr['vnf_cfg']
 204
 205     @property
 206     def error(self):
 207         return self._error
 208
 209     @error.setter
 210     def error(self, value):
 211         self._error = value
 212
 213
 214 class RiftCMConfigPluginBase(object):
 215     """
 216         Abstract base class for the NSM Configuration agent plugin.
 217         There will be single instance of this plugin for each plugin type.
 218     """
 219
 220     def __init__(self, dts, log, loop, project, config_agent):
 221         self._dts = dts
 222         self._log = log
 223         self._loop = loop
 224         self._project = project
 225         self._config_agent = config_agent
 226
 227     @property
 228     def agent_type(self):
 229         raise NotImplementedError
 230
 231     @property
 232     def name(self):
 233         raise NotImplementedError
 234
 235     @property
 236     def agent_data(self):
 237         raise NotImplementedError
 238
 239     @property
 240     def dts(self):
 241         return self._dts
 242
 243     @property
 244     def log(self):
 245         return self._log
 246
 247     @property
 248     def loop(self):
 249         return self._loop
 250
 251     @property
 252     def nsm(self):
 253         return self._nsm
 254
 255
 256     def vnfr(self, vnfr_id):
 257         raise NotImplementedError
 258
 259     @abc.abstractmethod
 260     def get_Service_name(self):
 261         """ Get the service name specific to the plugin """
 262         pass
 263
 264     @abc.abstractmethod
 265     @asyncio.coroutine
 266     def apply_config(self, agent_nsr, agent_vnfr, config, rpc_ip):
 267         """ Notification on configuration of an NSR """
 268         pass
 269
 270     @abc.abstractmethod
 271     @asyncio.coroutine
 272     def apply_ns_config(self, agent_nsr, agent_vnfrs, config, rpc_ip):
 273         """ Notification on configuration of an NSR """
 274         pass
 275
 276     @abc.abstractmethod
 277     @asyncio.coroutine
 278     def notify_create_vlr(self, agent_nsr, vld):
 279         """ Notification on creation of an VL """
 280         pass
 281
 282     @abc.abstractmethod
 283     @asyncio.coroutine
 284     def notify_create_vnfr(self, agent_nsr, agent_vnfr):
 285         """ Notification on creation of an VNFR """
 286         pass
 287
 288     @abc.abstractmethod
 289     @asyncio.coroutine
 290     def notify_instantiate_vnfr(self, agent_nsr, agent_vnfr):
 291         """ Notify instantiation of the virtual network function """
 292         pass
 293
 294     @abc.abstractmethod
 295     @asyncio.coroutine
 296     def notify_instantiate_vlr(self, agent_nsr, vl):
 297         """ Notify instantiate of the virtual link"""
 298         pass
 299
 300     @abc.abstractmethod
 301     @asyncio.coroutine
 302     def notify_terminate_vnfr(self, agent_nsr, agent_vnfr):
 303         """Notify termination of the VNF """
 304         pass
 305
 306     @abc.abstractmethod
 307     @asyncio.coroutine
 308     def notify_terminate_vlr(self, agent_nsr, vlr):
 309         """Notify termination of the Virtual Link Record"""
 310         pass
 311
 312     @abc.abstractmethod
 313     @asyncio.coroutine
 314     def apply_initial_config(self, vnfr_id, vnf):
 315         """Apply initial configuration"""
 316         pass
 317
 318     @abc.abstractmethod
 319     @asyncio.coroutine
 320     def get_config_status(self, vnfr_id):
 321         """Get the status for the VNF"""
 322         pass
 323
 324     @abc.abstractmethod
 325     def get_action_status(self, execution_id):
 326         """Get the action exection status"""
 327         pass
 328
 329     @abc.abstractmethod
 330     @asyncio.coroutine
 331     def vnf_config_primitive(self, nsr_id, vnfr_id, primitive, output):
 332         """Apply config primitive on a VNF"""
 333         pass
 334
 335     @abc.abstractmethod
 336     def is_vnfr_managed(self, vnfr_id):
 337         """ Check if VNR is managed by config agent """
 338         pass
 339
 340     @abc.abstractmethod
 341     def add_vnfr_managed(self, agent_vnfr):
 342         """ Add VNR to be managed by this config agent """
 343         pass
 344
 345     def get_service_status(self, vnfr_id):
 346         """Get the status of the service"""
 347         return None
 348
 349     # Helper functions
 350
 351     def convert_value(self, value, type_='STRING'):
 352         if type_ == 'STRING':
 353             if value.startswith('file://'):
 354                 p = urlparse(value)
 355                 with open(p[2], 'r') as f:
 356                     val = f.read()
 357                     return(val)
 358             return str(value)
 359
 360         if type_ == 'INTEGER':
 361             return int(value)
 362
 363         if type_ == 'BOOLEAN':
 364             return (value == 1) or (value.lower() == 'true')
 365
 366         return value
 367
 368     @asyncio.coroutine
 369     def _read_dts(self, path, do_trace=False):
 370         xpath = self._project.add_project(path)
 371         self._log.debug("_read_dts path = %s", xpath)
 372         flags = rwdts.XactFlag.MERGE
 373         res_iter = yield from self._dts.query_read(
 374                 xpath, flags=flags
 375                 )
 376
 377         results = []
 378         try:
 379             for i in res_iter:
 380                 result = yield from i
 381                 if result is not None:
 382                     results.append(result.result)
 383         except:
 384             pass
 385
 386         return results
 387
 388
 389     @asyncio.coroutine
 390     def get_xpath(self, xpath):
 391         self._log.debug("Attempting to get xpath: {}".format(xpath))
 392         resp = yield from self._read_dts(xpath, False)
 393         if len(resp) > 0:
 394             self._log.debug("Got DTS resp: {}".format(resp[0]))
 395             return resp[0]
 396         return None
 397
 398     @asyncio.coroutine
 399     def get_nsr(self, id):
 400         self._log.debug("Attempting to get NSR: %s", id)
 401         nsrl = yield from self._read_dts(XPaths.nsr_opdata(id), False)
 402         nsr = None
 403         if len(nsrl) > 0:
 404             nsr =  nsrl[0].as_dict()
 405         return nsr
 406
 407     @asyncio.coroutine
 408     def get_nsr_config(self, id):
 409         self._log.debug("Attempting to get config NSR: %s", id)
 410         nsrl = yield from self._read_dts(XPaths.nsr_config(id), False)
 411         nsr = None
 412         if len(nsrl) > 0:
 413             nsr =  nsrl[0]
 414         return nsr
 415
 416     @asyncio.coroutine
 417     def get_vnfr(self, id):
 418         self._log.debug("Attempting to get VNFR: %s", id)
 419         vnfrl = yield from self._read_dts(XPaths.vnfr_opdata(id), do_trace=False)
 420         vnfr_msg = None
 421         if len(vnfrl) > 0:
 422             vnfr_msg = vnfrl[0]
 423         return vnfr_msg
 424
 425     @asyncio.coroutine
 426     def exec_script(self, script, data):
 427         """Execute a shell script with the data as yaml input file"""
 428         self._log.debug("Execute script {} with data {}".
 429                         format(script, data))
 430
 431         #Make the script executable if it is not.
 432         perm = os.stat(script).st_mode
 433         if not (perm & stat.S_IXUSR):
 434             self._log.warning("script {} without execute permission: {}".
 435                                format(script, perm))
 436             os.chmod(script, perm | stat.S_IXUSR)
 437
 438         tmp_file = None
 439         with tempfile.NamedTemporaryFile(delete=False) as tmp_file:
 440             tmp_file.write(yaml.dump(data, default_flow_style=True)
 441                     .encode("UTF-8"))
 442
 443         cmd = "{} {}".format(script, tmp_file.name)
 444         self._log.debug("Running the CMD: {}".format(cmd))
 445
 446         try:
 447             proc = yield from asyncio.create_subprocess_shell(
 448                 cmd,
 449                 stdout=asyncio.subprocess.PIPE,
 450                 stderr=asyncio.subprocess.PIPE)
 451             rc = yield from proc.wait()
 452             script_out, script_err = yield from proc.communicate()
 453
 454         except Exception as e:
 455             msg = "Script {} caused exception: {}". \
 456                   format(script, e)
 457             self._log.exception(msg)
 458             rc = 1
 459             script_err = msg
 460             script_out = ''
 461
 462         finally:
 463             # Remove the tempfile created
 464             try:
 465                 if rc == 0:
 466                     os.remove(tmp_file.name)
 467             except OSError as e:
 468                 self._log.info("Error removing tempfile {}: {}".
 469                                 format(tmp_file.name, e))
 470
 471         if rc != 0:
 472             if not os.path.exists(script) :
 473                 self._log.error("Script {} not found: ".format(script))
 474             else:
 475                 self._log.error("Script {}: rc={}\nStdOut:{}\nStdErr:{} \nPermissions on script: {}".
 476                                 format(script, rc, script_out, script_err, stat.filemode(os.stat(script).st_mode)))
 477
 478         return rc, script_err
 479
 480     @asyncio.coroutine
 481     def invoke(self, method, *args):
 482         try:
 483             rc = None
 484             self._log.debug("Config agent plugin: method {} with args {}: {}".
 485                             format(method, args, self))
 486
 487             # TBD - Do a better way than string compare to find invoke the method
 488             if method == 'notify_create_nsr':
 489                 rc = yield from self.notify_create_nsr(args[0], args[1])
 490             elif method == 'notify_create_vlr':
 491                 rc = yield from self.notify_create_vlr(args[0], args[1], args[2])
 492             elif method == 'notify_create_vnfr':
 493                 rc = yield from self.notify_create_vnfr(args[0], args[1])
 494             elif method == 'notify_instantiate_nsr':
 495                 rc = yield from self.notify_instantiate_nsr(args[0])
 496             elif method == 'notify_instantiate_vnfr':
 497                 rc = yield from self.notify_instantiate_vnfr(args[0], args[1])
 498             elif method == 'notify_instantiate_vlr':
 499                 rc = yield from self.notify_instantiate_vlr(args[0], args[1])
 500             elif method == 'notify_nsr_active':
 501                 rc = yield from self.notify_nsr_active(args[0], args[1])
 502             elif method == 'notify_terminate_nsr':
 503                 rc = yield from self.notify_terminate_nsr(args[0])
 504             elif method == 'notify_terminate_vnfr':
 505                 rc = yield from self.notify_terminate_vnfr(args[0], args[1])
 506             elif method == 'notify_terminate_vlr':
 507                 rc = yield from self.notify_terminate_vlr(args[0], args[1])
 508             elif method == 'apply_initial_config':
 509                 rc = yield from self.apply_initial_config(args[0], args[1])
 510             elif method == 'apply_config':
 511                 rc = yield from self.apply_config(args[0], args[1], args[2])
 512             elif method == 'get_config_status':
 513                 rc = yield from self.get_config_status(args[0], args[1])
 514             else:
 515                 self._log.error("Unknown method %s invoked on config agent plugin",
 516                                 method)
 517         except Exception as e:
 518             self._log.exception("Caught exception while invoking method: %s, "
 519                                 "Exception: %s", method, str(e))
 520             raise e
 521
 522         return rc