mock.Mock(return_value=(0, {}, ""))
ssh.from_node.return_value = ssh_mock
+ mgmt = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]['mgmt-interface']
+ interfaces = \
+ self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]['vdu'][0]['external-interface']
self.resource_profile = \
- ResourceProfile(self.VNFD['vnfd:vnfd-catalog']['vnfd'][0],
- [1, 2, 3])
+ ResourceProfile(mgmt, interfaces, [1, 2, 3])
def test___init__(self):
self.assertEqual(True, self.resource_profile.enable)
reskey = ["", "cpufreq", "cpufreq-0"]
value = "metric:10"
val = self.resource_profile.get_cpu_data(reskey, value)
- self.assertEqual(val, ['0', 'cpufreq', '10', 'metric'])
+ self.assertIsNotNone(val)
def test_get_cpu_data_error(self):
reskey = ["", "", ""]
value = "metric:10"
val = self.resource_profile.get_cpu_data(reskey, value)
- self.assertEqual(val, ['error', 'Invalid', ''])
+ self.assertEqual(val, ('error', 'Invalid', '', ''))
def test__start_collectd(self):
with mock.patch("yardstick.ssh.SSH") as ssh:
ssh_mock.execute = \
mock.Mock(return_value=(0, "", ""))
ssh.from_node.return_value = ssh_mock
+ mgmt = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]['mgmt-interface']
+ interfaces = \
+ self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]['vdu'][0]['external-interface']
resource_profile = \
- ResourceProfile(self.VNFD['vnfd:vnfd-catalog']['vnfd'][0],
- [1, 2, 3])
+ ResourceProfile(mgmt, interfaces, [1, 2, 3])
+ resource_profile._prepare_collectd_conf = mock.Mock()
self.assertIsNone(
resource_profile._start_collectd(ssh_mock, "/opt/nsb_bin"))
- def test_initiate_systemagent(self):
+ def test__prepare_collectd_conf_BM(self):
with mock.patch("yardstick.ssh.SSH") as ssh:
ssh_mock = mock.Mock(autospec=ssh.SSH)
ssh_mock.execute = \
mock.Mock(return_value=(0, "", ""))
ssh.from_node.return_value = ssh_mock
+ mgmt = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]['mgmt-interface']
+ interfaces = \
+ self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]['vdu'][0]['external-interface']
resource_profile = \
- ResourceProfile(self.VNFD['vnfd:vnfd-catalog']['vnfd'][0],
- [1, 2, 3])
+ ResourceProfile(mgmt, interfaces, [1, 2, 3])
+ resource_profile._provide_config_file = mock.Mock()
+ self.assertIsNone(
+ resource_profile._prepare_collectd_conf("/opt/nsb_bin"))
+
+ def test__prepare_collectd_conf_managed_ovs_dpdk(self):
+ with mock.patch("yardstick.ssh.SSH") as ssh:
+ ssh_mock = mock.Mock(autospec=ssh.SSH)
+ ssh_mock.execute = \
+ mock.Mock(return_value=(0, "", ""))
+ ssh.from_node.return_value = ssh_mock
+ mgmt = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]['mgmt-interface']
+ interfaces = \
+ self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]['vdu'][0]['external-interface']
+ resource_profile = \
+ ResourceProfile(mgmt, interfaces, [1, 2, 3])
+ resource_profile._provide_config_file = mock.Mock()
+ self.assertIsNone(
+ resource_profile._prepare_collectd_conf("/opt/nsb_bin"))
+
+ def test__prepare_collectd_conf_ovs_dpdk(self):
+ with mock.patch("yardstick.ssh.SSH") as ssh:
+ ssh_mock = mock.Mock(autospec=ssh.SSH)
+ ssh_mock.execute = \
+ mock.Mock(return_value=(0, "", ""))
+ ssh.from_node.return_value = ssh_mock
+ mgmt = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]['mgmt-interface']
+ interfaces = \
+ self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]['vdu'][0]['external-interface']
+ resource_profile = \
+ ResourceProfile(mgmt, interfaces, [1, 2, 3])
+ resource_profile._provide_config_file = mock.Mock()
+ self.assertIsNone(
+ resource_profile._prepare_collectd_conf("/opt/nsb_bin"))
+
+ def test__prepare_collectd_conf_managed_sriov(self):
+ with mock.patch("yardstick.ssh.SSH") as ssh:
+ ssh_mock = mock.Mock(autospec=ssh.SSH)
+ ssh_mock.execute = \
+ mock.Mock(return_value=(0, "", ""))
+ ssh.from_node.return_value = ssh_mock
+ mgmt = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]['mgmt-interface']
+ interfaces = \
+ self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]['vdu'][0]['external-interface']
+ resource_profile = \
+ ResourceProfile(mgmt, interfaces, [1, 2, 3])
+ resource_profile._provide_config_file = mock.Mock()
+ self.assertIsNone(
+ resource_profile._prepare_collectd_conf("/opt/nsb_bin"))
+
+ def test__prepare_collectd_conf_sriov(self):
+ with mock.patch("yardstick.ssh.SSH") as ssh:
+ ssh_mock = mock.Mock(autospec=ssh.SSH)
+ ssh_mock.execute = \
+ mock.Mock(return_value=(0, "", ""))
+ ssh.from_node.return_value = ssh_mock
+ mgmt = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]['mgmt-interface']
+ interfaces = \
+ self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]['vdu'][0]['external-interface']
+ resource_profile = \
+ ResourceProfile(mgmt, interfaces, [1, 2, 3])
+ resource_profile._provide_config_file = mock.Mock()
+ self.assertIsNone(
+ resource_profile._prepare_collectd_conf("/opt/nsb_bin"))
+
+ @mock.patch("yardstick.network_services.nfvi.resource.open")
+ @mock.patch("yardstick.network_services.nfvi.resource.tempfile")
+ @mock.patch("yardstick.network_services.nfvi.resource.os")
+ def test__provide_config_file(self, mock_open, mock_tempfile, mock_os):
+ with mock.patch("yardstick.ssh.SSH") as ssh:
+ ssh_mock = mock.Mock(autospec=ssh.SSH)
+ ssh_mock.execute = \
+ mock.Mock(return_value=(0, "", ""))
+ ssh.from_node.return_value = ssh_mock
+ mgmt = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]['mgmt-interface']
+ interfaces = \
+ self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]['vdu'][0]['external-interface']
+ resource_profile = \
+ ResourceProfile(mgmt, interfaces, [1, 2, 3])
+ resource_profile._prepare_collectd_conf = mock.Mock()
+ resource_profile.connection = ssh_mock
+ resource_profile.connection.put = \
+ mock.Mock(return_value=(0, "", ""))
+ mock_tempfile.mkstemp = mock.Mock(return_value=["test", ""])
+ self.assertIsNone(
+ resource_profile._provide_config_file("/opt/nsb_bin",
+ "collectd.cfg", {}))
+
+ @mock.patch("yardstick.network_services.nfvi.resource.open")
+ def test_initiate_systemagent(self, mock_open):
+ with mock.patch("yardstick.ssh.SSH") as ssh:
+ ssh_mock = mock.Mock(autospec=ssh.SSH)
+ ssh_mock.execute = \
+ mock.Mock(return_value=(0, "", ""))
+ ssh.from_node.return_value = ssh_mock
+ mgmt = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]['mgmt-interface']
+ interfaces = \
+ self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]['vdu'][0]['external-interface']
+ resource_profile = \
+ ResourceProfile(mgmt, interfaces, [1, 2, 3])
+ resource_profile._start_collectd = mock.Mock()
self.assertIsNone(
resource_profile.initiate_systemagent("/opt/nsb_bin"))
+ def test__parse_hugepages(self):
+ reskey = ["cpu", "cpuFreq"]
+ value = "timestamp:12345"
+ res = self.resource_profile.parse_hugepages(reskey, value)
+ self.assertEqual({'cpu/cpuFreq': '12345'}, res)
+
+ def test__parse_dpdkstat(self):
+ reskey = ["dpdk0", "0"]
+ value = "tx:12345"
+ res = self.resource_profile.parse_dpdkstat(reskey, value)
+ self.assertEqual({'dpdk0/0': '12345'}, res)
+
+ def test__parse_virt(self):
+ reskey = ["vm0", "cpu"]
+ value = "load:45"
+ res = self.resource_profile.parse_virt(reskey, value)
+ self.assertEqual({'vm0/cpu': '45'}, res)
+
+ def test__parse_ovs_stats(self):
+ reskey = ["ovs", "stats"]
+ value = "tx:45"
+ res = self.resource_profile.parse_ovs_stats(reskey, value)
+ self.assertEqual({'ovs/stats': '45'}, res)
+
def test_parse_collectd_result(self):
res = self.resource_profile.parse_collectd_result({}, [0, 1, 2])
- self.assertDictEqual(res, {'timestamp': '', 'cpu': {}, 'memory': {}})
+ expected_result = {'cpu': {}, 'dpdkstat': {}, 'hugepages': {},
+ 'memory': {}, 'ovs_stats': {}, 'timestamp': '',
+ 'virt': {}}
+ self.assertDictEqual(res, expected_result)
+
+ def test_parse_collectd_result_cpu(self):
+ metric = {"nsb_stats/cpu/0/ipc": "101"}
+ self.resource_profile.get_cpu_data = mock.Mock(return_value=[1,
+ "ipc",
+ "1234",
+ ""])
+ res = self.resource_profile.parse_collectd_result(metric, [0, 1, 2])
+ expected_result = {'cpu': {1: {'ipc': '1234'}}, 'dpdkstat': {}, 'hugepages': {},
+ 'memory': {}, 'ovs_stats': {}, 'timestamp': '',
+ 'virt': {}}
+ self.assertDictEqual(res, expected_result)
+
+ def test_parse_collectd_result_memory(self):
+ metric = {"nsb_stats/memory/bw": "101"}
+ res = self.resource_profile.parse_collectd_result(metric, [0, 1, 2])
+ expected_result = {'cpu': {}, 'dpdkstat': {}, 'hugepages': {},
+ 'memory': {'bw': '101'}, 'ovs_stats': {}, 'timestamp': '',
+ 'virt': {}}
+ self.assertDictEqual(res, expected_result)
+
+ def test_parse_collectd_result_hugepage(self):
+ metric = {"nsb_stats/hugepages/free": "101"}
+ self.resource_profile.parse_hugepages = \
+ mock.Mock(return_value={"free": "101"})
+ res = self.resource_profile.parse_collectd_result(metric, [0, 1, 2])
+ expected_result = {'cpu': {}, 'dpdkstat': {}, 'hugepages': {'free':
+ '101'},
+ 'memory': {}, 'ovs_stats': {}, 'timestamp': '',
+ 'virt': {}}
+ self.assertDictEqual(res, expected_result)
+
+ def test_parse_collectd_result_dpdk_virt_ovs(self):
+ metric = {"nsb_stats/dpdkstat/tx": "101",
+ "nsb_stats/ovs_stats/tx": "101",
+ "nsb_stats/virt/virt/memory": "101"}
+ self.resource_profile.parse_dpdkstat = \
+ mock.Mock(return_value={"tx": "101"})
+ self.resource_profile.parse_virt = \
+ mock.Mock(return_value={"memory": "101"})
+ self.resource_profile.parse_ovs_stats = \
+ mock.Mock(return_value={"tx": "101"})
+ res = self.resource_profile.parse_collectd_result(metric, [0, 1, 2])
+ expected_result = {'cpu': {}, 'dpdkstat': {'tx': '101'}, 'hugepages': {},
+ 'memory': {}, 'ovs_stats': {'tx': '101'}, 'timestamp': '',
+ 'virt': {'memory': '101'}}
+ self.assertDictEqual(res, expected_result)
+
+ def test_amqp_process_for_nfvi_kpi(self):
+ self.resource_profile.amqp_client = \
+ mock.MagicMock(side_effect=[None, mock.MagicMock()])
+ self.resource_profile.run_collectd_amqp = \
+ mock.Mock(return_value=0)
+ res = self.resource_profile.amqp_process_for_nfvi_kpi()
+ self.assertEqual(None, res)
+
+ def test_amqp_collect_nfvi_kpi(self):
+ self.resource_profile.amqp_client = \
+ mock.MagicMock(side_effect=[None, mock.MagicMock()])
+ self.resource_profile.run_collectd_amqp = \
+ mock.Mock(return_value=0)
+ self.resource_profile.parse_collectd_result = mock.Mock()
+ res = self.resource_profile.amqp_collect_nfvi_kpi()
+ self.assertIsNotNone(res)
def test_run_collectd_amqp(self):
_queue = multiprocessing.Queue()
resource.AmqpConsumer = mock.Mock(autospec=collectd)
- self.assertIsNone(self.resource_profile.run_collectd_amqp(_queue))
+ self.assertIsNone(self.resource_profile.run_collectd_amqp())
def test_start(self):
self.assertIsNone(self.resource_profile.start())
Hostname "nsb_stats"
FQDNLookup true
-Interval 5
+Interval {interval}
##############################################################################
# LoadPlugin section #
# Specify what features to activate. #
##############################################################################
-LoadPlugin amqp
-LoadPlugin cpu
-LoadPlugin intel_rdt
-LoadPlugin memory
+#LoadPlugin syslog
+{loadplugin}
##############################################################################
# Plugin configuration #
# ription of those options is available in the collectd.conf(5) manual page. #
##############################################################################
+#<Plugin syslog>
+# LogLevel debug
+#</Plugin>
+
<Plugin amqp>
<Publish "name">
Host "0.0.0.0"
<Plugin cpu>
ReportByCpu true
ReportByState true
- ValuesPercentage false
+ ValuesPercentage true
</Plugin>
<Plugin memory>
ValuesPercentage false
</Plugin>
-<LoadPlugin intel_rdt>
- Interval 5
-</LoadPlugin>
<Plugin "intel_rdt">
Cores ""
</Plugin>
-<Plugin memcached>
- <Instance "local">
- Host "127.0.0.1"
- Port "11211"
- </Instance>
+<Plugin hugepages>
+ ReportPerNodeHP true
+ ReportRootHP true
+ ValuesPages true
+ ValuesBytes false
+ ValuesPercentage false
+</Plugin>
+
+<Plugin hugepages>
+ ReportPerNodeHP true
+ ReportRootHP true
+ ValuesPages true
+ ValuesBytes false
+ ValuesPercentage false
+</Plugin>
+
+<Plugin dpdkstat>
+ <EAL>
+ Coremask "0x1"
+ MemoryChannels "4"
+ ProcessType "secondary"
+ FilePrefix "rte"
+ </EAL>
+ SharedMemObj "dpdk_collectd_stats_0"
+ EnabledPortMask 0xffff
+ {dpdk_interface}
+</Plugin>
+
+<Plugin virt>
+ Domain "samplevnf"
+</Plugin>
+
+<Plugin ovs_stats>
+ Port "6640"
+ Address "127.0.0.1"
+ Socket "/usr/local/var/run/openvswitch/db.sock"
+ Bridges "br0" "br_ext"
</Plugin>
<Include "/etc/collectd/collectd.conf.d">
exit 1;
fi
+echo "setup proxy..."
+http_proxy=$1
+https_proxy=$2
+if [[ "$http_proxy" != "" ]]; then
+ export http_proxy=$http_proxy
+ export https_proxy=$http_proxy
+fi
+
+if [[ "$https_proxy" != "" ]]; then
+ export https_proxy=$https_proxy
+fi
+
echo "Install required libraries to run collectd..."
-pkg=(git flex bison build-essential pkg-config automake autotools-dev libltdl-dev librabbitmq-dev rabbitmq-server)
+pkg=(git flex bison build-essential pkg-config automake autotools-dev libltdl-dev librabbitmq-dev rabbitmq-server cmake)
for i in "${pkg[@]}"; do
dpkg-query -W --showformat='${Status}\n' "${i}"|grep "install ok installed"
if [ "$?" -eq "1" ]; then
rm -rf intel-cmt-cat >/dev/null
git clone https://github.com/01org/intel-cmt-cat.git
pushd intel-cmt-cat
- git checkout tags/v1.5 -b v1.5
make install PREFIX=/usr
popd
echo "Done."
fi
-which /opt/nsb_bin/collectd/collectd >/dev/null
+ls /usr/lib/libdpdk.so >/dev/null
+if [ $? -eq 0 ]
+then
+ echo "DPDK already installed. Done"
+else
+ pushd .
+
+ echo "Get dpdk and install..."
+ mkdir -p $INSTALL_NSB_BIN
+ rm -rf "$INSTALL_NSB_BIN"/dpdk >/dev/null
+ git clone http://dpdk.org/git/dpdk
+ pushd dpdk
+ mkdir -p /mnt/huge
+ mount -t hugetlbfs nodev /mnt/huge
+ sed -i 's/CONFIG_RTE_BUILD_SHARED_LIB=n/CONFIG_RTE_BUILD_SHARED_LIB=y/g' config/common_base
+ sed -i 's/CONFIG_RTE_EAL_PMD_PATH=""/CONFIG_RTE_EAL_PMD_PATH="\/usr\/lib\/dpdk-pmd\/"/g' config/common_base
+
+ echo "Build dpdk v16.04"
+ make config T=x86_64-native-linuxapp-gcc
+ make
+ sudo make install prefix=/usr
+ mkdir -p /usr/lib/dpdk-pmd
+ find /usr/lib -type f -name 'librte_pmd*' | while read path ; do ln -s $path /usr/lib/dpdk-pmd/`echo $path | grep -o 'librte_.*so'` ; done
+
+ echo "Disable ASLR."
+ echo 0 > /proc/sys/kernel/randomize_va_space
+ make install PREFIX=/usr
+ popd
+
+ popd
+ echo "Done."
+fi
+
+which $INSTALL_NSB_BIN/yajl > /dev/null
+if [ -f "/usr/local/lib/libyajl.so.2.1.1" ]
+then
+ echo "ovs stats libs already installed."
+else
+ echo "installing ovs stats libraries"
+ pushd .
+
+ cd $INSTALL_NSB_BIN
+ git clone https://github.com/lloyd/yajl.git
+ pushd yajl
+ ./configure
+ make
+ make install
+ popd
+
+ popd
+fi
+
+which $INSTALL_NSB_BIN/collectd/collectd >/dev/null
if [ $? -eq 0 ]
then
echo "Collectd already installed. Done"
git clone https://github.com/collectd/collectd.git
pushd collectd
git stash
- git checkout -b collectd 43a4db3b3209f497a0ba408aebf8aee385c6262d
+ git checkout -n nfvi 47c86ace348a1d7a5352a83d10935209f89aa4f5
./build.sh
- ./configure --with-libpqos=/usr/
+ ./configure --with-libpqos=/usr/ --with-libdpdk=/usr --with-libyajl=/usr/local --enable-debug --enable-dpdkstat --enable-virt --enable-ovs_stats
make install > /dev/null
popd
echo "Done."
""" Resource collection definitions """
from __future__ import absolute_import
+from __future__ import print_function
+import tempfile
import logging
+import os
import os.path
import re
import multiprocessing
+from collections import Sequence
+
from oslo_config import cfg
from yardstick import ssh
from yardstick.network_services.nfvi.collectd import AmqpConsumer
from yardstick.network_services.utils import provision_tool
+LOG = logging.getLogger(__name__)
+
CONF = cfg.CONF
ZMQ_OVS_PORT = 5567
ZMQ_POLLING_TIME = 12000
+LIST_PLUGINS_ENABLED = ["amqp", "cpu", "cpufreq", "intel_rdt", "memory",
+ "hugepages", "dpdkstat", "virt", "ovs_stats"]
class ResourceProfile(object):
This profile adds a resource at the beginning of the test session
"""
- def __init__(self, vnfd, cores):
+ def __init__(self, mgmt, interfaces=None, cores=None):
self.enable = True
self.connection = None
- self.cores = cores
+ self.cores = cores if isinstance(cores, Sequence) else []
+ self._queue = multiprocessing.Queue()
+ self.amqp_client = None
+ self.interfaces = interfaces if isinstance(interfaces, Sequence) else []
- mgmt_interface = vnfd.get("mgmt-interface")
# why the host or ip?
- self.vnfip = mgmt_interface.get("host", mgmt_interface["ip"])
- self.connection = ssh.SSH.from_node(mgmt_interface,
- overrides={"ip": self.vnfip})
+ self.vnfip = mgmt.get("host", mgmt["ip"])
+ self.connection = ssh.SSH.from_node(mgmt, overrides={"ip": self.vnfip})
self.connection.wait()
err, pid, _ = self.connection.execute("pgrep -f %s" % process)
return [err == 0, pid]
- def run_collectd_amqp(self, queue):
+ def run_collectd_amqp(self):
""" run amqp consumer to collect the NFVi data """
- amqp = \
- AmqpConsumer('amqp://admin:admin@{}:5672/%2F'.format(self.vnfip),
- queue)
+ amqp_url = 'amqp://admin:admin@{}:5672/%2F'.format(self.vnfip)
+ amqp = AmqpConsumer(amqp_url, self._queue)
try:
amqp.run()
except (AttributeError, RuntimeError, KeyboardInterrupt):
amqp.stop()
@classmethod
- def get_cpu_data(cls, reskey, value):
+ def parse_simple_resource(cls, key, value):
+ return {'/'.join(key): value.split(":")[1]}
+
+ @classmethod
+ def get_cpu_data(cls, key_split, value):
""" Get cpu topology of the host """
pattern = r"-(\d+)"
- if "cpufreq" in reskey[1]:
- match = re.search(pattern, reskey[2], re.MULTILINE)
- metric = reskey[1]
+ if "cpufreq" in key_split[0]:
+ metric = key_split[0]
+ source = key_split[1]
else:
- match = re.search(pattern, reskey[1], re.MULTILINE)
- metric = reskey[2]
+ metric = key_split[1]
+ source = key_split[0]
+
+ match = re.search(pattern, source, re.MULTILINE)
+ if not match:
+ return "error", "Invalid", "", ""
+
+ time, value = value.split(":")
+ return str(match.group(1)), metric, value, time
+
+ @classmethod
+ def parse_hugepages(cls, key, value):
+ return cls.parse_simple_resource(key, value)
- time, val = re.split(":", value)
- if match:
- return [str(match.group(1)), metric, val, time]
+ @classmethod
+ def parse_dpdkstat(cls, key, value):
+ return cls.parse_simple_resource(key, value)
+
+ @classmethod
+ def parse_virt(cls, key, value):
+ return cls.parse_simple_resource(key, value)
- return ["error", "Invalid", ""]
+ @classmethod
+ def parse_ovs_stats(cls, key, value):
+ return cls.parse_simple_resource(key, value)
- def parse_collectd_result(self, metrics, listcores):
+ def parse_collectd_result(self, metrics, core_list):
""" convert collectd data into json"""
- res = {"cpu": {}, "memory": {}}
+ result = {
+ "cpu": {},
+ "memory": {},
+ "hugepages": {},
+ "dpdkstat": {},
+ "virt": {},
+ "ovs_stats": {},
+ }
testcase = ""
for key, value in metrics.items():
- reskey = key.rsplit("/")
- if "cpu" in reskey[1] or "intel_rdt" in reskey[1]:
- cpu_key, name, metric, testcase = \
- self.get_cpu_data(reskey, value)
- if cpu_key in listcores:
- res["cpu"].setdefault(cpu_key, {}).update({name: metric})
- elif "memory" in reskey[1]:
- val = re.split(":", value)[1]
- res["memory"].update({reskey[2]: val})
- res["timestamp"] = testcase
-
- return res
-
- def amqp_collect_nfvi_kpi(self, _queue=multiprocessing.Queue()):
+ key_split = key.split("/")
+ res_key_iter = (key for key in key_split if "nsb_stats" not in key)
+ res_key0 = next(res_key_iter)
+ res_key1 = next(res_key_iter)
+
+ if "cpu" in res_key0 or "intel_rdt" in res_key0:
+ cpu_key, name, metric, testcase = self.get_cpu_data(key_split, value)
+ if cpu_key in core_list:
+ result["cpu"].setdefault(cpu_key, {}).update({name: metric})
+
+ elif "memory" in res_key0:
+ result["memory"].update({res_key1: value.split(":")[0]})
+
+ elif "hugepages" in res_key0:
+ result["hugepages"].update(self.parse_hugepages(key, value))
+
+ elif "dpdkstat" in res_key0:
+ result["dpdkstat"].update(self.parse_dpdkstat(key, value))
+
+ elif "virt" in res_key1:
+ result["virt"].update(self.parse_virt(key, value))
+
+ elif "ovs_stats" in res_key0:
+ result["ovs_stats"].update(self.parse_ovs_stats(key, value))
+
+ result["timestamp"] = testcase
+
+ return result
+
+ def amqp_process_for_nfvi_kpi(self):
""" amqp collect and return nfvi kpis """
- try:
- metric = {}
- amqp_client = \
- multiprocessing.Process(target=self.run_collectd_amqp,
- args=(_queue,))
- amqp_client.start()
- amqp_client.join(7)
- amqp_client.terminate()
-
- while not _queue.empty():
- metric.update(_queue.get())
- except (AttributeError, RuntimeError, TypeError, ValueError):
- logging.debug("Failed to get NFVi stats...")
- msg = {}
- else:
- msg = self.parse_collectd_result(metric, self.cores)
+ if self.amqp_client is None:
+ self.amqp_client = \
+ multiprocessing.Process(target=self.run_collectd_amqp)
+ self.amqp_client.start()
+ def amqp_collect_nfvi_kpi(self):
+ """ amqp collect and return nfvi kpis """
+ metric = {}
+ while not self._queue.empty():
+ metric.update(self._queue.get())
+ msg = self.parse_collectd_result(metric, self.cores)
return msg
- @classmethod
- def _start_collectd(cls, connection, bin_path):
- connection.execute('pkill -9 collectd')
+ def _provide_config_file(self, bin_path, nfvi_cfg, kwargs):
+ with open(os.path.join(bin_path, nfvi_cfg), 'r') as cfg:
+ template = cfg.read()
+ cfg, cfg_content = tempfile.mkstemp()
+ with os.fdopen(cfg, "w+") as cfg:
+ cfg.write(template.format(**kwargs))
+ cfg_file = os.path.join(bin_path, nfvi_cfg)
+ self.connection.put(cfg_content, cfg_file)
+
+ def _prepare_collectd_conf(self, bin_path):
+ """ Prepare collectd conf """
+ loadplugin = "\n".join("LoadPlugin {0}".format(plugin)
+ for plugin in LIST_PLUGINS_ENABLED)
+
+ interfaces = "\n".join("PortName '{0[name]}'".format(interface)
+ for interface in self.interfaces)
+
+ kwargs = {
+ "interval": '25',
+ "loadplugin": loadplugin,
+ "dpdk_interface": interfaces,
+ }
+
+ self._provide_config_file(bin_path, 'collectd.conf', kwargs)
+
+ def _start_collectd(self, connection, bin_path):
+ LOG.debug("Starting collectd to collect NFVi stats")
+ # temp disable
+ return
+ connection.execute('sudo pkill -9 collectd')
collectd = os.path.join(bin_path, "collectd.sh")
provision_tool(connection, collectd)
- provision_tool(connection, os.path.join(bin_path, "collectd.conf"))
+ self._prepare_collectd_conf(bin_path)
# Reset amqp queue
+ LOG.debug("reset and setup amqp to collect data from collectd")
+ connection.execute("sudo rm -rf /var/lib/rabbitmq/mnesia/rabbit*")
connection.execute("sudo service rabbitmq-server start")
connection.execute("sudo rabbitmqctl stop_app")
connection.execute("sudo rabbitmqctl reset")
connection.execute("sudo service rabbitmq-server restart")
# Run collectd
- connection.execute(collectd)
- connection.execute(os.path.join(bin_path, "collectd", "collectd"))
+
+ http_proxy = os.environ.get('http_proxy', '')
+ https_proxy = os.environ.get('https_proxy', '')
+ connection.execute("sudo %s '%s' '%s'" %
+ (collectd, http_proxy, https_proxy))
+ LOG.debug("Start collectd service.....")
+ connection.execute(
+ "sudo %s" % os.path.join(bin_path, "collectd", "collectd"))
+ LOG.debug("Done")
def initiate_systemagent(self, bin_path):
""" Start system agent for NFVi collection on host """
def start(self):
""" start nfvi collection """
if self.enable:
- logging.debug("Start NVFi metric collection...")
+ LOG.debug("Start NVFi metric collection...")
def stop(self):
""" stop nfvi collection """
- if self.enable:
- agent = "collectd"
- logging.debug("Stop resource monitor...")
- status, pid = self.check_if_sa_running(agent)
- if status:
- self.connection.execute('kill -9 %s' % pid)
- self.connection.execute('pkill -9 %s' % agent)
- self.connection.execute('service rabbitmq-server stop')
- self.connection.execute("sudo rabbitmqctl stop_app")
+ if not self.enable:
+ return
+
+ agent = "collectd"
+ LOG.debug("Stop resource monitor...")
+
+ if self.amqp_client is not None:
+ self.amqp_client.terminate()
+
+ status, pid = self.check_if_sa_running(agent)
+ if status == 0:
+ return
+
+ self.connection.execute('sudo kill -9 %s' % pid)
+ self.connection.execute('sudo pkill -9 %s' % agent)
+ self.connection.execute('sudo service rabbitmq-server stop')
+ self.connection.execute("sudo rabbitmqctl stop_app")
self.socket = 1
cores = self._validate_cpu_cfg()
- return ResourceProfile(self.vnfd_helper, cores)
+ return ResourceProfile(self.vnfd_helper.mgmt_interface,
+ interfaces=self.vnfd_helper.interfaces, cores=cores)
def _detect_drivers(self):
interfaces = self.vnfd_helper.interfaces
Code Review / yardstick.git / commitdiff