X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=NG-RO%2Fosm_ng_ro%2Fns.py;h=70217d691c5647f6fb6b2494150ed471d8032eb0;hb=6a6e3344cc0d68064a592941e33cdc6629eb3405;hp=c9791895c6ff2731df5e4207eff33ad0f3f65cb8;hpb=79ac6df793d0c47de98f6b3a57026aef6bdbb1f3;p=osm%2FRO.git diff --git a/NG-RO/osm_ng_ro/ns.py b/NG-RO/osm_ng_ro/ns.py index c9791895..70217d69 100644 --- a/NG-RO/osm_ng_ro/ns.py +++ b/NG-RO/osm_ng_ro/ns.py @@ -17,12 +17,13 @@ ## from http import HTTPStatus +from itertools import product import logging from random import choice as random_choice from threading import Lock from time import time from traceback import format_exc as traceback_format_exc -from typing import Any, Dict, Tuple, Type +from typing import Any, Dict, List, Optional, Tuple, Type from uuid import uuid4 from cryptography.hazmat.backends import default_backend as crypto_default_backend @@ -30,6 +31,7 @@ from cryptography.hazmat.primitives import serialization as crypto_serialization from cryptography.hazmat.primitives.asymmetric import rsa from jinja2 import ( Environment, + select_autoescape, StrictUndefined, TemplateError, TemplateNotFound, @@ -76,8 +78,8 @@ def get_process_id(): if text_id: return text_id - except Exception: - pass + except Exception as error: + logging.exception(f"{error} occured while getting process id") # Return a random id return "".join(random_choice("0123456789abcdef") for _ in range(12)) @@ -379,7 +381,10 @@ class Ns(object): str: [description] """ try: - env = Environment(undefined=StrictUndefined) + env = Environment( + undefined=StrictUndefined, + autoescape=select_autoescape(default_for_string=True, default=True), + ) template = env.from_string(cloud_init_content) return template.render(params or {}) @@ -642,44 +647,57 @@ class Ns(object): Tuple[Dict[str, Any], bool]: [description] """ numa = {} + numa_list = [] epa_vcpu_set = False if guest_epa_quota.get("numa-node-policy"): numa_node_policy = guest_epa_quota.get("numa-node-policy") if numa_node_policy.get("node"): - numa_node = numa_node_policy["node"][0] - - if numa_node.get("num-cores"): - numa["cores"] = numa_node["num-cores"] - epa_vcpu_set = True - - paired_threads = numa_node.get("paired-threads", {}) - if paired_threads.get("num-paired-threads"): - numa["paired-threads"] = int( - numa_node["paired-threads"]["num-paired-threads"] - ) - epa_vcpu_set = True + for numa_node in numa_node_policy["node"]: + vcpu_list = [] + if numa_node.get("id"): + numa["id"] = int(numa_node["id"]) + + if numa_node.get("vcpu"): + for vcpu in numa_node.get("vcpu"): + vcpu_id = int(vcpu.get("id")) + vcpu_list.append(vcpu_id) + numa["vcpu"] = vcpu_list + + if numa_node.get("num-cores"): + numa["cores"] = numa_node["num-cores"] + epa_vcpu_set = True + + paired_threads = numa_node.get("paired-threads", {}) + if paired_threads.get("num-paired-threads"): + numa["paired_threads"] = int( + numa_node["paired-threads"]["num-paired-threads"] + ) + epa_vcpu_set = True - if paired_threads.get("paired-thread-ids"): - numa["paired-threads-id"] = [] + if paired_threads.get("paired-thread-ids"): + numa["paired-threads-id"] = [] - for pair in paired_threads["paired-thread-ids"]: - numa["paired-threads-id"].append( - ( - str(pair["thread-a"]), - str(pair["thread-b"]), + for pair in paired_threads["paired-thread-ids"]: + numa["paired-threads-id"].append( + ( + str(pair["thread-a"]), + str(pair["thread-b"]), + ) ) - ) - if numa_node.get("num-threads"): - numa["threads"] = int(numa_node["num-threads"]) - epa_vcpu_set = True + if numa_node.get("num-threads"): + numa["threads"] = int(numa_node["num-threads"]) + epa_vcpu_set = True - if numa_node.get("memory-mb"): - numa["memory"] = max(int(int(numa_node["memory-mb"]) / 1024), 1) + if numa_node.get("memory-mb"): + numa["memory"] = max(int(int(numa_node["memory-mb"]) / 1024), 1) - return numa, epa_vcpu_set + numa_list.append(numa) + numa = {} + + return numa_list, epa_vcpu_set @staticmethod def _process_guest_epa_cpu_pinning_params( @@ -704,9 +722,12 @@ class Ns(object): guest_epa_quota.get("cpu-pinning-policy") == "DEDICATED" and not epa_vcpu_set ): + # Pinning policy "REQUIRE" uses threads as host should support SMT architecture + # Pinning policy "ISOLATE" uses cores as host should not support SMT architecture + # Pinning policy "PREFER" uses threads in case host supports SMT architecture numa[ "cores" - if guest_epa_quota.get("cpu-thread-pinning-policy") != "PREFER" + if guest_epa_quota.get("cpu-thread-pinning-policy") == "ISOLATE" else "threads" ] = max(vcpu_count, 1) local_epa_vcpu_set = True @@ -727,23 +748,39 @@ class Ns(object): """ extended = {} numa = {} + numa_list = [] if target_flavor.get("guest-epa"): guest_epa = target_flavor["guest-epa"] - numa, epa_vcpu_set = Ns._process_guest_epa_numa_params( + numa_list, epa_vcpu_set = Ns._process_guest_epa_numa_params( guest_epa_quota=guest_epa ) if guest_epa.get("mempage-size"): extended["mempage-size"] = guest_epa.get("mempage-size") + if guest_epa.get("cpu-pinning-policy"): + extended["cpu-pinning-policy"] = guest_epa.get("cpu-pinning-policy") + + if guest_epa.get("cpu-thread-pinning-policy"): + extended["cpu-thread-pinning-policy"] = guest_epa.get( + "cpu-thread-pinning-policy" + ) + + if guest_epa.get("numa-node-policy"): + if guest_epa.get("numa-node-policy").get("mem-policy"): + extended["mem-policy"] = guest_epa.get("numa-node-policy").get( + "mem-policy" + ) + tmp_numa, epa_vcpu_set = Ns._process_guest_epa_cpu_pinning_params( guest_epa_quota=guest_epa, vcpu_count=int(target_flavor.get("vcpu-count", 1)), epa_vcpu_set=epa_vcpu_set, ) - numa.update(tmp_numa) + for numa in numa_list: + numa.update(tmp_numa) extended.update( Ns._process_guest_epa_quota_params( @@ -753,7 +790,7 @@ class Ns(object): ) if numa: - extended["numas"] = [numa] + extended["numas"] = numa_list return extended @@ -776,18 +813,35 @@ class Ns(object): Returns: Dict[str, Any]: [description] """ + db = kwargs.get("db") + target_vdur = {} + flavor_data = { "disk": int(target_flavor["storage-gb"]), "ram": int(target_flavor["memory-mb"]), "vcpus": int(target_flavor["vcpu-count"]), } - target_vdur = {} for vnf in indata.get("vnf", []): for vdur in vnf.get("vdur", []): - if vdur.get("ns-flavor-id") == target_flavor["id"]: + if vdur.get("ns-flavor-id") == target_flavor.get("id"): target_vdur = vdur + if db and isinstance(indata.get("vnf"), list): + vnfd_id = indata.get("vnf")[0].get("vnfd-id") + vnfd = db.get_one("vnfds", {"_id": vnfd_id}) + # check if there is persistent root disk + for vdu in vnfd.get("vdu", ()): + if vdu["name"] == target_vdur.get("vdu-name"): + for vsd in vnfd.get("virtual-storage-desc", ()): + if vsd.get("id") == vdu.get("virtual-storage-desc", [[]])[0]: + root_disk = vsd + if ( + root_disk.get("type-of-storage") + == "persistent-storage:persistent-storage" + ): + flavor_data["disk"] = 0 + for storage in target_vdur.get("virtual-storages", []): if ( storage.get("type-of-storage") @@ -898,7 +952,7 @@ class Ns(object): "id": vim_info.get("vim_network_id"), }, } - elif target_vld.get("mgmt-network"): + elif target_vld.get("mgmt-network") and not vim_info.get("provider_network"): extra_dict["find_params"] = { "mgmt": True, "name": target_vld["id"], @@ -923,122 +977,197 @@ class Ns(object): return extra_dict @staticmethod - def _process_vdu_params( - target_vdu: Dict[str, Any], - indata: Dict[str, Any], - vim_info: Dict[str, Any], - target_record_id: str, - **kwargs: Dict[str, Any], - ) -> Dict[str, Any]: - """Function to process VDU parameters. + def find_persistent_root_volumes( + vnfd: dict, + target_vdu: dict, + vdu_instantiation_volumes_list: list, + disk_list: list, + ) -> Dict[str, any]: + """Find the persistent root volumes and add them to the disk_list + by parsing the instantiation parameters. Args: - target_vdu (Dict[str, Any]): [description] - indata (Dict[str, Any]): [description] - vim_info (Dict[str, Any]): [description] - target_record_id (str): [description] + vnfd (dict): VNF descriptor + target_vdu (dict): processed VDU + vdu_instantiation_volumes_list (list): instantiation parameters for the each VDU as a list + disk_list (list): to be filled up Returns: - Dict[str, Any]: [description] + persistent_root_disk (dict): Details of persistent root disk + """ - vnfr_id = kwargs.get("vnfr_id") - nsr_id = kwargs.get("nsr_id") - vnfr = kwargs.get("vnfr") - vdu2cloud_init = kwargs.get("vdu2cloud_init") - tasks_by_target_record_id = kwargs.get("tasks_by_target_record_id") - logger = kwargs.get("logger") - db = kwargs.get("db") - fs = kwargs.get("fs") - ro_nsr_public_key = kwargs.get("ro_nsr_public_key") + persistent_root_disk = {} + # There can be only one root disk, when we find it, it will return the result - vnf_preffix = "vnfrs:{}".format(vnfr_id) - ns_preffix = "nsrs:{}".format(nsr_id) - image_text = ns_preffix + ":image." + target_vdu["ns-image-id"] - flavor_text = ns_preffix + ":flavor." + target_vdu["ns-flavor-id"] - extra_dict = {"depends_on": [image_text, flavor_text]} - net_list = [] + for vdu, vsd in product( + vnfd.get("vdu", ()), vnfd.get("virtual-storage-desc", ()) + ): + if ( + vdu["name"] == target_vdu["vdu-name"] + and vsd.get("id") == vdu.get("virtual-storage-desc", [[]])[0] + ): + root_disk = vsd + if ( + root_disk.get("type-of-storage") + == "persistent-storage:persistent-storage" + ): + for vdu_volume in vdu_instantiation_volumes_list: - for iface_index, interface in enumerate(target_vdu["interfaces"]): - if interface.get("ns-vld-id"): - net_text = ns_preffix + ":vld." + interface["ns-vld-id"] - elif interface.get("vnf-vld-id"): - net_text = vnf_preffix + ":vld." + interface["vnf-vld-id"] - else: - logger.error( - "Interface {} from vdu {} not connected to any vld".format( - iface_index, target_vdu["vdu-name"] - ) - ) + if ( + vdu_volume["vim-volume-id"] + and root_disk["id"] == vdu_volume["name"] + ): - continue # interface not connected to any vld + persistent_root_disk[vsd["id"]] = { + "vim_volume_id": vdu_volume["vim-volume-id"], + "image_id": vdu.get("sw-image-desc"), + } - extra_dict["depends_on"].append(net_text) + disk_list.append(persistent_root_disk[vsd["id"]]) - if "port-security-enabled" in interface: - interface["port_security"] = interface.pop("port-security-enabled") + return persistent_root_disk - if "port-security-disable-strategy" in interface: - interface["port_security_disable_strategy"] = interface.pop( - "port-security-disable-strategy" - ) + else: - net_item = { - x: v - for x, v in interface.items() - if x - in ( - "name", - "vpci", - "port_security", - "port_security_disable_strategy", - "floating_ip", - ) - } - net_item["net_id"] = "TASK-" + net_text - net_item["type"] = "virtual" + if root_disk.get("size-of-storage"): + persistent_root_disk[vsd["id"]] = { + "image_id": vdu.get("sw-image-desc"), + "size": root_disk.get("size-of-storage"), + "keep": Ns.is_volume_keeping_required(root_disk), + } - # TODO mac_address: used for SR-IOV ifaces #TODO for other types - # TODO floating_ip: True/False (or it can be None) - if interface.get("type") in ("SR-IOV", "PCI-PASSTHROUGH"): - # mark the net create task as type data - if deep_get( - tasks_by_target_record_id, - net_text, - "extra_dict", - "params", - "net_type", - ): - tasks_by_target_record_id[net_text]["extra_dict"]["params"][ - "net_type" - ] = "data" + disk_list.append(persistent_root_disk[vsd["id"]]) - net_item["use"] = "data" - net_item["model"] = interface["type"] - net_item["type"] = interface["type"] - elif ( - interface.get("type") == "OM-MGMT" - or interface.get("mgmt-interface") - or interface.get("mgmt-vnf") + return persistent_root_disk + + @staticmethod + def find_persistent_volumes( + persistent_root_disk: dict, + target_vdu: dict, + vdu_instantiation_volumes_list: list, + disk_list: list, + ) -> None: + """Find the ordinary persistent volumes and add them to the disk_list + by parsing the instantiation parameters. + + Args: + persistent_root_disk: persistent root disk dictionary + target_vdu: processed VDU + vdu_instantiation_volumes_list: instantiation parameters for the each VDU as a list + disk_list: to be filled up + + """ + # Find the ordinary volumes which are not added to the persistent_root_disk + persistent_disk = {} + for disk in target_vdu.get("virtual-storages", {}): + if ( + disk.get("type-of-storage") == "persistent-storage:persistent-storage" + and disk["id"] not in persistent_root_disk.keys() ): - net_item["use"] = "mgmt" - else: - # if interface.get("type") in ("VIRTIO", "E1000", "PARAVIRT"): - net_item["use"] = "bridge" - net_item["model"] = interface.get("type") + for vdu_volume in vdu_instantiation_volumes_list: - if interface.get("ip-address"): - net_item["ip_address"] = interface["ip-address"] + if vdu_volume["vim-volume-id"] and disk["id"] == vdu_volume["name"]: - if interface.get("mac-address"): - net_item["mac_address"] = interface["mac-address"] + persistent_disk[disk["id"]] = { + "vim_volume_id": vdu_volume["vim-volume-id"], + } + disk_list.append(persistent_disk[disk["id"]]) - net_list.append(net_item) + else: + if disk["id"] not in persistent_disk.keys(): + persistent_disk[disk["id"]] = { + "size": disk.get("size-of-storage"), + "keep": Ns.is_volume_keeping_required(disk), + } + disk_list.append(persistent_disk[disk["id"]]) - if interface.get("mgmt-vnf"): - extra_dict["mgmt_vnf_interface"] = iface_index - elif interface.get("mgmt-interface"): - extra_dict["mgmt_vdu_interface"] = iface_index + @staticmethod + def is_volume_keeping_required(virtual_storage_desc: Dict[str, Any]) -> bool: + """Function to decide keeping persistent volume + upon VDU deletion. + Args: + virtual_storage_desc (Dict[str, Any]): virtual storage description dictionary + + Returns: + bool (True/False) + """ + + if not virtual_storage_desc.get("vdu-storage-requirements"): + return False + for item in virtual_storage_desc.get("vdu-storage-requirements", {}): + if item.get("key") == "keep-volume" and item.get("value") == "true": + return True + return False + + @staticmethod + def _sort_vdu_interfaces(target_vdu: dict) -> None: + """Sort the interfaces according to position number. + + Args: + target_vdu (dict): Details of VDU to be created + + """ + # If the position info is provided for all the interfaces, it will be sorted + # according to position number ascendingly. + sorted_interfaces = sorted( + target_vdu["interfaces"], + key=lambda x: (x.get("position") is None, x.get("position")), + ) + target_vdu["interfaces"] = sorted_interfaces + + @staticmethod + def _partially_locate_vdu_interfaces(target_vdu: dict) -> None: + """Only place the interfaces which has specific position. + + Args: + target_vdu (dict): Details of VDU to be created + + """ + # If the position info is provided for some interfaces but not all of them, the interfaces + # which has specific position numbers will be placed and others' positions will not be taken care. + if any( + i.get("position") + 1 + for i in target_vdu["interfaces"] + if i.get("position") is not None + ): + n = len(target_vdu["interfaces"]) + sorted_interfaces = [-1] * n + k, m = 0, 0 + + while k < n: + if target_vdu["interfaces"][k].get("position") is not None: + if any(i.get("position") == 0 for i in target_vdu["interfaces"]): + idx = target_vdu["interfaces"][k]["position"] + 1 + else: + idx = target_vdu["interfaces"][k]["position"] + sorted_interfaces[idx - 1] = target_vdu["interfaces"][k] + k += 1 + + while m < n: + if target_vdu["interfaces"][m].get("position") is None: + idy = sorted_interfaces.index(-1) + sorted_interfaces[idy] = target_vdu["interfaces"][m] + m += 1 + + target_vdu["interfaces"] = sorted_interfaces + + @staticmethod + def _prepare_vdu_cloud_init( + target_vdu: dict, vdu2cloud_init: dict, db: object, fs: object + ) -> Dict: + """Fill cloud_config dict with cloud init details. + + Args: + target_vdu (dict): Details of VDU to be created + vdu2cloud_init (dict): Cloud init dict + db (object): DB object + fs (object): FS object + + Returns: + cloud_config (dict): Cloud config details of VDU + + """ # cloud config cloud_config = {} @@ -1060,6 +1189,191 @@ class Ns(object): if target_vdu.get("boot-data-drive"): cloud_config["boot-data-drive"] = target_vdu.get("boot-data-drive") + return cloud_config + + @staticmethod + def _check_vld_information_of_interfaces( + interface: dict, ns_preffix: str, vnf_preffix: str + ) -> Optional[str]: + """Prepare the net_text by the virtual link information for vnf and ns level. + Args: + interface (dict): Interface details + ns_preffix (str): Prefix of NS + vnf_preffix (str): Prefix of VNF + + Returns: + net_text (str): information of net + + """ + net_text = "" + if interface.get("ns-vld-id"): + net_text = ns_preffix + ":vld." + interface["ns-vld-id"] + elif interface.get("vnf-vld-id"): + net_text = vnf_preffix + ":vld." + interface["vnf-vld-id"] + + return net_text + + @staticmethod + def _prepare_interface_port_security(interface: dict) -> None: + """ + + Args: + interface (dict): Interface details + + """ + if "port-security-enabled" in interface: + interface["port_security"] = interface.pop("port-security-enabled") + + if "port-security-disable-strategy" in interface: + interface["port_security_disable_strategy"] = interface.pop( + "port-security-disable-strategy" + ) + + @staticmethod + def _create_net_item_of_interface(interface: dict, net_text: str) -> dict: + """Prepare net item including name, port security, floating ip etc. + + Args: + interface (dict): Interface details + net_text (str): information of net + + Returns: + net_item (dict): Dict including net details + + """ + + net_item = { + x: v + for x, v in interface.items() + if x + in ( + "name", + "vpci", + "port_security", + "port_security_disable_strategy", + "floating_ip", + ) + } + net_item["net_id"] = "TASK-" + net_text + net_item["type"] = "virtual" + + return net_item + + @staticmethod + def _prepare_type_of_interface( + interface: dict, tasks_by_target_record_id: dict, net_text: str, net_item: dict + ) -> None: + """Fill the net item type by interface type such as SR-IOV, OM-MGMT, bridge etc. + + Args: + interface (dict): Interface details + tasks_by_target_record_id (dict): Task details + net_text (str): information of net + net_item (dict): Dict including net details + + """ + # TODO mac_address: used for SR-IOV ifaces #TODO for other types + # TODO floating_ip: True/False (or it can be None) + + if interface.get("type") in ("SR-IOV", "PCI-PASSTHROUGH"): + # Mark the net create task as type data + if deep_get( + tasks_by_target_record_id, + net_text, + "extra_dict", + "params", + "net_type", + ): + tasks_by_target_record_id[net_text]["extra_dict"]["params"][ + "net_type" + ] = "data" + + net_item["use"] = "data" + net_item["model"] = interface["type"] + net_item["type"] = interface["type"] + + elif ( + interface.get("type") == "OM-MGMT" + or interface.get("mgmt-interface") + or interface.get("mgmt-vnf") + ): + net_item["use"] = "mgmt" + + else: + # If interface.get("type") in ("VIRTIO", "E1000", "PARAVIRT"): + net_item["use"] = "bridge" + net_item["model"] = interface.get("type") + + @staticmethod + def _prepare_vdu_interfaces( + target_vdu: dict, + extra_dict: dict, + ns_preffix: str, + vnf_preffix: str, + logger: object, + tasks_by_target_record_id: dict, + net_list: list, + ) -> None: + """Prepare the net_item and add net_list, add mgmt interface to extra_dict. + + Args: + target_vdu (dict): VDU to be created + extra_dict (dict): Dictionary to be filled + ns_preffix (str): NS prefix as string + vnf_preffix (str): VNF prefix as string + logger (object): Logger Object + tasks_by_target_record_id (dict): Task details + net_list (list): Net list of VDU + """ + for iface_index, interface in enumerate(target_vdu["interfaces"]): + + net_text = Ns._check_vld_information_of_interfaces( + interface, ns_preffix, vnf_preffix + ) + if not net_text: + # Interface not connected to any vld + logger.error( + "Interface {} from vdu {} not connected to any vld".format( + iface_index, target_vdu["vdu-name"] + ) + ) + continue + + extra_dict["depends_on"].append(net_text) + + Ns._prepare_interface_port_security(interface) + + net_item = Ns._create_net_item_of_interface(interface, net_text) + + Ns._prepare_type_of_interface( + interface, tasks_by_target_record_id, net_text, net_item + ) + + if interface.get("ip-address"): + net_item["ip_address"] = interface["ip-address"] + + if interface.get("mac-address"): + net_item["mac_address"] = interface["mac-address"] + + net_list.append(net_item) + + if interface.get("mgmt-vnf"): + extra_dict["mgmt_vnf_interface"] = iface_index + elif interface.get("mgmt-interface"): + extra_dict["mgmt_vdu_interface"] = iface_index + + @staticmethod + def _prepare_vdu_ssh_keys( + target_vdu: dict, ro_nsr_public_key: dict, cloud_config: dict + ) -> None: + """Add ssh keys to cloud config. + + Args: + target_vdu (dict): Details of VDU to be created + ro_nsr_public_key (dict): RO NSR public Key + cloud_config (dict): Cloud config details + + """ ssh_keys = [] if target_vdu.get("ssh-keys"): @@ -1071,26 +1385,70 @@ class Ns(object): if ssh_keys: cloud_config["key-pairs"] = ssh_keys - persistent_root_disk = {} - disk_list = [] - vnfd_id = vnfr["vnfd-id"] - vnfd = db.get_one("vnfds", {"_id": vnfd_id}) + @staticmethod + def _select_persistent_root_disk(vsd: dict, vdu: dict) -> dict: + """Selects the persistent root disk if exists. + Args: + vsd (dict): Virtual storage descriptors in VNFD + vdu (dict): VNF descriptor + + Returns: + root_disk (dict): Selected persistent root disk + """ + if vsd.get("id") == vdu.get("virtual-storage-desc", [[]])[0]: + root_disk = vsd + if root_disk.get( + "type-of-storage" + ) == "persistent-storage:persistent-storage" and root_disk.get( + "size-of-storage" + ): + return root_disk + + @staticmethod + def _add_persistent_root_disk_to_disk_list( + vnfd: dict, target_vdu: dict, persistent_root_disk: dict, disk_list: list + ) -> None: + """Find the persistent root disk and add to disk list. + + Args: + vnfd (dict): VNF descriptor + target_vdu (dict): Details of VDU to be created + persistent_root_disk (dict): Details of persistent root disk + disk_list (list): Disks of VDU + + """ for vdu in vnfd.get("vdu", ()): if vdu["name"] == target_vdu["vdu-name"]: for vsd in vnfd.get("virtual-storage-desc", ()): - if vsd.get("id") == vdu.get("virtual-storage-desc", [[]])[0]: - root_disk = vsd - if root_disk.get( - "type-of-storage" - ) == "persistent-storage:persistent-storage" and root_disk.get( - "size-of-storage" - ): - persistent_root_disk[vsd["id"]] = { - "image_id": vdu.get("sw-image-desc"), - "size": root_disk["size-of-storage"], - } - disk_list.append(persistent_root_disk[vsd["id"]]) + root_disk = Ns._select_persistent_root_disk(vsd, vdu) + if not root_disk: + continue + + persistent_root_disk[vsd["id"]] = { + "image_id": vdu.get("sw-image-desc"), + "size": root_disk["size-of-storage"], + "keep": Ns.is_volume_keeping_required(root_disk), + } + disk_list.append(persistent_root_disk[vsd["id"]]) + break + + @staticmethod + def _add_persistent_ordinary_disks_to_disk_list( + target_vdu: dict, + persistent_root_disk: dict, + persistent_ordinary_disk: dict, + disk_list: list, + ) -> None: + """Fill the disk list by adding persistent ordinary disks. + + Args: + target_vdu (dict): Details of VDU to be created + persistent_root_disk (dict): Details of persistent root disk + persistent_ordinary_disk (dict): Details of persistent ordinary disk + disk_list (list): Disks of VDU + + """ if target_vdu.get("virtual-storages"): for disk in target_vdu["virtual-storages"]: if ( @@ -1098,21 +1456,158 @@ class Ns(object): == "persistent-storage:persistent-storage" and disk["id"] not in persistent_root_disk.keys() ): - disk_list.append({"size": disk["size-of-storage"]}) + persistent_ordinary_disk[disk["id"]] = { + "size": disk["size-of-storage"], + "keep": Ns.is_volume_keeping_required(disk), + } + disk_list.append(persistent_ordinary_disk[disk["id"]]) + @staticmethod + def _prepare_vdu_affinity_group_list( + target_vdu: dict, extra_dict: dict, ns_preffix: str + ) -> List[Dict[str, any]]: + """Process affinity group details to prepare affinity group list. + + Args: + target_vdu (dict): Details of VDU to be created + extra_dict (dict): Dictionary to be filled + ns_preffix (str): Prefix as string + + Returns: + + affinity_group_list (list): Affinity group details + + """ affinity_group_list = [] if target_vdu.get("affinity-or-anti-affinity-group-id"): - affinity_group = {} for affinity_group_id in target_vdu["affinity-or-anti-affinity-group-id"]: + affinity_group = {} affinity_group_text = ( ns_preffix + ":affinity-or-anti-affinity-group." + affinity_group_id ) + if not isinstance(extra_dict.get("depends_on"), list): + raise NsException("Invalid extra_dict format.") + extra_dict["depends_on"].append(affinity_group_text) affinity_group["affinity_group_id"] = "TASK-" + affinity_group_text affinity_group_list.append(affinity_group) + return affinity_group_list + + @staticmethod + def _process_vdu_params( + target_vdu: Dict[str, Any], + indata: Dict[str, Any], + vim_info: Dict[str, Any], + target_record_id: str, + **kwargs: Dict[str, Any], + ) -> Dict[str, Any]: + """Function to process VDU parameters. + + Args: + target_vdu (Dict[str, Any]): [description] + indata (Dict[str, Any]): [description] + vim_info (Dict[str, Any]): [description] + target_record_id (str): [description] + + Returns: + Dict[str, Any]: [description] + """ + vnfr_id = kwargs.get("vnfr_id") + nsr_id = kwargs.get("nsr_id") + vnfr = kwargs.get("vnfr") + vdu2cloud_init = kwargs.get("vdu2cloud_init") + tasks_by_target_record_id = kwargs.get("tasks_by_target_record_id") + logger = kwargs.get("logger") + db = kwargs.get("db") + fs = kwargs.get("fs") + ro_nsr_public_key = kwargs.get("ro_nsr_public_key") + + vnf_preffix = "vnfrs:{}".format(vnfr_id) + ns_preffix = "nsrs:{}".format(nsr_id) + image_text = ns_preffix + ":image." + target_vdu["ns-image-id"] + flavor_text = ns_preffix + ":flavor." + target_vdu["ns-flavor-id"] + extra_dict = {"depends_on": [image_text, flavor_text]} + net_list = [] + + persistent_root_disk = {} + persistent_ordinary_disk = {} + vdu_instantiation_volumes_list = [] + disk_list = [] + vnfd_id = vnfr["vnfd-id"] + vnfd = db.get_one("vnfds", {"_id": vnfd_id}) + + # If the position info is provided for all the interfaces, it will be sorted + # according to position number ascendingly. + if all( + True if i.get("position") is not None else False + for i in target_vdu["interfaces"] + ): + + Ns._sort_vdu_interfaces(target_vdu) + + # If the position info is provided for some interfaces but not all of them, the interfaces + # which has specific position numbers will be placed and others' positions will not be taken care. + else: + + Ns._partially_locate_vdu_interfaces(target_vdu) + + # If the position info is not provided for the interfaces, interfaces will be attached + # according to the order in the VNFD. + Ns._prepare_vdu_interfaces( + target_vdu, + extra_dict, + ns_preffix, + vnf_preffix, + logger, + tasks_by_target_record_id, + net_list, + ) + + # cloud config + cloud_config = Ns._prepare_vdu_cloud_init(target_vdu, vdu2cloud_init, db, fs) + + # Prepare VDU ssh keys + Ns._prepare_vdu_ssh_keys(target_vdu, ro_nsr_public_key, cloud_config) + + if target_vdu.get("additionalParams"): + vdu_instantiation_volumes_list = ( + target_vdu.get("additionalParams").get("OSM").get("vdu_volumes") + ) + + if vdu_instantiation_volumes_list: + + # Find the root volumes and add to the disk_list + persistent_root_disk = Ns.find_persistent_root_volumes( + vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list + ) + + # Find the ordinary volumes which are not added to the persistent_root_disk + # and put them to the disk list + Ns.find_persistent_volumes( + persistent_root_disk, + target_vdu, + vdu_instantiation_volumes_list, + disk_list, + ) + + else: + # Vdu_instantiation_volumes_list is empty + # First get add the persistent root disks to disk_list + Ns._add_persistent_root_disk_to_disk_list( + vnfd, target_vdu, persistent_root_disk, disk_list + ) + # Add the persistent non-root disks to disk_list + Ns._add_persistent_ordinary_disks_to_disk_list( + target_vdu, persistent_root_disk, persistent_ordinary_disk, disk_list + ) + + affinity_group_list = Ns._prepare_vdu_affinity_group_list( + target_vdu, extra_dict, ns_preffix + ) + extra_dict["params"] = { "name": "{}-{}-{}-{}".format( indata["name"][:16], @@ -1230,7 +1725,9 @@ class Ns(object): "floating_ip", ) } - existing_ifaces = existing_vdu["vim_info"][target_id].get("interfaces", []) + existing_ifaces = existing_vdu["vim_info"][target_id].get( + "interfaces_backup", [] + ) net_id = next( ( i["vim_net_id"] @@ -1455,7 +1952,7 @@ class Ns(object): target_record_id = "{}.{}".format(db_record, existing_item["id"]) item_ = item - if target_vim.startswith("sdn"): + if target_vim.startswith("sdn") or target_vim.startswith("wim"): # item must be sdn-net instead of net if target_vim is a sdn item_ = "sdn_net" target_record_id += ".sdn" @@ -1503,19 +2000,27 @@ class Ns(object): target_record_id = "{}.{}".format(db_record, target_item["id"]) item_ = item - if target_vim.startswith("sdn"): + if target_vim.startswith("sdn") or target_vim.startswith("wim"): # item must be sdn-net instead of net if target_vim is a sdn item_ = "sdn_net" target_record_id += ".sdn" kwargs = {} - self.logger.warning( + self.logger.debug( "ns.calculate_diff_items target_item={}".format(target_item) ) - if process_params == Ns._process_vdu_params: - self.logger.warning( - "calculate_diff_items self.fs={}".format(self.fs) + if process_params == Ns._process_flavor_params: + kwargs.update( + { + "db": self.db, + } + ) + self.logger.debug( + "calculate_diff_items for flavor kwargs={}".format(kwargs) ) + + if process_params == Ns._process_vdu_params: + self.logger.debug("calculate_diff_items self.fs={}".format(self.fs)) kwargs.update( { "vnfr_id": vnfr_id, @@ -1529,7 +2034,7 @@ class Ns(object): "ro_nsr_public_key": ro_nsr_public_key, } ) - self.logger.warning("calculate_diff_items kwargs={}".format(kwargs)) + self.logger.debug("calculate_diff_items kwargs={}".format(kwargs)) extra_dict = process_params( target_item, @@ -1667,7 +2172,7 @@ class Ns(object): extra_dict=change.get("extra_dict", None), ) - self.logger.warning("ns.define_all_tasks task={}".format(task)) + self.logger.debug("ns.define_all_tasks task={}".format(task)) tasks_by_target_record_id[change["target_record_id"]] = task db_new_tasks.append(task) @@ -1693,12 +2198,17 @@ class Ns(object): target_id = db_task.pop("target_id") common_id = db_task.get("common_id") + # Do not chek tasks with vim_status DELETED + # because in manual heealing there are two tasks for the same vdur: + # one with vim_status deleted and the other one with the actual VM status. + if common_id: if self.db.set_one( "ro_tasks", q_filter={ "target_id": target_id, "tasks.common_id": common_id, + "vim_info.vim_status.ne": "DELETED", }, update_dict={"to_check_at": now, "modified_at": now}, push={"tasks": db_task}, @@ -1711,6 +2221,7 @@ class Ns(object): q_filter={ "target_id": target_id, "tasks.target_record": db_task["target_record"], + "vim_info.vim_status.ne": "DELETED", }, update_dict={"to_check_at": now, "modified_at": now}, push={"tasks": db_task}, @@ -1745,7 +2256,7 @@ class Ns(object): for db_task in db_new_tasks: target_id = db_task.pop("target_id") - self.logger.warning("target_id={} db_task={}".format(target_id, db_task)) + self.logger.debug("target_id={} db_task={}".format(target_id, db_task)) action = db_task.get("action", None) @@ -1760,10 +2271,13 @@ class Ns(object): db_ro_task["vim_info"]["created_items"] = db_task.get( "created_items", {} ) + db_ro_task["vim_info"]["volumes_to_hold"] = db_task.get( + "volumes_to_hold", [] + ) db_ro_task["vim_info"]["vim_id"] = db_task.get("vim_id", None) nb_ro_tasks += 1 - self.logger.warning("upload_all_tasks db_ro_task={}".format(db_ro_task)) + self.logger.debug("upload_all_tasks db_ro_task={}".format(db_ro_task)) self.db.create("ro_tasks", db_ro_task) self.logger.debug( @@ -1774,16 +2288,23 @@ class Ns(object): def _prepare_created_items_for_healing( self, - target_id, - existing_vdu, + nsr_id, + target_record, ): - # Only ports are considered because created volumes are persistent - ports_list = {} - vim_interfaces = existing_vdu["vim_info"][target_id].get("interfaces", []) - for iface in vim_interfaces: - ports_list["port:" + iface["vim_interface_id"]] = True + created_items = {} + # Get created_items from ro_task + ro_tasks = self.db.get_list("ro_tasks", {"tasks.nsr_id": nsr_id}) + for ro_task in ro_tasks: + for task in ro_task["tasks"]: + if ( + task["target_record"] == target_record + and task["action"] == "CREATE" + and ro_task["vim_info"]["created_items"] + ): + created_items = ro_task["vim_info"]["created_items"] + break - return ports_list + return created_items def _prepare_persistent_volumes_for_healing( self, @@ -1871,7 +2392,7 @@ class Ns(object): target_record = f"{db_record}.{item_index}.vim_info.{target_vim}" created_items = self._prepare_created_items_for_healing( - target_vim, existing_instance + nsr_id, target_record ) volumes_to_hold = self._prepare_persistent_volumes_for_healing( @@ -1927,6 +2448,20 @@ class Ns(object): # The CREATE task depens on the DELETE task extra_dict["depends_on"] = [delete_task_id] + # Add volumes created from created_items if any + # Ports should be deleted with delete task and automatically created with create task + volumes = {} + for k, v in created_items.items(): + try: + k_item, _, k_id = k.partition(":") + if k_item == "volume": + volumes[k] = v + except Exception as e: + self.logger.error( + "Error evaluating created item {}: {}".format(k, e) + ) + extra_dict["previous_created_volumes"] = volumes + deployment_info = { "action_id": action_id, "nsr_id": nsr_id, @@ -2316,6 +2851,99 @@ class Ns(object): return None, None, True + def rebuild_start_stop_task( + self, + vdu_id, + vnf_id, + vdu_index, + action_id, + nsr_id, + task_index, + target_vim, + extra_dict, + ): + self._assign_vim(target_vim) + target_record = "vnfrs:{}:vdur.{}".format(vnf_id, vdu_index) + target_record_id = "vnfrs:{}:vdur.{}".format(vnf_id, vdu_id) + deployment_info = { + "action_id": action_id, + "nsr_id": nsr_id, + "task_index": task_index, + } + + task = Ns._create_task( + deployment_info=deployment_info, + target_id=target_vim, + item="update", + action="EXEC", + target_record=target_record, + target_record_id=target_record_id, + extra_dict=extra_dict, + ) + return task + + def rebuild_start_stop( + self, session, action_dict, version, nsr_id, *args, **kwargs + ): + task_index = 0 + extra_dict = {} + now = time() + action_id = action_dict.get("action_id", str(uuid4())) + step = "" + logging_text = "Task deploy nsr_id={} action_id={} ".format(nsr_id, action_id) + self.logger.debug(logging_text + "Enter") + + action = list(action_dict.keys())[0] + task_dict = action_dict.get(action) + vim_vm_id = action_dict.get(action).get("vim_vm_id") + + if action_dict.get("stop"): + action = "shutoff" + db_new_tasks = [] + try: + step = "lock the operation & do task creation" + with self.write_lock: + extra_dict["params"] = { + "vim_vm_id": vim_vm_id, + "action": action, + } + task = self.rebuild_start_stop_task( + task_dict["vdu_id"], + task_dict["vnf_id"], + task_dict["vdu_index"], + action_id, + nsr_id, + task_index, + task_dict["target_vim"], + extra_dict, + ) + db_new_tasks.append(task) + step = "upload Task to db" + self.upload_all_tasks( + db_new_tasks=db_new_tasks, + now=now, + ) + self.logger.debug( + logging_text + "Exit. Created {} tasks".format(len(db_new_tasks)) + ) + return ( + {"status": "ok", "nsr_id": nsr_id, "action_id": action_id}, + action_id, + True, + ) + except Exception as e: + if isinstance(e, (DbException, NsException)): + self.logger.error( + logging_text + "Exit Exception while '{}': {}".format(step, e) + ) + else: + e = traceback_format_exc() + self.logger.critical( + logging_text + "Exit Exception while '{}': {}".format(step, e), + exc_info=True, + ) + raise NsException(e) + def get_deploy(self, session, indata, version, nsr_id, action_id, *args, **kwargs): nsrs = self.db.get_list("nsrs", {}) return_data = [] @@ -2335,3 +2963,215 @@ class Ns(object): return_data.append(task["action_id"]) return return_data, None, True + + def migrate_task( + self, vdu, vnf, vdu_index, action_id, nsr_id, task_index, extra_dict + ): + target_vim, vim_info = next(k_v for k_v in vdu["vim_info"].items()) + self._assign_vim(target_vim) + target_record = "vnfrs:{}:vdur.{}".format(vnf["_id"], vdu_index) + target_record_id = "vnfrs:{}:vdur.{}".format(vnf["_id"], vdu["id"]) + deployment_info = { + "action_id": action_id, + "nsr_id": nsr_id, + "task_index": task_index, + } + + task = Ns._create_task( + deployment_info=deployment_info, + target_id=target_vim, + item="migrate", + action="EXEC", + target_record=target_record, + target_record_id=target_record_id, + extra_dict=extra_dict, + ) + + return task + + def migrate(self, session, indata, version, nsr_id, *args, **kwargs): + task_index = 0 + extra_dict = {} + now = time() + action_id = indata.get("action_id", str(uuid4())) + step = "" + logging_text = "Task deploy nsr_id={} action_id={} ".format(nsr_id, action_id) + self.logger.debug(logging_text + "Enter") + try: + vnf_instance_id = indata["vnfInstanceId"] + step = "Getting vnfrs from db" + db_vnfr = self.db.get_one("vnfrs", {"_id": vnf_instance_id}) + vdu = indata.get("vdu") + migrateToHost = indata.get("migrateToHost") + db_new_tasks = [] + + with self.write_lock: + if vdu is not None: + vdu_id = indata["vdu"]["vduId"] + vdu_count_index = indata["vdu"].get("vduCountIndex", 0) + for vdu_index, vdu in enumerate(db_vnfr["vdur"]): + if ( + vdu["vdu-id-ref"] == vdu_id + and vdu["count-index"] == vdu_count_index + ): + extra_dict["params"] = { + "vim_vm_id": vdu["vim-id"], + "migrate_host": migrateToHost, + "vdu_vim_info": vdu["vim_info"], + } + step = "Creating migration task for vdu:{}".format(vdu) + task = self.migrate_task( + vdu, + db_vnfr, + vdu_index, + action_id, + nsr_id, + task_index, + extra_dict, + ) + db_new_tasks.append(task) + task_index += 1 + break + else: + + for vdu_index, vdu in enumerate(db_vnfr["vdur"]): + extra_dict["params"] = { + "vim_vm_id": vdu["vim-id"], + "migrate_host": migrateToHost, + "vdu_vim_info": vdu["vim_info"], + } + step = "Creating migration task for vdu:{}".format(vdu) + task = self.migrate_task( + vdu, + db_vnfr, + vdu_index, + action_id, + nsr_id, + task_index, + extra_dict, + ) + db_new_tasks.append(task) + task_index += 1 + + self.upload_all_tasks( + db_new_tasks=db_new_tasks, + now=now, + ) + + self.logger.debug( + logging_text + "Exit. Created {} tasks".format(len(db_new_tasks)) + ) + return ( + {"status": "ok", "nsr_id": nsr_id, "action_id": action_id}, + action_id, + True, + ) + except Exception as e: + if isinstance(e, (DbException, NsException)): + self.logger.error( + logging_text + "Exit Exception while '{}': {}".format(step, e) + ) + else: + e = traceback_format_exc() + self.logger.critical( + logging_text + "Exit Exception while '{}': {}".format(step, e), + exc_info=True, + ) + raise NsException(e) + + def verticalscale_task( + self, vdu, vnf, vdu_index, action_id, nsr_id, task_index, extra_dict + ): + target_vim, vim_info = next(k_v for k_v in vdu["vim_info"].items()) + self._assign_vim(target_vim) + target_record = "vnfrs:{}:vdur.{}".format(vnf["_id"], vdu_index) + target_record_id = "vnfrs:{}:vdur.{}".format(vnf["_id"], vdu["id"]) + deployment_info = { + "action_id": action_id, + "nsr_id": nsr_id, + "task_index": task_index, + } + + task = Ns._create_task( + deployment_info=deployment_info, + target_id=target_vim, + item="verticalscale", + action="EXEC", + target_record=target_record, + target_record_id=target_record_id, + extra_dict=extra_dict, + ) + return task + + def verticalscale(self, session, indata, version, nsr_id, *args, **kwargs): + task_index = 0 + extra_dict = {} + now = time() + action_id = indata.get("action_id", str(uuid4())) + step = "" + logging_text = "Task deploy nsr_id={} action_id={} ".format(nsr_id, action_id) + self.logger.debug(logging_text + "Enter") + try: + VnfFlavorData = indata.get("changeVnfFlavorData") + vnf_instance_id = VnfFlavorData["vnfInstanceId"] + step = "Getting vnfrs from db" + db_vnfr = self.db.get_one("vnfrs", {"_id": vnf_instance_id}) + vduid = VnfFlavorData["additionalParams"]["vduid"] + vduCountIndex = VnfFlavorData["additionalParams"]["vduCountIndex"] + virtualMemory = VnfFlavorData["additionalParams"]["virtualMemory"] + numVirtualCpu = VnfFlavorData["additionalParams"]["numVirtualCpu"] + sizeOfStorage = VnfFlavorData["additionalParams"]["sizeOfStorage"] + flavor_dict = { + "name": vduid + "-flv", + "ram": virtualMemory, + "vcpus": numVirtualCpu, + "disk": sizeOfStorage, + } + db_new_tasks = [] + step = "Creating Tasks for vertical scaling" + with self.write_lock: + for vdu_index, vdu in enumerate(db_vnfr["vdur"]): + if ( + vdu["vdu-id-ref"] == vduid + and vdu["count-index"] == vduCountIndex + ): + extra_dict["params"] = { + "vim_vm_id": vdu["vim-id"], + "flavor_dict": flavor_dict, + } + task = self.verticalscale_task( + vdu, + db_vnfr, + vdu_index, + action_id, + nsr_id, + task_index, + extra_dict, + ) + db_new_tasks.append(task) + task_index += 1 + break + self.upload_all_tasks( + db_new_tasks=db_new_tasks, + now=now, + ) + self.logger.debug( + logging_text + "Exit. Created {} tasks".format(len(db_new_tasks)) + ) + return ( + {"status": "ok", "nsr_id": nsr_id, "action_id": action_id}, + action_id, + True, + ) + except Exception as e: + if isinstance(e, (DbException, NsException)): + self.logger.error( + logging_text + "Exit Exception while '{}': {}".format(step, e) + ) + else: + e = traceback_format_exc() + self.logger.critical( + logging_text + "Exit Exception while '{}': {}".format(step, e), + exc_info=True, + ) + raise NsException(e)