X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_lcm%2Fns.py;h=1e56e321c3ba547fdec38fe3428ffe448d108f34;hb=7ecbc3423bc9e08a662780b2eec10ee7cfe78b74;hp=139c8e41a3af2c6ada7c40ff284223c992d38f3a;hpb=78e3ec6beb92885718b426b2d123d10a69d1c6bb;p=osm%2FLCM.git diff --git a/osm_lcm/ns.py b/osm_lcm/ns.py index 139c8e4..1e56e32 100644 --- a/osm_lcm/ns.py +++ b/osm_lcm/ns.py @@ -1236,6 +1236,34 @@ class NsLcm(LcmBase): self.set_vnfr_at_error(db_vnfrs, str(e)) raise + async def wait_kdu_up(self, logging_text, nsr_id, vnfr_id, kdu_name): + """ + Wait for kdu to be up, get ip address + :param logging_text: prefix use for logging + :param nsr_id: + :param vnfr_id: + :param kdu_name: + :return: IP address + """ + + # self.logger.debug(logging_text + "Starting wait_kdu_up") + nb_tries = 0 + + while nb_tries < 360: + db_vnfr = self.db.get_one("vnfrs", {"_id": vnfr_id}) + kdur = next((x for x in get_iterable(db_vnfr, "kdur") if x.get("name") == kdu_name), None) + if not kdur: + raise LcmException("Not found vnfr_id={}, kdu_name={}".format(vnfr_id, kdu_name)) + if kdur.get("status"): + if kdur["status"] in ("READY", "ENABLED"): + return kdur.get("ip-address") + else: + raise LcmException("target KDU={} is in error state".format(kdu_name)) + + await asyncio.sleep(10, loop=self.loop) + nb_tries += 1 + raise LcmException("Timeout waiting KDU={} instantiated".format(kdu_name)) + async def wait_vm_up_insert_key_ro(self, logging_text, nsr_id, vnfr_id, vdu_id, vdu_index, pub_key=None, user=None): """ Wait for ip addres at RO, and optionally, insert public key in virtual machine @@ -1581,8 +1609,11 @@ class NsLcm(LcmBase): # n2vc_redesign STEP 5.1 # wait for RO (ip-address) Insert pub_key into VM if vnfr_id: - 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) + 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 @@ -1850,11 +1881,11 @@ class NsLcm(LcmBase): # wait for any previous tasks in process await self.lcm_tasks.waitfor_related_HA('ns', 'nslcmops', nslcmop_id) - stage[1] = "Sync filesystem from database" + stage[1] = "Sync filesystem from database." self.fs.sync() # TODO, make use of partial sync, only for the needed packages # STEP 0: Reading database (nslcmops, nsrs, nsds, vnfrs, vnfds) - stage[1] = "Reading from database" + stage[1] = "Reading from database." # nsState="BUILDING", currentOperation="INSTANTIATING", currentOperationID=nslcmop_id db_nsr_update["detailed-status"] = "creating" db_nsr_update["operational-status"] = "init" @@ -1872,7 +1903,7 @@ class NsLcm(LcmBase): ) # read from db: operation - stage[1] = "Getting nslcmop={} from db".format(nslcmop_id) + stage[1] = "Getting nslcmop={} from db.".format(nslcmop_id) db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id}) ns_params = db_nslcmop.get("operationParams") if ns_params and ns_params.get("timeout_ns_deploy"): @@ -1881,15 +1912,15 @@ class NsLcm(LcmBase): timeout_ns_deploy = self.timeout.get("ns_deploy", self.timeout_ns_deploy) # read from db: ns - stage[1] = "Getting nsr={} from db".format(nsr_id) + stage[1] = "Getting nsr={} from db.".format(nsr_id) db_nsr = self.db.get_one("nsrs", {"_id": nsr_id}) - stage[1] = "Getting nsd={} from db".format(db_nsr["nsd-id"]) + stage[1] = "Getting nsd={} from db.".format(db_nsr["nsd-id"]) nsd = self.db.get_one("nsds", {"_id": db_nsr["nsd-id"]}) db_nsr["nsd"] = nsd # nsr_name = db_nsr["name"] # TODO short-name?? # read from db: vnf's of this ns - stage[1] = "Getting vnfrs from db" + stage[1] = "Getting vnfrs from db." self.logger.debug(logging_text + stage[1]) db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id}) @@ -1903,10 +1934,11 @@ class NsLcm(LcmBase): db_vnfrs[vnfr["member-vnf-index-ref"]] = vnfr # vnf's dict indexed by member-index: '1', '2', etc vnfd_id = vnfr["vnfd-id"] # vnfd uuid for this vnf vnfd_ref = vnfr["vnfd-ref"] # vnfd name for this vnf + # if we haven't this vnfd, read it from db if vnfd_id not in db_vnfds: # read from db - stage[1] = "Getting vnfd={} id='{}' from db".format(vnfd_id, vnfd_ref) + stage[1] = "Getting vnfd={} id='{}' from db.".format(vnfd_id, vnfd_ref) self.logger.debug(logging_text + stage[1]) vnfd = self.db.get_one("vnfds", {"_id": vnfd_id}) @@ -1939,6 +1971,7 @@ class NsLcm(LcmBase): # set state to INSTANTIATED. When instantiated NBI will not delete directly db_nsr_update["_admin.nsState"] = "INSTANTIATED" self.update_db_2("nsrs", nsr_id, db_nsr_update) + self.db.set_list("vnfrs", {"nsr-id-ref": nsr_id}, {"_admin.nsState": "INSTANTIATED"}) # n2vc_redesign STEP 2 Deploy Network Scenario stage[0] = 'Stage 2/5: deployment of KDUs, VMs and execution environments.' @@ -1947,7 +1980,7 @@ class NsLcm(LcmBase): stage=stage ) - stage[1] = "Deploying KDUs," + stage[1] = "Deploying KDUs." # self.logger.debug(logging_text + "Before deploy_kdus") # Call to deploy_kdus in case exists the "vdu:kdu" param await self.deploy_kdus( @@ -2188,8 +2221,8 @@ class NsLcm(LcmBase): if error_list: error_detail = ". ".join(error_list) self.logger.error(logging_text + error_detail) - error_description_nslcmop = 'Stage: {}. Detail: {}'.format(stage[0], error_detail) - error_description_nsr = 'Operation: INSTANTIATING.{}, Stage {}'.format(nslcmop_id, stage[0]) + error_description_nslcmop = '{} Detail: {}'.format(stage[0], error_detail) + error_description_nsr = 'Operation: INSTANTIATING.{}, {}'.format(nslcmop_id, stage[0]) db_nsr_update["detailed-status"] = error_description_nsr + " Detail: " + error_detail db_nslcmop_update["detailed-status"] = error_detail @@ -2295,7 +2328,7 @@ class NsLcm(LcmBase): db_nsr = self.db.get_one("nsrs", {"_id": nsr_id}) # for each defined NS relation, find the VCA's related - for r in ns_relations: + for r in ns_relations.copy(): from_vca_ee_id = None to_vca_ee_id = None from_vca_endpoint = None @@ -2340,7 +2373,7 @@ class NsLcm(LcmBase): pass # for each defined VNF relation, find the VCA's related - for r in vnf_relations: + for r in vnf_relations.copy(): from_vca_ee_id = None to_vca_ee_id = None from_vca_endpoint = None @@ -2373,11 +2406,11 @@ class NsLcm(LcmBase): if vca.get('vdu_id') == r.get('entities')[0].get('id'): if vca_status.get('status') == 'BROKEN': # peer broken: remove relation from list - ns_relations.remove(r) + vnf_relations.remove(r) if vca.get('vdu_id') == r.get('entities')[1].get('id'): if vca_status.get('status') == 'BROKEN': # peer broken: remove relation from list - ns_relations.remove(r) + vnf_relations.remove(r) except Exception: # ignore pass @@ -2395,7 +2428,7 @@ class NsLcm(LcmBase): self.logger.warn(logging_text + ' ERROR adding relations: {}'.format(e)) return False - async def _install_kdu(self, nsr_id: str, nsr_db_path: str, vnfr_data: dict, kdu_index: int, kdur: dict, kdud: dict, + async def _install_kdu(self, nsr_id: str, nsr_db_path: str, vnfr_data: dict, kdu_index: int, kdud: dict, vnfd: dict, k8s_instance_info: dict, k8params: dict = None, timeout: int = 600): try: @@ -2423,8 +2456,9 @@ class NsLcm(LcmBase): namespace=k8s_instance_info["namespace"]) # Obtain management service info (if exists) + vnfr_update_dict = {} if services: - vnfr_update_dict = {"kdur.{}.services".format(kdu_index): services} + vnfr_update_dict["kdur.{}.services".format(kdu_index)] = services mgmt_services = [service for service in kdud.get("service", []) if service.get("mgmt-service")] for mgmt_service in mgmt_services: for service in services: @@ -2446,12 +2480,30 @@ class NsLcm(LcmBase): else: self.logger.warn("Mgmt service name: {} not found".format(mgmt_service["name"])) - self.update_db_2("vnfrs", vnfr_data.get("_id"), vnfr_update_dict) + vnfr_update_dict["kdur.{}.status".format(kdu_index)] = "READY" + self.update_db_2("vnfrs", vnfr_data.get("_id"), vnfr_update_dict) + + kdu_config = kdud.get("kdu-configuration") + if kdu_config and kdu_config.get("initial-config-primitive") and kdu_config.get("juju") is None: + initial_config_primitive_list = kdu_config.get("initial-config-primitive") + initial_config_primitive_list.sort(key=lambda val: int(val["seq"])) + + for initial_config_primitive in initial_config_primitive_list: + primitive_params_ = self._map_primitive_params(initial_config_primitive, {}, {}) + + await asyncio.wait_for( + self.k8scluster_map[k8sclustertype].exec_primitive( + cluster_uuid=k8s_instance_info["k8scluster-uuid"], + kdu_instance=kdu_instance, + primitive_name=initial_config_primitive["name"], + params=primitive_params_, db_dict={}), + timeout=timeout) except Exception as e: # Prepare update db with error and raise exception try: self.update_db_2("nsrs", nsr_id, {nsr_db_path + ".detailed-status": str(e)}) + self.update_db_2("vnfrs", vnfr_data.get("_id"), {"kdur.{}.status".format(kdu_index): "ERROR"}) except Exception: # ignore to keep original exception pass @@ -2465,14 +2517,22 @@ class NsLcm(LcmBase): k8scluster_id_2_uuic = {"helm-chart": {}, "juju-bundle": {}} - def _get_cluster_id(cluster_id, cluster_type): + async def _get_cluster_id(cluster_id, cluster_type): nonlocal k8scluster_id_2_uuic if cluster_id in k8scluster_id_2_uuic[cluster_type]: return k8scluster_id_2_uuic[cluster_type][cluster_id] + # check if K8scluster is creating and wait look if previous tasks in process + task_name, task_dependency = self.lcm_tasks.lookfor_related("k8scluster", cluster_id) + if task_dependency: + text = "Waiting for related tasks '{}' on k8scluster {} to be completed".format(task_name, cluster_id) + self.logger.debug(logging_text + text) + await asyncio.wait(task_dependency, timeout=3600) + db_k8scluster = self.db.get_one("k8sclusters", {"_id": cluster_id}, fail_on_empty=False) if not db_k8scluster: raise LcmException("K8s cluster {} cannot be found".format(cluster_id)) + k8s_id = deep_get(db_k8scluster, ("_admin", cluster_type, "id")) if not k8s_id: raise LcmException("K8s cluster '{}' has not been initialized for '{}'".format(cluster_id, @@ -2522,7 +2582,7 @@ class NsLcm(LcmBase): k8s_cluster_id = kdur["k8s-cluster"]["id"] step = "Synchronize repos for k8s cluster '{}'".format(k8s_cluster_id) - cluster_uuid = _get_cluster_id(k8s_cluster_id, k8sclustertype) + cluster_uuid = await _get_cluster_id(k8s_cluster_id, k8sclustertype) # Synchronize repos if k8sclustertype == "helm-chart" and cluster_uuid not in updated_cluster_list: @@ -2553,7 +2613,7 @@ class NsLcm(LcmBase): self.update_db_2("nsrs", nsr_id, db_nsr_update) task = asyncio.ensure_future( - self._install_kdu(nsr_id, db_path, vnfr_data, kdu_index, kdur, kdud, db_vnfds[vnfd_id], + self._install_kdu(nsr_id, db_path, vnfr_data, kdu_index, kdud, db_vnfds[vnfd_id], k8s_instance_info, k8params=desc_params, timeout=600)) self.lcm_tasks.register("ns", nsr_id, nslcmop_id, "instantiate_KDU-{}".format(index), task) task_instantiation_info[task] = "Deploying KDU {}".format(kdur["kdu-name"]) @@ -3193,7 +3253,6 @@ class NsLcm(LcmBase): self.logger.debug(logging_text + stage[1]) # self.logger.debug("nsr_deployed: {}".format(nsr_deployed)) for vca_index, vca in enumerate(get_iterable(nsr_deployed, "VCA")): - self.logger.debug("vca_index: {}, vca: {}".format(vca_index, vca)) config_descriptor = None if not vca or not vca.get("ee_id"): continue @@ -3215,8 +3274,11 @@ class NsLcm(LcmBase): vca_type = vca.get("type") exec_terminate_primitives = (not operation_params.get("skip_terminate_primitives") and vca.get("needed_terminate")) - # For helm we must destroy_ee - destroy_ee = "True" if vca_type == "helm" else "False" + # For helm we must destroy_ee. Also for native_charm, as juju_model cannot be deleted if there are + # pending native charms + destroy_ee = True if vca_type in ("helm", "native_charm") else False + # self.logger.debug(logging_text + "vca_index: {}, ee_id: {}, vca_type: {} destroy_ee: {}".format( + # vca_index, vca.get("ee_id"), vca_type, destroy_ee)) task = asyncio.ensure_future( self.destroy_N2VC(logging_text, db_nslcmop, vca, config_descriptor, vca_index, destroy_ee, exec_terminate_primitives)) @@ -3224,13 +3286,13 @@ class NsLcm(LcmBase): # wait for pending tasks of terminate primitives if tasks_dict_info: - self.logger.debug(logging_text + 'Waiting for terminate primitive pending tasks...') + self.logger.debug(logging_text + 'Waiting for tasks {}'.format(list(tasks_dict_info.keys()))) error_list = await self._wait_for_tasks(logging_text, tasks_dict_info, min(self.timeout_charm_delete, timeout_ns_terminate), stage, nslcmop_id) + tasks_dict_info.clear() if error_list: return # raise LcmException("; ".join(error_list)) - tasks_dict_info.clear() # remove All execution environments at once stage[0] = "Stage 3/3 delete all." @@ -3303,8 +3365,8 @@ class NsLcm(LcmBase): if error_list: error_detail = "; ".join(error_list) # self.logger.error(logging_text + error_detail) - error_description_nslcmop = 'Stage: {}. Detail: {}'.format(stage[0], error_detail) - error_description_nsr = 'Operation: TERMINATING.{}, Stage {}.'.format(nslcmop_id, stage[0]) + error_description_nslcmop = '{} Detail: {}'.format(stage[0], error_detail) + error_description_nsr = 'Operation: TERMINATING.{}, {}.'.format(nslcmop_id, stage[0]) db_nsr_update["operational-status"] = "failed" db_nsr_update["detailed-status"] = error_description_nsr + " Detail: " + error_detail @@ -3338,6 +3400,12 @@ class NsLcm(LcmBase): operation_state=nslcmop_operation_state, other_update=db_nslcmop_update, ) + if ns_state == "NOT_INSTANTIATED": + try: + self.db.set_list("vnfrs", {"nsr-id-ref": nsr_id}, {"_admin.nsState": "NOT_INSTANTIATED"}) + except DbException as e: + self.logger.warn(logging_text + 'Error writing VNFR status for nsr-id-ref: {} -> {}'. + format(nsr_id, e)) if operation_params: autoremove = operation_params.get("autoremove", False) if nslcmop_operation_state: @@ -3596,11 +3664,15 @@ class NsLcm(LcmBase): config_primitive_desc = config_primitive break - if not config_primitive_desc and not (kdu_name and primitive in ("upgrade", "rollback", "status")): - raise LcmException("Primitive {} not found at [ns|vnf|vdu]-configuration:config-primitive ". - format(primitive)) - primitive_name = config_primitive_desc.get("execution-environment-primitive", primitive) - ee_descriptor_id = config_primitive_desc.get("execution-environment-ref") + if not config_primitive_desc: + if not (kdu_name and primitive in ("upgrade", "rollback", "status")): + raise LcmException("Primitive {} not found at [ns|vnf|vdu]-configuration:config-primitive ". + format(primitive)) + primitive_name = primitive + ee_descriptor_id = None + else: + primitive_name = config_primitive_desc.get("execution-environment-primitive", primitive) + ee_descriptor_id = config_primitive_desc.get("execution-environment-ref") if vnf_index: if vdu_id: