step = "filling nsr from input data"
nsr_id = str(uuid4())
- nsr_descriptor = self._create_nsr_descriptor_from_nsd(nsd, ns_request, nsr_id)
+ nsr_descriptor = self._create_nsr_descriptor_from_nsd(nsd, ns_request, nsr_id, session)
# Create VNFRs
needed_vnfds = {}
return ns_k8s_namespace
- def _create_nsr_descriptor_from_nsd(self, nsd, ns_request, nsr_id):
+ def _create_nsr_descriptor_from_nsd(self, nsd, ns_request, nsr_id, session):
now = time()
additional_params, _ = self._format_additional_params(ns_request, descriptor=nsd)
"vnfd-id-ref": vnf_profile.get("vnfd-id")
})
- vnfd = self.db.get_one("vnfds",
- {"id": vnf_profile.get("vnfd-id")},
- fail_on_empty=True,
- fail_on_more=True)
+ vnfd = self._get_vnfd_from_db(vnf_profile.get("vnfd-id"), session)
for vdu in vnfd.get("vdu", ()):
flavor_data = {}
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
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)
vdur = {
"name": icp.get("id"),
}
- if "port-security-enabled" in icp:
- vdu_icp["port-security-enabled"] = icp["port-security-enabled"]
-
- if "port-security-disable-strategy" in icp:
- vdu_icp["port-security-disable-strategy"] = icp["port-security-disable-strategy"]
-
vdur["internal-connection-point"].append(vdu_icp)
for iface in icp.get("virtual-network-interface-requirement", ()):
vdu_iface = {x: iface[x] for x in iface_fields if iface.get(x) is not None}
vdu_iface["internal-connection-point-ref"] = vdu_icp["id"]
+ if "port-security-enabled" in icp:
+ vdu_iface["port-security-enabled"] = icp["port-security-enabled"]
+
+ if "port-security-disable-strategy" in icp:
+ vdu_iface["port-security-disable-strategy"] = icp["port-security-disable-strategy"]
+
for ext_cp in vnfd.get("ext-cpd", ()):
if not ext_cp.get("int-cpd"):
continue
continue
if icp["id"] == ext_cp["int-cpd"].get("cpd"):
vdu_iface["external-connection-point-ref"] = ext_cp.get("id")
+
+ if "port-security-enabled" in ext_cp:
+ vdu_iface["port-security-enabled"] = (
+ ext_cp["port-security-enabled"]
+ )
+
+ if "port-security-disable-strategy" in ext_cp:
+ vdu_iface["port-security-disable-strategy"] = (
+ ext_cp["port-security-disable-strategy"]
+ )
+
break
if vnfd_mgmt_cp and vdu_iface.get("external-connection-point-ref") == vnfd_mgmt_cp:
vdu_iface["mgmt-vnf"] = True
- vdu_iface["mgmt-interface"] = True # TODO change to mgmt-vdu
+ vdu_iface["mgmt-interface"] = True
+
+ for ecp in vdu_mgmt_cp:
+ if vdu_iface.get("external-connection-point-ref") == ecp:
+ vdu_iface["mgmt-interface"] = True
if iface.get("virtual-interface"):
vdu_iface.update(deepcopy(iface["virtual-interface"]))
if nsr_flavor_desc:
vdur["ns-flavor-id"] = nsr_flavor_desc["id"]
- count = int(vdu.get("count", 1))
+ if vdu_instantiation_level:
+ count = vdu_instantiation_level.get("number-of-instances")
+ else:
+ count = 1
+
for index in range(0, count):
vdur = deepcopy(vdur)
for iface in vdur["interfaces"]:
- if iface.get("ip-address"):
+ if iface.get("ip-address") and index != 0:
iface["ip-address"] = increment_ip_mac(iface["ip-address"])
- if iface.get("mac-address"):
+ if iface.get("mac-address") and index != 0:
iface["mac-address"] = increment_ip_mac(iface["mac-address"])
vdur["_id"] = str(uuid4())
descriptor_configuration = utils.find_in_list(
configs,
lambda config: config["id"] == indata["vdu_id"]
- ).get("config-primitive")
+ )
elif indata.get("kdu_name"):
self._check_valid_kdu(vnfd, indata["kdu_name"])
descriptor_configuration = utils.find_in_list(
configs,
lambda config: config["id"] == indata.get("kdu_name")
- ).get("config-primitive")
+ )
else:
descriptor_configuration = utils.find_in_list(
configs,
lambda config: config["id"] == vnfd["id"]
- ).get("config-primitive")
+ )
+ if descriptor_configuration is not None:
+ descriptor_configuration = descriptor_configuration.get("config-primitive")
else: # use a NSD
descriptor_configuration = nsd.get("ns-configuration", {}).get("config-primitive")