+ flavor_data = {}
+ guest_epa = {}
+ # Find this vdu compute and storage descriptors
+ vdu_virtual_compute = {}
+ vdu_virtual_storage = {}
+ for vcd in vnfd.get("virtual-compute-desc", ()):
+ if vcd.get("id") == vdu.get("virtual-compute-desc"):
+ vdu_virtual_compute = vcd
+ for vsd in vnfd.get("virtual-storage-desc", ()):
+ if vsd.get("id") == vdu.get("virtual-storage-desc", [[]])[0]:
+ vdu_virtual_storage = vsd
+ # Get this vdu vcpus, memory and storage info for flavor_data
+ if vdu_virtual_compute.get("virtual-cpu", {}).get(
+ "num-virtual-cpu"
+ ):
+ flavor_data["vcpu-count"] = vdu_virtual_compute["virtual-cpu"][
+ "num-virtual-cpu"
+ ]
+ if vdu_virtual_compute.get("virtual-memory", {}).get("size"):
+ flavor_data["memory-mb"] = (
+ float(vdu_virtual_compute["virtual-memory"]["size"])
+ * 1024.0
+ )
+ if vdu_virtual_storage.get("size-of-storage"):
+ flavor_data["storage-gb"] = vdu_virtual_storage[
+ "size-of-storage"
+ ]
+ # Get this vdu EPA info for guest_epa
+ if vdu_virtual_compute.get("virtual-cpu", {}).get("cpu-quota"):
+ guest_epa["cpu-quota"] = vdu_virtual_compute["virtual-cpu"][
+ "cpu-quota"
+ ]
+ if vdu_virtual_compute.get("virtual-cpu", {}).get("pinning"):
+ vcpu_pinning = vdu_virtual_compute["virtual-cpu"]["pinning"]
+ if vcpu_pinning.get("thread-policy"):
+ guest_epa["cpu-thread-pinning-policy"] = vcpu_pinning[
+ "thread-policy"
+ ]
+ if vcpu_pinning.get("policy"):
+ cpu_policy = (
+ "SHARED"
+ if vcpu_pinning["policy"] == "dynamic"
+ else "DEDICATED"
+ )
+ guest_epa["cpu-pinning-policy"] = cpu_policy
+ if vdu_virtual_compute.get("virtual-memory", {}).get("mem-quota"):
+ guest_epa["mem-quota"] = vdu_virtual_compute["virtual-memory"][
+ "mem-quota"
+ ]
+ if vdu_virtual_compute.get("virtual-memory", {}).get(
+ "mempage-size"
+ ):
+ guest_epa["mempage-size"] = vdu_virtual_compute[
+ "virtual-memory"
+ ]["mempage-size"]
+ if vdu_virtual_compute.get("virtual-memory", {}).get(
+ "numa-node-policy"
+ ):
+ guest_epa["numa-node-policy"] = vdu_virtual_compute[
+ "virtual-memory"
+ ]["numa-node-policy"]
+ if vdu_virtual_storage.get("disk-io-quota"):
+ guest_epa["disk-io-quota"] = vdu_virtual_storage[
+ "disk-io-quota"
+ ]
+
+ if guest_epa:
+ flavor_data["guest-epa"] = guest_epa
+
+ flavor_data["name"] = vdu["id"][:56] + "-flv"
+ flavor_data["id"] = str(len(nsr_descriptor["flavor"]))
+ nsr_descriptor["flavor"].append(flavor_data)
+
+ sw_image_id = vdu.get("sw-image-desc")
+ if sw_image_id:
+ image_data = self._get_image_data_from_vnfd(vnfd, sw_image_id)
+ self._add_image_to_nsr(nsr_descriptor, image_data)
+
+ # also add alternative images to the list of images
+ for alt_image in vdu.get("alternative-sw-image-desc", ()):
+ image_data = self._get_image_data_from_vnfd(vnfd, alt_image)
+ self._add_image_to_nsr(nsr_descriptor, image_data)
+
+ for vld in nsr_vld:
+ vld["vnfd-connection-point-ref"] = all_vld_connection_point_data.get(
+ vld.get("id"), []
+ )
+ vld["name"] = vld["id"]
+ nsr_descriptor["vld"] = nsr_vld
+
+ return nsr_descriptor
+
+ def _get_image_data_from_vnfd(self, vnfd, sw_image_id):
+ sw_image_desc = utils.find_in_list(
+ vnfd.get("sw-image-desc", ()), lambda sw: sw["id"] == sw_image_id
+ )
+ image_data = {}
+ if sw_image_desc.get("image"):
+ image_data["image"] = sw_image_desc["image"]
+ if sw_image_desc.get("checksum"):
+ image_data["image_checksum"] = sw_image_desc["checksum"]["hash"]
+ if sw_image_desc.get("vim-type"):
+ image_data["vim-type"] = sw_image_desc["vim-type"]
+ return image_data
+
+ def _add_image_to_nsr(self, nsr_descriptor, image_data):
+ """
+ Adds image to nsr checking first it is not already added
+ """
+ img = next(
+ (
+ f
+ for f in nsr_descriptor["image"]
+ if all(f.get(k) == image_data[k] for k in image_data)
+ ),
+ None,
+ )
+ if not img:
+ image_data["id"] = str(len(nsr_descriptor["image"]))
+ nsr_descriptor["image"].append(image_data)
+
+ def _create_vnfr_descriptor_from_vnfd(
+ self,
+ nsd,
+ vnfd,
+ vnfd_id,
+ vnf_index,
+ nsr_descriptor,
+ ns_request,
+ ns_k8s_namespace,
+ ):
+ vnfr_id = str(uuid4())
+ nsr_id = nsr_descriptor["id"]
+ now = time()
+ additional_params, vnf_params = self._format_additional_params(
+ ns_request, vnf_index, descriptor=vnfd
+ )
+
+ vnfr_descriptor = {
+ "id": vnfr_id,
+ "_id": vnfr_id,
+ "nsr-id-ref": nsr_id,
+ "member-vnf-index-ref": vnf_index,
+ "additionalParamsForVnf": additional_params,
+ "created-time": now,
+ # "vnfd": vnfd, # at OSM model.but removed to avoid data duplication TODO: revise
+ "vnfd-ref": vnfd_id,
+ "vnfd-id": vnfd["_id"], # not at OSM model, but useful
+ "vim-account-id": None,
+ "vca-id": None,
+ "vdur": [],
+ "connection-point": [],
+ "ip-address": None, # mgmt-interface filled by LCM
+ }
+ vnf_k8s_namespace = ns_k8s_namespace
+ if vnf_params:
+ if vnf_params.get("k8s-namespace"):
+ vnf_k8s_namespace = vnf_params["k8s-namespace"]
+ if vnf_params.get("config-units"):
+ vnfr_descriptor["config-units"] = vnf_params["config-units"]
+
+ # Create vld
+ if vnfd.get("int-virtual-link-desc"):
+ vnfr_descriptor["vld"] = []
+ for vnfd_vld in vnfd.get("int-virtual-link-desc"):
+ vnfr_descriptor["vld"].append({key: vnfd_vld[key] for key in vnfd_vld})
+
+ for cp in vnfd.get("ext-cpd", ()):
+ vnf_cp = {
+ "name": cp.get("id"),
+ "connection-point-id": cp.get("int-cpd", {}).get("cpd"),
+ "connection-point-vdu-id": cp.get("int-cpd", {}).get("vdu-id"),
+ "id": cp.get("id"),
+ # "ip-address", "mac-address" # filled by LCM
+ # vim-id # TODO it would be nice having a vim port id
+ }
+ vnfr_descriptor["connection-point"].append(vnf_cp)
+
+ # Create k8s-cluster information
+ # TODO: Validate if a k8s-cluster net can have more than one ext-cpd ?
+ if vnfd.get("k8s-cluster"):
+ vnfr_descriptor["k8s-cluster"] = vnfd["k8s-cluster"]
+ all_k8s_cluster_nets_cpds = {}
+ for cpd in get_iterable(vnfd.get("ext-cpd")):
+ if cpd.get("k8s-cluster-net"):
+ all_k8s_cluster_nets_cpds[cpd.get("k8s-cluster-net")] = cpd.get(
+ "id"
+ )
+ for net in get_iterable(vnfr_descriptor["k8s-cluster"].get("nets")):
+ if net.get("id") in all_k8s_cluster_nets_cpds:
+ net["external-connection-point-ref"] = all_k8s_cluster_nets_cpds[
+ net.get("id")
+ ]
+
+ # update kdus
+ for kdu in get_iterable(vnfd.get("kdu")):
+ additional_params, kdu_params = self._format_additional_params(
+ ns_request, vnf_index, kdu_name=kdu["name"], descriptor=vnfd
+ )
+ kdu_k8s_namespace = vnf_k8s_namespace
+ kdu_model = kdu_params.get("kdu_model") if kdu_params else None
+ if kdu_params and kdu_params.get("k8s-namespace"):
+ kdu_k8s_namespace = kdu_params["k8s-namespace"]
+ kdu_deployment_name = ""
+ if kdu_params and kdu_params.get("kdu-deployment-name"):
+ kdu_deployment_name = kdu_params.get("kdu-deployment-name")
+
+ kdur = {
+ "additionalParams": additional_params,
+ "k8s-namespace": kdu_k8s_namespace,
+ "kdu-deployment-name": kdu_deployment_name,
+ "kdu-name": kdu["name"],
+ # TODO "name": "" Name of the VDU in the VIM
+ "ip-address": None, # mgmt-interface filled by LCM
+ "k8s-cluster": {},
+ }
+ if kdu_params and kdu_params.get("config-units"):
+ kdur["config-units"] = kdu_params["config-units"]
+ if kdu.get("helm-version"):
+ kdur["helm-version"] = kdu["helm-version"]
+ for k8s_type in ("helm-chart", "juju-bundle"):
+ if kdu.get(k8s_type):
+ kdur[k8s_type] = kdu_model or kdu[k8s_type]
+ if not vnfr_descriptor.get("kdur"):
+ vnfr_descriptor["kdur"] = []
+ vnfr_descriptor["kdur"].append(kdur)
+
+ vnfd_mgmt_cp = vnfd.get("mgmt-cp")
+
+ for vdu in vnfd.get("vdu", ()):
+ vdu_mgmt_cp = []
+ try:
+ configs = vnfd.get("df")[0]["lcm-operations-configuration"][
+ "operate-vnf-op-config"
+ ]["day1-2"]
+ vdu_config = utils.find_in_list(
+ configs, lambda config: config["id"] == vdu["id"]
+ )
+ except Exception:
+ vdu_config = None
+
+ try:
+ vdu_instantiation_level = utils.find_in_list(
+ vnfd.get("df")[0]["instantiation-level"][0]["vdu-level"],
+ lambda a_vdu_profile: a_vdu_profile["vdu-id"] == vdu["id"],
+ )
+ except Exception:
+ vdu_instantiation_level = None
+
+ if vdu_config:
+ external_connection_ee = utils.filter_in_list(
+ vdu_config.get("execution-environment-list", []),
+ lambda ee: "external-connection-point-ref" in ee,
+ )
+ for ee in external_connection_ee:
+ vdu_mgmt_cp.append(ee["external-connection-point-ref"])
+
+ additional_params, vdu_params = self._format_additional_params(
+ ns_request, vnf_index, vdu_id=vdu["id"], descriptor=vnfd
+ )
+
+ try:
+ vdu_virtual_storage_descriptors = utils.filter_in_list(
+ vnfd.get("virtual-storage-desc", []),
+ lambda stg_desc: stg_desc["id"] in vdu["virtual-storage-desc"]
+ )
+ except Exception:
+ vdu_virtual_storage_descriptors = []
+ vdur = {
+ "vdu-id-ref": vdu["id"],
+ # TODO "name": "" Name of the VDU in the VIM
+ "ip-address": None, # mgmt-interface filled by LCM
+ # "vim-id", "flavor-id", "image-id", "management-ip" # filled by LCM
+ "internal-connection-point": [],
+ "interfaces": [],
+ "additionalParams": additional_params,
+ "vdu-name": vdu["name"],
+ "virtual-storages": vdu_virtual_storage_descriptors
+ }
+ if vdu_params and vdu_params.get("config-units"):
+ vdur["config-units"] = vdu_params["config-units"]
+ if deep_get(vdu, ("supplemental-boot-data", "boot-data-drive")):
+ vdur["boot-data-drive"] = vdu["supplemental-boot-data"][
+ "boot-data-drive"
+ ]
+ if vdu.get("pdu-type"):
+ vdur["pdu-type"] = vdu["pdu-type"]
+ vdur["name"] = vdu["pdu-type"]
+ # TODO volumes: name, volume-id
+ for icp in vdu.get("int-cpd", ()):
+ vdu_icp = {
+ "id": icp["id"],
+ "connection-point-id": icp["id"],
+ "name": icp.get("id"),
+ }
+
+ vdur["internal-connection-point"].append(vdu_icp)
+
+ for iface in icp.get("virtual-network-interface-requirement", ()):
+ iface_fields = ("name", "mac-address")
+ vdu_iface = {
+ x: iface[x] for x in iface_fields if iface.get(x) is not None