return workload_executions
- def record_workload_params(self, job_id, params):
+ def record_workload_params(self, params):
"""
"""
+ if (self.job_id is None):
+ self.create_job_id()
+
with db_mutex:
db = sqlite3.connect(JobDB.db_name)
param,
value)
values (?, ?, ?)""",
- (job_id,
+ (self.job_id,
param,
value,))
db.commit()
--- /dev/null
+##############################################################################
+# Copyright (c) 2016 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
+##############################################################################
+
+import json
+import requests
+import os
+
+
+def get_installer_type(logger=None):
+ """
+ Get installer type (fuel, apex, joid, compass)
+ """
+ try:
+ installer = os.environ['INSTALLER_TYPE']
+ except KeyError:
+ if logger:
+ logger.error("Impossible to retrieve the installer type")
+ installer = "Unknown_installer"
+
+ return installer
+
+
+def push_results_to_db(db_url, project, case_name, logger, pod_name,
+ version, scenario, criteria, build_tag, payload):
+ """
+ POST results to the Result target DB
+ """
+ url = db_url + "/results"
+ installer = get_installer_type(logger)
+ params = {"project_name": project, "case_name": case_name,
+ "pod_name": pod_name, "installer": installer,
+ "version": version, "scenario": scenario, "criteria": criteria,
+ "build_tag": build_tag, "details": payload}
+
+ headers = {'Content-Type': 'application/json'}
+ try:
+ if logger:
+ logger.debug("Pushing results to %s" % (url))
+ r = requests.post(url, data=json.dumps(params), headers=headers)
+ if logger:
+ logger.debug(r)
+ logger.debug(r.status_code)
+ logger.debug(r.content)
+ return True
+ except Exception, e:
+ logger.error("Error [push_results_to_db('%s', '%s', '%s', " +
+ "'%s', '%s', '%s', '%s', '%s', '%s')]:" %
+ (db_url, project, case_name, pod_name, version,
+ scenario, criteria, build_tag, payload), e)
+ return False
##############################################################################
-# Copyright (c) 2015 EMC and others.
+# Copyright (c) 2016 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
##############################################################################
-# Copyright (c) 2015 EMC and others.
+# Copyright (c) 2016 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
thread.join()
self._test_executor.slaves = slaves
- job_id = self._test_executor.execute()
params = metadata
params['agent_count'] = self.agent_count
params['public_network'] = self.public_network
params['volume_size'] = self.volume_size
- self.job_db.record_workload_params(job_id, params)
+ job_id = self._test_executor.execute(params)
return job_id
##############################################################################
-# Copyright (c) 2015 EMC and others.
+# Copyright (c) 2016 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
from os.path import isfile, join
from storperf.carbon.converter import Converter
from storperf.carbon.emitter import CarbonMetricTransmitter
+from storperf.db import test_results_db
+from storperf.db.graphite_db import GraphiteDB
from storperf.db.job_db import JobDB
from storperf.fio.fio_invoker import FIOInvoker
+from storperf.utilities import dictionary
from threading import Thread
import copy
import imp
self.precondition = True
self.deadline = None
self.warm = True
+ self.metadata = None
self._queue_depths = [1, 4, 8]
self._block_sizes = [512, 4096, 16384]
self.event_listeners = set()
return imp.load_source(mname, no_ext + '.py')
return None
- def execute(self):
+ def execute(self, metadata):
self.job_db.create_job_id()
- self._workload_thread = Thread(target=self.execute_workloads, args=())
+ self.job_db.record_workload_params(metadata)
+ self.metadata = metadata
+ self._workload_thread = Thread(target=self.execute_workloads,
+ args=())
self._workload_thread.start()
return self.job_db.job_id
def execute_workloads(self):
self._terminated = False
+ self.logger.info("Starting job %s" % (self.job_db.job_id))
+
+ start_time = time.time()
+
for workload_module in self.workload_modules:
workload_name = getattr(workload_module, "__name__")
+ self.logger.info("Starting workload %s" % (workload_name))
+
constructorMethod = getattr(workload_module, workload_name)
workload = constructorMethod()
if (self.filename is not None):
self._workload_executors = []
+ self.logger.info("Completed workload %s" % (workload_name))
+ self.logger.info("Completed job %s" % (self.job_db.job_id))
+ end_time = time.time()
+ pod_name = dictionary.get_key_from_dict(self.metadata,
+ 'pod_name',
+ 'Unknown')
+ version = dictionary.get_key_from_dict(self.metadata,
+ 'version',
+ 'Unknown')
+ scenario = dictionary.get_key_from_dict(self.metadata,
+ 'scenario',
+ 'Unknown')
+ build_tag = dictionary.get_key_from_dict(self.metadata,
+ 'build_tag',
+ 'Unknown')
+ duration = end_time - start_time
+ test_db = os.environ.get('TEST_DB_URL')
+
+ if test_db is not None:
+ # I really do not like doing this. As our threads just
+ # terminated, their final results are still being spooled
+ # off to Carbon. Need to give that a little time to finish
+ time.sleep(5)
+ self.logger.info("Pushing results to %s" % (test_db))
+
+ payload = self.metadata
+ payload['timestart'] = start_time
+ payload['duration'] = duration
+ payload['status'] = 'OK'
+ graphite_db = GraphiteDB()
+ payload['metrics'] = graphite_db.fetch_averages(self.job_db.job_id)
+ criteria = {}
+ criteria['block_sizes'] = self.block_sizes
+ criteria['queue_depths'] = self.block_sizes
+
+ try:
+ test_results_db.push_results_to_db(test_db,
+ "storperf",
+ "Latency Test",
+ self.logger,
+ pod_name,
+ version,
+ scenario,
+ criteria,
+ build_tag,
+ payload)
+ except:
+ self.logger.exception("Error pushing results into Database")
+
def execute_on_node(self, workload):
invoker = FIOInvoker()
+##############################################################################
+# Copyright (c) 2016 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 storperf.db.graphite_db import GraphiteDB
-import this
import unittest
self.graphdb = GraphiteDB()
self.graphdb._job_db = self
- def test_wilcard_pattern(self):
+ def test_wildcard_pattern(self):
workload = "job_id"
expected = "job_id.*.*.*.*.*.*"
actual = self.graphdb.make_fullname_pattern(workload)
self.assertEqual(expected, actual, actual)
- def test_no_wilcard_pattern(self):
+ def test_no_wildcard_pattern(self):
workload = "job_id.workload.host.queue-depth.1.block-size.16"
actual = self.graphdb.make_fullname_pattern(workload)
self.assertEqual(workload, actual, actual)
##############################################################################
-# Copyright (c) 2015 EMC and others.
+# Copyright (c) 2016 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
def test_job_params(self):
expected = {"a": "1", "b": "2"}
- self.job.record_workload_params("ABCD", expected)
- actual = self.job.fetch_workload_params("ABCD")
+ self.job.job_id = "ABCD"
+ self.job.record_workload_params(expected)
+ actual = self.job.fetch_workload_params(self.job.job_id)
self.assertEqual(expected, actual)
--- /dev/null
+##############################################################################
+# Copyright (c) 2016 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
+##############################################################################
--- /dev/null
+##############################################################################
+# Copyright (c) 2016 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
+##############################################################################
+
+import unittest
+from storperf.utilities import dictionary
+
+
+class DictionaryTest(unittest.TestCase):
+
+ def setUp(self):
+ self.dictionary = {}
+ self.dictionary['key'] = 'value'
+ pass
+
+ def test_get_no_default(self):
+ expected = None
+ actual = dictionary.get_key_from_dict(self.dictionary, 'no-key')
+ self.assertEqual(expected, actual)
+
+ def test_get_with_default(self):
+ expected = 'value 2'
+ actual = dictionary.get_key_from_dict(
+ self.dictionary, 'no-key', expected)
+ self.assertEqual(expected, actual)
+
+ def test_get_with_value(self):
+ expected = 'value'
+ actual = dictionary.get_key_from_dict(
+ self.dictionary, 'key')
+ self.assertEqual(expected, actual)
+
+ def test_get_with_value_and_default(self):
+ expected = 'value'
+ actual = dictionary.get_key_from_dict(
+ self.dictionary, 'key', 'value 2')
+ self.assertEqual(expected, actual)
--- /dev/null
+##############################################################################
+# Copyright (c) 2016 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
+##############################################################################
--- /dev/null
+##############################################################################
+# Copyright (c) 2016 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
+##############################################################################
+
+
+def get_key_from_dict(dictionary, key, default_value=None):
+ if key in dictionary:
+ return dictionary[key]
+ else:
+ return default_value
Code Review / storperf.git / commitdiff