rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/RiftCA.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 asyncio
  18 import concurrent.futures
  19 import re
  20 import tempfile
  21 import yaml
  22 import os
  23
  24 from gi.repository import (
  25     RwDts as rwdts,
  26 )
  27
  28 from . import riftcm_config_plugin
  29 from . import rwconman_events as Events
  30
  31 class RiftCAConfigPlugin(riftcm_config_plugin.RiftCMConfigPluginBase):
  32     """
  33         Implementation of the riftcm_config_plugin.RiftCMConfigPluginBase
  34     """
  35     def __init__(self, dts, log, loop, account):
  36         riftcm_config_plugin.RiftCMConfigPluginBase.__init__(self, dts, log, loop, account)
  37         self._name = account.name
  38         self._type = riftcm_config_plugin.DEFAULT_CAP_TYPE
  39         self._rift_install_dir = os.environ['RIFT_INSTALL']
  40         self._rift_artif_dir = os.environ['RIFT_ARTIFACTS']
  41         self._rift_vnfs = {}
  42         self._tasks = {}
  43
  44         # Instantiate events that will handle RiftCA configuration requests
  45         self._events = Events.ConfigManagerEvents(dts, log, loop, self)
  46
  47     @property
  48     def name(self):
  49         return self._name
  50
  51     @property
  52     def agent_type(self):
  53         return self._type
  54
  55     @asyncio.coroutine
  56     def notify_create_vlr(self, agent_nsr, agent_vnfr, vld, vlr):
  57         """
  58         Notification of create VL record
  59         """
  60         pass
  61
  62     @asyncio.coroutine
  63     def is_vnf_configurable(self, agent_vnfr):
  64         '''
  65         This needs to be part of abstract class
  66         '''
  67         loop_count = 10
  68         while loop_count:
  69             loop_count -= 1
  70             # Set this VNF's configurability status (need method to check)
  71             yield from asyncio.sleep(2, loop=self._loop)
  72
  73     def riftca_log(self, name, level, log_str, *args):
  74         getattr(self._log, level)('RiftCA:({}) {}'.format(name, log_str), *args)
  75
  76     @asyncio.coroutine
  77     def notify_create_vnfr(self, agent_nsr, agent_vnfr):
  78         """
  79         Notification of create Network VNF record
  80         """
  81         # Deploy the charm if specified for the vnf
  82         self._log.debug("Rift config agent: create vnfr nsr={}  vnfr={}"
  83                         .format(agent_nsr.name, agent_vnfr.name))
  84         try:
  85             self._loop.create_task(self.is_vnf_configurable(agent_vnfr))
  86         except Exception as e:
  87             self._log.debug("Rift config agent: vnf_configuration error for VNF:%s/%s: %s",
  88                             agent_nsr.name, agent_vnfr.name, str(e))
  89             return False
  90
  91         return True
  92
  93     @asyncio.coroutine
  94     def notify_instantiate_vnfr(self, agent_nsr, agent_vnfr):
  95         """
  96         Notification of Instantiate NSR with the passed nsr id
  97         """
  98         pass
  99
 100     @asyncio.coroutine
 101     def notify_instantiate_vlr(self, agent_nsr, agent_vnfr, vlr):
 102         """
 103         Notification of Instantiate NSR with the passed nsr id
 104         """
 105         pass
 106
 107     @asyncio.coroutine
 108     def notify_terminate_vnfr(self, agent_nsr, agent_vnfr):
 109         """
 110         Notification of Terminate the network service
 111         """
 112
 113     @asyncio.coroutine
 114     def notify_terminate_vlr(self, agent_nsr, agent_vnfr, vlr):
 115         """
 116         Notification of Terminate the virtual link
 117         """
 118         pass
 119
 120     @asyncio.coroutine
 121     def vnf_config_primitive(self, agent_nsr, agent_vnfr, primitive, output):
 122         '''
 123         primitives support by RiftCA
 124         '''
 125         pass
 126
 127     @asyncio.coroutine
 128     def apply_config(self, config, nsr, vnfr, rpc_ip):
 129         """ Notification on configuration of an NSR """
 130         pass
 131
 132     @asyncio.coroutine
 133     def apply_ns_config(self, agent_nsr, agent_vnfrs, rpc_ip):
 134         """Hook: Runs the user defined script. Feeds all the necessary data
 135         for the script thro' yaml file.
 136
 137         Args:
 138             rpc_ip (YangInput_Nsr_ExecNsConfigPrimitive): The input data.
 139             nsr (NetworkServiceRecord): Description
 140             vnfrs (dict): VNFR ID => VirtualNetworkFunctionRecord
 141         """
 142
 143         def xlate(tag, tags):
 144             # TBD
 145             if tag is None or tags is None:
 146                 return tag
 147             val = tag
 148             if re.search('<.*>', tag):
 149                 try:
 150                     if tag == '<rw_mgmt_ip>':
 151                         val = tags['rw_mgmt_ip']
 152                 except KeyError as e:
 153                     self._log.info("RiftCA: Did not get a value for tag %s, e=%s",
 154                                    tag, e)
 155             return val
 156
 157         def get_meta(agent_nsr, agent_vnfrs):
 158             unit_names, initial_params, vnfr_index_map, vnfr_data_map = {}, {}, {}, {}
 159
 160             for vnfr_id in agent_nsr.vnfr_ids:
 161                 vnfr = agent_vnfrs[vnfr_id]
 162
 163                 # index->vnfr ref
 164                 vnfr_index_map[vnfr.member_vnf_index] = vnfr_id
 165                 vnfr_data_dict = dict()
 166                 if 'mgmt_interface' in vnfr.vnfr:
 167                     vnfr_data_dict['mgmt_interface'] = vnfr.vnfr['mgmt_interface']
 168
 169                 vnfr_data_dict['connection_point'] = []
 170                 if 'connection_point' in vnfr.vnfr:
 171                     for cp in vnfr.vnfr['connection_point']:
 172                         cp_dict = dict()
 173                         cp_dict['name'] = cp['name']
 174                         cp_dict['ip_address'] = cp['ip_address']
 175                         vnfr_data_dict['connection_point'].append(cp_dict)
 176
 177                 vnfr_data_dict['vdur'] = []
 178                 vdu_data = [(vdu['name'], vdu['management_ip'], vdu['vm_management_ip'], vdu['id'])
 179                         for vdu in vnfr.vnfr['vdur']]
 180
 181                 for data in vdu_data:
 182                     data = dict(zip(['name', 'management_ip', 'vm_management_ip', 'id'] , data))
 183                     vnfr_data_dict['vdur'].append(data)
 184
 185                 vnfr_data_map[vnfr.member_vnf_index] = vnfr_data_dict
 186                 # Unit name
 187                 unit_names[vnfr_id] = vnfr.name
 188                 # Flatten the data for simplicity
 189                 param_data = {}
 190                 if 'initial_config_primitive' in vnfr.vnf_configuration:
 191                     for primitive in vnfr.vnf_configuration['initial_config_primitive']:
 192                         for parameter in primitive.parameter:
 193                             value = xlate(parameter.value, vnfr.tags)
 194                             param_data[parameter.name] = value
 195
 196                 initial_params[vnfr_id] = param_data
 197
 198
 199             return unit_names, initial_params, vnfr_index_map, vnfr_data_map
 200
 201         unit_names, init_data, vnfr_index_map, vnfr_data_map = get_meta(agent_nsr, agent_vnfrs)
 202         # The data consists of 4 sections
 203         # 1. Account data
 204         # 2. The input passed.
 205         # 3. Unit names (keyed by vnfr ID).
 206         # 4. Initial config data (keyed by vnfr ID).
 207         data = dict()
 208         data['config_agent'] = dict(
 209                 name=self._name,
 210                 )
 211         data["rpc_ip"] = rpc_ip.as_dict()
 212         data["unit_names"] = unit_names
 213         data["init_config"] = init_data
 214         data["vnfr_index_map"] = vnfr_index_map
 215         data["vnfr_data_map"] = vnfr_data_map
 216
 217         tmp_file = None
 218         with tempfile.NamedTemporaryFile(delete=False) as tmp_file:
 219             tmp_file.write(yaml.dump(data, default_flow_style=True)
 220                     .encode("UTF-8"))
 221
 222         # Get the full path to the script
 223         script = ''
 224         if rpc_ip.user_defined_script[0] == '/':
 225             # The script has full path, use as is
 226             script = rpc_ip.user_defined_script
 227         else:
 228             script = os.path.join(self._rift_artif_dir, 'launchpad/libs', agent_nsr.nsd_id, 'scripts',
 229                                   rpc_ip.user_defined_script)
 230             self._log.debug("Rift config agent: Checking for script in %s", script)
 231             if not os.path.exists(script):
 232                 self._log.debug("Rift config agent: Did not find scipt %s", script)
 233                 script = os.path.join(self._rift_install_dir, 'usr/bin', rpc_ip.user_defined_script)
 234
 235         cmd = "{} {}".format(script, tmp_file.name)
 236         self._log.debug("Rift config agent: Running the CMD: {}".format(cmd))
 237
 238         coro = asyncio.create_subprocess_shell(cmd, loop=self._loop,
 239                                                stderr=asyncio.subprocess.PIPE)
 240         process = yield from coro
 241         err = yield from process.stderr.read()
 242         task = self._loop.create_task(process.wait())
 243
 244         return task, err
 245
 246     @asyncio.coroutine
 247     def apply_initial_config(self, agent_nsr, agent_vnfr):
 248         """
 249         Apply the initial configuration
 250         """
 251         rc = False
 252         self._log.debug("Rift config agent: Apply initial config to VNF:%s/%s",
 253                         agent_nsr.name, agent_vnfr.name)
 254         try:
 255             if agent_vnfr.id in self._rift_vnfs.keys():
 256                 # Check if VNF instance is configurable (TBD - future)
 257                 ### Remove this once is_vnf_configurable() is implemented
 258                 agent_vnfr.set_to_configurable()
 259                 if agent_vnfr.is_configurable:
 260                     # apply initial config for the vnfr
 261                     rc = yield from self._events.apply_vnf_config(agent_vnfr.vnf_cfg)
 262                 else:
 263                     self._log.info("Rift config agent: VNF:%s/%s is not configurable yet!",
 264                                    agent_nsr.name, agent_vnfr.name)
 265         except Exception as e:
 266             self._log.error("Rift config agent: Error on initial configuration to VNF:{}/{}, e {}"
 267                             .format(agent_nsr.name, agent_vnfr.name, str(e)))
 268
 269             self._log.exception(e)
 270             return rc
 271
 272         return rc
 273
 274     def is_vnfr_managed(self, vnfr_id):
 275         try:
 276             if vnfr_id in self._rift_vnfs:
 277                 return True
 278         except Exception as e:
 279             self._log.debug("Rift config agent: Is VNFR {} managed: {}".
 280                             format(vnfr_id, e))
 281         return False
 282
 283     def add_vnfr_managed(self, agent_vnfr):
 284         if agent_vnfr.id not in self._rift_vnfs.keys():
 285             self._log.info("Rift config agent: add vnfr={}/{}".format(agent_vnfr.name, agent_vnfr.id))
 286             self._rift_vnfs[agent_vnfr.id] = agent_vnfr
 287
 288     @asyncio.coroutine
 289     def get_config_status(self, agent_nsr, agent_vnfr):
 290             if agent_vnfr.id in self._rift_vnfs.keys():
 291                 return 'configured'
 292             return 'unknown'
 293
 294
 295     def get_action_status(self, execution_id):
 296         ''' Get the action status for an execution ID
 297             *** Make sure this is NOT a asyncio coroutine function ***
 298         '''
 299         return None