X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;ds=sidebyside;f=osm_lcm%2Fns.py;h=9800f9e5f934ddeb64995d1ac30d101c784ba095;hb=305b5c4df29e2fe284957ae5f160896b35d34bc9;hp=d6184fd55cf759b445b4aced915ddcfe12313ce9;hpb=a96dd9c3f653a8055187adb6671ffc34f228b8a5;p=osm%2FLCM.git diff --git a/osm_lcm/ns.py b/osm_lcm/ns.py index d6184fd..9800f9e 100644 --- a/osm_lcm/ns.py +++ b/osm_lcm/ns.py @@ -17,7 +17,7 @@ ## import asyncio -from typing import Any, Dict +from typing import Any, Dict, List import yaml import logging import logging.handlers @@ -96,6 +96,7 @@ from n2vc.n2vc_juju_conn import N2VCJujuConnector from n2vc.exceptions import N2VCException, N2VCNotFound, K8sException from osm_lcm.lcm_helm_conn import LCMHelmConn +from osm_lcm.prometheus import parse_job from copy import copy, deepcopy from time import time @@ -123,7 +124,7 @@ class NsLcm(LcmBase): SUBOPERATION_STATUS_SKIP = -3 task_name_deploy_vca = "Deploying VCA" - def __init__(self, msg, lcm_tasks, config, loop, prometheus=None): + def __init__(self, msg, lcm_tasks, config, loop): """ Init, Connect to database, filesystem storage, and messaging :param config: two level dictionary with configuration. Top level should contain 'database', 'storage', @@ -200,8 +201,6 @@ class NsLcm(LcmBase): "helm-v3": self.conn_helm_ee, } - self.prometheus = prometheus - # create RO client self.RO = NgRoClient(self.loop, **self.ro_config) @@ -540,7 +539,7 @@ class NsLcm(LcmBase): vdur_copy = deepcopy(vdur) vdur_copy["status"] = "BUILD" vdur_copy["status-detailed"] = None - vdur_copy["ip-address"]: None + vdur_copy["ip-address"] = None vdur_copy["_id"] = str(uuid4()) vdur_copy["count-index"] += count + 1 vdur_copy["id"] = "{}-{}".format( @@ -823,6 +822,37 @@ class NsLcm(LcmBase): if vld_params.get("common_id"): target_vld["common_id"] = vld_params.get("common_id") + # modify target["ns"]["vld"] with instantiation parameters to override vnf vim-account + def update_ns_vld_target(target, ns_params): + for vnf_params in ns_params.get("vnf", ()): + if vnf_params.get("vimAccountId"): + target_vnf = next( + ( + vnfr + for vnfr in db_vnfrs.values() + if vnf_params["member-vnf-index"] + == vnfr["member-vnf-index-ref"] + ), + None, + ) + vdur = next((vdur for vdur in target_vnf.get("vdur", ())), None) + 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"], + ) + if target_vld: + if vnf_params.get("vimAccountId") not in a_vld.get( + "vim_info", {} + ): + target["ns"]["vld"][a_index].get("vim_info").update( + { + "vim:{}".format(vnf_params["vimAccountId"]): { + "vim_network_name": "" + } + } + ) + nslcmop_id = db_nslcmop["_id"] target = { "name": db_nsr["name"], @@ -837,6 +867,10 @@ class NsLcm(LcmBase): image["vim_info"] = {} for flavor in target["flavor"]: flavor["vim_info"] = {} + if db_nsr.get("affinity-or-anti-affinity-group"): + target["affinity-or-anti-affinity-group"] = deepcopy(db_nsr["affinity-or-anti-affinity-group"]) + for affinity_or_anti_affinity_group in target["affinity-or-anti-affinity-group"]: + affinity_or_anti_affinity_group["vim_info"] = {} if db_nslcmop.get("lcmOperationType") != "instantiate": # get parameters of instantiation: @@ -938,6 +972,8 @@ class NsLcm(LcmBase): vld_params.update(vld_instantiation_params) parse_vld_instantiation_params(target_vim, target_vld, vld_params, None) target["ns"]["vld"].append(target_vld) + # Update the target ns_vld if vnf vim_account is overriden by instantiation params + update_ns_vld_target(target, ns_params) for vnfr in db_vnfrs.values(): vnfd = find_in_list( @@ -1127,6 +1163,13 @@ class NsLcm(LcmBase): if target_vim not in ns_image["vim_info"]: ns_image["vim_info"][target_vim] = {} + # Affinity groups + if vdur.get("affinity-or-anti-affinity-group-id"): + for ags_id in vdur["affinity-or-anti-affinity-group-id"]: + ns_ags = target["affinity-or-anti-affinity-group"][int(ags_id)] + if target_vim not in ns_ags["vim_info"]: + ns_ags["vim_info"][target_vim] = {} + vdur["vim_info"] = {target_vim: {}} # instantiation parameters # if vnf_params: @@ -1666,7 +1709,8 @@ class NsLcm(LcmBase): base_folder["folder"], base_folder["pkg-dir"], "charms" - if vca_type in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm") + if vca_type + in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm") else "helm-charts", vca_name, ) @@ -1674,7 +1718,8 @@ class NsLcm(LcmBase): artifact_path = "{}/Scripts/{}/{}/".format( base_folder["folder"], "charms" - if vca_type in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm") + if vca_type + in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm") else "helm-charts", vca_name, ) @@ -1987,7 +2032,7 @@ class NsLcm(LcmBase): # STEP 7 Configure metrics if vca_type == "helm" or vca_type == "helm-v3": - prometheus_jobs = await self.add_prometheus_metrics( + prometheus_jobs = await self.extract_prometheus_scrape_jobs( ee_id=ee_id, artifact_path=artifact_path, ee_config_descriptor=ee_config_descriptor, @@ -2002,6 +2047,15 @@ class NsLcm(LcmBase): {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) @@ -2266,6 +2320,10 @@ class NsLcm(LcmBase): # read from db: operation stage[1] = "Getting nslcmop={} from db.".format(nslcmop_id) db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id}) + if db_nslcmop["operationParams"].get("additionalParamsForVnf"): + db_nslcmop["operationParams"]["additionalParamsForVnf"] = json.loads( + db_nslcmop["operationParams"]["additionalParamsForVnf"] + ) ns_params = db_nslcmop.get("operationParams") if ns_params and ns_params.get("timeout_ns_deploy"): timeout_ns_deploy = ns_params["timeout_ns_deploy"] @@ -2276,8 +2334,10 @@ class NsLcm(LcmBase): # read from db: ns stage[1] = "Getting nsr={} from db.".format(nsr_id) + self.logger.debug(logging_text + stage[1]) db_nsr = self.db.get_one("nsrs", {"_id": nsr_id}) 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 @@ -2293,6 +2353,14 @@ class NsLcm(LcmBase): # for each vnf in ns, read vnfd for vnfr in db_vnfrs_list: + if vnfr.get("kdur"): + kdur_list = [] + for kdur in vnfr["kdur"]: + if kdur.get("additionalParams"): + kdur["additionalParams"] = json.loads(kdur["additionalParams"]) + kdur_list.append(kdur) + vnfr["kdur"] = kdur_list + db_vnfrs[vnfr["member-vnf-index-ref"]] = vnfr vnfd_id = vnfr["vnfd-id"] vnfd_ref = vnfr["vnfd-ref"] @@ -2735,15 +2803,39 @@ class NsLcm(LcmBase): nsd: Dict[str, Any], vca: DeployedVCA, cached_vnfds: Dict[str, Any], - ): + ) -> List[Relation]: relations = [] db_ns_relations = get_ns_configuration_relation_list(nsd) for r in db_ns_relations: + provider_dict = None + requirer_dict = None + if all(key in r for key in ("provider", "requirer")): + provider_dict = r["provider"] + requirer_dict = r["requirer"] + elif "entities" in r: + provider_id = r["entities"][0]["id"] + provider_dict = { + "nsr-id": nsr_id, + "endpoint": r["entities"][0]["endpoint"], + } + if provider_id != nsd["id"]: + provider_dict["vnf-profile-id"] = provider_id + requirer_id = r["entities"][1]["id"] + requirer_dict = { + "nsr-id": nsr_id, + "endpoint": r["entities"][1]["endpoint"], + } + if requirer_id != nsd["id"]: + requirer_dict["vnf-profile-id"] = requirer_id + else: + raise Exception( + "provider/requirer or entities must be included in the relation." + ) relation_provider = self._update_ee_relation_data_with_implicit_data( - nsr_id, nsd, r["provider"], cached_vnfds + nsr_id, nsd, provider_dict, cached_vnfds ) relation_requirer = self._update_ee_relation_data_with_implicit_data( - nsr_id, nsd, r["requirer"], cached_vnfds + nsr_id, nsd, requirer_dict, cached_vnfds ) provider = EERelation(relation_provider) requirer = EERelation(relation_requirer) @@ -2759,7 +2851,7 @@ class NsLcm(LcmBase): nsd: Dict[str, Any], vca: DeployedVCA, cached_vnfds: Dict[str, Any], - ): + ) -> List[Relation]: relations = [] vnf_profile = get_vnf_profile(nsd, vca.vnf_profile_id) vnf_profile_id = vnf_profile["id"] @@ -2767,11 +2859,37 @@ class NsLcm(LcmBase): db_vnfd = self._get_vnfd(vnfd_id, cached_vnfds) db_vnf_relations = get_relation_list(db_vnfd, vnfd_id) for r in db_vnf_relations: + provider_dict = None + requirer_dict = None + if all(key in r for key in ("provider", "requirer")): + provider_dict = r["provider"] + requirer_dict = r["requirer"] + elif "entities" in r: + provider_id = r["entities"][0]["id"] + provider_dict = { + "nsr-id": nsr_id, + "vnf-profile-id": vnf_profile_id, + "endpoint": r["entities"][0]["endpoint"], + } + if provider_id != vnfd_id: + provider_dict["vdu-profile-id"] = provider_id + requirer_id = r["entities"][1]["id"] + requirer_dict = { + "nsr-id": nsr_id, + "vnf-profile-id": vnf_profile_id, + "endpoint": r["entities"][1]["endpoint"], + } + if requirer_id != vnfd_id: + requirer_dict["vdu-profile-id"] = requirer_id + else: + raise Exception( + "provider/requirer or entities must be included in the relation." + ) relation_provider = self._update_ee_relation_data_with_implicit_data( - nsr_id, nsd, r["provider"], cached_vnfds, vnf_profile_id=vnf_profile_id + nsr_id, nsd, provider_dict, cached_vnfds, vnf_profile_id=vnf_profile_id ) relation_requirer = self._update_ee_relation_data_with_implicit_data( - nsr_id, nsd, r["requirer"], cached_vnfds, vnf_profile_id=vnf_profile_id + nsr_id, nsd, requirer_dict, cached_vnfds, vnf_profile_id=vnf_profile_id ) provider = EERelation(relation_provider) requirer = EERelation(relation_requirer) @@ -3900,8 +4018,9 @@ class NsLcm(LcmBase): "nsrs", db_nslcmop["nsInstanceId"], {db_update_entry: False} ) - if vca_deployed.get("prometheus_jobs") and self.prometheus: - await self.prometheus.update(remove_jobs=vca_deployed["prometheus_jobs"]) + # Delete Prometheus Jobs if any + # This uses NSR_ID, so it will destroy any jobs under this index + self.db.del_list("prometheus_jobs", {"nsr_id": db_nslcmop["nsInstanceId"]}) if destroy_ee: await self.vca_map[vca_type].delete_execution_environment( @@ -4768,6 +4887,10 @@ class NsLcm(LcmBase): step = "Getting information from database" db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id}) db_nsr = self.db.get_one("nsrs", {"_id": nsr_id}) + if db_nslcmop["operationParams"].get("primitive_params"): + db_nslcmop["operationParams"]["primitive_params"] = json.loads( + db_nslcmop["operationParams"]["primitive_params"] + ) nsr_deployed = db_nsr["_admin"].get("deployed") vnf_index = db_nslcmop["operationParams"].get("member_vnf_index") @@ -4785,6 +4908,13 @@ class NsLcm(LcmBase): db_vnfr = self.db.get_one( "vnfrs", {"member-vnf-index-ref": vnf_index, "nsr-id-ref": nsr_id} ) + if db_vnfr.get("kdur"): + kdur_list = [] + for kdur in db_vnfr["kdur"]: + if kdur.get("additionalParams"): + kdur["additionalParams"] = json.loads(kdur["additionalParams"]) + kdur_list.append(kdur) + db_vnfr["kdur"] = kdur_list step = "Getting vnfd from database" db_vnfd = self.db.get_one("vnfds", {"_id": db_vnfr["vnfd-id"]}) @@ -5274,7 +5404,7 @@ class NsLcm(LcmBase): for kdu_delta in delta.get("kdu-resource-delta", {}): kdu_profile = get_kdu_resource_profile(db_vnfd, kdu_delta["id"]) kdu_name = kdu_profile["kdu-name"] - resource_name = kdu_profile["resource-name"] + resource_name = kdu_profile.get("resource-name", "") # Might have different kdus in the same delta # Should have list for each kdu @@ -5290,7 +5420,6 @@ class NsLcm(LcmBase): and kdur.get("helm-version") == "v2" ): k8s_cluster_type = "helm-chart" - raise NotImplementedError elif kdur.get("juju-bundle"): k8s_cluster_type = "juju-bundle" else: @@ -5320,7 +5449,13 @@ class NsLcm(LcmBase): kdu_instance = deployed_kdu.get("kdu-instance") instance_num = await self.k8scluster_map[ k8s_cluster_type - ].get_scale_count(resource_name, kdu_instance, vca_id=vca_id) + ].get_scale_count( + resource_name, + kdu_instance, + vca_id=vca_id, + cluster_uuid=deployed_kdu.get("k8scluster-uuid"), + kdu_model=deployed_kdu.get("kdu-model"), + ) kdu_replica_count = instance_num + kdu_delta.get( "number-of-instances", 1 ) @@ -5404,7 +5539,7 @@ class NsLcm(LcmBase): for kdu_delta in delta.get("kdu-resource-delta", {}): kdu_profile = get_kdu_resource_profile(db_vnfd, kdu_delta["id"]) kdu_name = kdu_profile["kdu-name"] - resource_name = kdu_profile["resource-name"] + resource_name = kdu_profile.get("resource-name", "") if not scaling_info["kdu-delete"].get(kdu_name, None): scaling_info["kdu-delete"][kdu_name] = [] @@ -5418,7 +5553,6 @@ class NsLcm(LcmBase): and kdur.get("helm-version") == "v2" ): k8s_cluster_type = "helm-chart" - raise NotImplementedError elif kdur.get("juju-bundle"): k8s_cluster_type = "juju-bundle" else: @@ -5446,7 +5580,13 @@ class NsLcm(LcmBase): kdu_instance = deployed_kdu.get("kdu-instance") instance_num = await self.k8scluster_map[ k8s_cluster_type - ].get_scale_count(resource_name, kdu_instance, vca_id=vca_id) + ].get_scale_count( + resource_name, + kdu_instance, + vca_id=vca_id, + cluster_uuid=deployed_kdu.get("k8scluster-uuid"), + kdu_model=deployed_kdu.get("kdu-model"), + ) kdu_replica_count = instance_num - kdu_delta.get( "number-of-instances", 1 ) @@ -5645,7 +5785,7 @@ class NsLcm(LcmBase): ] = "Deleting the execution environments" scale_process = "VCA" for vca_info in vca_scaling_info: - if vca_info["type"] == "delete": + if vca_info["type"] == "delete" and not vca_info.get("osm_kdu_id"): member_vnf_index = str(vca_info["member-vnf-index"]) self.logger.debug( logging_text + "vdu info: {}".format(vca_info) @@ -5658,14 +5798,6 @@ class NsLcm(LcmBase): ] = "Scaling member_vnf_index={}, vdu_id={}, vdu_index={} ".format( member_vnf_index, vdu_id, vdu_index ) - else: - vdu_index = 0 - kdu_id = vca_info["osm_kdu_id"] - stage[ - 1 - ] = "Scaling member_vnf_index={}, kdu_id={}, vdu_index={} ".format( - member_vnf_index, kdu_id, vdu_index - ) stage[2] = step = "Scaling in VCA" self._write_op_status(op_id=nslcmop_id, stage=stage) vca_update = db_nsr["_admin"]["deployed"]["VCA"] @@ -5781,7 +5913,7 @@ class NsLcm(LcmBase): ] = "Creating new execution environments" scale_process = "VCA" for vca_info in vca_scaling_info: - if vca_info["type"] == "create": + if vca_info["type"] == "create" and not vca_info.get("osm_kdu_id"): member_vnf_index = str(vca_info["member-vnf-index"]) self.logger.debug( logging_text + "vdu info: {}".format(vca_info) @@ -5869,43 +6001,6 @@ class NsLcm(LcmBase): task_instantiation_info=tasks_dict_info, stage=stage, ) - else: - kdu_name = vca_info["osm_kdu_id"] - descriptor_config = get_configuration(db_vnfd, kdu_name) - if descriptor_config: - vdu_id = None - kdu_index = int(vca_info["kdu_index"]) - vdu_name = None - kdur = next( - x - for x in db_vnfr["kdur"] - if x["kdu-name"] == kdu_name - ) - deploy_params_kdu = {"OSM": get_osm_params(db_vnfr)} - if kdur.get("additionalParams"): - deploy_params_kdu = parse_yaml_strings( - kdur["additionalParams"] - ) - - self._deploy_n2vc( - logging_text=logging_text, - db_nsr=db_nsr, - db_vnfr=db_vnfr, - nslcmop_id=nslcmop_id, - nsr_id=nsr_id, - nsi_id=nsi_id, - vnfd_id=vnfd_id, - vdu_id=vdu_id, - kdu_name=kdu_name, - member_vnf_index=member_vnf_index, - vdu_index=kdu_index, - vdu_name=vdu_name, - deploy_params=deploy_params_kdu, - descriptor_config=descriptor_config, - base_folder=base_folder, - task_instantiation_info=tasks_dict_info, - stage=stage, - ) # SCALE-UP VCA - END scale_process = None @@ -6152,6 +6247,7 @@ class NsLcm(LcmBase): ) cluster_uuid = deployed_kdu["k8scluster-uuid"] kdu_instance = deployed_kdu["kdu-instance"] + kdu_model = deployed_kdu.get("kdu-model") scale = int(kdu_scaling_info["scale"]) k8s_cluster_type = kdu_scaling_info["k8s-cluster-type"] @@ -6206,6 +6302,10 @@ class NsLcm(LcmBase): scale, kdu_scaling_info["resource-name"], vca_id=vca_id, + cluster_uuid=cluster_uuid, + kdu_model=kdu_model, + atomic=True, + db_dict=db_dict, ), timeout=self.timeout_vca_on_error, ) @@ -6289,11 +6389,9 @@ class NsLcm(LcmBase): db_vnfr, None, vdu_scaling_info["vdu-delete"], mark_delete=False ) - async def add_prometheus_metrics( + async def extract_prometheus_scrape_jobs( self, ee_id, artifact_path, ee_config_descriptor, vnfr_id, nsr_id, target_ip ): - if not self.prometheus: - return # look if exist a file called 'prometheus*.j2' and artifact_content = self.fs.dir_ls(artifact_path) job_file = next( @@ -6320,7 +6418,7 @@ class NsLcm(LcmBase): "EXPORTER_POD_IP": host_name, "EXPORTER_POD_PORT": host_port, } - job_list = self.prometheus.parse_job(job_data, variables) + job_list = parse_job(job_data, variables) # ensure job_name is using the vnfr_id. Adding the metadata nsr_id for job in job_list: if ( @@ -6329,9 +6427,8 @@ class NsLcm(LcmBase): ): job["job_name"] = vnfr_id + "_" + str(randint(1, 10000)) job["nsr_id"] = nsr_id - job_dict = {jl["job_name"]: jl for jl in job_list} - if await self.prometheus.update(job_dict): - return list(job_dict.keys()) + job["vnfr_id"] = vnfr_id + return job_list def get_vca_cloud_and_credentials(self, vim_account_id: str) -> (str, str): """