if self._vnffgr_state == VnffgRecordState.INIT:
vnffgr_dict = {"id": self._vnffgr_id,
- "nsd_id": self._nsr.nsd_id,
"vnffgd_id_ref": self._vnffgd_msg.id,
"vnffgd_name_ref": self._vnffgd_msg.name,
"sdn_account": self._sdn_account_name,
vnffgr = NsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr.from_dict(vnffgr_dict)
elif self._vnffgr_state == VnffgRecordState.TERMINATED:
vnffgr_dict = {"id": self._vnffgr_id,
- "nsd_id": self._nsr.nsd_id,
"vnffgd_id_ref": self._vnffgd_msg.id,
"vnffgd_name_ref": self._vnffgd_msg.name,
"sdn_account": self._sdn_account_name,
self._log.exception("Fetching VNFFGR for VNFFG with id %s failed", self._vnffgr_id)
self._vnffgr_state = VnffgRecordState.FAILED
vnffgr_dict = {"id": self._vnffgr_id,
- "nsd_id": self._nsr.nsd_id,
"vnffgd_id_ref": self._vnffgd_msg.id,
"vnffgd_name_ref": self._vnffgd_msg.name,
"sdn_account": self._sdn_account_name,
def vnffgr_create_msg(self):
""" Virtual Link Record message for Creating VLR in VNS """
vnffgr_dict = {"id": self._vnffgr_id,
- "nsd_id": self._nsr.nsd_id,
"vnffgd_id_ref": self._vnffgd_msg.id,
"vnffgd_name_ref": self._vnffgd_msg.name,
"sdn_account": self._sdn_account_name,
class VirtualLinkRecord(object):
""" Virtual Link Records class"""
+ XPATH = "D,/vlr:vlr-catalog/vlr:vlr"
@staticmethod
@asyncio.coroutine
def create_record(dts, log, loop, nsr_name, vld_msg, cloud_account_name, om_datacenter, ip_profile, nsr_id, restart_mode=False):
self._vlr_id = str(uuid.uuid4())
self._state = VlRecordState.INIT
self._prev_state = None
+ self._create_time = int(time.time())
@property
def xpath(self):
"nsr_id_ref": self._nsr_id,
"vld_ref": self.vld_msg.id,
"name": self.name,
+ "create_time": self._create_time,
"cloud_account": self.cloud_account_name,
"om_datacenter": self.om_datacenter_name,
}
@asyncio.coroutine
def instantiate(self):
""" Instantiate this VL """
-
self._log.debug("Instaniating VLR key %s, vld %s",
self.xpath, self._vld_msg)
vlr = None
self._group_instance_id = group_instance_id
self._placement_groups = placement_groups
self._config_status = NsrYang.ConfigStates.INIT
+ self._create_time = int(time.time())
self._prev_state = VnfRecordState.INIT
self._state = VnfRecordState.INIT
vnfr_dict = {
"id": self.id,
"nsr_id_ref": self._nsr_id,
- "vnfd_ref": self.vnfd.id,
"name": self.name,
"cloud_account": self._cloud_account_name,
"om_datacenter": self._om_datacenter_name,
vnfr_dict.update(vnfd_copy_dict)
vnfr = RwVnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr.from_dict(vnfr_dict)
+ vnfr.vnfd = VnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr_Vnfd.from_dict(self.vnfd.as_dict(),
+ ignore_missing_keys=True)
vnfr.member_vnf_index_ref = self.member_vnf_index
vnfr.vnf_configuration.from_dict(self._vnfd.vnf_configuration.as_dict())
cpr = VnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr_ConnectionPoint()
cpr.name = conn_p.name
cpr.type_yang = conn_p.type_yang
+ if conn_p.has_field('port_security_enabled'):
+ cpr.port_security_enabled = conn_p.port_security_enabled
+
vlr_ref = find_vlr_for_cp(conn_p)
if vlr_ref is None:
msg = "Failed to find VLR for cp = %s" % conn_p.name
cpr.vlr_ref = vlr_ref.id
self.vnfr_msg.connection_point.append(cpr)
self._log.debug("Connection point [%s] added, vnf id=%s vnfd id=%s",
- cpr, self.vnfr_msg.id, self.vnfr_msg.vnfd_ref)
+ cpr, self.vnfr_msg.id, self.vnfr_msg.vnfd.id)
if not self.restart_mode:
yield from self._dts.query_create(self.xpath,
""" Network service record """
XPATH = "D,/nsr:ns-instance-opdata/nsr:nsr"
- def __init__(self, dts, log, loop, nsm, nsm_plugin, nsr_cfg_msg, sdn_account_name, key_pairs, restart_mode=False):
+ def __init__(self, dts, log, loop, nsm, nsm_plugin, nsr_cfg_msg, sdn_account_name, key_pairs, restart_mode=False,
+ vlr_handler=None):
self._dts = dts
self._log = log
self._loop = loop
self._nsr_cfg_msg = nsr_cfg_msg
self._nsm_plugin = nsm_plugin
self._sdn_account_name = sdn_account_name
+ self._vlr_handler = vlr_handler
self._nsd = None
self._nsr_msg = None
self._is_active = False
self._vl_phase_completed = False
self._vnf_phase_completed = False
+ self.vlr_uptime_tasks = {}
# Initalise the state to init
self._vnf_phase_completed = True
self._op_status.set_state(state)
+ self._nsm_plugin.set_state(self.id, state)
@property
def id(self):
for vlr in self._vlrs:
yield from self.nsm_plugin.instantiate_vl(self, vlr)
vlr.state = VlRecordState.ACTIVE
+ self.vlr_uptime_tasks[vlr.id] = self._loop.create_task(self.vlr_uptime_update(vlr))
+
+
+ def vlr_uptime_update(self, vlr):
+ try:
+
+ vlr_ = RwVlrYang.YangData_Vlr_VlrCatalog_Vlr.from_dict({'id': vlr.id})
+ while True:
+ vlr_.uptime = int(time.time()) - vlr._create_time
+ yield from self._vlr_handler.update(None, VirtualLinkRecord.vlr_xpath(vlr), vlr_)
+ yield from asyncio.sleep(2, loop=self._loop)
+ except asyncio.CancelledError:
+ self._log.debug("Received cancellation request for vlr_uptime_update task")
+ yield from self._vlr_handler.delete(None, VirtualLinkRecord.vlr_xpath(vlr))
+
@asyncio.coroutine
def create(self, config_xact):
""" This function creates VLs for every VLD in the NSD
associated with this NSR"""
for vld in self.nsd_msg.vld:
+
self._log.debug("Found vld %s in nsr id %s", vld, self.id)
cloud_account_list = self._extract_cloud_accounts_for_vl(vld)
for cloud_account,om_datacenter in cloud_account_list:
# Fetch the VNFD associated with this VNF
placement_groups = self.get_placement_groups(vnfd_msg, const_vnfd)
self._log.info("Cloud Account for VNF %d is %s",const_vnfd.member_vnf_index,cloud_account_name)
- self._log.info("Launching VNF: %s (Member Index: %s) in NSD plancement Groups: %s",
+ self._log.info("Launching VNF: %s (Member Index: %s) in NSD plancement Groups: %s, restart mode self.restart_mode %s",
vnfd_msg.name,
const_vnfd.member_vnf_index,
- [ group.name for group in placement_groups])
+ [ group.name for group in placement_groups],
+ self.restart_mode)
vnfr = yield from VirtualNetworkFunctionRecord.create_record(self._dts,
self._log,
self._loop,
for vlr in self.vlrs:
yield from self.nsm_plugin.terminate_vl(vlr)
vlr.state = VlRecordState.TERMINATED
+ if vlr.id in self.vlr_uptime_tasks:
+ self.vlr_uptime_tasks[vlr.id].cancel()
self._log.debug("Terminating network service id %s", self.id)
nsr.config_status = self.map_config_status()
nsr.config_status_details = self._config_status_details
nsr.create_time = self._create_time
+ nsr.uptime = int(time.time()) - self._create_time
for cfg_prim in self.nsd_msg.service_primitive:
cfg_prim = NsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_ServicePrimitive.from_dict(
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
- """
- ScalingGroupInstance = NsrYang.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup_Instance
- ScalingGroup = NsrYang.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup
-
- xpath = ('C,/nsr:ns-instance-config/nsr:nsr[nsr:id="{}"]').format(
- msg.nsr_id_ref)
- instance = ScalingGroupInstance.from_dict({"id": msg.instance_id})
-
- @asyncio.coroutine
- def get_nsr_scaling_group():
- results = yield from self._dts.query_read(xpath, rwdts.XactFlag.MERGE)
-
- for result in results:
- res = yield from result
- nsr_config = res.result
-
- for scaling_group in nsr_config.scaling_group:
- if scaling_group.scaling_group_name_ref == msg.scaling_group_name_ref:
- break
- else:
- scaling_group = nsr_config.scaling_group.add()
- scaling_group.scaling_group_name_ref = msg.scaling_group_name_ref
-
- return (nsr_config, scaling_group)
-
- @asyncio.coroutine
- def update_config(nsr_config):
- xml = self._ncclient.convert_to_xml(RwNsrYang, nsr_config)
- xml = '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(xml)
- yield from self._ncclient.connect()
- yield from self._ncclient.manager.edit_config(target="running", config=xml, default_operation="replace")
-
- @asyncio.coroutine
- def scale_out():
- nsr_config, scaling_group = yield from get_nsr_scaling_group()
- scaling_group.instance.append(instance)
- yield from update_config(nsr_config)
-
- @asyncio.coroutine
- def scale_in():
- nsr_config, scaling_group = yield from get_nsr_scaling_group()
- scaling_group.instance.remove(instance)
- yield from update_config(nsr_config)
-
- if action == ScalingRpcHandler.ACTION.SCALE_OUT:
- self._loop.create_task(scale_out())
- else:
- self._loop.create_task(scale_in())
-
- # Opdata based calls, disabled for now!
- # if action == ScalingRpcHandler.ACTION.SCALE_OUT:
- # self.scale_nsr_out(
- # msg.nsr_id_ref,
- # msg.scaling_group_name_ref,
- # msg.instance_id,
- # xact)
- # else:
- # self.scale_nsr_in(
- # msg.nsr_id_ref,
- # msg.scaling_group_name_ref,
- # msg.instance_id)
-
def nsr_update_cfg(self, nsr_id, msg):
nsr = self._nsrs[nsr_id]
nsr.nsr_cfg_msg= msg
self._log.error(msg)
raise NetworkServiceRecordError(msg)
- self._log.info("Create NetworkServiceRecord nsr id %s from nsd_id %s",
+ self._log.info("Create NetworkServiceRecord nsr id %s from nsd_id %s, restart mode %s",
nsr_msg.id,
- nsr_msg.nsd.id)
+ nsr_msg.nsd.id,
+ restart_mode)
nsm_plugin = self._ro_plugin_selector.ro_plugin
sdn_account_name = self._cloud_account_handler.get_cloud_account_sdn_name(nsr_msg.cloud_account)
nsr_msg,
sdn_account_name,
key_pairs,
- restart_mode=restart_mode
+ restart_mode=restart_mode,
+ vlr_handler=self._ro_plugin_selector._records_publisher._vlr_pub_hdlr
)
self._nsrs[nsr_msg.id] = nsr
nsm_plugin.create_nsr(nsr_msg, nsr_msg.nsd, key_pairs)
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:
- if self.callback:
- self.callback(xact_info.xact, msg, self.ACTION.SCALE_IN)
-
- 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)
-
- 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
-
- if self.callback:
- self.callback(xact_info.xact, msg, self.ACTION.SCALE_OUT)
-
- 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)
-
- 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