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