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
8 # http://www.apache.org/licenses/LICENSE-2.0
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.
16 # ============LICENSE_END=========================================================
22 from instantiate import read_sdnc_model_details
23 from config import Config
24 import onapsdk.constants as const
26 from onapsdk.sdc.vendor import Vendor
27 from onapsdk.sdc.vsp import Vsp
28 from onapsdk.sdc.vf import Vf
29 from onapsdk.sdc.pnf import Pnf
30 from onapsdk.sdc.service import Service, ServiceInstantiationType
34 logger = logging.getLogger("")
35 logger.setLevel(logging.DEBUG)
36 fh = logging.StreamHandler()
37 fh_formatter = logging.Formatter('%(asctime)s %(levelname)s %(lineno)d:%(filename)s(%(process)d) - %(message)s')
38 fh.setFormatter(fh_formatter)
42 def retrieve_service(service_name: str):
43 logger.info("Retrieve service from SDC before onboarding")
44 services = Service.get_all()
46 for found_service in services:
47 if found_service.name == service_name:
48 logging.info(f"Service {found_service.name} found in SDC, onboarding will not be executed")
53 def onboard_vendor(vendor_name: str = "demo_vendor"):
54 logger.info("******** Onboard Vendor *******")
55 vendor = Vendor(vendor_name)
60 def onboard_vsp(vsp_name, vsp_file, vendor):
61 logger.info(f"******** Onboard VSP - {vsp_name} *******")
62 mypath = os.path.dirname(os.path.realpath(__file__))
63 vsp_path = os.path.join(mypath, vsp_file)
66 vsp = Vsp(name=vsp_name, vendor=vendor, package=open(vsp_path, 'rb'))
67 except FileNotFoundError:
68 logger.error(f"No vsp file was found for {vsp_name}!")
74 def onboard_pnf(pnf_name, vsp_name, vsp_file, vendor):
75 logger.info(f"******** Onboard PNF - {pnf_name} *******")
76 pnf_vsp = onboard_vsp(vsp_name=vsp_name, vsp_file=vsp_file, vendor=vendor)
77 pnf = Pnf(name=pnf_name, vsp=pnf_vsp)
82 def onboard_vnf(vnf_name, vsp_name, vsp_file, vendor):
83 logger.info(f"******** Onboard VNF - {vnf_name} *******")
84 vnf_vsp = onboard_vsp(vsp_name=vsp_name, vsp_file=vsp_file, vendor=vendor)
85 vnf = Vf(name=vnf_name, vsp=vnf_vsp)
91 def create_service(service_name, is_macro: bool = True):
92 logger.info("******** Create Service *******")
94 svc = Service(name=service_name,
95 instantiation_type=ServiceInstantiationType.MACRO)
97 svc = Service(name=service_name,
98 instantiation_type=ServiceInstantiationType.A_LA_CARTE)
103 def set_properties(service, xnf):
104 sdnc_model_name, sdnc_model_version = read_sdnc_model_details(Config.VSPFILE)
105 if service.status == const.DRAFT:
106 logger.info("******** Set SDNC properties for VF ********")
107 component = service.get_component(xnf)
108 prop = component.get_property("sdnc_model_version")
109 prop.value = sdnc_model_version
110 prop = component.get_property("sdnc_artifact_name")
111 prop.value = Config.SDNC_ARTIFACT_NAME
112 prop = component.get_property("sdnc_model_name")
113 prop.value = sdnc_model_name
114 prop = component.get_property("controller_actor")
116 prop = component.get_property("skip_post_instantiation_configuration")
117 prop.value = Config.SKIP_POST_INSTANTIATION
120 def check_distribution_status(service):
121 logger.info("******** Check Service Distribution *******")
122 distribution_completed = False
125 while distribution_completed is False and nb_try < nb_try_max:
126 distribution_completed = service.distributed
127 if distribution_completed is True:
128 logger.info(f"Service Distribution for {service.name} is successfully finished")
130 logger.info(f"Service Distribution for {service.name} ongoing, Wait for 60 s")
134 if distribution_completed is False:
135 logger.error(f"Service Distribution for {service.name} failed !!", )
140 retrieve_service(service_name=Config.SERVICENAME)
141 logger.info("******** SERVICE DESIGN *******")
142 vendor = onboard_vendor(vendor_name=Config.VENDOR)
143 service = create_service(Config.SERVICENAME, Config.MACRO_INSTANTIATION)
145 pnf = onboard_pnf(pnf_name=Config.PNF_NAME,
146 vsp_name=Config.PNF_VSP_NAME,
147 vsp_file=Config.PNF_VSP_FILE,
149 service.add_resource(pnf)
150 vnf = onboard_vnf(vnf_name=Config.VFNAME,
151 vsp_name=Config.VSPNAME,
152 vsp_file=Config.VSPFILE,
154 service.add_resource(vnf)
155 set_properties(service, vnf)
158 check_distribution_status(service)
161 if __name__ == "__main__":