Do not start collectd twice when SampleVNF is running on Baremetal

JIRA: YARDSTICK-1167

Change-Id: I7591bbb2a84ee4039a20c5da2914f1e374299015
Signed-off-by: Chornyi, TarasX <tarasx.chornyi@intel.com>
(cherry picked from commit ba2ca2819a8b0312627742d126030c3dac15052f)

7 files changed, 175 insertions, 75 deletions

diff --git a/tests/unit/network_services/vnf_generic/vnf/test_base.py b/tests/unit/network_services/vnf_generic/vnf/test_base.py
index 664373f8f..9ef6473f0 100644
--- a/tests/unit/network_services/vnf_generic/vnf/test_base.py
+++ b/tests/unit/network_services/vnf_generic/vnf/test_base.py
@@ -215,9 +215,11 @@ class TestGenericVNF(unittest.TestCase):
         with self.assertRaises(TypeError) as exc:
             # pylint: disable=abstract-class-instantiated
             base.GenericVNF('vnf1', VNFD['vnfd:vnfd-catalog']['vnfd'][0])
-        msg = ("Can't instantiate abstract class GenericVNF with abstract "
-               "methods collect_kpi, instantiate, scale, terminate, "
-               "wait_for_instantiate")
+
+        msg = ("Can't instantiate abstract class GenericVNF with abstract methods "
+               "collect_kpi, instantiate, scale, start_collect, "
+               "stop_collect, terminate, wait_for_instantiate")
+
         self.assertEqual(msg, str(exc.exception))
 
 
diff --git a/tests/unit/network_services/vnf_generic/vnf/test_sample_vnf.py b/tests/unit/network_services/vnf_generic/vnf/test_sample_vnf.py
index eb59c2837..38a043d00 100644
--- a/tests/unit/network_services/vnf_generic/vnf/test_sample_vnf.py
+++ b/tests/unit/network_services/vnf_generic/vnf/test_sample_vnf.py
@@ -1714,6 +1714,64 @@ class TestSampleVnf(unittest.TestCase):
         self.assertIsNone(sample_vnf.instantiate(scenario_cfg, {}))
         self.assertEqual(sample_vnf.nfvi_context, context2)
 
+    def test__update_collectd_options(self):
+        scenario_cfg = {'options':
+                            {'collectd':
+                                 {'interval': 3,
+                                  'plugins':
+                                      {'plugin3': {'param': 3}}},
+                             'vnf__0':
+                                 {'collectd':
+                                      {'interval': 2,
+                                       'plugins':
+                                           {'plugin3': {'param': 2},
+                                            'plugin2': {'param': 2}}}}}}
+        context_cfg = {'nodes':
+                           {'vnf__0':
+                                {'collectd':
+                                     {'interval': 1,
+                                      'plugins':
+                                          {'plugin3': {'param': 1},
+                                           'plugin2': {'param': 1},
+                                           'plugin1': {'param': 1}}}}}}
+        expected = {'interval': 1,
+                    'plugins':
+                        {'plugin3': {'param': 1},
+                         'plugin2': {'param': 1},
+                         'plugin1': {'param': 1}}}
+
+        vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
+        sample_vnf = SampleVNF('vnf__0', vnfd)
+        sample_vnf._update_collectd_options(scenario_cfg, context_cfg)
+        self.assertEqual(sample_vnf.setup_helper.collectd_options, expected)
+
+    def test__update_options(self):
+        options1 = {'interval': 1,
+                    'param1': 'value1',
+                    'plugins':
+                        {'plugin3': {'param': 3},
+                         'plugin2': {'param': 1},
+                         'plugin1': {'param': 1}}}
+        options2 = {'interval': 2,
+                    'param2': 'value2',
+                    'plugins':
+                        {'plugin4': {'param': 4},
+                         'plugin2': {'param': 2},
+                         'plugin1': {'param': 2}}}
+        expected = {'interval': 1,
+                    'param1': 'value1',
+                    'param2': 'value2',
+                    'plugins':
+                        {'plugin4': {'param': 4},
+                         'plugin3': {'param': 3},
+                         'plugin2': {'param': 1},
+                         'plugin1': {'param': 1}}}
+
+        vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
+        sample_vnf = SampleVNF('vnf1', vnfd)
+        sample_vnf._update_options(options2, options1)
+        self.assertEqual(options2, expected)
+
     @mock.patch("yardstick.network_services.vnf_generic.vnf.sample_vnf.time")
     @mock.patch("yardstick.ssh.SSH")
     def test_wait_for_instantiate_empty_queue(self, ssh, *args):
diff --git a/yardstick/benchmark/scenarios/networking/vnf_generic.py b/yardstick/benchmark/scenarios/networking/vnf_generic.py
index be2fa3f3b..78f866e25 100644
--- a/yardstick/benchmark/scenarios/networking/vnf_generic.py
+++ b/yardstick/benchmark/scenarios/networking/vnf_generic.py
@@ -441,7 +441,7 @@ class NetworkServiceTestCase(scenario_base.Scenario):
             traffic_gen.listen_traffic(self.traffic_profile)
 
         # register collector with yardstick for KPI collection.
-        self.collector = Collector(self.vnfs, self.context_cfg["nodes"], self.traffic_profile)
+        self.collector = Collector(self.vnfs)
         self.collector.start()
 
         # Start the actual traffic
diff --git a/yardstick/network_services/collector/subscriber.py b/yardstick/network_services/collector/subscriber.py
index 7e18302eb..322b3f5a2 100644
--- a/yardstick/network_services/collector/subscriber.py
+++ b/yardstick/network_services/collector/subscriber.py
@@ -14,42 +14,29 @@
 """This module implements stub for publishing results in yardstick format."""
 import logging
 
-from yardstick.network_services.nfvi.resource import ResourceProfile
-from yardstick.network_services.utils import get_nsb_option
-
 LOG = logging.getLogger(__name__)
 
 
 class Collector(object):
     """Class that handles dictionary of results in yardstick-plot format."""
 
-    def __init__(self, vnfs, nodes, traffic_profile, timeout=3600):
+    def __init__(self, vnfs):
         super(Collector, self).__init__()
-        self.traffic_profile = traffic_profile
         self.vnfs = vnfs
-        self.nodes = nodes
-        self.timeout = timeout
-        self.bin_path = get_nsb_option('bin_path', '')
-        self.resource_profiles = {node_name: ResourceProfile.make_from_node(node, self.timeout)
-                                  for node_name, node in self.nodes.items()
-                                  if node.get("collectd")}
 
     def start(self):
-        """Nothing to do, yet"""
-        for resource in self.resource_profiles.values():
-            resource.initiate_systemagent(self.bin_path)
-            resource.start()
-            resource.amqp_process_for_nfvi_kpi()
+        for vnf in self.vnfs:
+            vnf.start_collect()
 
     def stop(self):
-        """Nothing to do, yet"""
-        for resource in self.resource_profiles.values():
-            resource.stop()
+        for vnf in self.vnfs:
+            vnf.stop_collect()
 
     def get_kpi(self):
         """Returns dictionary of results in yardstick-plot format
 
-        :return:
+        :return: (dict) dictionary of kpis collected from the VNFs;
+                 the keys are the names of the VNFs.
         """
         results = {}
         for vnf in self.vnfs:
@@ -58,17 +45,4 @@ class Collector(object):
             LOG.debug("collect KPI for %s", vnf.name)
             results[vnf.name] = vnf.collect_kpi()
 
-        for node_name, resource in self.resource_profiles.items():
-            # Result example:
-            # {"VNF1: { "tput" : [1000, 999] }, "VNF2": { "latency": 100 }}
-            LOG.debug("collect KPI for %s", node_name)
-            if resource.check_if_system_agent_running("collectd")[0] != 0:
-                continue
-
-            try:
-                results[node_name] = {"core": resource.amqp_collect_nfvi_kpi()}
-                LOG.debug("%s collect KPIs %s", node_name, results[node_name]['core'])
-            # NOTE(elfoley): catch a more specific error
-            except Exception as exc:  # pylint: disable=broad-except
-                LOG.exception(exc)
         return results
diff --git a/yardstick/network_services/vnf_generic/vnf/base.py b/yardstick/network_services/vnf_generic/vnf/base.py
index a776b0989..9ceac3167 100644
--- a/yardstick/network_services/vnf_generic/vnf/base.py
+++ b/yardstick/network_services/vnf_generic/vnf/base.py
@@ -195,6 +195,18 @@ class GenericVNF(object):
         :return: {"kpi": value, "kpi2": value}
         """
 
+    @abc.abstractmethod
+    def start_collect(self):
+        """Start KPI collection
+        :return: None
+        """
+
+    @abc.abstractmethod
+    def stop_collect(self):
+        """Stop KPI collection
+        :return: None
+        """
+
 
 @six.add_metaclass(abc.ABCMeta)
 class GenericTrafficGen(GenericVNF):
@@ -254,3 +266,23 @@ class GenericTrafficGen(GenericVNF):
         :return: True/False
         """
         pass
+
+    def start_collect(self):
+        """Start KPI collection.
+
+        Traffic measurements are always collected during injection.
+
+        Optional.
+
+        :return: True/False
+        """
+        pass
+
+    def stop_collect(self):
+        """Stop KPI collection.
+
+        Optional.
+
+        :return: True/False
+        """
+        pass
diff --git a/yardstick/network_services/vnf_generic/vnf/sample_vnf.py b/yardstick/network_services/vnf_generic/vnf/sample_vnf.py
index d8b9625fb..34b0260b4 100644
--- a/yardstick/network_services/vnf_generic/vnf/sample_vnf.py
+++ b/yardstick/network_services/vnf_generic/vnf/sample_vnf.py
@@ -59,6 +59,7 @@ class SetupEnvHelper(object):
         self.vnfd_helper = vnfd_helper
         self.ssh_helper = ssh_helper
         self.scenario_helper = scenario_helper
+        self.collectd_options = {}
 
     def build_config(self):
         raise NotImplementedError
@@ -224,12 +225,6 @@ class DpdkVnfSetupEnvHelper(SetupEnvHelper):
         if exit_status == 0:
             return
 
-    def get_collectd_options(self):
-        options = self.scenario_helper.all_options.get("collectd", {})
-        # override with specific node settings
-        options.update(self.scenario_helper.options.get("collectd", {}))
-        return options
-
     def _setup_resources(self):
         # what is this magic?  how do we know which socket is for which port?
         # what about quad-socket?
@@ -242,11 +237,11 @@ class DpdkVnfSetupEnvHelper(SetupEnvHelper):
         # this won't work because we don't have DPDK port numbers yet
         ports = sorted(self.vnfd_helper.interfaces, key=self.vnfd_helper.port_num)
         port_names = (intf["name"] for intf in ports)
-        collectd_options = self.get_collectd_options()
-        plugins = collectd_options.get("plugins", {})
+        plugins = self.collectd_options.get("plugins", {})
+        interval = self.collectd_options.get("interval")
         # we must set timeout to be the same as the VNF otherwise KPIs will die before VNF
         return ResourceProfile(self.vnfd_helper.mgmt_interface, port_names=port_names,
-                               plugins=plugins, interval=collectd_options.get("interval"),
+                               plugins=plugins, interval=interval,
                                timeout=self.scenario_helper.timeout)
 
     def _check_interface_fields(self):
@@ -666,6 +661,7 @@ class SampleVNF(GenericVNF):
         pass
 
     def instantiate(self, scenario_cfg, context_cfg):
+        self._update_collectd_options(scenario_cfg, context_cfg)
         self.scenario_helper.scenario_cfg = scenario_cfg
         self.context_cfg = context_cfg
         self.nfvi_context = Context.get_context_from_server(self.scenario_helper.nodes[self.name])
@@ -677,6 +673,54 @@ class SampleVNF(GenericVNF):
         self.resource_helper.setup()
         self._start_vnf()
 
+    def _update_collectd_options(self, scenario_cfg, context_cfg):
+        """Update collectd configuration options
+        This function retrieves all collectd options contained in the test case
+
+        definition builds a single dictionary combining them. The following fragment
+        represents a test case with the collectd options and priorities (1 highest, 3 lowest):
+        ---
+        schema: yardstick:task:0.1
+        scenarios:
+        - type: NSPerf
+          nodes:
+            tg__0: trafficgen_1.yardstick
+            vnf__0: vnf.yardstick
+          options:
+            collectd:
+              <options>  # COLLECTD priority 3
+            vnf__0:
+              collectd:
+                plugins:
+                    load
+                <options> # COLLECTD priority 2
+        context:
+          type: Node
+          name: yardstick
+          nfvi_type: baremetal
+          file: /etc/yardstick/nodes/pod_ixia.yaml  # COLLECTD priority 1
+        """
+        scenario_options = scenario_cfg.get('options', {})
+        generic_options = scenario_options.get('collectd', {})
+        scenario_node_options = scenario_options.get(self.name, {})\
+            .get('collectd', {})
+        context_node_options = context_cfg.get('nodes', {})\
+            .get(self.name, {}).get('collectd', {})
+
+        options = generic_options
+        self._update_options(options, scenario_node_options)
+        self._update_options(options, context_node_options)
+
+        self.setup_helper.collectd_options = options
+
+    def _update_options(self, options, additional_options):
+        """Update collectd options and plugins dictionary"""
+        for k, v in additional_options.items():
+            if isinstance(v, dict) and k in options:
+                options[k].update(v)
+            else:
+                options[k] = v
+
     def wait_for_instantiate(self):
         buf = []
         time.sleep(self.WAIT_TIME)  # Give some time for config to load
@@ -692,7 +736,6 @@ class SampleVNF(GenericVNF):
                     LOG.info("%s VNF is up and running.", self.APP_NAME)
                     self._vnf_up_post()
                     self.queue_wrapper.clear()
-                    self.resource_helper.start_collect()
                     return self._vnf_process.exitcode
 
                 if "PANIC" in message:
@@ -705,6 +748,12 @@ class SampleVNF(GenericVNF):
             # by other VNF output
             self.q_in.put('\r\n')
 
+    def start_collect(self):
+        self.resource_helper.start_collect()
+
+    def stop_collect(self):
+        self.resource_helper.stop_collect()
+
     def _build_run_kwargs(self):
         self.run_kwargs = {
             'stdin': self.queue_wrapper,
diff --git a/yardstick/tests/unit/network_services/collector/test_subscriber.py b/yardstick/tests/unit/network_services/collector/test_subscriber.py
index 9b9649979..14e26f7fe 100644
--- a/yardstick/tests/unit/network_services/collector/test_subscriber.py
+++ b/yardstick/tests/unit/network_services/collector/test_subscriber.py
@@ -17,6 +17,7 @@ import unittest
 import mock
 
 from yardstick.network_services.collector import subscriber
+from yardstick import ssh
 
 
 class MockVnfAprrox(object):
@@ -37,57 +38,41 @@ class MockVnfAprrox(object):
 
 class CollectorTestCase(unittest.TestCase):
 
-    NODES = {
-        'node1': {},
-        'node2': {
-            'ip': '1.2.3.4',
-            'collectd': {
-                'plugins': {'abc': 12, 'def': 34},
-                'interval': 987,
-            },
-        },
-    }
-    TRAFFIC_PROFILE = {
-        'key1': 'value1',
-    }
-
     def setUp(self):
         vnf = MockVnfAprrox()
-        self.ssh_patch = mock.patch('yardstick.network_services.nfvi.resource.ssh', autospec=True)
+        vnf.start_collect = mock.Mock()
+        vnf.stop_collect = mock.Mock()
+        self.ssh_patch = mock.patch.object(ssh, 'AutoConnectSSH')
         mock_ssh = self.ssh_patch.start()
         mock_instance = mock.Mock()
         mock_instance.execute.return_value = 0, '', ''
-        mock_ssh.AutoConnectSSH.from_node.return_value = mock_instance
-        self.collector = subscriber.Collector([vnf], self.NODES, self.TRAFFIC_PROFILE, 1800)
+        mock_ssh.from_node.return_value = mock_instance
+        self.collector = subscriber.Collector([vnf])
 
     def tearDown(self):
         self.ssh_patch.stop()
 
     def test___init__(self, *_):
         vnf = MockVnfAprrox()
-        collector = subscriber.Collector([vnf], {}, {})
+        collector = subscriber.Collector([vnf])
         self.assertEqual(len(collector.vnfs), 1)
-        self.assertEqual(collector.traffic_profile, {})
-
-    def test___init___with_data(self, *_):
-        self.assertEqual(len(self.collector.vnfs), 1)
-        self.assertDictEqual(self.collector.traffic_profile, self.TRAFFIC_PROFILE)
-        self.assertEqual(len(self.collector.resource_profiles), 1)
-
-    def test___init___negative(self, *_):
-        pass
 
     def test_start(self, *_):
         self.assertIsNone(self.collector.start())
+        for vnf in self.collector.vnfs:
+            vnf.start_collect.assert_called_once()
 
     def test_stop(self, *_):
         self.assertIsNone(self.collector.stop())
+        for vnf in self.collector.vnfs:
+            vnf.stop_collect.assert_called_once()
 
     def test_get_kpi(self, *_):
         result = self.collector.get_kpi()
 
+        self.assertEqual(1, len(result))
+        self.assertEqual(4, len(result["vnf__1"]))
         self.assertEqual(result["vnf__1"]["pkt_in_up_stream"], 100)
         self.assertEqual(result["vnf__1"]["pkt_drop_up_stream"], 5)
         self.assertEqual(result["vnf__1"]["pkt_in_down_stream"], 50)
         self.assertEqual(result["vnf__1"]["pkt_drop_down_stream"], 40)
-        self.assertIn('node2', result)