lcm/ns/biz/ns_instant.py

   1 # Copyright 2016-2017 ZTE Corporation.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14 import json
  15 import logging
  16 import time
  17 import traceback
  18 import uuid
  19 from threading import Thread
  20
  21 from rest_framework import status
  22
  23 from lcm.pub.config.config import WORKFLOW_OPTION
  24 from lcm.pub.database.models import DefPkgMappingModel, ServiceBaseInfoModel, InputParamMappingModel
  25 from lcm.pub.database.models import NSInstModel, VNFFGInstModel, WFPlanModel
  26 from lcm.pub.exceptions import NSLCMException
  27 from lcm.pub.msapi import activiti
  28 from lcm.pub.msapi import sdc_run_catalog
  29 from lcm.pub.msapi.catalog import get_process_id
  30 from lcm.pub.msapi.catalog import get_servicetemplate_id, get_servicetemplate
  31 from lcm.pub.msapi.extsys import select_vnfm
  32 from lcm.pub.msapi.wso2bpel import workflow_run
  33 from lcm.pub.utils.jobutil import JobUtil
  34 from lcm.pub.utils.values import ignore_case_get
  35 from lcm.workflows import build_in
  36
  37 logger = logging.getLogger(__name__)
  38
  39
  40 class BuildInWorkflowThread(Thread):
  41     def __init__(self, plan_input):
  42         Thread.__init__(self)
  43         self.plan_input = plan_input
  44
  45     def run(self):
  46         build_in.run_ns_instantiate(self.plan_input)
  47
  48
  49 class InstantNSService(object):
  50     def __init__(self, ns_inst_id, plan_content):
  51         self.ns_inst_id = ns_inst_id
  52         self.req_data = plan_content
  53
  54     def do_biz(self):
  55         try:
  56             job_id = JobUtil.create_job("NS", "NS_INST", self.ns_inst_id)
  57             logger.debug('ns-instant(%s) workflow starting...' % self.ns_inst_id)
  58             logger.debug('req_data=%s' % self.req_data)
  59             ns_inst = NSInstModel.objects.get(id=self.ns_inst_id)
  60
  61             input_parameters = []
  62             for key, val in self.req_data['additionalParamForNs'].items():
  63                 input_parameters.append({"key": key, "value": val})
  64
  65             vim_id = ''
  66             if 'location' in self.req_data['additionalParamForNs']:
  67                 vim_id = self.req_data['additionalParamForNs']['location']
  68             location_constraints = []
  69             if 'locationConstraints' in self.req_data:
  70                 location_constraints = self.req_data['locationConstraints']
  71
  72             JobUtil.add_job_status(job_id, 5, 'Start query nsd(%s)' % ns_inst.nspackage_id)
  73             dst_plan = sdc_run_catalog.parse_nsd(ns_inst.nspackage_id, input_parameters)
  74             logger.debug('tosca plan dest: %s' % dst_plan)
  75
  76             NSInstModel.objects.filter(id=self.ns_inst_id).update(nsd_model=dst_plan)
  77
  78             params_json = json.JSONEncoder().encode(self.req_data["additionalParamForNs"])
  79             # start
  80             params_vnf = []
  81             plan_dict = json.JSONDecoder().decode(dst_plan)
  82             for vnf in ignore_case_get(plan_dict, "ns_vnfs"):
  83                 vnfd_id = vnf['properties']['id']
  84                 vnfm_type = vnf['properties'].get("nf_type", "undefined")
  85                 vimid = self.get_vnf_vim_id(vim_id, location_constraints, vnfd_id)
  86                 vnfm_info = select_vnfm(vnfm_type=vnfm_type, vim_id=vimid)
  87                 params_vnf.append({
  88                     "vnfProfileId": vnf["vnf_id"],
  89                     "additionalParam": {
  90                         "properties": json.JSONEncoder().encode(vnf['properties']),
  91                         "vimId": vimid,
  92                         "vnfmInstanceId": vnfm_info["vnfmId"],
  93                         "vnfmType": vnfm_type,
  94                         "inputs": params_json
  95                     }
  96                 })
  97             # end
  98
  99             self.set_vl_vim_id(vim_id, location_constraints, plan_dict)
 100             dst_plan = json.JSONEncoder().encode(plan_dict)
 101             logger.debug('tosca plan dest add vimid:%s' % dst_plan)
 102             NSInstModel.objects.filter(id=self.ns_inst_id).update(nsd_model=dst_plan)
 103
 104             vnf_params_json = json.JSONEncoder().encode(params_vnf)
 105             plan_input = {
 106                 'jobId': job_id,
 107                 'nsInstanceId': self.ns_inst_id,
 108                 'object_context': dst_plan,
 109                 'object_additionalParamForNs': params_json,
 110                 'object_additionalParamForVnf': vnf_params_json
 111             }
 112             plan_input.update(**self.get_model_count(dst_plan))
 113             plan_input["sdnControllerId"] = ignore_case_get(
 114                 self.req_data['additionalParamForNs'], "sdncontroller")
 115
 116             ServiceBaseInfoModel(service_id=self.ns_inst_id,
 117                                  service_name=ns_inst.name,
 118                                  service_type='NFVO',
 119                                  description=ns_inst.description,
 120                                  active_status='--',
 121                                  status=ns_inst.status,
 122                                  creator='--',
 123                                  create_time=int(time.time() * 1000)).save()
 124
 125             if WORKFLOW_OPTION == "wso2":
 126                 service_tpl = get_servicetemplate(ns_inst.nsd_id)
 127                 DefPkgMappingModel(service_id=self.ns_inst_id,
 128                                    service_def_id=service_tpl['csarId'],
 129                                    template_name=service_tpl['templateName'],
 130                                    template_id=service_tpl['serviceTemplateId']).save()
 131
 132                 for key, val in self.req_data['additionalParamForNs'].items():
 133                     InputParamMappingModel(service_id=self.ns_inst_id,
 134                                            input_key=key,
 135                                            input_value=val).save()
 136
 137                 for vnffg in ignore_case_get(plan_dict, "vnffgs"):
 138                     VNFFGInstModel(vnffgdid=vnffg["vnffg_id"],
 139                                    vnffginstid=str(uuid.uuid4()),
 140                                    nsinstid=self.ns_inst_id,
 141                                    endpointnumber=0).save()
 142             else:
 143                 # TODO:
 144                 pass
 145
 146             if WORKFLOW_OPTION == "wso2":
 147                 return self.start_wso2_workflow(job_id, ns_inst, plan_input)
 148             elif WORKFLOW_OPTION == "activiti":
 149                 return self.start_activiti_workflow()
 150             else:
 151                 return self.start_buildin_workflow(job_id, plan_input)
 152
 153         except Exception as e:
 154             logger.error(traceback.format_exc())
 155             logger.error("ns-instant(%s) workflow error:%s" % (self.ns_inst_id, e.message))
 156             JobUtil.add_job_status(job_id, 255, 'NS instantiation failed: %s' % e.message)
 157             return dict(data={'error': e.message}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
 158
 159     def start_wso2_workflow(self, job_id, ns_inst, plan_input):
 160         servicetemplate_id = get_servicetemplate_id(ns_inst.nsd_id)
 161         process_id = get_process_id('init', servicetemplate_id)
 162         data = {"processId": process_id, "params": {"planInput": plan_input}}
 163         logger.debug('ns-instant(%s) workflow data:%s' % (self.ns_inst_id, data))
 164
 165         ret = workflow_run(data)
 166         logger.info("ns-instant(%s) workflow result:%s" % (self.ns_inst_id, ret))
 167         JobUtil.add_job_status(job_id, 10, 'NS inst(%s) wso2 workflow started: %s' % (
 168             self.ns_inst_id, ret.get('status')))
 169         if ret.get('status') == 1:
 170             return dict(data={'jobId': job_id}, status=status.HTTP_200_OK)
 171         return dict(data={'error': ret['message']}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
 172
 173     def start_activiti_workflow(self, job_id, plan_input):
 174         plans = WFPlanModel.objects.filter()
 175         if not plans:
 176             raise NSLCMException("No plan is found, you should deploy plan first!")
 177         data = {
 178             "processId": plans[0].process_id,
 179             "params": plan_input
 180         }
 181         ret = activiti.exec_workflow(data)
 182         logger.info("ns-instant(%s) workflow result:%s" % (self.ns_inst_id, ret))
 183         JobUtil.add_job_status(job_id, 10, 'NS inst(%s) activiti workflow started: %s' % (
 184             self.ns_inst_id, ret.get('status')))
 185         if ret.get('status') == 1:
 186             return dict(data={'jobId': job_id}, status=status.HTTP_200_OK)
 187         return dict(data={'error': ret['message']}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
 188
 189     def start_buildin_workflow(self, job_id, plan_input):
 190         JobUtil.add_job_status(job_id, 10, 'NS inst(%s) buildin workflow started.' % self.ns_inst_id)
 191         BuildInWorkflowThread(plan_input).start()
 192         return dict(data={'jobId': job_id}, status=status.HTTP_200_OK)
 193
 194     @staticmethod
 195     def get_vnf_vim_id(vim_id, location_constraints, vnfdid):
 196         for location in location_constraints:
 197             if "vnfProfileId" in location and vnfdid == location["vnfProfileId"]:
 198                 return location["locationConstraints"]["vimId"]
 199         if vim_id:
 200             return vim_id
 201         raise NSLCMException("No Vim info is found for vnf(%s)." % vnfdid)
 202
 203     @staticmethod
 204     def set_vl_vim_id(vim_id, location_constraints, plan_dict):
 205         if "ns_vls" not in plan_dict:
 206             logger.debug("No vl is found in nsd.")
 207             return
 208         vl_vnf = {}
 209         for vnf in ignore_case_get(plan_dict, "ns_vnfs"):
 210             if "dependencies" in vnf:
 211                 for depend in vnf["dependencies"]:
 212                     vl_vnf[depend["vl_id"]] = vnf['properties']['id']
 213         vnf_vim = {}
 214         for location in location_constraints:
 215             if "vnfProfileId" in location:
 216                 vnfd_id = location["vnfProfileId"]
 217                 vnf_vim[vnfd_id] = location["locationConstraints"]["vimId"]
 218         for vl in plan_dict["ns_vls"]:
 219             vnfdid = ignore_case_get(vl_vnf, vl["vl_id"])
 220             vimid = ignore_case_get(vnf_vim, vnfdid)
 221             if not vimid:
 222                 vimid = vim_id
 223             if not vimid:
 224                 raise NSLCMException("No Vim info for vl(%s) of vnf(%s)." % (vl["vl_id"], vnfdid))
 225             if "location_info" not in vl["properties"]:
 226                 vl["properties"]["location_info"] = {}
 227             vl["properties"]["location_info"]["vimid"] = vimid
 228
 229     @staticmethod
 230     def get_model_count(context):
 231         data = json.JSONDecoder().decode(context)
 232         vls = len(data.get('ns_vls', []))
 233         sfcs = len(data.get('fps', []))
 234         vnfs = len(data.get('ns_vnfs', []))
 235         return {'vlCount': str(vls), 'sfcCount': str(sfcs), 'vnfCount': str(vnfs)}