+ 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"):
+ ssh_keys += target_vdu.get("ssh-keys")
+
+ if target_vdu.get("ssh-access-required"):
+ ssh_keys.append(ro_nsr_public_key)
+
+ if ssh_keys:
+ cloud_config["key-pairs"] = ssh_keys
+
+ @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", ()):
+ 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,
+ extra_dict: dict,
+ vnf_id: str = None,
+ nsr_id: str = None,
+ ) -> 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 (
+ disk.get("type-of-storage")
+ == "persistent-storage:persistent-storage"
+ and disk["id"] not in persistent_root_disk.keys()
+ ):
+ name, multiattach = Ns.is_shared_volume(disk, vnf_id)
+ persistent_ordinary_disk[disk["id"]] = {
+ "name": name,
+ "size": disk["size-of-storage"],
+ "keep": Ns.is_volume_keeping_required(disk),
+ "multiattach": multiattach,
+ }
+ disk_list.append(persistent_ordinary_disk[disk["id"]])
+ if multiattach: # VDU creation has to wait for shared volumes
+ extra_dict["depends_on"].append(
+ f"nsrs:{nsr_id}:shared-volumes.{name}"
+ )
+
+ @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"):
+ 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"]
+ extra_dict = {"depends_on": [image_text]}
+ net_list = []
+ persistent_root_disk = {}
+ persistent_ordinary_disk = {}
+ vdu_instantiation_volumes_list = []
+ vdu_instantiation_flavor_id = None
+ 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")
+ )
+ vdu_instantiation_flavor_id = (
+ target_vdu.get("additionalParams").get("OSM", {}).get("vim_flavor_id")
+ )
+
+ # flavor id
+ if vdu_instantiation_flavor_id:
+ flavor_id = vdu_instantiation_flavor_id
+ else:
+ flavor_text = ns_preffix + ":flavor." + target_vdu["ns-flavor-id"]
+ flavor_id = "TASK-" + flavor_text
+ extra_dict["depends_on"].append(flavor_text)
+
+ 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,
+ extra_dict,
+ vnfd["id"],
+ nsr_id,
+ )
+
+ affinity_group_list = Ns._prepare_vdu_affinity_group_list(
+ target_vdu, extra_dict, ns_preffix
+ )
+
+ extra_dict["params"] = {
+ "name": "{}-{}-{}-{}".format(
+ indata["name"][:16],
+ vnfr["member-vnf-index-ref"][:16],
+ target_vdu["vdu-name"][:32],
+ target_vdu.get("count-index") or 0,
+ ),
+ "description": target_vdu["vdu-name"],
+ "start": True,
+ "image_id": "TASK-" + image_text,
+ "flavor_id": flavor_id,
+ "affinity_group_list": affinity_group_list,
+ "net_list": net_list,
+ "cloud_config": cloud_config or None,
+ "disk_list": disk_list,
+ "availability_zone_index": None, # TODO
+ "availability_zone_list": None, # TODO
+ }
+ return extra_dict
+
+ @staticmethod
+ def _process_shared_volumes_params(
+ target_shared_volume: Dict[str, Any],
+ indata: Dict[str, Any],
+ vim_info: Dict[str, Any],
+ target_record_id: str,
+ **kwargs: Dict[str, Any],
+ ) -> Dict[str, Any]:
+ extra_dict = {}
+ shared_volume_data = {
+ "size": target_shared_volume["size-of-storage"],
+ "name": target_shared_volume["id"],
+ "type": target_shared_volume["type-of-storage"],
+ "keep": Ns.is_volume_keeping_required(target_shared_volume),
+ }
+ extra_dict["params"] = shared_volume_data
+ return extra_dict
+
+ @staticmethod
+ def _process_affinity_group_params(
+ target_affinity_group: Dict[str, Any],
+ indata: Dict[str, Any],
+ vim_info: Dict[str, Any],
+ target_record_id: str,