from n2vc.utils import get_ee_id_components, generate_random_alfanum_string
from n2vc.vca.connection import get_connection
from retrying_async import retry
+from typing import Tuple
class N2VCJujuConnector(N2VCConnector):
return get_ee_id_components(ee_id)
- def _get_application_name(self, namespace: str) -> str:
- """
- Build application name from namespace
- :param namespace:
- :return: app-vnf-<vnf id>-vdu-<vdu-id>-cnt-<vdu-count>-<random_value>
+ @staticmethod
+ def _find_charm_level(vnf_id: str, vdu_id: str) -> str:
+ """Decides the charm level.
+ Args:
+ vnf_id (str): VNF id
+ vdu_id (str): VDU id
+
+ Returns:
+ charm_level (str): ns-level or vnf-level or vdu-level
"""
+ if vdu_id and not vnf_id:
+ raise N2VCException(message="If vdu-id exists, vnf-id should be provided.")
+ if vnf_id and vdu_id:
+ return "vdu-level"
+ if vnf_id and not vdu_id:
+ return "vnf-level"
+ if not vnf_id and not vdu_id:
+ return "ns-level"
+
+ @staticmethod
+ def _generate_backward_compatible_application_name(
+ vnf_id: str, vdu_id: str, vdu_count: str
+ ) -> str:
+ """Generate backward compatible application name
+ by limiting the app name to 50 characters.
- # TODO: Enforce the Juju 50-character application limit
+ Args:
+ vnf_id (str): VNF ID
+ vdu_id (str): VDU ID
+ vdu_count (str): vdu-count-index
- # split namespace components
- _, _, vnf_id, vdu_id, vdu_count = self._get_namespace_components(
- namespace=namespace
- )
+ Returns:
+ application_name (str): generated application name
+ """
if vnf_id is None or len(vnf_id) == 0:
vnf_id = ""
else:
application_name = "app-{}{}{}-{}".format(
vnf_id, vdu_id, vdu_count, random_suffix
)
+ return application_name
+
+ @staticmethod
+ def _generate_application_name(
+ charm_level: str,
+ vnfrs: dict,
+ vca_records: list,
+ vnf_count: str = None,
+ vdu_count: str = None,
+ ) -> str:
+ """Generate application name to make the relevant charm of VDU/KDU
+ in the VNFD descriptor become clearly visible.
+ Limiting the app name to 50 characters.
+
+ Args:
+ charm_level (str): VNF ID
+ vnfrs (dict): VDU ID
+ vca_records (list): db_nsr["_admin"]["deployed"]["VCA"] as list
+ vnf_count (str): vnf count index
+ vdu_count (str): vdu count index
+
+ Returns:
+ application_name (str): generated application name
+
+ """
+ application_name = ""
+ if charm_level == "ns-level":
+ if len(vca_records) != 1:
+ raise N2VCException(message="One VCA record is expected.")
+ # Only one VCA record is expected if it's ns-level charm.
+ # Shorten the charm name to its first 40 characters.
+ charm_name = vca_records[0]["charm_name"][:40]
+ if not charm_name:
+ raise N2VCException(message="Charm name should be provided.")
+ application_name = charm_name + "-ns"
+
+ elif charm_level == "vnf-level":
+ if len(vca_records) < 1:
+ raise N2VCException(message="One or more VCA record is expected.")
+ # If VNF is scaled, more than one VCA record may be included in vca_records
+ # but ee_descriptor_id is same.
+ # Shorten the ee_descriptor_id and member-vnf-index-ref
+ # to first 12 characters.
+ application_name = (
+ vca_records[0]["ee_descriptor_id"][:12]
+ + "-"
+ + vnf_count
+ + "-"
+ + vnfrs["member-vnf-index-ref"][:12]
+ + "-vnf"
+ )
+ elif charm_level == "vdu-level":
+ if len(vca_records) < 1:
+ raise N2VCException(message="One or more VCA record is expected.")
+ vdu_profile_id = vnfrs["vdur"][int(vdu_count)]["vdu-id-ref"]
+ # If vnf/vdu is scaled, more than one VCA record may be included in vca_records
+ # but ee_descriptor_id is same.
+ # Shorten the ee_descriptor_id, member-vnf-index-ref and vdu_profile_id
+ # to first 12 characters.
+ application_name = (
+ vca_records[0]["ee_descriptor_id"][:12]
+ + "-"
+ + vnf_count
+ + "-"
+ + vnfrs["member-vnf-index-ref"][:12]
+ + "-"
+ + vdu_profile_id[:12]
+ + "-"
+ + vdu_count
+ + "-vdu"
+ )
+
+ return application_name
+
+ def _get_vnf_count_and_record(
+ self, charm_level: str, vnf_id_and_count: str
+ ) -> Tuple[str, dict]:
+ """Get the vnf count and VNF record depend on charm level
+
+ Args:
+ charm_level (str)
+ vnf_id_and_count (str)
+
+ Returns:
+ (vnf_count (str), db_vnfr(dict)) as Tuple
+
+ """
+ vnf_count = ""
+ db_vnfr = {}
+
+ if charm_level in ("vnf-level", "vdu-level"):
+ vnf_id = "-".join(vnf_id_and_count.split("-")[:-1])
+ vnf_count = vnf_id_and_count.split("-")[-1]
+ db_vnfr = self.db.get_one("vnfrs", {"_id": vnf_id})
+
+ # If the charm is ns level, it returns empty vnf_count and db_vnfr
+ return vnf_count, db_vnfr
+
+ @staticmethod
+ def _get_vca_records(charm_level: str, db_nsr: dict, db_vnfr: dict) -> list:
+ """Get the VCA records from db_nsr dict
+
+ Args:
+ charm_level (str): level of charm
+ db_nsr (dict): NS record from database
+ db_vnfr (dict): VNF record from database
+
+ Returns:
+ vca_records (list): List of VCA record dictionaries
+
+ """
+ vca_records = {}
+ if charm_level == "ns-level":
+ vca_records = list(
+ filter(
+ lambda vca_record: vca_record["target_element"] == "ns",
+ db_nsr["_admin"]["deployed"]["VCA"],
+ )
+ )
+ elif charm_level in ["vnf-level", "vdu-level"]:
+ vca_records = list(
+ filter(
+ lambda vca_record: vca_record["member-vnf-index"]
+ == db_vnfr["member-vnf-index-ref"],
+ db_nsr["_admin"]["deployed"]["VCA"],
+ )
+ )
+
+ return vca_records
+
+ def _get_application_name(self, namespace: str) -> str:
+ """Build application name from namespace
+
+ Application name structure:
+ NS level: <charm-name>-ns
+ VNF level: <ee-name>-z<vnf-ordinal-scale-number>-<vnf-profile-id>-vnf
+ VDU level: <ee-name>-z<vnf-ordinal-scale-number>-<vnf-profile-id>-
+ <vdu-profile-id>-z<vdu-ordinal-scale-number>-vdu
+
+ Application naming for backward compatibility (old structure):
+ NS level: app-<random_value>
+ VNF level: app-vnf-<vnf-id>-z<ordinal-scale-number>-<random_value>
+ VDU level: app-vnf-<vnf-id>-z<vnf-ordinal-scale-number>-vdu-
+ <vdu-id>-cnt-<vdu-count>-z<vdu-ordinal-scale-number>-<random_value>
+
+ Args:
+ namespace (str)
+
+ Returns:
+ application_name (str)
+
+ """
+ # split namespace components
+ (
+ nsi_id,
+ ns_id,
+ vnf_id_and_count,
+ vdu_id,
+ vdu_count,
+ ) = self._get_namespace_components(namespace=namespace)
+
+ if not ns_id:
+ raise N2VCException(message="ns-id should be provided.")
+
+ charm_level = self._find_charm_level(vnf_id_and_count, vdu_id)
+ db_nsr = self.db.get_one("nsrs", {"_id": ns_id})
+ vnf_count, db_vnfr = self._get_vnf_count_and_record(
+ charm_level, vnf_id_and_count
+ )
+ vca_records = self._get_vca_records(charm_level, db_nsr, db_vnfr)
+
+ if all("charm_name" in vca_record.keys() for vca_record in vca_records):
+ application_name = self._generate_application_name(
+ charm_level,
+ db_vnfr,
+ vca_records,
+ vnf_count=vnf_count,
+ vdu_count=vdu_count,
+ )
+ else:
+ application_name = self._generate_backward_compatible_application_name(
+ vnf_id_and_count, vdu_id, vdu_count
+ )
return N2VCJujuConnector._format_app_name(application_name)
projects / osm / N2VC.git / blobdiff