X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_lcm%2Fns.py;h=0b69062ac96ae4798951b7c94db81b6a96494b9a;hb=refs%2Fchanges%2F37%2F12637%2F2;hp=2e9c1bcc5fc79b0cea9e601833a3458200d7ddee;hpb=b827de931cbe2452bcca76b8b24371b4a8afe2aa;p=osm%2FLCM.git diff --git a/osm_lcm/ns.py b/osm_lcm/ns.py index 2e9c1bc..0b69062 100644 --- a/osm_lcm/ns.py +++ b/osm_lcm/ns.py @@ -134,6 +134,7 @@ class NsLcm(LcmBase): ) # timeout for some progress in a primitive execution timeout_migrate = 1800 # default global timeout for migrating vnfs timeout_operate = 1800 # default global timeout for migrating vnfs + timeout_verticalscale = 1800 # default global timeout for Vertical Sclaing SUBOPERATION_STATUS_NOT_FOUND = -1 SUBOPERATION_STATUS_NEW = -2 SUBOPERATION_STATUS_SKIP = -3 @@ -224,6 +225,8 @@ class NsLcm(LcmBase): "termination": self.RO.status, "migrate": self.RO.status, "healing": self.RO.recreate_status, + "verticalscale": self.RO.status, + "start_stop_rebuild": self.RO.status, } @staticmethod @@ -421,7 +424,7 @@ class NsLcm(LcmBase): @staticmethod def _parse_cloud_init(cloud_init_text, additional_params, vnfd_id, vdu_id): try: - env = Environment(undefined=StrictUndefined) + env = Environment(undefined=StrictUndefined, autoescape=True) template = env.from_string(cloud_init_text) return template.render(additional_params or {}) except UndefinedError as e: @@ -897,18 +900,55 @@ class NsLcm(LcmBase): get_iterable(vdur, "interfaces"), lambda iface: iface.get("ns-vld-id") == a_vld["name"], ) + + vld_params = find_in_list( + get_iterable(ns_params, "vld"), + lambda v_vld: v_vld["name"] in (a_vld["name"], a_vld["id"]), + ) if target_vld: + if vnf_params.get("vimAccountId") not in a_vld.get( "vim_info", {} ): + target_vim_network_list = [ + v for _, v in a_vld.get("vim_info").items() + ] + target_vim_network_name = next( + ( + item.get("vim_network_name", "") + for item in target_vim_network_list + ), + "", + ) + target["ns"]["vld"][a_index].get("vim_info").update( { "vim:{}".format(vnf_params["vimAccountId"]): { - "vim_network_name": "" + "vim_network_name": target_vim_network_name, } } ) + if vld_params: + for param in ("vim-network-name", "vim-network-id"): + if vld_params.get(param) and isinstance( + vld_params[param], dict + ): + for vim, vim_net in vld_params[ + param + ].items(): + other_target_vim = "vim:" + vim + populate_dict( + target["ns"]["vld"][a_index].get( + "vim_info" + ), + ( + other_target_vim, + param.replace("-", "_"), + ), + vim_net, + ) + nslcmop_id = db_nslcmop["_id"] target = { "name": db_nsr["name"], @@ -5196,12 +5236,17 @@ class NsLcm(LcmBase): parts = kdu_model.split(sep=":") if len(parts) == 2: kdu_model = parts[0] + if desc_params.get("kdu_atomic_upgrade"): + atomic_upgrade = desc_params.get("kdu_atomic_upgrade").lower() in ("yes", "true", "1") + del desc_params["kdu_atomic_upgrade"] + else: + atomic_upgrade = True detailed_status = await asyncio.wait_for( self.k8scluster_map[kdu["k8scluster-type"]].upgrade( cluster_uuid=kdu.get("k8scluster-uuid"), kdu_instance=kdu.get("kdu-instance"), - atomic=True, + atomic=atomic_upgrade, kdu_model=kdu_model, params=desc_params, db_dict=db_dict, @@ -7236,13 +7281,17 @@ class NsLcm(LcmBase): db_vnfr = self.db.get_one("vnfrs", {"_id": vnf_id}) vim_account_id = db_vnfr.get("vim-account-id") vim_info_key = "vim:" + vim_account_id + vdu_id = additional_param["vdu_id"] + vdurs = [item for item in db_vnfr["vdur"] if item["vdu-id-ref"] == vdu_id] vdur = find_in_list( - db_vnfr["vdur"], lambda vdu: vdu["count-index"] == additional_param["count-index"] + vdurs, lambda vdu: vdu["count-index"] == additional_param["count-index"] ) if vdur: vdu_vim_name = vdur["name"] vim_vm_id = vdur["vim_info"][vim_info_key]["vim_id"] target_vim, _ = next(k_v for k_v in vdur["vim_info"].items()) + else: + raise LcmException("Target vdu is not found") self.logger.info("vdu_vim_name >> {} ".format(vdu_vim_name)) # wait for any previous tasks in process stage[1] = "Waiting for previous operations to terminate" @@ -7281,7 +7330,8 @@ class NsLcm(LcmBase): self.logger.info("response from RO: {}".format(result_dict)) action_id = result_dict["action_id"] await self._wait_ng_ro( - nsr_id, action_id, nslcmop_id, start_deploy, self.timeout_operate + nsr_id, action_id, nslcmop_id, start_deploy, + self.timeout_operate, None, "start_stop_rebuild", ) return "COMPLETED", "Done" except (ROclient.ROClientException, DbException, LcmException) as e: @@ -7473,17 +7523,22 @@ class NsLcm(LcmBase): self.update_db_2("nsrs", nsr_id, db_nsr_update) step = "Sending heal order to VIM" - task_ro = asyncio.ensure_future( - self.heal_RO( + #task_ro = asyncio.ensure_future( + # self.heal_RO( + # logging_text=logging_text, + # nsr_id=nsr_id, + # db_nslcmop=db_nslcmop, + # stage=stage, + # ) + #) + #self.lcm_tasks.register("ns", nsr_id, nslcmop_id, "heal_RO", task_ro) + #tasks_dict_info[task_ro] = "Healing at VIM" + await self.heal_RO( logging_text=logging_text, nsr_id=nsr_id, db_nslcmop=db_nslcmop, stage=stage, ) - ) - self.lcm_tasks.register("ns", nsr_id, nslcmop_id, "heal_RO", task_ro) - tasks_dict_info[task_ro] = "Healing at VIM" - # VCA tasks # read from db: nsd stage[1] = "Getting nsd={} from db.".format(db_nsr["nsd-id"]) @@ -7517,7 +7572,17 @@ class NsLcm(LcmBase): member_vnf_index = db_vnfr.get("member-vnf-index-ref") # Check each target VDU and deploy N2VC - for target_vdu in target_vnf["additionalParams"].get("vdu", None): + target_vdu_list = target_vnf.get("additionalParams", {}).get("vdu", []) + if not target_vdu_list: + # Codigo nuevo para crear diccionario + target_vdu_list = [] + for existing_vdu in db_vnfr.get("vdur"): + vdu_name = existing_vdu.get("vdu-name", None) + vdu_index = existing_vdu.get("count-index", 0) + vdu_run_day1 = target_vnf.get("additionalParams", {}).get("run-day1", False) + vdu_to_be_healed = {"vdu-id": vdu_name, "count-index": vdu_index, "run-day1": vdu_run_day1} + target_vdu_list.append(vdu_to_be_healed) + for target_vdu in target_vdu_list: deploy_params_vdu = target_vdu # Set run-day1 vnf level value if not vdu level value exists if not deploy_params_vdu.get("run-day1") and target_vnf["additionalParams"].get("run-day1"): @@ -8337,3 +8402,99 @@ class NsLcm(LcmBase): await asyncio.sleep(15, loop=self.loop) else: # timeout_ns_deploy raise NgRoException("Timeout waiting ns to deploy") + + async def vertical_scale(self, nsr_id, nslcmop_id): + """ + Vertical Scale the VDUs in a NS + + :param: nsr_id: NS Instance ID + :param: nslcmop_id: nslcmop ID of migrate + + """ + # Try to lock HA task here + task_is_locked_by_me = self.lcm_tasks.lock_HA("ns", "nslcmops", nslcmop_id) + if not task_is_locked_by_me: + return + logging_text = "Task ns={} vertical scale ".format(nsr_id) + self.logger.debug(logging_text + "Enter") + # get all needed from database + db_nslcmop = None + db_nslcmop_update = {} + nslcmop_operation_state = None + db_nsr_update = {} + target = {} + exc = None + # in case of error, indicates what part of scale was failed to put nsr at error status + start_deploy = time() + + try: + # wait for any previous tasks in process + step = "Waiting for previous operations to terminate" + await self.lcm_tasks.waitfor_related_HA('ns', 'nslcmops', nslcmop_id) + + self._write_ns_status( + nsr_id=nsr_id, + ns_state=None, + current_operation="VerticalScale", + current_operation_id=nslcmop_id + ) + step = "Getting nslcmop from database" + self.logger.debug(step + " after having waited for previous tasks to be completed") + db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id}) + operationParams = db_nslcmop.get("operationParams") + target = {} + target.update(operationParams) + desc = await self.RO.vertical_scale(nsr_id, target) + self.logger.debug("RO return > {}".format(desc)) + action_id = desc["action_id"] + await self._wait_ng_ro( + nsr_id, action_id, nslcmop_id, start_deploy, self.timeout_verticalscale, + operation="verticalscale" + ) + except (ROclient.ROClientException, DbException, LcmException) as e: + self.logger.error("Exit Exception {}".format(e)) + exc = e + except asyncio.CancelledError: + self.logger.error("Cancelled Exception while '{}'".format(step)) + exc = "Operation was cancelled" + except Exception as e: + exc = traceback.format_exc() + self.logger.critical("Exit Exception {} {}".format(type(e).__name__, e), exc_info=True) + finally: + self._write_ns_status( + nsr_id=nsr_id, + ns_state=None, + current_operation="IDLE", + current_operation_id=None, + ) + if exc: + db_nslcmop_update[ + "detailed-status" + ] = "FAILED {}: {}".format(step, exc) + nslcmop_operation_state = "FAILED" + else: + nslcmop_operation_state = "COMPLETED" + db_nslcmop_update["detailed-status"] = "Done" + db_nsr_update["detailed-status"] = "Done" + + self._write_op_status( + op_id=nslcmop_id, + stage="", + error_message="", + operation_state=nslcmop_operation_state, + other_update=db_nslcmop_update, + ) + if nslcmop_operation_state: + try: + msg = { + "nsr_id": nsr_id, + "nslcmop_id": nslcmop_id, + "operationState": nslcmop_operation_state, + } + await self.msg.aiowrite("ns", "verticalscaled", msg, loop=self.loop) + except Exception as e: + self.logger.error( + logging_text + "kafka_write notification Exception {}".format(e) + ) + self.logger.debug(logging_text + "Exit") + self.lcm_tasks.remove("ns", nsr_id, nslcmop_id, "ns_verticalscale")