return ns_k8s_namespace
- def _add_flavor_to_nsr(self, vdu, vnfd, nsr_descriptor, member_vnf_index, revision=None):
+ def _add_flavor_to_nsr(
+ self, vdu, vnfd, nsr_descriptor, member_vnf_index, revision=None
+ ):
flavor_data = {}
guest_epa = {}
# Find this vdu compute and storage descriptors
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"
- ):
+ 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
+ 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"
- ]
+ 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"
- ]
+ 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"
- ]
+ 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"
+ "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"
+ 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(
- "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 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
revision = revision if revision is not None else 1
- flavor_data["name"] = vdu["id"][:56] + "-" + member_vnf_index + "-" + str(revision) + "-flv"
+ flavor_data["name"] = (
+ vdu["id"][:56] + "-" + member_vnf_index + "-" + str(revision) + "-flv"
+ )
flavor_data["id"] = str(len(nsr_descriptor["flavor"]))
nsr_descriptor["flavor"].append(flavor_data)
"configurationStatus": None,
"vcaStatus": None,
"nsd": {k: v for k, v in nsd.items()},
- "datacenter": ns_request["vimAccountId"],
+ "vimdatacenter": ns_request.get("vimAccountId"),
+ "paasdatacenter": ns_request.get("paasAccountId"),
"resource-orchestrator": "osmopenmano",
"description": ns_request.get("nsDescription", ""),
"constituent-vnfr-ref": [],
"vnfd-ref": vnfd_id,
"vnfd-id": vnfd["_id"], # not at OSM model, but useful
"vim-account-id": None,
+ "paas-account-id": None,
"vca-id": None,
"vdur": [],
"connection-point": [],
if "revision" in vnfd:
vnfr_descriptor["revision"] = vnfd["revision"]
-
vnf_k8s_namespace = ns_k8s_namespace
if vnf_params:
if vnf_params.get("k8s-namespace"):
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"]
+ lambda stg_desc: stg_desc["id"] in vdu["virtual-storage-desc"],
)
except Exception:
vdu_virtual_storage_descriptors = []
"interfaces": [],
"additionalParams": additional_params,
"vdu-name": vdu["name"],
- "virtual-storages": vdu_virtual_storage_descriptors
+ "virtual-storages": vdu_virtual_storage_descriptors,
}
if vdu_params and vdu_params.get("config-units"):
vdur["config-units"] = vdu_params["config-units"]
vdur["alt-image-ids"] = alt_image_ids
revision = revision if revision is not None else 1
- flavor_data_name = vdu["id"][:56] + "-" + vnf_index + "-" + str(revision) + "-flv"
+ flavor_data_name = (
+ vdu["id"][:56] + "-" + vnf_index + "-" + str(revision) + "-flv"
+ )
nsr_flavor_desc = utils.find_in_list(
nsr_descriptor["flavor"],
lambda flavor: flavor["name"] == flavor_data_name,
:param filter_q: dict: query parameter containing vcaStatus-refresh as true or false
:return: None
"""
- time_now, time_delta = time(), time() - ns_instance_content["_admin"]["modified"]
- force_refresh = isinstance(filter_q, dict) and filter_q.get('vcaStatusRefresh') == 'true'
+ time_now, time_delta = (
+ time(),
+ time() - ns_instance_content["_admin"]["modified"],
+ )
+ force_refresh = (
+ isinstance(filter_q, dict) and filter_q.get("vcaStatusRefresh") == "true"
+ )
threshold_reached = time_delta > 120
if force_refresh or threshold_reached:
operation, _id = "vca_status_refresh", ns_instance_content["_id"]
ns_instance_content["_admin"]["modified"] = time_now
self.db.set_one(self.topic, {"_id": _id}, ns_instance_content)
nslcmop_desc = NsLcmOpTopic._create_nslcmop(_id, operation, None)
- self.format_on_new(nslcmop_desc, session["project_id"], make_public=session["public"])
+ self.format_on_new(
+ nslcmop_desc, session["project_id"], make_public=session["public"]
+ )
nslcmop_desc["_admin"].pop("nsState")
self.msg.write("ns", operation, nslcmop_desc)
return
def _check_heal_ns_operation(self, indata, nsr):
return
- def _check_instantiate_ns_operation(self, indata, nsr, session):
- vnf_member_index_to_vnfd = {} # map between vnf_member_index to vnf descriptor.
- vim_accounts = []
- wim_accounts = []
- nsd = nsr["nsd"]
- self._check_valid_vim_account(indata["vimAccountId"], vim_accounts, session)
- self._check_valid_wim_account(indata.get("wimAccountId"), wim_accounts, session)
+ def _validate_vnf_in_indata(
+ self,
+ indata: dict,
+ vim_accounts: dict = None,
+ paas_accounts: dict = None,
+ nsr: dict = None,
+ session: dict = None,
+ ):
+ """Validate the VNF parameters in indata.
+
+ Args:
+ indata (dict): Input data dictionary
+ vim_accounts (dict): VIM accounts dictionary
+ paas_accounts (dict) PaaS accounts dictionary
+ nsr (dict): Network service record dictionary
+ session (dict): Dictionary contains "username", "admin", "force", "public",
+ "project_id", "set_project"
+ """
+ # Map between vnf_member_index to vnf descriptor.
+ vnf_member_index_to_vnfd = {}
+
for in_vnf in get_iterable(indata.get("vnf")):
member_vnf_index = in_vnf["member-vnf-index"]
+
if vnf_member_index_to_vnfd.get(member_vnf_index):
vnfd = vnf_member_index_to_vnfd[member_vnf_index]
+
else:
vnfd = self._get_vnfd_from_vnf_member_index(
member_vnf_index, nsr["_id"]
vnf_member_index_to_vnfd[
member_vnf_index
] = vnfd # add to cache, avoiding a later look for
+
self._check_vnf_instantiation_params(in_vnf, vnfd)
+
if in_vnf.get("vimAccountId"):
self._check_valid_vim_account(
in_vnf["vimAccountId"], vim_accounts, session
)
+ elif in_vnf.get("paasAccountId"):
+ self._check_valid_paas_account(
+ in_vnf["paasAccountId"], paas_accounts, session
+ )
+
+ def _validate_vld_in_indata(
+ self,
+ indata: dict,
+ session: dict = None,
+ nsd: dict = None,
+ wim_accounts: dict = None,
+ ):
+ """Validate the Virtual Link Descriptor parameters in indata.
+
+ Args:
+ indata (dict): Input data dictionary
+ session (dict): Dictionary contains "username", "admin", "force", "public",
+ "project_id", "set_project"
+ nsd (dict): Network service descriptor dictionary
+ wim_accounts (dict): WIM accounts dictionary
+
+ Raises:
+ EngineException
+ """
for in_vld in get_iterable(indata.get("vld")):
self._check_valid_wim_account(
in_vld.get("wimAccountId"), wim_accounts, session
)
)
+ def _check_instantiate_ns_operation(
+ self, indata: dict, nsr: dict, session: dict
+ ) -> None:
+ """Validate the parameters for NS instantiate operation
+
+ Args:
+ indata (dict): Input data dictionary
+ nsr (dict): Network service record dictionary
+ session (dict): Dictionary contains "username", "admin", "force", "public", "project_id", "set_project"
+ """
+ vim_accounts, wim_accounts, paas_accounts = [], [], []
+ nsd = nsr["nsd"]
+ self._check_valid_vim_account(indata.get("vimAccountId"), vim_accounts, session)
+ self._check_valid_wim_account(indata.get("wimAccountId"), wim_accounts, session)
+ self._check_valid_paas_account(indata.get("paasAccountId"), paas_accounts, session)
+
+ self._validate_vnf_in_indata(
+ indata,
+ vim_accounts=vim_accounts,
+ paas_accounts=paas_accounts,
+ nsr=nsr,
+ session=session,
+ )
+ self._validate_vld_in_indata(
+ indata, session=session, nsd=nsd, wim_accounts=wim_accounts
+ )
+
def _get_vnfd_from_vnf_member_index(self, member_vnf_index, nsr_id):
# Obtain vnf descriptor. The vnfr is used to get the vnfd._id used for this member_vnf_index
vnfr = self.db.get_one(
"nsd:constituent-vnfd".format(member_vnf_index)
)
- ## Backwards compatibility: if there is no revision, get it from the one and only VNFD entry
+ # Backwards compatibility: if there is no revision, get it from the one and only VNFD entry
if "revision" in vnfr:
vnfd_revision = vnfr["vnfd-id"] + ":" + str(vnfr["revision"])
- vnfd = self.db.get_one("vnfds_revisions", {"_id": vnfd_revision}, fail_on_empty=False)
+ vnfd = self.db.get_one(
+ "vnfds_revisions", {"_id": vnfd_revision}, fail_on_empty=False
+ )
else:
- vnfd = self.db.get_one("vnfds", {"_id": vnfr["vnfd-id"]}, fail_on_empty=False)
+ vnfd = self.db.get_one(
+ "vnfds", {"_id": vnfr["vnfd-id"]}, fail_on_empty=False
+ )
if not vnfd:
raise EngineException(
)
)
- def _check_valid_vim_account(self, vim_account, vim_accounts, session):
- if vim_account in vim_accounts:
- return
+ def _check_valid_vim_account(
+ self, vim_account: str, vim_accounts: list, session: dict
+ ) -> list:
+ """Validate the given VIM account whether present in the project or not,
+ add the given vim_account to vim_accounts list.
+
+ Args:
+ vim_account (str): VIM Account to be used
+ vim_accounts (list): List of VIM accounts
+ session (dict): Contains "username", "admin", "force", "public", "project_id", "set_project"
+
+ Raises:
+ EngineException
+ """
try:
- db_filter = self._get_project_filter(session)
- db_filter["_id"] = vim_account
- self.db.get_one("vim_accounts", db_filter)
+ if vim_account and vim_account not in vim_accounts:
+ db_filter = self._get_project_filter(session)
+ db_filter["_id"] = vim_account
+ self.db.get_one("vim_accounts", db_filter)
+ vim_accounts.append(vim_account)
except Exception:
raise EngineException(
"Invalid vimAccountId='{}' not present for the project".format(
vim_account
)
)
- vim_accounts.append(vim_account)
+
+ def _check_valid_paas_account(
+ self, paas_account: str, paas_accounts: list, session: dict
+ ) -> list:
+ """Validate the given PaaS account whether present in the project or not,
+ add the given paas_account to paas_accounts list and return.
+
+ Args:
+ paas_account (str): PaaS Account to be used
+ paas_accounts (list): List of PaaS accounts
+ session (dict): Contains "username", "admin", "force", "public", "project_id", "set_project"
+
+ Raises:
+ EngineException
+ """
+ try:
+ if paas_account and paas_account not in paas_accounts:
+ db_filter = self._get_project_filter(session)
+ db_filter["_id"] = paas_account
+ self.db.get_one("paas", db_filter)
+ paas_accounts.append(paas_account)
+ except Exception:
+ raise EngineException(
+ "Invalid paasAccountId='{}' not present for the project".format(
+ paas_account
+ )
+ )
def _get_vim_account(self, vim_id: str, session):
try:
return self.db.get_one("vim_accounts", db_filter)
except Exception:
raise EngineException(
- "Invalid vimAccountId='{}' not present for the project".format(
- vim_id
- )
+ "Invalid vimAccountId='{}' not present for the project".format(vim_id)
)
- def _check_valid_wim_account(self, wim_account, wim_accounts, session):
- if not isinstance(wim_account, str):
- return
- if wim_account in wim_accounts:
- return
+ def _check_valid_wim_account(
+ self, wim_account: str, wim_accounts: list, session: dict
+ ) -> list:
+ """Validate the given WIM account whether present in the project or not,
+ add the given wim_account to wim_accounts list and return.
+
+ Args:
+ wim_account (str): WIM Account to be used
+ wim_accounts (list): List of WIM accounts
+ session (dict): Contains "username", "admin", "force", "public", "project_id", "set_project"
+
+ Raises:
+ EngineException
+ """
try:
- db_filter = self._get_project_filter(session)
- db_filter["_id"] = wim_account
- self.db.get_one("wim_accounts", db_filter)
+ if isinstance(wim_account, str) and wim_account not in wim_accounts:
+ db_filter = self._get_project_filter(session)
+ db_filter["_id"] = wim_account
+ self.db.get_one("wim_accounts", db_filter)
+ wim_accounts.append(wim_account)
except Exception:
raise EngineException(
"Invalid wimAccountId='{}' not present for the project".format(
wim_account
)
)
- wim_accounts.append(wim_account)
def _look_for_pdu(
self, session, rollback, vnfr, vim_account, vnfr_update, vnfr_update_rollback
"ns-vld-id": NSD vld where this interface is connected.
NOTE: One, and only one between 'vnf-vld-id' and 'ns-vld-id' contains a value. The other will be None
"""
-
ifaces_forcing_vim_network = []
+ if not vim_account:
+ return ifaces_forcing_vim_network
for vdur_index, vdur in enumerate(get_iterable(vnfr.get("vdur"))):
if not vdur.get("pdu-type"):
continue
"ns-vld-id": NSD vld where this interface is connected.
NOTE: One, and only one between 'vnf-vld-id' and 'ns-vld-id' contains a value. The other will be None
"""
-
ifaces_forcing_vim_network = []
+
+ if not vim_account:
+ return ifaces_forcing_vim_network
+
if not vnfr.get("kdur"):
return ifaces_forcing_vim_network
for cpd in vlc.get("constituent-cpd-id", ()):
if cpd.get("ip-address"):
step = "Storing ip-address info"
- vld_fixed_ip_connection_point_data.update({vlc.get("virtual-link-profile-id") + '.' + cpd.get("constituent-base-element-id"): {
- "vnfd-connection-point-ref": cpd.get(
- "constituent-cpd-id"),
- "ip-address": cpd.get(
- "ip-address")}})
+ vld_fixed_ip_connection_point_data.update(
+ {
+ vlc.get("virtual-link-profile-id")
+ + "."
+ + cpd.get("constituent-base-element-id"): {
+ "vnfd-connection-point-ref": cpd.get(
+ "constituent-cpd-id"
+ ),
+ "ip-address": cpd.get("ip-address"),
+ }
+ }
+ )
# Inserting ip address to vnfr
if len(vld_fixed_ip_connection_point_data) > 0:
vnfrs = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
for item in vld_fixed_ip_connection_point_data.keys():
step = "Filtering vnfrs"
- vnfr = next(filter(lambda vnfr: vnfr["member-vnf-index-ref"] == item.split('.')[1], vnfrs), None)
+ vnfr = next(
+ filter(
+ lambda vnfr: vnfr["member-vnf-index-ref"]
+ == item.split(".")[1],
+ vnfrs,
+ ),
+ None,
+ )
if vnfr:
vnfr_update = {}
for vdur_index, vdur in enumerate(vnfr["vdur"]):
for iface_index, iface in enumerate(vdur["interfaces"]):
step = "Looking for matched interface"
if (
- iface.get("external-connection-point-ref")
- == vld_fixed_ip_connection_point_data[item].get("vnfd-connection-point-ref") and
- iface.get("ns-vld-id") == item.split('.')[0]
-
+ iface.get("external-connection-point-ref")
+ == vld_fixed_ip_connection_point_data[item].get(
+ "vnfd-connection-point-ref"
+ )
+ and iface.get("ns-vld-id") == item.split(".")[0]
):
vnfr_update_text = "vdur.{}.interfaces.{}".format(
vdur_index, iface_index
step = "Storing info in order to update vnfr"
vnfr_update[
vnfr_update_text + ".ip-address"
- ] = increment_ip_mac(
- vld_fixed_ip_connection_point_data[item].get("ip-address"),
- vdur.get("count-index", 0), )
+ ] = increment_ip_mac(
+ vld_fixed_ip_connection_point_data[item].get(
+ "ip-address"
+ ),
+ vdur.get("count-index", 0),
+ )
vnfr_update[vnfr_update_text + ".fixed-ip"] = True
step = "updating vnfr at database"
self.db.set_one("vnfrs", {"_id": vnfr["_id"]}, vnfr_update)
except (
- ValidationError,
- EngineException,
- DbException,
- MsgException,
- FsException,
+ ValidationError,
+ EngineException,
+ DbException,
+ MsgException,
+ FsException,
) as e:
raise type(e)("{} while '{}'".format(e, step), http_code=e.http_code)
- def _update_vnfrs(self, session, rollback, nsr, indata):
- # get vnfr
- nsr_id = nsr["_id"]
- vnfrs = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
+ @staticmethod
+ def _get_vld_icp_from_instantiation_params(
+ vnf_inst_params: dict, vnfr: dict, vnfr_update: dict
+ ) -> dict:
+ """Get vnf.internal-vld.internal-conection-point instantiation params to update vnfr.vdur.interfaces
+ Args:
+ vnf_inst_params (dict): VNF instantiation parameters dictionary
+ vnfr (dict): VNF record dictionary
+ vnfr_update (dict): VNF record update dictionary
- for vnfr in vnfrs:
- vnfr_update = {}
- vnfr_update_rollback = {}
- member_vnf_index = vnfr["member-vnf-index-ref"]
- # update vim-account-id
+ Returns:
+ vnfr_update (dict): VNF record update dictionary
+ """
+ for ivld_inst_param in get_iterable(vnf_inst_params.get("internal-vld")):
+ for icp_inst_param in get_iterable(
+ ivld_inst_param.get("internal-connection-point")
+ ):
+ # Look for iface
+ for vdur_index, vdur in enumerate(vnfr["vdur"]):
+ for iface_index, iface in enumerate(vdur["interfaces"]):
+ if (
+ iface.get("internal-connection-point-ref")
+ == icp_inst_param["id-ref"]
+ ):
+ vnfr_update_text = "vdur.{}.interfaces.{}".format(
+ vdur_index, iface_index
+ )
+ if icp_inst_param.get("ip-address"):
+ vnfr_update[
+ vnfr_update_text + ".ip-address"
+ ] = increment_ip_mac(
+ icp_inst_param.get("ip-address"),
+ vdur.get("count-index", 0),
+ )
+ vnfr_update[vnfr_update_text + ".fixed-ip"] = True
+ if icp_inst_param.get("mac-address"):
+ vnfr_update[
+ vnfr_update_text + ".mac-address"
+ ] = increment_ip_mac(
+ icp_inst_param.get("mac-address"),
+ vdur.get("count-index", 0),
+ )
+ vnfr_update[vnfr_update_text + ".fixed-mac"] = True
+ break
+ return vnfr_update
- vim_account = indata["vimAccountId"]
- vca_id = self._get_vim_account(vim_account, session).get("vca")
- # check instantiate parameters
- for vnf_inst_params in get_iterable(indata.get("vnf")):
- if vnf_inst_params["member-vnf-index"] != member_vnf_index:
+ @staticmethod
+ def _get_vdu_interfaces_from_instantiation_params(
+ vnfr_update: dict, vnf_inst_params: dict, vnfr: dict
+ ) -> dict:
+ """Get vnf.vdu.interface instantiation params to update vnfr.vdur.interfaces ip, mac
+ Args:
+ vnfr_update (dict): VNF record update dictionary
+ vnf_inst_params (dict): VNF instantiation parameters dictionary
+ vnfr (dict): VNF record dictionary
+
+ Returns:
+ vnfr_update (dict): VNF record update dictionary
+ """
+ for vdu_inst_param in get_iterable(vnf_inst_params.get("vdu")):
+ for vdur_index, vdur in enumerate(vnfr["vdur"]):
+ if vdu_inst_param["id"] != vdur["vdu-id-ref"]:
continue
- if vnf_inst_params.get("vimAccountId"):
- vim_account = vnf_inst_params.get("vimAccountId")
- vca_id = self._get_vim_account(vim_account, session).get("vca")
+ for iface_inst_param in get_iterable(vdu_inst_param.get("interface")):
+ iface_index, _ = next(
+ i
+ for i in enumerate(vdur["interfaces"])
+ if i[1]["name"] == iface_inst_param["name"]
+ )
+ vnfr_update_text = "vdur.{}.interfaces.{}".format(
+ vdur_index, iface_index
+ )
+ if iface_inst_param.get("ip-address"):
+ vnfr_update[
+ vnfr_update_text + ".ip-address"
+ ] = increment_ip_mac(
+ iface_inst_param.get("ip-address"),
+ vdur.get("count-index", 0),
+ )
+ vnfr_update[vnfr_update_text + ".fixed-ip"] = True
+ if iface_inst_param.get("mac-address"):
+ vnfr_update[
+ vnfr_update_text + ".mac-address"
+ ] = increment_ip_mac(
+ iface_inst_param.get("mac-address"),
+ vdur.get("count-index", 0),
+ )
+ vnfr_update[vnfr_update_text + ".fixed-mac"] = True
+ if iface_inst_param.get("floating-ip-required"):
+ vnfr_update[vnfr_update_text + ".floating-ip-required"] = True
+ return vnfr_update
- # get vnf.vdu.interface instantiation params to update vnfr.vdur.interfaces ip, mac
- for vdu_inst_param in get_iterable(vnf_inst_params.get("vdu")):
- for vdur_index, vdur in enumerate(vnfr["vdur"]):
- if vdu_inst_param["id"] != vdur["vdu-id-ref"]:
- continue
- for iface_inst_param in get_iterable(
- vdu_inst_param.get("interface")
+ @staticmethod
+ def _get_ip_address_from_instantiation_params(
+ indata: dict, member_vnf_index: str, vnfr: dict, vnfr_update: dict
+ ) -> dict:
+ """Get ip address from instantiation parameters.vld.vnfd-connection-point-ref
+ Args:
+ indata (dict): Input data dictionary
+ member_vnf_index (str): VNF index as an identifier
+ vnfr (dict): VNF record dictionary
+ vnfr_update (dict): VNF record update dictionary to keep the updates
+
+ Returns:
+ vnfr_update (dict): VNF record update dictionary to keep the updates
+ """
+ for vld_inst_param in get_iterable(indata.get("vld")):
+ for vnfcp_inst_param in get_iterable(
+ vld_inst_param.get("vnfd-connection-point-ref")
+ ):
+ if vnfcp_inst_param["member-vnf-index-ref"] != member_vnf_index:
+ continue
+ # look for iface
+ for vdur_index, vdur in enumerate(vnfr["vdur"]):
+ for iface_index, iface in enumerate(vdur["interfaces"]):
+ if (
+ iface.get("external-connection-point-ref")
+ == vnfcp_inst_param["vnfd-connection-point-ref"]
):
- iface_index, _ = next(
- i
- for i in enumerate(vdur["interfaces"])
- if i[1]["name"] == iface_inst_param["name"]
- )
vnfr_update_text = "vdur.{}.interfaces.{}".format(
vdur_index, iface_index
)
- if iface_inst_param.get("ip-address"):
+ if vnfcp_inst_param.get("ip-address"):
vnfr_update[
vnfr_update_text + ".ip-address"
] = increment_ip_mac(
- iface_inst_param.get("ip-address"),
+ vnfcp_inst_param.get("ip-address"),
vdur.get("count-index", 0),
)
vnfr_update[vnfr_update_text + ".fixed-ip"] = True
- if iface_inst_param.get("mac-address"):
+ if vnfcp_inst_param.get("mac-address"):
vnfr_update[
vnfr_update_text + ".mac-address"
] = increment_ip_mac(
- iface_inst_param.get("mac-address"),
+ vnfcp_inst_param.get("mac-address"),
vdur.get("count-index", 0),
)
vnfr_update[vnfr_update_text + ".fixed-mac"] = True
- if iface_inst_param.get("floating-ip-required"):
- vnfr_update[
- vnfr_update_text + ".floating-ip-required"
- ] = True
- # get vnf.internal-vld.internal-conection-point instantiation params to update vnfr.vdur.interfaces
- # TODO update vld with the ip-profile
- for ivld_inst_param in get_iterable(
- vnf_inst_params.get("internal-vld")
- ):
- for icp_inst_param in get_iterable(
- ivld_inst_param.get("internal-connection-point")
- ):
- # look for iface
- for vdur_index, vdur in enumerate(vnfr["vdur"]):
- for iface_index, iface in enumerate(vdur["interfaces"]):
- if (
- iface.get("internal-connection-point-ref")
- == icp_inst_param["id-ref"]
- ):
- vnfr_update_text = "vdur.{}.interfaces.{}".format(
- vdur_index, iface_index
- )
- if icp_inst_param.get("ip-address"):
- vnfr_update[
- vnfr_update_text + ".ip-address"
- ] = increment_ip_mac(
- icp_inst_param.get("ip-address"),
- vdur.get("count-index", 0),
- )
- vnfr_update[
- vnfr_update_text + ".fixed-ip"
- ] = True
- if icp_inst_param.get("mac-address"):
- vnfr_update[
- vnfr_update_text + ".mac-address"
- ] = increment_ip_mac(
- icp_inst_param.get("mac-address"),
- vdur.get("count-index", 0),
- )
- vnfr_update[
- vnfr_update_text + ".fixed-mac"
- ] = True
- break
- # get ip address from instantiation parameters.vld.vnfd-connection-point-ref
- for vld_inst_param in get_iterable(indata.get("vld")):
- for vnfcp_inst_param in get_iterable(
- vld_inst_param.get("vnfd-connection-point-ref")
- ):
- if vnfcp_inst_param["member-vnf-index-ref"] != member_vnf_index:
- continue
- # look for iface
- for vdur_index, vdur in enumerate(vnfr["vdur"]):
- for iface_index, iface in enumerate(vdur["interfaces"]):
- if (
- iface.get("external-connection-point-ref")
- == vnfcp_inst_param["vnfd-connection-point-ref"]
- ):
- vnfr_update_text = "vdur.{}.interfaces.{}".format(
- vdur_index, iface_index
+ break
+ return vnfr_update
+
+ @staticmethod
+ def _update_indata_to_use_concrete_vim_network(
+ ifaces_forcing_vim_network: list, indata: dict, member_vnf_index: str
+ ) -> dict:
+ """Update indata in case pdu forces to use a concrete vim-network-name
+ Args:
+ ifaces_forcing_vim_network (list): List of interfaces that are connected to an existing VIM network.
+ indata (dict): Input data dictionary
+ member_vnf_index (str): VNF index as an identifier
+ Returns:
+ indata (dict): Input data dictionary
+ """
+ for iface_info in ifaces_forcing_vim_network:
+ if iface_info.get("ns-vld-id"):
+ if "vld" not in indata:
+ indata["vld"] = []
+ indata["vld"].append(
+ {
+ key: iface_info[key]
+ for key in ("name", "vim-network-name", "vim-network-id")
+ if iface_info.get(key)
+ }
+ )
+
+ elif iface_info.get("vnf-vld-id"):
+ if "vnf" not in indata:
+ indata["vnf"] = []
+ indata["vnf"].append(
+ {
+ "member-vnf-index": member_vnf_index,
+ "internal-vld": [
+ {
+ key: iface_info[key]
+ for key in (
+ "name",
+ "vim-network-name",
+ "vim-network-id",
)
- if vnfcp_inst_param.get("ip-address"):
- vnfr_update[
- vnfr_update_text + ".ip-address"
- ] = increment_ip_mac(
- vnfcp_inst_param.get("ip-address"),
- vdur.get("count-index", 0),
- )
- vnfr_update[vnfr_update_text + ".fixed-ip"] = True
- if vnfcp_inst_param.get("mac-address"):
- vnfr_update[
- vnfr_update_text + ".mac-address"
- ] = increment_ip_mac(
- vnfcp_inst_param.get("mac-address"),
- vdur.get("count-index", 0),
- )
- vnfr_update[vnfr_update_text + ".fixed-mac"] = True
- break
+ if iface_info.get(key)
+ }
+ ],
+ }
+ )
+ return indata
+
+ def _update_vnfrs(
+ self, session: dict, rollback: list, nsr: dict, indata: dict
+ ) -> None:
+ """Update VNF record using indata.
+ Args:
+ session (dict): Contains "username", "admin", "force", "public", "project_id", "set_project"
+ rollback (list): List with the database modifications to rollback if needed
+ nsr (dict): NS record to get the related VNF record.
+ indata (dict): Input data dictionary
+
+ Returns:
+ None
+
+ Raises:
+ None
+ """
+ # Get vnfr
+ nsr_id = nsr["_id"]
+ vnfrs = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
- vnfr_update["vim-account-id"] = vim_account
- vnfr_update_rollback["vim-account-id"] = vnfr.get("vim-account-id")
+ for vnfr in vnfrs:
+ vnfr_update = {}
+ vnfr_update_rollback = {}
+ member_vnf_index = vnfr["member-vnf-index-ref"]
+
+ # Update vim-account-id
+ vim_account = indata.get("vimAccountId")
+ paas_account = indata.get("paasAccountId")
+ vca_id = None
+ if vim_account:
+ vca_id = self._get_vim_account(vim_account, session).get("vca")
+
+ # Check instantiate parameters
+ for vnf_inst_params in get_iterable(indata.get("vnf")):
+ if vnf_inst_params["member-vnf-index"] != member_vnf_index:
+ continue
+ if vnf_inst_params.get("vimAccountId"):
+ vim_account = vnf_inst_params["vimAccountId"]
+ vca_id = self._get_vim_account(vim_account, session).get("vca")
+ elif vnf_inst_params.get("paasAccountId"):
+ paas_account = vnf_inst_params["paasAccountId"]
+
+ # Get vnf.vdu.interface instantiation params to update vnfr.vdur.interfaces ip, mac
+ vnfr_update = (
+ NsLcmOpTopic._get_vdu_interfaces_from_instantiation_params(
+ vnfr_update, vnf_inst_params, vnfr
+ )
+ )
+
+ # Get vnf.internal-vld.internal-conection-point instantiation params to update vnfr.vdur.interfaces
+ # TODO update vld with the ip-profile
+ vnfr_update = NsLcmOpTopic._get_vld_icp_from_instantiation_params(
+ vnf_inst_params, vnfr, vnfr_update
+ )
+
+ # Get ip address from instantiation parameters.vld.vnfd-connection-point-ref
+ vnfr_update = NsLcmOpTopic._get_ip_address_from_instantiation_params(
+ indata, member_vnf_index, vnfr, vnfr_update
+ )
+
+ if vim_account:
+ vnfr_update["vim-account-id"] = vim_account
+ vnfr_update_rollback["vim-account-id"] = vnfr.get("vim-account-id")
+ elif paas_account:
+ vnfr_update["paas-account-id"] = paas_account
+ vnfr_update_rollback["paas-account-id"] = vnfr.get("paas-account-id")
if vca_id:
vnfr_update["vca-id"] = vca_id
vnfr_update_rollback["vca-id"] = vnfr.get("vca-id")
- # get pdu
+ # Get pdu
ifaces_forcing_vim_network = self._look_for_pdu(
session, rollback, vnfr, vim_account, vnfr_update, vnfr_update_rollback
)
- # get kdus
+ # Get kdus
ifaces_forcing_vim_network += self._look_for_k8scluster(
session, rollback, vnfr, vim_account, vnfr_update, vnfr_update_rollback
)
- # update database vnfr
+
+ # Update vnfr in database
self.db.set_one("vnfrs", {"_id": vnfr["_id"]}, vnfr_update)
rollback.append(
{
}
)
- # Update indada in case pdu forces to use a concrete vim-network-name
+ # Update indata in case pdu forces to use a concrete vim-network-name
# TODO check if user has already insert a vim-network-name and raises an error
if not ifaces_forcing_vim_network:
continue
- for iface_info in ifaces_forcing_vim_network:
- if iface_info.get("ns-vld-id"):
- if "vld" not in indata:
- indata["vld"] = []
- indata["vld"].append(
- {
- key: iface_info[key]
- for key in ("name", "vim-network-name", "vim-network-id")
- if iface_info.get(key)
- }
- )
-
- elif iface_info.get("vnf-vld-id"):
- if "vnf" not in indata:
- indata["vnf"] = []
- indata["vnf"].append(
- {
- "member-vnf-index": member_vnf_index,
- "internal-vld": [
- {
- key: iface_info[key]
- for key in (
- "name",
- "vim-network-name",
- "vim-network-id",
- )
- if iface_info.get(key)
- }
- ],
- }
- )
+ indata = NsLcmOpTopic._update_indata_to_use_concrete_vim_network(
+ ifaces_forcing_vim_network, indata, member_vnf_index
+ )
@staticmethod
def _create_nslcmop(nsr_id, operation, params):
HTTPStatus.CONFLICT,
)
self._check_ns_operation(session, nsr, operation, indata)
- if (indata.get("primitive_params")):
+ if indata.get("primitive_params"):
indata["primitive_params"] = json.dumps(indata["primitive_params"])
- elif (indata.get("additionalParamsForVnf")):
- indata["additionalParamsForVnf"] = json.dumps(indata["additionalParamsForVnf"])
+ elif indata.get("additionalParamsForVnf"):
+ indata["additionalParamsForVnf"] = json.dumps(
+ indata["additionalParamsForVnf"]
+ )
if operation == "instantiate":
self._update_vnfrs_from_nsd(nsr)
vnfd = self.db.get_one("vnfds", {"_id": vnfd_id})
nsd = self.db.get_one("nsds", {"_id": nsr["nsd-id"]})
ns_request = nsr["instantiate_params"]
- vnfr = self.db.get_one("vnfrs", {"_id": indata["changeVnfPackageData"]["vnfInstanceId"]})
+ vnfr = self.db.get_one(
+ "vnfrs", {"_id": indata["changeVnfPackageData"]["vnfInstanceId"]}
+ )
latest_vnfd_revision = vnfd["_admin"].get("revision", 1)
vnfr_vnfd_revision = vnfr.get("revision", 1)
if latest_vnfd_revision != vnfr_vnfd_revision:
old_vnfd_id = vnfd_id + ":" + str(vnfr_vnfd_revision)
- old_db_vnfd = self.db.get_one("vnfds_revisions", {"_id": old_vnfd_id})
+ old_db_vnfd = self.db.get_one(
+ "vnfds_revisions", {"_id": old_vnfd_id}
+ )
old_sw_version = old_db_vnfd.get("software-version", "1.0")
new_sw_version = vnfd.get("software-version", "1.0")
if new_sw_version != old_sw_version:
vnf_index = vnfr["member-vnf-index-ref"]
self.logger.info("nsr {}".format(nsr))
for vdu in vnfd["vdu"]:
- self.nsrtopic._add_flavor_to_nsr(vdu, vnfd, nsr, vnf_index, latest_vnfd_revision)
+ self.nsrtopic._add_flavor_to_nsr(
+ vdu, vnfd, nsr, vnf_index, latest_vnfd_revision
+ )
sw_image_id = vdu.get("sw-image-desc")
if sw_image_id:
- image_data = self.nsrtopic._get_image_data_from_vnfd(vnfd, sw_image_id)
+ image_data = self.nsrtopic._get_image_data_from_vnfd(
+ vnfd, sw_image_id
+ )
self.nsrtopic._add_image_to_nsr(nsr, image_data)
for alt_image in vdu.get("alternative-sw-image-desc", ()):
- image_data = self.nsrtopic._get_image_data_from_vnfd(vnfd, alt_image)
+ image_data = self.nsrtopic._get_image_data_from_vnfd(
+ vnfd, alt_image
+ )
self.nsrtopic._add_image_to_nsr(nsr, image_data)
nsr_update["image"] = nsr["image"]
nsr_update["flavor"] = nsr["flavor"]
self.db.set_one("nsrs", {"_id": nsr["_id"]}, nsr_update)
- ns_k8s_namespace = self.nsrtopic._get_ns_k8s_namespace(nsd, ns_request, session)
- vnfr_descriptor = self.nsrtopic._create_vnfr_descriptor_from_vnfd(
- nsd,
- vnfd,
- vnfd_id,
- vnf_index,
- nsr,
- ns_request,
- ns_k8s_namespace,
- latest_vnfd_revision,
+ ns_k8s_namespace = self.nsrtopic._get_ns_k8s_namespace(
+ nsd, ns_request, session
+ )
+ vnfr_descriptor = (
+ self.nsrtopic._create_vnfr_descriptor_from_vnfd(
+ nsd,
+ vnfd,
+ vnfd_id,
+ vnf_index,
+ nsr,
+ ns_request,
+ ns_k8s_namespace,
+ latest_vnfd_revision,
+ )
)
indata["newVdur"] = vnfr_descriptor["vdur"]
nslcmop_desc = self._create_nslcmop(nsInstanceId, operation, indata)
projects / osm / NBI.git / commitdiff