VnfrYang,
RwVnfrYang,
RwNsmYang,
- RwsdnYang,
+ RwsdnalYang,
RwDts as rwdts,
RwTypes,
ProtobufC,
vnfr = yield from self._nsr.fetch_vnfr(nsr_vnfr.xpath)
self._log.debug("Received VNFR is %s", vnfr)
- sff = RwsdnYang.VNFFGSff()
+ sff = RwsdnalYang.VNFFGSff()
sff_list[nsr_vnfr.vnfd.id] = sff
sff.name = nsr_vnfr.name
sff.function_type = nsr_vnfr.vnfd.service_function_chain
(member_vnfd.member_vnf_index_ref == const_vnfd.member_vnf_index):
group_info = self.resolve_placement_group_cloud_construct(group)
if group_info is None:
- self._log.error("Could not resolve cloud-construct for placement group: %s", group.name)
+ self._log.info("Could not resolve cloud-construct for placement group: %s", group.name)
### raise PlacementGroupError("Could not resolve cloud-construct for placement group: {}".format(group.name))
else:
self._log.info("Successfully resolved cloud construct for placement group: %s for VNF: %s (Member Index: %s)",
self._cloud_account_handler = cloud_account_handler
self._ro_plugin_selector = ro_plugin_selector
- self._ncclient = rift.mano.ncclient.NcClient(
- host="127.0.0.1",
- port=2022,
- username="admin",
- password="admin",
- loop=self._loop)
+ # Intialize the set of variables for implementing Scaling RPC using REST.
+ self._headers = {"content-type":"application/json", "accept":"application/json"}
+ #This will break when we have rbac in the rift code and admin user password is changed or admin it self is removed.
+ self._user = 'admin'
+ self._password = 'admin'
+ self._ip = 'localhost'
+ self._rport = 8008
+ self._conf_url = "https://{ip}:{port}/api/config". \
+ format(ip=self._ip,
+ port=self._rport)
+
self._nsrs = {}
self._nsds = {}
self._vnfds = {}
VnfrDtsHandler(dts, log, loop, self),
NsdRefCountDtsHandler(dts, log, loop, self),
NsrDtsHandler(dts, log, loop, self),
+ ScalingRpcHandler(log, dts, loop, self.scale_rpc_callback),
NsrRpcDtsHandler(dts,log,loop,self),
self._vnfd_dts_handler,
self.cfgmgr_obj,
self._loop.create_task(nsr.delete_scale_group_instance(scale_group_name, instance_id))
+ def scale_rpc_callback(self, xact, msg, action):
+ """Callback handler for RPC calls
+ Args:
+ xact : Transaction Handler
+ msg : RPC input
+ action : Scaling Action
+ """
+ def get_scaling_group_information():
+ scaling_group_url = "{url}/ns-instance-config/nsr/{nsr_id}".format(url=self._conf_url, nsr_id=msg.nsr_id_ref)
+ output = requests.get(scaling_group_url, headers=self._headers, auth=(self._user, self._password), verify=False)
+ if output.text == None or len(output.text) == 0:
+ self.log.error("nsr id %s information not present", self._nsr_id)
+ return None
+ scaling_group_info = json.loads(output.text)
+ return scaling_group_info
+
+ def config_scaling_group_information(scaling_group_info):
+ data_str = json.dumps(scaling_group_info)
+ self.log.debug("scaling group Info %s", data_str)
+
+ scale_out_url = "{url}/ns-instance-config/nsr/{nsr_id}".format(url=self._conf_url, nsr_id=msg.nsr_id_ref)
+ response = requests.put(scale_out_url, data=data_str, verify=False, auth=(self._user, self._password), headers=self._headers)
+ response.raise_for_status()
+
+ def scale_out():
+ scaling_group_info = get_scaling_group_information()
+ if scaling_group_info is None:
+ return
+
+ scaling_group_present = False
+ if "scaling-group" in scaling_group_info["nsr:nsr"]:
+ scaling_group_array = scaling_group_info["nsr:nsr"]["scaling-group"]
+ for scaling_group in scaling_group_array:
+ if scaling_group["scaling-group-name-ref"] == msg.scaling_group_name_ref:
+ scaling_group_present = True
+ if 'instance' not in scaling_group:
+ scaling_group['instance'] = []
+ for instance in scaling_group['instance']:
+ if instance["id"] == int(msg.instance_id):
+ self.log.error("scaling group with instance id %s exists for scale out", msg.instance_id)
+ return
+ scaling_group["instance"].append({"id": int(msg.instance_id)})
+
+ if not scaling_group_present:
+ scaling_group_info["nsr:nsr"]["scaling-group"] = [{"scaling-group-name-ref": msg.scaling_group_name_ref, "instance": [{"id": msg.instance_id}]}]
+
+ config_scaling_group_information(scaling_group_info)
+ return
+
+ def scale_in():
+ scaling_group_info = get_scaling_group_information()
+ if scaling_group_info is None:
+ return
+
+ scaling_group_array = scaling_group_info["nsr:nsr"]["scaling-group"]
+ scaling_group_present = False
+ instance_id_present = False
+ for scaling_group in scaling_group_array:
+ if scaling_group["scaling-group-name-ref"] == msg.scaling_group_name_ref:
+ scaling_group_present = True
+ if 'instance' in scaling_group:
+ instance_array = scaling_group["instance"];
+ for index in range(len(instance_array)):
+ if instance_array[index]["id"] == int(msg.instance_id):
+ instance_array.pop(index)
+ instance_id_present = True
+ break
+
+ if not scaling_group_present:
+ self.log.error("Scaling group %s doesnot exists for scale in", msg.scaling_group_name_ref)
+ return
+
+ if not instance_id_present:
+ self.log.error("Instance id %s doesnot exists for scale in", msg.instance_id)
+ return
+
+ config_scaling_group_information(scaling_group_info)
+ return
+
+ if action == ScalingRpcHandler.ACTION.SCALE_OUT:
+ self._loop.run_in_executor(None, scale_out)
+ else:
+ self._loop.run_in_executor(None, scale_in)
+
def nsr_update_cfg(self, nsr_id, msg):
nsr = self._nsrs[nsr_id]
nsr.nsr_cfg_msg= msg
path = VirtualLinkRecord.vlr_xpath(vlr)
return (yield from self._vlr_pub_hdlr.delete(xact, path))
+
+class ScalingRpcHandler(mano_dts.DtsHandler):
+ """ The Network service Monitor DTS handler """
+ SCALE_IN_INPUT_XPATH = "I,/nsr:exec-scale-in"
+ SCALE_IN_OUTPUT_XPATH = "O,/nsr:exec-scale-in"
+
+ SCALE_OUT_INPUT_XPATH = "I,/nsr:exec-scale-out"
+ SCALE_OUT_OUTPUT_XPATH = "O,/nsr:exec-scale-out"
+
+ ACTION = Enum('ACTION', 'SCALE_IN SCALE_OUT')
+
+ def __init__(self, log, dts, loop, callback=None):
+ super().__init__(log, dts, loop)
+ self.callback = callback
+ self.last_instance_id = defaultdict(int)
+
+ @asyncio.coroutine
+ def register(self):
+
+ @asyncio.coroutine
+ def on_scale_in_prepare(xact_info, action, ks_path, msg):
+ assert action == rwdts.QueryAction.RPC
+
+ try:
+ rpc_op = NsrYang.YangOutput_Nsr_ExecScaleIn.from_dict({
+ "instance_id": msg.instance_id})
+
+ xact_info.respond_xpath(
+ rwdts.XactRspCode.ACK,
+ self.__class__.SCALE_IN_OUTPUT_XPATH,
+ rpc_op)
+
+ if self.callback:
+ self.callback(xact_info.xact, msg, self.ACTION.SCALE_IN)
+ except Exception as e:
+ self.log.exception(e)
+ xact_info.respond_xpath(
+ rwdts.XactRspCode.NACK,
+ self.__class__.SCALE_IN_OUTPUT_XPATH)
+
+ @asyncio.coroutine
+ def on_scale_out_prepare(xact_info, action, ks_path, msg):
+ assert action == rwdts.QueryAction.RPC
+
+ try:
+ scaling_group = msg.scaling_group_name_ref
+ if not msg.instance_id:
+ last_instance_id = self.last_instance_id[scale_group]
+ msg.instance_id = last_instance_id + 1
+ self.last_instance_id[scale_group] += 1
+
+ rpc_op = NsrYang.YangOutput_Nsr_ExecScaleOut.from_dict({
+ "instance_id": msg.instance_id})
+
+ xact_info.respond_xpath(
+ rwdts.XactRspCode.ACK,
+ self.__class__.SCALE_OUT_OUTPUT_XPATH,
+ rpc_op)
+
+ if self.callback:
+ self.callback(xact_info.xact, msg, self.ACTION.SCALE_OUT)
+ except Exception as e:
+ self.log.exception(e)
+ xact_info.respond_xpath(
+ rwdts.XactRspCode.NACK,
+ self.__class__.SCALE_OUT_OUTPUT_XPATH)
+
+ scale_in_hdl = rift.tasklets.DTS.RegistrationHandler(
+ on_prepare=on_scale_in_prepare)
+ scale_out_hdl = rift.tasklets.DTS.RegistrationHandler(
+ on_prepare=on_scale_out_prepare)
+
+ with self.dts.group_create() as group:
+ group.register(
+ xpath=self.__class__.SCALE_IN_INPUT_XPATH,
+ handler=scale_in_hdl,
+ flags=rwdts.Flag.PUBLISHER)
+ group.register(
+ xpath=self.__class__.SCALE_OUT_INPUT_XPATH,
+ handler=scale_out_hdl,
+ flags=rwdts.Flag.PUBLISHER)
+
+
class NsmTasklet(rift.tasklets.Tasklet):
"""
The network service manager tasklet
projects / osm / SO.git / blobdiff