tutorials/ApacheCNF/automation/onboard.py

   1 # ============LICENSE_START=======================================================
   2 # Copyright (C) 2021 Orange
   3 # ================================================================================
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #      http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15 #
  16 # ============LICENSE_END=========================================================
  17
  18 import logging
  19 import os
  20 import time
  21 import zipfile
  22 from io import BytesIO
  23
  24 import yaml
  25
  26 from config import Config, VariablesDict
  27 import onapsdk.constants as const
  28
  29 from onapsdk.sdc.vendor import Vendor
  30 from onapsdk.sdc.vsp import Vsp
  31 from onapsdk.sdc.vf import Vf
  32 from onapsdk.sdc.pnf import Pnf
  33 from onapsdk.sdc.service import Service, ServiceInstantiationType
  34 from onapsdk.exceptions import ResourceNotFound
  35
  36 logger = logging.getLogger()
  37 logger.setLevel(logging.DEBUG)
  38
  39
  40 def retrieve_service(service_name: str):
  41     logger.info("Retrieve service from SDC before onboarding")
  42     services = Service.get_all()
  43
  44     for found_service in services:
  45         if found_service.name == service_name:
  46             logging.info(f"Service {found_service.name} found in SDC, onboarding will not be executed")
  47             exit(0)
  48     return
  49
  50
  51 def onboard_vendor(vendor_name: str = "demo_vendor"):
  52     logger.info("******** Onboard Vendor *******")
  53     try:
  54         vendor = next(vendor for vendor in Vendor.get_all() if vendor.name.lower() == vendor_name.lower())
  55     except (StopIteration, ResourceNotFound):
  56         vendor = Vendor(vendor_name)
  57         vendor.onboard()
  58     return vendor
  59
  60
  61 def onboard_vsp(vsp_name, vsp_file, vendor):
  62     logger.info(f"******** Onboard VSP - {vsp_name} *******")
  63     mypath = os.path.dirname(os.path.realpath(__file__))
  64     vsp_path = os.path.join(mypath, vsp_file)
  65     vsp = None
  66     try:
  67         vsp = Vsp(name=vsp_name, vendor=vendor, package=open(vsp_path, 'rb'))
  68     except FileNotFoundError:
  69         logger.error(f"No vsp file was found for {vsp_name}!")
  70         exit(1)
  71     vsp.onboard()
  72     return vsp
  73
  74
  75 def onboard_pnf(pnf_name, vsp_name, vsp_file, vendor_name):
  76     logger.info(f"******** Onboard PNF - {pnf_name} *******")
  77     vendor = onboard_vendor(vendor_name=vendor_name)
  78     try:
  79         pnf = next(_pnf for _pnf in Pnf.get_all() if _pnf.name == pnf_name)
  80         logger.info("PNF with provided name exists in ONAP SDC, onboarding will not be executed")
  81     except (StopIteration, ResourceNotFound):
  82         pnf_vsp = onboard_vsp(vsp_name=vsp_name, vsp_file=vsp_file, vendor=vendor)
  83         pnf = Pnf(name=pnf_name, vsp=pnf_vsp)
  84         pnf.onboard()
  85     return pnf
  86
  87
  88 def onboard_vnf(vnf_name, vsp_name, vsp_file, vendor_name):
  89     logger.info(f"******** Onboard VNF - {vnf_name} *******")
  90     vendor = onboard_vendor(vendor_name=vendor_name)
  91     try:
  92         vnf = next(_vf for _vf in Vf.get_all() if _vf.name == vnf_name)
  93         if vnf.status != "Certified":
  94             logger.error("Selected VNF is not certified. Try Onboard VF with new name.")
  95             exit(1)
  96         logger.info("VNF with provided name exists in ONAP SDC, onboarding will not  be executed")
  97     except (StopIteration, ResourceNotFound):
  98         vnf_vsp = onboard_vsp(vsp_name=vsp_name, vsp_file=vsp_file, vendor=vendor)
  99         vnf = Vf(name=vnf_name, vsp=vnf_vsp)
 100         vnf.create()
 101         vnf.onboard()
 102     return vnf
 103
 104
 105 def create_service(service_name, is_macro: bool = True):
 106     logger.info("******** Create Service *******")
 107     if is_macro:
 108         service = Service(name=service_name,
 109                           instantiation_type=ServiceInstantiationType.MACRO)
 110     else:
 111         service = Service(name=service_name,
 112                           instantiation_type=ServiceInstantiationType.A_LA_CARTE)
 113     service.create()
 114     return service
 115
 116
 117 def read_sdnc_model_details(file):
 118     mypath = os.path.dirname(os.path.realpath(__file__))
 119     file_path = os.path.join(mypath, file)
 120     try:
 121         with zipfile.ZipFile(file_path, 'r') as package:
 122             try:
 123                 cba_io = BytesIO(package.read("CBA.zip"))
 124                 with zipfile.ZipFile(cba_io) as cba:
 125                     with cba.open('TOSCA-Metadata/TOSCA.meta') as meta_file:
 126                         tosca_meta = yaml.load(meta_file, Loader=yaml.SafeLoader)
 127                         sdnc_model_name = tosca_meta.get("Template-Name")
 128                         sdnc_model_version = tosca_meta.get("Template-Version")
 129                         return sdnc_model_name, sdnc_model_version
 130             except KeyError:
 131                 logger.info("No CBA file was found")
 132                 return None, None
 133     except FileNotFoundError:
 134         logger.error("No vsp file was found!")
 135         exit(1)
 136
 137
 138 def set_properties(service, xnf, vsp_details):
 139     sdnc_model_name, sdnc_model_version = read_sdnc_model_details(vsp_details["vsp_file"])
 140     if sdnc_model_name and sdnc_model_version:
 141         if service.status == const.DRAFT:
 142             logger.info("******** Set SDNC properties for VF ********")
 143             component = service.get_component(xnf)
 144             prop = component.get_property("sdnc_model_name")
 145             prop.value = sdnc_model_name
 146             prop = component.get_property("sdnc_model_version")
 147             prop.value = sdnc_model_version
 148             prop = component.get_property("controller_actor")
 149             prop.value = "CDS"
 150             prop = component.get_property("sdnc_artifact_name")
 151             prop.value = vsp_details["sdnc_artifact_name"]
 152             prop = component.get_property("skip_post_instantiation_configuration")
 153             prop.value = vsp_details["skip_post_instantiation_configuration"]
 154
 155
 156 def check_distribution_status(service):
 157     logger.info("******** Check Service Distribution *******")
 158     distribution_completed = False
 159     nb_try = 0
 160     nb_try_max = 10
 161     while distribution_completed is False and nb_try < nb_try_max:
 162         distribution_completed = service.distributed
 163         if distribution_completed is True:
 164             logger.info(f"Service Distribution for {service.name} is successfully finished")
 165             break
 166         logger.info(f"Service Distribution for {service.name} ongoing, Wait for 60 s")
 167         time.sleep(60)
 168         nb_try += 1
 169
 170     if distribution_completed is False:
 171         logger.error(f"Service Distribution for {service.name} failed !!", )
 172         exit(1)
 173
 174
 175 def main():
 176     config = Config(env_dict=VariablesDict.env_variable)
 177     retrieve_service(service_name=config.service_model["model_name"])
 178
 179     logger.info("******** SERVICE DESIGN *******")
 180     service = create_service(service_name=config.service_model["model_name"],
 181                              is_macro=config.service_model["macro_orchestration"])
 182     vnfs = config.service_model.get("vnfs")
 183     if vnfs:
 184         for vnf in vnfs:
 185             new_vnf = onboard_vnf(vnf_name=vnf["model_name"],
 186                                   vsp_name="VSP" + "_" + vnf["model_name"],
 187                                   vsp_file=vnf["vsp"]["vsp_file"],
 188                                   vendor_name=vnf["vsp"]["vendor"])
 189             service.add_resource(new_vnf)
 190             set_properties(service=service, xnf=new_vnf, vsp_details=vnf["vsp"])
 191
 192     pnfs = config.service_model.get("pnfs")
 193     if pnfs:
 194         for pnf in pnfs:
 195             new_pnf = onboard_pnf(pnf_name=pnf["model_name"],
 196                                   vsp_name="VSP" + "_" + pnf["model_name"],
 197                                   vsp_file=pnf["vsp"]["vsp_file"],
 198                                   vendor_name=pnf["vsp"]["vendor"])
 199             service.add_resource(new_pnf)
 200             set_properties(service=service, xnf=new_pnf, vsp_details=pnf["vsp"])
 201
 202     service.checkin()
 203     service.onboard()
 204     check_distribution_status(service)
 205
 206
 207 if __name__ == "__main__":
 208     sh = logging.StreamHandler()
 209     sh_formatter = logging.Formatter('%(asctime)s %(levelname)s %(lineno)d:%(filename)s(%(process)d) - %(message)s')
 210     sh.setFormatter(sh_formatter)
 211     logger.addHandler(sh)
 212
 213     main()