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
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],
"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"],
"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"),
"size": root_disk.get("size-of-storage"),
+ "keep": Ns.is_volume_keeping_required(root_disk),
}
disk_list.append(persistent_root_disk[vsd["id"]])
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"],
}
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"]])
+ @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.
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 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"]])
):
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"]])
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 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
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")
task_index += 1
break
else:
-
for vdu_index, vdu in enumerate(db_vnfr["vdur"]):
extra_dict["params"] = {
"vim_vm_id": vdu["vim-id"],