Implement scaling RPCs using REST rather than netconf
[osm/SO.git] / rwlaunchpad / plugins / rwnsm / rift / tasklets / rwnsmtasklet / rwnsmtasklet.py
index 718d2b1..85b31a5 100755 (executable)
@@ -51,7 +51,7 @@ from gi.repository import (
     VnfrYang,
     RwVnfrYang,
     RwNsmYang,
-    RwsdnYang,
+    RwsdnalYang,
     RwDts as rwdts,
     RwTypes,
     ProtobufC,
@@ -220,7 +220,6 @@ class VnffgRecord(object):
 
         if self._vnffgr_state == VnffgRecordState.INIT:
             vnffgr_dict = {"id": self._vnffgr_id,
-                           "nsd_id": self._nsr.nsd_id,
                            "vnffgd_id_ref": self._vnffgd_msg.id,
                            "vnffgd_name_ref": self._vnffgd_msg.name,
                            "sdn_account": self._sdn_account_name,
@@ -229,7 +228,6 @@ class VnffgRecord(object):
             vnffgr = NsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr.from_dict(vnffgr_dict)
         elif self._vnffgr_state == VnffgRecordState.TERMINATED:
             vnffgr_dict = {"id": self._vnffgr_id,
-                           "nsd_id": self._nsr.nsd_id,
                            "vnffgd_id_ref": self._vnffgd_msg.id,
                            "vnffgd_name_ref": self._vnffgd_msg.name,
                            "sdn_account": self._sdn_account_name,
@@ -243,7 +241,6 @@ class VnffgRecord(object):
                 self._log.exception("Fetching VNFFGR for VNFFG with id %s failed", self._vnffgr_id)
                 self._vnffgr_state = VnffgRecordState.FAILED
                 vnffgr_dict = {"id": self._vnffgr_id,
-                               "nsd_id": self._nsr.nsd_id,
                                "vnffgd_id_ref": self._vnffgd_msg.id,
                                "vnffgd_name_ref": self._vnffgd_msg.name,
                                "sdn_account": self._sdn_account_name,
@@ -257,7 +254,6 @@ class VnffgRecord(object):
     def vnffgr_create_msg(self):
         """ Virtual Link Record message for Creating VLR in VNS """
         vnffgr_dict = {"id": self._vnffgr_id,
-                       "nsd_id": self._nsr.nsd_id,
                        "vnffgd_id_ref": self._vnffgd_msg.id,
                        "vnffgd_name_ref": self._vnffgd_msg.name,
                        "sdn_account": self._sdn_account_name,
@@ -387,7 +383,7 @@ class VnffgRecord(object):
                     vnfr = yield from self._nsr.fetch_vnfr(nsr_vnfr.xpath)
                     self._log.debug("Received VNFR is %s", vnfr)
 
-                sff =  RwsdnYang.VNFFGSff()
+                sff =  RwsdnalYang.VNFFGSff()
                 sff_list[nsr_vnfr.vnfd.id] = sff
                 sff.name = nsr_vnfr.name
                 sff.function_type = nsr_vnfr.vnfd.service_function_chain
@@ -796,6 +792,7 @@ class VirtualNetworkFunctionRecord(object):
         self._group_instance_id = group_instance_id
         self._placement_groups = placement_groups
         self._config_status = NsrYang.ConfigStates.INIT
+        self._create_time = int(time.time())
 
         self._prev_state = VnfRecordState.INIT
         self._state = VnfRecordState.INIT
@@ -960,8 +957,8 @@ class VirtualNetworkFunctionRecord(object):
         vnfr_dict.update(vnfd_copy_dict)
 
         vnfr = RwVnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr.from_dict(vnfr_dict)
-
-        vnfr.vnfd = VnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr_Vnfd.from_dict(self.vnfd.as_dict())
+        vnfr.vnfd = VnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr_Vnfd.from_dict(self.vnfd.as_dict(),
+                                                                          ignore_missing_keys=True)
         vnfr.member_vnf_index_ref = self.member_vnf_index
         vnfr.vnf_configuration.from_dict(self._vnfd.vnf_configuration.as_dict())
 
@@ -1075,6 +1072,9 @@ class VirtualNetworkFunctionRecord(object):
             cpr = VnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr_ConnectionPoint()
             cpr.name = conn_p.name
             cpr.type_yang = conn_p.type_yang
+            if conn_p.has_field('port_security_enabled'):
+              cpr.port_security_enabled = conn_p.port_security_enabled
+
             vlr_ref = find_vlr_for_cp(conn_p)
             if vlr_ref is None:
                 msg = "Failed to find VLR for cp = %s" % conn_p.name
@@ -1266,6 +1266,7 @@ class NetworkServiceRecord(object):
         self._is_active = False
         self._vl_phase_completed = False
         self._vnf_phase_completed = False
+        self.vlr_uptime_tasks = {}
 
 
         # Initalise the state to init
@@ -1295,6 +1296,7 @@ class NetworkServiceRecord(object):
             self._vnf_phase_completed = True
 
         self._op_status.set_state(state)
+        self._nsm_plugin.set_state(self.id, state)
 
     @property
     def id(self):
@@ -1457,17 +1459,20 @@ class NetworkServiceRecord(object):
         for vlr in self._vlrs:
             yield from self.nsm_plugin.instantiate_vl(self, vlr)
             vlr.state = VlRecordState.ACTIVE
-            self._loop.create_task(self.vlr_uptime_update(vlr))
+            self.vlr_uptime_tasks[vlr.id] = self._loop.create_task(self.vlr_uptime_update(vlr))
 
 
     def vlr_uptime_update(self, vlr):
-        vlr_ = RwVlrYang.YangData_Vlr_VlrCatalog_Vlr.from_dict({'id': vlr.id})
-        while True:
-            if vlr.state not in [VlRecordState.INIT, VlRecordState.INSTANTIATION_PENDING, VlRecordState.ACTIVE]:
-                return
-            vlr_.uptime = int(time.time()) - vlr._create_time
-            yield from self._vlr_handler.update(None, VirtualLinkRecord.vlr_xpath(vlr), vlr_)
-            yield from asyncio.sleep(2, loop=self._loop)
+        try:
+
+            vlr_ = RwVlrYang.YangData_Vlr_VlrCatalog_Vlr.from_dict({'id': vlr.id})
+            while True:
+                vlr_.uptime = int(time.time()) - vlr._create_time
+                yield from self._vlr_handler.update(None, VirtualLinkRecord.vlr_xpath(vlr), vlr_)
+                yield from asyncio.sleep(2, loop=self._loop)
+        except asyncio.CancelledError:
+            self._log.debug("Received cancellation request for vlr_uptime_update task")
+            yield from self._vlr_handler.delete(None, VirtualLinkRecord.vlr_xpath(vlr))
 
 
     @asyncio.coroutine
@@ -2006,7 +2011,7 @@ class NetworkServiceRecord(object):
                    (member_vnfd.member_vnf_index_ref == const_vnfd.member_vnf_index):
                     group_info = self.resolve_placement_group_cloud_construct(group)
                     if group_info is None:
-                        self._log.error("Could not resolve cloud-construct for placement group: %s", group.name)
+                        self._log.info("Could not resolve cloud-construct for placement group: %s", group.name)
                         ### raise PlacementGroupError("Could not resolve cloud-construct for placement group: {}".format(group.name))
                     else:
                         self._log.info("Successfully resolved cloud construct for placement group: %s for VNF: %s (Member Index: %s)",
@@ -2021,10 +2026,11 @@ class NetworkServiceRecord(object):
         # Fetch the VNFD associated with this VNF
         placement_groups = self.get_placement_groups(vnfd_msg, const_vnfd)
         self._log.info("Cloud Account for VNF %d is %s",const_vnfd.member_vnf_index,cloud_account_name)
-        self._log.info("Launching VNF: %s (Member Index: %s) in NSD plancement Groups: %s",
+        self._log.info("Launching VNF: %s (Member Index: %s) in NSD plancement Groups: %s, restart mode self.restart_mode %s",
                        vnfd_msg.name,
                        const_vnfd.member_vnf_index,
-                       [ group.name for group in placement_groups])
+                       [ group.name for group in placement_groups],
+                       self.restart_mode)
         vnfr = yield from VirtualNetworkFunctionRecord.create_record(self._dts,
                                             self._log,
                                             self._loop,
@@ -2417,6 +2423,8 @@ class NetworkServiceRecord(object):
             for vlr in self.vlrs:
                 yield from self.nsm_plugin.terminate_vl(vlr)
                 vlr.state = VlRecordState.TERMINATED
+                if vlr.id in self.vlr_uptime_tasks:
+                    self.vlr_uptime_tasks[vlr.id].cancel()
 
         self._log.debug("Terminating network service id %s", self.id)
 
@@ -3611,13 +3619,18 @@ class NsManager(object):
         self._cloud_account_handler = cloud_account_handler
 
         self._ro_plugin_selector = ro_plugin_selector
-        self._ncclient = rift.mano.ncclient.NcClient(
-              host="127.0.0.1",
-              port=2022,
-              username="admin",
-              password="admin",
-              loop=self._loop)
 
+        # Intialize the set of variables for implementing Scaling RPC using REST.
+        self._headers = {"content-type":"application/json", "accept":"application/json"}
+        #This will break when we have rbac in the rift code and admin user password is changed or admin it self is removed.
+        self._user = 'admin'
+        self._password = 'admin'
+        self._ip = 'localhost'
+        self._rport = 8008
+        self._conf_url = "https://{ip}:{port}/api/config". \
+                       format(ip=self._ip,
+                              port=self._rport)
+        
         self._nsrs = {}
         self._nsds = {}
         self._vnfds = {}
@@ -3749,66 +3762,82 @@ class NsManager(object):
             msg : RPC input
             action : Scaling Action
         """
-        ScalingGroupInstance = NsrYang.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup_Instance
-        ScalingGroup = NsrYang.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup
-
-        xpath = ('C,/nsr:ns-instance-config/nsr:nsr[nsr:id="{}"]').format(
-                          msg.nsr_id_ref)
-        instance = ScalingGroupInstance.from_dict({"id": msg.instance_id})
-
-        @asyncio.coroutine
-        def get_nsr_scaling_group():
-            results = yield from self._dts.query_read(xpath, rwdts.XactFlag.MERGE)
-
-            for result in results:
-                res = yield from result
-                nsr_config = res.result
-
-            for scaling_group in nsr_config.scaling_group:
-                if scaling_group.scaling_group_name_ref == msg.scaling_group_name_ref:
-                    break
-            else:
-                scaling_group = nsr_config.scaling_group.add()
-                scaling_group.scaling_group_name_ref = msg.scaling_group_name_ref
-
-            return (nsr_config, scaling_group)
-
-        @asyncio.coroutine
-        def update_config(nsr_config):
-            xml = self._ncclient.convert_to_xml(RwNsrYang, nsr_config)
-            xml = '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml)
-            yield from self._ncclient.connect()
-            yield from self._ncclient.manager.edit_config(target="running", config=xml, default_operation="replace")
-
-        @asyncio.coroutine
+        def get_scaling_group_information():
+            scaling_group_url = "{url}/ns-instance-config/nsr/{nsr_id}".format(url=self._conf_url, nsr_id=msg.nsr_id_ref)
+            output = requests.get(scaling_group_url, headers=self._headers, auth=(self._user, self._password), verify=False)
+            if output.text == None or len(output.text) == 0:
+                self.log.error("nsr id %s information not present", self._nsr_id)
+                return None
+            scaling_group_info = json.loads(output.text)
+            return scaling_group_info
+        
+        def config_scaling_group_information(scaling_group_info):
+            data_str = json.dumps(scaling_group_info)
+            self.log.debug("scaling group Info %s", data_str)
+
+            scale_out_url = "{url}/ns-instance-config/nsr/{nsr_id}".format(url=self._conf_url, nsr_id=msg.nsr_id_ref)
+            response = requests.put(scale_out_url, data=data_str, verify=False, auth=(self._user, self._password), headers=self._headers)
+            response.raise_for_status()
+            
         def scale_out():
-            nsr_config, scaling_group = yield from get_nsr_scaling_group()
-            scaling_group.instance.append(instance)
-            yield from update_config(nsr_config)
+            scaling_group_info = get_scaling_group_information()
+            if scaling_group_info is None:
+                return
+                    
+            scaling_group_present = False
+            if "scaling-group" in scaling_group_info["nsr:nsr"]:
+                scaling_group_array = scaling_group_info["nsr:nsr"]["scaling-group"]
+                for scaling_group in scaling_group_array:
+                    if scaling_group["scaling-group-name-ref"] == msg.scaling_group_name_ref:
+                        scaling_group_present = True
+                        if 'instance' not in scaling_group:
+                            scaling_group['instance'] = []
+                        for instance in scaling_group['instance']:
+                            if instance["id"] == int(msg.instance_id):
+                                self.log.error("scaling group with instance id %s exists for scale out", msg.instance_id)
+                                return 
+                        scaling_group["instance"].append({"id": int(msg.instance_id)})       
+            
+            if not scaling_group_present:
+                scaling_group_info["nsr:nsr"]["scaling-group"] = [{"scaling-group-name-ref": msg.scaling_group_name_ref, "instance": [{"id": msg.instance_id}]}]
+                
+            config_scaling_group_information(scaling_group_info)
+            return
 
-        @asyncio.coroutine
         def scale_in():
-            nsr_config, scaling_group = yield from get_nsr_scaling_group()
-            scaling_group.instance.remove(instance)
-            yield from update_config(nsr_config)
+            scaling_group_info = get_scaling_group_information()
+            if scaling_group_info is None:
+                return
+            
+            scaling_group_array = scaling_group_info["nsr:nsr"]["scaling-group"]
+            scaling_group_present = False
+            instance_id_present = False
+            for scaling_group in scaling_group_array:
+                if scaling_group["scaling-group-name-ref"] == msg.scaling_group_name_ref:
+                    scaling_group_present = True
+                    if 'instance' in scaling_group:
+                        instance_array = scaling_group["instance"];
+                        for index in range(len(instance_array)):       
+                            if instance_array[index]["id"] == int(msg.instance_id):
+                                instance_array.pop(index)
+                                instance_id_present = True
+                                break
+    
+            if not scaling_group_present:
+                self.log.error("Scaling group %s doesnot exists for scale in", msg.scaling_group_name_ref)
+                return
+            
+            if not instance_id_present:
+                self.log.error("Instance id %s doesnot exists for scale in", msg.instance_id)
+                return
+            
+            config_scaling_group_information(scaling_group_info)
+            return
 
         if action == ScalingRpcHandler.ACTION.SCALE_OUT:
-            self._loop.create_task(scale_out())
+            self._loop.run_in_executor(None, scale_out)
         else:
-            self._loop.create_task(scale_in())
-
-        # Opdata based calls, disabled for now!
-        # if action == ScalingRpcHandler.ACTION.SCALE_OUT:
-        #     self.scale_nsr_out(
-        #           msg.nsr_id_ref,
-        #           msg.scaling_group_name_ref,
-        #           msg.instance_id,
-        #           xact)
-        # else:
-        #     self.scale_nsr_in(
-        #           msg.nsr_id_ref,
-        #           msg.scaling_group_name_ref,
-        #           msg.instance_id)
+            self._loop.run_in_executor(None, scale_in)
 
     def nsr_update_cfg(self, nsr_id, msg):
         nsr = self._nsrs[nsr_id]
@@ -3839,9 +3868,10 @@ class NsManager(object):
             self._log.error(msg)
             raise NetworkServiceRecordError(msg)
 
-        self._log.info("Create NetworkServiceRecord nsr id %s from nsd_id %s",
+        self._log.info("Create NetworkServiceRecord nsr id %s from nsd_id %s, restart mode %s",
                        nsr_msg.id,
-                       nsr_msg.nsd.id)
+                       nsr_msg.nsd.id,
+                       restart_mode)
 
         nsm_plugin = self._ro_plugin_selector.ro_plugin
         sdn_account_name = self._cloud_account_handler.get_cloud_account_sdn_name(nsr_msg.cloud_account)
@@ -4117,7 +4147,10 @@ class NsManager(object):
 
         # Terminate the instances/networks assocaited with this nw service
         self._log.debug("Terminating the network service %s", nsr_id)
-        yield from self._nsrs[nsr_id].terminate()
+        try :
+            yield from self._nsrs[nsr_id].terminate()
+        except Exception as e:
+            self.log.exception("Failed to terminate NSR[id=%s]", nsr_id)
 
         # Unref the NSD
         yield from self.nsd_unref_by_nsr_id(nsr_id)
@@ -4203,9 +4236,6 @@ class ScalingRpcHandler(mano_dts.DtsHandler):
             assert action == rwdts.QueryAction.RPC
 
             try:
-                if self.callback:
-                    self.callback(xact_info.xact, msg, self.ACTION.SCALE_IN)
-
                 rpc_op = NsrYang.YangOutput_Nsr_ExecScaleIn.from_dict({
                       "instance_id": msg.instance_id})
 
@@ -4214,6 +4244,8 @@ class ScalingRpcHandler(mano_dts.DtsHandler):
                     self.__class__.SCALE_IN_OUTPUT_XPATH,
                     rpc_op)
 
+                if self.callback:
+                    self.callback(xact_info.xact, msg, self.ACTION.SCALE_IN)
             except Exception as e:
                 self.log.exception(e)
                 xact_info.respond_xpath(
@@ -4231,9 +4263,6 @@ class ScalingRpcHandler(mano_dts.DtsHandler):
                     msg.instance_id  = last_instance_id + 1
                     self.last_instance_id[scale_group] += 1
 
-                if self.callback:
-                    self.callback(xact_info.xact, msg, self.ACTION.SCALE_OUT)
-
                 rpc_op = NsrYang.YangOutput_Nsr_ExecScaleOut.from_dict({
                       "instance_id": msg.instance_id})
 
@@ -4242,6 +4271,8 @@ class ScalingRpcHandler(mano_dts.DtsHandler):
                     self.__class__.SCALE_OUT_OUTPUT_XPATH,
                     rpc_op)
 
+                if self.callback:
+                    self.callback(xact_info.xact, msg, self.ACTION.SCALE_OUT)
             except Exception as e:
                 self.log.exception(e)
                 xact_info.respond_xpath(