MQ startup process refactor

JIRA: YARDSTICK-1112

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

4 files changed, 66 insertions, 33 deletions

diff --git a/tests/unit/network_services/nfvi/test_resource.py b/tests/unit/network_services/nfvi/test_resource.py
index f5f7f0fe7..741f9a6cc 100644
--- a/tests/unit/network_services/nfvi/test_resource.py
+++ b/tests/unit/network_services/nfvi/test_resource.py
@@ -19,6 +19,8 @@ import unittest
 
 from yardstick.network_services.nfvi.resource import ResourceProfile
 from yardstick.network_services.nfvi import resource, collectd
+from yardstick.common.exceptions import ResourceCommandError
+from yardstick.common.exceptions import SSHError
 
 
 class TestResourceProfile(unittest.TestCase):
@@ -128,8 +130,31 @@ class TestResourceProfile(unittest.TestCase):
         self.assertEqual(val, ('error', 'Invalid', '', ''))
 
     def test__start_collectd(self):
-            self.assertIsNone(
-                self.resource_profile._start_collectd(self.ssh_mock, "/opt/nsb_bin"))
+        ssh_mock = mock.Mock()
+        ssh_mock.execute = mock.Mock(return_value=(0, "", ""))
+        self.assertIsNone(self.resource_profile._start_collectd(ssh_mock,
+                                                                "/opt/nsb_bin"))
+
+        ssh_mock.execute = mock.Mock(side_effect=SSHError)
+        with self.assertRaises(SSHError):
+            self.resource_profile._start_collectd(ssh_mock, "/opt/nsb_bin")
+
+        ssh_mock.execute = mock.Mock(return_value=(1, "", ""))
+        self.assertIsNone(self.resource_profile._start_collectd(ssh_mock,
+                                                                "/opt/nsb_bin"))
+
+    def test__start_rabbitmq(self):
+        ssh_mock = mock.Mock()
+        ssh_mock.execute = mock.Mock(return_value=(0, "RabbitMQ", ""))
+        self.assertIsNone(self.resource_profile._start_rabbitmq(ssh_mock))
+
+        ssh_mock.execute = mock.Mock(return_value=(0, "", ""))
+        with self.assertRaises(ResourceCommandError):
+            self.resource_profile._start_rabbitmq(ssh_mock)
+
+        ssh_mock.execute = mock.Mock(return_value=(1, "", ""))
+        with self.assertRaises(ResourceCommandError):
+            self.resource_profile._start_rabbitmq(ssh_mock)
 
     def test__prepare_collectd_conf(self):
             self.assertIsNone(
@@ -154,11 +179,12 @@ class TestResourceProfile(unittest.TestCase):
 
     def test_initiate_systemagent(self):
         self.resource_profile._start_collectd = mock.Mock()
+        self.resource_profile._start_rabbitmq = mock.Mock()
         self.assertIsNone(
             self.resource_profile.initiate_systemagent("/opt/nsb_bin"))
 
     def test_initiate_systemagent_raise(self):
-        self.resource_profile._start_collectd = mock.Mock(side_effect=RuntimeError)
+        self.resource_profile._start_rabbitmq = mock.Mock(side_effect=RuntimeError)
         with self.assertRaises(RuntimeError):
             self.resource_profile.initiate_systemagent("/opt/nsb_bin")
 
diff --git a/yardstick/common/exceptions.py b/yardstick/common/exceptions.py
index 0c6272c14..3def05e9d 100644
--- a/yardstick/common/exceptions.py
+++ b/yardstick/common/exceptions.py
@@ -64,6 +64,10 @@ class YardstickException(Exception):
         return False
 
 
+class ResourceCommandError(YardstickException):
+    message = 'Command: "%(command)s" Failed, stderr: "%(stderr)s"'
+
+
 class FunctionNotImplemented(YardstickException):
     message = ('The function "%(function_name)s" is not implemented in '
                '"%(class_name)" class.')
diff --git a/yardstick/network_services/nfvi/resource.py b/yardstick/network_services/nfvi/resource.py
index dc5c46a86..0c0bf223a 100644
--- a/yardstick/network_services/nfvi/resource.py
+++ b/yardstick/network_services/nfvi/resource.py
@@ -27,6 +27,7 @@ from oslo_config import cfg
 from oslo_utils.encodeutils import safe_decode
 
 from yardstick import ssh
+from yardstick.common.exceptions import ResourceCommandError
 from yardstick.common.task_template import finalize_for_yaml
 from yardstick.common.utils import validate_non_string_sequence
 from yardstick.network_services.nfvi.collectd import AmqpConsumer
@@ -249,45 +250,46 @@ class ResourceProfile(object):
         if status != 0:
             LOG.error("cannot find OVS socket %s", socket_path)
 
+    def _start_rabbitmq(self, connection):
+        # Reset amqp queue
+        LOG.debug("reset and setup amqp to collect data from collectd")
+        # ensure collectd.conf.d exists to avoid error/warning
+        cmd_list = ["sudo mkdir -p /etc/collectd/collectd.conf.d",
+                    "sudo service rabbitmq-server restart",
+                    "sudo rabbitmqctl stop_app",
+                    "sudo rabbitmqctl reset",
+                    "sudo rabbitmqctl start_app",
+                    "sudo rabbitmqctl add_user admin admin",
+                    "sudo rabbitmqctl authenticate_user admin admin",
+                    "sudo rabbitmqctl set_permissions -p / admin '.*' '.*' '.*'"
+                    ]
+        for cmd in cmd_list:
+                exit_status, stdout, stderr = connection.execute(cmd)
+                if exit_status != 0:
+                    raise ResourceCommandError(command=cmd, stderr=stderr)
+
+        # check stdout for "sudo rabbitmqctl status" command
+        cmd = "sudo rabbitmqctl status"
+        _, stdout, stderr = connection.execute(cmd)
+        if not re.search("RabbitMQ", stdout):
+            LOG.error("rabbitmqctl status don't have RabbitMQ in running apps")
+            raise ResourceCommandError(command=cmd, stderr=stderr)
+
     def _start_collectd(self, connection, bin_path):
         LOG.debug("Starting collectd to collect NFVi stats")
-        connection.execute('sudo pkill -x -9 collectd')
         collectd_path = os.path.join(bin_path, "collectd", "sbin", "collectd")
         config_file_path = os.path.join(bin_path, "collectd", "etc")
+        self._prepare_collectd_conf(config_file_path)
+
+        connection.execute('sudo pkill -x -9 collectd')
         exit_status = connection.execute("which %s > /dev/null 2>&1" % collectd_path)[0]
         if exit_status != 0:
             LOG.warning("%s is not present disabling", collectd_path)
-            # disable auto-provisioning because it requires Internet access
-            # collectd_installer = os.path.join(bin_path, "collectd.sh")
-            # provision_tool(connection, collectd)
-            # http_proxy = os.environ.get('http_proxy', '')
-            # https_proxy = os.environ.get('https_proxy', '')
-            # connection.execute("sudo %s '%s' '%s'" % (
-            #     collectd_installer, http_proxy, https_proxy))
             return
         if "ovs_stats" in self.plugins:
             self._setup_ovs_stats(connection)
 
         LOG.debug("Starting collectd to collect NFVi stats")
-        # ensure collectd.conf.d exists to avoid error/warning
-        connection.execute("sudo mkdir -p /etc/collectd/collectd.conf.d")
-        self._prepare_collectd_conf(config_file_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 rabbitmqctl start_app")
-        connection.execute("sudo service rabbitmq-server restart")
-
-        LOG.debug("Creating admin user for rabbitmq in order to collect data from collectd")
-        connection.execute("sudo rabbitmqctl delete_user guest")
-        connection.execute("sudo rabbitmqctl add_user admin admin")
-        connection.execute("sudo rabbitmqctl authenticate_user admin admin")
-        connection.execute("sudo rabbitmqctl set_permissions -p / admin '.*' '.*' '.*'")
-
         LOG.debug("Start collectd service..... %s second timeout", self.timeout)
         # intel_pmu plug requires large numbers of files open, so try to set
         # ulimit -n to a large value
@@ -299,9 +301,10 @@ class ResourceProfile(object):
         """ Start system agent for NFVi collection on host """
         if self.enable:
             try:
+                self._start_rabbitmq(self.connection)
                 self._start_collectd(self.connection, bin_path)
-            except Exception:
-                LOG.exception("Exception during collectd start")
+            except ResourceCommandError as e:
+                LOG.exception("Exception during collectd and rabbitmq start: %s", str(e))
                 raise
 
     def start(self):
diff --git a/yardstick/tests/unit/network_services/collector/test_subscriber.py b/yardstick/tests/unit/network_services/collector/test_subscriber.py
index 14e26f7fe..4ec2ea7a2 100644
--- a/yardstick/tests/unit/network_services/collector/test_subscriber.py
+++ b/yardstick/tests/unit/network_services/collector/test_subscriber.py
@@ -58,7 +58,7 @@ class CollectorTestCase(unittest.TestCase):
         self.assertEqual(len(collector.vnfs), 1)
 
     def test_start(self, *_):
-        self.assertIsNone(self.collector.start())
+        self.collector.start()
         for vnf in self.collector.vnfs:
             vnf.start_collect.assert_called_once()