-#
+#
# Copyright 2016 RIFT.IO Inc
#
# Licensed under the Apache License, Version 2.0 (the "License");
import time
import uuid
import yaml
+import requests
+import json
from collections import deque
import gi
gi.require_version('RwYang', '1.0')
-gi.require_version('RwNsdYang', '1.0')
+gi.require_version('ProjectNsdYang', '1.0')
gi.require_version('RwDts', '1.0')
gi.require_version('RwNsmYang', '1.0')
gi.require_version('RwNsrYang', '1.0')
RwYang,
RwNsrYang,
NsrYang,
- NsdYang,
+ ProjectNsdYang as NsdYang,
RwVlrYang,
VnfrYang,
RwVnfrYang,
import rift.mano.ncclient
import rift.mano.config_data.config
import rift.mano.dts as mano_dts
+from rift.mano.utils.project import (
+ ManoProject,
+ ProjectHandler,
+ get_add_delete_update_cfgs,
+ DEFAULT_PROJECT,
+ )
from . import rwnsm_conman as conman
from . import cloud
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,
"operational_status": 'init',
}
- vnffgr = NsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr.from_dict(vnffgr_dict)
+ vnffgr = NsrYang.YangData_RwProject_Project_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,
"operational_status": 'terminated',
}
- vnffgr = NsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr.from_dict(vnffgr_dict)
+ vnffgr = NsrYang.YangData_RwProject_Project_NsInstanceOpdata_Nsr_Vnffgr.from_dict(vnffgr_dict)
else:
try:
vnffgr = self._vnffgmgr.fetch_vnffgr(self._vnffgr_id)
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,
"operational_status": 'failed',
}
- vnffgr = NsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr.from_dict(vnffgr_dict)
+ vnffgr = NsrYang.YangData_RwProject_Project_NsInstanceOpdata_Nsr_Vnffgr.from_dict(vnffgr_dict)
return vnffgr
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,
}
- vnffgr = NsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_Vnffgr.from_dict(vnffgr_dict)
+ vnffgr = NsrYang.YangData_RwProject_Project_NsInstanceOpdata_Nsr_Vnffgr.from_dict(vnffgr_dict)
for rsp in self._vnffgd_msg.rsp:
vnffgr_rsp = vnffgr.rsp.add()
vnffgr_rsp.id = str(uuid.uuid4())
vnfd = [vnfr.vnfd for vnfr in self._nsr.vnfrs.values() if vnfr.vnfd.id == rsp_cp_ref.vnfd_id_ref]
self._log.debug("VNFD message during VNFFG instantiation is %s",vnfd)
if len(vnfd) > 0 and vnfd[0].has_field('service_function_type'):
- self._log.debug("Service Function Type for VNFD ID %s is %s",rsp_cp_ref.vnfd_id_ref, vnfd[0].service_function_type)
+ self._log.debug("Service Function Type for VNFD ID %s is %s",
+ rsp_cp_ref.vnfd_id_ref, vnfd[0].service_function_type)
else:
- self._log.error("Service Function Type not available for VNFD ID %s; Skipping in chain",rsp_cp_ref.vnfd_id_ref)
+ self._log.error("Service Function Type not available for VNFD ID %s; Skipping in chain",
+ rsp_cp_ref.vnfd_id_ref)
continue
vnfr_cp_ref = vnffgr_rsp.vnfr_connection_point_ref.add()
self._log.info("Received vnf op status is %s; retrying",vnfr.operational_status)
if vnfr.operational_status == 'failed':
self._log.error("Fetching VNFR for %s failed", vnfr.id)
- raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" % (self.id, vnfr.id))
+ raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" %
+ (self.id, vnfr.id))
yield from asyncio.sleep(2, loop=self._loop)
vnfr = yield from self._nsr.fetch_vnfr(nsr_vnfr.xpath)
self._log.debug("Received VNFR is %s", vnfr)
rsp_id_ref = _rsp[0].id
rsp_name = _rsp[0].name
else:
- self._log.error("RSP with ID %s not found during classifier creation for classifier id %s",vnffgd_classifier.rsp_id_ref,vnffgd_classifier.id)
+ self._log.error("RSP with ID %s not found during classifier creation for classifier id %s",
+ vnffgd_classifier.rsp_id_ref,vnffgd_classifier.id)
continue
vnffgr_classifier = vnffgr.classifier.add()
vnffgr_classifier.id = vnffgd_classifier.id
self._log.info("Received vnf op status is %s; retrying",vnfr.operational_status)
if vnfr.operational_status == 'failed':
self._log.error("Fetching VNFR for %s failed", vnfr.id)
- raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" % (self.id, vnfr.id))
+ raise NsrInstantiationFailed("Failed NS %s instantiation due to VNFR %s failure" %
+ (self.id, vnfr.id))
yield from asyncio.sleep(2, loop=self._loop)
vnfr = yield from self._nsr.fetch_vnfr(nsr_vnfr.xpath)
self._log.debug("Received VNFR is %s", vnfr)
for ext_intf in vdu.external_interface:
if ext_intf.name == vnffgr_classifier.vnfr_connection_point_ref:
vnffgr_classifier.vm_id = vdu.vim_id
- self._log.debug("VIM ID for CP %s in VNFR %s is %s",cp.name,nsr_vnfr.id,
- vnfr_cp_ref.connection_point_params.vm_id)
+ self._log.debug("VIM ID for CP %s in VNFR %s is %s",
+ cp.name,nsr_vnfr.id,
+ vnfr_cp_ref.connection_point_params.vm_id)
break
self._log.info("VNFFGR msg to be sent is %s", vnffgr)
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):
+ def create_record(dts, log, loop, project, nsr_name, vld_msg,
+ cloud_account_name, om_datacenter, ip_profile,
+ nsr_id, restart_mode=False):
"""Creates a new VLR object based on the given data.
If restart mode is enabled, then we look for existing records in the
dts,
log,
loop,
+ project,
nsr_name,
vld_msg,
cloud_account_name,
if restart_mode:
res_iter = yield from dts.query_read(
- "D,/vlr:vlr-catalog/vlr:vlr",
+ project.add_project("D,/vlr:vlr-catalog/vlr:vlr"),
rwdts.XactFlag.MERGE)
for fut in res_iter:
return vlr_obj
- def __init__(self, dts, log, loop, nsr_name, vld_msg, cloud_account_name, om_datacenter, ip_profile, nsr_id):
+ def __init__(self, dts, log, loop, project, nsr_name, vld_msg,
+ cloud_account_name, om_datacenter, ip_profile, nsr_id):
self._dts = dts
self._log = log
self._loop = loop
+ self._project = project
self._nsr_name = nsr_name
self._vld_msg = vld_msg
self._cloud_account_name = cloud_account_name
self._vlr_id = str(uuid.uuid4())
self._state = VlRecordState.INIT
self._prev_state = None
-
+ self._create_time = int(time.time())
+
@property
def xpath(self):
""" path for this object """
- return "D,/vlr:vlr-catalog/vlr:vlr[vlr:id = '{}']".format(self._vlr_id)
+ return self._project.add_project("D,/vlr:vlr-catalog/vlr:vlr[vlr:id = '{}']".
+ format(self._vlr_id))
@property
def id(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,
}
vlr_dict['ip_profile_params' ] = self._ip_profile.ip_profile_params.as_dict()
vlr_dict.update(vld_copy_dict)
- vlr = RwVlrYang.YangData_Vlr_VlrCatalog_Vlr.from_dict(vlr_dict)
+ vlr = RwVlrYang.YangData_RwProject_Project_VlrCatalog_Vlr.from_dict(vlr_dict)
return vlr
def reset_id(self, vlr_id):
def create_nsr_vlr_msg(self, vnfrs):
""" The VLR message"""
- nsr_vlr = RwNsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_Vlr()
+ nsr_vlr = RwNsrYang.YangData_RwProject_Project_NsInstanceOpdata_Nsr_Vlr()
nsr_vlr.vlr_ref = self._vlr_id
nsr_vlr.assigned_subnet = self.assigned_subnet
nsr_vlr.cloud_account = self.cloud_account_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
@staticmethod
@asyncio.coroutine
- def create_record(dts, log, loop, vnfd, const_vnfd_msg, nsd_id, nsr_name,
+ def create_record(dts, log, loop, project, vnfd, const_vnfd_msg, nsd_id, nsr_name,
cloud_account_name, om_datacenter_name, nsr_id, group_name, group_instance_id,
placement_groups, restart_mode=False):
"""Creates a new VNFR object based on the given data.
dts,
log,
loop,
+ project,
vnfd,
const_vnfd_msg,
nsd_id,
if restart_mode:
res_iter = yield from dts.query_read(
- "D,/vnfr:vnfr-catalog/vnfr:vnfr",
+ project.add_project("D,/vnfr:vnfr-catalog/vnfr:vnfr"),
rwdts.XactFlag.MERGE)
for fut in res_iter:
dts,
log,
loop,
+ project,
vnfd,
const_vnfd_msg,
nsd_id,
self._dts = dts
self._log = log
self._loop = loop
+ self._project = project
self._vnfd = vnfd
self._const_vnfd_msg = const_vnfd_msg
self._nsd_id = nsd_id
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
@property
def xpath(self):
""" VNFR xpath """
- return "D,/vnfr:vnfr-catalog/vnfr:vnfr[vnfr:id = '{}']".format(self.id)
+ return self._project.add_project("D,/vnfr:vnfr-catalog/vnfr:vnfr[vnfr:id = '{}']"
+ .format(self.id))
@property
def vnfr_msg(self):
@property
def const_vnfr_msg(self):
""" VNFR message """
- return RwNsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_ConstituentVnfrRef(vnfr_id=self.id,cloud_account=self.cloud_account_name,om_datacenter=self._om_datacenter_name)
+ return RwNsrYang.YangData_RwProject_Project_NsInstanceOpdata_Nsr_ConstituentVnfrRef(
+ vnfr_id=self.id, cloud_account=self.cloud_account_name,
+ om_datacenter=self._om_datacenter_name)
@property
def vnfd(self):
@staticmethod
def vnfr_xpath(vnfr):
""" Get the VNFR path from VNFR """
- return (VirtualNetworkFunctionRecord.XPATH + "[vnfr:id = '{}']").format(vnfr.id)
+ return (self._project.add_project(VirtualNetworkFunctionRecord.XPATH) +
+ "[vnfr:id = '{}']").format(vnfr.id)
@property
def config_type(self):
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 = RwVnfrYang.YangData_RwProject_Project_VnfrCatalog_Vnfr.from_dict(vnfr_dict)
+ vnfr.vnfd = VnfrYang.YangData_RwProject_Project_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())
format(self.name, self.vnfr_msg))
yield from self._dts.query_update(
self.xpath,
- rwdts.XactFlag.TRACE,
+ 0, #rwdts.XactFlag.TRACE,
self.vnfr_msg
)
return None
# For every connection point in the VNFD fill in the identifier
+ self._log.debug("Add connection point for VNF %s: %s",
+ self.vnfr_msg.name, self._vnfd.connection_point)
for conn_p in self._vnfd.connection_point:
- cpr = VnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr_ConnectionPoint()
+ cpr = VnfrYang.YangData_RwProject_Project_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,
self._log.info("Created VNF with xpath %s and vnfr %s",
self.xpath, self.vnfr_msg)
- self._log.info("Instantiated VNFR with xpath %s and vnfd %s, vnfr %s",
- self.xpath, self._vnfd, self.vnfr_msg)
-
@asyncio.coroutine
def update_state(self, vnfr_msg):
""" Update this VNFR"""
event_list = []
idx = 1
for entry in self._events:
- event = RwNsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_OperationalEvents()
+ event = RwNsrYang.YangData_RwProject_Project_NsInstanceOpdata_Nsr_OperationalEvents()
event.id = idx
idx += 1
event.timestamp, event.event, event.description, event.details = entry
""" 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, restart_mode=False):
+ def __init__(self, dts, log, loop, nsm, nsm_plugin, nsr_cfg_msg,
+ sdn_account_name, key_pairs, project, 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._project = project
self._nsd = None
self._nsr_msg = None
self._nsr_regh = None
+ self._key_pairs = key_pairs
self._vlrs = []
self._vnfrs = {}
self._vnfds = {}
self._is_active = False
self._vl_phase_completed = False
self._vnf_phase_completed = False
+ self.vlr_uptime_tasks = {}
# Initalise the state to init
self.set_state(NetworkServiceRecordState.INIT)
- self.substitute_input_parameters = InputParameterSubstitution(self._log)
+ self.substitute_input_parameters = InputParameterSubstitution(self._log, self._project)
@property
def nsm_plugin(self):
self._vnf_phase_completed = True
self._op_status.set_state(state)
+ self._nsm_plugin.set_state(self.id, state)
@property
def id(self):
for group_info in self._nsr_cfg_msg.nsd_placement_group_maps:
if group_info.placement_group_ref == input_group.name:
- group = VnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr_PlacementGroupsInfo()
+ group = VnfrYang.YangData_RwProject_Project_VnfrCatalog_Vnfr_PlacementGroupsInfo()
group_dict = {k:v for k,v in
group_info.as_dict().items() if k != 'placement_group_ref'}
for param in copy_dict:
""" Fetch Cloud Account for the passed vnfd id """
if self._nsr_cfg_msg.vnf_cloud_account_map:
vim_accounts = [(vnf.cloud_account,vnf.om_datacenter) for vnf in self._nsr_cfg_msg.vnf_cloud_account_map \
- if vnfd_member_index == vnf.member_vnf_index_ref]
+ if str(vnfd_member_index) == vnf.member_vnf_index_ref]
if vim_accounts and vim_accounts[0]:
return vim_accounts[0]
return (self.cloud_account_name,self.om_datacenter_name)
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_RwProject_Project_VlrCatalog_Vlr.from_dict({'id': vlr.id})
+ while True:
+ vlr_.uptime = int(time.time()) - vlr._create_time
+ xpath = self._project.add_project(VirtualLinkRecord.vlr_xpath(vlr))
+ yield from self._vlr_handler.update(None, xpath, vlr_)
+ yield from asyncio.sleep(2, loop=self._loop)
+ except asyncio.CancelledError:
+ self._log.debug("Received cancellation request for vlr_uptime_update task")
+ xpath = self._project.add_project(VirtualLinkRecord.vlr_xpath(vlr))
+ yield from self._vlr_handler.delete(None, xpath)
+
@asyncio.coroutine
def create(self, config_xact):
self._vnffgrs[vnffgr.id] = vnffgr
def resolve_vld_ip_profile(self, nsd_msg, vld):
+ self._log.debug("Receieved ip profile ref is %s",vld.ip_profile_ref)
if not vld.has_field('ip_profile_ref'):
return None
- profile = [ profile for profile in nsd_msg.ip_profiles if profile.name == vld.ip_profile_ref ]
+ profile = [profile for profile in nsd_msg.ip_profiles if profile.name == vld.ip_profile_ref]
return profile[0] if profile else None
@asyncio.coroutine
def _create_vls(self, vld, cloud_account,om_datacenter):
"""Create a VLR in the cloud account specified using the given VLD
-
+
Args:
vld : VLD yang obj
cloud_account : Cloud account name
-
+
Returns:
VirtualLinkRecord
"""
self._dts,
self._log,
self._loop,
+ self._project,
self.name,
vld,
cloud_account,
""" 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:
if vlr is None:
cloud_account_list = self._extract_cloud_accounts_for_vl(vld)
- for account in cloud_account_list:
- vlr = yield from self._create_vls(vld, account)
+ for account,om_datacenter in cloud_account_list:
+ vlr = yield from self._create_vls(vld, account,om_datacenter)
self._vlrs.append(vlr)
vlr.state = VlRecordState.INSTANTIATION_PENDING
for group in self.nsd_msg.placement_groups:
for member_vnfd in group.member_vnfd:
if (member_vnfd.vnfd_id_ref == vnfd_msg.id) and \
- (member_vnfd.member_vnf_index_ref == const_vnfd.member_vnf_index):
+ (member_vnfd.member_vnf_index_ref == str(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)
# 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,
+ self._project,
vnfd_msg,
const_vnfd,
self.nsd_id,
@property
def nsr_xpath(self):
""" Returns the xpath associated with this NSR """
- return(
+ return self._project.add_project((
"D,/nsr:ns-instance-opdata" +
"/nsr:nsr[nsr:ns-instance-config-ref = '{}']"
- ).format(self.id)
+ ).format(self.id))
@staticmethod
def xpath_from_nsr(nsr):
""" Returns the xpath associated with this NSR op data"""
- return (NetworkServiceRecord.XPATH +
- "[nsr:ns-instance-config-ref = '{}']").format(nsr.id)
+ return self._project.add_project((NetworkServiceRecord.XPATH +
+ "[nsr:ns-instance-config-ref = '{}']").format(nsr.id))
@property
def nsd_xpath(self):
""" Return NSD config xpath."""
- return(
- "C,/nsd:nsd-catalog/nsd:nsd[nsd:id = '{}']"
- ).format(self.nsd_id)
+ return self._project.add_project((
+ "C,/project-nsd:nsd-catalog/project-nsd:nsd[project-nsd:id = '{}']"
+ ).format(self.nsd_id))
@asyncio.coroutine
def instantiate(self, config_xact):
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)
def create_msg(self):
""" The network serice record as a message """
nsr_dict = {"ns_instance_config_ref": self.id}
- nsr = RwNsrYang.YangData_Nsr_NsInstanceOpdata_Nsr.from_dict(nsr_dict)
+ nsr = RwNsrYang.YangData_RwProject_Project_NsInstanceOpdata_Nsr.from_dict(nsr_dict)
#nsr.cloud_account = self.cloud_account_name
nsr.sdn_account = self._sdn_account_name
nsr.name_ref = self.name
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(
+ cfg_prim = NsrYang.YangData_RwProject_Project_NsInstanceOpdata_Nsr_ServicePrimitive.from_dict(
cfg_prim.as_dict())
nsr.service_primitive.append(cfg_prim)
This class is responsible for substituting input parameters into an NSD.
"""
- def __init__(self, log):
+ def __init__(self, log, project):
"""Create an instance of InputParameterSubstitution
Arguments:
"""
self.log = log
+ self.project = project
def __call__(self, nsd, nsr_config):
"""Substitutes input parameters from the NSR config into the NSD
# to be modified
optional_input_parameters = set()
for input_parameter in nsd.input_parameter_xpath:
- optional_input_parameters.add(input_parameter.xpath)
+ optional_input_parameters.add(self.project.add_project(input_parameter.xpath))
# Apply the input parameters to the descriptor
if nsr_config.input_parameter:
@staticmethod
def path_for_id(nsd_id):
""" Return path for the passed nsd_id"""
- return "C,/nsd:nsd-catalog/nsd:nsd[nsd:id = '{}'".format(nsd_id)
+ return self._nsm._project.add_project(
+ "C,/project-nsd:nsd-catalog/project-nsd:nsd[project-nsd:id = '{}'".
+ format(nsd_id))
def path(self):
""" Return the message associated with this NetworkServiceDescriptor"""
class NsdDtsHandler(object):
""" The network service descriptor DTS handler """
- XPATH = "C,/nsd:nsd-catalog/nsd:nsd"
+ XPATH = "C,/project-nsd:nsd-catalog/project-nsd:nsd"
def __init__(self, dts, log, loop, nsm):
self._dts = dts
self._nsm = nsm
self._regh = None
+ self._project = nsm._project
@property
def regh(self):
def register(self):
""" Register for Nsd create/update/delete/read requests from dts """
+ if self._regh:
+ self._log.warning("DTS handler already registered for project {}".
+ format(self._project.name))
+ return
+
def on_apply(dts, acg, xact, action, scratch):
"""Apply the configuration"""
is_recovery = xact.xact is None and action == rwdts.AppconfAction.INSTALL
self._log.debug("Got nsd apply cfg (xact:%s) (action:%s)",
xact, action)
- # Create/Update an NSD record
- for cfg in self._regh.get_xact_elements(xact):
- # Only interested in those NSD cfgs whose ID was received in prepare callback
- if cfg.id in scratch.get('nsds', []) or is_recovery:
- self._nsm.update_nsd(cfg)
+
+ if self._regh:
+ # Create/Update an NSD record
+ for cfg in self._regh.get_xact_elements(xact):
+ # Only interested in those NSD cfgs whose ID was received in prepare callback
+ if cfg.id in scratch.get('nsds', []) or is_recovery:
+ self._nsm.update_nsd(cfg)
+
+ else:
+ self._log.error("No reg handle for {} for project {}".
+ format(self.__class__, self._project.name))
scratch.pop('nsds', None)
except Exception as e:
self._log.error("Exception in cleaning up NSD libs {}: {}".
format(nsd_id, e))
- self._log.excpetion(e)
+ self._log.exception(e)
@asyncio.coroutine
def on_prepare(dts, acg, xact, xact_info, ks_path, msg, scratch):
# Need a list in scratch to store NSDs to create/update later
# acg._scratch['nsds'] = list()
self._regh = acg.register(
- xpath=NsdDtsHandler.XPATH,
+ xpath=self._project.add_project(NsdDtsHandler.XPATH),
flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY | rwdts.Flag.CACHE,
on_prepare=on_prepare)
+ def deregister(self):
+ self._log.debug("De-register NSD handler for project {}".
+ format(self._project.name))
+ if self._regh:
+ self._regh.deregister()
+ self._regh = None
+
class VnfdDtsHandler(object):
""" DTS handler for VNFD config changes """
- XPATH = "C,/vnfd:vnfd-catalog/vnfd:vnfd"
+ XPATH = "C,/project-vnfd:vnfd-catalog/project-vnfd:vnfd"
def __init__(self, dts, log, loop, nsm):
self._dts = dts
self._loop = loop
self._nsm = nsm
self._regh = None
+ self._project = nsm._project
@property
def regh(self):
def register(self):
""" Register for VNFD configuration"""
+ if self._regh:
+ self._log.warning("DTS handler already registered for project {}".
+ format(self._project.name))
+ return
+
@asyncio.coroutine
def on_apply(dts, acg, xact, action, scratch):
"""Apply the configuration"""
self._log.debug("Got NSM VNFD apply (xact: %s) (action: %s)(scr: %s)",
xact, action, scratch)
- # Create/Update a VNFD record
- for cfg in self._regh.get_xact_elements(xact):
- # Only interested in those VNFD cfgs whose ID was received in prepare callback
- if cfg.id in scratch.get('vnfds', []):
- self._nsm.update_vnfd(cfg)
+ if self._regh:
+ # Create/Update a VNFD record
+ for cfg in self._regh.get_xact_elements(xact):
+ # Only interested in those VNFD cfgs whose ID was received in prepare callback
+ if cfg.id in scratch.get('vnfds', []):
+ self._nsm.update_vnfd(cfg)
+
+ for cfg in self._regh.elements:
+ if cfg.id in scratch.get('deleted_vnfds', []):
+ yield from self._nsm.delete_vnfd(cfg.id)
- for cfg in self._regh.elements:
- if cfg.id in scratch.get('deleted_vnfds', []):
- yield from self._nsm.delete_vnfd(cfg.id)
+ else:
+ self._log.error("Reg handle none for {} in project {}".
+ format(self.__class__, self._project))
scratch.pop('vnfds', None)
scratch.pop('deleted_vnfds', None)
xact_info.respond_xpath(rwdts.XactRspCode.ACK)
+ xpath = self._project.add_project(VnfdDtsHandler.XPATH)
self._log.debug(
- "Registering for VNFD config using xpath: %s",
- VnfdDtsHandler.XPATH,
- )
+ "Registering for VNFD config using xpath {} for project {}"
+ .format(xpath, self._project))
acg_hdl = rift.tasklets.AppConfGroup.Handler(on_apply=on_apply)
with self._dts.appconf_group_create(handler=acg_hdl) as acg:
# Need a list in scratch to store VNFDs to create/update later
# acg._scratch['vnfds'] = list()
# acg._scratch['deleted_vnfds'] = list()
self._regh = acg.register(
- xpath=VnfdDtsHandler.XPATH,
+ xpath=xpath,
flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY,
on_prepare=on_prepare)
+ def deregister(self):
+ self._log.debug("De-register VNFD handler for project {}".
+ format(self._project.name))
+ if self._regh:
+ self._regh.deregister()
+ self._regh = None
+
+
class NsrRpcDtsHandler(object):
""" The network service instantiation RPC DTS handler """
EXEC_NSR_CONF_XPATH = "I,/nsr:start-network-service"
EXEC_NSR_CONF_O_XPATH = "O,/nsr:start-network-service"
NETCONF_IP_ADDRESS = "127.0.0.1"
NETCONF_PORT = 2022
+ RESTCONF_PORT = 8888
NETCONF_USER = "admin"
NETCONF_PW = "admin"
+ REST_BASE_V2_URL = 'https://{}:{}/v2/api/'.format("127.0.0.1",8888)
def __init__(self, dts, log, loop, nsm):
self._dts = dts
self._ns_regh = None
self._manager = None
+ self._nsr_config_url = NsrRpcDtsHandler.REST_BASE_V2_URL + \
+ 'config/project/{}/ns-instance-config'. \
+ format(self._nsm._project.name)
self._model = RwYang.Model.create_libncx()
self._model.load_schema_ypbc(RwNsrYang.get_schema())
raise NsrInstantiationFailed("Failed to connect to Launchpad within %s seconds" %
timeout_secs)
+ def _apply_ns_instance_config(self,payload_dict):
+ #self._log.debug("At apply NS instance config with payload %s",payload_dict)
+ req_hdr= {'accept':'application/vnd.yang.data+json',
+ 'content-type':'application/vnd.yang.data+json'}
+ response=requests.post(self._nsr_config_url, headers=req_hdr,
+ auth=('admin', 'admin'),data=payload_dict,verify=False)
+ return response
+
@asyncio.coroutine
def register(self):
""" Register for NS monitoring read from dts """
+ if self._ns_regh:
+ self._log.warning("RPC already registered for project {}".
+ format(self._project.name))
+ return
+
@asyncio.coroutine
def on_ns_config_prepare(xact_info, action, ks_path, msg):
""" prepare callback from dts start-network-service"""
assert action == rwdts.QueryAction.RPC
rpc_ip = msg
+
+ if not self._nsm._project.rpc_check(msg, xact_info=xact_info):
+ return
+
rpc_op = NsrYang.YangOutput_Nsr_StartNetworkService.from_dict({
- "nsr_id":str(uuid.uuid4())
- })
-
- if not ('name' in rpc_ip and 'nsd_ref' in rpc_ip and 'cloud_account' in rpc_ip):
- self._log.error("Mandatory parameters name or nsd_ref or cloud account not found in start-network-service {}".format(rpc_ip))
-
+ "nsr_id":str(uuid.uuid4()),
+ "project_name": msg.prject_name,
+ })
+
+ if not ('name' in rpc_ip and 'nsd_ref' in rpc_ip and
+ ('cloud_account' in rpc_ip or 'om_datacenter' in rpc_ip)):
+ self._log.error("Mandatory parameters name or nsd_ref or " +
+ "cloud account not found in start-network-service {}".
+ format(rpc_ip))
+
self._log.debug("start-network-service RPC input: {}".format(rpc_ip))
try:
# Add used value to the pool
self._log.debug("RPC output: {}".format(rpc_op))
+
nsd_copy = self.nsm.get_nsd(rpc_ip.nsd_ref)
- if not self._manager:
- self._manager = yield from self._connect()
-
+ #if not self._manager:
+ # self._manager = yield from self._connect()
+
self._log.debug("Configuring ns-instance-config with name %s nsd-ref: %s",
rpc_ip.name, rpc_ip.nsd_ref)
ns_instance_config_dict = {"id":rpc_op.nsr_id, "admin_status":"ENABLED"}
ns_instance_config_copy_dict = {k:v for k, v in rpc_ip.as_dict().items()
- if k in RwNsrYang.YangData_Nsr_NsInstanceConfig_Nsr().fields}
+ if k in RwNsrYang.YangData_RwProject_Project_NsInstanceConfig_Nsr().fields}
ns_instance_config_dict.update(ns_instance_config_copy_dict)
- ns_instance_config = RwNsrYang.YangData_Nsr_NsInstanceConfig_Nsr.from_dict(ns_instance_config_dict)
- ns_instance_config.nsd = NsrYang.YangData_Nsr_NsInstanceConfig_Nsr_Nsd()
+ ns_instance_config = RwNsrYang.YangData_RwProject_Project_NsInstanceConfig_Nsr.from_dict(ns_instance_config_dict)
+ ns_instance_config.nsd = NsrYang.YangData_RwProject_Project_NsInstanceConfig_Nsr_Nsd()
ns_instance_config.nsd.from_dict(nsd_copy.msg.as_dict())
- xml = ns_instance_config.to_xml_v2(self._model)
- netconf_xml = self.wrap_netconf_config_xml(xml)
-
- self._log.debug("Sending configure ns-instance-config xml to %s: %s",
- netconf_xml, NsrRpcDtsHandler.NETCONF_IP_ADDRESS)
-
- response = yield from self._manager.edit_config(
- target="running",
- config=netconf_xml,
- )
- self._log.debug("Received edit config response: %s", str(response))
+ payload_dict = ns_instance_config.to_json(self._model)
+ #xml = ns_instance_config.to_xml_v2(self._model)
+ #netconf_xml = self.wrap_netconf_config_xml(xml)
+
+ #self._log.debug("Sending configure ns-instance-config xml to %s: %s",
+ # netconf_xml, NsrRpcDtsHandler.NETCONF_IP_ADDRESS)
+ self._log.debug("Sending configure ns-instance-config json to %s: %s",
+ self._nsr_config_url,ns_instance_config)
+
+ #response = yield from self._manager.edit_config(
+ # target="running",
+ # config=netconf_xml,
+ # )
+ response = yield from self._loop.run_in_executor(
+ None,
+ self._apply_ns_instance_config,
+ payload_dict
+ )
+ response.raise_for_status()
+ self._log.debug("Received edit config response: %s", response.json())
xact_info.respond_xpath(rwdts.XactRspCode.ACK,
NsrRpcDtsHandler.EXEC_NSR_CONF_O_XPATH,
flags=rwdts.Flag.PUBLISHER,
)
+ def deregister(self):
+ self._log.debug("De-register NSR RPC for project {}".
+ format(self._nsm._project.name))
+ if self._ns_regh:
+ self._ns_regh.deregister()
+ self._ns_regh = None
+
class NsrDtsHandler(object):
""" The network service DTS handler """
NSR_XPATH = "C,/nsr:ns-instance-config/nsr:nsr"
SCALE_INSTANCE_XPATH = "C,/nsr:ns-instance-config/nsr:nsr/nsr:scaling-group/nsr:instance"
+ KEY_PAIR_XPATH = "C,/nsr:key-pair"
def __init__(self, dts, log, loop, nsm):
self._dts = dts
self._log = log
self._loop = loop
self._nsm = nsm
+ self._project = self._nsm._project
self._nsr_regh = None
self._scale_regh = None
+ self._key_pair_regh = None
@property
def nsm(self):
def register(self):
""" Register for Nsr create/update/delete/read requests from dts """
+ if self._nsr_regh:
+ self._log.warning("DTS handler already registered for project {}".
+ format(self._project.name))
+ return
+
def nsr_id_from_keyspec(ks):
- nsr_path_entry = NsrYang.YangData_Nsr_NsInstanceConfig_Nsr.schema().keyspec_to_entry(ks)
+ nsr_path_entry = NsrYang.YangData_RwProject_Project_NsInstanceConfig_Nsr.schema().keyspec_to_entry(ks)
nsr_id = nsr_path_entry.key00.id
return nsr_id
def group_name_from_keyspec(ks):
- group_path_entry = NsrYang.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup.schema().keyspec_to_entry(ks)
+ group_path_entry = NsrYang.YangData_RwProject_Project_NsInstanceConfig_Nsr_ScalingGroup.schema().keyspec_to_entry(ks)
group_name = group_path_entry.key00.scaling_group_name_ref
return group_name
for vld in vl_delta["deleted"]:
yield from self._nsm.nsr_terminate_vl(nsr_id, vld)
- def get_add_delete_update_cfgs(dts_member_reg, xact, key_name, scratch):
- # Unfortunately, it is currently difficult to figure out what has exactly
- # changed in this xact without Pbdelta support (RIFT-4916)
- # As a workaround, we can fetch the pre and post xact elements and
- # perform a comparison to figure out adds/deletes/updates
- xact_cfgs = list(dts_member_reg.get_xact_elements(xact))
- curr_cfgs = list(dts_member_reg.elements)
-
- xact_key_map = {getattr(cfg, key_name): cfg for cfg in xact_cfgs}
- curr_key_map = {getattr(cfg, key_name): cfg for cfg in curr_cfgs}
-
- # Find Adds
- added_keys = set(xact_key_map) - set(curr_key_map)
- added_cfgs = [xact_key_map[key] for key in added_keys]
-
- # Find Deletes
- deleted_keys = set(curr_key_map) - set(xact_key_map)
- deleted_cfgs = [curr_key_map[key] for key in deleted_keys]
-
- # Find Updates
- updated_keys = set(curr_key_map) & set(xact_key_map)
- updated_cfgs = [xact_key_map[key] for key in updated_keys
- if xact_key_map[key] != curr_key_map[key]]
-
- return added_cfgs, deleted_cfgs, updated_cfgs
+ def get_nsr_key_pairs(dts_member_reg, xact):
+ key_pairs = {}
+ for instance_cfg, keyspec in dts_member_reg.get_xact_elements(xact, include_keyspec=True):
+ self._log.debug("Key pair received is {} KS: {}".format(instance_cfg, keyspec))
+ xpath = keyspec.to_xpath(RwNsrYang.get_schema())
+ key_pairs[instance_cfg.name] = instance_cfg
+ return key_pairs
def on_apply(dts, acg, xact, action, scratch):
"""Apply the configuration"""
self._log.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
xact, action, scratch)
- def handle_create_nsr(msg, restart_mode=False):
+ def handle_create_nsr(msg, key_pairs=None, restart_mode=False):
# Handle create nsr requests """
# Do some validations
if not msg.has_field("nsd"):
self._log.debug("Creating NetworkServiceRecord %s from nsr config %s",
msg.id, msg.as_dict())
- nsr = self.nsm.create_nsr(msg, restart_mode=restart_mode)
+ nsr = self.nsm.create_nsr(msg, key_pairs=key_pairs, restart_mode=restart_mode)
return nsr
def handle_delete_nsr(msg):
def begin_instantiation(nsr):
# Begin instantiation
self._log.info("Beginning NS instantiation: %s", nsr.id)
- yield from self._nsm.instantiate_ns(nsr.id, xact)
+ try:
+ yield from self._nsm.instantiate_ns(nsr.id, xact)
+ except Exception as e:
+ self._log.exception("NS instantiation: {}".format(e))
+ raise e
self._log.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)",
xact, action, scratch)
if action == rwdts.AppconfAction.INSTALL and xact.id is None:
- for element in self._nsr_regh.elements:
- nsr = handle_create_nsr(element, restart_mode=True)
- self._loop.create_task(begin_instantiation(nsr))
+ key_pairs = []
+ if self._key_pair_regh:
+ for element in self._key_pair_regh.elements:
+ key_pairs.append(element)
+ else:
+ self._log.error("Reg handle none for key pair in project {}".
+ format(self._project))
+
+ if self._nsr_regh:
+ for element in self._nsr_regh.elements:
+ nsr = handle_create_nsr(element, key_pairs, restart_mode=True)
+ self._loop.create_task(begin_instantiation(nsr))
+ else:
+ self._log.error("Reg handle none for NSR in project {}".
+ format(self._project))
(added_msgs, deleted_msgs, updated_msgs) = get_add_delete_update_cfgs(self._nsr_regh,
xact,
- "id",
- scratch)
+ "id")
self._log.debug("Added: %s, Deleted: %s, Updated: %s", added_msgs,
deleted_msgs, updated_msgs)
for msg in added_msgs:
if msg.id not in self._nsm.nsrs:
self._log.info("Create NSR received in on_apply to instantiate NS:%s", msg.id)
- nsr = handle_create_nsr(msg)
+ key_pairs = get_nsr_key_pairs(self._key_pair_regh, xact)
+ nsr = handle_create_nsr(msg,key_pairs)
self._loop.create_task(begin_instantiation(nsr))
for msg in deleted_msgs:
acg.handle.prepare_complete_ok(xact_info.handle)
- self._log.debug("Registering for NSR config using xpath: %s",
- NsrDtsHandler.NSR_XPATH)
+ xpath = self._project.add_project(NsrDtsHandler.NSR_XPATH)
+ self._log.debug("Registering for NSR config using xpath: {}".
+ format(xpath))
acg_hdl = rift.tasklets.AppConfGroup.Handler(on_apply=on_apply)
with self._dts.appconf_group_create(handler=acg_hdl) as acg:
- self._nsr_regh = acg.register(xpath=NsrDtsHandler.NSR_XPATH,
- flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY | rwdts.Flag.CACHE,
- on_prepare=on_prepare)
+ self._nsr_regh = acg.register(
+ xpath=xpath,
+ flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY | rwdts.Flag.CACHE,
+ on_prepare=on_prepare
+ )
self._scale_regh = acg.register(
- xpath=NsrDtsHandler.SCALE_INSTANCE_XPATH,
- flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY| rwdts.Flag.CACHE,
- )
+ xpath=self._project.add_project(NsrDtsHandler.SCALE_INSTANCE_XPATH),
+ flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY| rwdts.Flag.CACHE,
+ )
+
+ self._key_pair_regh = acg.register(
+ xpath=self._project.add_project(NsrDtsHandler.KEY_PAIR_XPATH),
+ flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY | rwdts.Flag.CACHE,
+ )
+
+ def deregister(self):
+ self._log.debug("De-register NSR config for project {}".
+ format(self._project.name))
+ if self._nsr_regh:
+ self._nsr_regh.deregister()
+ self._nsr_regh = None
+ if self._scale_regh:
+ self._scale_regh.deregister()
+ self._scale_regh = None
+ if self._key_pair_regh:
+ self._key_pair_regh.deregister()
+ self._key_pair_regh = None
class NsrOpDataDtsHandler(object):
self._log = log
self._loop = loop
self._nsm = nsm
+
+ self._project = nsm._project
self._regh = None
@property
@asyncio.coroutine
def register(self):
""" Register for Nsr op data publisher registration"""
- self._log.debug("Registering Nsr op data path %s as publisher",
- NsrOpDataDtsHandler.XPATH)
+ if self._regh:
+ self._log.warning("NSR op data handler already registered for project {}".
+ format(self._project.name))
+ return
+
+ xpath = self._project.add_project(NsrOpDataDtsHandler.XPATH)
+ self._log.debug("Registering Nsr op data path {} as publisher".
+ format(xpath))
hdl = rift.tasklets.DTS.RegistrationHandler()
handlers = rift.tasklets.Group.Handler()
with self._dts.group_create(handler=handlers) as group:
- self._regh = group.register(xpath=NsrOpDataDtsHandler.XPATH,
+ self._regh = group.register(xpath=xpath,
handler=hdl,
flags=rwdts.Flag.PUBLISHER | rwdts.Flag.NO_PREP_READ | rwdts.Flag.DATASTORE)
+ def deregister(self):
+ self._log.debug("De-register NSR opdata for project {}".
+ format(self._project.name))
+ if self._regh:
+ self._regh.deregister()
+ self._regh = None
+
@asyncio.coroutine
- def create(self, path, msg):
+ def create(self, xpath, msg):
"""
Create an NS record in DTS with the path and message
"""
+ path = self._project.add_project(xpath)
self._log.debug("Creating NSR %s:%s", path, msg)
self.regh.create_element(path, msg)
self._log.debug("Created NSR, %s:%s", path, msg)
@asyncio.coroutine
- def update(self, path, msg, flags=rwdts.XactFlag.REPLACE):
+ def update(self, xpath, msg, flags=rwdts.XactFlag.REPLACE):
"""
Update an NS record in DTS with the path and message
"""
+ path = self._project.add_project(xpath)
self._log.debug("Updating NSR, %s:%s regh = %s", path, msg, self.regh)
self.regh.update_element(path, msg, flags)
self._log.debug("Updated NSR, %s:%s", path, msg)
@asyncio.coroutine
- def delete(self, path):
+ def delete(self, xpath):
"""
Update an NS record in DTS with the path and message
"""
+ path = self._project.add_project(xpath)
self._log.debug("Deleting NSR path:%s", path)
self.regh.delete_element(path)
self._log.debug("Deleted NSR path:%s", path)
@asyncio.coroutine
def register(self):
""" Register for vnfr create/update/delete/ advises from dts """
+ if self._regh:
+ self._log.warning("VNFR DTS handler already registered for project {}".
+ format(self._project.name))
+ return
+
def on_commit(xact_info):
""" The transaction has been committed """
xact_info, action, ks_path, msg
)
- schema = VnfrYang.YangData_Vnfr_VnfrCatalog_Vnfr.schema()
+ schema = VnfrYang.YangData_RwProject_Project_VnfrCatalog_Vnfr.schema()
path_entry = schema.keyspec_to_entry(ks_path)
- if path_entry.key00.id not in self._nsm._vnfrs:
- self._log.error("%s request for non existent record path %s",
- action, xpath)
+ if not path_entry or (path_entry.key00.id not in self._nsm._vnfrs):
+ # Check if this is a monitoring param xpath
+ if 'vnfr:monitoring-param' not in xpath:
+ self._log.error("%s request for non existent record path %s",
+ action, xpath)
xact_info.respond_xpath(rwdts.XactRspCode.NA, xpath)
return
- self._log.debug("Deleting VNFR with id %s", path_entry.key00.id)
if action == rwdts.QueryAction.CREATE or action == rwdts.QueryAction.UPDATE:
yield from self._nsm.update_vnfr(msg)
elif action == rwdts.QueryAction.DELETE:
hdl = rift.tasklets.DTS.RegistrationHandler(on_commit=on_commit,
on_prepare=on_prepare,)
with self._dts.group_create() as group:
- self._regh = group.register(xpath=VnfrDtsHandler.XPATH,
+ self._regh = group.register(xpath=self._nsm._project.add_project(
+ VnfrDtsHandler.XPATH),
handler=hdl,
flags=(rwdts.Flag.SUBSCRIBER),)
+ def deregister(self):
+ self._log.debug("De-register VNFR for project {}".
+ format(self._nsm._project.name))
+ if self._regh:
+ self._regh.deregister()
+ self._regh = None
class NsdRefCountDtsHandler(object):
""" The NSD Ref Count DTS handler """
@asyncio.coroutine
def register(self):
""" Register for NSD ref count read from dts """
+ if self._regh:
+ self._log.warning("NSD ref DTS handler already registered for project {}".
+ format(self._project.name))
+ return
+
@asyncio.coroutine
def on_prepare(xact_info, action, ks_path, msg):
xpath = ks_path.to_xpath(RwNsrYang.get_schema())
if action == rwdts.QueryAction.READ:
- schema = RwNsrYang.YangData_Nsr_NsInstanceOpdata_NsdRefCount.schema()
+ schema = RwNsrYang.YangData_RwProject_Project_NsInstanceOpdata_NsdRefCount.schema()
path_entry = schema.keyspec_to_entry(ks_path)
nsd_list = yield from self._nsm.get_nsd_refcount(path_entry.key00.nsd_id_ref)
for xpath, msg in nsd_list:
hdl = rift.tasklets.DTS.RegistrationHandler(on_prepare=on_prepare,)
with self._dts.group_create() as group:
- self._regh = group.register(xpath=NsdRefCountDtsHandler.XPATH,
+ self._regh = group.register(xpath=self._nsm._project.add_project(
+ NsdRefCountDtsHandler.XPATH),
handler=hdl,
flags=rwdts.Flag.PUBLISHER,)
+ def deregister(self):
+ self._log.debug("De-register NSD Ref count for project {}".
+ format(self._nsm._project.name))
+ if self._regh:
+ self._regh.deregister()
+ self._regh = None
+
class NsManager(object):
""" The Network Service Manager class"""
- def __init__(self, dts, log, loop,
+ def __init__(self, dts, log, loop, project,
nsr_handler, vnfr_handler, vlr_handler, ro_plugin_selector,
vnffgmgr, vnfd_pub_handler, cloud_account_handler):
self._dts = dts
self._log = log
self._loop = loop
+ self._project = project
self._nsr_handler = nsr_handler
self._vnfr_pub_handler = vnfr_handler
self._vlr_pub_handler = vlr_handler
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),
+ ScalingRpcHandler(log, dts, loop, self._project,
+ self.scale_rpc_callback),
+ NsrRpcDtsHandler(dts, log, loop, self),
self._vnfd_dts_handler,
self.cfgmgr_obj,
]
for dts_handle in self._dts_handlers:
yield from dts_handle.register()
+ def deregister(self):
+ """ Register all static DTS handlers """
+ for dts_handle in self._dts_handlers:
+ dts_handle.deregister()
+
def get_ns_by_nsr_id(self, nsr_id):
""" get NSR by nsr id """
msg : RPC input
action : Scaling Action
"""
- ScalingGroupInstance = NsrYang.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup_Instance
- ScalingGroup = NsrYang.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup
+ ScalingGroupInstance = NsrYang.YangData_RwProject_Project_NsInstanceConfig_Nsr_ScalingGroup_Instance
+ ScalingGroup = NsrYang.YangData_RwProject_Project_NsInstanceConfig_Nsr_ScalingGroup
+
+ xpath = self._project.add_project(
+ ('C,/nsr:ns-instance-config/nsr:nsr[nsr:id="{}"]').
+ format(msg.nsr_id_ref))
- xpath = ('C,/nsr:ns-instance-config/nsr:nsr[nsr:id="{}"]').format(
- msg.nsr_id_ref)
- instance = ScalingGroupInstance.from_dict({"id": msg.instance_id})
+ instance = ScalingGroupInstance.from_dict({
+ "id": msg.instance_id,
+ "project_name": self._project.name,})
@asyncio.coroutine
def get_nsr_scaling_group():
# 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
# Not calling in a separate task as this is called from a separate task
yield from nsr.delete_vl_instance(vld)
- def create_nsr(self, nsr_msg, restart_mode=False):
+ def create_nsr(self, nsr_msg, key_pairs=None,restart_mode=False):
""" Create an NSR instance """
if nsr_msg.id in self._nsrs:
msg = "NSR id %s already exists" % nsr_msg.id
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)
nsm_plugin,
nsr_msg,
sdn_account_name,
- restart_mode=restart_mode
+ key_pairs,
+ self._project,
+ 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)
+ nsm_plugin.create_nsr(nsr_msg, nsr_msg.nsd, key_pairs)
return nsr
@asyncio.coroutine
def get_nsr_config(self, nsd_id):
- xpath = "C,/nsr:ns-instance-config"
+ xpath = self._project.add_project("C,/nsr:ns-instance-config")
results = yield from self._dts.query_read(xpath, rwdts.XactFlag.MERGE)
for result in results:
self.create_nsd(nsd)
else:
self._log.debug("Updating NSD id = %s, nsd = %s", nsd.id, nsd)
- self._nsds[nsd.id].update(nsd)
+ self._nsds[nsd.id].update(nsd)
def delete_nsd(self, nsd_id):
""" Delete the Network service descriptor with the passed id """
""" Update the virtual network function descriptor """
self._log.debug("Update virtual network function descriptor- %s", vnfd)
- # Hack to remove duplicates from leaf-lists - to be fixed by RIFT-6511
- for ivld in vnfd.internal_vld:
- ivld.internal_connection_point_ref = list(set(ivld.internal_connection_point_ref))
if vnfd.id not in self._vnfds:
self._log.debug("No VNFD found - creating VNFD id = %s", vnfd.id)
def nsd_refcount_xpath(nsd_id):
""" xpath for ref count entry """
- return (NsdRefCountDtsHandler.XPATH +
+ return (self._project.add_project(NsdRefCountDtsHandler.XPATH) +
"[rw-nsr:nsd-id-ref = '{}']").format(nsd_id)
nsd_list = []
if nsd_id is None or nsd_id == "":
for nsd in self._nsds.values():
- nsd_msg = RwNsrYang.YangData_Nsr_NsInstanceOpdata_NsdRefCount()
+ nsd_msg = RwNsrYang.YangData_RwProject_Project_NsInstanceOpdata_NsdRefCount()
nsd_msg.nsd_id_ref = nsd.id
nsd_msg.instance_ref_count = nsd.ref_count
nsd_list.append((nsd_refcount_xpath(nsd.id), nsd_msg))
elif nsd_id in self._nsds:
- nsd_msg = RwNsrYang.YangData_Nsr_NsInstanceOpdata_NsdRefCount()
+ nsd_msg = RwNsrYang.YangData_RwProject_Project_NsInstanceOpdata_NsdRefCount()
nsd_msg.nsd_id_ref = self._nsds[nsd_id].id
nsd_msg.instance_ref_count = self._nsds[nsd_id].ref_count
nsd_list.append((nsd_refcount_xpath(nsd_id), nsd_msg))
# Terminate the instances/networks assocaited with this nw service
self._log.debug("Terminating the network service %s", nsr_id)
- yield from self._nsrs[nsr_id].terminate()
+ try :
+ yield from self._nsrs[nsr_id].terminate()
+ except Exception as e:
+ self.log.exception("Failed to terminate NSR[id=%s]", nsr_id)
# Unref the NSD
yield from self.nsd_unref_by_nsr_id(nsr_id)
""" This class provides a publisher interface that allows plugin objects
to publish NSR/VNFR/VLR"""
- def __init__(self, dts, log, loop, nsr_pub_hdlr, vnfr_pub_hdlr, vlr_pub_hdlr):
+ def __init__(self, dts, log, loop, project, nsr_pub_hdlr,
+ vnfr_pub_hdlr, vlr_pub_hdlr,):
self._dts = dts
self._log = log
self._loop = loop
+ self._project = project
self._nsr_pub_hdlr = nsr_pub_hdlr
self._vlr_pub_hdlr = vlr_pub_hdlr
self._vnfr_pub_hdlr = vnfr_pub_hdlr
@asyncio.coroutine
def publish_vlr(self, xact, vlr):
""" Publish a VLR """
- path = VirtualLinkRecord.vlr_xpath(vlr)
+ path = self._project.add_project(VirtualLinkRecord.vlr_xpath(vlr))
return (yield from self._vlr_pub_hdlr.update(xact, path, vlr))
@asyncio.coroutine
def unpublish_vlr(self, xact, vlr):
""" Unpublish a VLR """
- path = VirtualLinkRecord.vlr_xpath(vlr)
+ path = self._project.add_project(VirtualLinkRecord.vlr_xpath(vlr))
return (yield from self._vlr_pub_hdlr.delete(xact, path))
ACTION = Enum('ACTION', 'SCALE_IN SCALE_OUT')
- def __init__(self, log, dts, loop, callback=None):
- super().__init__(log, dts, loop)
+ def __init__(self, log, dts, loop, project, callback=None):
+ super().__init__(log, dts, loop, project)
self.callback = callback
self.last_instance_id = defaultdict(int)
+ self._regh_in = None
+ self._regh_out = None
@asyncio.coroutine
def register(self):
+ if self._regh_in:
+ self._log.warning("RPC already registered for project {}".
+ format(self._project.name))
+ return
+
@asyncio.coroutine
def on_scale_in_prepare(xact_info, action, ks_path, msg):
assert action == rwdts.QueryAction.RPC
try:
+ if not self._project.rpc_check(msg, xact_info=xact_info):
+ return
+
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})
+ "instance_id": msg.instance_id,
+ "project_name": self._project.name,})
xact_info.respond_xpath(
rwdts.XactRspCode.ACK,
assert action == rwdts.QueryAction.RPC
try:
+ if not self._project.rpc_check(msg, xact_info=xact_info):
+ return
+
scaling_group = msg.scaling_group_name_ref
if not msg.instance_id:
last_instance_id = self.last_instance_id[scale_group]
self.callback(xact_info.xact, msg, self.ACTION.SCALE_OUT)
rpc_op = NsrYang.YangOutput_Nsr_ExecScaleOut.from_dict({
- "instance_id": msg.instance_id})
+ "instance_id": msg.instance_id,
+ "project_name": self._project.name,})
xact_info.respond_xpath(
rwdts.XactRspCode.ACK,
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
- """
- def __init__(self, *args, **kwargs):
- super(NsmTasklet, self).__init__(*args, **kwargs)
- self.rwlog.set_category("rw-mano-log")
- self.rwlog.set_subcategory("nsm")
+ self._regh_in = group.register(
+ xpath=self.__class__.SCALE_IN_INPUT_XPATH,
+ handler=scale_in_hdl,
+ flags=rwdts.Flag.PUBLISHER)
+ self._regh_out = group.register(
+ xpath=self.__class__.SCALE_OUT_INPUT_XPATH,
+ handler=scale_out_hdl,
+ flags=rwdts.Flag.PUBLISHER)
+
+ def deregister(self):
+ self._log.debug("De-register scale RPCs for project {}".
+ format(self._project.name))
+ if self._regh_in:
+ self._regh_in.deregister()
+ self._regh_in = None
+ if self._regh_out:
+ self._regh_out.deregister()
+ self._regh_out = None
+
+
+class NsmProject(ManoProject):
+
+ def __init__(self, name, tasklet, **kw):
+ super(NsmProject, self).__init__(tasklet.log, name)
+ self.update(tasklet)
- self._dts = None
self._nsm = None
self._ro_plugin_selector = None
self._vnffgmgr = None
- self._nsr_handler = None
+ self._nsr_pub_handler = None
self._vnfr_pub_handler = None
self._vlr_pub_handler = None
self._vnfd_pub_handler = None
self._records_publisher_proxy = None
- def start(self):
- """ The task start callback """
- super(NsmTasklet, self).start()
- self.log.info("Starting NsmTasklet")
-
- self.log.debug("Registering with dts")
- self._dts = rift.tasklets.DTS(self.tasklet_info,
- RwNsmYang.get_schema(),
- self.loop,
- self.on_dts_state_change)
-
- self.log.debug("Created DTS Api GI Object: %s", self._dts)
-
- def stop(self):
- try:
- self._dts.deinit()
- except Exception:
- print("Caught Exception in NSM stop:", sys.exc_info()[0])
- raise
-
- def on_instance_started(self):
- """ Task instance started callback """
- self.log.debug("Got instance started callback")
-
@asyncio.coroutine
- def init(self):
- """ Task init callback """
- self.log.debug("Got instance started callback")
-
- self.log.debug("creating config account handler")
-
- self._nsr_pub_handler = publisher.NsrOpDataDtsHandler(self._dts, self.log, self.loop)
+ def register(self):
+ self._nsr_pub_handler = publisher.NsrOpDataDtsHandler(
+ self._dts, self.log, self.loop, self)
yield from self._nsr_pub_handler.register()
- self._vnfr_pub_handler = publisher.VnfrPublisherDtsHandler(self._dts, self.log, self.loop)
+ self._vnfr_pub_handler = publisher.VnfrPublisherDtsHandler(
+ self._dts, self.log, self.loop, self)
yield from self._vnfr_pub_handler.register()
- self._vlr_pub_handler = publisher.VlrPublisherDtsHandler(self._dts, self.log, self.loop)
+ self._vlr_pub_handler = publisher.VlrPublisherDtsHandler(
+ self._dts, self.log, self.loop, self)
yield from self._vlr_pub_handler.register()
- manifest = self.tasklet_info.get_pb_manifest()
+ manifest = self._tasklet.tasklet_info.get_pb_manifest()
use_ssl = manifest.bootstrap_phase.rwsecurity.use_ssl
ssl_cert = manifest.bootstrap_phase.rwsecurity.cert
ssl_key = manifest.bootstrap_phase.rwsecurity.key
- self._vnfd_pub_handler = publisher.VnfdPublisher(use_ssl, ssl_cert, ssl_key, self.loop)
+ self._vnfd_pub_handler = publisher.VnfdPublisher(
+ use_ssl, ssl_cert, ssl_key, self.loop, self)
self._records_publisher_proxy = NsmRecordsPublisherProxy(
self._dts,
self.log,
self.loop,
+ self,
self._nsr_pub_handler,
self._vnfr_pub_handler,
self._vlr_pub_handler,
self._dts,
self.log,
self.loop,
+ self,
self._records_publisher_proxy,
)
yield from self._ro_plugin_selector.register()
self._cloud_account_handler = cloud.CloudAccountConfigSubscriber(
self._log,
self._dts,
- self.log_hdl)
+ self.log_hdl,
+ self,
+ )
yield from self._cloud_account_handler.register()
- self._vnffgmgr = rwvnffgmgr.VnffgMgr(self._dts, self.log, self.log_hdl, self.loop)
+ self._vnffgmgr = rwvnffgmgr.VnffgMgr(self._dts, self.log, self.log_hdl, self.loop, self)
yield from self._vnffgmgr.register()
self._nsm = NsManager(
self._dts,
self.log,
self.loop,
+ self,
self._nsr_pub_handler,
self._vnfr_pub_handler,
self._vlr_pub_handler,
self._ro_plugin_selector,
self._vnffgmgr,
self._vnfd_pub_handler,
- self._cloud_account_handler
+ self._cloud_account_handler,
)
yield from self._nsm.register()
+ def deregister(self):
+ self._log.debug("Project {} de-register".format(self.name))
+ self._nsm.deregister()
+ self._vnffgmgr.deregister()
+ self._cloud_account_handler.deregister()
+ self._ro_plugin_selector.deregister()
+ self._nsm = None
+
+ @asyncio.coroutine
+ def delete_prepare(self):
+ # Check if any NS instance is present
+ if self._nsm and self._nsm._nsrs:
+ return False
+ return True
+
+
+class NsmTasklet(rift.tasklets.Tasklet):
+ """
+ The network service manager tasklet
+ """
+ def __init__(self, *args, **kwargs):
+ super(NsmTasklet, self).__init__(*args, **kwargs)
+ self.rwlog.set_category("rw-mano-log")
+ self.rwlog.set_subcategory("nsm")
+
+ self._dts = None
+ self.project_handler = None
+ self.projects = {}
+
+ @property
+ def dts(self):
+ return self._dts
+
+ def start(self):
+ """ The task start callback """
+ super(NsmTasklet, self).start()
+ self.log.info("Starting NsmTasklet")
+
+ self.log.debug("Registering with dts")
+ self._dts = rift.tasklets.DTS(self.tasklet_info,
+ RwNsmYang.get_schema(),
+ self.loop,
+ self.on_dts_state_change)
+
+ self.log.debug("Created DTS Api GI Object: %s", self._dts)
+
+ def stop(self):
+ try:
+ self._dts.deinit()
+ except Exception:
+ print("Caught Exception in NSM stop:", sys.exc_info()[0])
+ raise
+
+ def on_instance_started(self):
+ """ Task instance started callback """
+ self.log.debug("Got instance started callback")
+
+ @asyncio.coroutine
+ def init(self):
+ """ Task init callback """
+ self.log.debug("Got instance started callback")
+
+ self.log.debug("creating project handler")
+ self.project_handler = ProjectHandler(self, NsmProject)
+ self.project_handler.register()
+
+
+
@asyncio.coroutine
def run(self):
""" Task run callback """