diff options
| -rwxr-xr-x | ci/daily.sh | 36 | ||||
| -rw-r--r-- | storperf/test_executor.py | 6 | ||||
| -rw-r--r-- | storperf/utilities/data_handler.py | 19 | ||||
| -rw-r--r-- | storperf/utilities/thread_gate.py | 60 | ||||
| -rw-r--r-- | tests/utilities_tests/data_handler_test.py | 11 |
5 files changed, 73 insertions, 59 deletions
diff --git a/ci/daily.sh b/ci/daily.sh index e3b64cc..1e77d67 100755 --- a/ci/daily.sh +++ b/ci/daily.sh @@ -96,32 +96,22 @@ do | awk '/Status/ {print $2}' | sed 's/"//g'` done -for WORKLOAD in ws wr rs rr rw +export QUEUE_DEPTH=1,2,8 +export BLOCK_SIZE=2048,8192,16384 +export WORKLOAD=ws,wr,rs,rr,rw +export SCENARIO_NAME="${CINDER_BACKEND}_${WORKLOAD}" + +JOB=`$WORKSPACE/ci/start_job.sh \ + | awk '/job_id/ {print $2}' | sed 's/"//g'` +JOB_STATUS=`curl -s -X GET "http://127.0.0.1:5000/api/v1.0/jobs?id=$JOB&type=status" \ + | awk '/Status/ {print $2}' | sed 's/"//g'` +while [ "$JOB_STATUS" != "Completed" ] do - for BLOCK_SIZE in 2048 8192 16384 - do - for QUEUE_DEPTH in 1 2 8 - do - export QUEUE_DEPTH - export BLOCK_SIZE - export WORKLOAD - export SCENARIO_NAME="${CINDER_BACKEND}_${WORKLOAD}" - - JOB=`$WORKSPACE/ci/start_job.sh \ - | awk '/job_id/ {print $2}' | sed 's/"//g'` - JOB_STATUS=`curl -s -X GET "http://127.0.0.1:5000/api/v1.0/jobs?id=$JOB&type=status" \ - | awk '/Status/ {print $2}' | sed 's/"//g'` - while [ "$JOB_STATUS" != "Completed" ] - do - sleep 60 - JOB_STATUS=`curl -s -X GET "http://127.0.0.1:5000/api/v1.0/jobs?id=$JOB&type=status" \ - | awk '/Status/ {print $2}' | sed 's/"//g'` - done - done - done + sleep 60 + JOB_STATUS=`curl -s -X GET "http://127.0.0.1:5000/api/v1.0/jobs?id=$JOB&type=status" \ + | awk '/Status/ {print $2}' | sed 's/"//g'` done - echo "Deleting stack for cleanup" curl -X DELETE --header 'Accept: application/json' 'http://127.0.0.1:5000/api/v1.0/configurations' diff --git a/storperf/test_executor.py b/storperf/test_executor.py index 9c9393f..cda6c78 100644 --- a/storperf/test_executor.py +++ b/storperf/test_executor.py @@ -177,9 +177,9 @@ class TestExecutor(object): def terminate(self): self._terminated = True self.end_time = time.time() - return self._terminate_current_run() + return self.terminate_current_run() - def _terminate_current_run(self): + def terminate_current_run(self): self.logger.info("Terminating current run") terminated_hosts = [] for workload in self._workload_executors: @@ -243,7 +243,7 @@ class TestExecutor(object): if self.deadline is not None \ and not workload_name.startswith("_"): event = scheduler.enter(self.deadline * 60, 1, - self._terminate_current_run, + self.terminate_current_run, ()) t = Thread(target=scheduler.run, args=()) t.start() diff --git a/storperf/utilities/data_handler.py b/storperf/utilities/data_handler.py index ebc1bfd..0aae3b1 100644 --- a/storperf/utilities/data_handler.py +++ b/storperf/utilities/data_handler.py @@ -17,6 +17,7 @@ from storperf.utilities import math as math from storperf.utilities import steady_state as SteadyState from time import sleep import time +import json class DataHandler(object): @@ -61,13 +62,21 @@ class DataHandler(object): if not steady: steady_state = False - executor.metadata['report_data'] = metrics - executor.metadata['steady_state'] = steady_state + workload = '.'.join(executor.current_workload.split('.')[1:6]) + + if 'report_data' not in executor.metadata: + executor.metadata['report_data'] = {} + + if 'steady_state' not in executor.metadata: + executor.metadata['steady_state'] = {} + + executor.metadata['report_data'][workload] = metrics + executor.metadata['steady_state'][workload] = steady_state workload_name = executor.current_workload.split('.')[1] if steady_state and not workload_name.startswith('_'): - executor.terminate() + executor.terminate_current_run() def _lookup_prior_data(self, executor, metric, io_type): workload = executor.current_workload @@ -112,7 +121,7 @@ class DataHandler(object): duration = latest_timestamp - earliest_timestamp if (duration < 60 * self.samples): self.logger.debug("Only %s minutes of samples, ignoring" % - (duration / 60,)) + ((duration / 60 + 1),)) return False return SteadyState.steady_state(data_series) @@ -160,6 +169,6 @@ class DataHandler(object): scenario, criteria, build_tag, - payload) + json.dumps(payload)) except: self.logger.exception("Error pushing results into Database") diff --git a/storperf/utilities/thread_gate.py b/storperf/utilities/thread_gate.py index 295b8be..38acbb1 100644 --- a/storperf/utilities/thread_gate.py +++ b/storperf/utilities/thread_gate.py @@ -12,6 +12,7 @@ number of callers. """ import logging import time +from threading import Lock class FailureToReportException(Exception): @@ -26,6 +27,7 @@ class ThreadGate(object): self._timeout = timeout self._registrants = {} self._creation_time = time.time() + self._lock = Lock() """ Calling this method returns a true or false, indicating that enough @@ -33,31 +35,33 @@ class ThreadGate(object): """ def report(self, gate_id): - now = time.time() - self._registrants[gate_id] = now - ready = True - self.logger.debug("Gate report for %s", gate_id) - - total_missing = self._gate_size - len(self._registrants) - if total_missing > 0: - self.logger.debug("Not all registrants have reported in") - time_since_creation = now - self._creation_time - if (time_since_creation > (self._timeout * 2)): - self.logger.error("%s registrant(s) have never reported in", - total_missing) - raise FailureToReportException - return False - - for k, v in self._registrants.items(): - time_since_last_report = now - v - if time_since_last_report > self._timeout: - self.logger.debug("Registrant %s last reported %s ago", - k, time_since_last_report) - ready = False - - self.logger.debug("Gate pass? %s", ready) - - if ready: - self._registrants.clear() - - return ready + with self._lock: + now = time.time() + self._registrants[gate_id] = now + ready = True + self.logger.debug("Gate report for %s", gate_id) + + total_missing = self._gate_size - len(self._registrants) + if total_missing > 0: + self.logger.debug("Not all registrants have reported in") + time_since_creation = now - self._creation_time + if (time_since_creation > (self._timeout * 2)): + self.logger.error( + "%s registrant(s) have never reported in", + total_missing) + raise FailureToReportException + return False + + for k, v in self._registrants.items(): + time_since_last_report = now - v + if time_since_last_report > self._timeout: + self.logger.debug("Registrant %s last reported %s ago", + k, time_since_last_report) + ready = False + + self.logger.debug("Gate pass? %s", ready) + + if ready: + self._registrants.clear() + + return ready diff --git a/tests/utilities_tests/data_handler_test.py b/tests/utilities_tests/data_handler_test.py index b175c87..8115c6d 100644 --- a/tests/utilities_tests/data_handler_test.py +++ b/tests/utilities_tests/data_handler_test.py @@ -57,6 +57,9 @@ class DataHandlerTest(unittest.TestCase): def terminate(self): self._terminated = True + def terminate_current_run(self): + self._terminated = True + @mock.patch("time.time") @mock.patch.dict(os.environ, {'TEST_DB_URL': 'mock'}) @mock.patch("storperf.db.graphite_db.GraphiteDB.fetch_series") @@ -163,18 +166,22 @@ class DataHandlerTest(unittest.TestCase): self.assertEqual(False, self._terminated) self.assertEqual(expected_slope, self.metadata['report_data'] + ['rw.queue-depth.8.block-size.8192'] ['lat.mean'] ['read'] ['slope']) self.assertEqual(expected_range, self.metadata['report_data'] + ['rw.queue-depth.8.block-size.8192'] ['lat.mean'] ['read'] ['range']) self.assertEqual(expected_average, self.metadata['report_data'] + ['rw.queue-depth.8.block-size.8192'] ['lat.mean'] ['read'] ['average']) self.assertEqual(series, self.metadata['report_data'] + ['rw.queue-depth.8.block-size.8192'] ['lat.mean'] ['read'] ['series']) @@ -211,18 +218,22 @@ class DataHandlerTest(unittest.TestCase): self.data_handler.data_event(self) self.assertEqual(expected_slope, self.metadata['report_data'] + ['rw.queue-depth.8.block-size.8192'] ['lat.mean'] ['read'] ['slope']) self.assertEqual(expected_range, self.metadata['report_data'] + ['rw.queue-depth.8.block-size.8192'] ['lat.mean'] ['read'] ['range']) self.assertEqual(expected_average, self.metadata['report_data'] + ['rw.queue-depth.8.block-size.8192'] ['lat.mean'] ['read'] ['average']) self.assertEqual(series, self.metadata['report_data'] + ['rw.queue-depth.8.block-size.8192'] ['lat.mean'] ['read'] ['series']) |