from re import (
match,
) # For checking that additional parameter names are valid Jinja2 identifiers
+from osm_nbi.temporal.nbi_temporal import NbiTemporal
__author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
def __init__(self, db, fs, msg, auth):
BaseTopic.__init__(self, db, fs, msg, auth)
- def _check_descriptor_dependencies(self, session, descriptor):
- """
- Check that the dependent descriptors exist on a new descriptor or edition
- :param session: client session information
- :param descriptor: descriptor to be inserted or edit
- :return: None or raises exception
- """
- if not descriptor.get("nsdId"):
- return
- nsd_id = descriptor["nsdId"]
- if not self.get_item_list(session, "nsds", {"id": nsd_id}):
- raise EngineException(
- "Descriptor error at nsdId='{}' references a non exist nsd".format(
- nsd_id
- ),
- http_code=HTTPStatus.CONFLICT,
- )
-
@staticmethod
def format_on_new(content, project_id=None, make_public=False):
BaseTopic.format_on_new(content, project_id=project_id, make_public=make_public)
for config in df["lcm-operations-configuration"][
"operate-vnf-op-config"
].get("day1-2", []):
- # Verify the target object (VNF|NS|VDU|KDU) where we need to populate
+ # Verify the target object (VNF|NS|VDU|KDU) where we need to populate
# the params with the additional ones given by the user
if config.get("id") == selector:
for primitive in get_iterable(
"<rw_mgmt_ip>",
"<VDU_SCALE_INFO>",
"<ns_config_info>",
+ "<OSM>",
):
continue
if (
EngineException, ValidationError, DbException, FsException, MsgException.
Note: Exceptions are not captured on purpose. They should be captured at called
"""
+ step = "checking quotas" # first step must be defined outside try
try:
- step = "checking quotas"
self.check_quota(session)
step = "validating input parameters"
"configurationStatus": None,
"vcaStatus": None,
"nsd": {k: v for k, v in nsd.items()},
- "vimdatacenter": ns_request.get("vimAccountId"),
- "paasdatacenter": ns_request.get("paasAccountId"),
+ "datacenter": ns_request["vimAccountId"],
"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": [],
def __init__(self, db, fs, msg, auth):
BaseTopic.__init__(self, db, fs, msg, auth)
self.nsrtopic = NsrTopic(db, fs, msg, auth)
+ self.temporal = NbiTemporal()
def _check_ns_operation(self, session, nsr, operation, indata):
"""
vnfd_id_2update = indata["changeVnfPackageData"]["vnfdId"]
if vnf_instance_id not in nsr["constituent-vnfr-ref"]:
-
raise EngineException(
f"Error in validating ns-update request: vnf {vnf_instance_id} does not "
f"belong to NS {ns_instance_id}",
# Check the given vnfd-id belongs to given vnf instance
if constituent_vnfd_id and (vnfd_id_2update != constituent_vnfd_id):
-
raise EngineException(
f"Error in validating ns-update request: vnfd-id {vnfd_id_2update} does not "
f"match with the vnfd-id: {constituent_vnfd_id} of VNF instance: {vnf_instance_id}",
def _check_heal_ns_operation(self, indata, nsr):
return
- 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 = {}
-
+ 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)
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(
)
)
- 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
- """
+ def _check_valid_vim_account(self, vim_account, vim_accounts, session):
+ if vim_account in vim_accounts:
+ return
try:
- 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)
+ db_filter = self._get_project_filter(session)
+ db_filter["_id"] = vim_account
+ self.db.get_one("vim_accounts", db_filter)
except Exception:
raise EngineException(
"Invalid vimAccountId='{}' not present for the project".format(
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
- )
- )
+ vim_accounts.append(vim_account)
def _get_vim_account(self, vim_id: str, session):
try:
"Invalid vimAccountId='{}' not present for the project".format(vim_id)
)
- 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
- """
+ 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
try:
- 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)
+ db_filter = self._get_project_filter(session)
+ db_filter["_id"] = wim_account
+ self.db.get_one("wim_accounts", db_filter)
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
+ ifaces_forcing_vim_network = []
if not vnfr.get("kdur"):
return ifaces_forcing_vim_network
return ifaces_forcing_vim_network
def _update_vnfrs_from_nsd(self, nsr):
+ step = "Getting vnf_profiles from nsd" # first step must be defined outside try
try:
nsr_id = nsr["_id"]
nsd = nsr["nsd"]
- step = "Getting vnf_profiles from nsd"
vnf_profiles = nsd.get("df", [{}])[0].get("vnf-profile", ())
vld_fixed_ip_connection_point_data = {}
) as e:
raise type(e)("{} while '{}'".format(e, step), http_code=e.http_code)
- @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
-
- 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
+ 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_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
+ 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 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"]:
+ 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:
continue
- 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
-
- @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
+ if vnf_inst_params.get("vimAccountId"):
+ vim_account = vnf_inst_params.get("vimAccountId")
+ vca_id = self._get_vim_account(vim_account, session).get("vca")
- 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"]
+ # 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")
):
+ 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 vnfcp_inst_param.get("ip-address"):
+ if iface_inst_param.get("ip-address"):
vnfr_update[
vnfr_update_text + ".ip-address"
] = increment_ip_mac(
- vnfcp_inst_param.get("ip-address"),
+ iface_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"):
+ if iface_inst_param.get("mac-address"):
vnfr_update[
vnfr_update_text + ".mac-address"
] = increment_ip_mac(
- vnfcp_inst_param.get("mac-address"),
+ iface_inst_param.get("mac-address"),
vdur.get("count-index", 0),
)
vnfr_update[vnfr_update_text + ".fixed-mac"] = True
- 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 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})
-
- 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
+ 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
- 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
- )
+ 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
+ )
+ 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 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")
+ vnfr_update["vim-account-id"] = vim_account
+ vnfr_update_rollback["vim-account-id"] = vnfr.get("vim-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 vnfr in database
+ # update database vnfr
self.db.set_one("vnfrs", {"_id": vnfr["_id"]}, vnfr_update)
rollback.append(
{
}
)
- # Update indata in case pdu forces to use a concrete vim-network-name
+ # Update indada 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
- indata = NsLcmOpTopic._update_indata_to_use_concrete_vim_network(
- ifaces_forcing_vim_network, indata, member_vnf_index
- )
+ 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)
+ }
+ ],
+ }
+ )
@staticmethod
def _create_nslcmop(nsr_id, operation, params):
self.db.create("nslcmops", nslcmop_desc)
rollback.append({"topic": "nslcmops", "_id": _id})
if not slice_object:
+ if "instantiate_params" in nsr:
+ if "vimAccountId" in nsr["instantiate_params"]:
+ vim = self._get_vim_account(
+ vim_id=nsr["instantiate_params"]["vimAccountId"],
+ session=session,
+ )
+ if vim["vim_type"] == "paas":
+ self.logger.info("Starting {} workflow".format(operation))
+ self.temporal.start_ns_workflow(nslcmop_desc)
+ return _id, None
self.msg.write("ns", operation, nslcmop_desc)
return _id, None
+ except ValidationError as e: # TODO remove try Except, it is captured at nbi.py
+ raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)
+ return _id, None
except ValidationError as e: # TODO remove try Except, it is captured at nbi.py
raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)
# except DbException as e:
additional_params[k] = "!!yaml " + safe_dump(v)
return additional_params
- def _check_descriptor_dependencies(self, session, descriptor):
- """
- Check that the dependent descriptors exist on a new descriptor or edition
- :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
- :param descriptor: descriptor to be inserted or edit
- :return: None or raises exception
- """
- if not descriptor.get("nst-ref"):
- return
- nstd_id = descriptor["nst-ref"]
- if not self.get_item_list(session, "nsts", {"id": nstd_id}):
- raise EngineException(
- "Descriptor error at nst-ref='{}' references a non exist nstd".format(
- nstd_id
- ),
- http_code=HTTPStatus.CONFLICT,
- )
-
def check_conflict_on_del(self, session, _id, db_content):
"""
Check that NSI is not instantiated
:return: the _id of nsi descriptor created at database
"""
+ step = "checking quotas" # first step must be defined outside try
try:
- step = "checking quotas"
self.check_quota(session)
step = ""
self.db.create("nsis", nsi_descriptor)
rollback.append({"topic": "nsis", "_id": nsi_id})
return nsi_id, None
+ except ValidationError as e:
+ raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)
except Exception as e: # TODO remove try Except, it is captured at nbi.py
self.logger.exception(
"Exception {} at NsiTopic.new()".format(e), exc_info=True
)
raise EngineException("Error {}: {}".format(step, e))
- except ValidationError as e:
- raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)
def edit(self, session, _id, indata=None, kwargs=None, content=None):
raise EngineException(
projects / osm / NBI.git / blobdiff