RIFT OSM R1 Initial Submission
Signed-off-by: Jeremy Mordkoff <jeremy.mordkoff@riftio.com>
diff --git a/rwcm/CMakeLists.txt b/rwcm/CMakeLists.txt
new file mode 100644
index 0000000..581c9bc
--- /dev/null
+++ b/rwcm/CMakeLists.txt
@@ -0,0 +1,35 @@
+#
+# Copyright 2016 RIFT.IO Inc
+#
+# 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.
+#
+# Author(s): Manish Patel
+# Creation Date: 10/28/2015
+#
+
+cmake_minimum_required(VERSION 2.8)
+
+set(PKG_NAME rwcm)
+set(PKG_VERSION 1.0)
+set(PKG_RELEASE 1)
+set(PKG_LONG_NAME ${PKG_NAME}-${PKG_VERSION})
+
+set(subdirs
+ plugins
+ test
+ )
+
+##
+# Include the subdirs
+##
+rift_add_subdirs(SUBDIR_LIST ${subdirs})
diff --git a/rwcm/plugins/CMakeLists.txt b/rwcm/plugins/CMakeLists.txt
new file mode 100644
index 0000000..1522628
--- /dev/null
+++ b/rwcm/plugins/CMakeLists.txt
@@ -0,0 +1,30 @@
+#
+# Copyright 2016 RIFT.IO Inc
+#
+# 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.
+#
+# Author(s): Manish Patel
+# Creation Date: 10/29/2015
+#
+
+cmake_minimum_required(VERSION 2.8)
+
+set(subdirs
+ yang
+ rwconman
+ )
+
+##
+# Include the subdirs
+##
+rift_add_subdirs(SUBDIR_LIST ${subdirs})
diff --git a/rwcm/plugins/rwconman/CMakeLists.txt b/rwcm/plugins/rwconman/CMakeLists.txt
new file mode 100644
index 0000000..adeb27c
--- /dev/null
+++ b/rwcm/plugins/rwconman/CMakeLists.txt
@@ -0,0 +1,57 @@
+#
+# Copyright 2016 RIFT.IO Inc
+#
+# 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.
+#
+# Author(s): Manish Patel
+# Creation Date: 10/28/2015
+#
+
+include(rift_plugin)
+
+set(TASKLET_NAME rwconmantasklet)
+set(CONMAN_INSTALL "etc/conman")
+
+##
+# Install translation script in demos
+##
+install(
+ FILES
+ rift/tasklets/${TASKLET_NAME}/xlate_cfg.py
+ rift/tasklets/${TASKLET_NAME}/xlate_tags.yml
+ DESTINATION ${CONMAN_INSTALL}
+ COMPONENT ${PKG_LONG_NAME})
+
+
+##
+# This function creates an install target for the plugin artifacts
+##
+rift_install_python_plugin(${TASKLET_NAME} ${TASKLET_NAME}.py)
+
+# Workaround RIFT-6485 - rpmbuild defaults to python2 for
+# anything not in a site-packages directory so we have to
+# install the plugin implementation in site-packages and then
+# import it from the actual plugin.
+rift_python_install_tree(
+ FILES
+ rift/tasklets/${TASKLET_NAME}/__init__.py
+ rift/tasklets/${TASKLET_NAME}/${TASKLET_NAME}.py
+ rift/tasklets/${TASKLET_NAME}/rwconman_config.py
+ rift/tasklets/${TASKLET_NAME}/rwconman_events.py
+ rift/tasklets/${TASKLET_NAME}/jujuconf.py
+ rift/tasklets/${TASKLET_NAME}/RiftCA.py
+ rift/tasklets/${TASKLET_NAME}/riftcm_config_plugin.py
+ rift/tasklets/${TASKLET_NAME}/RiftCM_rpc.py
+ rift/tasklets/${TASKLET_NAME}/rwconman_conagent.py
+ COMPONENT ${PKG_LONG_NAME}
+ PYTHON3_ONLY)
diff --git a/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/RiftCA.py b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/RiftCA.py
new file mode 100644
index 0000000..4a95a7d
--- /dev/null
+++ b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/RiftCA.py
@@ -0,0 +1,299 @@
+#
+# Copyright 2016 RIFT.IO Inc
+#
+# 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 asyncio
+import concurrent.futures
+import re
+import tempfile
+import yaml
+import os
+
+from gi.repository import (
+ RwDts as rwdts,
+)
+
+from . import riftcm_config_plugin
+from . import rwconman_events as Events
+
+class RiftCAConfigPlugin(riftcm_config_plugin.RiftCMConfigPluginBase):
+ """
+ Implementation of the riftcm_config_plugin.RiftCMConfigPluginBase
+ """
+ def __init__(self, dts, log, loop, account):
+ riftcm_config_plugin.RiftCMConfigPluginBase.__init__(self, dts, log, loop, account)
+ self._name = account.name
+ self._type = riftcm_config_plugin.DEFAULT_CAP_TYPE
+ self._rift_install_dir = os.environ['RIFT_INSTALL']
+ self._rift_artif_dir = os.environ['RIFT_ARTIFACTS']
+ self._rift_vnfs = {}
+ self._tasks = {}
+
+ # Instantiate events that will handle RiftCA configuration requests
+ self._events = Events.ConfigManagerEvents(dts, log, loop, self)
+
+ @property
+ def name(self):
+ return self._name
+
+ @property
+ def agent_type(self):
+ return self._type
+
+ @asyncio.coroutine
+ def notify_create_vlr(self, agent_nsr, agent_vnfr, vld, vlr):
+ """
+ Notification of create VL record
+ """
+ pass
+
+ @asyncio.coroutine
+ def is_vnf_configurable(self, agent_vnfr):
+ '''
+ This needs to be part of abstract class
+ '''
+ loop_count = 10
+ while loop_count:
+ loop_count -= 1
+ # Set this VNF's configurability status (need method to check)
+ yield from asyncio.sleep(2, loop=self._loop)
+
+ def riftca_log(self, name, level, log_str, *args):
+ getattr(self._log, level)('RiftCA:({}) {}'.format(name, log_str), *args)
+
+ @asyncio.coroutine
+ def notify_create_vnfr(self, agent_nsr, agent_vnfr):
+ """
+ Notification of create Network VNF record
+ """
+ # Deploy the charm if specified for the vnf
+ self._log.debug("Rift config agent: create vnfr nsr={} vnfr={}"
+ .format(agent_nsr.name, agent_vnfr.name))
+ try:
+ self._loop.create_task(self.is_vnf_configurable(agent_vnfr))
+ except Exception as e:
+ self._log.debug("Rift config agent: vnf_configuration error for VNF:%s/%s: %s",
+ agent_nsr.name, agent_vnfr.name, str(e))
+ return False
+
+ return True
+
+ @asyncio.coroutine
+ def notify_instantiate_vnfr(self, agent_nsr, agent_vnfr):
+ """
+ Notification of Instantiate NSR with the passed nsr id
+ """
+ pass
+
+ @asyncio.coroutine
+ def notify_instantiate_vlr(self, agent_nsr, agent_vnfr, vlr):
+ """
+ Notification of Instantiate NSR with the passed nsr id
+ """
+ pass
+
+ @asyncio.coroutine
+ def notify_terminate_vnfr(self, agent_nsr, agent_vnfr):
+ """
+ Notification of Terminate the network service
+ """
+
+ @asyncio.coroutine
+ def notify_terminate_vlr(self, agent_nsr, agent_vnfr, vlr):
+ """
+ Notification of Terminate the virtual link
+ """
+ pass
+
+ @asyncio.coroutine
+ def vnf_config_primitive(self, agent_nsr, agent_vnfr, primitive, output):
+ '''
+ primitives support by RiftCA
+ '''
+ pass
+
+ @asyncio.coroutine
+ def apply_config(self, config, nsr, vnfr, rpc_ip):
+ """ Notification on configuration of an NSR """
+ pass
+
+ @asyncio.coroutine
+ def apply_ns_config(self, agent_nsr, agent_vnfrs, rpc_ip):
+ """Hook: Runs the user defined script. Feeds all the necessary data
+ for the script thro' yaml file.
+
+ Args:
+ rpc_ip (YangInput_Nsr_ExecNsConfigPrimitive): The input data.
+ nsr (NetworkServiceRecord): Description
+ vnfrs (dict): VNFR ID => VirtualNetworkFunctionRecord
+ """
+
+ def xlate(tag, tags):
+ # TBD
+ if tag is None or tags is None:
+ return tag
+ val = tag
+ if re.search('<.*>', tag):
+ try:
+ if tag == '<rw_mgmt_ip>':
+ val = tags['rw_mgmt_ip']
+ except KeyError as e:
+ self._log.info("RiftCA: Did not get a value for tag %s, e=%s",
+ tag, e)
+ return val
+
+ def get_meta(agent_nsr, agent_vnfrs):
+ unit_names, initial_params, vnfr_index_map, vnfr_data_map = {}, {}, {}, {}
+
+ for vnfr_id in agent_nsr.vnfr_ids:
+ vnfr = agent_vnfrs[vnfr_id]
+
+ # index->vnfr ref
+ vnfr_index_map[vnfr.member_vnf_index] = vnfr_id
+ vnfr_data_dict = dict()
+ if 'mgmt_interface' in vnfr.vnfr:
+ vnfr_data_dict['mgmt_interface'] = vnfr.vnfr['mgmt_interface']
+
+ vnfr_data_dict['connection_point'] = []
+ if 'connection_point' in vnfr.vnfr:
+ for cp in vnfr.vnfr['connection_point']:
+ cp_dict = dict()
+ cp_dict['name'] = cp['name']
+ cp_dict['ip_address'] = cp['ip_address']
+ vnfr_data_dict['connection_point'].append(cp_dict)
+
+ vnfr_data_dict['vdur'] = []
+ vdu_data = [(vdu['name'], vdu['management_ip'], vdu['vm_management_ip'], vdu['id'])
+ for vdu in vnfr.vnfr['vdur']]
+
+ for data in vdu_data:
+ data = dict(zip(['name', 'management_ip', 'vm_management_ip', 'id'] , data))
+ vnfr_data_dict['vdur'].append(data)
+
+ vnfr_data_map[vnfr.member_vnf_index] = vnfr_data_dict
+ # Unit name
+ unit_names[vnfr_id] = vnfr.name
+ # Flatten the data for simplicity
+ param_data = {}
+ if 'initial_config_primitive' in vnfr.vnf_configuration:
+ for primitive in vnfr.vnf_configuration['initial_config_primitive']:
+ for parameter in primitive.parameter:
+ value = xlate(parameter.value, vnfr.tags)
+ param_data[parameter.name] = value
+
+ initial_params[vnfr_id] = param_data
+
+
+ return unit_names, initial_params, vnfr_index_map, vnfr_data_map
+
+ unit_names, init_data, vnfr_index_map, vnfr_data_map = get_meta(agent_nsr, agent_vnfrs)
+ # The data consists of 4 sections
+ # 1. Account data
+ # 2. The input passed.
+ # 3. Unit names (keyed by vnfr ID).
+ # 4. Initial config data (keyed by vnfr ID).
+ data = dict()
+ data['config_agent'] = dict(
+ name=self._name,
+ )
+ data["rpc_ip"] = rpc_ip.as_dict()
+ data["unit_names"] = unit_names
+ data["init_config"] = init_data
+ data["vnfr_index_map"] = vnfr_index_map
+ data["vnfr_data_map"] = vnfr_data_map
+
+ tmp_file = None
+ with tempfile.NamedTemporaryFile(delete=False) as tmp_file:
+ tmp_file.write(yaml.dump(data, default_flow_style=True)
+ .encode("UTF-8"))
+
+ # Get the full path to the script
+ script = ''
+ if rpc_ip.user_defined_script[0] == '/':
+ # The script has full path, use as is
+ script = rpc_ip.user_defined_script
+ else:
+ script = os.path.join(self._rift_artif_dir, 'launchpad/libs', agent_nsr.nsd_id, 'scripts',
+ rpc_ip.user_defined_script)
+ self._log.debug("Rift config agent: Checking for script in %s", script)
+ if not os.path.exists(script):
+ self._log.debug("Rift config agent: Did not find scipt %s", script)
+ script = os.path.join(self._rift_install_dir, 'usr/bin', rpc_ip.user_defined_script)
+
+ cmd = "{} {}".format(script, tmp_file.name)
+ self._log.debug("Rift config agent: Running the CMD: {}".format(cmd))
+
+ coro = asyncio.create_subprocess_shell(cmd, loop=self._loop,
+ stderr=asyncio.subprocess.PIPE)
+ process = yield from coro
+ err = yield from process.stderr.read()
+ task = self._loop.create_task(process.wait())
+
+ return task, err
+
+ @asyncio.coroutine
+ def apply_initial_config(self, agent_nsr, agent_vnfr):
+ """
+ Apply the initial configuration
+ """
+ rc = False
+ self._log.debug("Rift config agent: Apply initial config to VNF:%s/%s",
+ agent_nsr.name, agent_vnfr.name)
+ try:
+ if agent_vnfr.id in self._rift_vnfs.keys():
+ # Check if VNF instance is configurable (TBD - future)
+ ### Remove this once is_vnf_configurable() is implemented
+ agent_vnfr.set_to_configurable()
+ if agent_vnfr.is_configurable:
+ # apply initial config for the vnfr
+ rc = yield from self._events.apply_vnf_config(agent_vnfr.vnf_cfg)
+ else:
+ self._log.info("Rift config agent: VNF:%s/%s is not configurable yet!",
+ agent_nsr.name, agent_vnfr.name)
+ except Exception as e:
+ self._log.error("Rift config agent: Error on initial configuration to VNF:{}/{}, e {}"
+ .format(agent_nsr.name, agent_vnfr.name, str(e)))
+
+ self._log.exception(e)
+ return rc
+
+ return rc
+
+ def is_vnfr_managed(self, vnfr_id):
+ try:
+ if vnfr_id in self._rift_vnfs:
+ return True
+ except Exception as e:
+ self._log.debug("Rift config agent: Is VNFR {} managed: {}".
+ format(vnfr_id, e))
+ return False
+
+ def add_vnfr_managed(self, agent_vnfr):
+ if agent_vnfr.id not in self._rift_vnfs.keys():
+ self._log.info("Rift config agent: add vnfr={}/{}".format(agent_vnfr.name, agent_vnfr.id))
+ self._rift_vnfs[agent_vnfr.id] = agent_vnfr
+
+ @asyncio.coroutine
+ def get_config_status(self, agent_nsr, agent_vnfr):
+ if agent_vnfr.id in self._rift_vnfs.keys():
+ return 'configured'
+ return 'unknown'
+
+
+ def get_action_status(self, execution_id):
+ ''' Get the action status for an execution ID
+ *** Make sure this is NOT a asyncio coroutine function ***
+ '''
+ return None
diff --git a/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/RiftCM_rpc.py b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/RiftCM_rpc.py
new file mode 100644
index 0000000..9155d84
--- /dev/null
+++ b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/RiftCM_rpc.py
@@ -0,0 +1,350 @@
+
+#
+# Copyright 2016 RIFT.IO Inc
+#
+# 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 asyncio
+import time
+
+import rift.mano.config_agent
+import gi
+gi.require_version('RwDts', '1.0')
+gi.require_version('RwNsrYang', '1.0')
+from gi.repository import (
+ RwDts as rwdts,
+ NsrYang,
+)
+
+class RiftCMRPCHandler(object):
+ """ The Network service Monitor DTS handler """
+ EXEC_NS_CONF_XPATH = "I,/nsr:exec-ns-service-primitive"
+ EXEC_NS_CONF_O_XPATH = "O,/nsr:exec-ns-service-primitive"
+
+ GET_NS_CONF_XPATH = "I,/nsr:get-ns-service-primitive-values"
+ GET_NS_CONF_O_XPATH = "O,/nsr:get-ns-service-primitive-values"
+
+ def __init__(self, dts, log, loop, nsm):
+ self._dts = dts
+ self._log = log
+ self._loop = loop
+ self._nsm = nsm
+
+ self._ns_regh = None
+ self._vnf_regh = None
+ self._get_ns_conf_regh = None
+
+ self.job_manager = rift.mano.config_agent.ConfigAgentJobManager(dts, log, loop, nsm)
+
+ @property
+ def reghs(self):
+ """ Return registration handles """
+ return (self._ns_regh, self._vnf_regh, self._get_ns_conf_regh)
+
+ @property
+ def nsm(self):
+ """ Return the NS manager instance """
+ return self._nsm
+
+ def prepare_meta(self, rpc_ip):
+
+ try:
+ nsr_id = rpc_ip.nsr_id_ref
+ nsr = self._nsm.nsrs[nsr_id]
+ vnfrs = {}
+ for vnfr in nsr.vnfrs:
+ vnfr_id = vnfr.id
+ # vnfr is a dict containing all attributes
+ vnfrs[vnfr_id] = vnfr
+
+ return nsr, vnfrs
+ except KeyError as e:
+ raise ValueError("Record not found", str(e))
+
+ @asyncio.coroutine
+ def _get_ns_cfg_primitive(self, nsr_id, ns_cfg_name):
+ nsd_msg = yield from self._nsm.get_nsd(nsr_id)
+
+ def get_nsd_cfg_prim(name):
+ for ns_cfg_prim in nsd_msg.service_primitive:
+ if ns_cfg_prim.name == name:
+ return ns_cfg_prim
+ return None
+
+ ns_cfg_prim_msg = get_nsd_cfg_prim(ns_cfg_name)
+ if ns_cfg_prim_msg is not None:
+ ret_cfg_prim_msg = ns_cfg_prim_msg.deep_copy()
+ return ret_cfg_prim_msg
+ return None
+
+ @asyncio.coroutine
+ def _get_vnf_primitive(self, vnfr_id, nsr_id, primitive_name):
+ vnf = self._nsm.get_vnfr_msg(vnfr_id, nsr_id)
+ self._log.debug("vnfr_msg:%s", vnf)
+ if vnf:
+ self._log.debug("nsr/vnf {}/{}, vnf_configuration: %s",
+ vnf.vnf_configuration)
+ for primitive in vnf.vnf_configuration.service_primitive:
+ if primitive.name == primitive_name:
+ return primitive
+
+ raise ValueError("Could not find nsr/vnf {}/{} primitive {}"
+ .format(nsr_id, vnfr_id, primitive_name))
+
+ @asyncio.coroutine
+ def register(self):
+ """ Register for NS monitoring read from dts """
+ yield from self.job_manager.register()
+
+ @asyncio.coroutine
+ def on_ns_config_prepare(xact_info, action, ks_path, msg):
+ """ prepare callback from dts exec-ns-service-primitive"""
+ assert action == rwdts.QueryAction.RPC
+ rpc_ip = msg
+ rpc_op = NsrYang.YangOutput_Nsr_ExecNsServicePrimitive.from_dict({
+ "triggered_by": rpc_ip.triggered_by,
+ "create_time": int(time.time()),
+ "parameter": [param.as_dict() for param in rpc_ip.parameter],
+ "parameter_group": [pg.as_dict() for pg in rpc_ip.parameter_group]
+ })
+
+ try:
+ ns_cfg_prim_name = rpc_ip.name
+ nsr_id = rpc_ip.nsr_id_ref
+ nsr = self._nsm.nsrs[nsr_id]
+
+ nsd_cfg_prim_msg = yield from self._get_ns_cfg_primitive(nsr_id, ns_cfg_prim_name)
+
+ def find_nsd_vnf_prim_param_pool(vnf_index, vnf_prim_name, param_name):
+ for vnf_prim_group in nsd_cfg_prim_msg.vnf_primitive_group:
+ if vnf_prim_group.member_vnf_index_ref != vnf_index:
+ continue
+
+ for vnf_prim in vnf_prim_group.primitive:
+ if vnf_prim.name != vnf_prim_name:
+ continue
+
+ try:
+ nsr_param_pool = nsr.param_pools[pool_param.parameter_pool]
+ except KeyError:
+ raise ValueError("Parameter pool %s does not exist in nsr" % vnf_prim.parameter_pool)
+
+ self._log.debug("Found parameter pool %s for vnf index(%s), vnf_prim_name(%s), param_name(%s)",
+ nsr_param_pool, vnf_index, vnf_prim_name, param_name)
+ return nsr_param_pool
+
+ self._log.debug("Could not find parameter pool for vnf index(%s), vnf_prim_name(%s), param_name(%s)",
+ vnf_index, vnf_prim_name, param_name)
+ return None
+
+ rpc_op.nsr_id_ref = nsr_id
+ rpc_op.name = ns_cfg_prim_name
+
+ nsr, vnfrs = self.prepare_meta(rpc_ip)
+ rpc_op.job_id = nsr.job_id
+
+ # Copy over the NS level Parameters
+
+ # Give preference to user defined script.
+ if nsd_cfg_prim_msg and nsd_cfg_prim_msg.has_field("user_defined_script"):
+ rpc_ip.user_defined_script = nsd_cfg_prim_msg.user_defined_script
+
+ tasks = []
+ for config_plugin in self.nsm.config_agent_plugins:
+ task, err = yield from config_plugin.apply_ns_config(
+ nsr,
+ vnfrs,
+ rpc_ip)
+ tasks.append(task)
+ if err:
+ rpc_op.job_status_details = err.decode()
+
+ self.job_manager.add_job(rpc_op, tasks)
+ else:
+ # Otherwise create VNF primitives.
+ for vnf in rpc_ip.vnf_list:
+ vnf_op = rpc_op.vnf_out_list.add()
+ vnf_member_idx = vnf.member_vnf_index_ref
+ vnfr_id = vnf.vnfr_id_ref
+ vnf_op.vnfr_id_ref = vnfr_id
+ vnf_op.member_vnf_index_ref = vnf_member_idx
+
+ for primitive in vnf.vnf_primitive:
+ op_primitive = vnf_op.vnf_out_primitive.add()
+ op_primitive.name = primitive.name
+ op_primitive.execution_id = ''
+ op_primitive.execution_status = 'completed'
+ op_primitive.execution_error_details = ''
+
+ # Copy over the VNF pimitive's input parameters
+ for param in primitive.parameter:
+ output_param = op_primitive.parameter.add()
+ output_param.name = param.name
+ output_param.value = param.value
+
+ self._log.debug("%s:%s Got primitive %s:%s",
+ nsr_id, vnf.member_vnf_index_ref, primitive.name, primitive.parameter)
+
+ nsd_vnf_primitive = yield from self._get_vnf_primitive(
+ vnfr_id,
+ nsr_id,
+ primitive.name
+ )
+ for param in nsd_vnf_primitive.parameter:
+ if not param.has_field("parameter_pool"):
+ continue
+
+ try:
+ nsr_param_pool = nsr.param_pools[param.parameter_pool]
+ except KeyError:
+ raise ValueError("Parameter pool %s does not exist in nsr" % param.parameter_pool)
+ nsr_param_pool.add_used_value(param.value)
+
+ for config_plugin in self.nsm.config_agent_plugins:
+ yield from config_plugin.vnf_config_primitive(nsr_id,
+ vnfr_id,
+ primitive,
+ op_primitive)
+
+ self.job_manager.add_job(rpc_op)
+
+ # Get NSD
+ # Find Config Primitive
+ # For each vnf-primitive with parameter pool
+ # Find parameter pool
+ # Add used value to the pool
+ self._log.debug("RPC output: {}".format(rpc_op))
+ xact_info.respond_xpath(rwdts.XactRspCode.ACK,
+ RiftCMRPCHandler.EXEC_NS_CONF_O_XPATH,
+ rpc_op)
+ except Exception as e:
+ self._log.error("Exception processing the "
+ "exec-ns-service-primitive: {}".format(e))
+ self._log.exception(e)
+ xact_info.respond_xpath(rwdts.XactRspCode.NACK,
+ RiftCMRPCHandler.EXEC_NS_CONF_O_XPATH)
+
+ @asyncio.coroutine
+ def on_get_ns_config_values_prepare(xact_info, action, ks_path, msg):
+ assert action == rwdts.QueryAction.RPC
+ nsr_id = msg.nsr_id_ref
+ cfg_prim_name = msg.name
+ try:
+ nsr = self._nsm.nsrs[nsr_id]
+
+ rpc_op = NsrYang.YangOutput_Nsr_GetNsServicePrimitiveValues()
+
+ ns_cfg_prim_msg = yield from self._get_ns_cfg_primitive(nsr_id, cfg_prim_name)
+
+ # Get pool values for NS-level parameters
+ for ns_param in ns_cfg_prim_msg.parameter:
+ if not ns_param.has_field("parameter_pool"):
+ continue
+
+ try:
+ nsr_param_pool = nsr.param_pools[ns_param.parameter_pool]
+ except KeyError:
+ raise ValueError("Parameter pool %s does not exist in nsr" % ns_param.parameter_pool)
+
+ new_ns_param = rpc_op.ns_parameter.add()
+ new_ns_param.name = ns_param.name
+ new_ns_param.value = str(nsr_param_pool.get_next_unused_value())
+
+ # Get pool values for NS-level parameters
+ for vnf_prim_group in ns_cfg_prim_msg.vnf_primitive_group:
+ rsp_prim_group = rpc_op.vnf_primitive_group.add()
+ rsp_prim_group.member_vnf_index_ref = vnf_prim_group.member_vnf_index_ref
+ if vnf_prim_group.has_field("vnfd_id_ref"):
+ rsp_prim_group.vnfd_id_ref = vnf_prim_group.vnfd_id_ref
+
+ for index, vnf_prim in enumerate(vnf_prim_group.primitive):
+ rsp_prim = rsp_prim_group.primitive.add()
+ rsp_prim.name = vnf_prim.name
+ rsp_prim.index = index
+ vnf_primitive = yield from self._get_vnf_primitive(
+ vnf_prim_group.vnfd_id_ref,
+ nsr_id,
+ vnf_prim.name
+ )
+ for param in vnf_primitive.parameter:
+ if not param.has_field("parameter_pool"):
+ continue
+
+ # Get pool values for NS-level parameters
+ for ns_param in ns_cfg_prim_msg.parameter:
+ if not ns_param.has_field("parameter_pool"):
+ continue
+
+ try:
+ nsr_param_pool = nsr.param_pools[ns_param.parameter_pool]
+ except KeyError:
+ raise ValueError("Parameter pool %s does not exist in nsr" % ns_param.parameter_pool)
+
+ new_ns_param = rpc_op.ns_parameter.add()
+ new_ns_param.name = ns_param.name
+ new_ns_param.value = str(nsr_param_pool.get_next_unused_value())
+
+ # Get pool values for NS-level parameters
+ for vnf_prim_group in ns_cfg_prim_msg.vnf_primitive_group:
+ rsp_prim_group = rpc_op.vnf_primitive_group.add()
+ rsp_prim_group.member_vnf_index_ref = vnf_prim_group.member_vnf_index_ref
+ if vnf_prim_group.has_field("vnfd_id_ref"):
+ rsp_prim_group.vnfd_id_ref = vnf_prim_group.vnfd_id_ref
+
+ for index, vnf_prim in enumerate(vnf_prim_group.primitive):
+ rsp_prim = rsp_prim_group.primitive.add()
+ rsp_prim.name = vnf_prim.name
+ rsp_prim.index = index
+ vnf_primitive = yield from self._get_vnf_primitive(
+ nsr_id,
+ vnf_prim_group.member_vnf_index_ref,
+ vnf_prim.name
+ )
+ for param in vnf_primitive.parameter:
+ if not param.has_field("parameter_pool"):
+ continue
+
+ try:
+ nsr_param_pool = nsr.param_pools[param.parameter_pool]
+ except KeyError:
+ raise ValueError("Parameter pool %s does not exist in nsr" % vnf_prim.parameter_pool)
+
+ vnf_param = rsp_prim.parameter.add()
+ vnf_param.name = param.name
+ vnf_param.value = str(nsr_param_pool.get_next_unused_value())
+
+ self._log.debug("RPC output: {}".format(rpc_op))
+ xact_info.respond_xpath(rwdts.XactRspCode.ACK,
+ RiftCMRPCHandler.GET_NS_CONF_O_XPATH, rpc_op)
+ except Exception as e:
+ self._log.error("Exception processing the "
+ "get-ns-service-primitive-values: {}".format(e))
+ self._log.exception(e)
+ xact_info.respond_xpath(rwdts.XactRspCode.NACK,
+ RiftCMRPCHandler.GET_NS_CONF_O_XPATH)
+
+ hdl_ns = rift.tasklets.DTS.RegistrationHandler(on_prepare=on_ns_config_prepare,)
+ hdl_ns_get = rift.tasklets.DTS.RegistrationHandler(on_prepare=on_get_ns_config_values_prepare,)
+
+ with self._dts.group_create() as group:
+ self._ns_regh = group.register(xpath=RiftCMRPCHandler.EXEC_NS_CONF_XPATH,
+ handler=hdl_ns,
+ flags=rwdts.Flag.PUBLISHER,
+ )
+ self._get_ns_conf_regh = group.register(xpath=RiftCMRPCHandler.GET_NS_CONF_XPATH,
+ handler=hdl_ns_get,
+ flags=rwdts.Flag.PUBLISHER,
+ )
+
+
diff --git a/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/__init__.py b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/__init__.py
new file mode 100644
index 0000000..88db365
--- /dev/null
+++ b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/__init__.py
@@ -0,0 +1 @@
+from .rwconmantasklet import ConfigManagerTasklet
diff --git a/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/jujuconf.py b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/jujuconf.py
new file mode 100644
index 0000000..1c32fc9
--- /dev/null
+++ b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/jujuconf.py
@@ -0,0 +1,655 @@
+#
+# Copyright 2016 RIFT.IO Inc
+#
+# 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 asyncio
+import re
+import tempfile
+import yaml
+import os
+
+import rift.mano.utils.juju_api as juju
+from . import riftcm_config_plugin
+
+
+# Charm service name accepts only a to z and -.
+def get_vnf_unique_name(nsr_name, vnfr_short_name, member_vnf_index):
+ name = "{}-{}-{}".format(nsr_name, vnfr_short_name, member_vnf_index)
+ new_name = ''
+ for c in name:
+ if c.isdigit():
+ c = chr(97 + int(c))
+ elif not c.isalpha():
+ c = "-"
+ new_name += c
+ return new_name.lower()
+
+
+class JujuConfigPlugin(riftcm_config_plugin.RiftCMConfigPluginBase):
+ """
+ Juju implementation of the riftcm_config_plugin.RiftCMConfigPluginBase
+ """
+ def __init__(self, dts, log, loop, account):
+ riftcm_config_plugin.RiftCMConfigPluginBase.__init__(self, dts, log, loop, account)
+ self._name = account.name
+ self._type = 'juju'
+ self._ip_address = account.juju.ip_address
+ self._port = account.juju.port
+ self._user = account.juju.user
+ self._secret = account.juju.secret
+ self._rift_install_dir = os.environ['RIFT_INSTALL']
+ self._rift_artif_dir = os.environ['RIFT_ARTIFACTS']
+
+ ############################################################
+ # This is wrongfully overloaded with 'juju' private data. #
+ # Really need to separate agent_vnfr from juju vnfr data. #
+ # Currently, this holds agent_vnfr, which has actual vnfr, #
+ # then this juju overloads actual vnfr with its own #
+ # dictionary elemetns (WRONG!!!) #
+ self._juju_vnfs = {}
+ ############################################################
+
+ self._tasks = {}
+ self._api = juju.JujuApi(log, loop,
+ self._ip_address, self._port,
+ self._user, self._secret)
+
+ @property
+ def name(self):
+ return self._name
+
+ @property
+ def agent_type(self):
+ return self._type
+
+ @property
+ def api(self):
+ return self._api
+
+ @property
+ def vnfr(self, vnfr_id):
+ try:
+ vnfr = self._juju_vnfs[vnfr_id].vnfr
+ except KeyError:
+ self._log.error("jujuCA: Did not find VNFR %s in juju plugin", vnfr_id)
+ return None
+
+ return vnfr
+
+ def juju_log(self, level, name, log_str, *args):
+ if name is not None:
+ g_log_str = 'jujuCA:({}) {}'.format(name, log_str)
+ else:
+ g_log_str = 'jujuCA: {}'.format(log_str)
+ getattr(self._log, level)(g_log_str, *args)
+
+ # TBD: Do a better, similar to config manager
+ def xlate(self, tag, tags):
+ # TBD
+ if tag is None:
+ return tag
+ val = tag
+ if re.search('<.*>', tag):
+ self._log.debug("jujuCA: Xlate value %s", tag)
+ try:
+ if tag == '<rw_mgmt_ip>':
+ val = tags['rw_mgmt_ip']
+ except KeyError as e:
+ self._log.info("jujuCA: Did not get a value for tag %s, e=%s",
+ tag, e)
+ return val
+
+ @asyncio.coroutine
+ def notify_create_vlr(self, agent_nsr, agent_vnfr, vld, vlr):
+ """
+ Notification of create VL record
+ """
+ return True
+
+ @asyncio.coroutine
+ def notify_create_vnfr(self, agent_nsr, agent_vnfr):
+ """
+ Notification of create Network VNF record
+ Returns True if configured using config_agent
+ """
+ # Deploy the charm if specified for the vnf
+ self._log.debug("jujuCA: create vnfr nsr=%s vnfr=%s",
+ agent_nsr.name, agent_vnfr.name)
+ self._log.debug("jujuCA: Config = %s",
+ agent_vnfr.vnf_configuration)
+ try:
+ vnf_config = agent_vnfr.vnfr_msg.vnf_configuration
+ self._log.debug("jujuCA: vnf_configuration = %s", vnf_config)
+ if not vnf_config.has_field('juju'):
+ return False
+ charm = vnf_config.juju.charm
+ self._log.debug("jujuCA: charm = %s", charm)
+ except Exception as e:
+ self._log.Error("jujuCA: vnf_configuration error for vnfr {}: {}".
+ format(agent_vnfr.name, e))
+ return False
+
+ # Prepare unique name for this VNF
+ vnf_unique_name = get_vnf_unique_name(agent_nsr.name,
+ agent_vnfr.name,
+ agent_vnfr.member_vnf_index)
+ if vnf_unique_name in self._tasks:
+ self._log.warn("jujuCA: Service %s already deployed",
+ vnf_unique_name)
+
+ vnfr_dict = agent_vnfr.vnfr
+ vnfr_dict.update({'vnf_juju_name': vnf_unique_name,
+ 'charm': charm,
+ 'nsr_id': agent_nsr.id,
+ 'member_vnf_index': agent_vnfr.member_vnf_index,
+ 'tags': {},
+ 'active': False,
+ 'config': vnf_config,
+ 'vnfr_name' : agent_vnfr.name})
+ self._log.debug("jujuCA: Charm %s for vnf %s to be deployed as %s",
+ charm, agent_vnfr.name, vnf_unique_name)
+
+ # Find the charm directory
+ try:
+ path = os.path.join(self._rift_artif_dir,
+ 'launchpad/libs',
+ agent_vnfr.vnfr_msg.vnfd_ref,
+ 'charms/trusty',
+ charm)
+ self._log.debug("jujuCA: Charm dir is {}".format(path))
+ if not os.path.isdir(path):
+ self._log.error("jujuCA: Did not find the charm directory at {}".
+ format(path))
+ path = None
+ except Exception as e:
+ self.log.exception(e)
+ return False
+
+ if vnf_unique_name not in self._tasks:
+ self._tasks[vnf_unique_name] = {}
+
+ self._tasks[vnf_unique_name]['deploy'] = self.loop.create_task(
+ self.api.deploy_service(charm, vnf_unique_name, path=path))
+
+ self._log.debug("jujuCA: Deploying service %s",
+ vnf_unique_name)
+
+ return True
+
+ @asyncio.coroutine
+ def notify_instantiate_vnfr(self, agent_nsr, agent_vnfr):
+ """
+ Notification of Instantiate NSR with the passed nsr id
+ """
+ return True
+
+ @asyncio.coroutine
+ def notify_instantiate_vlr(self, agent_nsr, agent_vnfr, vlr):
+ """
+ Notification of Instantiate NSR with the passed nsr id
+ """
+ return True
+
+ @asyncio.coroutine
+ def notify_terminate_nsr(self, agent_nsr, agent_vnfr):
+ """
+ Notification of Terminate the network service
+ """
+ return True
+
+ @asyncio.coroutine
+ def notify_terminate_vnfr(self, agent_nsr, agent_vnfr):
+ """
+ Notification of Terminate the network service
+ """
+ self._log.debug("jujuCA: Terminate VNFr {}, current vnfrs={}".
+ format(agent_vnfr.name, self._juju_vnfs))
+ try:
+ vnfr = agent_vnfr.vnfr
+ service = vnfr['vnf_juju_name']
+
+ self._log.debug ("jujuCA: Terminating VNFr %s, %s",
+ agent_vnfr.name, service)
+ self._tasks[service]['destroy'] = self.loop.create_task(
+ self.api.destroy_service(service)
+ )
+
+ del self._juju_vnfs[agent_vnfr.id]
+ self._log.debug ("jujuCA: current vnfrs={}".
+ format(self._juju_vnfs))
+ if service in self._tasks:
+ tasks = []
+ for action in self._tasks[service].keys():
+ #if self.check_task_status(service, action):
+ tasks.append(action)
+ del tasks
+ except KeyError as e:
+ self._log.debug ("jujuCA: Termiating charm service for VNFr {}, e={}".
+ format(agent_vnfr.name, e))
+ except Exception as e:
+ self._log.error("jujuCA: Exception terminating charm service for VNFR {}: {}".
+ format(agent_vnfr.name, e))
+
+ return True
+
+ @asyncio.coroutine
+ def notify_terminate_vlr(self, agent_nsr, agent_vnfr, vlr):
+ """
+ Notification of Terminate the virtual link
+ """
+ return True
+
+ def check_task_status(self, service, action):
+ #self.log.debug("jujuCA: check task status for %s, %s" % (service, action))
+ try:
+ task = self._tasks[service][action]
+ if task.done():
+ self.log.debug("jujuCA: Task for %s, %s done" % (service, action))
+ e = task.exception()
+ if e:
+ self.log.error("jujuCA: Error in task for {} and {} : {}".
+ format(service, action, e))
+ raise Exception(e)
+ r= task.result()
+ if r:
+ self.log.debug("jujuCA: Task for {} and {}, returned {}".
+ format(service, action,r))
+ return True
+ else:
+ self.log.debug("jujuCA: task {}, {} not done".
+ format(service, action))
+ return False
+ except KeyError as e:
+ self.log.error("jujuCA: KeyError for task for {} and {}: {}".
+ format(service, action, e))
+ except Exception as e:
+ self.log.error("jujuCA: Error for task for {} and {}: {}".
+ format(service, action, e))
+ raise
+ return True
+
+ @asyncio.coroutine
+ def vnf_config_primitive(self, nsr_id, vnfr_id, primitive, output):
+ self._log.debug("jujuCA: VNF config primititve {} for nsr {}, vnfr_id {}".
+ format(primitive, nsr_id, vnfr_id))
+ output.execution_status = "failed"
+ output.execution_id = ''
+ output.execution_error_details = ''
+
+ try:
+ vnfr = self._juju_vnfs[vnfr_id].vnfr
+ except KeyError:
+ self._log.error("jujuCA: Did not find VNFR %s in juju plugin",
+ vnfr_id)
+ return
+
+ try:
+ service = vnfr['vnf_juju_name']
+ vnf_config = vnfr['config']
+ self._log.debug("VNF config %s", vnf_config)
+ configs = vnf_config.service_primitive
+ for config in configs:
+ if config.name == primitive.name:
+ self._log.debug("jujuCA: Found the config primitive %s",
+ config.name)
+ params = {}
+ for parameter in primitive.parameter:
+ if parameter.value:
+ val = self.xlate(parameter.value, vnfr['tags'])
+ # TBD do validation of the parameters
+ data_type = 'string'
+ found = False
+ for ca_param in config.parameter:
+ if ca_param.name == parameter.name:
+ data_type = ca_param.data_type
+ found = True
+ break
+ if data_type == 'integer':
+ val = int(parameter.value)
+ if not found:
+ self._log.warn("jujuCA: Did not find parameter {} for {}".
+ format(parameter, config.name))
+ params.update({parameter.name: val})
+
+ if config.name == 'config':
+ if len(params):
+ self._log.debug("jujuCA: applying config with params {} for service {}".
+ format(params, service))
+
+ rc = yield from self.api.apply_config(params, service=service)
+
+ if rc:
+ output.execution_status = "completed"
+ self._log.debug("jujuCA: applied config {} on {}".
+ format(params, service))
+ else:
+ output.execution_status = 'failed'
+ output.execution_error_Details = \
+ 'Failed to apply config: {}'.format(params)
+ self._log.error("jujuCA: Error applying config {} on service {}".
+ format(params, service))
+ else:
+ self._log.warn("jujuCA: Did not find valid paramaters for config : {}".
+ format(primitive.parameter))
+ else:
+ self._log.debug("jujuCA: Execute action {} on service {} with params {}".
+ format(config.name, service, params))
+
+ resp = yield from self.api.execute_action(config.name,
+ params,
+ service=service)
+
+ if resp:
+ if 'error' in resp:
+ output.execution_error_details = resp['error']['Message']
+ else:
+ output.execution_id = resp['action']['tag']
+ output.execution_status = resp['status']
+ if output.execution_status == 'failed':
+ output.execution_error_details = resp['message']
+ self._log.debug("jujuCA: execute action {} on service {} returned {}".
+ format(config.name, service, output.execution_status))
+ else:
+ self._log.error("jujuCA: error executing action {} for {} with {}".
+ format(config.name, service, params))
+ output.execution_id = ''
+ output.execution_status = 'failed'
+ output.execution_error_details = "Failed to queue the action"
+ break
+
+ except KeyError as e:
+ self._log.info("VNF %s does not have config primititves, e=%s", vnfr_id, e)
+
+ @asyncio.coroutine
+ def apply_config(self, agent_nsr, agent_vnfr, config, rpc_ip):
+ """ Notification on configuration of an NSR """
+ pass
+
+ @asyncio.coroutine
+ def apply_ns_config(self, agent_nsr, agent_vnfrs, rpc_ip):
+ """
+
+ ###### TBD - This really does not belong here. Looks more like NS level script ####
+ ###### apply_config should be called for a particular VNF only here ###############
+
+ Hook: Runs the user defined script. Feeds all the necessary data
+ for the script thro' yaml file.
+
+ Args:
+ rpc_ip (YangInput_Nsr_ExecNsConfigPrimitive): The input data.
+ nsr (NetworkServiceRecord): Description
+ vnfrs (dict): VNFR ID => VirtualNetworkFunctionRecord
+
+ """
+ def get_meta(agent_nsr):
+ unit_names, initial_params, vnfr_index_map = {}, {}, {}
+
+ for vnfr_id in agent_nsr.vnfr_ids:
+ juju_vnf = self._juju_vnfs[vnfr_id].vnfr
+
+ # Vnfr -> index ref
+ vnfr_index_map[vnfr_id] = juju_vnf['member_vnf_index']
+
+ # Unit name
+ unit_names[vnfr_id] = juju_vnf['vnf_juju_name']
+
+ # Flatten the data for simplicity
+ param_data = {}
+ self._log.debug("Juju Config:%s", juju_vnf['config'])
+ for primitive in juju_vnf['config'].initial_config_primitive:
+ for parameter in primitive.parameter:
+ value = self.xlate(parameter.value, juju_vnf['tags'])
+ param_data[parameter.name] = value
+
+ initial_params[vnfr_id] = param_data
+
+
+ return unit_names, initial_params, vnfr_index_map
+
+ unit_names, init_data, vnfr_index_map = get_meta(agent_nsr)
+
+ # The data consists of 4 sections
+ # 1. Account data
+ # 2. The input passed.
+ # 3. Juju unit names (keyed by vnfr ID).
+ # 4. Initial config data (keyed by vnfr ID).
+ data = dict()
+ data['config_agent'] = dict(
+ name=self._name,
+ host=self._ip_address,
+ port=self._port,
+ user=self._user,
+ secret=self._secret
+ )
+ data["rpc_ip"] = rpc_ip.as_dict()
+ data["unit_names"] = unit_names
+ data["init_config"] = init_data
+ data["vnfr_index_map"] = vnfr_index_map
+
+ tmp_file = None
+ with tempfile.NamedTemporaryFile(delete=False) as tmp_file:
+ tmp_file.write(yaml.dump(data, default_flow_style=True)
+ .encode("UTF-8"))
+
+ self._log.debug("jujuCA: Creating a temp file: {} with input data".format(
+ tmp_file.name))
+
+ # Get the full path to the script
+ script = ''
+ if rpc_ip.user_defined_script[0] == '/':
+ # The script has full path, use as is
+ script = rpc_ip.user_defined_script
+ else:
+ script = os.path.join(self._rift_artif_dir, 'launchpad/libs', agent_nsr.id, 'scripts',
+ rpc_ip.user_defined_script)
+ self.log.debug("jujuCA: Checking for script in %s", script)
+ if not os.path.exists(script):
+ script = os.path.join(self._rift_install_dir, 'usr/bin', rpc_ip.user_defined_script)
+
+ cmd = "{} {}".format(rpc_ip.user_defined_script, tmp_file.name)
+ self._log.debug("jujuCA: Running the CMD: {}".format(cmd))
+
+ coro = asyncio.create_subprocess_shell(cmd, loop=self._loop,
+ stderr=asyncio.subprocess.PIPE)
+ process = yield from coro
+ err = yield from process.stderr.read()
+ task = self._loop.create_task(process.wait())
+
+ return task, err
+
+ @asyncio.coroutine
+ def apply_initial_config(self, agent_nsr, agent_vnfr):
+ """
+ Apply the initial configuration
+ Expect config directives mostly, not actions
+ Actions in initial config may not work based on charm design
+ """
+
+ vnfr = agent_vnfr.vnfr
+ service = vnfr['vnf_juju_name']
+
+ rc = yield from self.api.is_service_up(service=service)
+ if not rc:
+ return False
+
+ action_ids = []
+ try:
+ vnf_cat = agent_vnfr.vnfr_msg
+ if vnf_cat and vnf_cat.mgmt_interface.ip_address:
+ vnfr['tags'].update({'rw_mgmt_ip': vnf_cat.mgmt_interface.ip_address})
+ self._log.debug("jujuCA:(%s) tags: %s", vnfr['vnf_juju_name'], vnfr['tags'])
+
+ config = {}
+ try:
+ for primitive in vnfr['config'].initial_config_primitive:
+ self._log.debug("jujuCA:(%s) Initial config primitive %s", vnfr['vnf_juju_name'], primitive)
+ if primitive.name == 'config':
+ for param in primitive.parameter:
+ if vnfr['tags']:
+ val = self.xlate(param.value, vnfr['tags'])
+ config.update({param.name: val})
+ except KeyError as e:
+ self._log.exception("jujuCA:(%s) Initial config error(%s): config=%s",
+ vnfr['vnf_juju_name'], str(e), config)
+ config = None
+ return False
+
+ if config:
+ self.juju_log('info', vnfr['vnf_juju_name'],
+ "Applying Initial config:%s",
+ config)
+
+ rc = yield from self.api.apply_config(config, service=service)
+ if rc is False:
+ self.log.error("Service {} is in error state".format(service))
+ return False
+
+
+ # Apply any actions specified as part of initial config
+ for primitive in vnfr['config'].initial_config_primitive:
+ if primitive.name != 'config':
+ self._log.debug("jujuCA:(%s) Initial config action primitive %s",
+ vnfr['vnf_juju_name'], primitive)
+ action = primitive.name
+ params = {}
+ for param in primitive.parameter:
+ val = self.xlate(param.value, vnfr['tags'])
+ params.update({param.name: val})
+
+ self._log.info("jujuCA:(%s) Action %s with params %s",
+ vnfr['vnf_juju_name'], action, params)
+
+ resp = yield from self.api.execute_actions(action, params,
+ service=service)
+ if 'error' in resp:
+ self._log.error("Applying initial config failed: {}".
+ format(resp))
+ return False
+
+ action_ids.append(resp['action']['tag'])
+
+ except KeyError as e:
+ self._log.info("Juju config agent(%s): VNFR %s not managed by Juju",
+ vnfr['vnf_juju_name'], agent_vnfr.id)
+ return False
+ except Exception as e:
+ self._log.exception("jujuCA:(%s) Exception juju apply_initial_config for VNFR {}: {}".
+ format(vnfr['vnf_juju_name'], agent_vnfr.id, e))
+ return False
+
+ # Check if all actions completed
+ pending = True
+ while pending:
+ pending = False
+ for act in action_ids:
+ resp = yield from self.api.get_action_status(act, service=service)
+ if 'error' in resp:
+ self._log.error("Initial config failed: {}".format(resp))
+ return False
+
+ if resp['status'] == 'failed':
+ self._log.error("Initial config action failed: {}".format(resp))
+ return False
+
+ if resp['status'] == 'pending':
+ pending = True
+
+ return True
+
+ def add_vnfr_managed(self, agent_vnfr):
+ if agent_vnfr.id not in self._juju_vnfs.keys():
+ self._log.info("juju config agent: add vnfr={}/{}".
+ format(agent_vnfr.name, agent_vnfr.id))
+ self._juju_vnfs[agent_vnfr.id] = agent_vnfr
+
+ def is_vnfr_managed(self, vnfr_id):
+ try:
+ if vnfr_id in self._juju_vnfs:
+ return True
+ except Exception as e:
+ self._log.debug("jujuCA: Is VNFR {} managed: {}".
+ format(vnfr_id, e))
+ return False
+
+ @asyncio.coroutine
+ def is_configured(self, vnfr_id):
+ try:
+ agent_vnfr = self._juju_vnfs[vnfr_id]
+ vnfr = agent_vnfr.vnfr
+ if vnfr['active']:
+ return True
+
+ vnfr = self._juju_vnfs[vnfr_id].vnfr
+ service = vnfr['vnf_juju_name']
+ resp = self.api.is_service_active(service=service)
+ self._juju_vnfs[vnfr_id]['active'] = resp
+ self._log.debug("jujuCA: Service state for {} is {}".
+ format(service, resp))
+ return resp
+
+ except KeyError:
+ self._log.debug("jujuCA: VNFR id {} not found in config agent".
+ format(vnfr_id))
+ return False
+ except Exception as e:
+ self._log.error("jujuCA: VNFR id {} is_configured: {}".
+ format(vnfr_id, e))
+ return False
+
+ @asyncio.coroutine
+ def get_config_status(self, agent_nsr, agent_vnfr):
+ """Get the configuration status for the VNF"""
+ rc = 'unknown'
+
+ try:
+ vnfr = agent_vnfr.vnfr
+ service = vnfr['vnf_juju_name']
+ except KeyError:
+ # This VNF is not managed by Juju
+ return rc
+
+ rc = 'configuring'
+
+ if not self.check_task_status(service, 'deploy'):
+ return rc
+
+ try:
+ resp = yield from self.api.get_service_status(service=service)
+ self._log.debug("jujuCA: Get service %s status? %s", service, resp)
+
+ if resp == 'error':
+ return 'error'
+ if resp == 'active':
+ return 'configured'
+ except KeyError:
+ self._log.error("jujuCA: Check unknown service %s status", service)
+ except Exception as e:
+ self._log.error("jujuCA: Caught exception when checking for service is active: %s", e)
+ self._log.exception(e)
+
+ return rc
+
+ def get_action_status(self, execution_id):
+ ''' Get the action status for an execution ID
+ *** Make sure this is NOT a asyncio coroutine function ***
+ '''
+
+ try:
+ return self.api._get_action_status(execution_id)
+ except Exception as e:
+ self._log.error("jujuCA: Error fetching execution status for %s",
+ execution_id)
+ self._log.exception(e)
+ raise e
diff --git a/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/riftcm_config_plugin.py b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/riftcm_config_plugin.py
new file mode 100644
index 0000000..1540360
--- /dev/null
+++ b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/riftcm_config_plugin.py
@@ -0,0 +1,307 @@
+#
+# Copyright 2016 RIFT.IO Inc
+#
+# 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 asyncio
+import abc
+
+# Default config agent plugin type
+DEFAULT_CAP_TYPE = "riftca"
+
+class RiftCMnsr(object):
+ '''
+ Agent class for NSR
+ created for Agents to use objects from NSR
+ '''
+ def __init__(self, nsr_dict, cfg):
+ self._nsr = nsr_dict
+ self._cfg = cfg
+ self._vnfrs = []
+ self._vnfrs_msg = []
+ self._vnfr_ids = {}
+ self._job_id = 0
+
+ @property
+ def name(self):
+ return self._nsr['name_ref']
+
+ @property
+ def nsd_name(self):
+ return self._nsr['nsd_name_ref']
+
+ @property
+ def nsd_id(self):
+ return self._nsr['nsd_ref']
+
+ @property
+ def id(self):
+ return self._nsr['ns_instance_config_ref']
+
+ @property
+ def nsr_dict(self):
+ return self._nsr
+
+ @property
+ def nsr_cfg_msg(self):
+ return self._cfg
+
+ @property
+ def job_id(self):
+ ''' Get a new job id for config primitive'''
+ self._job_id += 1
+ return self._job_id
+
+ @property
+ def vnfrs(self):
+ return self._vnfrs
+
+ @property
+ def member_vnf_index(self):
+ return self._vnfr['member_vnf_index_ref']
+
+ def add_vnfr(self, vnfr, vnfr_msg):
+ if vnfr['id'] in self._vnfr_ids.keys():
+ agent_vnfr = self._vnfr_ids[vnfr['id']]
+ else:
+ agent_vnfr = RiftCMvnfr(self.name, vnfr, vnfr_msg)
+ self._vnfrs.append(agent_vnfr)
+ self._vnfrs_msg.append(vnfr_msg)
+ self._vnfr_ids[agent_vnfr.id] = agent_vnfr
+ return agent_vnfr
+
+ @property
+ def vnfr_ids(self):
+ return self._vnfr_ids
+
+class RiftCMvnfr(object):
+ '''
+ Agent base class for VNFR processing
+ '''
+ def __init__(self, nsr_name, vnfr_dict, vnfr_msg):
+ self._vnfr = vnfr_dict
+ self._vnfr_msg = vnfr_msg
+ self._nsr_name = nsr_name
+ self._configurable = False
+
+ @property
+ def nsr_name(self):
+ return self._nsr_name
+
+ @property
+ def vnfr(self):
+ return self._vnfr
+
+ @property
+ def vnfr_msg(self):
+ return self._vnfr_msg
+
+ @property
+ def name(self):
+ return self._vnfr['short_name']
+
+ @property
+ def tags(self):
+ try:
+ return self._vnfr['tags']
+ except KeyError:
+ return None
+
+ @property
+ def id(self):
+ return self._vnfr['id']
+
+ @property
+ def member_vnf_index(self):
+ return self._vnfr['member_vnf_index_ref']
+
+ @property
+ def vnf_configuration(self):
+ return self._vnfr['vnf_configuration']
+
+ @property
+ def xpath(self):
+ """ VNFR xpath """
+ return "D,/vnfr:vnfr-catalog/vnfr:vnfr[vnfr:id = '{}']".format(self.id)
+
+ def set_to_configurable(self):
+ self._configurable = True
+
+ @property
+ def is_configurable(self):
+ return self._configurable
+
+ @property
+ def vnf_cfg(self):
+ return self._vnfr['vnf_cfg']
+
+class RiftCMConfigPluginBase(object):
+ """
+ Abstract base class for the NSM Configuration agent plugin.
+ There will be single instance of this plugin for each plugin type.
+ """
+
+ def __init__(self, dts, log, loop, config_agent):
+ self._dts = dts
+ self._log = log
+ self._loop = loop
+ self._config_agent = config_agent
+
+ @property
+ def agent_type(self):
+ raise NotImplementedError
+
+ @property
+ def name(self):
+ raise NotImplementedError
+
+ @property
+ def dts(self):
+ return self._dts
+
+ @property
+ def log(self):
+ return self._log
+
+ @property
+ def loop(self):
+ return self._loop
+
+ @property
+ def nsm(self):
+ return self._nsm
+
+
+ @abc.abstractmethod
+ @asyncio.coroutine
+ def apply_config(self, agent_nsr, agent_vnfr, config, rpc_ip):
+ """ Notification on configuration of an NSR """
+ pass
+
+ @abc.abstractmethod
+ @asyncio.coroutine
+ def apply_ns_config(self, agent_nsr, agent_vnfrs, config, rpc_ip):
+ """ Notification on configuration of an NSR """
+ pass
+
+ @abc.abstractmethod
+ @asyncio.coroutine
+ def notify_create_vlr(self, agent_nsr, vld):
+ """ Notification on creation of an VL """
+ pass
+
+ @abc.abstractmethod
+ @asyncio.coroutine
+ def notify_create_vnfr(self, agent_nsr, agent_vnfr):
+ """ Notification on creation of an VNFR """
+ pass
+
+ @abc.abstractmethod
+ @asyncio.coroutine
+ def notify_instantiate_vnfr(self, agent_nsr, agent_vnfr):
+ """ Notify instantiation of the virtual network function """
+ pass
+
+ @abc.abstractmethod
+ @asyncio.coroutine
+ def notify_instantiate_vlr(self, agent_nsr, vl):
+ """ Notify instantiate of the virtual link"""
+ pass
+
+ @abc.abstractmethod
+ @asyncio.coroutine
+ def notify_terminate_vnfr(self, agent_nsr, agent_vnfr):
+ """Notify termination of the VNF """
+ pass
+
+ @abc.abstractmethod
+ @asyncio.coroutine
+ def notify_terminate_vlr(self, agent_nsr, vlr):
+ """Notify termination of the Virtual Link Record"""
+ pass
+
+ @abc.abstractmethod
+ @asyncio.coroutine
+ def apply_initial_config(self, vnfr_id, vnf):
+ """Apply initial configuration"""
+ pass
+
+ @abc.abstractmethod
+ @asyncio.coroutine
+ def get_config_status(self, vnfr_id):
+ """Get the status for the VNF"""
+ pass
+
+ @abc.abstractmethod
+ def get_action_status(self, execution_id):
+ """Get the action exection status"""
+ pass
+
+ @abc.abstractmethod
+ @asyncio.coroutine
+ def vnf_config_primitive(self, nsr_id, vnfr_id, primitive, output):
+ """Apply config primitive on a VNF"""
+ pass
+
+ @abc.abstractmethod
+ def is_vnfr_managed(self, vnfr_id):
+ """ Check if VNR is managed by config agent """
+ pass
+
+ @abc.abstractmethod
+ def add_vnfr_managed(self, agent_vnfr):
+ """ Add VNR to be managed by this config agent """
+ pass
+
+ @asyncio.coroutine
+ def invoke(self, method, *args):
+ try:
+ rc = None
+ self._log.debug("Config agent plugin: method {} with args {}: {}".
+ format(method, args, self))
+
+ # TBD - Do a better way than string compare to find invoke the method
+ if method == 'notify_create_nsr':
+ rc = yield from self.notify_create_nsr(args[0], args[1])
+ elif method == 'notify_create_vlr':
+ rc = yield from self.notify_create_vlr(args[0], args[1], args[2])
+ elif method == 'notify_create_vnfr':
+ rc = yield from self.notify_create_vnfr(args[0], args[1])
+ elif method == 'notify_instantiate_nsr':
+ rc = yield from self.notify_instantiate_nsr(args[0])
+ elif method == 'notify_instantiate_vnfr':
+ rc = yield from self.notify_instantiate_vnfr(args[0], args[1])
+ elif method == 'notify_instantiate_vlr':
+ rc = yield from self.notify_instantiate_vlr(args[0], args[1])
+ elif method == 'notify_nsr_active':
+ rc = yield from self.notify_nsr_active(args[0], args[1])
+ elif method == 'notify_terminate_nsr':
+ rc = yield from self.notify_terminate_nsr(args[0])
+ elif method == 'notify_terminate_vnfr':
+ rc = yield from self.notify_terminate_vnfr(args[0], args[1])
+ elif method == 'notify_terminate_vlr':
+ rc = yield from self.notify_terminate_vlr(args[0], args[1])
+ elif method == 'apply_initial_config':
+ rc = yield from self.apply_initial_config(args[0], args[1])
+ elif method == 'apply_config':
+ rc = yield from self.apply_config(args[0], args[1], args[2])
+ elif method == 'get_config_status':
+ rc = yield from self.get_config_status(args[0], args[1])
+ else:
+ self._log.error("Unknown method %s invoked on config agent plugin",
+ method)
+ except Exception as e:
+ self._log.error("Caught exception while invoking method: %s, Exception: %s", method, str(e))
+ raise
+ return rc
diff --git a/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/rwconman_conagent.py b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/rwconman_conagent.py
new file mode 100644
index 0000000..543e51b
--- /dev/null
+++ b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/rwconman_conagent.py
@@ -0,0 +1,263 @@
+#
+# Copyright 2016 RIFT.IO Inc
+#
+# 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 asyncio
+import rift.tasklets
+
+from gi.repository import (
+ RwConfigAgentYang as rwcfg_agent,
+)
+
+from .riftcm_config_plugin import DEFAULT_CAP_TYPE
+from . import RiftCA
+from . import jujuconf
+import rift.mano.config_agent
+
+
+class ConfigAgentError(Exception):
+ pass
+
+
+class ConfigAgentExistsError(ConfigAgentError):
+ pass
+
+
+class UnknownAgentTypeError(Exception):
+ pass
+
+
+class ConfigAgentVnfrAddError(Exception):
+ pass
+
+
+class ConfigAgentVnfrTypeError(Exception):
+ pass
+
+
+class ConfigAccountHandler(object):
+ def __init__(self, dts, log, loop, on_add_config_agent, on_delete_config_agent):
+ self._log = log
+ self._dts = dts
+ self._loop = loop
+ self._on_add_config_agent = on_add_config_agent
+ self._on_delete_config_agent = on_delete_config_agent
+
+ self._log.debug("creating config account handler")
+ self.cloud_cfg_handler = rift.mano.config_agent.ConfigAgentSubscriber(
+ self._dts, self._log,
+ rift.mano.config_agent.ConfigAgentCallbacks(
+ on_add_apply=self.on_config_account_added,
+ on_delete_apply=self.on_config_account_deleted,
+ )
+ )
+
+ def on_config_account_deleted(self, account):
+ self._log.debug("config account deleted: %s", account.name)
+ self._on_delete_config_agent(account)
+
+ def on_config_account_added(self, account):
+ self._log.debug("config account added")
+ self._log.debug(account.as_dict())
+ self._on_add_config_agent(account)
+
+ @asyncio.coroutine
+ def register(self):
+ self.cloud_cfg_handler.register()
+
+class RiftCMConfigPlugins(object):
+ """ NSM Config Agent Plugins """
+ def __init__(self):
+ self._plugin_classes = {
+ "juju": jujuconf.JujuConfigPlugin,
+ "riftca": RiftCA.RiftCAConfigPlugin,
+ }
+
+ @property
+ def plugins(self):
+ """ Plugin info """
+ return self._plugin_classes
+
+ def __getitem__(self, name):
+ """ Get item """
+ return self._plugin_classes[name]
+
+ def register(self, plugin_name, plugin_class, *args):
+ """ Register a plugin to this Nsm"""
+ self._plugin_classes[plugin_name] = plugin_class
+
+ def deregister(self, plugin_name, plugin_class, *args):
+ """ Deregister a plugin to this Nsm"""
+ if plugin_name in self._plugin_classes:
+ del self._plugin_classes[plugin_name]
+
+ def class_by_plugin_name(self, name):
+ """ Get class by plugin name """
+ return self._plugin_classes[name]
+
+
+class RiftCMConfigAgent(object):
+ def __init__(self, dts, log, loop, parent):
+ self._dts = dts
+ self._log = log
+ self._loop = loop
+ self._ConfigManagerConfig = parent
+
+ self._config_plugins = RiftCMConfigPlugins()
+ self._config_handler = ConfigAccountHandler(
+ self._dts, self._log, self._loop, self._on_config_agent, self._on_config_agent_delete)
+ self._plugin_instances = {}
+ self._default_account_added = False
+
+ @asyncio.coroutine
+ def invoke_config_agent_plugins(self, method, nsr, vnfr, *args):
+ # Invoke the methods on all config agent plugins registered
+ rc = False
+ for agent in self._plugin_instances.values():
+ if not agent.is_vnfr_managed(vnfr.id):
+ continue
+ try:
+ self._log.debug("Invoke {} on {}".format(method, agent.name))
+ rc = yield from agent.invoke(method, nsr, vnfr, *args)
+ break
+ except Exception as e:
+ self._log.error("Error invoking {} on {} : {}".
+ format(method, agent.name, e))
+ raise
+
+ self._log.info("vnfr({}), method={}, return rc={}"
+ .format(vnfr.name, method, rc))
+ return rc
+
+ def is_vnfr_config_agent_managed(self, vnfr):
+ if (not vnfr.has_field('netconf') and
+ not vnfr.has_field('juju') and
+ not vnfr.has_field('script')):
+ return False
+
+ for agent in self._plugin_instances.values():
+ try:
+ if agent.is_vnfr_managed(vnfr.id):
+ return True
+ except Exception as e:
+ self._log.debug("Check if VNFR {} is config agent managed: {}".
+ format(vnfr.name, e))
+ return False
+
+ def _on_config_agent(self, config_agent):
+ self._log.debug("Got nsm plugin config agent account: %s", config_agent)
+ try:
+ cap_name = config_agent.name
+ cap_inst = self._config_plugins.class_by_plugin_name(
+ config_agent.account_type)
+ except KeyError as e:
+ msg = "Config agent nsm plugin type not found: {}". \
+ format(config_agent.account_type)
+ self._log.error(msg)
+ raise UnknownAgentTypeError(msg)
+
+ # Check to see if the plugin was already instantiated
+ if cap_name in self._plugin_instances:
+ self._log.debug("Config agent nsm plugin {} already instantiated. " \
+ "Using existing.". format(cap_name))
+ else:
+ # Otherwise, instantiate a new plugin using the config agent account
+ self._log.debug("Instantiting new config agent using class: %s", cap_inst)
+ new_instance = cap_inst(self._dts, self._log, self._loop, config_agent)
+ self._plugin_instances[cap_name] = new_instance
+
+ # TODO (pjoseph): See why this was added, as this deletes the
+ # Rift plugin account when Juju account is added
+ # if self._default_account_added:
+ # # If the user has provided a config account, chuck the default one.
+ # if self.DEFAULT_CAP_TYPE in self._plugin_instances:
+ # del self._plugin_instances[self.DEFAULT_CAP_TYPE]
+
+ def _on_config_agent_delete(self, config_agent):
+ self._log.debug("Got nsm plugin config agent delete, account: %s, type: %s",
+ config_agent.name, config_agent.account_type)
+ cap_name = config_agent.account_type
+ if cap_name in self._plugin_instances:
+ self._log.debug("Config agent nsm plugin exists, deleting it.")
+ del self._plugin_instances[cap_name]
+ else:
+ self._log.error("Error deleting - Config Agent nsm plugin %s does not exist.", cap_name)
+
+
+ @asyncio.coroutine
+ def register(self):
+ self._log.debug("Registering for config agent nsm plugin manager")
+ yield from self._config_handler.register()
+
+ account = rwcfg_agent.ConfigAgentAccount()
+ account.account_type = DEFAULT_CAP_TYPE
+ account.name = "RiftCA"
+ self._on_config_agent(account)
+ self._default_account_added = True
+
+ # Also grab any account already configured
+ config_agents = yield from self._ConfigManagerConfig.cmdts_obj.get_config_agents(name=None)
+ for account in config_agents:
+ self._on_config_agent(account)
+
+ def set_config_agent(self, nsr, vnfr, method):
+ if method == 'juju':
+ agent_type = 'juju'
+ elif method in ['netconf', 'script']:
+ agent_type = DEFAULT_CAP_TYPE
+ else:
+ msg = "Unsupported configuration method ({}) for VNF:{}/{}". \
+ format(method, nsr.name, vnfr.name)
+ self._log.error(msg)
+ raise UnknownAgentTypeError(msg)
+
+ try:
+ acc_map = nsr.nsr_cfg_msg.vnf_cloud_account_map
+ except AttributeError:
+ self._log.debug("Did not find cloud account map for NS {}".
+ format(nsr.name))
+ acc_map = []
+
+ for vnfd in acc_map:
+ if vnfd.config_agent_account is not None:
+ if vnfd.member_vnf_index_ref == vnfr.vnfr_msg.member_index:
+ for agent in self._plugin_instances:
+ # Find the plugin with the same name
+ if agent == vnfd.config_agent_account:
+ # Check if the types are same
+ if self._plugin_instances[agent].agent_type != agent_type:
+ msg = "VNF {} specified config agent {} is not of type {}". \
+ format(vnfr.name, agent, agent_type)
+ self._log.error(msg)
+ raise ConfigAgentVnfrTypeError(msg)
+
+ self._plugin_instances[agent].add_vnfr_managed(vnfr)
+ self._log.debug("Added vnfr {} as config plugin {} managed".
+ format(vnfr.name, agent))
+ return
+
+ # If no config agent specified for the VNF, use the
+ # first available of the same type
+ for agent in self._plugin_instances:
+ if self._plugin_instances[agent].agent_type == agent_type:
+ self._plugin_instances[agent].add_vnfr_managed(vnfr)
+ self._log.debug("Added vnfr {} as config plugin {} managed".
+ format(vnfr.name, agent))
+ return
+
+ msg = "Error finding config agent of type {} for VNF {}". \
+ format(agent_type, vnfr.name)
+ self._log.error(msg)
+ raise ConfigAgentVnfrAddError(msg)
diff --git a/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/rwconman_config.py b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/rwconman_config.py
new file mode 100644
index 0000000..4848e9e
--- /dev/null
+++ b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/rwconman_config.py
@@ -0,0 +1,1472 @@
+
+#
+# Copyright 2016 RIFT.IO Inc
+#
+# 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 asyncio
+import os
+import stat
+import subprocess
+import sys
+import tempfile
+import yaml
+
+from gi.repository import (
+ RwDts as rwdts,
+ RwConmanYang as conmanY,
+ ProtobufC,
+)
+
+import rift.tasklets
+
+from . import rwconman_conagent as conagent
+from . import RiftCM_rpc
+from . import riftcm_config_plugin
+
+if sys.version_info < (3, 4, 4):
+ asyncio.ensure_future = asyncio.async
+
+def get_vnf_unique_name(nsr_name, vnfr_short_name, member_vnf_index):
+ return "{}.{}.{}".format(nsr_name, vnfr_short_name, member_vnf_index)
+
+class ConmanConfigError(Exception):
+ pass
+
+
+class InitialConfigError(ConmanConfigError):
+ pass
+
+
+def log_this_vnf(vnf_cfg):
+ log_vnf = ""
+ used_item_list = ['nsr_name', 'vnfr_name', 'member_vnf_index', 'mgmt_ip_address']
+ for item in used_item_list:
+ if item in vnf_cfg:
+ if item == 'mgmt_ip_address':
+ log_vnf += "({})".format(vnf_cfg[item])
+ else:
+ log_vnf += "{}/".format(vnf_cfg[item])
+ return log_vnf
+
+class PretendNsm(object):
+ def __init__(self, dts, log, loop, parent):
+ self._dts = dts
+ self._log = log
+ self._loop = loop
+ self._parent = parent
+ self._nsrs = {}
+ self._nsr_dict = parent._nsr_dict
+ self._config_agent_plugins = []
+ self._nsd_msg = {}
+
+ @property
+ def nsrs(self):
+ # Expensive, instead use get_nsr, if you know id.
+ self._nsrs = {}
+ # Update the list of nsrs (agent nsr)
+ for id, nsr_obj in self._nsr_dict.items():
+ self._nsrs[id] = nsr_obj.agent_nsr
+ return self._nsrs
+
+ def get_nsr(self, nsr_id):
+ if nsr_id in self._nsr_dict:
+ nsr_obj = self._nsr_dict[nsr_id]
+ return nsr_obj._nsr
+ return None
+
+ def get_vnfr_msg(self, vnfr_id, nsr_id=None):
+ self._log.debug("get_vnfr_msg(vnfr=%s, nsr=%s)",
+ vnfr_id, nsr_id)
+ found = False
+ if nsr_id:
+ if nsr_id in self._nsr_dict:
+ nsr_obj = self._nsr_dict[nsr_id]
+ if vnfr_id in nsr_obj._vnfr_dict:
+ found = True
+ else:
+ for nsr_obj in self._nsr_dict.values():
+ if vnfr_id in nsr_obj._vnfr_dict:
+ # Found it
+ found = True
+ break
+ if found:
+ vnf_cfg = nsr_obj._vnfr_dict[vnfr_id]['vnf_cfg']
+ return vnf_cfg['agent_vnfr'].vnfr_msg
+ else:
+ return None
+
+ @asyncio.coroutine
+ def get_nsd(self, nsr_id):
+ if nsr_id not in self._nsd_msg:
+ nsr_config = yield from self._parent.cmdts_obj.get_nsr_config(nsr_id)
+ self._nsd_msg[nsr_id] = nsr_config.nsd
+ return self._nsd_msg[nsr_id]
+
+ @property
+ def config_agent_plugins(self):
+ self._config_agent_plugins = []
+ for agent in self._parent._config_agent_mgr._plugin_instances.values():
+ self._config_agent_plugins.append(agent)
+ return self._config_agent_plugins
+
+class ConfigManagerConfig(object):
+ def __init__(self, dts, log, loop, parent):
+ self._dts = dts
+ self._log = log
+ self._loop = loop
+ self._parent = parent
+ self._nsr_dict = {}
+ self.pending_cfg = {}
+ self.terminate_cfg = {}
+ self.pending_tasks = [] # User for NSRid get retry
+ # (mainly excercised at restart case)
+ self._config_xpath = "C,/cm-config"
+ self._opdata_xpath = "D,/rw-conman:cm-state"
+
+ self.cm_config = conmanY.SoConfig()
+ # RO specific configuration
+ self.ro_config = {}
+ for key in self.cm_config.ro_endpoint.fields:
+ self.ro_config[key] = None
+
+ # Initialize cm-state
+ self.cm_state = {}
+ self.cm_state['cm_nsr'] = []
+ self.cm_state['states'] = "Initialized"
+
+ # Initialize objects to register
+ self.cmdts_obj = ConfigManagerDTS(self._log, self._loop, self, self._dts)
+ self._config_agent_mgr = conagent.RiftCMConfigAgent(
+ self._dts,
+ self._log,
+ self._loop,
+ self,
+ )
+ self.reg_handles = [
+ self.cmdts_obj,
+ self._config_agent_mgr,
+ RiftCM_rpc.RiftCMRPCHandler(self._dts, self._log, self._loop,
+ PretendNsm(
+ self._dts, self._log, self._loop, self)),
+ ]
+
+ def is_nsr_valid(self, nsr_id):
+ if nsr_id in self._nsr_dict:
+ return True
+ return False
+
+ def add_to_pending_tasks(self, task):
+ if self.pending_tasks:
+ for p_task in self.pending_tasks:
+ if p_task['nsrid'] == task['nsrid']:
+ # Already queued
+ return
+ try:
+ self.pending_tasks.append(task)
+ self._log.debug("add_to_pending_tasks (nsrid:%s)",
+ task['nsrid'])
+ if len(self.pending_tasks) == 1:
+ self._loop.create_task(self.ConfigManagerConfig_pending_loop())
+ # TBD - change to info level
+ self._log.debug("Started pending_loop!")
+ except Exception as e:
+ self._log.error("Failed adding to pending tasks (%s)", str(e))
+
+ def del_from_pending_tasks(self, task):
+ try:
+ self.pending_tasks.remove(task)
+ except Exception as e:
+ self._log.error("Failed removing from pending tasks (%s)", str(e))
+
+ @asyncio.coroutine
+ def ConfigManagerConfig_pending_loop(self):
+ loop_sleep = 2
+ while True:
+ yield from asyncio.sleep(loop_sleep, loop=self._loop)
+ """
+ This pending task queue is ordred by events,
+ must finish previous task successfully to be able to go on to the next task
+ """
+ if self.pending_tasks:
+ self._log.debug("self.pending_tasks len=%s", len(self.pending_tasks))
+ task = self.pending_tasks[0]
+ done = False
+ if 'nsrid' in task:
+ nsrid = task['nsrid']
+ self._log.debug("Will execute pending task for NSR id(%s)", nsrid)
+ try:
+ # Try to configure this NSR
+ task['retries'] -= 1
+ done = yield from self.config_NSR(nsrid)
+ self._log.info("self.config_NSR status=%s", done)
+
+ except Exception as e:
+ self._log.error("Failed(%s) configuring NSR(%s)," \
+ "retries remained:%d!",
+ str(e), nsrid, task['retries'])
+ finally:
+ self.pending_tasks.remove(task)
+
+ if done:
+ self._log.debug("Finished pending task NSR id(%s):", nsrid)
+ else:
+ self._log.error("Failed configuring NSR(%s), retries remained:%d!",
+ nsrid, task['retries'])
+
+ # Failed, re-insert (append at the end)
+ # this failed task to be retried later
+ # If any retries remained.
+ if task['retries']:
+ self.pending_tasks.append(task)
+ else:
+ self._log.debug("Stopped pending_loop!")
+ break
+
+ @asyncio.coroutine
+ def register(self):
+ yield from self.register_cm_state_opdata()
+
+ # Initialize all handles that needs to be registered
+ for reg in self.reg_handles:
+ yield from reg.register()
+
+ @asyncio.coroutine
+ def register_cm_state_opdata(self):
+
+ def state_to_string(state):
+ state_dict = {
+ conmanY.RecordState.INIT : "init",
+ conmanY.RecordState.RECEIVED : "received",
+ conmanY.RecordState.CFG_PROCESS : "cfg_process",
+ conmanY.RecordState.CFG_PROCESS_FAILED : "cfg_process_failed",
+ conmanY.RecordState.CFG_SCHED : "cfg_sched",
+ conmanY.RecordState.CFG_DELAY : "cfg_delay",
+ conmanY.RecordState.CONNECTING : "connecting",
+ conmanY.RecordState.FAILED_CONNECTION : "failed_connection",
+ conmanY.RecordState.NETCONF_CONNECTED : "netconf_connected",
+ conmanY.RecordState.NETCONF_SSH_CONNECTED : "netconf_ssh_connected",
+ conmanY.RecordState.RESTCONF_CONNECTED : "restconf_connected",
+ conmanY.RecordState.CFG_SEND : "cfg_send",
+ conmanY.RecordState.CFG_FAILED : "cfg_failed",
+ conmanY.RecordState.READY_NO_CFG : "ready_no_cfg",
+ conmanY.RecordState.READY : "ready",
+ }
+ return state_dict[state]
+
+ @asyncio.coroutine
+ def on_prepare(xact_info, action, ks_path, msg):
+
+ self._log.debug("Received cm-state: msg=%s, action=%s", msg, action)
+
+ if action == rwdts.QueryAction.READ:
+ show_output = conmanY.CmOpdata()
+ show_output.from_dict(self.cm_state)
+ self._log.debug("Responding to SHOW cm-state: %s", self.cm_state)
+ xact_info.respond_xpath(rwdts.XactRspCode.ACK,
+ xpath=self._opdata_xpath,
+ msg=show_output)
+ else:
+ xact_info.respond_xpath(rwdts.XactRspCode.ACK)
+
+ self._log.info("Registering for cm-opdata xpath: %s",
+ self._opdata_xpath)
+
+ try:
+ handler=rift.tasklets.DTS.RegistrationHandler(on_prepare=on_prepare)
+ yield from self._dts.register(xpath=self._opdata_xpath,
+ handler=handler,
+ flags=rwdts.Flag.PUBLISHER)
+ self._log.info("Successfully registered for opdata(%s)", self._opdata_xpath)
+ except Exception as e:
+ self._log.error("Failed to register for opdata as (%s)", e)
+
+ @asyncio.coroutine
+ def process_nsd_vnf_configuration(self, nsr_obj, vnfr):
+
+ def get_config_method(vnf_config):
+ cfg_types = ['netconf', 'juju', 'script']
+ for method in cfg_types:
+ if method in vnf_config:
+ return method
+ return None
+
+ def get_cfg_file_extension(method, configuration_options):
+ ext_dict = {
+ "netconf" : "xml",
+ "script" : {
+ "bash" : "sh",
+ "expect" : "exp",
+ },
+ "juju" : "yml"
+ }
+
+ if method == "netconf":
+ return ext_dict[method]
+ elif method == "script":
+ return ext_dict[method][configuration_options['script_type']]
+ elif method == "juju":
+ return ext_dict[method]
+ else:
+ return "cfg"
+
+ # This is how the YAML file should look like,
+ # This routine will be called for each VNF, so keep appending the file.
+ # priority order is determined by the number,
+ # hence no need to generate the file in that order. A dictionary will be
+ # used that will take care of the order by number.
+ '''
+ 1 : <== This is priority
+ name : trafsink_vnfd
+ member_vnf_index : 2
+ configuration_delay : 120
+ configuration_type : netconf
+ configuration_options :
+ username : admin
+ password : admin
+ port : 2022
+ target : running
+ 2 :
+ name : trafgen_vnfd
+ member_vnf_index : 1
+ configuration_delay : 0
+ configuration_type : netconf
+ configuration_options :
+ username : admin
+ password : admin
+ port : 2022
+ target : running
+ '''
+
+ # Save some parameters needed as short cuts in flat structure (Also generated)
+ vnf_cfg = vnfr['vnf_cfg']
+ # Prepare unique name for this VNF
+ vnf_cfg['vnf_unique_name'] = get_vnf_unique_name(
+ vnf_cfg['nsr_name'], vnfr['short_name'], vnfr['member_vnf_index_ref'])
+
+ nsr_obj.cfg_path_prefix = '{}/{}_{}'.format(
+ nsr_obj.this_nsr_dir, vnfr['short_name'], vnfr['member_vnf_index_ref'])
+ nsr_vnfr = '{}/{}_{}'.format(
+ vnf_cfg['nsr_name'], vnfr['short_name'], vnfr['member_vnf_index_ref'])
+
+ # Get vnf_configuration from vnfr
+ vnf_config = vnfr['vnf_configuration']
+
+ self._log.debug("vnf_configuration = %s", vnf_config)
+
+ # Create priority dictionary
+ cfg_priority_order = 0
+ if ('config_attributes' in vnf_config and
+ 'config_priority' in vnf_config['config_attributes']):
+ cfg_priority_order = vnf_config['config_attributes']['config_priority']
+
+ if cfg_priority_order not in nsr_obj.nsr_cfg_config_attributes_dict:
+ # No VNFR with this priority yet, initialize the list
+ nsr_obj.nsr_cfg_config_attributes_dict[cfg_priority_order] = []
+
+ method = get_config_method(vnf_config)
+ if method is not None:
+ # Create all sub dictionaries first
+ config_priority = {
+ 'id' : vnfr['id'],
+ 'name' : vnfr['short_name'],
+ 'member_vnf_index' : vnfr['member_vnf_index_ref'],
+ }
+
+ if 'config_delay' in vnf_config['config_attributes']:
+ config_priority['configuration_delay'] = vnf_config['config_attributes']['config_delay']
+ vnf_cfg['config_delay'] = config_priority['configuration_delay']
+
+ configuration_options = {}
+ self._log.debug("config method=%s", method)
+ config_priority['configuration_type'] = method
+ vnf_cfg['config_method'] = method
+
+ # Set config agent based on method
+ self._config_agent_mgr.set_config_agent(
+ nsr_obj.agent_nsr, vnf_cfg['agent_vnfr'], method)
+
+ cfg_opt_list = [
+ 'port', 'target', 'script_type', 'ip_address', 'user', 'secret',
+ ]
+ for cfg_opt in cfg_opt_list:
+ if cfg_opt in vnf_config[method]:
+ configuration_options[cfg_opt] = vnf_config[method][cfg_opt]
+ vnf_cfg[cfg_opt] = configuration_options[cfg_opt]
+
+ cfg_opt_list = ['mgmt_ip_address', 'username', 'password']
+ for cfg_opt in cfg_opt_list:
+ if cfg_opt in vnf_config['config_access']:
+ configuration_options[cfg_opt] = vnf_config['config_access'][cfg_opt]
+ vnf_cfg[cfg_opt] = configuration_options[cfg_opt]
+
+ # Add to the cp_dict
+ vnf_cp_dict = nsr_obj._cp_dict[vnfr['member_vnf_index_ref']]
+ vnf_cp_dict['rw_mgmt_ip'] = vnf_cfg['mgmt_ip_address']
+ vnf_cp_dict['rw_username'] = vnf_cfg['username']
+ vnf_cp_dict['rw_password'] = vnf_cfg['password']
+
+
+ # TBD - see if we can neatly include the config in "config_attributes" file, no need though
+ #config_priority['config_template'] = vnf_config['config_template']
+ # Create config file
+ vnf_cfg['juju_script'] = os.path.join(self._parent.cfg_dir, 'juju_if.py')
+
+ if 'config_template' in vnf_config:
+ vnf_cfg['cfg_template'] = '{}_{}_template.cfg'.format(nsr_obj.cfg_path_prefix, config_priority['configuration_type'])
+ vnf_cfg['cfg_file'] = '{}.{}'.format(nsr_obj.cfg_path_prefix, get_cfg_file_extension(method, configuration_options))
+ vnf_cfg['xlate_script'] = os.path.join(self._parent.cfg_dir, 'xlate_cfg.py')
+ try:
+ # Now write this template into file
+ with open(vnf_cfg['cfg_template'], "w") as cf:
+ cf.write(vnf_config['config_template'])
+ except Exception as e:
+ self._log.error("Processing NSD, failed to generate configuration template : %s (Error : %s)",
+ vnf_config['config_template'], str(e))
+ raise
+
+ self._log.debug("VNF endpoint so far: %s", vnf_cfg)
+
+ # Populate filled up dictionary
+ config_priority['configuration_options'] = configuration_options
+ nsr_obj.nsr_cfg_config_attributes_dict[cfg_priority_order].append(config_priority)
+ nsr_obj.num_vnfs_to_cfg += 1
+ nsr_obj._vnfr_dict[vnf_cfg['vnf_unique_name']] = vnfr
+ nsr_obj._vnfr_dict[vnfr['id']] = vnfr
+
+ self._log.debug("VNF:(%s) config_attributes = %s",
+ log_this_vnf(vnfr['vnf_cfg']),
+ nsr_obj.nsr_cfg_config_attributes_dict)
+ else:
+ self._log.info("VNF:(%s) is not to be configured by Configuration Manager!",
+ log_this_vnf(vnfr['vnf_cfg']))
+ yield from nsr_obj.update_vnf_cm_state(vnfr, conmanY.RecordState.READY_NO_CFG)
+
+ # Update the cm-state
+ nsr_obj.populate_vm_state_from_vnf_cfg()
+
+ @asyncio.coroutine
+ def config_NSR(self, id):
+
+ def my_yaml_dump(config_attributes_dict, yf):
+
+ yaml_dict = dict(sorted(config_attributes_dict.items()))
+ yf.write(yaml.dump(yaml_dict, default_flow_style=False))
+
+ nsr_dict = self._nsr_dict
+ self._log.info("Configure NSR, id = %s", id)
+
+ #####################TBD###########################
+ # yield from self._config_agent_mgr.invoke_config_agent_plugins('notify_create_nsr', self.id, self._nsd)
+ # yield from self._config_agent_mgr.invoke_config_agent_plugins('notify_nsr_active', self.id, self._vnfrs)
+
+ try:
+ if id not in nsr_dict:
+ nsr_obj = ConfigManagerNSR(self._log, self._loop, self, id)
+ nsr_dict[id] = nsr_obj
+ else:
+ self._log.info("NSR(%s) is already initialized!", id)
+ nsr_obj = nsr_dict[id]
+ except Exception as e:
+ self._log.error("Failed creating NSR object for (%s) as (%s)", id, str(e))
+ raise
+
+ # Try to configure this NSR only if not already processed
+ if nsr_obj.cm_nsr['state'] != nsr_obj.state_to_string(conmanY.RecordState.INIT):
+ self._log.debug("NSR(%s) is already processed, state=%s",
+ nsr_obj.nsr_name, nsr_obj.cm_nsr['state'])
+ yield from nsr_obj.publish_cm_state()
+ return True
+
+ cmdts_obj = self.cmdts_obj
+ try:
+ # Fetch NSR
+ nsr = yield from cmdts_obj.get_nsr(id)
+ self._log.debug("Full NSR : %s", nsr)
+ if nsr['operational_status'] != "running":
+ self._log.info("NSR(%s) is not ready yet!", nsr['nsd_name_ref'])
+ return False
+ self._nsr = nsr
+
+ # Create Agent NSR class
+ nsr_config = yield from cmdts_obj.get_nsr_config(id)
+ self._log.debug("NSR {} config: {}".format(id, nsr_config))
+ nsr_obj.agent_nsr = riftcm_config_plugin.RiftCMnsr(nsr, nsr_config)
+
+ try:
+ yield from nsr_obj.update_ns_cm_state(conmanY.RecordState.RECEIVED)
+
+ # Parse NSR
+ if nsr is not None:
+ nsr_obj.set_nsr_name(nsr['nsd_name_ref'])
+ nsr_dir = os.path.join(self._parent.cfg_dir, nsr_obj.nsr_name)
+ self._log.info("Checking NS config directory: %s", nsr_dir)
+ if not os.path.isdir(nsr_dir):
+ os.makedirs(nsr_dir)
+ # self._log.critical("NS %s is not to be configured by Service Orchestrator!", nsr_obj.nsr_name)
+ # yield from nsr_obj.update_ns_cm_state(conmanY.RecordState.READY_NO_CFG)
+ # return
+
+ nsr_obj.set_config_dir(self)
+
+ for const_vnfr in nsr['constituent_vnfr_ref']:
+ self._log.debug("Fetching VNFR (%s)", const_vnfr['vnfr_id'])
+ vnfr_msg = yield from cmdts_obj.get_vnfr(const_vnfr['vnfr_id'])
+ if vnfr_msg:
+ vnfr = vnfr_msg.as_dict()
+ self._log.info("create VNF:{}/{}".format(nsr_obj.nsr_name, vnfr['short_name']))
+ agent_vnfr = yield from nsr_obj.add_vnfr(vnfr, vnfr_msg)
+
+ # Preserve order, self.process_nsd_vnf_configuration()
+ # sets up the config agent based on the method
+ yield from self.process_nsd_vnf_configuration(nsr_obj, vnfr)
+ yield from self._config_agent_mgr.invoke_config_agent_plugins(
+ 'notify_create_vnfr',
+ nsr_obj.agent_nsr,
+ agent_vnfr)
+
+ #####################TBD###########################
+ # self._log.debug("VNF active. Apply initial config for vnfr {}".format(vnfr.name))
+ # yield from self._config_agent_mgr.invoke_config_agent_plugins('apply_initial_config',
+ # vnfr.id, vnfr)
+ # yield from self._config_agent_mgr.invoke_config_agent_plugins('notify_terminate_vnf', self.id, vnfr)
+
+ except Exception as e:
+ self._log.error("Failed processing NSR (%s) as (%s)", nsr_obj.nsr_name, str(e))
+ self._log.exception(e)
+ yield from nsr_obj.update_ns_cm_state(conmanY.RecordState.CFG_PROCESS_FAILED)
+ raise
+
+ try:
+ # Generate config_config_attributes.yaml (For debug reference)
+ with open(nsr_obj.config_attributes_file, "w") as yf:
+ my_yaml_dump(nsr_obj.nsr_cfg_config_attributes_dict, yf)
+ except Exception as e:
+ self._log.error("NS:(%s) failed to write config attributes file as (%s)", nsr_obj.nsr_name, str(e))
+
+ try:
+ # Generate nsr_xlate_dict.yaml (For debug reference)
+ with open(nsr_obj.xlate_dict_file, "w") as yf:
+ yf.write(yaml.dump(nsr_obj._cp_dict, default_flow_style=False))
+ except Exception as e:
+ self._log.error("NS:(%s) failed to write nsr xlate tags file as (%s)", nsr_obj.nsr_name, str(e))
+
+ self._log.debug("Starting to configure each VNF")
+
+ # Check if this NS has input parametrs
+ self._log.info("Checking NS configuration order: %s", nsr_obj.config_attributes_file)
+
+ if os.path.exists(nsr_obj.config_attributes_file):
+ # Apply configuration is specified order
+ try:
+ # Go in loop to configure by specified order
+ self._log.info("Using Dynamic configuration input parametrs for NS: %s", nsr_obj.nsr_name)
+
+ # cfg_delay = nsr_obj.nsr_cfg_config_attributes_dict['configuration_delay']
+ # if cfg_delay:
+ # self._log.info("Applying configuration delay for NS (%s) ; %d seconds",
+ # nsr_obj.nsr_name, cfg_delay)
+ # yield from asyncio.sleep(cfg_delay, loop=self._loop)
+
+ for config_attributes_dict in nsr_obj.nsr_cfg_config_attributes_dict.values():
+ # Iterate through each priority level
+ for vnf_config_attributes_dict in config_attributes_dict:
+ # Iterate through each vnfr at this priority level
+
+ # Make up vnf_unique_name with vnfd name and member index
+ #vnfr_name = "{}.{}".format(nsr_obj.nsr_name, vnf_config_attributes_dict['name'])
+ vnf_unique_name = get_vnf_unique_name(
+ nsr_obj.nsr_name,
+ vnf_config_attributes_dict['name'],
+ str(vnf_config_attributes_dict['member_vnf_index']),
+ )
+ self._log.info("NS (%s) : VNF (%s) - Processing configuration attributes",
+ nsr_obj.nsr_name, vnf_unique_name)
+
+ # Find vnfr for this vnf_unique_name
+ if vnf_unique_name not in nsr_obj._vnfr_dict:
+ self._log.error("NS (%s) - Can not find VNF to be configured: %s", nsr_obj.nsr_name, vnf_unique_name)
+ else:
+ # Save this unique VNF's config input parameters
+ nsr_obj.vnf_config_attributes_dict[vnf_unique_name] = vnf_config_attributes_dict
+ nsr_obj.ConfigVNF(nsr_obj._vnfr_dict[vnf_unique_name])
+
+ # Now add the entire NS to the pending config list.
+ self._log.info("Scheduling NSR:{} configuration".format(nsr_obj.nsr_name))
+ self._parent.add_to_pending(nsr_obj)
+ self._parent.add_nsr_obj(nsr_obj)
+
+ except Exception as e:
+ self._log.error("Failed processing input parameters for NS (%s) as %s", nsr_obj.nsr_name, str(e))
+ raise
+ else:
+ self._log.error("No configuration input parameters for NSR (%s)", nsr_obj.nsr_name)
+
+ except Exception as e:
+ self._log.error("Failed to configure NS (%s) as (%s)", nsr_obj.nsr_name, str(e))
+ yield from nsr_obj.update_ns_cm_state(conmanY.RecordState.CFG_PROCESS_FAILED)
+ raise
+
+ return True
+
+ @asyncio.coroutine
+ def terminate_NSR(self, id):
+ nsr_dict = self._nsr_dict
+ if id not in nsr_dict:
+ self._log.error("NSR(%s) does not exist!", id)
+ return
+ else:
+ # Remove this NSR if we have it on pending task list
+ for task in self.pending_tasks:
+ if task['nsrid'] == id:
+ self.del_from_pending_tasks(task)
+
+ # Remove this object from global list
+ nsr_obj = nsr_dict.pop(id, None)
+
+ # Remove this NS cm-state from global status list
+ self.cm_state['cm_nsr'].remove(nsr_obj.cm_nsr)
+
+ # Also remove any scheduled configuration event
+ for nsr_obj_p in self._parent.pending_cfg:
+ if nsr_obj_p == nsr_obj:
+ assert id == nsr_obj_p._nsr_id
+ #self._parent.pending_cfg.remove(nsr_obj_p)
+ # Mark this as being deleted so we do not try to configure it if we are in cfg_delay (will wake up and continue to process otherwise)
+ nsr_obj_p.being_deleted = True
+ self._log.info("Removed scheduled configuration for NSR(%s)", nsr_obj.nsr_name)
+
+ self._parent.remove_nsr_obj(id)
+
+ # Call Config Agent to clean up for each VNF
+ for agent_vnfr in nsr_obj.agent_nsr.vnfrs:
+ yield from self._config_agent_mgr.invoke_config_agent_plugins(
+ 'notify_terminate_vnfr',
+ nsr_obj.agent_nsr,
+ agent_vnfr)
+
+ # publish delete cm-state (cm-nsr)
+ yield from nsr_obj.delete_cm_nsr()
+
+ #####################TBD###########################
+ # yield from self._config_agent_mgr.invoke_config_agent_plugins('notify_terminate_ns', self.id)
+
+ self._log.info("NSR(%s/%s) is deleted", nsr_obj.nsr_name, id)
+
+ @asyncio.coroutine
+ def process_ns_initial_config(self, nsr_obj):
+ '''Apply the initial-config-primitives specified in NSD'''
+
+ def get_input_file(parameters):
+ inp = {}
+
+ # Add NSR name to file
+ inp['nsr_name'] = nsr_obj.nsr_name
+
+ # TODO (pjoseph): Add config agents, we need to identify which all
+ # config agents are required from this NS and provide only those
+ inp['config-agent'] = {}
+
+ # Add parameters for initial config
+ inp['parameter'] = {}
+ for parameter in parameters:
+ try:
+ inp['parameter'][parameter['name']] = parameter['value']
+ except KeyError as e:
+ self._log.info("NSR {} initial config parameter {} with no value: {}".
+ format(nsr_obj.nsr_name, parameter, e))
+
+
+ # Add vnfrs specific data
+ inp['vnfr'] = {}
+ for vnfr in nsr_obj.vnfrs:
+ v = {}
+
+ v['name'] = vnfr['name']
+ v['mgmt_ip_address'] = vnfr['vnf_cfg']['mgmt_ip_address']
+ v['mgmt_port'] = vnfr['vnf_cfg']['port']
+
+ if 'dashboard_url' in vnfr:
+ v['dashboard_url'] = vnfr['dashboard_url']
+
+ if 'connection_point' in vnfr:
+ v['connection_point'] = []
+ for cp in vnfr['connection_point']:
+ v['connection_point'].append(
+ {
+ 'name': cp['name'],
+ 'ip_address': cp['ip_address'],
+ }
+ )
+
+ v['vdur'] = []
+ vdu_data = [(vdu['name'], vdu['management_ip'], vdu['vm_management_ip'], vdu['id'])
+ for vdu in vnfr['vdur']]
+
+ for data in vdu_data:
+ data = dict(zip(['name', 'management_ip', 'vm_management_ip', 'id'] , data))
+ v['vdur'].append(data)
+
+ inp['vnfr'][vnfr['member_vnf_index_ref']] = v
+
+ self._log.debug("Input data for NSR {}: {}".
+ format(nsr_obj.nsr_name, inp))
+
+ # Convert to YAML string
+ yaml_string = yaml.dump(inp, default_flow_style=False)
+
+ # Write the inputs as yaml file
+ tmp_file = None
+ with tempfile.NamedTemporaryFile(delete=False) as tmp_file:
+ tmp_file.write(yaml_string.encode("UTF-8"))
+ self._log.debug("Input file created for NSR {}: {}".
+ format(nsr_obj.nsr_name, tmp_file.name))
+
+ return tmp_file.name
+
+ def get_script_file(script_name, nsd_name, nsd_id):
+ # Get the full path to the script
+ script = ''
+ # If script name starts with /, assume it is full path
+ if script_name[0] == '/':
+ # The script has full path, use as is
+ script = script_name
+ else:
+ script = os.path.join(os.environ['RIFT_ARTIFACTS'],
+ 'launchpad/packages/nsd',
+ nsd_id,
+ nsd_name,
+ 'scripts',
+ script_name)
+ self._log.debug("Checking for script at %s", script)
+ if not os.path.exists(script):
+ self._log.debug("Did not find script %s", script)
+ script = os.path.join(os.environ['RIFT_INSTALL'],
+ 'usr/bin',
+ script_name)
+
+ # Seen cases in jenkins, where the script execution fails
+ # with permission denied. Setting the permission on script
+ # to make sure it has execute permission
+ perm = os.stat(script).st_mode
+ if not (perm & stat.S_IXUSR):
+ self._log.warn("NSR {} initial config script {} " \
+ "without execute permission: {}".
+ format(nsr_id, script, perm))
+ os.chmod(script, perm | stat.S_IXUSR)
+ return script
+
+ nsr_id = nsr_obj.nsr_id
+ nsr_name = nsr_obj.nsr_name
+ self._log.debug("Apply initial config for NSR {}({})".
+ format(nsr_name, nsr_id))
+
+ # Fetch NSR
+ nsr = yield from self.cmdts_obj.get_nsr(nsr_id)
+ if nsr is not None:
+ nsd = yield from self.cmdts_obj.get_nsd(nsr_id)
+
+ try:
+ # Check if initial config is present
+ # TODO (pjoseph): Sort based on seq
+ for conf in nsr['initial_config_primitive']:
+ self._log.debug("Parameter conf: {}".
+ format(conf))
+
+ parameters = []
+ try:
+ parameters = conf['parameter']
+ except Exception as e:
+ self._log.debug("Parameter conf: {}, e: {}".
+ format(conf, e))
+ pass
+
+ inp_file = get_input_file(parameters)
+
+ script = get_script_file(conf['user_defined_script'],
+ nsd.name,
+ nsd.id)
+
+ cmd = "{0} {1}".format(script, inp_file)
+ self._log.debug("Running the CMD: {}".format(cmd))
+
+ process = yield from asyncio. \
+ create_subprocess_shell(cmd, loop=self._loop)
+ yield from process.wait()
+ if process.returncode:
+ msg = "NSR {} initial config using {} failed with {}". \
+ format(nsr_name, script, process.returncode)
+ self._log.error(msg)
+ raise InitialConfigError(msg)
+ else:
+ os.remove(inp_file)
+
+ except KeyError as e:
+ self._log.debug("Did not find initial config {}".
+ format(e))
+
+
+class ConfigManagerNSR(object):
+ def __init__(self, log, loop, parent, id):
+ self._log = log
+ self._loop = loop
+ self._rwcal = None
+ self._vnfr_dict = {}
+ self._cp_dict = {}
+ self._nsr_id = id
+ self._parent = parent
+ self._log.info("Instantiated NSR entry for id=%s", id)
+ self.nsr_cfg_config_attributes_dict = {}
+ self.vnf_config_attributes_dict = {}
+ self.num_vnfs_to_cfg = 0
+ self._vnfr_list = []
+ self.vnf_cfg_list = []
+ self.this_nsr_dir = None
+ self.being_deleted = False
+ self.dts_obj = self._parent.cmdts_obj
+
+ # Initialize cm-state for this NS
+ self.cm_nsr = {}
+ self.cm_nsr['cm_vnfr'] = []
+ self.cm_nsr['id'] = id
+ self.cm_nsr['state'] = self.state_to_string(conmanY.RecordState.INIT)
+ self.cm_nsr['state_details'] = None
+
+ self.set_nsr_name('Not Set')
+
+ # Add this NSR cm-state object to global cm-state
+ parent.cm_state['cm_nsr'].append(self.cm_nsr)
+
+ # Place holders for NSR & VNFR classes
+ self.agent_nsr = None
+
+ @property
+ def nsr_opdata_xpath(self):
+ ''' Returns full xpath for this NSR cm-state opdata '''
+ return(
+ "D,/rw-conman:cm-state" +
+ "/rw-conman:cm-nsr[rw-conman:id='{}']"
+ ).format(self._nsr_id)
+
+ @property
+ def vnfrs(self):
+ return self._vnfr_list
+
+ @property
+ def parent(self):
+ return self._parent
+
+ @property
+ def nsr_id(self):
+ return self._nsr_id
+
+ @asyncio.coroutine
+ def publish_cm_state(self):
+ ''' This function publishes cm_state for this NSR '''
+
+ cm_state = conmanY.CmOpdata()
+ cm_state_nsr = cm_state.cm_nsr.add()
+ cm_state_nsr.from_dict(self.cm_nsr)
+ #with self._dts.transaction() as xact:
+ yield from self.dts_obj.update(self.nsr_opdata_xpath, cm_state_nsr)
+ self._log.info("Published cm-state with xpath %s and nsr %s",
+ self.nsr_opdata_xpath,
+ cm_state_nsr)
+
+ @asyncio.coroutine
+ def delete_cm_nsr(self):
+ ''' This function publishes cm_state for this NSR '''
+
+ yield from self.dts_obj.delete(self.nsr_opdata_xpath)
+ self._log.info("Deleted cm-nsr with xpath %s",
+ self.nsr_opdata_xpath)
+
+ def set_nsr_name(self, name):
+ self.nsr_name = name
+ self.cm_nsr['name'] = name
+
+ def set_config_dir(self, caller):
+ self.this_nsr_dir = os.path.join(
+ caller._parent.cfg_dir, self.nsr_name, caller._nsr['name_ref'])
+ if not os.path.exists(self.this_nsr_dir):
+ os.makedirs(self.this_nsr_dir)
+ self._log.debug("NSR:(%s), Created configuration directory(%s)",
+ caller._nsr['name_ref'], self.this_nsr_dir)
+ self.config_attributes_file = os.path.join(self.this_nsr_dir, "configuration_config_attributes.yml")
+ self.xlate_dict_file = os.path.join(self.this_nsr_dir, "nsr_xlate_dict.yml")
+
+ def xlate_conf(self, vnfr, vnf_cfg):
+
+ # If configuration type is not already set, try to read from attributes
+ if vnf_cfg['interface_type'] is None:
+ # Prepare unique name for this VNF
+ vnf_unique_name = get_vnf_unique_name(
+ vnf_cfg['nsr_name'],
+ vnfr['short_name'],
+ vnfr['member_vnf_index_ref'],
+ )
+
+ # Find this particular (unique) VNF's config attributes
+ if (vnf_unique_name in self.vnf_config_attributes_dict):
+ vnf_cfg_config_attributes_dict = self.vnf_config_attributes_dict[vnf_unique_name]
+ vnf_cfg['interface_type'] = vnf_cfg_config_attributes_dict['configuration_type']
+ if 'configuration_options' in vnf_cfg_config_attributes_dict:
+ cfg_opts = vnf_cfg_config_attributes_dict['configuration_options']
+ for key, value in cfg_opts.items():
+ vnf_cfg[key] = value
+
+ cfg_path_prefix = '{}/{}/{}_{}'.format(
+ self._parent._parent.cfg_dir,
+ vnf_cfg['nsr_name'],
+ vnfr['short_name'],
+ vnfr['member_vnf_index_ref'],
+ )
+
+ vnf_cfg['cfg_template'] = '{}_{}_template.cfg'.format(cfg_path_prefix, vnf_cfg['interface_type'])
+ vnf_cfg['cfg_file'] = '{}.cfg'.format(cfg_path_prefix)
+ vnf_cfg['xlate_script'] = self._parent._parent.cfg_dir + '/xlate_cfg.py'
+
+ self._log.debug("VNF endpoint so far: %s", vnf_cfg)
+
+ self._log.info("Checking cfg_template %s", vnf_cfg['cfg_template'])
+ if os.path.exists(vnf_cfg['cfg_template']):
+ return True
+ return False
+
+ def ConfigVNF(self, vnfr):
+
+ vnf_cfg = vnfr['vnf_cfg']
+ vnf_cm_state = self.find_or_create_vnfr_cm_state(vnf_cfg)
+
+ if (vnf_cm_state['state'] == self.state_to_string(conmanY.RecordState.READY_NO_CFG)
+ or
+ vnf_cm_state['state'] == self.state_to_string(conmanY.RecordState.READY)):
+ self._log.warning("NS/VNF (%s/%s) is already configured! Skipped.", self.nsr_name, vnfr['short_name'])
+ return
+
+ #UPdate VNF state
+ vnf_cm_state['state'] = self.state_to_string(conmanY.RecordState.CFG_PROCESS)
+
+ # Now translate the configuration for iP addresses
+ try:
+ # Add cp_dict members (TAGS) for this VNF
+ self._cp_dict['rw_mgmt_ip'] = vnf_cfg['mgmt_ip_address']
+ self._cp_dict['rw_username'] = vnf_cfg['username']
+ self._cp_dict['rw_password'] = vnf_cfg['password']
+ ############################################################
+ # TBD - Need to lookup above 3 for a given VNF, not global #
+ # Once we do that no need to dump below file again before #
+ # each VNF configuration translation. #
+ # This will require all existing config templates to be #
+ # changed for above three tags to include member index #
+ ############################################################
+ try:
+ nsr_obj = vnf_cfg['nsr_obj']
+ # Generate config_config_attributes.yaml (For debug reference)
+ with open(nsr_obj.xlate_dict_file, "w") as yf:
+ yf.write(yaml.dump(nsr_obj._cp_dict, default_flow_style=False))
+ except Exception as e:
+ self._log.error("NS:(%s) failed to write nsr xlate tags file as (%s)", nsr_obj.nsr_name, str(e))
+
+ if 'cfg_template' in vnf_cfg:
+ script_cmd = 'python3 {} -i {} -o {} -x "{}"'.format(vnf_cfg['xlate_script'], vnf_cfg['cfg_template'], vnf_cfg['cfg_file'], self.xlate_dict_file)
+ self._log.debug("xlate script command (%s)", script_cmd)
+ #xlate_msg = subprocess.check_output(script_cmd).decode('utf-8')
+ xlate_msg = subprocess.check_output(script_cmd, shell=True).decode('utf-8')
+ self._log.info("xlate script output (%s)", xlate_msg)
+ except Exception as e:
+ vnf_cm_state['state'] = self.state_to_string(conmanY.RecordState.CFG_PROCESS_FAILED)
+ self._log.error("Failed to execute translation script for VNF: %s with (%s)", log_this_vnf(vnf_cfg), str(e))
+ return
+
+ self._log.info("Applying config to VNF: %s = %s!", log_this_vnf(vnf_cfg), vnf_cfg)
+ try:
+ #self.vnf_cfg_list.append(vnf_cfg)
+ self._log.debug("Scheduled configuration!")
+ vnf_cm_state['state'] = self.state_to_string(conmanY.RecordState.CFG_SCHED)
+ except Exception as e:
+ self._log.error("Failed apply_vnf_config to VNF: %s as (%s)", log_this_vnf(vnf_cfg), str(e))
+ vnf_cm_state['state'] = self.state_to_string(conmanY.RecordState.CFG_PROCESS_FAILED)
+ raise
+
+ def add(self, nsr):
+ self._log.info("Adding NS Record for id=%s", id)
+ self._nsr = nsr
+
+ def sample_cm_state(self):
+ return (
+ {
+ 'cm_nsr': [
+ {
+ 'cm_vnfr': [
+ {
+ 'cfg_location': 'location1',
+ 'cfg_type': 'script',
+ 'connection_point': [
+ {'ip_address': '1.1.1.1', 'name': 'vnf1cp1'},
+ {'ip_address': '1.1.1.2', 'name': 'vnf1cp2'}
+ ],
+ 'id': 'vnfrid1',
+ 'mgmt_interface': {'ip_address': '7.1.1.1',
+ 'port': 1001},
+ 'name': 'vnfrname1',
+ 'state': 'init'
+ },
+ {
+ 'cfg_location': 'location2',
+ 'cfg_type': 'netconf',
+ 'connection_point': [{'ip_address': '2.1.1.1', 'name': 'vnf2cp1'},
+ {'ip_address': '2.1.1.2', 'name': 'vnf2cp2'}],
+ 'id': 'vnfrid2',
+ 'mgmt_interface': {'ip_address': '7.1.1.2',
+ 'port': 1001},
+ 'name': 'vnfrname2',
+ 'state': 'init'}
+ ],
+ 'id': 'nsrid1',
+ 'name': 'nsrname1',
+ 'state': 'init'}
+ ],
+ 'states': 'Initialized, '
+ })
+
+ def populate_vm_state_from_vnf_cfg(self):
+ # Fill in each VNFR from this nsr object
+ vnfr_list = self._vnfr_list
+ for vnfr in vnfr_list:
+ vnf_cfg = vnfr['vnf_cfg']
+ vnf_cm_state = self.find_vnfr_cm_state(vnfr['id'])
+
+ if vnf_cm_state:
+ # Fill in VNF management interface
+ vnf_cm_state['mgmt_interface']['ip_address'] = vnf_cfg['mgmt_ip_address']
+ vnf_cm_state['mgmt_interface']['port'] = vnf_cfg['port']
+
+ # Fill in VNF configuration details
+ vnf_cm_state['cfg_type'] = vnf_cfg['config_method']
+ vnf_cm_state['cfg_location'] = vnf_cfg['cfg_file']
+
+ # Fill in each connection-point for this VNF
+ if "connection_point" in vnfr:
+ cp_list = vnfr['connection_point']
+ for cp_item_dict in cp_list:
+ vnf_cm_state['connection_point'].append(
+ {
+ 'name' : cp_item_dict['name'],
+ 'ip_address' : cp_item_dict['ip_address'],
+ }
+ )
+
+ def state_to_string(self, state):
+ state_dict = {
+ conmanY.RecordState.INIT : "init",
+ conmanY.RecordState.RECEIVED : "received",
+ conmanY.RecordState.CFG_PROCESS : "cfg_process",
+ conmanY.RecordState.CFG_PROCESS_FAILED : "cfg_process_failed",
+ conmanY.RecordState.CFG_SCHED : "cfg_sched",
+ conmanY.RecordState.CFG_DELAY : "cfg_delay",
+ conmanY.RecordState.CONNECTING : "connecting",
+ conmanY.RecordState.FAILED_CONNECTION : "failed_connection",
+ conmanY.RecordState.NETCONF_CONNECTED : "netconf_connected",
+ conmanY.RecordState.NETCONF_SSH_CONNECTED : "netconf_ssh_connected",
+ conmanY.RecordState.RESTCONF_CONNECTED : "restconf_connected",
+ conmanY.RecordState.CFG_SEND : "cfg_send",
+ conmanY.RecordState.CFG_FAILED : "cfg_failed",
+ conmanY.RecordState.READY_NO_CFG : "ready_no_cfg",
+ conmanY.RecordState.READY : "ready",
+ }
+ return state_dict[state]
+
+ def find_vnfr_cm_state(self, id):
+ if self.cm_nsr['cm_vnfr']:
+ for vnf_cm_state in self.cm_nsr['cm_vnfr']:
+ if vnf_cm_state['id'] == id:
+ return vnf_cm_state
+ return None
+
+ def find_or_create_vnfr_cm_state(self, vnf_cfg):
+ vnfr = vnf_cfg['vnfr']
+ vnf_cm_state = self.find_vnfr_cm_state(vnfr['id'])
+
+ if vnf_cm_state is None:
+ # Not found, Create and Initialize this VNF cm-state
+ vnf_cm_state = {
+ 'id' : vnfr['id'],
+ 'name' : vnfr['short_name'],
+ 'state' : self.state_to_string(conmanY.RecordState.RECEIVED),
+ 'mgmt_interface' :
+ {
+ 'ip_address' : vnf_cfg['mgmt_ip_address'],
+ 'port' : vnf_cfg['port'],
+ },
+ 'cfg_type' : vnf_cfg['config_method'],
+ 'cfg_location' : vnf_cfg['cfg_file'],
+ 'connection_point' : [],
+ }
+ self.cm_nsr['cm_vnfr'].append(vnf_cm_state)
+
+ # Publish newly created cm-state
+
+
+ return vnf_cm_state
+
+ @asyncio.coroutine
+ def get_vnf_cm_state(self, vnfr):
+ if vnfr:
+ vnf_cm_state = self.find_vnfr_cm_state(vnfr['id'])
+ if vnf_cm_state:
+ return vnf_cm_state['state']
+ return False
+
+ @asyncio.coroutine
+ def update_vnf_cm_state(self, vnfr, state):
+ if vnfr:
+ vnf_cm_state = self.find_vnfr_cm_state(vnfr['id'])
+ if vnf_cm_state is None:
+ self._log.error("No opdata found for NS/VNF:%s/%s!",
+ self.nsr_name, vnfr['short_name'])
+ return
+
+ if vnf_cm_state['state'] != self.state_to_string(state):
+ old_state = vnf_cm_state['state']
+ vnf_cm_state['state'] = self.state_to_string(state)
+ # Publish new state
+ yield from self.publish_cm_state()
+ self._log.info("VNF ({}/{}/{}) state change: {} -> {}"
+ .format(self.nsr_name,
+ vnfr['short_name'],
+ vnfr['member_vnf_index_ref'],
+ old_state,
+ vnf_cm_state['state']))
+
+ else:
+ self._log.error("No VNFR supplied for state update (NS=%s)!",
+ self.nsr_name)
+
+ @property
+ def get_ns_cm_state(self):
+ return self.cm_nsr['state']
+
+ @asyncio.coroutine
+ def update_ns_cm_state(self, state, state_details=None):
+ if self.cm_nsr['state'] != self.state_to_string(state):
+ old_state = self.cm_nsr['state']
+ self.cm_nsr['state'] = self.state_to_string(state)
+ self.cm_nsr['state_details'] = state_details if state_details is not None else None
+ self._log.info("NS ({}) state change: {} -> {}"
+ .format(self.nsr_name,
+ old_state,
+ self.cm_nsr['state']))
+ # Publish new state
+ yield from self.publish_cm_state()
+
+ @asyncio.coroutine
+ def add_vnfr(self, vnfr, vnfr_msg):
+
+ @asyncio.coroutine
+ def populate_subnets_from_vlr(id):
+ try:
+ # Populate cp_dict with VLR subnet info
+ vlr = yield from self.dts_obj.get_vlr(id)
+ if vlr is not None and 'assigned_subnet' in vlr:
+ subnet = {vlr.name:vlr.assigned_subnet}
+ self._cp_dict[vnfr['member_vnf_index_ref']].update(subnet)
+ self._cp_dict.update(subnet)
+ self._log.debug("VNF:(%s) Updated assigned subnet = %s",
+ vnfr['short_name'], subnet)
+ except Exception as e:
+ self._log.error("VNF:(%s) VLR Error = %s",
+ vnfr['short_name'], e)
+
+ if vnfr['id'] not in self._vnfr_dict:
+ self._log.info("NSR(%s) : Adding VNF Record for name=%s, id=%s", self._nsr_id, vnfr['short_name'], vnfr['id'])
+ # Add this vnfr to the list for show, or single traversal
+ self._vnfr_list.append(vnfr)
+ else:
+ self._log.warning("NSR(%s) : VNF Record for name=%s, id=%s already exists, overwriting", self._nsr_id, vnfr['short_name'], vnfr['id'])
+
+ # Make vnfr available by id as well as by name
+ unique_name = get_vnf_unique_name(self.nsr_name, vnfr['short_name'], vnfr['member_vnf_index_ref'])
+ self._vnfr_dict[unique_name] = vnfr
+ self._vnfr_dict[vnfr['id']] = vnfr
+
+ # Create vnf_cfg dictionary with default values
+ vnf_cfg = {
+ 'nsr_obj' : self,
+ 'vnfr' : vnfr,
+ 'agent_vnfr' : self.agent_nsr.add_vnfr(vnfr, vnfr_msg),
+ 'nsr_name' : self.nsr_name,
+ 'nsr_id' : self._nsr_id,
+ 'vnfr_name' : vnfr['short_name'],
+ 'member_vnf_index' : vnfr['member_vnf_index_ref'],
+ 'port' : 0,
+ 'username' : 'admin',
+ 'password' : 'admin',
+ 'config_method' : 'None',
+ 'protocol' : 'None',
+ 'mgmt_ip_address' : '0.0.0.0',
+ 'cfg_file' : 'None',
+ 'cfg_retries' : 0,
+ 'script_type' : 'bash',
+ }
+
+ # Update the mgmt ip address
+ # In case the config method is none, this is not
+ # updated later
+ try:
+ vnf_cfg['mgmt_ip_address'] = vnfr_msg.mgmt_interface.ip_address
+ vnf_cfg['port'] = vnfr_msg.mgmt_interface.port
+ except Exception as e:
+ self._log.warn(
+ "VNFR {}({}), unable to retrieve mgmt ip address: {}".
+ format(vnfr['short_name'], vnfr['id'], e))
+
+ vnfr['vnf_cfg'] = vnf_cfg
+ self.find_or_create_vnfr_cm_state(vnf_cfg)
+
+ '''
+ Build the connection-points list for this VNF (self._cp_dict)
+ '''
+ # Populate global CP list self._cp_dict from VNFR
+ cp_list = []
+ if 'connection_point' in vnfr:
+ cp_list = vnfr['connection_point']
+
+ self._cp_dict[vnfr['member_vnf_index_ref']] = {}
+ if 'vdur' in vnfr:
+ for vdur in vnfr['vdur']:
+ if 'internal_connection_point' in vdur:
+ cp_list += vdur['internal_connection_point']
+
+ for cp_item_dict in cp_list:
+ # Populate global dictionary
+ self._cp_dict[
+ cp_item_dict['name']
+ ] = cp_item_dict['ip_address']
+
+ # Populate unique member specific dictionary
+ self._cp_dict[
+ vnfr['member_vnf_index_ref']
+ ][
+ cp_item_dict['name']
+ ] = cp_item_dict['ip_address']
+
+ # Fill in the subnets from vlr
+ if 'vlr_ref' in cp_item_dict:
+ ### HACK: Internal connection_point do not have VLR reference
+ yield from populate_subnets_from_vlr(cp_item_dict['vlr_ref'])
+
+ if 'internal_vlr' in vnfr:
+ for ivlr in vnfr['internal_vlr']:
+ yield from populate_subnets_from_vlr(ivlr['vlr_ref'])
+
+ # Update vnfr
+ vnf_cfg['agent_vnfr']._vnfr = vnfr
+ return vnf_cfg['agent_vnfr']
+
+
+class XPaths(object):
+ @staticmethod
+ def nsr_opdata(k=None):
+ return ("D,/nsr:ns-instance-opdata/nsr:nsr" +
+ ("[nsr:ns-instance-config-ref='{}']".format(k) if k is not None else ""))
+
+ @staticmethod
+ def nsd_msg(k=None):
+ return ("C,/nsd:nsd-catalog/nsd:nsd" +
+ "[nsd:id = '{}']".format(k) if k is not None else "")
+
+ @staticmethod
+ def vnfr_opdata(k=None):
+ return ("D,/vnfr:vnfr-catalog/vnfr:vnfr" +
+ ("[vnfr:id='{}']".format(k) if k is not None else ""))
+
+ @staticmethod
+ def config_agent(k=None):
+ return ("D,/rw-config-agent:config-agent/rw-config-agent:account" +
+ ("[rw-config-agent:name='{}']".format(k) if k is not None else ""))
+
+ @staticmethod
+ def nsr_config(k=None):
+ return ("C,/nsr:ns-instance-config/nsr:nsr[nsr:id='{}']".format(k) if k is not None else "")
+
+ @staticmethod
+ def vlr(k=None):
+ return ("D,/vlr:vlr-catalog/vlr:vlr[vlr:id='{}']".format(k) if k is not None else "")
+
+class ConfigManagerDTS(object):
+ ''' This class either reads from DTS or publishes to DTS '''
+
+ def __init__(self, log, loop, parent, dts):
+ self._log = log
+ self._loop = loop
+ self._parent = parent
+ self._dts = dts
+
+ @asyncio.coroutine
+ def _read_dts(self, xpath, do_trace=False):
+ self._log.debug("_read_dts path = %s", xpath)
+ flags = rwdts.XactFlag.MERGE
+ res_iter = yield from self._dts.query_read(
+ xpath, flags=flags
+ )
+
+ results = []
+ try:
+ for i in res_iter:
+ result = yield from i
+ if result is not None:
+ results.append(result.result)
+ except:
+ pass
+
+ return results
+
+
+ @asyncio.coroutine
+ def get_nsr(self, id):
+ self._log.debug("Attempting to get NSR: %s", id)
+ nsrl = yield from self._read_dts(XPaths.nsr_opdata(id), False)
+ nsr = None
+ if len(nsrl) > 0:
+ nsr = nsrl[0].as_dict()
+ return nsr
+
+ @asyncio.coroutine
+ def get_nsr_config(self, id):
+ self._log.debug("Attempting to get config NSR: %s", id)
+ nsrl = yield from self._read_dts(XPaths.nsr_config(id), False)
+ nsr = None
+ if len(nsrl) > 0:
+ nsr = nsrl[0]
+ return nsr
+
+ @asyncio.coroutine
+ def get_nsd_msg(self, id):
+ self._log.debug("Attempting to get NSD: %s", id)
+ nsdl = yield from self._read_dts(XPaths.nsd_msg(id), False)
+ nsd_msg = None
+ if len(nsdl) > 0:
+ nsd_msg = nsdl[0]
+ return nsd_msg
+
+ @asyncio.coroutine
+ def get_nsd(self, nsr_id):
+ self._log.debug("Attempting to get NSD for NSR: %s", id)
+ nsr_config = yield from self.get_nsr_config(nsr_id)
+ nsd_msg = nsr_config.nsd
+ return nsd_msg
+
+ @asyncio.coroutine
+ def get_vnfr(self, id):
+ self._log.debug("Attempting to get VNFR: %s", id)
+ vnfrl = yield from self._read_dts(XPaths.vnfr_opdata(id), do_trace=False)
+ vnfr_msg = None
+ if len(vnfrl) > 0:
+ vnfr_msg = vnfrl[0]
+ return vnfr_msg
+
+ @asyncio.coroutine
+ def get_vlr(self, id):
+ self._log.debug("Attempting to get VLR subnet: %s", id)
+ vlrl = yield from self._read_dts(XPaths.vlr(id), do_trace=True)
+ vlr_msg = None
+ if len(vlrl) > 0:
+ vlr_msg = vlrl[0]
+ return vlr_msg
+
+ @asyncio.coroutine
+ def get_config_agents(self, name):
+ self._log.debug("Attempting to get config_agents: %s", name)
+ cfgagentl = yield from self._read_dts(XPaths.config_agent(name), False)
+ return cfgagentl
+
+ @asyncio.coroutine
+ def update(self, path, msg, flags=rwdts.XactFlag.REPLACE):
+ """
+ Update a cm-state (cm-nsr) record in DTS with the path and message
+ """
+ self._log.debug("Updating cm-state %s:%s dts_pub_hdl = %s", path, msg, self.dts_pub_hdl)
+ self.dts_pub_hdl.update_element(path, msg, flags)
+ self._log.debug("Updated cm-state, %s:%s", path, msg)
+
+ @asyncio.coroutine
+ def delete(self, path):
+ """
+ Delete cm-nsr record in DTS with the path only
+ """
+ self._log.debug("Deleting cm-nsr %s dts_pub_hdl = %s", path, self.dts_pub_hdl)
+ self.dts_pub_hdl.delete_element(path)
+ self._log.debug("Deleted cm-nsr, %s", path)
+
+ @asyncio.coroutine
+ def register(self):
+ yield from self.register_to_publish()
+ yield from self.register_for_nsr()
+
+ @asyncio.coroutine
+ def register_to_publish(self):
+ ''' Register to DTS for publishing cm-state opdata '''
+
+ xpath = "D,/rw-conman:cm-state/rw-conman:cm-nsr"
+ self._log.debug("Registering to publish cm-state @ %s", xpath)
+ hdl = rift.tasklets.DTS.RegistrationHandler()
+ with self._dts.group_create() as group:
+ self.dts_pub_hdl = group.register(xpath=xpath,
+ handler=hdl,
+ flags=rwdts.Flag.PUBLISHER | rwdts.Flag.NO_PREP_READ)
+
+ @property
+ def nsr_xpath(self):
+ return "D,/nsr:ns-instance-opdata/nsr:nsr"
+
+ @asyncio.coroutine
+ def register_for_nsr(self):
+ """ Register for NSR changes """
+
+ @asyncio.coroutine
+ def on_prepare(xact_info, query_action, ks_path, msg):
+ """ This NSR is created """
+ self._log.debug("Received NSR instantiate on_prepare (%s:%s:%s)",
+ query_action,
+ ks_path,
+ msg)
+
+ if (query_action == rwdts.QueryAction.UPDATE or
+ query_action == rwdts.QueryAction.CREATE):
+ msg_dict = msg.as_dict()
+ # Update Each NSR/VNFR state)
+ if ('operational_status' in msg_dict and
+ msg_dict['operational_status'] == 'running'):
+ # Add to the task list
+ self._parent.add_to_pending_tasks({'nsrid' : msg_dict['ns_instance_config_ref'], 'retries' : 5})
+ elif query_action == rwdts.QueryAction.DELETE:
+ nsr_id = msg.ns_instance_config_ref
+ asyncio.ensure_future(self._parent.terminate_NSR(nsr_id), loop=self._loop)
+ else:
+ raise NotImplementedError(
+ "%s action on cm-state not supported",
+ query_action)
+
+ xact_info.respond_xpath(rwdts.XactRspCode.ACK)
+
+ try:
+ handler = rift.tasklets.DTS.RegistrationHandler(on_prepare=on_prepare)
+ self.dts_reg_hdl = yield from self._dts.register(self.nsr_xpath,
+ flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY,
+ handler=handler)
+ except Exception as e:
+ self._log.error("Failed to register for NSR changes as %s", str(e))
+
+
diff --git a/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/rwconman_events.py b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/rwconman_events.py
new file mode 100644
index 0000000..f292a68
--- /dev/null
+++ b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/rwconman_events.py
@@ -0,0 +1,367 @@
+
+#
+# Copyright 2016 RIFT.IO Inc
+#
+# 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 ncclient
+import ncclient.asyncio_manager
+import tornado.httpclient as tornadoh
+import asyncio.subprocess
+import asyncio
+import time
+import sys
+import os, stat
+
+import gi
+gi.require_version('RwDts', '1.0')
+gi.require_version('RwYang', '1.0')
+gi.require_version('RwConmanYang', '1.0')
+gi.require_version('RwNsrYang', '1.0')
+gi.require_version('RwVnfrYang', '1.0')
+
+from gi.repository import (
+ RwDts as rwdts,
+ RwYang,
+ RwConmanYang as conmanY,
+ RwNsrYang as nsrY,
+ RwVnfrYang as vnfrY,
+)
+
+import rift.tasklets
+
+if sys.version_info < (3, 4, 4):
+ asyncio.ensure_future = asyncio.async
+
+def log_this_vnf(vnf_cfg):
+ log_vnf = ""
+ used_item_list = ['nsr_name', 'vnfr_name', 'member_vnf_index', 'mgmt_ip_address']
+ for item in used_item_list:
+ if item in vnf_cfg:
+ if item == 'mgmt_ip_address':
+ log_vnf += "({})".format(vnf_cfg[item])
+ else:
+ log_vnf += "{}/".format(vnf_cfg[item])
+ return log_vnf
+
+class ConfigManagerROifConnectionError(Exception):
+ pass
+class ScriptError(Exception):
+ pass
+
+
+class ConfigManagerEvents(object):
+ def __init__(self, dts, log, loop, parent):
+ self._dts = dts
+ self._log = log
+ self._loop = loop
+ self._parent = parent
+ self._nsr_xpath = "/cm-state/cm-nsr"
+
+ @asyncio.coroutine
+ def register(self):
+ pass
+
+ @asyncio.coroutine
+ def update_vnf_state(self, vnf_cfg, state):
+ nsr_obj = vnf_cfg['nsr_obj']
+ yield from nsr_obj.update_vnf_cm_state(vnf_cfg['vnfr'], state)
+
+ @asyncio.coroutine
+ def apply_vnf_config(self, vnf_cfg):
+ self._log.debug("apply_vnf_config VNF:{}"
+ .format(log_this_vnf(vnf_cfg)))
+
+ if vnf_cfg['config_delay']:
+ yield from self.update_vnf_state(vnf_cfg, conmanY.RecordState.CFG_DELAY)
+ yield from asyncio.sleep(vnf_cfg['config_delay'], loop=self._loop)
+
+ # See if we are still alive!
+ if vnf_cfg['nsr_obj'].being_deleted:
+ # Don't do anything, just return
+ self._log.info("VNF : %s is being deleted, skipping configuration!",
+ log_this_vnf(vnf_cfg))
+ return True
+
+ yield from self.update_vnf_state(vnf_cfg, conmanY.RecordState.CFG_SEND)
+ try:
+ if vnf_cfg['config_method'] == 'netconf':
+ self._log.info("Creating ncc handle for VNF cfg = %s!", vnf_cfg)
+ self.ncc = ConfigManagerVNFnetconf(self._log, self._loop, self, vnf_cfg)
+ if vnf_cfg['protocol'] == 'ssh':
+ yield from self.ncc.connect_ssh()
+ else:
+ yield from self.ncc.connect()
+ yield from self.ncc.apply_edit_cfg()
+ elif vnf_cfg['config_method'] == 'rest':
+ if self.rcc is None:
+ self._log.info("Creating rcc handle for VNF cfg = %s!", vnf_cfg)
+ self.rcc = ConfigManagerVNFrestconf(self._log, self._loop, self, vnf_cfg)
+ self.ncc.apply_edit_cfg()
+ elif vnf_cfg['config_method'] == 'script':
+ self._log.info("Executing script for VNF cfg = %s!", vnf_cfg)
+ scriptc = ConfigManagerVNFscriptconf(self._log, self._loop, self, vnf_cfg)
+ yield from scriptc.apply_edit_cfg()
+ elif vnf_cfg['config_method'] == 'juju':
+ self._log.info("Executing juju config for VNF cfg = %s!", vnf_cfg)
+ jujuc = ConfigManagerVNFjujuconf(self._log, self._loop, self._parent, vnf_cfg)
+ yield from jujuc.apply_edit_cfg()
+ else:
+ self._log.error("Unknown configuration method(%s) received for %s",
+ vnf_cfg['config_method'], vnf_cfg['vnf_unique_name'])
+ yield from self.update_vnf_state(vnf_cfg, conmanY.RecordState.CFG_FAILED)
+ return True
+
+ #Update VNF state
+ yield from self.update_vnf_state(vnf_cfg, conmanY.RecordState.READY)
+ self._log.info("Successfully applied configuration to VNF: %s",
+ log_this_vnf(vnf_cfg))
+ except Exception as e:
+ self._log.error("Applying configuration(%s) file(%s) to VNF: %s failed as: %s",
+ vnf_cfg['config_method'],
+ vnf_cfg['cfg_file'],
+ log_this_vnf(vnf_cfg),
+ str(e))
+ #raise
+ return False
+
+ return True
+
+class ConfigManagerVNFscriptconf(object):
+
+ def __init__(self, log, loop, parent, vnf_cfg):
+ self._log = log
+ self._loop = loop
+ self._parent = parent
+ self._manager = None
+ self._vnf_cfg = vnf_cfg
+
+ #@asyncio.coroutine
+ def apply_edit_cfg(self):
+ vnf_cfg = self._vnf_cfg
+ self._log.debug("Attempting to apply scriptconf to VNF: %s", log_this_vnf(vnf_cfg))
+ try:
+ st = os.stat(vnf_cfg['cfg_file'])
+ os.chmod(vnf_cfg['cfg_file'], st.st_mode | stat.S_IEXEC)
+ #script_msg = subprocess.check_output(vnf_cfg['cfg_file'], shell=True).decode('utf-8')
+
+ proc = yield from asyncio.create_subprocess_exec(
+ vnf_cfg['script_type'], vnf_cfg['cfg_file'],
+ stdout=asyncio.subprocess.PIPE)
+ script_msg = yield from proc.stdout.read()
+ rc = yield from proc.wait()
+
+ if rc != 0:
+ raise ScriptError(
+ "script config returned error code : %s" % rc
+ )
+
+ self._log.debug("config script output (%s)", script_msg)
+ except Exception as e:
+ self._log.error("Error (%s) while executing script config for VNF: %s",
+ str(e), log_this_vnf(vnf_cfg))
+ raise
+
+class ConfigManagerVNFrestconf(object):
+
+ def __init__(self, log, loop, parent, vnf_cfg):
+ self._log = log
+ self._loop = loop
+ self._parent = parent
+ self._manager = None
+ self._vnf_cfg = vnf_cfg
+
+ def fetch_handle(self, response):
+ if response.error:
+ self._log.error("Failed to send HTTP config request - %s", response.error)
+ else:
+ self._log.debug("Sent HTTP config request - %s", response.body)
+
+ @asyncio.coroutine
+ def apply_edit_cfg(self):
+ vnf_cfg = self._vnf_cfg
+ self._log.debug("Attempting to apply restconf to VNF: %s", log_this_vnf(vnf_cfg))
+ try:
+ http_c = tornadoh.AsyncHTTPClient()
+ # TBD
+ # Read the config entity from file?
+ # Convert connectoin-point?
+ http_c.fetch("http://", self.fetch_handle)
+ except Exception as e:
+ self._log.error("Error (%s) while applying HTTP config", str(e))
+
+class ConfigManagerVNFnetconf(object):
+
+ def __init__(self, log, loop, parent, vnf_cfg):
+ self._log = log
+ self._loop = loop
+ self._parent = parent
+ self._manager = None
+ self._vnf_cfg = vnf_cfg
+
+ self._model = RwYang.Model.create_libncx()
+ self._model.load_schema_ypbc(conmanY.get_schema())
+
+ @asyncio.coroutine
+ def connect(self, timeout_secs=120):
+ vnf_cfg = self._vnf_cfg
+ start_time = time.time()
+ self._log.debug("connecting netconf .... %s", vnf_cfg)
+ while (time.time() - start_time) < timeout_secs:
+
+ try:
+ self._log.info("Attemping netconf connection to VNF: %s", log_this_vnf(vnf_cfg))
+
+ self._manager = yield from ncclient.asyncio_manager.asyncio_connect(
+ loop=self._loop,
+ host=vnf_cfg['mgmt_ip_address'],
+ port=vnf_cfg['port'],
+ username=vnf_cfg['username'],
+ password=vnf_cfg['password'],
+ allow_agent=False,
+ look_for_keys=False,
+ hostkey_verify=False,
+ )
+
+ self._log.info("Netconf connected to VNF: %s", log_this_vnf(vnf_cfg))
+ return
+
+ except ncclient.transport.errors.SSHError as e:
+ yield from self._parent.update_vnf_state(vnf_cfg, conmanY.RecordState.FAILED_CONNECTION)
+ self._log.error("Netconf connection to VNF: %s, failed: %s",
+ log_this_vnf(vnf_cfg), str(e))
+
+ yield from asyncio.sleep(2, loop=self._loop)
+
+ raise ConfigManagerROifConnectionError(
+ "Failed to connect to VNF: %s within %s seconds" %
+ (log_this_vnf(vnf_cfg), timeout_secs)
+ )
+
+ @asyncio.coroutine
+ def connect_ssh(self, timeout_secs=120):
+ vnf_cfg = self._vnf_cfg
+ start_time = time.time()
+
+ if (self._manager != None and self._manager.connected == True):
+ self._log.debug("Disconnecting previous session")
+ self._manager.close_session
+
+ self._log.debug("connecting netconf via SSH .... %s", vnf_cfg)
+ while (time.time() - start_time) < timeout_secs:
+
+ try:
+ yield from self._parent.update_vnf_state(vnf_cfg, conmanY.RecordState.CONNECTING)
+ self._log.debug("Attemping netconf connection to VNF: %s", log_this_vnf(vnf_cfg))
+
+ self._manager = ncclient.asyncio_manager.manager.connect_ssh(
+ host=vnf_cfg['mgmt_ip_address'],
+ port=vnf_cfg['port'],
+ username=vnf_cfg['username'],
+ password=vnf_cfg['password'],
+ allow_agent=False,
+ look_for_keys=False,
+ hostkey_verify=False,
+ )
+
+ yield from self._parent.update_vnf_state(vnf_cfg, conmanY.RecordState.NETCONF_SSH_CONNECTED)
+ self._log.debug("netconf over SSH connected to VNF: %s", log_this_vnf(vnf_cfg))
+ return
+
+ except ncclient.transport.errors.SSHError as e:
+ yield from self._parent.update_vnf_state(vnf_cfg, conmanY.RecordState.FAILED_CONNECTION)
+ self._log.error("Netconf connection to VNF: %s, failed: %s",
+ log_this_vnf(vnf_cfg), str(e))
+
+ yield from asyncio.sleep(2, loop=self._loop)
+
+ raise ConfigManagerROifConnectionError(
+ "Failed to connect to VNF: %s within %s seconds" %
+ (log_this_vnf(vnf_cfg), timeout_secs)
+ )
+
+ @asyncio.coroutine
+ def apply_edit_cfg(self):
+ vnf_cfg = self._vnf_cfg
+ self._log.debug("Attempting to apply netconf to VNF: %s", log_this_vnf(vnf_cfg))
+
+ if self._manager is None:
+ self._log.error("Netconf is not connected to VNF: %s, aborting!", log_this_vnf(vnf_cfg))
+ return
+
+ # Get config file contents
+ try:
+ with open(vnf_cfg['cfg_file']) as f:
+ configuration = f.read()
+ except Exception as e:
+ self._log.error("Reading contents of the configuration file(%s) failed: %s", vnf_cfg['cfg_file'], str(e))
+ return
+
+ try:
+ self._log.debug("apply_edit_cfg to VNF: %s", log_this_vnf(vnf_cfg))
+ xml = '<config xmlns:xc="urn:ietf:params:xml:ns:netconf:base:1.0">{}</config>'.format(configuration)
+ response = yield from self._manager.edit_config(xml, target='running')
+ if hasattr(response, 'xml'):
+ response_xml = response.xml
+ else:
+ response_xml = response.data_xml.decode()
+
+ self._log.debug("apply_edit_cfg response: %s", response_xml)
+ if '<rpc-error>' in response_xml:
+ raise ConfigManagerROifConnectionError("apply_edit_cfg response has rpc-error : %s",
+ response_xml)
+
+ self._log.debug("apply_edit_cfg Successfully applied configuration {%s}", xml)
+ except:
+ raise
+
+class ConfigManagerVNFjujuconf(object):
+
+ def __init__(self, log, loop, parent, vnf_cfg):
+ self._log = log
+ self._loop = loop
+ self._parent = parent
+ self._manager = None
+ self._vnf_cfg = vnf_cfg
+
+ #@asyncio.coroutine
+ def apply_edit_cfg(self):
+ vnf_cfg = self._vnf_cfg
+ self._log.debug("Attempting to apply juju conf to VNF: %s", log_this_vnf(vnf_cfg))
+ try:
+ args = ['python3',
+ vnf_cfg['juju_script'],
+ '--server', vnf_cfg['mgmt_ip_address'],
+ '--user', vnf_cfg['user'],
+ '--password', vnf_cfg['secret'],
+ '--port', str(vnf_cfg['port']),
+ vnf_cfg['cfg_file']]
+ self._log.error("juju script command (%s)", args)
+
+ proc = yield from asyncio.create_subprocess_exec(
+ *args,
+ stdout=asyncio.subprocess.PIPE)
+ juju_msg = yield from proc.stdout.read()
+ rc = yield from proc.wait()
+
+ if rc != 0:
+ raise ScriptError(
+ "Juju config returned error code : %s" % rc
+ )
+
+ self._log.debug("Juju config output (%s)", juju_msg)
+ except Exception as e:
+ self._log.error("Error (%s) while executing juju config", str(e))
+ raise
diff --git a/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/rwconman_test_config_template.cfg b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/rwconman_test_config_template.cfg
new file mode 100644
index 0000000..d5342c2
--- /dev/null
+++ b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/rwconman_test_config_template.cfg
@@ -0,0 +1,32 @@
+# This template has all supported TAGs.
+# This template can be used as input to the xlate_cfg.py script as follows:
+
+# python3 ./xlate_cfg.py -i ./rwconman_test_config_template.cfg -o ./rwconman_test_config.cfg -x ./rwconman_test_xlate_dict.yml
+
+
+# This is error
+#0. <rw_connection_point_name test/cp2>
+
+# Following are simple TAGs
+1. This is Management IP: <rw_mgmt_ip>
+2. This is Username: <rw_username>
+3. This is Password: <rw_password>
+4. This is globally unique connection point: <rw_connection_point_name test/cp1>
+
+# Following are colon separated complex TAGs
+5. This is connection point for a given VNF with unique member index: <rw_unique_index:rw_connection_point_name 2:test/cp1>
+6. This is converting connection point IP address into network address: <rw_connection_point:masklen_network test/cp1:24>
+7. This is converting connection point IP address into boadcast address: <rw_connection_point:masklen_broadcast test/cp1:24>
+
+# Following generated tuple with original connectino point name (Global only)
+8. This is not used anywhere: <rw_connection_point_tuple test/cp1>
+
+# Following are multi-colon separated complex TAGs
+9. This is converting connection point IP address into network address VNF with unique member index: <rw_unique_index:rw_connection_point:masklen_network 2:test/cp1:24>
+10. This is converting connection point IP address into network address VNF with unique member index: <rw_unique_index:rw_connection_point:masklen_broadcast 2:test/cp1:24>
+
+# Following test all of the above in single line
+11. All at once: START| rw_mgmt_ip: <rw_mgmt_ip> | rw_username: <rw_username> | rw_password: <rw_password> | global CP: <rw_connection_point_name test/cp1> | 1 CP: <rw_unique_index:rw_connection_point_name 1:test/cp1> | network: <rw_connection_point:masklen_network test/cp1:24> | broadcast: <rw_connection_point:masklen_broadcast test/cp1:24> | tuple: <rw_connection_point_tuple test/cp1> | 2 network: <rw_unique_index:rw_connection_point:masklen_network 2:test/cp1:24> | 2 broadcast: <rw_unique_index:rw_connection_point:masklen_broadcast 2:test/cp1:24> |END
+
+# Need to work on the solution for multiple pattern of same type in single line.
+
diff --git a/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/rwconman_test_xlate_dict.yml b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/rwconman_test_xlate_dict.yml
new file mode 100644
index 0000000..becbff1
--- /dev/null
+++ b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/rwconman_test_xlate_dict.yml
@@ -0,0 +1,8 @@
+1:
+ test/cp1: 11.0.0.1
+2:
+ test/cp1: 11.0.0.2
+test/cp1: 11.0.0.3
+rw_mgmt_ip: 1.1.1.1
+rw_username: admin
+rw_password: admin
diff --git a/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/rwconmantasklet.py b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/rwconmantasklet.py
new file mode 100755
index 0000000..7ea73c4
--- /dev/null
+++ b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/rwconmantasklet.py
@@ -0,0 +1,352 @@
+
+#
+# Copyright 2016 RIFT.IO Inc
+#
+# 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.
+#
+
+'''
+This file - ConfigManagerTasklet()
+|
++--|--> ConfigurationManager()
+ |
+ +--> rwconman_config.py - ConfigManagerConfig()
+ | |
+ | +--> ConfigManagerNSR()
+ |
+ +--> rwconman_events.py - ConfigManagerEvents()
+ |
+ +--> ConfigManagerROif()
+
+'''
+
+import asyncio
+import logging
+import os
+
+import gi
+gi.require_version('RwDts', '1.0')
+gi.require_version('RwConmanYang', '1.0')
+
+from gi.repository import (
+ RwDts as rwdts,
+ RwConmanYang as conmanY,
+)
+
+import rift.tasklets
+
+from . import rwconman_config as Config
+from . import rwconman_events as Event
+
+def log_this_vnf(vnf_cfg):
+ log_vnf = ""
+ used_item_list = ['nsr_name', 'vnfr_name', 'member_vnf_index', 'mgmt_ip_address']
+ for item in used_item_list:
+ if item in vnf_cfg:
+ if item == 'mgmt_ip_address':
+ log_vnf += "({})".format(vnf_cfg[item])
+ else:
+ log_vnf += "{}/".format(vnf_cfg[item])
+ return log_vnf
+
+class ConfigurationManager(object):
+ def __init__(self, log, loop, dts):
+ self._log = log
+ self._loop = loop
+ self._dts = dts
+ self.cfg_sleep = True
+ self.cfg_dir = os.path.join(os.environ["RIFT_INSTALL"], "etc/conman")
+ self._config = Config.ConfigManagerConfig(self._dts, self._log, self._loop, self)
+ self._event = Event.ConfigManagerEvents(self._dts, self._log, self._loop, self)
+ self.pending_cfg = []
+ self.pending_tasks = {}
+ self._nsr_objs = {}
+
+ self._handlers = [
+ self._config,
+ self._event,
+ ]
+
+
+ @asyncio.coroutine
+ def update_vnf_state(self, vnf_cfg, state):
+ nsr_obj = vnf_cfg['nsr_obj']
+ self._log.info("Updating cm-state for VNF(%s/%s) to:%s", nsr_obj.nsr_name, vnf_cfg['vnfr_name'], state)
+ yield from nsr_obj.update_vnf_cm_state(vnf_cfg['vnfr'], state)
+
+ @asyncio.coroutine
+ def update_ns_state(self, nsr_obj, state):
+ self._log.info("Updating cm-state for NS(%s) to:%s", nsr_obj.nsr_name, state)
+ yield from nsr_obj.update_ns_cm_state(state)
+
+ def add_to_pending(self, nsr_obj):
+
+ if (nsr_obj not in self.pending_cfg and
+ nsr_obj.cm_nsr['state'] == nsr_obj.state_to_string(conmanY.RecordState.RECEIVED)):
+
+ self._log.info("Adding NS={} to pending config list"
+ .format(nsr_obj.nsr_name))
+
+ # Build the list
+ nsr_obj.vnf_cfg_list = []
+ # Sort all the VNF by their configuration attribute priority
+ sorted_dict = dict(sorted(nsr_obj.nsr_cfg_config_attributes_dict.items()))
+ for config_attributes_dict in sorted_dict.values():
+ # Iterate through each priority level
+ for config_priority in config_attributes_dict:
+ # Iterate through each vnfr at this priority level
+ vnfr = nsr_obj._vnfr_dict[config_priority['id']]
+ self._log.debug("Adding VNF:(%s) to pending cfg list", log_this_vnf(vnfr['vnf_cfg']))
+ nsr_obj.vnf_cfg_list.append(vnfr['vnf_cfg'])
+ self.pending_cfg.append(nsr_obj)
+
+ def add_nsr_obj(self, nsr_obj):
+ self._log.debug("Adding nsr_obj (%s) to Configuration Manager", nsr_obj)
+ self._nsr_objs[nsr_obj.nsr_id] = nsr_obj
+
+ def remove_nsr_obj(self, nsr_id):
+ self._log.debug("Removing nsr_obj (%s) from Configuration Manager", nsr_id)
+ del self._nsr_objs[nsr_id]
+
+ def get_nsr_obj(self, nsr_id):
+ self._log.debug("Returning nsr_obj (%s) from Configuration Manager", self._nsr_objs[nsr_id])
+ return self._nsr_objs.get(nsr_id)
+
+ @asyncio.coroutine
+ def configuration_handler(self):
+ @asyncio.coroutine
+ def process_vnf_cfg(agent_vnfr, nsr_obj):
+ vnf_cfg = agent_vnfr.vnf_cfg
+ done = False
+
+ if vnf_cfg['cfg_retries']:
+ # This failed previously, lets give it some time
+ yield from asyncio.sleep(5, loop=self._loop)
+
+ vnf_cfg['cfg_retries'] += 1
+
+ # Check to see if this vnfr is managed
+ done = yield from self._config._config_agent_mgr.invoke_config_agent_plugins(
+ 'apply_initial_config',
+ nsr_obj.agent_nsr,
+ agent_vnfr)
+ self._log.debug("Apply configuration for VNF={} on attempt {} " \
+ "returned {}".format(log_this_vnf(vnf_cfg),
+ vnf_cfg['cfg_retries'],
+ done))
+
+ if done:
+ yield from self.update_vnf_state(vnf_cfg, conmanY.RecordState.READY)
+
+ else:
+ # Check to see if the VNF configure failed
+ status = yield from self._config._config_agent_mgr.invoke_config_agent_plugins(
+ 'get_config_status',
+ nsr_obj.agent_nsr,
+ agent_vnfr)
+
+ if status and status == 'error':
+ # Failed configuration
+ nsr_obj.vnf_failed = True
+ done = True
+ yield from self.update_vnf_state(vnf_cfg, conmanY.RecordState.CFG_FAILED)
+ self._log.error("Failed to apply configuration for VNF = {}"
+ .format(log_this_vnf(vnf_cfg)))
+
+ return done
+
+ @asyncio.coroutine
+ def process_nsr_obj(nsr_obj):
+ # Return status, this will be set to False is if we fail to configure any VNF
+ ret_status = True
+
+ # Reset VNF failed flag
+ nsr_obj.vnf_failed = False
+ vnf_cfg_list = nsr_obj.vnf_cfg_list
+ while vnf_cfg_list:
+ # Check to make sure the NSR is still valid
+ if nsr_obj.parent.is_nsr_valid(nsr_obj.nsr_id) is False:
+ self._log.info("NSR {} not found, could be terminated".
+ format(nsr_obj.nsr_id))
+ return
+
+ # Need while loop here, since we will be removing list item
+ vnf_cfg = vnf_cfg_list.pop(0)
+ self._log.info("Applying Pending Configuration for VNF = %s / %s",
+ log_this_vnf(vnf_cfg), vnf_cfg['agent_vnfr'])
+ vnf_done = yield from process_vnf_cfg(vnf_cfg['agent_vnfr'], nsr_obj)
+ self._log.debug("Applied Pending Configuration for VNF = {}, status={}"
+ .format(log_this_vnf(vnf_cfg), vnf_done))
+
+ if not vnf_done:
+ # We will retry, but we will give other VNF chance first since this one failed.
+ vnf_cfg_list.append(vnf_cfg)
+
+ if nsr_obj.vnf_failed:
+ # Atleast one VNF config failed
+ ret_status = False
+
+ if ret_status:
+ # Apply NS initial config if present
+ nsr_obj.nsr_failed = False
+ self._log.debug("Apply initial config on NSR {}".format(nsr_obj.nsr_name))
+ try:
+ yield from nsr_obj.parent.process_ns_initial_config(nsr_obj)
+ except Exception as e:
+ nsr_obj.nsr_failed = True
+ self._log.exception(e)
+ ret_status = False
+
+ # Set the config status for the NSR
+ if ret_status:
+ yield from nsr_obj.update_ns_cm_state(conmanY.RecordState.READY)
+ elif nsr_obj.vnf_failed or nsr_obj.nsr_failed:
+ yield from nsr_obj.update_ns_cm_state(conmanY.RecordState.CFG_FAILED)
+ return ret_status
+
+ # Basically, this loop will never end.
+ while True:
+ # Check the pending tasks are complete
+ # Store a list of tasks that are completed and
+ # remove from the pending_tasks list outside loop
+ ids = []
+ for nsr_id, task in self.pending_tasks.items():
+ if task.done():
+ ids.append(nsr_id)
+ e = task.exception()
+ if e:
+ self._log.error("Exception in configuring nsr {}: {}".
+ format(nsr_id, e))
+ nsr_obj = self.get_nsr_obj(nsr_id)
+ if nsr_obj:
+ yield from nsr_obj.update_ns_cm_state(conmanY.RecordState.CFG_FAILED, str(e))
+
+ else:
+ rc = task.result()
+ self._log.debug("NSR {} configured: {}".format(nsr_id, rc))
+ else:
+ self._log.debug("NSR {} still configuring".format(nsr_id))
+
+ # Remove the completed tasks
+ for nsr_id in ids:
+ self.pending_tasks.pop(nsr_id)
+
+ # TODO (pjoseph): Fix this
+ # Sleep before processing any NS (Why are we getting multiple NSR running DTS updates?)
+ # If the sleep is not 10 seconds it does not quite work, NSM is marking it 'running'
+ # wrongfully 10 seconds in advance?
+ yield from asyncio.sleep(10, loop=self._loop)
+
+ if self.pending_cfg:
+ # get first NS, pending_cfg is nsr_obj list
+ nsr_obj = self.pending_cfg[0]
+ nsr_done = False
+ if nsr_obj.being_deleted is False:
+ # Process this NS, returns back same obj is successfull or exceeded retries
+ try:
+ self._log.info("Processing NSR:{}".format(nsr_obj.nsr_name))
+
+ # Check if we already have a task running for this NSR
+ # Case where we are still configuring and terminate is called
+ if nsr_obj.nsr_id in self.pending_tasks:
+ self._log.error("NSR {} in state {} has a configure task running.".
+ format(nsr_obj.nsr_name, nsr_obj.get_ns_cm_state()))
+ # Terminate the task for this NSR
+ self.pending_tasks[nsr_obj.nsr_id].cancel()
+
+ yield from self.update_ns_state(nsr_obj, conmanY.RecordState.CFG_PROCESS)
+
+ # Call in a separate thread
+ self.pending_tasks[nsr_obj.nsr_id] = \
+ self._loop.create_task(
+ process_nsr_obj(nsr_obj)
+ )
+
+ # Remove this nsr_obj
+ self.pending_cfg.remove(nsr_obj)
+
+ except Exception as e:
+ self._log.error("Failed to process NSR as %s", str(e))
+ self._log.exception(e)
+
+
+ @asyncio.coroutine
+ def register(self):
+ # Perform register() for all handlers
+ for reg in self._handlers:
+ yield from reg.register()
+
+ asyncio.ensure_future(self.configuration_handler(), loop=self._loop)
+
+class ConfigManagerTasklet(rift.tasklets.Tasklet):
+ def __init__(self, *args, **kwargs):
+ super(ConfigManagerTasklet, self).__init__(*args, **kwargs)
+ self.rwlog.set_category("rw-conman-log")
+
+ self._dts = None
+ self._con_man = None
+
+ def start(self):
+ super(ConfigManagerTasklet, self).start()
+
+ self.log.debug("Registering with dts")
+
+ self._dts = rift.tasklets.DTS(self.tasklet_info,
+ conmanY.get_schema(),
+ self.loop,
+ self.on_dts_state_change)
+
+ self.log.debug("Created DTS Api GI Object: %s", self._dts)
+
+ def on_instance_started(self):
+ self.log.debug("Got instance started callback")
+
+ @asyncio.coroutine
+ def init(self):
+ self._log.info("Initializing the Configuration-Manager tasklet")
+ self._con_man = ConfigurationManager(self.log,
+ self.loop,
+ self._dts)
+ yield from self._con_man.register()
+
+ @asyncio.coroutine
+ def run(self):
+ pass
+
+ @asyncio.coroutine
+ def on_dts_state_change(self, state):
+ """Take action according to current dts state to transition
+ application into the corresponding application state
+
+ Arguments
+ state - current dts state
+ """
+ switch = {
+ rwdts.State.INIT: rwdts.State.REGN_COMPLETE,
+ rwdts.State.CONFIG: rwdts.State.RUN,
+ }
+
+ handlers = {
+ rwdts.State.INIT: self.init,
+ rwdts.State.RUN: self.run,
+ }
+
+ # Transition application to next state
+ handler = handlers.get(state, None)
+ if handler is not None:
+ yield from handler()
+
+ # Transition dts to next state
+ next_state = switch.get(state, None)
+ if next_state is not None:
+ self._dts.handle.set_state(next_state)
diff --git a/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/xlate_cfg.py b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/xlate_cfg.py
new file mode 100644
index 0000000..add8a9a
--- /dev/null
+++ b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/xlate_cfg.py
@@ -0,0 +1,245 @@
+#!/usr/bin/env python3
+
+#
+# Copyright 2016 RIFT.IO Inc
+#
+# 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.
+#
+
+
+'''
+This script will go through the input conffiguration template and convert all the matching "regular expression" and "strings"
+specified in xlate_cp_list & xlate_str_list with matching IP addresses passed in as dictionary to this script.
+
+-i Configuration template
+-o Output final configuration complete with IP addresses
+-x Xlate(Translate dictionary in string format
+-t TAGS to be translated
+
+'''
+
+import sys
+import getopt
+import ast
+import re
+import yaml
+import netaddr
+
+from inspect import getsourcefile
+import os.path
+
+xlate_dict = None
+
+def xlate_cp_list(line, cp_list):
+ for cp_string in cp_list:
+ match = re.search(cp_string, line)
+ if match is not None:
+ # resolve IP address using Connection Point dictionary
+ resolved_ip = xlate_dict[match.group(1)]
+ if resolved_ip is None:
+ print("No matching CP found: ", match.group(1))
+ exit(2)
+ else:
+ line = line[:match.start()] + resolved_ip + line[match.end():]
+ return line
+
+def xlate_multi_colon_list(line, multi_colon_list):
+ for ucp_string in multi_colon_list:
+ #print("Searching :", ucp_string)
+ match = re.search(ucp_string, line)
+ if match is not None:
+ #print("match :", match.group())
+ # resolve IP address using Connection Point dictionary for specified member (unique) index
+ ucp_str_list = match.group(1).split(':')
+ print("matched = {}, split list = {}".format(match.group(1), ucp_str_list))
+ if len(ucp_str_list) != 3:
+ print("Invalid TAG in the configuration: ", match.group(1))
+ exit(2)
+
+ # Traslate given CP address & mask into netaddr
+ if ucp_string.startswith('<rw_unique_index:rw_connection_point:masklen'):
+ member_vnf_index = int(ucp_str_list[0])
+ resolved_ip = xlate_dict[ucp_str_list[1]]
+ masklen = ucp_str_list[2]
+ if resolved_ip is None:
+ print("No matching CP found: ", ucp_str_list[1])
+ exit(2)
+ if int(masklen) <= 0:
+ print("Invalid mask length: ", masklen)
+ exit(2)
+ else:
+ # Generate netaddr
+ ip_str = resolved_ip + '/' + masklen
+ #print("ip_str:", ip_str)
+ ip = netaddr.IPNetwork(ip_str)
+ if ucp_string.startswith('<rw_unique_index:rw_connection_point:masklen_broadcast'):
+ # Traslate given CP address & mask into broadcast address
+ addr = ip.broadcast
+ if ucp_string.startswith('<rw_unique_index:rw_connection_point:masklen_network'):
+ # Traslate given CP address & mask into network address
+ addr = ip.network
+ line = line[:match.start()] + str(addr) + line[match.end():]
+ return line
+
+
+
+def xlate_colon_list(line, colon_list):
+ for ucp_string in colon_list:
+ #print("Searching :", ucp_string)
+ match = re.search(ucp_string, line)
+ if match is not None:
+ #print("match :", match.group())
+ # resolve IP address using Connection Point dictionary for specified member (unique) index
+ ucp_str_list = match.group(1).split(':')
+ #print("matched = {}, split list = {}".format(match.group(1), ucp_str_list))
+ if len(ucp_str_list) != 2:
+ print("Invalid TAG in the configuration: ", match.group(1))
+ exit(2)
+
+ # Unique Connection Point translation to IP
+ if ucp_string.startswith('<rw_unique_index:'):
+ member_vnf_index = int(ucp_str_list[0])
+ resolved_ip = xlate_dict[member_vnf_index][ucp_str_list[1]]
+ #print("member_vnf_index = {}, resolved_ip = {}", member_vnf_index, resolved_ip)
+ if resolved_ip is None:
+ print("For Unique index ({}), No matching CP found: {}", ucp_str_list[0], ucp_str_list[1])
+ exit(2)
+ else:
+ line = line[:match.start()] + resolved_ip + line[match.end():]
+
+ # Traslate given CP address & mask into netaddr
+ if ucp_string.startswith('<rw_connection_point:masklen'):
+ resolved_ip = xlate_dict[ucp_str_list[0]]
+ masklen = ucp_str_list[1]
+ if resolved_ip is None:
+ print("No matching CP found: ", ucp_str_list[0])
+ exit(2)
+ if int(masklen) <= 0:
+ print("Invalid mask length: ", masklen)
+ exit(2)
+ else:
+ # Generate netaddr
+ ip_str = resolved_ip + '/' + masklen
+ #print("ip_str:", ip_str)
+ ip = netaddr.IPNetwork(ip_str)
+
+ if ucp_string.startswith('<rw_connection_point:masklen_broadcast'):
+ # Traslate given CP address & mask into broadcast address
+ addr = ip.broadcast
+ if ucp_string.startswith('<rw_connection_point:masklen_network'):
+ # Traslate given CP address & mask into network address
+ addr = ip.network
+
+ line = line[:match.start()] + str(addr) + line[match.end():]
+ return line
+
+def xlate_cp_to_tuple_list(line, cp_to_tuple_list):
+ for cp_string in cp_to_tuple_list:
+ match = re.search(cp_string, line)
+ if match is not None:
+ # resolve IP address using Connection Point dictionary
+ resolved_ip = xlate_dict[match.group(1)]
+ if resolved_ip is None:
+ print("No matching CP found: ", match.group(1))
+ exit(2)
+ else:
+ line = line[:match.start()] + match.group(1) + ':' + resolved_ip + line[match.end():]
+ return line
+
+def xlate_str_list(line, str_list):
+ for replace_tag in str_list:
+ replace_string = replace_tag[1:-1]
+ line = line.replace(replace_tag, xlate_dict[replace_string])
+ return line
+
+
+def main(argv=sys.argv[1:]):
+ cfg_template = None
+ cfg_file = None
+ global xlate_dict
+ try:
+ opts, args = getopt.getopt(argv,"i:o:x:")
+ except getopt.GetoptError:
+ print("Check arguments {}".format(argv))
+ sys.exit(2)
+ for opt, arg in opts:
+ if opt == '-i':
+ cfg_template = arg
+ elif opt in ("-o"):
+ cfg_file = arg
+ elif opt in ("-x"):
+ xlate_arg = arg
+
+ # Read TAGS from yaml file
+ # Read the translation tags from yaml file
+ yml_dir = os.path.dirname(os.path.abspath(getsourcefile(lambda:0)))
+ tags_input_file = os.path.join(yml_dir, 'xlate_tags.yml')
+ with open(tags_input_file, "r") as ti:
+ xlate_tags = yaml.load(ti.read())
+
+ # Need to work on the solution for multiple pattern of same type in single line.
+ try:
+ with open(xlate_arg, "r") as ti:
+ xlate_dict = yaml.load(ti.read())
+ try:
+ with open(cfg_template, 'r') as r:
+ try:
+ with open(cfg_file, 'w') as w:
+ # Traslate
+ try:
+ # For each line
+ for line in r:
+ if line.startswith("#"):
+ # Skip comment lines
+ continue
+ #print("1.Line : ", line)
+ # For each Connection Point translation to IP
+ line = xlate_cp_list(line, xlate_tags['xlate_cp_list'])
+ #print("2.Line : ", line)
+
+ # For each colon(:) separated tag, i.e. 3 inputs in a tag.
+ line = xlate_multi_colon_list(line, xlate_tags['xlate_multi_colon_list'])
+ #print("2a.Line : ", line)
+
+ # For each colon(:) separated tag, i.e. 2 inputs in a tag.
+ line = xlate_colon_list(line, xlate_tags['xlate_colon_list'])
+ #print("3.Line : ", line)
+
+ # For each connection point to tuple replacement
+ line = xlate_cp_to_tuple_list(line, xlate_tags['xlate_cp_to_tuple_list'])
+ #print("4.Line : ", line)
+
+ # For each direct replacement (currently only management IP address for ping/pong)
+ line = xlate_str_list(line, xlate_tags['xlate_str_list'])
+ #print("5.Line : ", line)
+
+ # Finally write the modified line to the new config file
+ w.write(line)
+ except Exception as e:
+ print("Error ({}) on line: {}".format(str(e), line))
+ exit(2)
+ except Exception as e:
+ print("Failed to open for write: {}, error({})".format(cfg_file, str(e)))
+ exit(2)
+ except Exception as e:
+ print("Failed to open for read: {}, error({})".format(cfg_template, str(e)))
+ exit(2)
+ print("Wrote configuration file", cfg_file)
+ except Exception as e:
+ print("Could not translate dictionary, error: ", str(e))
+
+if __name__ == "__main__":
+ try:
+ main()
+ except Exception as e:
+ print(str(e))
diff --git a/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/xlate_tags.yml b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/xlate_tags.yml
new file mode 100644
index 0000000..412e91e
--- /dev/null
+++ b/rwcm/plugins/rwconman/rift/tasklets/rwconmantasklet/xlate_tags.yml
@@ -0,0 +1,58 @@
+# """
+# #
+# Copyright 2016 RIFT.IO Inc
+#
+# 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.
+#
+
+# @file xlate_tags.yml
+# @author Manish Patel (Manish.Patel@riftio.com)
+# @date 01/14/2016
+# """
+
+# This file contains the tags that needs translation
+# One can add some tags with processing limitations by the translation script.
+
+# Add Regular expressions here (connection-points received dynamically from VNFR)
+
+# Translate connection point names (Connection point name is read using RegEx)
+
+xlate_cp_list :
+ - <rw_connection_point_name (.*?)>
+
+# Literal string translations
+xlate_str_list :
+ - <rw_mgmt_ip>
+ - <rw_username>
+ - <rw_password>
+
+# This list contains 2 tags separated by colon (:)
+xlate_colon_list :
+ # Fetch CP from the member_index dictionary (I.e. CP of a particular VNF)
+ - <rw_unique_index:rw_connection_point_name (.*?)>
+ # Generate network address from CP address and mask (mask is expected to be a hard coded number in config)
+ - <rw_connection_point:masklen_network (.*?)>
+ # Generate broadcast address from CP address and mask (mask is expected to be a hard coded number in config)
+ - <rw_connection_point:masklen_broadcast (.*?)>
+
+# This list contains 3 tags separated by colon (:)
+xlate_multi_colon_list :
+ # Generate network address from CP of a particular VNF (mask is expected to be a hard coded number in config))
+ - <rw_unique_index:rw_connection_point:masklen_network (.*?)>
+ # Generate broadcast address from CP of a particular VNF (mask is expected to be a hard coded number in config))
+ - <rw_unique_index:rw_connection_point:masklen_broadcast (.*?)>
+
+# This translates connection point name and generates tuple with name:resolved IP
+xlate_cp_to_tuple_list :
+ - <rw_connection_point_tuple (.*?)>
+
diff --git a/rwcm/plugins/rwconman/rwconmantasklet.py b/rwcm/plugins/rwconman/rwconmantasklet.py
new file mode 100755
index 0000000..796c4af
--- /dev/null
+++ b/rwcm/plugins/rwconman/rwconmantasklet.py
@@ -0,0 +1,27 @@
+
+#
+# Copyright 2016 RIFT.IO Inc
+#
+# 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.
+#
+
+# Workaround RIFT-6485 - rpmbuild defaults to python2 for
+# anything not in a site-packages directory so we have to
+# install the plugin implementation in site-packages and then
+# import it from the actual plugin.
+
+import rift.tasklets.rwconmantasklet
+class Tasklet(rift.tasklets.rwconmantasklet.ConfigManagerTasklet):
+ pass
+
+# vim: sw=4
diff --git a/rwcm/plugins/yang/CMakeLists.txt b/rwcm/plugins/yang/CMakeLists.txt
new file mode 100644
index 0000000..cdb2734
--- /dev/null
+++ b/rwcm/plugins/yang/CMakeLists.txt
@@ -0,0 +1,44 @@
+#
+# Copyright 2016 RIFT.IO Inc
+#
+# 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.
+#
+# Author(s): Manish Patel
+# Creation Date: 10/28/2015
+#
+
+##
+# Yang targets
+##
+
+rift_generate_python_log_yang(
+ LOG_CATEGORY_NAME rw-conman-log
+ START_EVENT_ID 66000
+ OUT_YANG_FILE_VAR rw_conman_log_file
+ )
+
+rift_add_yang_target(
+ TARGET rw_conman_yang
+ YANG_FILES rw-conman.yang ${rw_conman_log_file}
+ COMPONENT ${PKG_LONG_NAME}
+ LIBRARIES
+ mano_yang_gen
+ mano-types_yang_gen
+ rwconfig_agent_yang_gen
+ DEPENDS
+ mano_yang
+ rwconfig_agent_yang
+ mano-types_yang
+)
+
+
diff --git a/rwcm/plugins/yang/rw-conman.tailf.yang b/rwcm/plugins/yang/rw-conman.tailf.yang
new file mode 100644
index 0000000..aabbdd5
--- /dev/null
+++ b/rwcm/plugins/yang/rw-conman.tailf.yang
@@ -0,0 +1,38 @@
+
+/*
+ *
+ * Copyright 2016 RIFT.IO Inc
+ *
+ * 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.
+ *
+ *
+ */
+
+module rw-conman-annotation
+{
+ namespace "http://riftio.com/ns/riftware-1.0/rw-conman-annotation";
+ prefix "rw-conman-ann";
+
+ import tailf-common {
+ prefix tailf;
+ }
+
+ import rw-conman {
+ prefix conman;
+ }
+
+ tailf:annotate "/conman:cm-state" {
+ tailf:callpoint rw_callpoint;
+ }
+
+}
diff --git a/rwcm/plugins/yang/rw-conman.yang b/rwcm/plugins/yang/rw-conman.yang
new file mode 100644
index 0000000..bb1555d
--- /dev/null
+++ b/rwcm/plugins/yang/rw-conman.yang
@@ -0,0 +1,260 @@
+
+/*
+ *
+ * Copyright 2016 RIFT.IO Inc
+ *
+ * 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.
+ *
+ */
+
+
+
+/**
+ * @file rw-conman.yang
+ * @author Manish Patel
+ * @date 2015/10/27
+ * @brief Service Orchestrator configuration yang
+ */
+
+module rw-conman
+{
+ namespace "http://riftio.com/ns/riftware-1.0/rw-conman";
+ prefix "rw-conman";
+
+ import rw-pb-ext {
+ prefix "rwpb";
+ }
+
+ import rw-cli-ext {
+ prefix "rwcli";
+ }
+
+ import nsr {
+ prefix "nsr";
+ }
+
+ import vnfr {
+ prefix "vnfr";
+ }
+
+ import rw-vlr {
+ prefix "rwvlr";
+ }
+
+ import rw-yang-types {
+ prefix "rwt";
+ }
+
+ import ietf-inet-types {
+ prefix "inet";
+ }
+
+ import ietf-yang-types {
+ prefix "yang";
+ }
+
+ import mano-base {
+ prefix "manobase";
+ }
+
+ import mano-types {
+ prefix "manotypes";
+ }
+
+ import rw-config-agent {
+ prefix "rw-config-agent";
+ }
+
+ revision 2015-10-27 {
+ description
+ "Initial revision.";
+ }
+
+ // typedef ro-endpoint-method {
+ // type enumeration {
+ // enum netconf;
+ // enum restconf;
+ // }
+ // }
+
+ grouping ro-endpoint {
+ // leaf ro-endpoint-method {
+ // description "interface between CM & RO, defaults to netconf";
+ // type ro-endpoint-method;
+ // default netconf;
+ // }
+ leaf ro-ip-address {
+ type inet:ip-address;
+ description "IP Address";
+ default "127.0.0.1";
+ }
+ leaf ro-port {
+ type inet:port-number;
+ description "Port Number";
+ default 2022;
+ }
+ leaf ro-username {
+ description "RO endpoint username";
+ type string;
+ default "admin";
+ }
+ leaf ro-password {
+ description "RO endpoint password";
+ type string;
+ default "admin";
+ }
+ }
+
+ grouping vnf-cfg-items {
+ leaf configuration-file {
+ description "Location of the confguration file on CM system";
+ type string;
+ }
+ leaf translator-script {
+ description "Script that translates the templates in the configuration-file using VNFR information
+ Currently, we only use IP address translations.
+ configuration will use connection point name instead of IP addresses.";
+ type string;
+ }
+ }
+
+ container cm-config {
+ description "Service Orchestrator specific configuration";
+ rwpb:msg-new "SoConfig";
+ rwcli:new-mode "cm-config";
+
+ container ro-endpoint {
+ description "Resource Orchestrator endpoint ip address";
+ rwpb:msg-new "RoEndpoint";
+ uses ro-endpoint;
+ }
+
+ //uses vnf-cfg-items;
+
+ list nsr {
+ key "id";
+ leaf id {
+ description "Indicates NSR bringup complete, now initiate configuration of the NSR";
+ type yang:uuid;
+ }
+ }
+ }// cm-config
+
+ // =================== SHOW ==================
+ typedef record-state {
+ type enumeration {
+ enum init;
+ enum received;
+ enum cfg-delay;
+ enum cfg-process;
+ enum cfg-process-failed;
+ enum cfg-sched;
+ enum connecting;
+ enum failed-connection;
+ enum netconf-connected;
+ enum netconf-ssh-connected;
+ enum restconf-connected;
+ enum cfg-send;
+ enum cfg-failed;
+ enum ready-no-cfg;
+ enum ready;
+ }
+ }
+
+ // TBD: Do we need this typedef, currently not used anywhere
+ typedef cfg-type {
+ type enumeration {
+ enum none;
+ enum scriptconf;
+ enum netconf;
+ enum restconf;
+ enum jujuconf;
+ }
+ }
+
+
+ // This is also used by RO (Resource Orchestrator) to indicate NSR is ready
+ // It will only fill in IDs
+ container cm-state {
+ rwpb:msg-new "CmOpdata";
+ config false;
+ description "CM NS & VNF states";
+
+ leaf states {
+ description "CM various states";
+ type string;
+ }
+
+ list cm-nsr {
+ description "List of NS Records";
+ key "id";
+ leaf id {
+ type yang:uuid;
+ }
+ leaf name {
+ description "NSR name.";
+ type string;
+ }
+ leaf state {
+ description "State of NSR";
+ type record-state;
+ }
+ leaf state-details {
+ description "Details of the state of NSR, in case of errors";
+ type string;
+ }
+
+ list cm-vnfr {
+ description "List of VNF Records within NS Record";
+ key "id";
+ leaf id {
+ type yang:uuid;
+ }
+ leaf name {
+ description "VNFR name.";
+ type string;
+ }
+ leaf state {
+ description "Last known state of this VNFR";
+ type record-state;
+ }
+ container mgmt-interface {
+ leaf ip-address {
+ type inet:ip-address;
+ }
+ leaf port {
+ type inet:port-number;
+ }
+ }
+ leaf cfg-type {
+ type string;
+ }
+ leaf cfg-location {
+ type inet:uri;
+ }
+ list connection-point {
+ key "name";
+ leaf name {
+ description "Connection Point name";
+ type string;
+ }
+ leaf ip-address {
+ description "IP address assigned to this connection point";
+ type inet:ip-address;
+ }
+ }
+ } // list VNFR
+ } // list NSR
+ } // cm-state
+
+} // rw-conman
diff --git a/rwcm/test/CMakeLists.txt b/rwcm/test/CMakeLists.txt
new file mode 100644
index 0000000..ead05af
--- /dev/null
+++ b/rwcm/test/CMakeLists.txt
@@ -0,0 +1,39 @@
+#
+# Copyright 2016 RIFT.IO Inc
+#
+# 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.
+#
+# Author(s): Manish Patel
+# Creation Date: 10/28/2015
+#
+
+cmake_minimum_required(VERSION 2.8)
+
+set(CONMAN_INSTALL "demos/conman")
+
+install(
+ FILES
+ start_cm_system.py
+ README.start_cm
+ DESTINATION ${CONMAN_INSTALL}
+ COMPONENT ${PKG_LONG_NAME})
+
+# set(NS_NAME ping_pong_nsd)
+# install(
+# FILES
+# ${NS_NAME}/configuration_input_params.yml
+# ${NS_NAME}/ping_vnfd_1_scriptconf_template.cfg
+# ${NS_NAME}/pong_vnfd_11_scriptconf_template.cfg
+# DESTINATION ${CONMAN_INSTALL}/${NS_NAME}
+# COMPONENT ${PKG_LONG_NAME})
+
diff --git a/rwcm/test/README.start_cm b/rwcm/test/README.start_cm
new file mode 100644
index 0000000..7a8098b
--- /dev/null
+++ b/rwcm/test/README.start_cm
@@ -0,0 +1,4 @@
+# Following example command line to launch the system in collapse mode.
+# Please tailor for expanded mode or any other requirements
+
+./start_cm_system.py -m ethsim -c --skip-prepare-vm
diff --git a/rwcm/test/cwims_juju_nsd/configuration_input_params.yml b/rwcm/test/cwims_juju_nsd/configuration_input_params.yml
new file mode 100644
index 0000000..bbbe5bc
--- /dev/null
+++ b/rwcm/test/cwims_juju_nsd/configuration_input_params.yml
@@ -0,0 +1,20 @@
+
+# This is input parameters file for Network Service configuration.
+# This file is formatted as below:
+
+# configuration_delay : 120 # Number of seconds to wait before applying configuration after NS is up
+# number_of_vnfs_to_be_configured : 1 # Total number of VNFs in this NS to be configured by Service Orchestrator
+# 1 : # Configuration Priority, order in which each VNF will be configured
+# name : vnfd_name # Name of the VNF
+# member_vnf_index : 11 # member index of the VNF that makes it unique (in case of multiple instances of same VNF)
+# configuration_type : scriptconf # Type of configuration (Currently supported values : scriptconf, netconf)
+#
+# Repeat VNF block for as many VNFs
+
+configuration_delay : 30
+number_of_vnfs_to_be_configured : 1
+1 :
+ name : cwims_vnfd
+ member_vnf_index : 1
+ configuration_type : jujuconf
+
diff --git a/rwcm/test/cwims_juju_nsd/cwaio_vnfd_1_juju_template.cfg b/rwcm/test/cwims_juju_nsd/cwaio_vnfd_1_juju_template.cfg
new file mode 100644
index 0000000..d32efe3
--- /dev/null
+++ b/rwcm/test/cwims_juju_nsd/cwaio_vnfd_1_juju_template.cfg
@@ -0,0 +1,23 @@
+ims-a:
+ deploy:
+ store: local
+ directory: /usr/rift/charms/cw-aio-proxy/trusty/
+ series: trusty
+ to: "lxc:0"
+
+ # Data under config passed as such during deployment
+ config:
+ proxied_ip: <rw_mgmt_ip>
+ home_domain: "ims.riftio.local"
+ base_number: "1234567000"
+ number_count: 1000
+
+ units:
+ - unit:
+ # Wait for each command to complete
+ wait: true
+ # Bail on failure
+ bail: true
+ actions:
+ - create-user: { number: "1234567001", password: "secret"}
+ - create-user: { number: "1234567002", password: "secret"}
diff --git a/rwcm/test/ping_pong_nsd/configuration_input_params.yml b/rwcm/test/ping_pong_nsd/configuration_input_params.yml
new file mode 100644
index 0000000..47c4fc3
--- /dev/null
+++ b/rwcm/test/ping_pong_nsd/configuration_input_params.yml
@@ -0,0 +1,23 @@
+
+# This is input parameters file for Network Service configuration.
+# This file is formatted as below:
+
+# configuration_delay : 120 # Number of seconds to wait before applying configuration after NS is up
+# number_of_vnfs_to_be_configured : 1 # Total number of VNFs in this NS to be configured by Service Orchestrator
+# 1 : # Configuration Priority, order in which each VNF will be configured
+# name : vnfd_name # Name of the VNF
+# member_vnf_index : 11 # member index of the VNF that makes it unique (in case of multiple instances of same VNF)
+# configuration_type : scriptconf # Type of configuration (Currently supported values : scriptconf, netconf)
+#
+# Repeat VNF block for as many VNFs
+
+configuration_delay : 30
+number_of_vnfs_to_be_configured : 2
+1 :
+ name : pong_vnfd
+ member_vnf_index : 2
+ configuration_type : scriptconf
+2 :
+ name : ping_vnfd
+ member_vnf_index : 1
+ configuration_type : scriptconf
diff --git a/rwcm/test/ping_pong_nsd/ping_vnfd_1_scriptconf_template.cfg b/rwcm/test/ping_pong_nsd/ping_vnfd_1_scriptconf_template.cfg
new file mode 100755
index 0000000..e6e9889
--- /dev/null
+++ b/rwcm/test/ping_pong_nsd/ping_vnfd_1_scriptconf_template.cfg
@@ -0,0 +1,54 @@
+#!/bin/bash
+
+# Rest API config
+ping_mgmt_ip='<rw_mgmt_ip>'
+ping_mgmt_port=18888
+
+# VNF specific configuration
+pong_server_ip='<rw_connection_point_name pong_vnfd/cp0>'
+ping_rate=5
+server_port=5555
+
+# Make rest API calls to configure VNF
+curl -D /dev/stdout \
+ -H "Accept: application/vnd.yang.data+xml" \
+ -H "Content-Type: application/vnd.yang.data+json" \
+ -X POST \
+ -d "{\"ip\":\"$pong_server_ip\", \"port\":$server_port}" \
+ http://${ping_mgmt_ip}:${ping_mgmt_port}/api/v1/ping/server
+rc=$?
+if [ $rc -ne 0 ]
+then
+ echo "Failed to set server info for ping!"
+ exit $rc
+fi
+
+curl -D /dev/stdout \
+ -H "Accept: application/vnd.yang.data+xml" \
+ -H "Content-Type: application/vnd.yang.data+json" \
+ -X POST \
+ -d "{\"rate\":$ping_rate}" \
+ http://${ping_mgmt_ip}:${ping_mgmt_port}/api/v1/ping/rate
+rc=$?
+if [ $rc -ne 0 ]
+then
+ echo "Failed to set ping rate!"
+ exit $rc
+fi
+
+output=$(curl -D /dev/stdout \
+ -H "Accept: application/vnd.yang.data+xml" \
+ -H "Content-Type: application/vnd.yang.data+json" \
+ -X POST \
+ -d "{\"enable\":true}" \
+ http://${ping_mgmt_ip}:${ping_mgmt_port}/api/v1/ping/adminstatus/state)
+if [[ $output == *"Internal Server Error"* ]]
+then
+ echo $output
+ exit 3
+else
+ echo $output
+fi
+
+
+exit 0
diff --git a/rwcm/test/ping_pong_nsd/pong_vnfd_11_scriptconf_template.cfg b/rwcm/test/ping_pong_nsd/pong_vnfd_11_scriptconf_template.cfg
new file mode 100755
index 0000000..28b01df
--- /dev/null
+++ b/rwcm/test/ping_pong_nsd/pong_vnfd_11_scriptconf_template.cfg
@@ -0,0 +1,42 @@
+#!/bin/bash
+
+# Rest API configuration
+pong_mgmt_ip='<rw_mgmt_ip>'
+pong_mgmt_port=18889
+
+# Test
+# username=<rw_username>
+# password=<rw_password>
+
+# VNF specific configuration
+pong_server_ip='<rw_connection_point_name pong_vnfd/cp0>'
+server_port=5555
+
+# Make Rest API calls to configure VNF
+curl -D /dev/stdout \
+ -H "Accept: application/vnd.yang.data+xml" \
+ -H "Content-Type: application/vnd.yang.data+json" \
+ -X POST \
+ -d "{\"ip\":\"$pong_server_ip\", \"port\":$server_port}" \
+ http://${pong_mgmt_ip}:${pong_mgmt_port}/api/v1/pong/server
+rc=$?
+if [ $rc -ne 0 ]
+then
+ echo "Failed to set server(own) info for pong!"
+ exit $rc
+fi
+
+curl -D /dev/stdout \
+ -H "Accept: application/vnd.yang.data+xml" \
+ -H "Content-Type: application/vnd.yang.data+json" \
+ -X POST \
+ -d "{\"enable\":true}" \
+ http://${pong_mgmt_ip}:${pong_mgmt_port}/api/v1/pong/adminstatus/state
+rc=$?
+if [ $rc -ne 0 ]
+then
+ echo "Failed to enable pong service!"
+ exit $rc
+fi
+
+exit 0
diff --git a/rwcm/test/rwso_test.py b/rwcm/test/rwso_test.py
new file mode 100755
index 0000000..e0c5011
--- /dev/null
+++ b/rwcm/test/rwso_test.py
@@ -0,0 +1,349 @@
+#!/usr/bin/env python3
+
+#
+# Copyright 2016 RIFT.IO Inc
+#
+# 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 asyncio
+import logging
+import os
+import sys
+import types
+import unittest
+import uuid
+
+import xmlrunner
+
+import gi.repository.CF as cf
+import gi.repository.RwDts as rwdts
+import gi.repository.RwMain as rwmain
+import gi.repository.RwManifestYang as rwmanifest
+import gi.repository.RwConmanYang as conmanY
+import gi.repository.RwLaunchpadYang as launchpadyang
+
+import rift.tasklets
+
+if sys.version_info < (3, 4, 4):
+ asyncio.ensure_future = asyncio.async
+
+
+class RWSOTestCase(unittest.TestCase):
+ """
+ DTS GI interface unittests
+
+ Note: Each tests uses a list of asyncio.Events for staging through the
+ test. These are required here because we are bring up each coroutine
+ ("tasklet") at the same time and are not implementing any re-try
+ mechanisms. For instance, this is used in numerous tests to make sure that
+ a publisher is up and ready before the subscriber sends queries. Such
+ event lists should not be used in production software.
+ """
+ rwmain = None
+ tinfo = None
+ schema = None
+ id_cnt = 0
+
+ @classmethod
+ def setUpClass(cls):
+ msgbroker_dir = os.environ.get('MESSAGE_BROKER_DIR')
+ router_dir = os.environ.get('ROUTER_DIR')
+ cm_dir = os.environ.get('SO_DIR')
+
+ manifest = rwmanifest.Manifest()
+ manifest.init_phase.settings.rwdtsrouter.single_dtsrouter.enable = True
+
+ cls.rwmain = rwmain.Gi.new(manifest)
+ cls.tinfo = cls.rwmain.get_tasklet_info()
+
+ # Run router in mainq. Eliminates some ill-diagnosed bootstrap races.
+ os.environ['RWDTS_ROUTER_MAINQ']='1'
+ cls.rwmain.add_tasklet(msgbroker_dir, 'rwmsgbroker-c')
+ cls.rwmain.add_tasklet(router_dir, 'rwdtsrouter-c')
+ cls.rwmain.add_tasklet(cm_dir, 'rwconmantasklet')
+
+ cls.log = rift.tasklets.logger_from_tasklet_info(cls.tinfo)
+ cls.log.setLevel(logging.DEBUG)
+
+ stderr_handler = logging.StreamHandler(stream=sys.stderr)
+ fmt = logging.Formatter(
+ '%(asctime)-23s %(levelname)-5s (%(name)s@%(process)d:%(filename)s:%(lineno)d) - %(message)s')
+ stderr_handler.setFormatter(fmt)
+ cls.log.addHandler(stderr_handler)
+ cls.schema = conmanY.get_schema()
+
+ def setUp(self):
+ def scheduler_tick(self, *args):
+ self.call_soon(self.stop)
+ self.run_forever()
+
+ self.loop = asyncio.new_event_loop()
+ self.loop.scheduler_tick = types.MethodType(scheduler_tick, self.loop)
+ self.loop.set_debug(True)
+ os.environ["PYTHONASYNCIODEBUG"] = "1"
+ asyncio_logger = logging.getLogger("asyncio")
+ asyncio_logger.setLevel(logging.DEBUG)
+
+ self.asyncio_timer = None
+ self.stop_timer = None
+ self.id_cnt += 1
+
+ @asyncio.coroutine
+ def wait_tasklets(self):
+ yield from asyncio.sleep(1, loop=self.loop)
+
+ def run_until(self, test_done, timeout=30):
+ """
+ Attach the current asyncio event loop to rwsched and then run the
+ scheduler until the test_done function returns True or timeout seconds
+ pass.
+
+ @param test_done - function which should return True once the test is
+ complete and the scheduler no longer needs to run.
+ @param timeout - maximum number of seconds to run the test.
+ """
+ def shutdown(*args):
+ if args:
+ self.log.debug('Shutting down loop due to timeout')
+
+ if self.asyncio_timer is not None:
+ self.tinfo.rwsched_tasklet.CFRunLoopTimerRelease(self.asyncio_timer)
+ self.asyncio_timer = None
+
+ if self.stop_timer is not None:
+ self.tinfo.rwsched_tasklet.CFRunLoopTimerRelease(self.stop_timer)
+ self.stop_timer = None
+
+ self.tinfo.rwsched_instance.CFRunLoopStop()
+
+ def tick(*args):
+ self.loop.call_later(0.1, self.loop.stop)
+ self.loop.run_forever()
+ if test_done():
+ shutdown()
+
+ self.asyncio_timer = self.tinfo.rwsched_tasklet.CFRunLoopTimer(
+ cf.CFAbsoluteTimeGetCurrent(),
+ 0.1,
+ tick,
+ None)
+
+ self.stop_timer = self.tinfo.rwsched_tasklet.CFRunLoopTimer(
+ cf.CFAbsoluteTimeGetCurrent() + timeout,
+ 0,
+ shutdown,
+ None)
+
+ self.tinfo.rwsched_tasklet.CFRunLoopAddTimer(
+ self.tinfo.rwsched_tasklet.CFRunLoopGetCurrent(),
+ self.stop_timer,
+ self.tinfo.rwsched_instance.CFRunLoopGetMainMode())
+
+ self.tinfo.rwsched_tasklet.CFRunLoopAddTimer(
+ self.tinfo.rwsched_tasklet.CFRunLoopGetCurrent(),
+ self.asyncio_timer,
+ self.tinfo.rwsched_instance.CFRunLoopGetMainMode())
+
+ self.tinfo.rwsched_instance.CFRunLoopRun()
+
+ self.assertTrue(test_done())
+
+ def new_tinfo(self, name):
+ """
+ Create a new tasklet info instance with a unique instance_id per test.
+ It is up to each test to use unique names if more that one tasklet info
+ instance is needed.
+
+ @param name - name of the "tasklet"
+ @return - new tasklet info instance
+ """
+ ret = self.rwmain.new_tasklet_info(name, RWSOTestCase.id_cnt)
+
+ log = rift.tasklets.logger_from_tasklet_info(ret)
+ log.setLevel(logging.DEBUG)
+
+ stderr_handler = logging.StreamHandler(stream=sys.stderr)
+ fmt = logging.Formatter(
+ '%(asctime)-23s %(levelname)-5s (%(name)s@%(process)d:%(filename)s:%(lineno)d) - %(message)s')
+ stderr_handler.setFormatter(fmt)
+ log.addHandler(stderr_handler)
+
+ return ret
+
+ def get_cloud_account_msg(self):
+ cloud_account = launchpadyang.CloudAccount()
+ cloud_account.name = "cloudy"
+ cloud_account.account_type = "mock"
+ cloud_account.mock.username = "rainy"
+ return cloud_account
+
+ def get_compute_pool_msg(self, name, pool_type):
+ pool_config = rmgryang.ResourcePools()
+ pool = pool_config.pools.add()
+ pool.name = name
+ pool.resource_type = "compute"
+ if pool_type == "static":
+ # Need to query CAL for resource
+ pass
+ else:
+ pool.max_size = 10
+ return pool_config
+
+ def get_network_pool_msg(self, name, pool_type):
+ pool_config = rmgryang.ResourcePools()
+ pool = pool_config.pools.add()
+ pool.name = name
+ pool.resource_type = "network"
+ if pool_type == "static":
+ # Need to query CAL for resource
+ pass
+ else:
+ pool.max_size = 4
+ return pool_config
+
+
+ def get_network_reserve_msg(self, xpath):
+ event_id = str(uuid.uuid4())
+ msg = rmgryang.VirtualLinkEventData()
+ msg.event_id = event_id
+ msg.request_info.name = "mynet"
+ msg.request_info.subnet = "1.1.1.0/24"
+ return msg, xpath.format(event_id)
+
+ def get_compute_reserve_msg(self,xpath):
+ event_id = str(uuid.uuid4())
+ msg = rmgryang.VDUEventData()
+ msg.event_id = event_id
+ msg.request_info.name = "mynet"
+ msg.request_info.image_id = "This is a image_id"
+ msg.request_info.vm_flavor.vcpu_count = 4
+ msg.request_info.vm_flavor.memory_mb = 8192*2
+ msg.request_info.vm_flavor.storage_gb = 40
+ c1 = msg.request_info.connection_points.add()
+ c1.name = "myport1"
+ c1.virtual_link_id = "This is a network_id"
+ return msg, xpath.format(event_id)
+
+ def test_create_resource_pools(self):
+ self.log.debug("STARTING - test_create_resource_pools")
+ tinfo = self.new_tinfo('poolconfig')
+ dts = rift.tasklets.DTS(tinfo, self.schema, self.loop)
+ pool_xpath = "C,/rw-resource-mgr:resource-mgr-config/rw-resource-mgr:resource-pools"
+ pool_records_xpath = "D,/rw-resource-mgr:resource-pool-records"
+ account_xpath = "C,/rw-launchpad:cloud-account"
+ compute_xpath = "D,/rw-resource-mgr:resource-mgmt/vdu-event/vdu-event-data[event-id='{}']"
+ network_xpath = "D,/rw-resource-mgr:resource-mgmt/vlink-event/vlink-event-data[event-id='{}']"
+
+ @asyncio.coroutine
+ def configure_cloud_account():
+ msg = self.get_cloud_account_msg()
+ self.log.info("Configuring cloud-account: %s",msg)
+ yield from dts.query_create(account_xpath,
+ rwdts.XactFlag.ADVISE,
+ msg)
+ yield from asyncio.sleep(3, loop=self.loop)
+
+ @asyncio.coroutine
+ def configure_compute_resource_pools():
+ msg = self.get_compute_pool_msg("virtual-compute", "dynamic")
+ self.log.info("Configuring compute-resource-pool: %s",msg)
+ yield from dts.query_create(pool_xpath,
+ rwdts.XactFlag.ADVISE,
+ msg)
+ yield from asyncio.sleep(3, loop=self.loop)
+
+
+ @asyncio.coroutine
+ def configure_network_resource_pools():
+ msg = self.get_network_pool_msg("virtual-network", "dynamic")
+ self.log.info("Configuring network-resource-pool: %s",msg)
+ yield from dts.query_create(pool_xpath,
+ rwdts.XactFlag.ADVISE,
+ msg)
+ yield from asyncio.sleep(3, loop=self.loop)
+
+
+ @asyncio.coroutine
+ def verify_resource_pools():
+ self.log.debug("Verifying test_create_resource_pools results")
+ res_iter = yield from dts.query_read(pool_records_xpath,)
+ for result in res_iter:
+ response = yield from result
+ records = response.result.records
+ #self.assertEqual(len(records), 2)
+ #names = [i.name for i in records]
+ #self.assertTrue('virtual-compute' in names)
+ #self.assertTrue('virtual-network' in names)
+ for record in records:
+ self.log.debug("Received Pool Record, Name: %s, Resource Type: %s, Pool Status: %s, Pool Size: %d, Busy Resources: %d",
+ record.name,
+ record.resource_type,
+ record.pool_status,
+ record.max_size,
+ record.busy_resources)
+ @asyncio.coroutine
+ def reserve_network_resources():
+ msg,xpath = self.get_network_reserve_msg(network_xpath)
+ self.log.debug("Sending create event to network-event xpath %s with msg: %s" % (xpath, msg))
+ yield from dts.query_create(xpath, rwdts.XactFlag.TRACE, msg)
+ yield from asyncio.sleep(3, loop=self.loop)
+ yield from dts.query_delete(xpath, rwdts.XactFlag.TRACE)
+
+ @asyncio.coroutine
+ def reserve_compute_resources():
+ msg,xpath = self.get_compute_reserve_msg(compute_xpath)
+ self.log.debug("Sending create event to compute-event xpath %s with msg: %s" % (xpath, msg))
+ yield from dts.query_create(xpath, rwdts.XactFlag.TRACE, msg)
+ yield from asyncio.sleep(3, loop=self.loop)
+ yield from dts.query_delete(xpath, rwdts.XactFlag.TRACE)
+
+ @asyncio.coroutine
+ def run_test():
+ yield from self.wait_tasklets()
+ yield from configure_cloud_account()
+ yield from configure_compute_resource_pools()
+ yield from configure_network_resource_pools()
+ yield from verify_resource_pools()
+ yield from reserve_network_resources()
+ yield from reserve_compute_resources()
+
+ future = asyncio.ensure_future(run_test(), loop=self.loop)
+ self.run_until(future.done)
+ if future.exception() is not None:
+ self.log.error("Caught exception during test")
+ raise future.exception()
+
+ self.log.debug("DONE - test_create_resource_pools")
+
+
+def main():
+ plugin_dir = os.path.join(os.environ["RIFT_INSTALL"], "usr/lib/rift/plugins")
+
+ if 'MESSAGE_BROKER_DIR' not in os.environ:
+ os.environ['MESSAGE_BROKER_DIR'] = os.path.join(plugin_dir, 'rwmsgbroker-c')
+
+ if 'ROUTER_DIR' not in os.environ:
+ os.environ['ROUTER_DIR'] = os.path.join(plugin_dir, 'rwdtsrouter-c')
+
+ if 'SO_DIR' not in os.environ:
+ os.environ['SO_DIR'] = os.path.join(plugin_dir, 'rwconmantasklet')
+
+ runner = xmlrunner.XMLTestRunner(output=os.environ["RIFT_MODULE_TEST"])
+ unittest.main(testRunner=runner)
+
+if __name__ == '__main__':
+ main()
+
+# vim: sw=4
diff --git a/rwcm/test/start_cm_system.py b/rwcm/test/start_cm_system.py
new file mode 100755
index 0000000..1975a0a
--- /dev/null
+++ b/rwcm/test/start_cm_system.py
@@ -0,0 +1,131 @@
+#!/usr/bin/env python
+
+#
+# Copyright 2016 RIFT.IO Inc
+#
+# 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 logging
+import os
+import sys
+
+import rift.vcs
+import rift.vcs.demo
+import rift.vcs.vms
+
+from rift.vcs.ext import ClassProperty
+
+logger = logging.getLogger(__name__)
+
+
+class ConfigManagerTasklet(rift.vcs.core.Tasklet):
+ """
+ This class represents SO tasklet.
+ """
+
+ def __init__(self, name='rwcmtasklet', uid=None):
+ """
+ Creates a PingTasklet object.
+
+ Arguments:
+ name - the name of the tasklet
+ uid - a unique identifier
+ """
+ super(ConfigManagerTasklet, self).__init__(name=name, uid=uid)
+
+ plugin_directory = ClassProperty('./usr/lib/rift/plugins/rwconmantasklet')
+ plugin_name = ClassProperty('rwconmantasklet')
+
+
+# Construct the system. This system consists of 1 cluster in 1
+# colony. The master cluster houses CLI and management VMs
+sysinfo = rift.vcs.SystemInfo(
+ colonies=[
+ rift.vcs.Colony(
+ clusters=[
+ rift.vcs.Cluster(
+ name='master',
+ virtual_machines=[
+ rift.vcs.VirtualMachine(
+ name='vm-so',
+ ip='127.0.0.1',
+ tasklets=[
+ rift.vcs.uAgentTasklet(),
+ ],
+ procs=[
+ rift.vcs.CliTasklet(),
+ rift.vcs.DtsRouterTasklet(),
+ rift.vcs.MsgBrokerTasklet(),
+ rift.vcs.RestconfTasklet(),
+ ConfigManagerTasklet()
+ ],
+ ),
+ ]
+ )
+ ]
+ )
+ ]
+ )
+
+
+# Define the generic portmap.
+port_map = {}
+
+
+# Define a mapping from the placeholder logical names to the real
+# port names for each of the different modes supported by this demo.
+port_names = {
+ 'ethsim': {
+ },
+ 'pci': {
+ }
+}
+
+
+# Define the connectivity between logical port names.
+port_groups = {}
+
+def main(argv=sys.argv[1:]):
+ logging.basicConfig(format='%(asctime)-15s %(levelname)s %(message)s')
+
+ # Create a parser which includes all generic demo arguments
+ parser = rift.vcs.demo.DemoArgParser()
+
+ args = parser.parse_args(argv)
+
+ #load demo info and create Demo object
+ demo = rift.vcs.demo.Demo(sysinfo=sysinfo,
+ port_map=port_map,
+ port_names=port_names,
+ port_groups=port_groups)
+
+ # Create the prepared system from the demo
+ system = rift.vcs.demo.prepared_system_from_demo_and_args(demo, args, netconf_trace_override=True)
+
+ # Start the prepared system
+ system.start()
+
+
+if __name__ == "__main__":
+ try:
+ main()
+ except rift.vcs.demo.ReservationError:
+ print("ERROR: unable to retrieve a list of IP addresses from the reservation system")
+ sys.exit(1)
+ except rift.vcs.demo.MissingModeError:
+ print("ERROR: you need to provide a mode to run the script")
+ sys.exit(1)
+ finally:
+ os.system("stty sane")
diff --git a/rwcm/test/tg_vrouter_ts_nsd/configuration_input_params.yml b/rwcm/test/tg_vrouter_ts_nsd/configuration_input_params.yml
new file mode 100644
index 0000000..b5a126f
--- /dev/null
+++ b/rwcm/test/tg_vrouter_ts_nsd/configuration_input_params.yml
@@ -0,0 +1,23 @@
+
+# This is input parameters file for Network Service configuration.
+# This file is formatted as below:
+
+# configuration_delay : 120 # Number of seconds to wait before applying configuration after NS is up
+# number_of_vnfs_to_be_configured : 1 # Total number of VNFs in this NS to be configured by Service Orchestrator
+# 1 : # Configuration Priority, order in which each VNF will be configured
+# name : vnfd_name # Name of the VNF
+# member_vnf_index : 11 # member index of the VNF that makes it unique (in case of multiple instances of same VNF)
+# configuration_type : scriptconf # Type of configuration (Currently supported values : scriptconf, netconf)
+#
+# Repeat VNF block for as many VNFs
+
+configuration_delay : 120
+number_of_vnfs_to_be_configured : 2
+1 :
+ name : trafsink_vnfd
+ member_vnf_index : 3
+ configuration_type : netconf
+2 :
+ name : trafgen_vnfd
+ member_vnf_index : 1
+ configuration_type : netconf
diff --git a/rwcm/test/tg_vrouter_ts_nsd/trafgen_vnfd_1_netconf_template.cfg b/rwcm/test/tg_vrouter_ts_nsd/trafgen_vnfd_1_netconf_template.cfg
new file mode 100644
index 0000000..02dfc85
--- /dev/null
+++ b/rwcm/test/tg_vrouter_ts_nsd/trafgen_vnfd_1_netconf_template.cfg
@@ -0,0 +1,79 @@
+ <vnf-config xmlns="http://riftio.com/ns/riftware-1.0/mano-base">
+ <vnf>
+ <name>trafgen</name>
+ <instance>0</instance>
+ <network-context xmlns="http://riftio.com/ns/riftware-1.0/rw-vnf-base-config">
+ <name>trafgen-lb</name>
+ <interface>
+ <name>N1TenGi-1</name>
+ <bind>
+ <port>trafgen_vnfd/cp0</port>
+ </bind>
+ </interface>
+ </network-context>
+ <port xmlns="http://riftio.com/ns/riftware-1.0/rw-vnf-base-config">
+ <name>trafgen_vnfd/cp0</name>
+ <open/>
+ <application>
+ <rx>rw_trafgen</rx>
+ <tx>rw_trafgen</tx>
+ </application>
+ <receive-q-length>2</receive-q-length>
+ <port-identity>
+ <ip-address><rw_connection_point_name trafgen_vnfd/cp0></ip-address>
+ <port-mode>direct</port-mode>
+ </port-identity>
+ <trafgen xmlns="http://riftio.com/ns/riftware-1.0/rw-trafgen">
+ <transmit-params>
+ <transmit-mode>
+ <range/>
+ </transmit-mode>
+ </transmit-params>
+ <range-template>
+ <destination-mac>
+ <dynamic>
+ <gateway><rw_connection_point_name vrouter_vnfd/cp0></gateway>
+ </dynamic>
+ </destination-mac>
+ <source-ip>
+ <start><rw_connection_point_name trafgen_vnfd/cp0></start>
+ <minimum><rw_connection_point_name trafgen_vnfd/cp0></minimum>
+ <maximum><rw_connection_point_name trafgen_vnfd/cp0></maximum>
+ <increment>1</increment>
+ </source-ip>
+ <destination-ip>
+ <start><rw_connection_point_name trafsink_vnfd/cp0></start>
+ <minimum><rw_connection_point_name trafsink_vnfd/cp0></minimum>
+ <maximum><rw_connection_point_name trafsink_vnfd/cp0></maximum>
+ <increment>1</increment>
+ </destination-ip>
+ <source-port>
+ <start>10000</start>
+ <minimum>10000</minimum>
+ <maximum>10128</maximum>
+ <increment>1</increment>
+ </source-port>
+ <destination-port>
+ <start>5678</start>
+ <minimum>5678</minimum>
+ <maximum>5678</maximum>
+ <increment>1</increment>
+ </destination-port>
+ <packet-size>
+ <start>512</start>
+ <minimum>512</minimum>
+ <maximum>512</maximum>
+ <increment>1</increment>
+ </packet-size>
+ </range-template>
+ </trafgen>
+ </port>
+ </vnf>
+ </vnf-config>
+ <logging xmlns="http://riftio.com/ns/riftware-1.0/rwlog-mgmt">
+ <sink>
+ <name>syslog</name>
+ <server-address><rw_mgmt_ip></server-address>
+ <port>514</port>
+ </sink>
+ </logging>
diff --git a/rwcm/test/tg_vrouter_ts_nsd/trafsink_vnfd_3_netconf_template.cfg b/rwcm/test/tg_vrouter_ts_nsd/trafsink_vnfd_3_netconf_template.cfg
new file mode 100644
index 0000000..6402201
--- /dev/null
+++ b/rwcm/test/tg_vrouter_ts_nsd/trafsink_vnfd_3_netconf_template.cfg
@@ -0,0 +1,42 @@
+ <vnf-config xmlns="http://riftio.com/ns/riftware-1.0/mano-base">
+ <vnf>
+ <name>trafsink</name>
+ <instance>0</instance>
+ <network-context xmlns="http://riftio.com/ns/riftware-1.0/rw-vnf-base-config">
+ <name>lb-trafsink</name>
+ <interface>
+ <name>N3TenGigi-1</name>
+ <bind>
+ <port>trafsink_vnfd/cp0</port>
+ </bind>
+ </interface>
+ </network-context>
+ <port xmlns="http://riftio.com/ns/riftware-1.0/rw-vnf-base-config">
+ <name>trafsink_vnfd/cp0</name>
+ <open/>
+ <application>
+ <rx>rw_trafgen</rx>
+ <tx>rw_trafgen</tx>
+ </application>
+ <receive-q-length>2</receive-q-length>
+ <port-identity>
+ <ip-address><rw_connection_point_name trafsink_vnfd/cp0></ip-address>
+ <port-mode>direct</port-mode>
+ </port-identity>
+ <trafgen xmlns="http://riftio.com/ns/riftware-1.0/rw-trafgen">
+ <receive-param>
+ <receive-echo>
+ <on/>
+ </receive-echo>
+ </receive-param>
+ </trafgen>
+ </port>
+ </vnf>
+ </vnf-config>
+ <logging xmlns="http://riftio.com/ns/riftware-1.0/rwlog-mgmt">
+ <sink>
+ <name>syslog</name>
+ <server-address><rw_mgmt_ip></server-address>
+ <port>514</port>
+ </sink>
+ </logging>