From: bravof Date: Mon, 7 Dec 2020 15:57:31 +0000 (-0300) Subject: fix(scale): manual scaling now works X-Git-Tag: release-v9.0-start~6 X-Git-Url: https://osm.etsi.org/gitweb/?a=commitdiff_plain;h=refs%2Fchanges%2F81%2F10081%2F2;p=osm%2FLCM.git fix(scale): manual scaling now works Change-Id: I1f54930996368626d292ace5b55e45e9472775a7 Signed-off-by: bravof --- diff --git a/osm_lcm/data_utils/vnfd.py b/osm_lcm/data_utils/vnfd.py index 5774060..668fab9 100644 --- a/osm_lcm/data_utils/vnfd.py +++ b/osm_lcm/data_utils/vnfd.py @@ -135,3 +135,23 @@ def get_vdu_index(vnfd, vdu_id): return vnfd.get("vdu", ()).index(target_vdu) else: return -1 + + +def get_scaling_aspect(vnfd): + return vnfd.get("df", ())[0].get("scaling-aspect", ()) + + +def get_number_of_instances(vnfd, vdu_id): + return list_utils.find_in_list( + vnfd.get( + "df", + () + )[0].get( + "instantiation-level", + () + )[0].get( + "vdu-level", + () + ), + lambda a_vdu: a_vdu["vdu-id"] == vdu_id + )["number-of-instances"] diff --git a/osm_lcm/ns.py b/osm_lcm/ns.py index ea2320f..a90f4e8 100644 --- a/osm_lcm/ns.py +++ b/osm_lcm/ns.py @@ -31,7 +31,7 @@ from osm_lcm.data_utils.nsd import get_vnf_profiles from osm_lcm.data_utils.vnfd import get_vnf_configuration, get_vdu_list, get_vdu_profile, \ get_ee_sorted_initial_config_primitive_list, get_ee_sorted_terminate_config_primitive_list, \ get_kdu_list, get_virtual_link_profiles, get_vdu, get_vdu_configuration, get_kdu_configuration, \ - get_vdu_index + get_vdu_index, get_scaling_aspect, get_number_of_instances from osm_lcm.data_utils.list_utils import find_in_list from osm_lcm.data_utils.vnfr import get_osm_params from osm_lcm.data_utils.dict_utils import parse_yaml_strings @@ -3627,7 +3627,6 @@ class NsLcm(LcmBase): # vdu_name = db_nslcmop["operationParams"].get("vdu_name") ####### - RO_nsr_id = nsr_deployed["RO"].get("nsr_id") vnf_index = db_nslcmop["operationParams"]["scaleVnfData"]["scaleByStepData"]["member-vnf-index"] scaling_group = db_nslcmop["operationParams"]["scaleVnfData"]["scaleByStepData"]["scaling-group-descriptor"] scaling_type = db_nslcmop["operationParams"]["scaleVnfData"]["scaleVnfType"] @@ -3644,10 +3643,13 @@ class NsLcm(LcmBase): db_vnfd = self.db.get_one("vnfds", {"_id": db_vnfr["vnfd-id"]}) step = "Getting scaling-group-descriptor" - for scaling_descriptor in db_vnfd["scaling-group-descriptor"]: - if scaling_descriptor["name"] == scaling_group: - break - else: + scaling_descriptor = find_in_list( + get_scaling_aspect( + db_vnfd + ), + lambda scale_desc: scale_desc["name"] == scaling_group + ) + if not scaling_descriptor: raise LcmException("input parameter 'scaleByStepData':'scaling-group-descriptor':'{}' is not present " "at vnfd:scaling-group-descriptor".format(scaling_group)) @@ -3668,63 +3670,94 @@ class NsLcm(LcmBase): RO_scaling_info = [] vdu_scaling_info = {"scaling_group_name": scaling_group, "vdu": []} if scaling_type == "SCALE_OUT": + if "aspect-delta-details" not in scaling_descriptor: + raise LcmException( + "Aspect delta details not fount in scaling descriptor {}".format( + scaling_descriptor["name"] + ) + ) # count if max-instance-count is reached - max_instance_count = scaling_descriptor.get("max-instance-count", 10) - # self.logger.debug("MAX_INSTANCE_COUNT is {}".format(max_instance_count)) - if nb_scale_op >= max_instance_count: - raise LcmException("reached the limit of {} (max-instance-count) " - "scaling-out operations for the " - "scaling-group-descriptor '{}'".format(nb_scale_op, scaling_group)) - - nb_scale_op += 1 + deltas = scaling_descriptor.get("aspect-delta-details")["deltas"] + vdu_scaling_info["scaling_direction"] = "OUT" vdu_scaling_info["vdu-create"] = {} - for vdu_scale_info in scaling_descriptor["vdu"]: - vdud = next(vdu for vdu in db_vnfd.get("vdu") if vdu["id"] == vdu_scale_info["vdu-id-ref"]) - vdu_index = len([x for x in db_vnfr.get("vdur", ()) - if x.get("vdu-id-ref") == vdu_scale_info["vdu-id-ref"] and - x.get("member-vnf-index-ref") == vnf_index]) - cloud_init_text = self._get_vdu_cloud_init_content(vdud, db_vnfd) - if cloud_init_text: - additional_params = self._get_vdu_additional_params(db_vnfr, vdud["id"]) or {} - cloud_init_list = [] - for x in range(vdu_scale_info.get("count", 1)): + for delta in deltas: + for vdu_delta in delta["vdu-delta"]: + vdud = get_vdu(db_vnfd, vdu_delta["id"]) + vdu_index = get_vdu_index(db_vnfr, vdu_delta["id"]) + cloud_init_text = self._get_vdu_cloud_init_content(vdud, db_vnfd) if cloud_init_text: - # TODO Information of its own ip is not available because db_vnfr is not updated. - additional_params["OSM"] = get_osm_params( - db_vnfr, - vdu_scale_info["vdu-id-ref"], - vdu_index + x + additional_params = self._get_vdu_additional_params(db_vnfr, vdud["id"]) or {} + cloud_init_list = [] + + vdu_profile = get_vdu_profile(db_vnfd, vdu_delta["id"]) + max_instance_count = 10 + if vdu_profile and "max-number-of-instances" in vdu_profile: + max_instance_count = vdu_profile.get("max-number-of-instances", 10) + + deafult_instance_num = get_number_of_instances(db_vnfd, vdud["id"]) + + nb_scale_op += vdu_delta.get("number-of-instances", 1) + + if nb_scale_op + deafult_instance_num > max_instance_count: + raise LcmException( + "reached the limit of {} (max-instance-count) " + "scaling-out operations for the " + "scaling-group-descriptor '{}'".format(nb_scale_op, scaling_group) ) - cloud_init_list.append( - self._parse_cloud_init( - cloud_init_text, - additional_params, - db_vnfd["id"], - vdud["id"] + for x in range(vdu_delta.get("number-of-instances", 1)): + if cloud_init_text: + # TODO Information of its own ip is not available because db_vnfr is not updated. + additional_params["OSM"] = get_osm_params( + db_vnfr, + vdu_delta["id"], + vdu_index + x ) - ) - RO_scaling_info.append({"osm_vdu_id": vdu_scale_info["vdu-id-ref"], "member-vnf-index": vnf_index, - "type": "create", "count": vdu_scale_info.get("count", 1)}) - if cloud_init_list: - RO_scaling_info[-1]["cloud_init"] = cloud_init_list - vdu_scaling_info["vdu-create"][vdu_scale_info["vdu-id-ref"]] = vdu_scale_info.get("count", 1) + cloud_init_list.append( + self._parse_cloud_init( + cloud_init_text, + additional_params, + db_vnfd["id"], + vdud["id"] + ) + ) + RO_scaling_info.append( + { + "osm_vdu_id": vdu_delta["id"], + "member-vnf-index": vnf_index, + "type": "create", + "count": vdu_delta.get("number-of-instances", 1) + } + ) + if cloud_init_list: + RO_scaling_info[-1]["cloud_init"] = cloud_init_list + vdu_scaling_info["vdu-create"][vdu_delta["id"]] = vdu_delta.get("number-of-instances", 1) elif scaling_type == "SCALE_IN": - # count if min-instance-count is reached - min_instance_count = 0 if "min-instance-count" in scaling_descriptor and scaling_descriptor["min-instance-count"] is not None: min_instance_count = int(scaling_descriptor["min-instance-count"]) - if nb_scale_op <= min_instance_count: - raise LcmException("reached the limit of {} (min-instance-count) scaling-in operations for the " - "scaling-group-descriptor '{}'".format(nb_scale_op, scaling_group)) - nb_scale_op -= 1 + vdu_scaling_info["scaling_direction"] = "IN" vdu_scaling_info["vdu-delete"] = {} - for vdu_scale_info in scaling_descriptor["vdu"]: - RO_scaling_info.append({"osm_vdu_id": vdu_scale_info["vdu-id-ref"], "member-vnf-index": vnf_index, - "type": "delete", "count": vdu_scale_info.get("count", 1)}) - vdu_scaling_info["vdu-delete"][vdu_scale_info["vdu-id-ref"]] = vdu_scale_info.get("count", 1) + deltas = scaling_descriptor.get("aspect-delta-details")["deltas"] + for delta in deltas: + for vdu_delta in delta["vdu-delta"]: + min_instance_count = 0 + vdu_profile = get_vdu_profile(db_vnfd, vdu_delta["id"]) + if vdu_profile and "min-number-of-instances" in vdu_profile: + min_instance_count = vdu_profile["min-number-of-instances"] + + deafult_instance_num = get_number_of_instances(db_vnfd, vdu_delta["id"]) + + nb_scale_op -= vdu_delta.get("number-of-instances", 1) + if nb_scale_op + deafult_instance_num < min_instance_count: + raise LcmException( + "reached the limit of {} (min-instance-count) scaling-in operations for the " + "scaling-group-descriptor '{}'".format(nb_scale_op, scaling_group) + ) + RO_scaling_info.append({"osm_vdu_id": vdu_delta["id"], "member-vnf-index": vnf_index, + "type": "delete", "count": vdu_delta.get("number-of-instances", 1)}) + vdu_scaling_info["vdu-delete"][vdu_delta["id"]] = vdu_delta.get("number-of-instances", 1) # update VDU_SCALING_INFO with the VDUs to delete ip_addresses vdu_delete = copy(vdu_scaling_info.get("vdu-delete")) @@ -3828,9 +3861,6 @@ class NsLcm(LcmBase): scale_process = "RO" if self.ro_config.get("ng"): await self._scale_ng_ro(logging_text, db_nsr, db_nslcmop, db_vnfr, vdu_scaling_info, stage) - else: - await self._RO_scale(logging_text, RO_nsr_id, RO_scaling_info, db_nslcmop, db_vnfr, - db_nslcmop_update, vdu_scaling_info) vdu_scaling_info.pop("vdu-create", None) vdu_scaling_info.pop("vdu-delete", None) @@ -3969,18 +3999,17 @@ class NsLcm(LcmBase): db_vnfrs = {} # read from db: vnfd's for every vnf - db_vnfds = {} # every vnfd data indexed by vnf id - db_vnfds = {} + db_vnfds = [] # for each vnf in ns, read vnfd for vnfr in self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id}): db_vnfrs[vnfr["member-vnf-index-ref"]] = vnfr vnfd_id = vnfr["vnfd-id"] # vnfd uuid for this vnf # if we haven't this vnfd, read it from db - if vnfd_id not in db_vnfds: + if not find_in_list(db_vnfds, lambda a_vnfd: a_vnfd["id"] == vnfd_id): # read from db vnfd = self.db.get_one("vnfds", {"_id": vnfd_id}) - db_vnfds[vnfd_id] = vnfd # vnfd's indexed by id + db_vnfds.append(vnfd) n2vc_key = self.n2vc.get_public_key() n2vc_key_list = [n2vc_key] self.scale_vnfr(db_vnfr, vdu_scaling_info.get("vdu-create"), vdu_scaling_info.get("vdu-delete"), @@ -3993,115 +4022,6 @@ class NsLcm(LcmBase): if vdu_scaling_info.get("vdu-delete"): self.scale_vnfr(db_vnfr, None, vdu_scaling_info["vdu-delete"], mark_delete=False) - async def _RO_scale(self, logging_text, RO_nsr_id, RO_scaling_info, db_nslcmop, db_vnfr, db_nslcmop_update, - vdu_scaling_info): - nslcmop_id = db_nslcmop["_id"] - nsr_id = db_nslcmop["nsInstanceId"] - vdu_create = vdu_scaling_info.get("vdu-create") - vdu_delete = vdu_scaling_info.get("vdu-delete") - # Scale RO retry check: Check if this sub-operation has been executed before - op_index = self._check_or_add_scale_suboperation( - db_nslcmop, db_vnfr["member-vnf-index-ref"], None, None, 'SCALE-RO', RO_nsr_id, RO_scaling_info) - if op_index == self.SUBOPERATION_STATUS_SKIP: - # Skip sub-operation - result = 'COMPLETED' - result_detail = 'Done' - self.logger.debug(logging_text + "Skipped sub-operation RO, result {} {}".format(result, result_detail)) - else: - if op_index == self.SUBOPERATION_STATUS_NEW: - # New sub-operation: Get index of this sub-operation - op_index = len(db_nslcmop.get('_admin', {}).get('operations')) - 1 - self.logger.debug(logging_text + "New sub-operation RO") - else: - # retry: Get registered params for this existing sub-operation - op = db_nslcmop.get('_admin', {}).get('operations', [])[op_index] - RO_nsr_id = op.get('RO_nsr_id') - RO_scaling_info = op.get('RO_scaling_info') - self.logger.debug(logging_text + "Sub-operation RO retry") - - RO_desc = await self.RO.create_action("ns", RO_nsr_id, {"vdu-scaling": RO_scaling_info}) - # wait until ready - RO_nslcmop_id = RO_desc["instance_action_id"] - db_nslcmop_update["_admin.deploy.RO"] = RO_nslcmop_id - - RO_task_done = False - step = detailed_status = "Waiting for VIM to scale. RO_task_id={}.".format(RO_nslcmop_id) - detailed_status_old = None - self.logger.debug(logging_text + step) - - deployment_timeout = 1 * 3600 # One hour - while deployment_timeout > 0: - if not RO_task_done: - desc = await self.RO.show("ns", item_id_name=RO_nsr_id, extra_item="action", - extra_item_id=RO_nslcmop_id) - - # deploymentStatus - self._on_update_ro_db(nsrs_id=nsr_id, ro_descriptor=desc) - - ns_status, ns_status_info = self.RO.check_action_status(desc) - if ns_status == "ERROR": - raise ROclient.ROClientException(ns_status_info) - elif ns_status == "BUILD": - detailed_status = step + "; {}".format(ns_status_info) - elif ns_status == "ACTIVE": - RO_task_done = True - self.scale_vnfr(db_vnfr, vdu_create=vdu_create, vdu_delete=vdu_delete) - step = detailed_status = "Waiting ns ready at RO. RO_id={}".format(RO_nsr_id) - self.logger.debug(logging_text + step) - else: - assert False, "ROclient.check_action_status returns unknown {}".format(ns_status) - else: - desc = await self.RO.show("ns", RO_nsr_id) - ns_status, ns_status_info = self.RO.check_ns_status(desc) - # deploymentStatus - self._on_update_ro_db(nsrs_id=nsr_id, ro_descriptor=desc) - - if ns_status == "ERROR": - raise ROclient.ROClientException(ns_status_info) - elif ns_status == "BUILD": - detailed_status = step + "; {}".format(ns_status_info) - elif ns_status == "ACTIVE": - step = detailed_status = \ - "Waiting for management IP address reported by the VIM. Updating VNFRs" - try: - # nsr_deployed["nsr_ip"] = RO.get_ns_vnf_info(desc) - self.ns_update_vnfr({db_vnfr["member-vnf-index-ref"]: db_vnfr}, desc) - break - except LcmExceptionNoMgmtIP: - pass - else: - assert False, "ROclient.check_ns_status returns unknown {}".format(ns_status) - if detailed_status != detailed_status_old: - self._update_suboperation_status( - db_nslcmop, op_index, 'COMPLETED', detailed_status) - detailed_status_old = db_nslcmop_update["detailed-status"] = detailed_status - self.update_db_2("nslcmops", nslcmop_id, db_nslcmop_update) - - await asyncio.sleep(5, loop=self.loop) - deployment_timeout -= 5 - if deployment_timeout <= 0: - self._update_suboperation_status( - db_nslcmop, nslcmop_id, op_index, 'FAILED', "Timeout when waiting for ns to get ready") - raise ROclient.ROClientException("Timeout waiting ns to be ready") - - # update VDU_SCALING_INFO with the obtained ip_addresses - if vdu_scaling_info["scaling_direction"] == "OUT": - for vdur in reversed(db_vnfr["vdur"]): - if vdu_scaling_info["vdu-create"].get(vdur["vdu-id-ref"]): - vdu_scaling_info["vdu-create"][vdur["vdu-id-ref"]] -= 1 - vdu_scaling_info["vdu"].append({ - "name": vdur["name"] or vdur.get("vdu-name"), - "vdu_id": vdur["vdu-id-ref"], - "interface": [] - }) - for interface in vdur["interfaces"]: - vdu_scaling_info["vdu"][-1]["interface"].append({ - "name": interface["name"], - "ip_address": interface["ip-address"], - "mac_address": interface.get("mac-address"), - }) - self._update_suboperation_status(db_nslcmop, op_index, 'COMPLETED', 'Done') - async def add_prometheus_metrics(self, ee_id, artifact_path, ee_config_descriptor, vnfr_id, nsr_id, target_ip): if not self.prometheus: return