+ Args:
+ ip_profile (Dict[str, Any]): [description]
+
+ Returns:
+ Dict[str, Any]: [description]
+ """
+ if not ip_profile:
+ return None
+
+ ro_ip_profile = {
+ "ip_version": "IPv4"
+ if "v4" in ip_profile.get("ip-version", "ipv4")
+ else "IPv6",
+ "subnet_address": ip_profile.get("subnet-address"),
+ "gateway_address": ip_profile.get("gateway-address"),
+ "dhcp_enabled": ip_profile.get("dhcp-params", {}).get("enabled", False),
+ "dhcp_start_address": ip_profile.get("dhcp-params", {}).get(
+ "start-address", None
+ ),
+ "dhcp_count": ip_profile.get("dhcp-params", {}).get("count", None),
+ }
+
+ if ip_profile.get("dns-server"):
+ ro_ip_profile["dns_address"] = ";".join(
+ [v["address"] for v in ip_profile["dns-server"] if v.get("address")]
+ )
+
+ if ip_profile.get("security-group"):
+ ro_ip_profile["security_group"] = ip_profile["security-group"]
+
+ return ro_ip_profile
+
+ @staticmethod
+ def _process_net_params(
+ target_vld: 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 network parameters.
+
+ Args:
+ target_vld (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]
+ """
+ extra_dict = {}
+
+ if vim_info.get("sdn"):
+ # vnf_preffix = "vnfrs:{}".format(vnfr_id)
+ # ns_preffix = "nsrs:{}".format(nsr_id)
+ # remove the ending ".sdn
+ vld_target_record_id, _, _ = target_record_id.rpartition(".")
+ extra_dict["params"] = {
+ k: vim_info[k]
+ for k in ("sdn-ports", "target_vim", "vlds", "type")
+ if vim_info.get(k)
+ }
+
+ # TODO needed to add target_id in the dependency.
+ if vim_info.get("target_vim"):
+ extra_dict["depends_on"] = [
+ f"{vim_info.get('target_vim')} {vld_target_record_id}"
+ ]
+
+ return extra_dict
+
+ if vim_info.get("vim_network_name"):
+ extra_dict["find_params"] = {
+ "filter_dict": {
+ "name": vim_info.get("vim_network_name"),
+ },
+ }
+ elif vim_info.get("vim_network_id"):
+ extra_dict["find_params"] = {
+ "filter_dict": {
+ "id": vim_info.get("vim_network_id"),
+ },
+ }
+ elif target_vld.get("mgmt-network"):
+ extra_dict["find_params"] = {
+ "mgmt": True,
+ "name": target_vld["id"],
+ }
+ else:
+ # create
+ extra_dict["params"] = {
+ "net_name": (
+ f"{indata.get('name')[:16]}-{target_vld.get('name', target_vld.get('id'))[:16]}"
+ ),
+ "ip_profile": Ns._ip_profile_to_ro(vim_info.get("ip_profile")),
+ "provider_network_profile": vim_info.get("provider_network"),
+ }
+
+ if not target_vld.get("underlay"):
+ extra_dict["params"]["net_type"] = "bridge"
+ else:
+ extra_dict["params"]["net_type"] = (
+ "ptp" if target_vld.get("type") == "ELINE" else "data"
+ )
+
+ 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.
+
+ 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 = []
+
+ 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"]
+ )
+ )
+
+ continue # interface not connected to any vld
+
+ extra_dict["depends_on"].append(net_text)
+
+ 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"
+ )
+
+ 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"
+
+ # 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")
+
+ 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
+
+ # cloud config
+ cloud_config = {}
+
+ if target_vdu.get("cloud-init"):
+ if target_vdu["cloud-init"] not in vdu2cloud_init:
+ vdu2cloud_init[target_vdu["cloud-init"]] = Ns._get_cloud_init(
+ db=db,
+ fs=fs,
+ location=target_vdu["cloud-init"],
+ )
+
+ cloud_content_ = vdu2cloud_init[target_vdu["cloud-init"]]
+ cloud_config["user-data"] = Ns._parse_jinja2(
+ cloud_init_content=cloud_content_,
+ params=target_vdu.get("additionalParams"),
+ context=target_vdu["cloud-init"],
+ )
+
+ if target_vdu.get("boot-data-drive"):
+ cloud_config["boot-data-drive"] = target_vdu.get("boot-data-drive")
+
+ 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
+
+ persistent_root_disk = {}
+ disk_list = []
+ vnfd_id = vnfr["vnfd-id"]
+ vnfd = db.get_one("vnfds", {"_id": vnfd_id})
+ 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"