X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_lcm%2Fns.py;h=3e6bafc5e9c813ecd085dad9d8d006e596c96fad;hb=afff693a2ae9757aee5f731ee12bca70cdb9a907;hp=d57b739b8f92c5d11e38f0581862f64012a70101;hpb=349aa468cc1fce58d85f50a8bf650a94f73f938e;p=osm%2FLCM.git diff --git a/osm_lcm/ns.py b/osm_lcm/ns.py index d57b739..3e6bafc 100644 --- a/osm_lcm/ns.py +++ b/osm_lcm/ns.py @@ -30,9 +30,11 @@ from jinja2 import ( TemplateNotFound, StrictUndefined, UndefinedError, + select_autoescape, ) from osm_lcm import ROclient +from osm_lcm.data_utils.lcm_config import LcmCfg from osm_lcm.data_utils.nsr import ( get_deployed_kdu, get_deployed_vca, @@ -58,6 +60,7 @@ from osm_lcm.lcm_utils import ( populate_dict, check_juju_bundle_existence, get_charm_artifact_path, + get_ee_id_parts, ) from osm_lcm.data_utils.nsd import ( get_ns_configuration_relation_list, @@ -82,6 +85,7 @@ from osm_lcm.data_utils.vnfd import ( get_juju_ee_ref, get_kdu_resource_profile, find_software_version, + check_helm_ee_in_ns, ) from osm_lcm.data_utils.list_utils import find_in_list from osm_lcm.data_utils.vnfr import ( @@ -102,6 +106,11 @@ from osm_common.fsbase import FsException from osm_lcm.data_utils.database.database import Database from osm_lcm.data_utils.filesystem.filesystem import Filesystem +from osm_lcm.data_utils.wim import ( + get_sdn_ports, + get_target_wim_attrs, + select_feasible_wim_account, +) from n2vc.n2vc_juju_conn import N2VCJujuConnector from n2vc.exceptions import N2VCException, N2VCNotFound, K8sException @@ -120,25 +129,12 @@ __author__ = "Alfonso Tierno " class NsLcm(LcmBase): - timeout_vca_on_error = ( - 5 * 60 - ) # Time for charm from first time at blocked,error status to mark as failed - timeout_ns_deploy = 2 * 3600 # default global timeout for deployment a ns - timeout_ns_terminate = 1800 # default global timeout for un deployment a ns - timeout_charm_delete = 10 * 60 - timeout_primitive = 30 * 60 # timeout for primitive execution - timeout_ns_update = 30 * 60 # timeout for ns update - timeout_progress_primitive = ( - 10 * 60 - ) # timeout for some progress in a primitive execution - timeout_migrate = 1800 # default global timeout for migrating vnfs - SUBOPERATION_STATUS_NOT_FOUND = -1 SUBOPERATION_STATUS_NEW = -2 SUBOPERATION_STATUS_SKIP = -3 task_name_deploy_vca = "Deploying VCA" - def __init__(self, msg, lcm_tasks, config, loop): + def __init__(self, msg, lcm_tasks, config: LcmCfg, loop): """ Init, Connect to database, filesystem storage, and messaging :param config: two level dictionary with configuration. Top level should contain 'database', 'storage', @@ -150,10 +146,9 @@ class NsLcm(LcmBase): self.fs = Filesystem().instance.fs self.loop = loop self.lcm_tasks = lcm_tasks - self.timeout = config["timeout"] - self.ro_config = config["ro_config"] - self.ng_ro = config["ro_config"].get("ng") - self.vca_config = config["VCA"].copy() + self.timeout = config.timeout + self.ro_config = config.RO + self.vca_config = config.VCA # create N2VC connector self.n2vc = N2VCJujuConnector( @@ -172,8 +167,8 @@ class NsLcm(LcmBase): ) self.k8sclusterhelm2 = K8sHelmConnector( - kubectl_command=self.vca_config.get("kubectlpath"), - helm_command=self.vca_config.get("helmpath"), + kubectl_command=self.vca_config.kubectlpath, + helm_command=self.vca_config.helmpath, log=self.logger, on_update_db=None, fs=self.fs, @@ -181,8 +176,8 @@ class NsLcm(LcmBase): ) self.k8sclusterhelm3 = K8sHelm3Connector( - kubectl_command=self.vca_config.get("kubectlpath"), - helm_command=self.vca_config.get("helm3path"), + kubectl_command=self.vca_config.kubectlpath, + helm_command=self.vca_config.helm3path, fs=self.fs, log=self.logger, db=self.db, @@ -190,8 +185,8 @@ class NsLcm(LcmBase): ) self.k8sclusterjuju = K8sJujuConnector( - kubectl_command=self.vca_config.get("kubectlpath"), - juju_command=self.vca_config.get("jujupath"), + kubectl_command=self.vca_config.kubectlpath, + juju_command=self.vca_config.jujupath, log=self.logger, loop=self.loop, on_update_db=self._on_update_k8s_db, @@ -216,7 +211,16 @@ class NsLcm(LcmBase): } # create RO client - self.RO = NgRoClient(self.loop, **self.ro_config) + self.RO = NgRoClient(self.loop, **self.ro_config.to_dict()) + + self.op_status_map = { + "instantiation": self.RO.status, + "termination": self.RO.status, + "migrate": self.RO.status, + "healing": self.RO.recreate_status, + "verticalscale": self.RO.status, + "start_stop_rebuild": self.RO.status, + } @staticmethod def increment_ip_mac(ip_mac, vm_index=1): @@ -282,7 +286,6 @@ class NsLcm(LcmBase): # vcaStatus db_dict = dict() db_dict["vcaStatus"] = status_dict - await self.n2vc.update_vca_status(db_dict["vcaStatus"], vca_id=vca_id) # update configurationStatus for this VCA try: @@ -390,15 +393,6 @@ class NsLcm(LcmBase): db_dict = dict() db_dict["vcaStatus"] = {nsr_id: vca_status} - if cluster_type in ("juju-bundle", "juju"): - # TODO -> this should be done in a more uniform way, I think in N2VC, in order to update the K8s VCA - # status in a similar way between Juju Bundles and Helm Charts on this side - await self.k8sclusterjuju.update_vca_status( - db_dict["vcaStatus"], - kdu_instance, - vca_id=vca_id, - ) - self.logger.debug( f"Obtained VCA status for cluster type '{cluster_type}': {vca_status}" ) @@ -413,7 +407,10 @@ 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=select_autoescape(default_for_string=True, default=True), + ) template = env.from_string(cloud_init_text) return template.render(additional_params or {}) except UndefinedError as e: @@ -850,9 +847,30 @@ class NsLcm(LcmBase): target_vld["vim_info"][target_sdn]["sdn-ports"] = vld_params[ "provider-network" ]["sdn-ports"] - if vld_params.get("wimAccountId"): - target_wim = "wim:{}".format(vld_params["wimAccountId"]) - target_vld["vim_info"][target_wim] = {} + + # check if WIM is needed; if needed, choose a feasible WIM able to connect VIMs + # if wim_account_id is specified in vld_params, validate if it is feasible. + wim_account_id, db_wim = select_feasible_wim_account( + db_nsr, db_vnfrs, target_vld, vld_params, self.logger + ) + + if wim_account_id: + # WIM is needed and a feasible one was found, populate WIM target and SDN ports + self.logger.info("WIM selected: {:s}".format(str(wim_account_id))) + # update vld_params with correct WIM account Id + vld_params["wimAccountId"] = wim_account_id + + target_wim = "wim:{}".format(wim_account_id) + target_wim_attrs = get_target_wim_attrs(nsr_id, target_vld, vld_params) + sdn_ports = get_sdn_ports(vld_params, db_wim) + if len(sdn_ports) > 0: + target_vld["vim_info"][target_wim] = target_wim_attrs + target_vld["vim_info"][target_wim]["sdn-ports"] = sdn_ports + + self.logger.debug( + "Target VLD with WIM data: {:s}".format(str(target_vld)) + ) + for param in ("vim-network-name", "vim-network-id"): if vld_params.get(param): if isinstance(vld_params[param], dict): @@ -884,23 +902,62 @@ class NsLcm(LcmBase): None, ) vdur = next((vdur for vdur in target_vnf.get("vdur", ())), None) + if not vdur: + return for a_index, a_vld in enumerate(target["ns"]["vld"]): target_vld = find_in_list( 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"], @@ -1239,11 +1296,18 @@ class NsLcm(LcmBase): target_vnf["vdur"] = vdur_list target["vnf"].append(target_vnf) + self.logger.debug("Send to RO > nsr_id={} target={}".format(nsr_id, target)) desc = await self.RO.deploy(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, timeout_ns_deploy, stage + nsr_id, + action_id, + nslcmop_id, + start_deploy, + timeout_ns_deploy, + stage, + operation="instantiation", ) # Updating NSR @@ -1267,12 +1331,13 @@ class NsLcm(LcmBase): start_time=None, timeout=600, stage=None, + operation=None, ): detailed_status_old = None db_nsr_update = {} start_time = start_time or time() while time() <= start_time + timeout: - desc_status = await self.RO.status(nsr_id, action_id) + desc_status = await self.op_status_map[operation](nsr_id, action_id) self.logger.debug("Wait NG RO > {}".format(desc_status)) if desc_status["status"] == "FAILED": raise NgRoException(desc_status["details"]) @@ -1323,7 +1388,13 @@ class NsLcm(LcmBase): # wait until done delete_timeout = 20 * 60 # 20 minutes await self._wait_ng_ro( - nsr_id, action_id, nslcmop_id, start_deploy, delete_timeout, stage + nsr_id, + action_id, + nslcmop_id, + start_deploy, + delete_timeout, + stage, + operation="termination", ) db_nsr_update["_admin.deployed.RO.nsr_delete_action_id"] = None @@ -1394,9 +1465,7 @@ class NsLcm(LcmBase): if ns_params and ns_params.get("timeout_ns_deploy"): timeout_ns_deploy = ns_params["timeout_ns_deploy"] else: - timeout_ns_deploy = self.timeout.get( - "ns_deploy", self.timeout_ns_deploy - ) + timeout_ns_deploy = self.timeout.ns_deploy # Check for and optionally request placement optimization. Database will be updated if placement activated stage[2] = "Waiting for Placement." @@ -1586,7 +1655,7 @@ class NsLcm(LcmBase): ro_vm_id = "{}-{}".format( db_vnfr["member-vnf-index-ref"], target_vdu_id ) # TODO add vdu_index - if self.ng_ro: + if self.ro_config.ng: target = { "action": { "action": "inject_ssh_key", @@ -1597,7 +1666,9 @@ class NsLcm(LcmBase): } desc = await self.RO.deploy(nsr_id, target) action_id = desc["action_id"] - await self._wait_ng_ro(nsr_id, action_id, timeout=600) + await self._wait_ng_ro( + nsr_id, action_id, timeout=600, operation="instantiation" + ) break else: # wait until NS is deployed at RO @@ -1711,6 +1782,7 @@ class NsLcm(LcmBase): vdu_id, kdu_name, vdu_index, + kdu_index, config_descriptor, deploy_params, base_folder, @@ -1847,6 +1919,7 @@ class NsLcm(LcmBase): db_dict=db_dict, config=osm_config, artifact_path=artifact_path, + chart_model=vca_name, vca_type=vca_type, ) else: @@ -2046,7 +2119,7 @@ class NsLcm(LcmBase): # for a KNF and not for its KDUs, the previous verification gives False, and the code # jumps to this block, meaning that there is the need to verify if the VNF is actually a VNF # or it is a KNF) - elif db_vnfr.get('vdur'): + elif db_vnfr.get("vdur"): rw_mgmt_ip = await self.wait_vm_up_insert_key_ro( logging_text, nsr_id, @@ -2123,6 +2196,8 @@ class NsLcm(LcmBase): # STEP 7 Configure metrics if vca_type == "helm" or vca_type == "helm-v3": + # TODO: review for those cases where the helm chart is a reference and + # is not part of the NF package prometheus_jobs = await self.extract_prometheus_scrape_jobs( ee_id=ee_id, artifact_path=artifact_path, @@ -2130,6 +2205,11 @@ class NsLcm(LcmBase): vnfr_id=vnfr_id, nsr_id=nsr_id, target_ip=rw_mgmt_ip, + vnf_member_index=db_vnfr.get("member-vnf-index-ref", ""), + vdu_id=vdu_id, + vdu_index=vdu_index, + kdu_name=kdu_name, + kdu_index=kdu_index, ) if prometheus_jobs: self.update_db_2( @@ -2419,9 +2499,7 @@ class NsLcm(LcmBase): if ns_params and ns_params.get("timeout_ns_deploy"): timeout_ns_deploy = ns_params["timeout_ns_deploy"] else: - timeout_ns_deploy = self.timeout.get( - "ns_deploy", self.timeout_ns_deploy - ) + timeout_ns_deploy = self.timeout.ns_deploy # read from db: ns stage[1] = "Getting nsr={} from db.".format(nsr_id) @@ -2522,8 +2600,8 @@ class NsLcm(LcmBase): # feature 1429. Add n2vc public key to needed VMs n2vc_key = self.n2vc.get_public_key() n2vc_key_list = [n2vc_key] - if self.vca_config.get("public_key"): - n2vc_key_list.append(self.vca_config["public_key"]) + if self.vca_config.public_key: + n2vc_key_list.append(self.vca_config.public_key) stage[1] = "Deploying NS at VIM." task_ro = asyncio.ensure_future( @@ -2546,6 +2624,17 @@ class NsLcm(LcmBase): stage[1] = "Deploying Execution Environments." self.logger.debug(logging_text + stage[1]) + # create namespace and certificate if any helm based EE is present in the NS + if check_helm_ee_in_ns(db_vnfds): + # TODO: create EE namespace + # create TLS certificates + await self.vca_map["helm-v3"].create_tls_certificate( + secret_name="ee-tls-{}".format(nsr_id), + dns_prefix="*", + nsr_id=nsr_id, + usage="server auth", + ) + nsi_id = None # TODO put nsi_id when this nsr belongs to a NSI for vnf_profile in get_vnf_profiles(nsd): vnfd_id = vnf_profile["vnfd-id"] @@ -2557,6 +2646,7 @@ class NsLcm(LcmBase): vdu_index = 0 vdu_name = None kdu_name = None + kdu_index = None # Get additional parameters deploy_params = {"OSM": get_osm_params(db_vnfr)} @@ -2580,6 +2670,7 @@ class NsLcm(LcmBase): kdu_name=kdu_name, member_vnf_index=member_vnf_index, vdu_index=vdu_index, + kdu_index=kdu_index, vdu_name=vdu_name, deploy_params=deploy_params, descriptor_config=descriptor_config, @@ -2612,6 +2703,7 @@ class NsLcm(LcmBase): if descriptor_config: vdu_name = None kdu_name = None + kdu_index = None for vdu_index in range(vdud_count): # TODO vnfr_params["rw_mgmt_ip"] = vdur["ip-address"] self._deploy_n2vc( @@ -2627,6 +2719,7 @@ class NsLcm(LcmBase): vnfd_id=vnfd_id, vdu_id=vdu_id, kdu_name=kdu_name, + kdu_index=kdu_index, member_vnf_index=member_vnf_index, vdu_index=vdu_index, vdu_name=vdu_name, @@ -2643,8 +2736,10 @@ class NsLcm(LcmBase): vdu_id = None vdu_index = 0 vdu_name = None - kdur = next( - x for x in db_vnfr["kdur"] if x["kdu-name"] == kdu_name + kdu_index, kdur = next( + x + for x in enumerate(db_vnfr["kdur"]) + if x[1]["kdu-name"] == kdu_name ) deploy_params_kdu = {"OSM": get_osm_params(db_vnfr)} if kdur.get("additionalParams"): @@ -2664,6 +2759,7 @@ class NsLcm(LcmBase): kdu_name=kdu_name, member_vnf_index=member_vnf_index, vdu_index=vdu_index, + kdu_index=kdu_index, vdu_name=vdu_name, deploy_params=deploy_params_kdu, descriptor_config=descriptor_config, @@ -2680,6 +2776,7 @@ class NsLcm(LcmBase): member_vnf_index = None vdu_id = None kdu_name = None + kdu_index = None vdu_index = 0 vdu_name = None @@ -2702,6 +2799,7 @@ class NsLcm(LcmBase): kdu_name=kdu_name, member_vnf_index=member_vnf_index, vdu_index=vdu_index, + kdu_index=kdu_index, vdu_name=vdu_name, deploy_params=deploy_params, descriptor_config=descriptor_config, @@ -2832,9 +2930,11 @@ class NsLcm(LcmBase): self.logger.debug(logging_text + "Exit") self.lcm_tasks.remove("ns", nsr_id, nslcmop_id, "ns_instantiate") - def _get_vnfd(self, vnfd_id: str, cached_vnfds: Dict[str, Any]): + def _get_vnfd(self, vnfd_id: str, projects_read: str, cached_vnfds: Dict[str, Any]): if vnfd_id not in cached_vnfds: - cached_vnfds[vnfd_id] = self.db.get_one("vnfds", {"id": vnfd_id}) + cached_vnfds[vnfd_id] = self.db.get_one( + "vnfds", {"id": vnfd_id, "_admin.projects_read": projects_read} + ) return cached_vnfds[vnfd_id] def _get_vnfr(self, nsr_id: str, vnf_profile_id: str, cached_vnfrs: Dict[str, Any]): @@ -2876,7 +2976,8 @@ class NsLcm(LcmBase): ]: vnf_profile = get_vnf_profile(nsd, ee_relation_data["vnf-profile-id"]) vnfd_id = vnf_profile["vnfd-id"] - db_vnfd = self._get_vnfd(vnfd_id, cached_vnfds) + project = nsd["_admin"]["projects_read"][0] + db_vnfd = self._get_vnfd(vnfd_id, project, cached_vnfds) entity_id = ( vnfd_id if ee_relation_level == EELevel.VNF @@ -2949,7 +3050,8 @@ class NsLcm(LcmBase): vnf_profile = get_vnf_profile(nsd, vca.vnf_profile_id) vnf_profile_id = vnf_profile["id"] vnfd_id = vnf_profile["vnfd-id"] - db_vnfd = self._get_vnfd(vnfd_id, cached_vnfds) + project = nsd["_admin"]["projects_read"][0] + db_vnfd = self._get_vnfd(vnfd_id, project, cached_vnfds) db_vnf_relations = get_relation_list(db_vnfd, vnfd_id) for r in db_vnf_relations: provider_dict = None @@ -3004,7 +3106,8 @@ class NsLcm(LcmBase): vnf_profiles, lambda vnf_profile: vnf_profile["id"] == ee_relation.vnf_profile_id, )["vnfd-id"] - db_vnfd = self._get_vnfd(vnfd_id, cached_vnfds) + project = nsd["_admin"]["projects_read"][0] + db_vnfd = self._get_vnfd(vnfd_id, project, cached_vnfds) kdu_resource_profile = get_kdu_resource_profile( db_vnfd, ee_relation.kdu_resource_profile_id ) @@ -3674,6 +3777,7 @@ class NsLcm(LcmBase): kdu_name, member_vnf_index, vdu_index, + kdu_index, vdu_name, deploy_params, descriptor_config, @@ -3689,10 +3793,16 @@ class NsLcm(LcmBase): self.logger.debug( logging_text + "_deploy_n2vc vnfd_id={}, vdu_id={}".format(vnfd_id, vdu_id) ) + + charm_name = "" + get_charm_name = False if "execution-environment-list" in descriptor_config: ee_list = descriptor_config.get("execution-environment-list", []) elif "juju" in descriptor_config: ee_list = [descriptor_config] # ns charms + if "execution-environment-list" not in descriptor_config: + # charm name is only required for ns charms + get_charm_name = True else: # other types as script are not supported ee_list = [] @@ -3706,6 +3816,8 @@ class NsLcm(LcmBase): ee_descriptor_id = ee_item.get("id") if ee_item.get("juju"): vca_name = ee_item["juju"].get("charm") + if get_charm_name: + charm_name = self.find_charm_name(db_nsr, str(vca_name)) vca_type = ( "lxc_proxy_charm" if ee_item["juju"].get("charm") is not None @@ -3764,6 +3876,7 @@ class NsLcm(LcmBase): "vdu_name": vdu_name, "type": vca_type, "ee_descriptor_id": ee_descriptor_id, + "charm_name": charm_name, } vca_index += 1 @@ -3791,6 +3904,7 @@ class NsLcm(LcmBase): vdu_id=vdu_id, kdu_name=kdu_name, vdu_index=vdu_index, + kdu_index=kdu_index, deploy_params=deploy_params, config_descriptor=descriptor_config, base_folder=base_folder, @@ -4162,7 +4276,7 @@ class NsLcm(LcmBase): try: await self.n2vc.delete_namespace( namespace=namespace, - total_timeout=self.timeout_charm_delete, + total_timeout=self.timeout.charm_delete, vca_id=vca_id, ) except N2VCNotFound: # already deleted. Skip @@ -4377,7 +4491,7 @@ class NsLcm(LcmBase): logging_text = "Task ns={} terminate={} ".format(nsr_id, nslcmop_id) self.logger.debug(logging_text + "Enter") - timeout_ns_terminate = self.timeout_ns_terminate + timeout_ns_terminate = self.timeout.ns_terminate db_nsr = None db_nslcmop = None operation_params = None @@ -4503,7 +4617,7 @@ class NsLcm(LcmBase): error_list = await self._wait_for_tasks( logging_text, tasks_dict_info, - min(self.timeout_charm_delete, timeout_ns_terminate), + min(self.timeout.charm_delete, timeout_ns_terminate), stage, nslcmop_id, ) @@ -4521,12 +4635,19 @@ class NsLcm(LcmBase): task_delete_ee = asyncio.ensure_future( asyncio.wait_for( self._delete_all_N2VC(db_nsr=db_nsr, vca_id=vca_id), - timeout=self.timeout_charm_delete, + timeout=self.timeout.charm_delete, ) ) # task_delete_ee = asyncio.ensure_future(self.n2vc.delete_namespace(namespace="." + nsr_id)) tasks_dict_info[task_delete_ee] = "Terminating all VCA" + # Delete Namespace and Certificates if necessary + if check_helm_ee_in_ns(list(db_vnfds_from_member_index.values())): + await self.vca_map["helm-v3"].delete_tls_certificate( + certificate_name=db_nslcmop["nsInstanceId"], + ) + # TODO: Delete namespace + # Delete from k8scluster stage[1] = "Deleting KDUs." self.logger.debug(logging_text + stage[1]) @@ -4560,7 +4681,7 @@ class NsLcm(LcmBase): # remove from RO stage[1] = "Deleting ns from VIM." - if self.ng_ro: + if self.ro_config.ng: task_delete_ro = asyncio.ensure_future( self._terminate_ng_ro( logging_text, nsr_deployed, nsr_id, nslcmop_id, stage @@ -4923,21 +5044,19 @@ class NsLcm(LcmBase): ee_id=ee_id, primitive_name=primitive, params_dict=primitive_params, - progress_timeout=self.timeout_progress_primitive, - total_timeout=self.timeout_primitive, + progress_timeout=self.timeout.progress_primitive, + total_timeout=self.timeout.primitive, db_dict=db_dict, vca_id=vca_id, vca_type=vca_type, ), - timeout=timeout or self.timeout_primitive, + timeout=timeout or self.timeout.primitive, ) # execution was OK break except asyncio.CancelledError: raise - except Exception as e: # asyncio.TimeoutError - if isinstance(e, asyncio.TimeoutError): - e = "Timeout" + except Exception as e: retries -= 1 if retries >= 0: self.logger.debug( @@ -4948,7 +5067,11 @@ class NsLcm(LcmBase): # wait and retry await asyncio.sleep(retries_interval, loop=self.loop) else: - return "FAILED", str(e) + if isinstance(e, asyncio.TimeoutError): + e = N2VCException( + message="Timed out waiting for action to complete" + ) + return "FAILED", getattr(e, "message", repr(e)) return "COMPLETED", output @@ -5035,7 +5158,7 @@ class NsLcm(LcmBase): primitive = db_nslcmop["operationParams"]["primitive"] primitive_params = db_nslcmop["operationParams"]["primitive_params"] timeout_ns_action = db_nslcmop["operationParams"].get( - "timeout_ns_action", self.timeout_primitive + "timeout_ns_action", self.timeout.primitive ) if vnf_index: @@ -5184,12 +5307,19 @@ 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, @@ -5283,7 +5413,7 @@ class NsLcm(LcmBase): ) self.logger.debug( logging_text - + " task Done with result {} {}".format( + + "Done with result {} {}".format( nslcmop_operation_state, detailed_status ) ) @@ -5381,33 +5511,39 @@ class NsLcm(LcmBase): "member-vnf-index": member_vnf_index, "type": "delete", "vdu_index": count_index, - }) + } + ) scaling_info["vdu-delete"][vdu["vdu-id-ref"]] = count_index scaling_info["vdu"].append( { "name": vdu.get("name") or vdu.get("vdu-name"), "vdu_id": vdu["vdu-id-ref"], "interface": [], - }) + } + ) for interface in vdu["interfaces"]: scaling_info["vdu"][index]["interface"].append( { "name": interface["name"], "ip_address": interface["ip-address"], "mac_address": interface.get("mac-address"), - }) + } + ) self.logger.info("NS update scaling info{}".format(scaling_info)) stage[2] = "Terminating VDUs" if scaling_info.get("vdu-delete"): # scale_process = "RO" - if self.ro_config.get("ng"): + if self.ro_config.ng: await self._scale_ng_ro( - logging_text, db_nsr, update_db_nslcmops, db_vnfr, scaling_info, stage + logging_text, + db_nsr, + update_db_nslcmops, + db_vnfr, + scaling_info, + stage, ) - async def remove_vnf( - self, nsr_id, nslcmop_id, vnf_instance_id - ): + async def remove_vnf(self, nsr_id, nslcmop_id, vnf_instance_id): """This method is to Remove VNF instances from NS. Args: @@ -5426,7 +5562,9 @@ class NsLcm(LcmBase): if check_vnfr_count > 1: stage = ["", "", ""] step = "Getting nslcmop from database" - self.logger.debug(step + " after having waited for previous tasks to be completed") + self.logger.debug( + step + " after having waited for previous tasks to be completed" + ) # db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id}) db_nsr = self.db.get_one("nsrs", {"_id": nsr_id}) db_vnfr = self.db.get_one("vnfrs", {"_id": vnf_instance_id}) @@ -5435,19 +5573,31 @@ class NsLcm(LcmBase): "vnfrs", {"member-vnf-index-ref": member_vnf_index, "nsr-id-ref": nsr_id}) """ update_db_nslcmops = self.db.get_one("nslcmops", {"_id": nslcmop_id}) - await self.terminate_vdus(db_vnfr, member_vnf_index, db_nsr, update_db_nslcmops, stage, logging_text) + await self.terminate_vdus( + db_vnfr, + member_vnf_index, + db_nsr, + update_db_nslcmops, + stage, + logging_text, + ) constituent_vnfr = db_nsr.get("constituent-vnfr-ref") constituent_vnfr.remove(db_vnfr.get("_id")) - db_nsr_update["constituent-vnfr-ref"] = db_nsr.get("constituent-vnfr-ref") + db_nsr_update["constituent-vnfr-ref"] = db_nsr.get( + "constituent-vnfr-ref" + ) self.update_db_2("nsrs", nsr_id, db_nsr_update) self.db.del_one("vnfrs", {"_id": db_vnfr.get("_id")}) self.update_db_2("nsrs", nsr_id, db_nsr_update) return "COMPLETED", "Done" else: step = "Terminate VNF Failed with" - raise LcmException("{} Cannot terminate the last VNF in this NS.".format( - vnf_instance_id)) + raise LcmException( + "{} Cannot terminate the last VNF in this NS.".format( + vnf_instance_id + ) + ) except (LcmException, asyncio.CancelledError): raise except Exception as e: @@ -5455,7 +5605,12 @@ class NsLcm(LcmBase): return "FAILED", "Error removing VNF {}".format(e) async def _ns_redeploy_vnf( - self, nsr_id, nslcmop_id, db_vnfd, db_vnfr, db_nsr, + self, + nsr_id, + nslcmop_id, + db_vnfd, + db_vnfr, + db_nsr, ): """This method updates and redeploys VNF instances @@ -5478,7 +5633,14 @@ class NsLcm(LcmBase): # Terminate old VNF resources update_db_nslcmops = self.db.get_one("nslcmops", {"_id": nslcmop_id}) - await self.terminate_vdus(db_vnfr, member_vnf_index, db_nsr, update_db_nslcmops, stage, logging_text) + await self.terminate_vdus( + db_vnfr, + member_vnf_index, + db_nsr, + update_db_nslcmops, + stage, + logging_text, + ) # old_vnfd_id = db_vnfr["vnfd-id"] # new_db_vnfd = self.db.get_one("vnfds", {"_id": vnfd_id}) @@ -5499,10 +5661,16 @@ class NsLcm(LcmBase): new_vdur = update_db_nslcmops["operationParams"]["newVdur"] # new_vdur = self._create_vdur_descriptor_from_vnfd(db_nsd, db_vnfd, old_db_vnfd, vnfd_id, db_nsr, member_vnf_index) # new_vnfr_update = {"vnfd-ref": new_vnfd_ref, "vnfd-id": new_vnfd_id, "connection-point": new_vnfr_cp, "vdur": new_vdur, "ip-address": ""} - new_vnfr_update = {"revision": latest_vnfd_revision, "connection-point": new_vnfr_cp, "vdur": new_vdur, "ip-address": ""} + new_vnfr_update = { + "revision": latest_vnfd_revision, + "connection-point": new_vnfr_cp, + "vdur": new_vdur, + "ip-address": "", + } self.update_db_2("vnfrs", db_vnfr["_id"], new_vnfr_update) updated_db_vnfr = self.db.get_one( - "vnfrs", {"member-vnf-index-ref": member_vnf_index, "nsr-id-ref": nsr_id} + "vnfrs", + {"member-vnf-index-ref": member_vnf_index, "nsr-id-ref": nsr_id}, ) # Instantiate new VNF resources @@ -5514,9 +5682,7 @@ class NsLcm(LcmBase): scaling_info["kdu-create"] = {} vdud_instantiate_list = db_vnfd["vdu"] for index, vdud in enumerate(vdud_instantiate_list): - cloud_init_text = self._get_vdu_cloud_init_content( - vdud, db_vnfd - ) + cloud_init_text = self._get_vdu_cloud_init_content(vdud, db_vnfd) if cloud_init_text: additional_params = ( self._get_vdu_additional_params(updated_db_vnfr, vdud["id"]) @@ -5545,11 +5711,17 @@ class NsLcm(LcmBase): } ) scaling_info["vdu-create"][vdud["id"]] = count_index - if self.ro_config.get("ng"): + if self.ro_config.ng: self.logger.debug( - "New Resources to be deployed: {}".format(scaling_info)) + "New Resources to be deployed: {}".format(scaling_info) + ) await self._scale_ng_ro( - logging_text, db_nsr, update_db_nslcmops, updated_db_vnfr, scaling_info, stage + logging_text, + db_nsr, + update_db_nslcmops, + updated_db_vnfr, + scaling_info, + stage, ) return "COMPLETED", "Done" except (LcmException, asyncio.CancelledError): @@ -5585,7 +5757,7 @@ class NsLcm(LcmBase): path=path, charm_id=charm_id, charm_type=charm_type, - timeout=timeout or self.timeout_ns_update, + timeout=timeout or self.timeout.ns_update, ) if output: @@ -5693,7 +5865,8 @@ class NsLcm(LcmBase): current_charm_artifact_path, target_charm_artifact_path, charm_artifact_paths, - ) = ([], [], []) + helm_artifacts, + ) = ([], [], [], []) step = "Checking if revision has changed in VNFD" if current_vnf_revision != latest_vnfd_revision: @@ -5715,24 +5888,34 @@ class NsLcm(LcmBase): step = ( "Get the charm-type, charm-id, ee-id if there is deployed VCA" ) - base_folder = latest_vnfd["_admin"]["storage"] + current_base_folder = current_vnfd["_admin"]["storage"] + latest_base_folder = latest_vnfd["_admin"]["storage"] - for charm_index, charm_deployed in enumerate( + for vca_index, vca_deployed in enumerate( get_iterable(nsr_deployed, "VCA") ): vnf_index = db_vnfr.get("member-vnf-index-ref") # Getting charm-id and charm-type - if charm_deployed.get("member-vnf-index") == vnf_index: - charm_id = self.get_vca_id(db_vnfr, db_nsr) - charm_type = charm_deployed.get("type") + if vca_deployed.get("member-vnf-index") == vnf_index: + vca_id = self.get_vca_id(db_vnfr, db_nsr) + vca_type = vca_deployed.get("type") + vdu_count_index = vca_deployed.get("vdu_count_index") # Getting ee-id - ee_id = charm_deployed.get("ee_id") + ee_id = vca_deployed.get("ee_id") step = "Getting descriptor config" + if current_vnfd.get("kdu"): + + search_key = "kdu_name" + else: + search_key = "vnfd_id" + + entity_id = vca_deployed.get(search_key) + descriptor_config = get_configuration( - current_vnfd, current_vnfd["id"] + current_vnfd, entity_id ) if "execution-environment-list" in descriptor_config: @@ -5752,20 +5935,52 @@ class NsLcm(LcmBase): step = "Setting Charm artifact paths" current_charm_artifact_path.append( get_charm_artifact_path( - base_folder, + current_base_folder, charm_name, - charm_type, + vca_type, current_vnf_revision, ) ) target_charm_artifact_path.append( get_charm_artifact_path( - base_folder, + latest_base_folder, charm_name, - charm_type, + vca_type, latest_vnfd_revision, ) ) + elif ee_item.get("helm-chart"): + # add chart to list and all parameters + step = "Getting helm chart name" + chart_name = ee_item.get("helm-chart") + if ( + ee_item.get("helm-version") + and ee_item.get("helm-version") == "v2" + ): + vca_type = "helm" + else: + vca_type = "helm-v3" + step = "Setting Helm chart artifact paths" + + helm_artifacts.append( + { + "current_artifact_path": get_charm_artifact_path( + current_base_folder, + chart_name, + vca_type, + current_vnf_revision, + ), + "target_artifact_path": get_charm_artifact_path( + latest_base_folder, + chart_name, + vca_type, + latest_vnfd_revision, + ), + "ee_id": ee_id, + "vca_index": vca_index, + "vdu_index": vdu_count_index, + } + ) charm_artifact_paths = zip( current_charm_artifact_path, target_charm_artifact_path @@ -5791,15 +6006,8 @@ class NsLcm(LcmBase): # based on new descriptor step = "Redeploying VNF" member_vnf_index = db_vnfr["member-vnf-index-ref"] - ( - result, - detailed_status - ) = await self._ns_redeploy_vnf( - nsr_id, - nslcmop_id, - latest_vnfd, - db_vnfr, - db_nsr + (result, detailed_status) = await self._ns_redeploy_vnf( + nsr_id, nslcmop_id, latest_vnfd, db_vnfr, db_nsr ) if result == "FAILED": nslcmop_operation_state = result @@ -5841,8 +6049,8 @@ class NsLcm(LcmBase): detailed_status, ) = await self._ns_charm_upgrade( ee_id=ee_id, - charm_id=charm_id, - charm_type=charm_type, + charm_id=vca_id, + charm_type=vca_type, path=self.fs.path + target_charm_path, timeout=timeout_seconds, ) @@ -5865,6 +6073,121 @@ class NsLcm(LcmBase): detailed_status = "Done" db_nslcmop_update["detailed-status"] = "Done" + # helm base EE + for item in helm_artifacts: + if not ( + item["current_artifact_path"] + and item["target_artifact_path"] + and self.check_charm_hash_changed( + item["current_artifact_path"], + item["target_artifact_path"], + ) + ): + continue + db_update_entry = "_admin.deployed.VCA.{}.".format( + item["vca_index"] + ) + vnfr_id = db_vnfr["_id"] + osm_config = {"osm": {"ns_id": nsr_id, "vnf_id": vnfr_id}} + db_dict = { + "collection": "nsrs", + "filter": {"_id": nsr_id}, + "path": db_update_entry, + } + vca_type, namespace, helm_id = get_ee_id_parts(item["ee_id"]) + await self.vca_map[vca_type].upgrade_execution_environment( + namespace=namespace, + helm_id=helm_id, + db_dict=db_dict, + config=osm_config, + artifact_path=item["target_artifact_path"], + vca_type=vca_type, + ) + vnf_id = db_vnfr.get("vnfd-ref") + config_descriptor = get_configuration(latest_vnfd, vnf_id) + self.logger.debug("get ssh key block") + rw_mgmt_ip = None + if deep_get( + config_descriptor, + ("config-access", "ssh-access", "required"), + ): + # Needed to inject a ssh key + user = deep_get( + config_descriptor, + ("config-access", "ssh-access", "default-user"), + ) + step = ( + "Install configuration Software, getting public ssh key" + ) + pub_key = await self.vca_map[ + vca_type + ].get_ee_ssh_public__key( + ee_id=ee_id, db_dict=db_dict, vca_id=vca_id + ) + + step = ( + "Insert public key into VM user={} ssh_key={}".format( + user, pub_key + ) + ) + self.logger.debug(logging_text + step) + + # wait for RO (ip-address) Insert pub_key into VM + rw_mgmt_ip = await self.wait_vm_up_insert_key_ro( + logging_text, + nsr_id, + vnfr_id, + None, + item["vdu_index"], + user=user, + pub_key=pub_key, + ) + + initial_config_primitive_list = config_descriptor.get( + "initial-config-primitive" + ) + config_primitive = next( + ( + p + for p in initial_config_primitive_list + if p["name"] == "config" + ), + None, + ) + if not config_primitive: + continue + + deploy_params = {"OSM": get_osm_params(db_vnfr)} + if rw_mgmt_ip: + deploy_params["rw_mgmt_ip"] = rw_mgmt_ip + if db_vnfr.get("additionalParamsForVnf"): + deploy_params.update( + parse_yaml_strings( + db_vnfr["additionalParamsForVnf"].copy() + ) + ) + primitive_params_ = self._map_primitive_params( + config_primitive, {}, deploy_params + ) + + step = "execute primitive '{}' params '{}'".format( + config_primitive["name"], primitive_params_ + ) + self.logger.debug(logging_text + step) + await self.vca_map[vca_type].exec_primitive( + ee_id=ee_id, + primitive_name=config_primitive["name"], + params_dict=primitive_params_, + db_dict=db_dict, + vca_id=vca_id, + vca_type=vca_type, + ) + + step = "Updating policies" + member_vnf_index = db_vnfr["member-vnf-index-ref"] + detailed_status = "Done" + db_nslcmop_update["detailed-status"] = "Done" + # If nslcmop_operation_state is None, so any operation is not failed. if not nslcmop_operation_state: nslcmop_operation_state = "COMPLETED" @@ -5886,7 +6209,9 @@ class NsLcm(LcmBase): db_vnfr = self.db.get_one("vnfrs", {"_id": vnf_instance_id}) member_vnf_index = db_vnfr["member-vnf-index-ref"] step = "Removing VNF" - (result, detailed_status) = await self.remove_vnf(nsr_id, nslcmop_id, vnf_instance_id) + (result, detailed_status) = await self.remove_vnf( + nsr_id, nslcmop_id, vnf_instance_id + ) if result == "FAILED": nslcmop_operation_state = result error_description_nslcmop = detailed_status @@ -5901,6 +6226,32 @@ class NsLcm(LcmBase): ) ) + elif update_type == "OPERATE_VNF": + vnf_id = db_nslcmop["operationParams"]["operateVnfData"][ + "vnfInstanceId" + ] + operation_type = db_nslcmop["operationParams"]["operateVnfData"][ + "changeStateTo" + ] + additional_param = db_nslcmop["operationParams"]["operateVnfData"][ + "additionalParam" + ] + (result, detailed_status) = await self.rebuild_start_stop( + nsr_id, nslcmop_id, vnf_id, additional_param, operation_type + ) + if result == "FAILED": + nslcmop_operation_state = result + error_description_nslcmop = detailed_status + db_nslcmop_update["detailed-status"] = detailed_status + if not nslcmop_operation_state: + nslcmop_operation_state = "COMPLETED" + self.logger.debug( + logging_text + + " task Done with result {} {}".format( + nslcmop_operation_state, detailed_status + ) + ) + # If nslcmop_operation_state is None, so any operation is not failed. # All operations are executed in overall. if not nslcmop_operation_state: @@ -6599,7 +6950,7 @@ class NsLcm(LcmBase): scaling_in=True, vca_id=vca_id, ), - timeout=self.timeout_charm_delete, + timeout=self.timeout.charm_delete, ) ) tasks_dict_info[task] = "Terminating VCA {}".format( @@ -6619,7 +6970,7 @@ class NsLcm(LcmBase): logging_text, tasks_dict_info, min( - self.timeout_charm_delete, self.timeout_ns_terminate + self.timeout.charm_delete, self.timeout.ns_terminate ), stage, nslcmop_id, @@ -6641,7 +6992,7 @@ class NsLcm(LcmBase): # SCALE RO - BEGIN if scaling_info.get("vdu-create") or scaling_info.get("vdu-delete"): scale_process = "RO" - if self.ro_config.get("ng"): + if self.ro_config.ng: await self._scale_ng_ro( logging_text, db_nsr, db_nslcmop, db_vnfr, scaling_info, stage ) @@ -6695,6 +7046,7 @@ class NsLcm(LcmBase): vdu_id = None vdu_name = None kdu_name = None + kdu_index = None self._deploy_n2vc( logging_text=logging_text + "member_vnf_index={} ".format(member_vnf_index), @@ -6706,6 +7058,7 @@ class NsLcm(LcmBase): vnfd_id=vnfd_id, vdu_id=vdu_id, kdu_name=kdu_name, + kdu_index=kdu_index, member_vnf_index=member_vnf_index, vdu_index=vdu_index, vdu_name=vdu_name, @@ -6732,6 +7085,7 @@ class NsLcm(LcmBase): if descriptor_config: vdu_name = None kdu_name = None + kdu_index = None stage[ 1 ] = "Scaling member_vnf_index={}, vdu_id={}, vdu_index={} ".format( @@ -6754,6 +7108,7 @@ class NsLcm(LcmBase): kdu_name=kdu_name, member_vnf_index=member_vnf_index, vdu_index=vdu_index, + kdu_index=kdu_index, vdu_name=vdu_name, deploy_params=deploy_params_vdu, descriptor_config=descriptor_config, @@ -6936,7 +7291,7 @@ class NsLcm(LcmBase): exc = await self._wait_for_tasks( logging_text, tasks_dict_info, - self.timeout_ns_deploy, + self.timeout.ns_deploy, stage, nslcmop_id, nsr_id=nsr_id, @@ -7051,23 +7406,27 @@ class NsLcm(LcmBase): primitive_name=terminate_config_primitive["name"], params=primitive_params_, db_dict=db_dict, + total_timeout=self.timeout.primitive, vca_id=vca_id, ), - timeout=600, + timeout=self.timeout.primitive + * self.timeout.primitive_outer_factor, ) await asyncio.wait_for( self.k8scluster_map[k8s_cluster_type].scale( - kdu_instance, - scale, - kdu_scaling_info["resource-name"], + kdu_instance=kdu_instance, + scale=scale, + resource_name=kdu_scaling_info["resource-name"], + total_timeout=self.timeout.scale_on_error, vca_id=vca_id, cluster_uuid=cluster_uuid, kdu_model=kdu_model, atomic=True, db_dict=db_dict, ), - timeout=self.timeout_vca_on_error, + timeout=self.timeout.scale_on_error + * self.timeout.scale_on_error_outer_factor, ) if kdu_scaling_info["type"] == "create": @@ -7142,7 +7501,7 @@ class NsLcm(LcmBase): n2vc_key_list, stage=stage, start_deploy=time(), - timeout_ns_deploy=self.timeout_ns_deploy, + timeout_ns_deploy=self.timeout.ns_deploy, ) if vdu_scaling_info.get("vdu-delete"): self.scale_vnfr( @@ -7150,8 +7509,42 @@ class NsLcm(LcmBase): ) async def extract_prometheus_scrape_jobs( - self, ee_id, artifact_path, ee_config_descriptor, vnfr_id, nsr_id, target_ip - ): + self, + ee_id: str, + artifact_path: str, + ee_config_descriptor: dict, + vnfr_id: str, + nsr_id: str, + target_ip: str, + vnf_member_index: str = "", + vdu_id: str = "", + vdu_index: int = None, + kdu_name: str = "", + kdu_index: int = None, + ) -> dict: + """Method to extract prometheus scrape jobs from EE's Prometheus template job file + This method will wait until the corresponding VDU or KDU is fully instantiated + + Args: + ee_id (str): Execution Environment ID + artifact_path (str): Path where the EE's content is (including the Prometheus template file) + ee_config_descriptor (dict): Execution Environment's configuration descriptor + vnfr_id (str): VNFR ID where this EE applies + nsr_id (str): NSR ID where this EE applies + target_ip (str): VDU/KDU instance IP address + vnf_member_index (str, optional): VNF index where this EE applies. Defaults to "". + vdu_id (str, optional): VDU ID where this EE applies. Defaults to "". + vdu_index (int, optional): VDU index where this EE applies. Defaults to None. + kdu_name (str, optional): KDU name where this EE applies. Defaults to "". + kdu_index (int, optional): KDU index where this EE applies. Defaults to None. + + Raises: + LcmException: When the VDU or KDU instance was not found in an hour + + Returns: + _type_: Prometheus jobs + """ + self.logger.debug(f"KDU: {kdu_name}; KDU INDEX: {kdu_index}") # look if exist a file called 'prometheus*.j2' and artifact_content = self.fs.dir_ls(artifact_path) job_file = next( @@ -7167,6 +7560,52 @@ class NsLcm(LcmBase): with self.fs.file_open((artifact_path, job_file), "r") as f: job_data = f.read() + vdur_name = "" + kdur_name = "" + for r in range(360): + db_vnfr = self.db.get_one("vnfrs", {"_id": vnfr_id}) + if vdu_id and vdu_index is not None: + vdur = next( + ( + x + for x in get_iterable(db_vnfr, "vdur") + if ( + x.get("vdu-id-ref") == vdu_id + and x.get("count-index") == vdu_index + ) + ), + {}, + ) + if vdur.get("name"): + vdur_name = vdur.get("name") + break + if kdu_name and kdu_index is not None: + kdur = next( + ( + x + for x in get_iterable(db_vnfr, "kdur") + if ( + x.get("kdu-name") == kdu_name + and x.get("count-index") == kdu_index + ) + ), + {}, + ) + if kdur.get("name"): + kdur_name = kdur.get("name") + break + + await asyncio.sleep(10, loop=self.loop) + else: + if vdu_id and vdu_index is not None: + raise LcmException( + f"Timeout waiting VDU with name={vdu_id} and index={vdu_index} to be intantiated" + ) + if kdu_name and kdu_index is not None: + raise LcmException( + f"Timeout waiting KDU with name={kdu_name} and index={kdu_index} to be intantiated" + ) + # TODO get_service _, _, service = ee_id.partition(".") # remove prefix "namespace." host_name = "{}-{}".format(service, ee_config_descriptor["metric-service"]) @@ -7177,6 +7616,10 @@ class NsLcm(LcmBase): "TARGET_IP": target_ip, "EXPORTER_POD_IP": host_name, "EXPORTER_POD_PORT": host_port, + "NSR_ID": nsr_id, + "VNF_MEMBER_INDEX": vnf_member_index, + "VDUR_NAME": vdur_name, + "KDUR_NAME": kdur_name, } job_list = parse_job(job_data, variables) # ensure job_name is using the vnfr_id. Adding the metadata nsr_id @@ -7190,6 +7633,93 @@ class NsLcm(LcmBase): job["vnfr_id"] = vnfr_id return job_list + async def rebuild_start_stop( + self, nsr_id, nslcmop_id, vnf_id, additional_param, operation_type + ): + logging_text = "Task ns={} {}={} ".format(nsr_id, operation_type, nslcmop_id) + self.logger.info(logging_text + "Enter") + stage = ["Preparing the environment", ""] + # database nsrs record + db_nsr_update = {} + vdu_vim_name = None + vim_vm_id = None + # in case of error, indicates what part of scale was failed to put nsr at error status + start_deploy = time() + try: + 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( + 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" + self.logger.info(stage[1]) + await self.lcm_tasks.waitfor_related_HA("ns", "nslcmops", nslcmop_id) + + stage[1] = "Reading from database." + self.logger.info(stage[1]) + self._write_ns_status( + nsr_id=nsr_id, + ns_state=None, + current_operation=operation_type.upper(), + current_operation_id=nslcmop_id, + ) + self._write_op_status(op_id=nslcmop_id, stage=stage, queuePosition=0) + + # read from db: ns + stage[1] = "Getting nsr={} from db.".format(nsr_id) + db_nsr_update["operational-status"] = operation_type + self.update_db_2("nsrs", nsr_id, db_nsr_update) + # Payload for RO + desc = { + operation_type: { + "vim_vm_id": vim_vm_id, + "vnf_id": vnf_id, + "vdu_index": additional_param["count-index"], + "vdu_id": vdur["id"], + "target_vim": target_vim, + "vim_account_id": vim_account_id, + } + } + stage[1] = "Sending rebuild request to RO... {}".format(desc) + self._write_op_status(op_id=nslcmop_id, stage=stage, queuePosition=0) + self.logger.info("ro nsr id: {}".format(nsr_id)) + result_dict = await self.RO.operate(nsr_id, desc, operation_type) + 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, + None, + "start_stop_rebuild", + ) + return "COMPLETED", "Done" + 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(stage)) + 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 + ) + return "FAILED", "Error in operate VNF {}".format(exc) + def get_vca_cloud_and_credentials(self, vim_account_id: str) -> (str, str): """ Get VCA Cloud and VCA Cloud Credentials for the VIM account @@ -7260,7 +7790,12 @@ class NsLcm(LcmBase): 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_migrate + nsr_id, + action_id, + nslcmop_id, + start_deploy, + self.timeout.migrate, + operation="migrate", ) except (ROclient.ROClientException, DbException, LcmException) as e: self.logger.error("Exit Exception {}".format(e)) @@ -7309,3 +7844,1069 @@ class NsLcm(LcmBase): ) self.logger.debug(logging_text + "Exit") self.lcm_tasks.remove("ns", nsr_id, nslcmop_id, "ns_migrate") + + async def heal(self, nsr_id, nslcmop_id): + """ + Heal NS + + :param nsr_id: ns instance to heal + :param nslcmop_id: operation to run + :return: + """ + + # 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={} heal={} ".format(nsr_id, nslcmop_id) + stage = ["", "", ""] + tasks_dict_info = {} + # ^ stage, step, VIM progress + self.logger.debug(logging_text + "Enter") + # get all needed from database + db_nsr = None + db_nslcmop_update = {} + db_nsr_update = {} + db_vnfrs = {} # vnf's info indexed by _id + exc = None + old_operational_status = "" + old_config_status = "" + nsi_id = None + 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="HEALING", + 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}) + + step = "Getting nsr from database" + db_nsr = self.db.get_one("nsrs", {"_id": nsr_id}) + old_operational_status = db_nsr["operational-status"] + old_config_status = db_nsr["config-status"] + + db_nsr_update = { + "_admin.deployed.RO.operational-status": "healing", + } + self.update_db_2("nsrs", nsr_id, db_nsr_update) + + step = "Sending heal order to VIM" + await self.heal_RO( + logging_text=logging_text, + nsr_id=nsr_id, + db_nslcmop=db_nslcmop, + stage=stage, + ) + # VCA tasks + # read from db: nsd + stage[1] = "Getting nsd={} from db.".format(db_nsr["nsd-id"]) + self.logger.debug(logging_text + stage[1]) + nsd = self.db.get_one("nsds", {"_id": db_nsr["nsd-id"]}) + self.fs.sync(db_nsr["nsd-id"]) + db_nsr["nsd"] = nsd + # read from db: vnfr's of this ns + step = "Getting vnfrs from db" + db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id}) + for vnfr in db_vnfrs_list: + db_vnfrs[vnfr["_id"]] = vnfr + self.logger.debug("ns.heal db_vnfrs={}".format(db_vnfrs)) + + # Check for each target VNF + target_list = db_nslcmop.get("operationParams", {}).get("healVnfData", {}) + for target_vnf in target_list: + # Find this VNF in the list from DB + vnfr_id = target_vnf.get("vnfInstanceId", None) + if vnfr_id: + db_vnfr = db_vnfrs[vnfr_id] + vnfd_id = db_vnfr.get("vnfd-id") + vnfd_ref = db_vnfr.get("vnfd-ref") + vnfd = self.db.get_one("vnfds", {"_id": vnfd_id}) + base_folder = vnfd["_admin"]["storage"] + vdu_id = None + vdu_index = 0 + vdu_name = None + kdu_name = None + nsi_id = None # TODO put nsi_id when this nsr belongs to a NSI + member_vnf_index = db_vnfr.get("member-vnf-index-ref") + + # Check each target VDU and deploy N2VC + 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"): + deploy_params_vdu["run-day1"] = target_vnf[ + "additionalParams" + ].get("run-day1") + vdu_name = target_vdu.get("vdu-id", None) + # TODO: Get vdu_id from vdud. + vdu_id = vdu_name + # For multi instance VDU count-index is mandatory + # For single session VDU count-indes is 0 + vdu_index = target_vdu.get("count-index", 0) + + # n2vc_redesign STEP 3 to 6 Deploy N2VC + stage[1] = "Deploying Execution Environments." + self.logger.debug(logging_text + stage[1]) + + # VNF Level charm. Normal case when proxy charms. + # If target instance is management machine continue with actions: recreate EE for native charms or reinject juju key for proxy charms. + descriptor_config = get_configuration(vnfd, vnfd_ref) + if descriptor_config: + # Continue if healed machine is management machine + vnf_ip_address = db_vnfr.get("ip-address") + target_instance = None + for instance in db_vnfr.get("vdur", None): + if ( + instance["vdu-name"] == vdu_name + and instance["count-index"] == vdu_index + ): + target_instance = instance + break + if vnf_ip_address == target_instance.get("ip-address"): + self._heal_n2vc( + logging_text=logging_text + + "member_vnf_index={}, vdu_name={}, vdu_index={} ".format( + member_vnf_index, vdu_name, vdu_index + ), + db_nsr=db_nsr, + db_vnfr=db_vnfr, + nslcmop_id=nslcmop_id, + nsr_id=nsr_id, + nsi_id=nsi_id, + vnfd_id=vnfd_ref, + vdu_id=None, + kdu_name=None, + member_vnf_index=member_vnf_index, + vdu_index=0, + vdu_name=None, + deploy_params=deploy_params_vdu, + descriptor_config=descriptor_config, + base_folder=base_folder, + task_instantiation_info=tasks_dict_info, + stage=stage, + ) + + # VDU Level charm. Normal case with native charms. + descriptor_config = get_configuration(vnfd, vdu_name) + if descriptor_config: + self._heal_n2vc( + logging_text=logging_text + + "member_vnf_index={}, vdu_name={}, vdu_index={} ".format( + member_vnf_index, vdu_name, vdu_index + ), + db_nsr=db_nsr, + db_vnfr=db_vnfr, + nslcmop_id=nslcmop_id, + nsr_id=nsr_id, + nsi_id=nsi_id, + vnfd_id=vnfd_ref, + vdu_id=vdu_id, + kdu_name=kdu_name, + member_vnf_index=member_vnf_index, + vdu_index=vdu_index, + vdu_name=vdu_name, + deploy_params=deploy_params_vdu, + descriptor_config=descriptor_config, + base_folder=base_folder, + task_instantiation_info=tasks_dict_info, + stage=stage, + ) + + except ( + ROclient.ROClientException, + DbException, + LcmException, + NgRoException, + ) as e: + self.logger.error(logging_text + "Exit Exception {}".format(e)) + exc = e + except asyncio.CancelledError: + self.logger.error( + logging_text + "Cancelled Exception while '{}'".format(step) + ) + exc = "Operation was cancelled" + except Exception as e: + exc = traceback.format_exc() + self.logger.critical( + logging_text + "Exit Exception {} {}".format(type(e).__name__, e), + exc_info=True, + ) + finally: + if tasks_dict_info: + stage[1] = "Waiting for healing pending tasks." + self.logger.debug(logging_text + stage[1]) + exc = await self._wait_for_tasks( + logging_text, + tasks_dict_info, + self.timeout.ns_deploy, + stage, + nslcmop_id, + nsr_id=nsr_id, + ) + if exc: + db_nslcmop_update[ + "detailed-status" + ] = error_description_nslcmop = "FAILED {}: {}".format(step, exc) + nslcmop_operation_state = "FAILED" + if db_nsr: + db_nsr_update["operational-status"] = old_operational_status + db_nsr_update["config-status"] = old_config_status + db_nsr_update[ + "detailed-status" + ] = "FAILED healing nslcmop={} {}: {}".format(nslcmop_id, step, exc) + for task, task_name in tasks_dict_info.items(): + if not task.done() or task.cancelled() or task.exception(): + if task_name.startswith(self.task_name_deploy_vca): + # A N2VC task is pending + db_nsr_update["config-status"] = "failed" + else: + # RO task is pending + db_nsr_update["operational-status"] = "failed" + else: + error_description_nslcmop = None + nslcmop_operation_state = "COMPLETED" + db_nslcmop_update["detailed-status"] = "Done" + db_nsr_update["detailed-status"] = "Done" + db_nsr_update["operational-status"] = "running" + db_nsr_update["config-status"] = "configured" + + self._write_op_status( + op_id=nslcmop_id, + stage="", + error_message=error_description_nslcmop, + operation_state=nslcmop_operation_state, + other_update=db_nslcmop_update, + ) + if db_nsr: + self._write_ns_status( + nsr_id=nsr_id, + ns_state=None, + current_operation="IDLE", + current_operation_id=None, + other_update=db_nsr_update, + ) + + if nslcmop_operation_state: + try: + msg = { + "nsr_id": nsr_id, + "nslcmop_id": nslcmop_id, + "operationState": nslcmop_operation_state, + } + await self.msg.aiowrite("ns", "healed", 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_heal") + + async def heal_RO( + self, + logging_text, + nsr_id, + db_nslcmop, + stage, + ): + """ + Heal at RO + :param logging_text: preffix text to use at logging + :param nsr_id: nsr identity + :param db_nslcmop: database content of ns operation, in this case, 'instantiate' + :param stage: list with 3 items: [general stage, tasks, vim_specific]. This task will write over vim_specific + :return: None or exception + """ + + def get_vim_account(vim_account_id): + nonlocal db_vims + if vim_account_id in db_vims: + return db_vims[vim_account_id] + db_vim = self.db.get_one("vim_accounts", {"_id": vim_account_id}) + db_vims[vim_account_id] = db_vim + return db_vim + + try: + start_heal = time() + ns_params = db_nslcmop.get("operationParams") + if ns_params and ns_params.get("timeout_ns_heal"): + timeout_ns_heal = ns_params["timeout_ns_heal"] + else: + timeout_ns_heal = self.timeout.ns_heal + + db_vims = {} + + nslcmop_id = db_nslcmop["_id"] + target = { + "action_id": nslcmop_id, + } + self.logger.warning( + "db_nslcmop={} and timeout_ns_heal={}".format( + db_nslcmop, timeout_ns_heal + ) + ) + target.update(db_nslcmop.get("operationParams", {})) + + self.logger.debug("Send to RO > nsr_id={} target={}".format(nsr_id, target)) + desc = await self.RO.recreate(nsr_id, target) + self.logger.debug("RO return > {}".format(desc)) + action_id = desc["action_id"] + # waits for RO to complete because Reinjecting juju key at ro can find VM in state Deleted + await self._wait_ng_ro( + nsr_id, + action_id, + nslcmop_id, + start_heal, + timeout_ns_heal, + stage, + operation="healing", + ) + + # Updating NSR + db_nsr_update = { + "_admin.deployed.RO.operational-status": "running", + "detailed-status": " ".join(stage), + } + self.update_db_2("nsrs", nsr_id, db_nsr_update) + self._write_op_status(nslcmop_id, stage) + self.logger.debug( + logging_text + "ns healed at RO. RO_id={}".format(action_id) + ) + + except Exception as e: + stage[2] = "ERROR healing at VIM" + # self.set_vnfr_at_error(db_vnfrs, str(e)) + self.logger.error( + "Error healing at VIM {}".format(e), + exc_info=not isinstance( + e, + ( + ROclient.ROClientException, + LcmException, + DbException, + NgRoException, + ), + ), + ) + raise + + def _heal_n2vc( + self, + logging_text, + db_nsr, + db_vnfr, + nslcmop_id, + nsr_id, + nsi_id, + vnfd_id, + vdu_id, + kdu_name, + member_vnf_index, + vdu_index, + vdu_name, + deploy_params, + descriptor_config, + base_folder, + task_instantiation_info, + stage, + ): + # launch instantiate_N2VC in a asyncio task and register task object + # Look where information of this charm is at database ._admin.deployed.VCA + # if not found, create one entry and update database + # fill db_nsr._admin.deployed.VCA. + + self.logger.debug( + logging_text + "_deploy_n2vc vnfd_id={}, vdu_id={}".format(vnfd_id, vdu_id) + ) + + charm_name = "" + get_charm_name = False + if "execution-environment-list" in descriptor_config: + ee_list = descriptor_config.get("execution-environment-list", []) + elif "juju" in descriptor_config: + ee_list = [descriptor_config] # ns charms + if "execution-environment-list" not in descriptor_config: + # charm name is only required for ns charms + get_charm_name = True + else: # other types as script are not supported + ee_list = [] + + for ee_item in ee_list: + self.logger.debug( + logging_text + + "_deploy_n2vc ee_item juju={}, helm={}".format( + ee_item.get("juju"), ee_item.get("helm-chart") + ) + ) + ee_descriptor_id = ee_item.get("id") + if ee_item.get("juju"): + vca_name = ee_item["juju"].get("charm") + if get_charm_name: + charm_name = self.find_charm_name(db_nsr, str(vca_name)) + vca_type = ( + "lxc_proxy_charm" + if ee_item["juju"].get("charm") is not None + else "native_charm" + ) + if ee_item["juju"].get("cloud") == "k8s": + vca_type = "k8s_proxy_charm" + elif ee_item["juju"].get("proxy") is False: + vca_type = "native_charm" + elif ee_item.get("helm-chart"): + vca_name = ee_item["helm-chart"] + if ee_item.get("helm-version") and ee_item.get("helm-version") == "v2": + vca_type = "helm" + else: + vca_type = "helm-v3" + else: + self.logger.debug( + logging_text + "skipping non juju neither charm configuration" + ) + continue + + vca_index = -1 + for vca_index, vca_deployed in enumerate( + db_nsr["_admin"]["deployed"]["VCA"] + ): + if not vca_deployed: + continue + if ( + vca_deployed.get("member-vnf-index") == member_vnf_index + and vca_deployed.get("vdu_id") == vdu_id + and vca_deployed.get("kdu_name") == kdu_name + and vca_deployed.get("vdu_count_index", 0) == vdu_index + and vca_deployed.get("ee_descriptor_id") == ee_descriptor_id + ): + break + else: + # not found, create one. + target = ( + "ns" if not member_vnf_index else "vnf/{}".format(member_vnf_index) + ) + if vdu_id: + target += "/vdu/{}/{}".format(vdu_id, vdu_index or 0) + elif kdu_name: + target += "/kdu/{}".format(kdu_name) + vca_deployed = { + "target_element": target, + # ^ target_element will replace member-vnf-index, kdu_name, vdu_id ... in a single string + "member-vnf-index": member_vnf_index, + "vdu_id": vdu_id, + "kdu_name": kdu_name, + "vdu_count_index": vdu_index, + "operational-status": "init", # TODO revise + "detailed-status": "", # TODO revise + "step": "initial-deploy", # TODO revise + "vnfd_id": vnfd_id, + "vdu_name": vdu_name, + "type": vca_type, + "ee_descriptor_id": ee_descriptor_id, + "charm_name": charm_name, + } + vca_index += 1 + + # create VCA and configurationStatus in db + db_dict = { + "_admin.deployed.VCA.{}".format(vca_index): vca_deployed, + "configurationStatus.{}".format(vca_index): dict(), + } + self.update_db_2("nsrs", nsr_id, db_dict) + + db_nsr["_admin"]["deployed"]["VCA"].append(vca_deployed) + + self.logger.debug("N2VC > NSR_ID > {}".format(nsr_id)) + self.logger.debug("N2VC > DB_NSR > {}".format(db_nsr)) + self.logger.debug("N2VC > VCA_DEPLOYED > {}".format(vca_deployed)) + + # Launch task + task_n2vc = asyncio.ensure_future( + self.heal_N2VC( + logging_text=logging_text, + vca_index=vca_index, + nsi_id=nsi_id, + db_nsr=db_nsr, + db_vnfr=db_vnfr, + vdu_id=vdu_id, + kdu_name=kdu_name, + vdu_index=vdu_index, + deploy_params=deploy_params, + config_descriptor=descriptor_config, + base_folder=base_folder, + nslcmop_id=nslcmop_id, + stage=stage, + vca_type=vca_type, + vca_name=vca_name, + ee_config_descriptor=ee_item, + ) + ) + self.lcm_tasks.register( + "ns", + nsr_id, + nslcmop_id, + "instantiate_N2VC-{}".format(vca_index), + task_n2vc, + ) + task_instantiation_info[ + task_n2vc + ] = self.task_name_deploy_vca + " {}.{}".format( + member_vnf_index or "", vdu_id or "" + ) + + async def heal_N2VC( + self, + logging_text, + vca_index, + nsi_id, + db_nsr, + db_vnfr, + vdu_id, + kdu_name, + vdu_index, + config_descriptor, + deploy_params, + base_folder, + nslcmop_id, + stage, + vca_type, + vca_name, + ee_config_descriptor, + ): + nsr_id = db_nsr["_id"] + db_update_entry = "_admin.deployed.VCA.{}.".format(vca_index) + vca_deployed_list = db_nsr["_admin"]["deployed"]["VCA"] + vca_deployed = db_nsr["_admin"]["deployed"]["VCA"][vca_index] + osm_config = {"osm": {"ns_id": db_nsr["_id"]}} + db_dict = { + "collection": "nsrs", + "filter": {"_id": nsr_id}, + "path": db_update_entry, + } + step = "" + try: + + element_type = "NS" + element_under_configuration = nsr_id + + vnfr_id = None + if db_vnfr: + vnfr_id = db_vnfr["_id"] + osm_config["osm"]["vnf_id"] = vnfr_id + + namespace = "{nsi}.{ns}".format(nsi=nsi_id if nsi_id else "", ns=nsr_id) + + if vca_type == "native_charm": + index_number = 0 + else: + index_number = vdu_index or 0 + + if vnfr_id: + element_type = "VNF" + element_under_configuration = vnfr_id + namespace += ".{}-{}".format(vnfr_id, index_number) + if vdu_id: + namespace += ".{}-{}".format(vdu_id, index_number) + element_type = "VDU" + element_under_configuration = "{}-{}".format(vdu_id, index_number) + osm_config["osm"]["vdu_id"] = vdu_id + elif kdu_name: + namespace += ".{}".format(kdu_name) + element_type = "KDU" + element_under_configuration = kdu_name + osm_config["osm"]["kdu_name"] = kdu_name + + # Get artifact path + if base_folder["pkg-dir"]: + artifact_path = "{}/{}/{}/{}".format( + base_folder["folder"], + base_folder["pkg-dir"], + "charms" + if vca_type + in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm") + else "helm-charts", + vca_name, + ) + else: + artifact_path = "{}/Scripts/{}/{}/".format( + base_folder["folder"], + "charms" + if vca_type + in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm") + else "helm-charts", + vca_name, + ) + + self.logger.debug("Artifact path > {}".format(artifact_path)) + + # get initial_config_primitive_list that applies to this element + initial_config_primitive_list = config_descriptor.get( + "initial-config-primitive" + ) + + self.logger.debug( + "Initial config primitive list > {}".format( + initial_config_primitive_list + ) + ) + + # add config if not present for NS charm + ee_descriptor_id = ee_config_descriptor.get("id") + self.logger.debug("EE Descriptor > {}".format(ee_descriptor_id)) + initial_config_primitive_list = get_ee_sorted_initial_config_primitive_list( + initial_config_primitive_list, vca_deployed, ee_descriptor_id + ) + + self.logger.debug( + "Initial config primitive list #2 > {}".format( + initial_config_primitive_list + ) + ) + # n2vc_redesign STEP 3.1 + # find old ee_id if exists + ee_id = vca_deployed.get("ee_id") + + vca_id = self.get_vca_id(db_vnfr, db_nsr) + # create or register execution environment in VCA. Only for native charms when healing + if vca_type == "native_charm": + step = "Waiting to VM being up and getting IP address" + self.logger.debug(logging_text + step) + rw_mgmt_ip = await self.wait_vm_up_insert_key_ro( + logging_text, + nsr_id, + vnfr_id, + vdu_id, + vdu_index, + user=None, + pub_key=None, + ) + credentials = {"hostname": rw_mgmt_ip} + # get username + username = deep_get( + config_descriptor, ("config-access", "ssh-access", "default-user") + ) + # TODO remove this when changes on IM regarding config-access:ssh-access:default-user were + # merged. Meanwhile let's get username from initial-config-primitive + if not username and initial_config_primitive_list: + for config_primitive in initial_config_primitive_list: + for param in config_primitive.get("parameter", ()): + if param["name"] == "ssh-username": + username = param["value"] + break + if not username: + raise LcmException( + "Cannot determine the username neither with 'initial-config-primitive' nor with " + "'config-access.ssh-access.default-user'" + ) + credentials["username"] = username + + # n2vc_redesign STEP 3.2 + # TODO: Before healing at RO it is needed to destroy native charm units to be deleted. + self._write_configuration_status( + nsr_id=nsr_id, + vca_index=vca_index, + status="REGISTERING", + element_under_configuration=element_under_configuration, + element_type=element_type, + ) + + step = "register execution environment {}".format(credentials) + self.logger.debug(logging_text + step) + ee_id = await self.vca_map[vca_type].register_execution_environment( + credentials=credentials, + namespace=namespace, + db_dict=db_dict, + vca_id=vca_id, + ) + + # update ee_id en db + db_dict_ee_id = { + "_admin.deployed.VCA.{}.ee_id".format(vca_index): ee_id, + } + self.update_db_2("nsrs", nsr_id, db_dict_ee_id) + + # for compatibility with MON/POL modules, the need model and application name at database + # TODO ask MON/POL if needed to not assuming anymore the format "model_name.application_name" + # Not sure if this need to be done when healing + """ + ee_id_parts = ee_id.split(".") + db_nsr_update = {db_update_entry + "ee_id": ee_id} + if len(ee_id_parts) >= 2: + model_name = ee_id_parts[0] + application_name = ee_id_parts[1] + db_nsr_update[db_update_entry + "model"] = model_name + db_nsr_update[db_update_entry + "application"] = application_name + """ + + # n2vc_redesign STEP 3.3 + # Install configuration software. Only for native charms. + step = "Install configuration Software" + + self._write_configuration_status( + nsr_id=nsr_id, + vca_index=vca_index, + status="INSTALLING SW", + element_under_configuration=element_under_configuration, + element_type=element_type, + # other_update=db_nsr_update, + other_update=None, + ) + + # TODO check if already done + self.logger.debug(logging_text + step) + config = None + if vca_type == "native_charm": + config_primitive = next( + (p for p in initial_config_primitive_list if p["name"] == "config"), + None, + ) + if config_primitive: + config = self._map_primitive_params( + config_primitive, {}, deploy_params + ) + await self.vca_map[vca_type].install_configuration_sw( + ee_id=ee_id, + artifact_path=artifact_path, + db_dict=db_dict, + config=config, + num_units=1, + vca_id=vca_id, + vca_type=vca_type, + ) + + # write in db flag of configuration_sw already installed + self.update_db_2( + "nsrs", nsr_id, {db_update_entry + "config_sw_installed": True} + ) + + # Not sure if this need to be done when healing + """ + # add relations for this VCA (wait for other peers related with this VCA) + await self._add_vca_relations( + logging_text=logging_text, + nsr_id=nsr_id, + vca_type=vca_type, + vca_index=vca_index, + ) + """ + + # if SSH access is required, then get execution environment SSH public + # if native charm we have waited already to VM be UP + if vca_type in ("k8s_proxy_charm", "lxc_proxy_charm", "helm", "helm-v3"): + pub_key = None + user = None + # self.logger.debug("get ssh key block") + if deep_get( + config_descriptor, ("config-access", "ssh-access", "required") + ): + # self.logger.debug("ssh key needed") + # Needed to inject a ssh key + user = deep_get( + config_descriptor, + ("config-access", "ssh-access", "default-user"), + ) + step = "Install configuration Software, getting public ssh key" + pub_key = await self.vca_map[vca_type].get_ee_ssh_public__key( + ee_id=ee_id, db_dict=db_dict, vca_id=vca_id + ) + + step = "Insert public key into VM user={} ssh_key={}".format( + user, pub_key + ) + else: + # self.logger.debug("no need to get ssh key") + step = "Waiting to VM being up and getting IP address" + self.logger.debug(logging_text + step) + + # n2vc_redesign STEP 5.1 + # wait for RO (ip-address) Insert pub_key into VM + # IMPORTANT: We need do wait for RO to complete healing operation. + await self._wait_heal_ro(nsr_id, self.timeout.ns_heal) + if vnfr_id: + if kdu_name: + rw_mgmt_ip = await self.wait_kdu_up( + logging_text, nsr_id, vnfr_id, kdu_name + ) + else: + rw_mgmt_ip = await self.wait_vm_up_insert_key_ro( + logging_text, + nsr_id, + vnfr_id, + vdu_id, + vdu_index, + user=user, + pub_key=pub_key, + ) + else: + rw_mgmt_ip = None # This is for a NS configuration + + self.logger.debug(logging_text + " VM_ip_address={}".format(rw_mgmt_ip)) + + # store rw_mgmt_ip in deploy params for later replacement + deploy_params["rw_mgmt_ip"] = rw_mgmt_ip + + # Day1 operations. + # get run-day1 operation parameter + runDay1 = deploy_params.get("run-day1", False) + self.logger.debug( + "Healing vnf={}, vdu={}, runDay1 ={}".format(vnfr_id, vdu_id, runDay1) + ) + if runDay1: + # n2vc_redesign STEP 6 Execute initial config primitive + step = "execute initial config primitive" + + # wait for dependent primitives execution (NS -> VNF -> VDU) + if initial_config_primitive_list: + await self._wait_dependent_n2vc( + nsr_id, vca_deployed_list, vca_index + ) + + # stage, in function of element type: vdu, kdu, vnf or ns + my_vca = vca_deployed_list[vca_index] + if my_vca.get("vdu_id") or my_vca.get("kdu_name"): + # VDU or KDU + stage[0] = "Stage 3/5: running Day-1 primitives for VDU." + elif my_vca.get("member-vnf-index"): + # VNF + stage[0] = "Stage 4/5: running Day-1 primitives for VNF." + else: + # NS + stage[0] = "Stage 5/5: running Day-1 primitives for NS." + + self._write_configuration_status( + nsr_id=nsr_id, vca_index=vca_index, status="EXECUTING PRIMITIVE" + ) + + self._write_op_status(op_id=nslcmop_id, stage=stage) + + check_if_terminated_needed = True + for initial_config_primitive in initial_config_primitive_list: + # adding information on the vca_deployed if it is a NS execution environment + if not vca_deployed["member-vnf-index"]: + deploy_params["ns_config_info"] = json.dumps( + self._get_ns_config_info(nsr_id) + ) + # TODO check if already done + primitive_params_ = self._map_primitive_params( + initial_config_primitive, {}, deploy_params + ) + + step = "execute primitive '{}' params '{}'".format( + initial_config_primitive["name"], primitive_params_ + ) + self.logger.debug(logging_text + step) + await self.vca_map[vca_type].exec_primitive( + ee_id=ee_id, + primitive_name=initial_config_primitive["name"], + params_dict=primitive_params_, + db_dict=db_dict, + vca_id=vca_id, + vca_type=vca_type, + ) + # Once some primitive has been exec, check and write at db if it needs to exec terminated primitives + if check_if_terminated_needed: + if config_descriptor.get("terminate-config-primitive"): + self.update_db_2( + "nsrs", + nsr_id, + {db_update_entry + "needed_terminate": True}, + ) + check_if_terminated_needed = False + + # TODO register in database that primitive is done + + # STEP 7 Configure metrics + # Not sure if this need to be done when healing + """ + if vca_type == "helm" or vca_type == "helm-v3": + prometheus_jobs = await self.extract_prometheus_scrape_jobs( + ee_id=ee_id, + artifact_path=artifact_path, + ee_config_descriptor=ee_config_descriptor, + vnfr_id=vnfr_id, + nsr_id=nsr_id, + target_ip=rw_mgmt_ip, + ) + if prometheus_jobs: + self.update_db_2( + "nsrs", + nsr_id, + {db_update_entry + "prometheus_jobs": prometheus_jobs}, + ) + + for job in prometheus_jobs: + self.db.set_one( + "prometheus_jobs", + {"job_name": job["job_name"]}, + job, + upsert=True, + fail_on_empty=False, + ) + + """ + step = "instantiated at VCA" + self.logger.debug(logging_text + step) + + self._write_configuration_status( + nsr_id=nsr_id, vca_index=vca_index, status="READY" + ) + + except Exception as e: # TODO not use Exception but N2VC exception + # self.update_db_2("nsrs", nsr_id, {db_update_entry + "instantiation": "FAILED"}) + if not isinstance( + e, (DbException, N2VCException, LcmException, asyncio.CancelledError) + ): + self.logger.error( + "Exception while {} : {}".format(step, e), exc_info=True + ) + self._write_configuration_status( + nsr_id=nsr_id, vca_index=vca_index, status="BROKEN" + ) + raise LcmException("{} {}".format(step, e)) from e + + async def _wait_heal_ro( + self, + nsr_id, + timeout=600, + ): + start_time = time() + while time() <= start_time + timeout: + db_nsr = self.db.get_one("nsrs", {"_id": nsr_id}) + operational_status_ro = db_nsr["_admin"]["deployed"]["RO"][ + "operational-status" + ] + self.logger.debug("Wait Heal RO > {}".format(operational_status_ro)) + if operational_status_ro != "healing": + break + 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")