return extra_dict
- @staticmethod
- def _ip_profile_to_ro(
- ip_profile: Dict[str, Any],
- ) -> Dict[str, Any]:
- """[summary]
-
- 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],
"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")),
+ "ip_profile": vim_info.get("ip_profile"),
"provider_network_profile": vim_info.get("provider_network"),
}
== "persistent-storage:persistent-storage"
):
for vdu_volume in vdu_instantiation_volumes_list:
-
if (
vdu_volume["vim-volume-id"]
and root_disk["id"] == vdu_volume["name"]
):
-
persistent_root_disk[vsd["id"]] = {
"vim_volume_id": vdu_volume["vim-volume-id"],
"image_id": vdu.get("sw-image-desc"),
return persistent_root_disk
else:
-
if root_disk.get("size-of-storage"):
persistent_root_disk[vsd["id"]] = {
"image_id": vdu.get("sw-image-desc"),
and disk["id"] not in persistent_root_disk.keys()
):
for vdu_volume in vdu_instantiation_volumes_list:
-
if vdu_volume["vim-volume-id"] and disk["id"] == vdu_volume["name"]:
-
persistent_disk[disk["id"]] = {
"vim_volume_id": vdu_volume["vim-volume-id"],
}
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
)
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]}
+ 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})
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
if target_vdu.get("additionalParams"):
vdu_instantiation_volumes_list = (
- target_vdu.get("additionalParams").get("OSM").get("vdu_volumes")
+ target_vdu.get("additionalParams").get("OSM", {}).get("vdu_volumes")
+ )
+ vdu_instantiation_flavor_id = (
+ target_vdu.get("additionalParams").get("OSM", {}).get("vim_flavor_id")
)
- if vdu_instantiation_volumes_list:
+ # 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
"description": target_vdu["vdu-name"],
"start": True,
"image_id": "TASK-" + image_text,
- "flavor_id": "TASK-" + flavor_text,
+ "flavor_id": flavor_id,
"affinity_group_list": affinity_group_list,
"net_list": net_list,
"cloud_config": cloud_config or None,
vim_details = yaml.safe_load(f"{vim_details_text}")
for iface_index, interface in enumerate(existing_vdu["interfaces"]):
-
if "port-security-enabled" in interface:
interface["port_security"] = interface.pop("port-security-enabled")
# Check each VNF of the target
for target_vnf in target_list:
- # Find this VNF in the list from DB
- vnfr_id = target_vnf.get("vnfInstanceId", None)
- if vnfr_id:
- existing_vnf = db_vnfrs.get(vnfr_id)
- db_record = "vnfrs:{}:{}".format(vnfr_id, db_path)
- # vim_account_id = existing_vnf.get("vim-account-id", "")
+ # Find this VNF in the list from DB, raise exception if vnfInstanceId is not found
+ vnfr_id = target_vnf["vnfInstanceId"]
+ existing_vnf = db_vnfrs.get(vnfr_id)
+ db_record = "vnfrs:{}:{}".format(vnfr_id, db_path)
+ # vim_account_id = existing_vnf.get("vim-account-id", "")
+ target_vdus = target_vnf.get("additionalParams", {}).get("vdu", [])
# Check each VDU of this VNF
- for target_vdu in target_vnf["additionalParams"].get("vdu", None):
+ if not target_vdus:
+ # Create target_vdu_list from DB, if VDUs are not specified
+ target_vdus = []
+ for existing_vdu in existing_vnf.get("vdur"):
+ vdu_name = existing_vdu.get("vdu-name", None)
+ vdu_index = existing_vdu.get("count-index", 0)
+ vdu_to_be_healed = {"vdu-id": vdu_name, "count-index": vdu_index}
+ target_vdus.append(vdu_to_be_healed)
+ for target_vdu in target_vdus:
vdu_name = target_vdu.get("vdu-id", None)
# For multi instance VDU count-index is mandatory
# For single session VDU count-indes is 0
task_index += 1
break
else:
-
for vdu_index, vdu in enumerate(db_vnfr["vdur"]):
extra_dict["params"] = {
"vim_vm_id": vdu["vim-id"],