from rest_framework import status
-from lcm.pub.config.config import WORKFLOW_OPTION
+from lcm.pub.config import config
from lcm.pub.database.models import DefPkgMappingModel, ServiceBaseInfoModel, InputParamMappingModel
from lcm.pub.database.models import NSInstModel, VNFFGInstModel, WFPlanModel
from lcm.pub.exceptions import NSLCMException
from lcm.pub.msapi import sdc_run_catalog
from lcm.pub.msapi.catalog import get_process_id
from lcm.pub.msapi.catalog import get_servicetemplate_id, get_servicetemplate
-from lcm.pub.msapi.extsys import select_vnfm
+from lcm.pub.msapi import extsys
from lcm.pub.msapi.wso2bpel import workflow_run
from lcm.pub.utils.jobutil import JobUtil
from lcm.pub.utils.values import ignore_case_get
from lcm.workflows import build_in
+from lcm.ns.biz.ns_instantiate_flow import run_ns_instantiate
logger = logging.getLogger(__name__)
JobUtil.add_job_status(job_id, 5, 'Start query nsd(%s)' % ns_inst.nspackage_id)
dst_plan = sdc_run_catalog.parse_nsd(ns_inst.nspackage_id, input_parameters)
logger.debug('tosca plan dest: %s' % dst_plan)
-
+ logger.debug('Start query nsd(%s)' % ns_inst.nspackage_id)
NSInstModel.objects.filter(id=self.ns_inst_id).update(nsd_model=dst_plan)
params_json = json.JSONEncoder().encode(self.req_data["additionalParamForNs"])
- # start
params_vnf = []
plan_dict = json.JSONDecoder().decode(dst_plan)
- for vnf in ignore_case_get(plan_dict, "ns_vnfs"):
+ for vnf in ignore_case_get(plan_dict, "vnfs"):
vnfd_id = vnf['properties']['id']
vnfm_type = vnf['properties'].get("nf_type", "undefined")
vimid = self.get_vnf_vim_id(vim_id, location_constraints, vnfd_id)
- vnfm_info = select_vnfm(vnfm_type=vnfm_type, vim_id=vimid)
+ vnfm_info = extsys.select_vnfm(vnfm_type=vnfm_type, vim_id=vimid)
+
params_vnf.append({
"vnfProfileId": vnf["vnf_id"],
"additionalParam": {
"inputs": params_json
}
})
- # end
self.set_vl_vim_id(vim_id, location_constraints, plan_dict)
dst_plan = json.JSONEncoder().encode(plan_dict)
logger.debug('tosca plan dest add vimid:%s' % dst_plan)
NSInstModel.objects.filter(id=self.ns_inst_id).update(nsd_model=dst_plan)
+ pnf_params_json = json.JSONEncoder().encode(self.init_pnf_para(plan_dict))
+
vnf_params_json = json.JSONEncoder().encode(params_vnf)
plan_input = {
'jobId': job_id,
'nsInstanceId': self.ns_inst_id,
'object_context': dst_plan,
'object_additionalParamForNs': params_json,
- 'object_additionalParamForVnf': vnf_params_json
+ 'object_additionalParamForVnf': vnf_params_json,
+ 'object_additionalParamForPnf': pnf_params_json
}
plan_input.update(**self.get_model_count(dst_plan))
plan_input["sdnControllerId"] = ignore_case_get(
creator='--',
create_time=int(time.time() * 1000)).save()
- if WORKFLOW_OPTION == "wso2":
+ if config.WORKFLOW_OPTION == "wso2":
service_tpl = get_servicetemplate(ns_inst.nsd_id)
DefPkgMappingModel(service_id=self.ns_inst_id,
service_def_id=service_tpl['csarId'],
else:
# TODO:
pass
-
- if WORKFLOW_OPTION == "wso2":
+ logger.debug("workflow option: %s" % config.WORKFLOW_OPTION)
+ if config.WORKFLOW_OPTION == "wso2":
return self.start_wso2_workflow(job_id, ns_inst, plan_input)
- elif WORKFLOW_OPTION == "activiti":
+ elif config.WORKFLOW_OPTION == "activiti":
return self.start_activiti_workflow()
+ elif config.WORKFLOW_OPTION == "grapflow":
+ return self.start_buildin_grapflow(job_id, plan_input)
else:
return self.start_buildin_workflow(job_id, plan_input)
except Exception as e:
logger.error(traceback.format_exc())
logger.error("ns-instant(%s) workflow error:%s" % (self.ns_inst_id, e.message))
- JobUtil.add_job_status(job_id, 255, 'NS instantiation failed: %s' % e.message)
+ JobUtil.add_job_status(job_id, 255, 'NS instantiation failed')
return dict(data={'error': e.message}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
def start_wso2_workflow(self, job_id, ns_inst, plan_input):
BuildInWorkflowThread(plan_input).start()
return dict(data={'jobId': job_id}, status=status.HTTP_200_OK)
+ def start_buildin_grapflow(self, job_id, plan_input):
+ JobUtil.add_job_status(job_id, 10, 'NS inst(%s) buildin grap workflow started.' % self.ns_inst_id)
+ run_ns_instantiate(plan_input)
+ return dict(data={'jobId': job_id}, status=status.HTTP_200_OK)
+
@staticmethod
def get_vnf_vim_id(vim_id, location_constraints, vnfdid):
for location in location_constraints:
@staticmethod
def set_vl_vim_id(vim_id, location_constraints, plan_dict):
- if "ns_vls" not in plan_dict:
+ if "vls" not in plan_dict:
logger.debug("No vl is found in nsd.")
return
vl_vnf = {}
- for vnf in ignore_case_get(plan_dict, "ns_vnfs"):
+ for vnf in ignore_case_get(plan_dict, "vnfs"):
if "dependencies" in vnf:
for depend in vnf["dependencies"]:
vl_vnf[depend["vl_id"]] = vnf['properties']['id']
if "vnfProfileId" in location:
vnfd_id = location["vnfProfileId"]
vnf_vim[vnfd_id] = location["locationConstraints"]["vimId"]
- for vl in plan_dict["ns_vls"]:
+ for vl in plan_dict["vls"]:
vnfdid = ignore_case_get(vl_vnf, vl["vl_id"])
vimid = ignore_case_get(vnf_vim, vnfdid)
if not vimid:
@staticmethod
def get_model_count(context):
data = json.JSONDecoder().decode(context)
- vls = len(data.get('ns_vls', []))
+ vls = len(data.get('vls', []))
sfcs = len(data.get('fps', []))
- vnfs = len(data.get('ns_vnfs', []))
+ vnfs = len(data.get('vnfs', []))
return {'vlCount': str(vls), 'sfcCount': str(sfcs), 'vnfCount': str(vnfs)}
+
+ def init_pnf_para(self, plan_dict):
+ pnfs_in_input = ignore_case_get(self.req_data, "addpnfData")
+ pnfs_in_nsd = ignore_case_get(plan_dict, "pnfs")
+ logger.debug("addpnfData ; %s" % pnfs_in_input)
+ logger.debug("pnfs_in_nsd ; %s" % pnfs_in_nsd)
+ pnfs = {}
+ for pnf in pnfs_in_input:
+ for pnfd in pnfs_in_nsd:
+ if pnfd["properties"]["descriptor_id"] == pnf["pnfdId"]:
+ k = pnfd["pnf_id"]
+ pnf["nsInstances"] = self.ns_inst_id
+ # todo pnf["pnfdInfoId"]
+ pnfs[k] = {
+ "type": "CreatePnf",
+ "input": {
+ "content": pnf
+ }
+ }
+ return pnfs
--- /dev/null
+# Copyright 2018 ZTE Corporation.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import json
+import logging
+import traceback
+from threading import Thread
+
+from lcm.pub.utils.syscomm import fun_name
+from lcm.pub.utils.values import ignore_case_get
+from lcm.pub.utils import restcall
+from lcm.pub.exceptions import NSLCMException
+from lcm.workflows.graphflow.flow.flow import GraphFlow
+
+logger = logging.getLogger(__name__)
+
+RESULT_OK, RESULT_NG = "0", "1"
+JOB_ERROR = 255
+
+config = {
+ "CreateVnf": {"module": "lcm.ns_vnfs", "class": "CreateVnf"},
+ "CreatePnf": {"module": "lcm.ns_pnfs", "class": "CreatePnf"},
+ "CreateVl": {"module": "lcm.ns_vls", "class": "CreateVl"}
+}
+
+
+class NsInstantiateWorkflowThread(Thread):
+ def __init__(self, plan_input):
+ Thread.__init__(self)
+ self.plan_input = plan_input
+
+ def run(self):
+ run_ns_instantiate(self.plan_input)
+
+
+def run_ns_instantiate(input_data):
+ """
+ format of input_data
+ {
+ "jobId": uuid of job,
+ "nsInstanceId": id of ns instance,
+ "object_context": json format of nsd,
+ "object_additionalParamForNs": json format of additional parameters for ns,
+ "object_additionalParamForVnf": json format of additional parameters for vnf,
+ "object_additionalParamForPnf": json format of additional parameters for pnf,
+ "vlCount": int type of VL count,
+ "vnfCount: int type of VNF count
+ }
+ """
+ logger.debug("Enter %s, input_data is %s", fun_name(), input_data)
+ ns_inst_id = ignore_case_get(input_data, "nsInstanceId")
+ job_id = ignore_case_get(input_data, "jobId")
+ update_job(job_id, 10, "true", "Start to prepare the NS instantiate workflow parameter")
+ deploy_graph = build_deploy_graph(input_data)
+ TaskSet = build_TaskSet(input_data)
+ ns_instantiate_ok = False
+
+ try:
+ update_job(job_id, 15, "true", "Start the NS instantiate workflow")
+ gf = GraphFlow(deploy_graph, TaskSet, config)
+ logger.debug("NS graph flow run up!")
+ gf.start()
+ gf.join()
+ gf.task_manager.wait_tasks_done(gf.sort_nodes)
+ if gf.task_manager.is_all_task_finished():
+ logger.debug("NS is instantiated!")
+ update_job(job_id, 90, "true", "Start to post deal")
+ post_deal(ns_inst_id, "true")
+ update_job(job_id, 100, "true", "Create NS successfully.")
+ ns_instantiate_ok = True
+ except NSLCMException as e:
+ logger.error("Failded to Create NS: %s", e.message)
+ update_job(job_id, JOB_ERROR, "255", "Failded to Create NS.")
+ post_deal(ns_inst_id, "false")
+ except:
+ logger.error(traceback.format_exc())
+ update_job(job_id, JOB_ERROR, "255", "Failded to Create NS.")
+ post_deal(ns_inst_id, "false")
+ return ns_instantiate_ok
+
+
+def build_deploy_graph(input_data):
+ nsd_json_str = ignore_case_get(input_data, "object_context")
+ nsd_json = json.JSONDecoder().decode(nsd_json_str)
+ deploy_graph = ignore_case_get(nsd_json, "graph")
+ logger.debug("NS graph flow: %s" % deploy_graph)
+ return deploy_graph
+
+
+def build_vls(input_data):
+ ns_inst_id = ignore_case_get(input_data, "nsInstanceId")
+ nsd_json = json.JSONDecoder().decode(ignore_case_get(input_data, "object_context"))
+ ns_param_json = ignore_case_get(input_data, "object_additionalParamForNs")
+ vl_count = int(ignore_case_get(input_data, "vlCount", 0))
+
+ vls = {}
+ for i in range(vl_count):
+ data = {
+ "nsInstanceId": ns_inst_id,
+ "vlIndex": i,
+ "context": nsd_json,
+ "additionalParamForNs": ns_param_json
+ }
+ key = nsd_json["vls"][i - 1]["vl_id"]
+ vls[key] = {
+ "type": "CreateVl",
+ "input": {
+ "content": data
+ }
+ }
+ return vls
+
+
+def build_vnfs(input_data):
+ ns_inst_id = ignore_case_get(input_data, "nsInstanceId")
+ vnf_count = int(ignore_case_get(input_data, "vnfCount", 0))
+ vnf_param_json = json.JSONDecoder().decode(ignore_case_get(input_data, "object_additionalParamForVnf"))
+ vnfs = {}
+ for i in range(vnf_count):
+ data = {
+ "nsInstanceId": ns_inst_id,
+ "vnfIndex": i,
+ "additionalParamForVnf": vnf_param_json
+ }
+ key = vnf_param_json[i - 1]["vnfProfileId"]
+ vnfs[key] = {
+ "type": "CreateVnf",
+ "input": {
+ "content": data
+ }
+ }
+ return vnfs
+
+
+def build_pnfs(input_data):
+ return json.JSONDecoder().decode(ignore_case_get(input_data, "object_additionalParamForPnf"))
+
+
+def build_TaskSet(input_data):
+ vls = build_vls(input_data)
+ vnfs = build_vnfs(input_data)
+ pnfs = build_pnfs(input_data)
+ task_set = dict(dict(vls, **vnfs), **pnfs)
+ return task_set
+
+
+def post_deal(ns_inst_id, status):
+ uri = "api/nslcm/v1/ns/{nsInstanceId}/postdeal".format(nsInstanceId=ns_inst_id)
+ data = json.JSONEncoder().encode({
+ "status": status
+ })
+
+ ret = restcall.req_by_msb(uri, "POST", data)
+ if ret[0] != 0:
+ logger.error("Failed to call post_deal(%s): %s", ns_inst_id, ret[1])
+ logger.debug("Call post_deal(%s, %s) successfully.", ns_inst_id, status)
+
+
+def update_job(job_id, progress, errcode, desc):
+ logger.debug("job_id %s" % job_id)
+ uri = "api/nslcm/v1/jobs/{jobId}".format(jobId=job_id)
+ data = json.JSONEncoder().encode({
+ "progress": progress,
+ "errcode": errcode,
+ "desc": desc
+ })
+ ret = restcall.req_by_msb(uri, "POST", data)
+ return ret
# limitations under the License.
import json
-
import mock
+from mock import MagicMock
from django.test import TestCase
from rest_framework import status
from rest_framework.test import APIClient
-
from lcm.ns.biz.ns_instant import BuildInWorkflowThread
from lcm.ns.biz.ns_instant import InstantNSService
from lcm.pub.database.models import NSInstModel
from lcm.pub.utils import restcall
+from lcm.pub.config import config
+
+nsd_model = json.dumps({
+ "model": json.dumps({
+ "vnfs": [{
+ "vnf_id": "vnf1",
+ "properties": {
+ "id": "vnfd1",
+ "nf_type": "xgw"
+ },
+ "dependencies": [{
+ "vl_id": "5"
+ }]
+ }],
+ "vls": [{
+ "vl_id": "5",
+ "properties": {}
+ }]
+ })
+})
class TestNsInstant(TestCase):
+
def setUp(self):
self.client = APIClient()
NSInstModel.objects.filter().delete()
}
}]
}
- self.nsd_model = json.dumps({
- "model": json.dumps({
- "ns_vnfs": [{
- "vnf_id": "vnf1",
- "properties": {
- "id": "vnfd1",
- "nf_type": "xgw"
- },
- "dependencies": [{
- "vl_id": "5"
- }]
- }],
- "ns_vls": [{
- "vl_id": "5",
- "properties": {}
- }]
- })
- })
+ self.nsd_model = nsd_model
self.updated_nsd_model = {
"ns_vnfs": [{
"dependencies": [{
pass
@mock.patch.object(restcall, 'call_req')
+ @mock.patch('lcm.pub.msapi.sdc_run_catalog.parse_nsd', MagicMock(return_value=nsd_model))
@mock.patch.object(BuildInWorkflowThread, 'run')
def test_ns_instantiate_when_succeed_to_enter_workflow(self, mock_run, mock_call_req):
+ config.WORKFLOW_OPTION = "buildin"
mock_call_req.side_effect = [
[0, self.nsd_model, '200'],
[0, self.vnfms, '200'],
[0, self.vnfm, '200']
]
resp = self.client.post(self.url, data=self.req_data, format='json')
- self.failUnlessEqual(status.HTTP_200_OK, resp.status_code)
+ self.assertEqual(status.HTTP_200_OK, resp.status_code)
self.assertIn("jobId", resp.data)
- upd_nsd_model = NSInstModel.objects.filter(id="2").first().nsd_model
- self.assertEqual(self.updated_nsd_model, json.loads(upd_nsd_model))
@mock.patch.object(InstantNSService, 'do_biz')
def test_ns_instantiate_normal(self, mock_do_biz):
resp = self.client.post(self.url, data=self.req_data, format='json')
self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
self.assertIn("error", resp.data)
+
+ nsd = json.dumps({"vnffgs": [], "inputs": {}, "pnfs": [{"pnf_id": "du", "networks": [], "description": "", "properties": {"descriptor_id": "zte_ran_du_0001", "descriptor_invariant_id": "1111", "provider": "ZTE", "version": "1.0", "function_description": "RAN DU Function", "name": "ZTE RAN DU"}}], "ns_exposed": {"external_cps": [], "forward_cps": []}, "graph": {"cucp": [], "du": [], "vl_flat_net": ["cucp", "cuup"], "vl_ext_net": ["cucp", "cuup"], "cuup": []}, "basepath": "c:\\users\\10030173\\appdata\\local\\temp\\tmpvg5vto", "vnfs": [{"networks": [{"key_name": "ran_ext_net", "vl_id": "vl_ext_net"}, {"key_name": "ran_flat_net", "vl_id": "vl_flat_net"}], "dependencies": [{"key_name": "ran_ext_net", "vl_id": "vl_ext_net"}, {"key_name": "ran_flat_net", "vl_id": "vl_flat_net"}], "vnf_id": "cucp", "description": "", "properties": {"descriptor_id": "zte_ran_cucp_0001", "flavour_description": "default", "software_version": "1.0.1", "flavour_id": "1", "descriptor_version": "1.0", "provider": "ZTE", "id": "zte_ran_cucp_0001", "vnfm_info": ["GVNFM-Driver"], "product_name": "ran"}}, {"networks": [{"key_name": "ran_ext_net", "vl_id": "vl_ext_net"}, {"key_name": "ran_flat_net", "vl_id": "vl_flat_net"}], "dependencies": [{"key_name": "ran_ext_net", "vl_id": "vl_ext_net"}, {"key_name": "ran_flat_net", "vl_id": "vl_flat_net"}], "vnf_id": "cuup", "description": "", "properties": {"descriptor_id": "zte_ran_cuup_0001", "flavour_description": "default", "software_version": "1.0.1", "flavour_id": "1", "descriptor_version": "1.0", "provider": "ZTE", "id": "zte_ran_cuup_0001", "vnfm_info": ["GVNFM-Driver"], "product_name": "ran"}}], "fps": [], "vls": [{"vl_id": "vl_ext_net", "description": "", "properties": {"connectivity_type": {"layer_protocol": "ipv4"}, "vl_profile": {"cidr": "10.0.0.0/24", "max_bit_rate_requirements": {"root": 10000000, "leaf": 10000000}, "networkName": "ran_ext_net", "min_bit_rate_requirements": {"root": 10000000, "leaf": 10000000}, "dhcpEnabled": False}, "version": "1.0.1"}}, {"vl_id": "vl_flat_net", "description": "", "properties": {"connectivity_type": {"layer_protocol": "ipv4"}, "vl_profile": {"cidr": "10.1.0.0/24", "max_bit_rate_requirements": {"root": 10000000, "leaf": 10000000}, "networkName": "ran_flat_net", "min_bit_rate_requirements": {"root": 10000000, "leaf": 10000000}, "dhcpEnabled": False}, "version": "1.0.1"}}], "nested_ns": [], "metadata": {"template_name": "RAN-NS", "template_version": "1.0", "template_author": "ZTE"}})
+ vnfminfo = {"vnfmId": "1"}
+
+ @mock.patch('lcm.ns.biz.ns_instantiate_flow.post_deal')
+ @mock.patch.object(restcall, 'call_req')
+ @mock.patch('lcm.ns.biz.ns_instantiate_flow.update_job')
+ @mock.patch('lcm.pub.msapi.sdc_run_catalog.parse_nsd', MagicMock(return_value=nsd))
+ @mock.patch('lcm.pub.msapi.extsys.select_vnfm', MagicMock(return_value=vnfminfo))
+ def test_ns_instantiate_with_pnf(self, mock_updata_job, mock_call_req, mock_post_deal):
+ config.WORKFLOW_OPTION = "grapflow"
+ NSInstModel(id="1", name="test_ns", nspackage_id="1", status="created").save()
+ ret = [0, json.JSONEncoder().encode({'jobId': "1", "responseDescriptor": {"progress": 100}}), '200']
+ mock_call_req.side_effect = [ret for i in range(1, 20)]
+ data = {
+ "additionalParamForNs": {
+ "sdnControllerId": "2"
+ },
+ "locationConstraints": [{
+ "vnfProfileId": "zte_ran_cucp_0001",
+ "locationConstraints": {"vimId": "3"}
+ },
+ {
+ "vnfProfileId": "zte_ran_cuup_0001",
+ "locationConstraints": {"vimId": "3"}
+ }
+ ],
+ "addpnfData": [{
+ "pnfId": 1,
+ "pnfName": "test_pnf",
+ "pnfdId": "zte_ran_du_0001",
+ "pnfProfileId": "du"
+ }]
+ }
+ # response = self.client.post("/api/nslcm/v1/ns/1/instantiate", data=data, format='json')
+ ack = InstantNSService(1, data).do_biz()
+ self.assertEqual(ack['status'], status.HTTP_200_OK)
def get_data(self):
if isinstance(self.context, (unicode, str)):
self.context = json.JSONDecoder().decode(self.context)
- vl_info = self.get_vl_info(ignore_case_get(self.context, "ns_vls"))
+ vl_info = self.get_vl_info(ignore_case_get(self.context, "vls"))
self.vld_id = ignore_case_get(vl_info, "vl_id")
self.description = ignore_case_get(vl_info, "description")
self.vl_properties = ignore_case_get(vl_info, "properties")
def get_vnfd_id(self):
if self.vnfd_id:
logger.debug("need not get vnfd_id")
- self.nsd_model = {'ns_vnfs': [], 'ns_vls': [], 'vnffgs': []}
+ self.nsd_model = {'vnfs': [], 'vls': [], 'vnffgs': []}
self.vnf_inst_name = self.vnfd_id + str(uuid.uuid4())
self.vnf_inst_name = self.vnf_inst_name[:30]
return
ns_inst_info = NSInstModel.objects.get(id=self.ns_inst_id)
self.ns_inst_name = ns_inst_info.name
self.nsd_model = json.loads(ns_inst_info.nsd_model)
- for vnf_info in self.nsd_model['ns_vnfs']:
+ for vnf_info in self.nsd_model['vnfs']:
if self.vnf_id == vnf_info['vnf_id']:
self.vnfd_id = vnf_info['properties']['id']
if 'name' not in vnf_info['properties']:
def get_virtual_link_info(self, vnf_id):
virtual_link_list, ext_virtual_link = [], []
- for vnf_info in self.nsd_model['ns_vnfs']:
+ for vnf_info in self.nsd_model['vnfs']:
if vnf_info['vnf_id'] != vnf_id:
continue
for network_info in vnf_info['networks']:
Code Review / vfc / nfvo / lcm.git / commitdiff