lcm/ns/biz/ns_instantiate_flow.py

   1 # Copyright 2018 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
  15 import json
  16 import logging
  17 import traceback
  18 from threading import Thread
  19
  20 from lcm.pub.utils.syscomm import fun_name
  21 from lcm.pub.utils.values import ignore_case_get
  22 from lcm.pub.utils import restcall
  23 from lcm.pub.exceptions import NSLCMException
  24 from lcm.workflows.graphflow.flow.flow import GraphFlow
  25 from lcm.ns.biz.ns_lcm_op_occ import NsLcmOpOcc
  26
  27 logger = logging.getLogger(__name__)
  28
  29 RESULT_OK, RESULT_NG = "0", "1"
  30 JOB_ERROR = 255
  31
  32 config = {
  33     "CreateVnf": {"module": "lcm.ns_vnfs", "class": "CreateVnf"},
  34     "CreatePnf": {"module": "lcm.ns_pnfs", "class": "CreatePnf"},
  35     "CreateVl": {"module": "lcm.ns_vls", "class": "CreateVl"}
  36 }
  37
  38
  39 class NsInstantiateWorkflowThread(Thread):
  40     def __init__(self, plan_input, occ_id):
  41         Thread.__init__(self)
  42         self.plan_input = plan_input
  43         self.occ_id = occ_id
  44
  45     def run(self):
  46         run_ns_instantiate(self.plan_input, self.occ_id)
  47
  48
  49 def run_ns_instantiate(input_data, occ_id):
  50     """
  51     format of input_data
  52     {
  53         "jobId": uuid of job,
  54         "nsInstanceId": id of ns instance,
  55         "object_context": json format of nsd,
  56         "object_additionalParamForNs": json format of additional parameters for ns,
  57         "object_additionalParamForVnf": json format of additional parameters for vnf,
  58         "object_additionalParamForPnf": json format of additional parameters for pnf,
  59         "vlCount": int type of VL count,
  60         "vnfCount: int type of VNF count
  61     }
  62     """
  63     logger.debug("Enter %s, input_data is %s", fun_name(), input_data)
  64     ns_inst_id = ignore_case_get(input_data, "nsInstanceId")
  65     job_id = ignore_case_get(input_data, "jobId")
  66     update_job(job_id, 10, "true", "Start to prepare the NS instantiate workflow parameter")
  67     deploy_graph = build_deploy_graph(input_data)
  68     TaskSet = build_TaskSet(input_data)
  69     ns_instantiate_ok = False
  70
  71     try:
  72         update_job(job_id, 15, "true", "Start the NS instantiate workflow")
  73         gf = GraphFlow(deploy_graph, TaskSet, config)
  74         logger.debug("NS graph flow run up!")
  75         gf.start()
  76         gf.join()
  77         gf.task_manager.wait_tasks_done(gf.sort_nodes)
  78         if gf.task_manager.is_all_task_finished():
  79             logger.debug("NS is instantiated!")
  80             update_job(job_id, 90, "true", "Start to post deal")
  81             post_deal(ns_inst_id, "true")
  82             update_job(job_id, 100, "true", "Create NS successfully.")
  83             NsLcmOpOcc.update(occ_id, "COMPLETED")
  84             ns_instantiate_ok = True
  85     except NSLCMException as e:
  86         logger.error("Failded to Create NS: %s", e.message)
  87         update_job(job_id, JOB_ERROR, "255", "Failded to Create NS.")
  88         NsLcmOpOcc.update(occ_id, operationState="FAILED", error=e.message)
  89         post_deal(ns_inst_id, "false")
  90     except Exception as e:
  91         logger.error(traceback.format_exc())
  92         update_job(job_id, JOB_ERROR, "255", "Failded to Create NS.")
  93         NsLcmOpOcc.update(occ_id, operationState="FAILED", error=e.message)
  94         post_deal(ns_inst_id, "false")
  95     return ns_instantiate_ok
  96
  97
  98 def build_deploy_graph(input_data):
  99     nsd_json_str = ignore_case_get(input_data, "object_context")
 100     nsd_json = json.JSONDecoder().decode(nsd_json_str)
 101     deploy_graph = ignore_case_get(nsd_json, "graph")
 102     logger.debug("NS graph flow: %s" % deploy_graph)
 103     return deploy_graph
 104
 105
 106 def build_vls(input_data):
 107     ns_inst_id = ignore_case_get(input_data, "nsInstanceId")
 108     nsd_json = json.JSONDecoder().decode(ignore_case_get(input_data, "object_context"))
 109     ns_param_json = ignore_case_get(input_data, "object_additionalParamForNs")
 110     vl_count = int(ignore_case_get(input_data, "vlCount", 0))
 111
 112     vls = {}
 113     for i in range(vl_count):
 114         data = {
 115             "nsInstanceId": ns_inst_id,
 116             "vlIndex": i,
 117             "context": nsd_json,
 118             "additionalParamForNs": ns_param_json
 119         }
 120         key = nsd_json["vls"][i - 1]["vl_id"]
 121         vls[key] = {
 122             "type": "CreateVl",
 123             "input": {
 124                     "content": data
 125             }
 126         }
 127     return vls
 128
 129
 130 def build_vnfs(input_data):
 131     ns_inst_id = ignore_case_get(input_data, "nsInstanceId")
 132     vnf_count = int(ignore_case_get(input_data, "vnfCount", 0))
 133     vnf_param_json = json.JSONDecoder().decode(ignore_case_get(input_data, "object_additionalParamForVnf"))
 134     vnfs = {}
 135     for i in range(vnf_count):
 136         data = {
 137             "nsInstanceId": ns_inst_id,
 138             "vnfIndex": i,
 139             "additionalParamForVnf": vnf_param_json
 140         }
 141         key = vnf_param_json[i - 1]["vnfProfileId"]
 142         vnfs[key] = {
 143             "type": "CreateVnf",
 144             "input": {
 145                     "content": data
 146             }
 147         }
 148     return vnfs
 149
 150
 151 def build_pnfs(input_data):
 152     return json.JSONDecoder().decode(ignore_case_get(input_data, "object_additionalParamForPnf"))
 153
 154
 155 def build_TaskSet(input_data):
 156     vls = build_vls(input_data)
 157     vnfs = build_vnfs(input_data)
 158     pnfs = build_pnfs(input_data)
 159     task_set = dict(dict(vls, **vnfs), **pnfs)
 160     return task_set
 161
 162
 163 def post_deal(ns_inst_id, status):
 164     uri = "api/nslcm/v1/ns/{nsInstanceId}/postdeal".format(nsInstanceId=ns_inst_id)
 165     data = json.JSONEncoder().encode({
 166         "status": status
 167     })
 168
 169     ret = restcall.req_by_msb(uri, "POST", data)
 170     if ret[0] != 0:
 171         logger.error("Failed to call post_deal(%s): %s", ns_inst_id, ret[1])
 172     logger.debug("Call post_deal(%s, %s) successfully.", ns_inst_id, status)
 173
 174
 175 def update_job(job_id, progress, errcode, desc):
 176     logger.debug("job_id %s" % job_id)
 177     uri = "api/nslcm/v1/jobs/{jobId}".format(jobId=job_id)
 178     data = json.JSONEncoder().encode({
 179         "progress": progress,
 180         "errcode": errcode,
 181         "desc": desc
 182     })
 183     ret = restcall.req_by_msb(uri, "POST", data)
 184     return ret