Remote slave agent workload

Add storperf master object to manage stack
lifecycle.
Add configuration db.
Creation of CLI vs. main so that ReST API
and CLI API can be kept clear.
Fixed License in files.
Changed DB objects to be thread safe.
Added ssh server to container if desired
for CLI.

Change-Id: Idfe84bfb7920e6ce19d27462593c21ea86e7b406
JIRA: STORPERF-29
Signed-off-by: Mark Beierl <mark.beierl@emc.com>
(cherry picked from commit 488a47d945d3ef3dfa9ee37ca0aac3b480ffc800)

1 files changed, 342 insertions, 0 deletions

diff --git a/storperf/storperf_master.py b/storperf/storperf_master.py
new file mode 100644
index 0000000..9e88a3c
--- /dev/null
+++ b/storperf/storperf_master.py
@@ -0,0 +1,342 @@
+##############################################################################
+# Copyright (c) 2015 EMC and others.
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+##############################################################################
+
+from threading import Thread
+import logging
+import os
+import subprocess
+
+from db.configuration_db import ConfigurationDB
+from test_executor import TestExecutor
+import cinderclient.v2 as cinderclient
+import heatclient.client as heatclient
+import keystoneclient.v2_0 as ksclient
+
+
+class ParameterError(Exception):
+    """ """
+
+
+class StorPerfMaster(object):
+
+    def __init__(self):
+        self.logger = logging.getLogger(__name__)
+
+        self.configuration_db = ConfigurationDB()
+
+        template_file = open("storperf/resources/hot/agent-group.yaml")
+        self._agent_group_hot = template_file.read()
+        template_file = open("storperf/resources/hot/storperf-agent.yaml")
+        self._agent_resource_hot = template_file.read()
+        self._hot_files = {
+            'storperf-agent.yaml': self._agent_resource_hot
+        }
+        self.logger.debug(
+            "Loaded agent-group template as: " + self._agent_group_hot)
+        self.logger.debug(
+            "Loaded agent-resource template as: " + self._agent_resource_hot)
+
+        self._username = os.environ.get('OS_USERNAME')
+        self._password = os.environ.get('OS_PASSWORD')
+        self._tenant_name = os.environ.get('OS_TENANT_NAME')
+        self._project_name = os.environ.get('OS_PROJECT_NAME')
+        self._auth_url = os.environ.get('OS_AUTH_URL')
+
+        self._cinder_client = None
+        self._heat_client = None
+        self._test_executor = TestExecutor()
+
+    @property
+    def volume_size(self):
+        value = self.configuration_db.get_configuration_value(
+            'stack',
+            'volume_size')
+        if (value is None):
+            self.volume_size = 1
+            value = 1
+        return int(value)
+
+    @volume_size.setter
+    def volume_size(self, value):
+        if (self.stack_id is not None):
+            raise ParameterError(
+                "ERROR: Cannot change volume size after stack is created")
+
+        self.configuration_db.set_configuration_value(
+            'stack',
+            'volume_size',
+            value)
+
+    @property
+    def agent_count(self):
+        value = self.configuration_db.get_configuration_value(
+            'stack',
+            'agent_count')
+
+        if (value is None):
+            self.agent_count = 1
+            value = 1
+        return int(value)
+
+    @agent_count.setter
+    def agent_count(self, value):
+        if (self.stack_id is not None):
+            raise ParameterError(
+                "ERROR: Cannot change agent count after stack is created")
+
+        self.configuration_db.set_configuration_value(
+            'stack',
+            'agent_count',
+            value)
+
+    @property
+    def agent_network(self):
+        return self.configuration_db.get_configuration_value(
+            'stack',
+            'agent_network')
+
+    @agent_network.setter
+    def agent_network(self, value):
+        if (self.stack_id is not None):
+            raise ParameterError(
+                "ERROR: Cannot change agent network after stack is created")
+
+        self.configuration_db.set_configuration_value(
+            'stack',
+            'agent_network',
+            value)
+
+    @property
+    def stack_id(self):
+        return self.configuration_db.get_configuration_value(
+            'stack',
+            'stack_id')
+
+    @stack_id.setter
+    def stack_id(self, value):
+        self.configuration_db.set_configuration_value(
+            'stack',
+            'stack_id',
+            value)
+
+    @property
+    def volume_quota(self):
+        self._attach_to_openstack()
+        quotas = self._cinder_client.quotas.get(self._tenant_name)
+        return int(quotas.volumes)
+
+    @property
+    def filename(self):
+        return self._test_executor.filename
+
+    @filename.setter
+    def filename(self, value):
+        self._test_executor.filename = value
+
+    @property
+    def precondition(self):
+        return self._test_executor.precondition
+
+    @precondition.setter
+    def precondition(self, value):
+        self._test_executor.precondition = value
+
+    @property
+    def warm_up(self):
+        return self._test_executor.warm
+
+    @warm_up.setter
+    def warm_up(self, value):
+        self._test_executor.warm = value
+
+    @property
+    def is_stack_created(self):
+        if (self.stack_id is not None):
+            self._attach_to_openstack()
+
+            stack = self._heat_client.stacks.get(self.stack_id)
+            status = getattr(stack, 'stack_status')
+
+            self.logger.info("Status=" + status)
+            if (status == u'CREATE_COMPLETE'):
+                return True
+
+        return False
+
+    @property
+    def workloads(self):
+        return self.configuration_db.get_configuration_value(
+            'workload',
+            'workloads')
+
+    @workloads.setter
+    def workloads(self, value):
+        self._test_executor.register_workloads(value)
+
+        self.configuration_db.set_configuration_value(
+            'workload',
+            'workloads',
+            str(self._test_executor.workload_modules))
+
+    def create_stack(self):
+        if (self.stack_id is not None):
+            raise ParameterError("ERROR: Stack has already been created")
+
+        self._attach_to_openstack()
+        if (self.agent_count > self.volume_quota):
+            message = "ERROR: Volume quota too low: " + \
+                str(self.agent_count) + " > " + str(self.volume_quota)
+            raise ParameterError(message)
+
+        stack = self._heat_client.stacks.create(
+            stack_name="StorPerfAgentGroup",
+            template=self._agent_group_hot,
+            files=self._hot_files,
+            parameters=self._make_parameters())
+
+        self.stack_id = stack['stack']['id']
+        pass
+
+    def validate_stack(self):
+        self._attach_to_openstack()
+        if (self.agent_count > self.volume_quota):
+            message = "ERROR: Volume quota too low: " + \
+                str(self.agent_count) + " > " + str(self.volume_quota)
+            self.logger.error(message)
+            raise ParameterError(message)
+
+        self._heat_client.stacks.preview(
+            stack_name="StorPerfAgentGroup",
+            template=self._agent_group_hot,
+            files=self._hot_files,
+            parameters=self._make_parameters())
+        return True
+
+    def wait_for_stack_creation(self):
+
+        pass
+
+    def delete_stack(self):
+        if (self.stack_id is None):
+            raise ParameterError("ERROR: Stack does not exist")
+
+        self._attach_to_openstack()
+
+        self._heat_client.stacks.delete(stack_id=self.stack_id)
+        self.stack_id = None
+
+        pass
+
+    def execute_workloads(self):
+
+        if (self.stack_id is None):
+            raise ParameterError("ERROR: Stack does not exist")
+
+        self._attach_to_openstack()
+
+        stack = self._heat_client.stacks.get(self.stack_id)
+        outputs = getattr(stack, 'outputs')
+        slaves = outputs[0]['output_value']
+
+        setup_threads = []
+
+        for slave in slaves:
+            t = Thread(target=self._setup_slave, args=(slave,))
+            setup_threads.append(t)
+            t.start()
+
+        for thread in setup_threads:
+            thread.join()
+
+        self._test_executor.slaves = slaves
+        return self._test_executor.execute()
+
+    def _setup_slave(self, slave):
+        logger = logging.getLogger(__name__ + ":" + slave)
+
+        logger.info("Initializing slave at " + slave)
+
+        args = ['scp', '-o', 'StrictHostKeyChecking=no',
+                '-i', 'storperf/resources/ssh/storperf_rsa',
+                '/lib/x86_64-linux-gnu/libaio.so.1',
+                'ubuntu@' + slave + ":"]
+
+        logger.debug(args)
+        proc = subprocess.Popen(args,
+                                universal_newlines=True,
+                                stdout=subprocess.PIPE,
+                                stderr=subprocess.PIPE)
+
+        (stdout, stderr) = proc.communicate()
+        if (len(stdout) > 0):
+            logger.debug(stdout.decode('utf-8').strip())
+        if (len(stderr) > 0):
+            logger.error(stderr.decode('utf-8').strip())
+
+        args = ['scp', '-o', 'StrictHostKeyChecking=no',
+                '-i', 'storperf/resources/ssh/storperf_rsa',
+                '/usr/local/bin/fio',
+                'ubuntu@' + slave + ":"]
+
+        logger.debug(args)
+        proc = subprocess.Popen(args,
+                                universal_newlines=True,
+                                stdout=subprocess.PIPE,
+                                stderr=subprocess.PIPE)
+
+        (stdout, stderr) = proc.communicate()
+        if (len(stdout) > 0):
+            logger.debug(stdout.decode('utf-8').strip())
+        if (len(stderr) > 0):
+            logger.error(stderr.decode('utf-8').strip())
+
+        args = ['ssh', '-o', 'StrictHostKeyChecking=no',
+                '-i', 'storperf/resources/ssh/storperf_rsa',
+                'ubuntu@' + slave,
+                'sudo cp -v libaio.so.1 /lib/x86_64-linux-gnu/libaio.so.1'
+                ]
+
+        logger.debug(args)
+        proc = subprocess.Popen(args,
+                                universal_newlines=True,
+                                stdout=subprocess.PIPE,
+                                stderr=subprocess.PIPE)
+
+        (stdout, stderr) = proc.communicate()
+        if (len(stdout) > 0):
+            logger.debug(stdout.decode('utf-8').strip())
+        if (len(stderr) > 0):
+            logger.error(stderr.decode('utf-8').strip())
+
+    def _make_parameters(self):
+        heat_parameters = {}
+        heat_parameters['agent_network'] = self.agent_network
+        heat_parameters['agent_count'] = self.agent_count
+        heat_parameters['volume_size'] = self.volume_size
+        return heat_parameters
+
+    def _attach_to_openstack(self):
+
+        if (self._cinder_client is None):
+            self._cinder_client = cinderclient.Client(
+                self._username, self._password, self._project_name,
+                self._auth_url, service_type='volumev2')
+            self._cinder_client.authenticate()
+
+        if (self._heat_client is None):
+            self._keystone_client = ksclient.Client(
+                auth_url=self._auth_url,
+                username=self._username,
+                password=self._password,
+                tenant_name=self._tenant_name)
+            heat_endpoint = self._keystone_client.service_catalog.url_for(
+                service_type='orchestration')
+            self._heat_client = heatclient.Client(
+                '1', endpoint=heat_endpoint,
+                token=self._keystone_client.auth_token)