diff options
Diffstat (limited to 'tutorials/ApacheCNF/automation/onboard.py')
| -rw-r--r-- | tutorials/ApacheCNF/automation/onboard.py | 213 |
1 files changed, 213 insertions, 0 deletions
diff --git a/tutorials/ApacheCNF/automation/onboard.py b/tutorials/ApacheCNF/automation/onboard.py new file mode 100644 index 00000000..7e27419c --- /dev/null +++ b/tutorials/ApacheCNF/automation/onboard.py @@ -0,0 +1,213 @@ +# ============LICENSE_START======================================================= +# Copyright (C) 2021 Orange +# ================================================================================ +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# ============LICENSE_END========================================================= + +import logging +import os +import time +import zipfile +from io import BytesIO + +import yaml + +from config import Config, VariablesDict +import onapsdk.constants as const + +from onapsdk.sdc.vendor import Vendor +from onapsdk.sdc.vsp import Vsp +from onapsdk.sdc.vf import Vf +from onapsdk.sdc.pnf import Pnf +from onapsdk.sdc.service import Service, ServiceInstantiationType +from onapsdk.exceptions import ResourceNotFound + +logger = logging.getLogger() +logger.setLevel(logging.DEBUG) + + +def retrieve_service(service_name: str): + logger.info("Retrieve service from SDC before onboarding") + services = Service.get_all() + + for found_service in services: + if found_service.name == service_name: + logging.info(f"Service {found_service.name} found in SDC, onboarding will not be executed") + exit(0) + return + + +def onboard_vendor(vendor_name: str = "demo_vendor"): + logger.info("******** Onboard Vendor *******") + try: + vendor = next(vendor for vendor in Vendor.get_all() if vendor.name.lower() == vendor_name.lower()) + except (StopIteration, ResourceNotFound): + vendor = Vendor(vendor_name) + vendor.onboard() + return vendor + + +def onboard_vsp(vsp_name, vsp_file, vendor): + logger.info(f"******** Onboard VSP - {vsp_name} *******") + mypath = os.path.dirname(os.path.realpath(__file__)) + vsp_path = os.path.join(mypath, vsp_file) + vsp = None + try: + vsp = Vsp(name=vsp_name, vendor=vendor, package=open(vsp_path, 'rb')) + except FileNotFoundError: + logger.error(f"No vsp file was found for {vsp_name}!") + exit(1) + vsp.onboard() + return vsp + + +def onboard_pnf(pnf_name, vsp_name, vsp_file, vendor_name): + logger.info(f"******** Onboard PNF - {pnf_name} *******") + vendor = onboard_vendor(vendor_name=vendor_name) + try: + pnf = next(_pnf for _pnf in Pnf.get_all() if _pnf.name == pnf_name) + logger.info("PNF with provided name exists in ONAP SDC, onboarding will not be executed") + except (StopIteration, ResourceNotFound): + pnf_vsp = onboard_vsp(vsp_name=vsp_name, vsp_file=vsp_file, vendor=vendor) + pnf = Pnf(name=pnf_name, vsp=pnf_vsp) + pnf.onboard() + return pnf + + +def onboard_vnf(vnf_name, vsp_name, vsp_file, vendor_name): + logger.info(f"******** Onboard VNF - {vnf_name} *******") + vendor = onboard_vendor(vendor_name=vendor_name) + try: + vnf = next(_vf for _vf in Vf.get_all() if _vf.name == vnf_name) + if vnf.status != "Certified": + logger.error("Selected VNF is not certified. Try Onboard VF with new name.") + exit(1) + logger.info("VNF with provided name exists in ONAP SDC, onboarding will not be executed") + except (StopIteration, ResourceNotFound): + vnf_vsp = onboard_vsp(vsp_name=vsp_name, vsp_file=vsp_file, vendor=vendor) + vnf = Vf(name=vnf_name, vsp=vnf_vsp) + vnf.create() + vnf.onboard() + return vnf + + +def create_service(service_name, is_macro: bool = True): + logger.info("******** Create Service *******") + if is_macro: + service = Service(name=service_name, + instantiation_type=ServiceInstantiationType.MACRO) + else: + service = Service(name=service_name, + instantiation_type=ServiceInstantiationType.A_LA_CARTE) + service.create() + return service + + +def read_sdnc_model_details(file): + mypath = os.path.dirname(os.path.realpath(__file__)) + file_path = os.path.join(mypath, file) + try: + with zipfile.ZipFile(file_path, 'r') as package: + try: + cba_io = BytesIO(package.read("CBA.zip")) + with zipfile.ZipFile(cba_io) as cba: + with cba.open('TOSCA-Metadata/TOSCA.meta') as meta_file: + tosca_meta = yaml.load(meta_file, Loader=yaml.SafeLoader) + sdnc_model_name = tosca_meta.get("Template-Name") + sdnc_model_version = tosca_meta.get("Template-Version") + return sdnc_model_name, sdnc_model_version + except KeyError: + logger.info("No CBA file was found") + return None, None + except FileNotFoundError: + logger.error("No vsp file was found!") + exit(1) + + +def set_properties(service, xnf, vsp_details): + sdnc_model_name, sdnc_model_version = read_sdnc_model_details(vsp_details["vsp_file"]) + if sdnc_model_name and sdnc_model_version: + if service.status == const.DRAFT: + logger.info("******** Set SDNC properties for VF ********") + component = service.get_component(xnf) + prop = component.get_property("sdnc_model_name") + prop.value = sdnc_model_name + prop = component.get_property("sdnc_model_version") + prop.value = sdnc_model_version + prop = component.get_property("controller_actor") + prop.value = "CDS" + prop = component.get_property("sdnc_artifact_name") + prop.value = vsp_details["sdnc_artifact_name"] + prop = component.get_property("skip_post_instantiation_configuration") + prop.value = vsp_details["skip_post_instantiation_configuration"] + + +def check_distribution_status(service): + logger.info("******** Check Service Distribution *******") + distribution_completed = False + nb_try = 0 + nb_try_max = 10 + while distribution_completed is False and nb_try < nb_try_max: + distribution_completed = service.distributed + if distribution_completed is True: + logger.info(f"Service Distribution for {service.name} is successfully finished") + break + logger.info(f"Service Distribution for {service.name} ongoing, Wait for 60 s") + time.sleep(60) + nb_try += 1 + + if distribution_completed is False: + logger.error(f"Service Distribution for {service.name} failed !!", ) + exit(1) + + +def main(): + config = Config(env_dict=VariablesDict.env_variable) + retrieve_service(service_name=config.service_model["model_name"]) + + logger.info("******** SERVICE DESIGN *******") + service = create_service(service_name=config.service_model["model_name"], + is_macro=config.service_model["macro_orchestration"]) + vnfs = config.service_model.get("vnfs") + if vnfs: + for vnf in vnfs: + new_vnf = onboard_vnf(vnf_name=vnf["model_name"], + vsp_name="VSP" + "_" + vnf["model_name"], + vsp_file=vnf["vsp"]["vsp_file"], + vendor_name=vnf["vsp"]["vendor"]) + service.add_resource(new_vnf) + set_properties(service=service, xnf=new_vnf, vsp_details=vnf["vsp"]) + + pnfs = config.service_model.get("pnfs") + if pnfs: + for pnf in pnfs: + new_pnf = onboard_pnf(pnf_name=pnf["model_name"], + vsp_name="VSP" + "_" + pnf["model_name"], + vsp_file=pnf["vsp"]["vsp_file"], + vendor_name=pnf["vsp"]["vendor"]) + service.add_resource(new_pnf) + set_properties(service=service, xnf=new_pnf, vsp_details=pnf["vsp"]) + + service.checkin() + service.onboard() + check_distribution_status(service) + + +if __name__ == "__main__": + sh = logging.StreamHandler() + sh_formatter = logging.Formatter('%(asctime)s %(levelname)s %(lineno)d:%(filename)s(%(process)d) - %(message)s') + sh.setFormatter(sh_formatter) + logger.addHandler(sh) + + main() |