X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FSO.git;a=blobdiff_plain;f=rwlaunchpad%2Fplugins%2Frwnsm%2Frift%2Ftasklets%2Frwnsmtasklet%2Frwnsmtasklet.py;h=afcb822a4ee103874d951067ab7452d5d9551e26;hp=b78a279b0896ad2cc64af0d1de84b9109a1127e2;hb=bfac7570738987efd4be7f76a1b0e6f8dd0386f7;hpb=85a3dc954226041e4db357fa807b1ffdb6953110 diff --git a/rwlaunchpad/plugins/rwnsm/rift/tasklets/rwnsmtasklet/rwnsmtasklet.py b/rwlaunchpad/plugins/rwnsm/rift/tasklets/rwnsmtasklet/rwnsmtasklet.py index b78a279b..afcb822a 100755 --- a/rwlaunchpad/plugins/rwnsm/rift/tasklets/rwnsmtasklet/rwnsmtasklet.py +++ b/rwlaunchpad/plugins/rwnsm/rift/tasklets/rwnsmtasklet/rwnsmtasklet.py @@ -51,7 +51,7 @@ from gi.repository import ( VnfrYang, RwVnfrYang, RwNsmYang, - RwsdnYang, + RwsdnalYang, RwDts as rwdts, RwTypes, ProtobufC, @@ -383,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 @@ -1701,7 +1701,6 @@ class NetworkServiceRecord(object): vnfr = yield from self.create_vnf_record(vnfd_msg, const_vnfd_msg, cloud_account_name, om_datacenter_name, group_name, index) scale_instance.add_vnfr(vnfr) vnfrs.append(vnfr) - return vnfrs @asyncio.coroutine @@ -1722,7 +1721,8 @@ class NetworkServiceRecord(object): format(group.name, index)) scale_instance.operational_status = "failed" else: - yield from self.instantiate_vnfs(vnfrs) + yield from self.instantiate_vnfs(vnfrs, scaleout=True) + except Exception as e: self._log.exception("Failed to begin instantiatiation of vnfs for scale group {}: {}". @@ -2011,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)", @@ -2026,11 +2026,10 @@ 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, restart mode self.restart_mode %s", + self._log.info("Launching VNF: %s (Member Index: %s) in NSD plancement Groups: %s", vnfd_msg.name, const_vnfd.member_vnf_index, - [ group.name for group in placement_groups], - self.restart_mode) + [ group.name for group in placement_groups]) vnfr = yield from VirtualNetworkFunctionRecord.create_record(self._dts, self._log, self._loop, @@ -2095,14 +2094,14 @@ class NetworkServiceRecord(object): return vnfr @asyncio.coroutine - def instantiate_vnfs(self, vnfrs): + def instantiate_vnfs(self, vnfrs, scaleout=False): """ This function instantiates VNFs for every VNF in this Network Service """ self._log.debug("Instantiating %u VNFs in NS %s", len(vnfrs), self.id) for vnf in vnfrs: self._log.debug("Instantiating VNF: %s in NS %s", vnf, self.id) - yield from self.nsm_plugin.instantiate_vnf(self, vnf) + yield from self.nsm_plugin.instantiate_vnf(self, vnf,scaleout) @asyncio.coroutine def instantiate_vnffgs(self): @@ -2351,9 +2350,12 @@ class NetworkServiceRecord(object): def on_instantiate_done(fut): # If the do_instantiate fails, then publish NSR with failed result - if fut.exception() is not None: - self._log.error("NSR instantiation failed for NSR id %s: %s", self.id, str(fut.exception())) - self._loop.create_task(self.instantiation_failed(failed_reason=str(fut.exception()))) + e = fut.exception() + if e is not None: + import traceback, sys + print(traceback.format_exception(None,e, e.__traceback__), file=sys.stderr, flush=True) + self._log.error("NSR instantiation failed for NSR id %s: %s", self.id, str(e)) + self._loop.create_task(self.instantiation_failed(failed_reason=str(e))) instantiate_task = self._loop.create_task(do_instantiate()) instantiate_task.add_done_callback(on_instantiate_done) @@ -3382,6 +3384,8 @@ class NsrDtsHandler(object): raise NsrVlUpdateError("NS config NSD should have atleast 1 VLD defined") if msg.has_field("scaling_group"): + self._log.debug("ScaleMsg %s", msg) + self._log.debug("NSSCALINGSTATE %s", nsr.state) if nsr.state != NetworkServiceRecordState.RUNNING: raise ScalingOperationError("Unable to perform scaling action when NS is not in running state") @@ -3619,13 +3623,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 = {} @@ -3641,6 +3650,7 @@ class NsManager(object): VnfrDtsHandler(dts, log, loop, self), NsdRefCountDtsHandler(dts, log, loop, self), NsrDtsHandler(dts, log, loop, self), + ScalingRpcHandler(log, dts, loop, self.scale_rpc_callback), NsrRpcDtsHandler(dts,log,loop,self), self._vnfd_dts_handler, self.cfgmgr_obj, @@ -3749,6 +3759,90 @@ class NsManager(object): self._loop.create_task(nsr.delete_scale_group_instance(scale_group_name, instance_id)) + def scale_rpc_callback(self, xact, msg, action): + """Callback handler for RPC calls + Args: + xact : Transaction Handler + msg : RPC input + action : Scaling Action + """ + 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(): + 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 + + def scale_in(): + 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.run_in_executor(None, scale_out) + else: + self._loop.run_in_executor(None, scale_in) + def nsr_update_cfg(self, nsr_id, msg): nsr = self._nsrs[nsr_id] nsr.nsr_cfg_msg= msg @@ -3773,15 +3867,15 @@ class NsManager(object): def create_nsr(self, nsr_msg, key_pairs=None,restart_mode=False): """ Create an NSR instance """ + self._log.debug("NSRMSG %s", nsr_msg) if nsr_msg.id in self._nsrs: msg = "NSR id %s already exists" % nsr_msg.id self._log.error(msg) raise NetworkServiceRecordError(msg) - self._log.info("Create NetworkServiceRecord nsr id %s from nsd_id %s, restart mode %s", + self._log.info("Create NetworkServiceRecord nsr id %s from nsd_id %s", nsr_msg.id, - nsr_msg.nsd.id, - restart_mode) + nsr_msg.nsd.id) nsm_plugin = self._ro_plugin_selector.ro_plugin sdn_account_name = self._cloud_account_handler.get_cloud_account_sdn_name(nsr_msg.cloud_account) @@ -4122,6 +4216,89 @@ class NsmRecordsPublisherProxy(object): path = VirtualLinkRecord.vlr_xpath(vlr) return (yield from self._vlr_pub_hdlr.delete(xact, path)) + +class ScalingRpcHandler(mano_dts.DtsHandler): + """ The Network service Monitor DTS handler """ + SCALE_IN_INPUT_XPATH = "I,/nsr:exec-scale-in" + SCALE_IN_OUTPUT_XPATH = "O,/nsr:exec-scale-in" + + SCALE_OUT_INPUT_XPATH = "I,/nsr:exec-scale-out" + SCALE_OUT_OUTPUT_XPATH = "O,/nsr:exec-scale-out" + + ACTION = Enum('ACTION', 'SCALE_IN SCALE_OUT') + + def __init__(self, log, dts, loop, callback=None): + super().__init__(log, dts, loop) + self.callback = callback + self.last_instance_id = defaultdict(int) + + @asyncio.coroutine + def register(self): + + @asyncio.coroutine + def on_scale_in_prepare(xact_info, action, ks_path, msg): + assert action == rwdts.QueryAction.RPC + + try: + rpc_op = NsrYang.YangOutput_Nsr_ExecScaleIn.from_dict({ + "instance_id": msg.instance_id}) + + xact_info.respond_xpath( + rwdts.XactRspCode.ACK, + 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( + rwdts.XactRspCode.NACK, + self.__class__.SCALE_IN_OUTPUT_XPATH) + + @asyncio.coroutine + def on_scale_out_prepare(xact_info, action, ks_path, msg): + assert action == rwdts.QueryAction.RPC + + try: + scaling_group = msg.scaling_group_name_ref + if not msg.instance_id: + last_instance_id = self.last_instance_id[scale_group] + msg.instance_id = last_instance_id + 1 + self.last_instance_id[scale_group] += 1 + + rpc_op = NsrYang.YangOutput_Nsr_ExecScaleOut.from_dict({ + "instance_id": msg.instance_id}) + + xact_info.respond_xpath( + rwdts.XactRspCode.ACK, + 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( + rwdts.XactRspCode.NACK, + self.__class__.SCALE_OUT_OUTPUT_XPATH) + + scale_in_hdl = rift.tasklets.DTS.RegistrationHandler( + on_prepare=on_scale_in_prepare) + scale_out_hdl = rift.tasklets.DTS.RegistrationHandler( + on_prepare=on_scale_out_prepare) + + with self.dts.group_create() as group: + group.register( + xpath=self.__class__.SCALE_IN_INPUT_XPATH, + handler=scale_in_hdl, + flags=rwdts.Flag.PUBLISHER) + group.register( + xpath=self.__class__.SCALE_OUT_INPUT_XPATH, + handler=scale_out_hdl, + flags=rwdts.Flag.PUBLISHER) + + class NsmTasklet(rift.tasklets.Tasklet): """ The network service manager tasklet