diff options
| -rw-r--r-- | storperf/db/configuration_db.py | 115 | ||||
| -rw-r--r-- | storperf/db/job_db.py | 264 |
2 files changed, 206 insertions, 173 deletions
diff --git a/storperf/db/configuration_db.py b/storperf/db/configuration_db.py index 649c186..700588d 100644 --- a/storperf/db/configuration_db.py +++ b/storperf/db/configuration_db.py @@ -8,74 +8,88 @@ ############################################################################## from _sqlite3 import OperationalError +from threading import Lock import logging import sqlite3 +db_mutex = Lock() + class ConfigurationDB(object): - db_name = "StorPerf.db" + db_name = "StorPerfConfig.db" def __init__(self): """ - Creates the StorPerf.db and configuration tables on demand + Creates the StorPerfConfig.db and configuration tables on demand """ self.logger = logging.getLogger(__name__) self.logger.debug("Connecting to " + ConfigurationDB.db_name) - db = sqlite3.connect(ConfigurationDB.db_name) - - cursor = db.cursor() - try: - cursor.execute('''CREATE TABLE configuration - (configuration_name text, - key text, - value text)''') - self.logger.debug("Created configuration table") - except OperationalError: - self.logger.debug("Configuration table exists") - - cursor.execute('SELECT * FROM configuration') + with db_mutex: + db = sqlite3.connect(ConfigurationDB.db_name) + + cursor = db.cursor() + try: + cursor.execute('''CREATE TABLE configuration + (configuration_name text, + key text, + value text)''') + self.logger.debug("Created configuration table") + except OperationalError: + self.logger.debug("Configuration table exists") + + cursor.execute('SELECT * FROM configuration') + db.commit() + db.close() def delete_configuration_value(self, configuration_name, key): """Deletes the value associated with the given key """ - db = sqlite3.connect(ConfigurationDB.db_name) - cursor = db.cursor() + with db_mutex: + db = sqlite3.connect(ConfigurationDB.db_name) + cursor = db.cursor() - cursor.execute( - "delete from configuration where configuration_name=? and key=?", - (configuration_name, key)) + cursor.execute( + "delete from configuration where configuration_name=? and key=?", + (configuration_name, key)) - self.logger.debug("Deleted " + configuration_name + ":" + key) + self.logger.debug("Deleted " + configuration_name + ":" + key) - db.commit() + db.commit() + db.close() def get_configuration_value(self, configuration_name, key): """Returns a string representation of the value stored with this key under the given configuration name. """ - db = sqlite3.connect(ConfigurationDB.db_name) - cursor = db.cursor() + with db_mutex: + db = sqlite3.connect(ConfigurationDB.db_name) + cursor = db.cursor() - cursor.execute( - """select value from configuration - where configuration_name = ? - and key = ?""", - (configuration_name, key,)) + cursor.execute( + """select value from configuration + where configuration_name = ? + and key = ?""", + (configuration_name, key,)) - row = cursor.fetchone() + row = cursor.fetchone() - if (row is None): - self.logger.debug( - configuration_name + ":" + key + " does not exist") - return None - else: - self.logger.debug( - configuration_name + ":" + key + " is " + str(row[0])) - return str(row[0]) + return_value = None + + if (row is None): + self.logger.debug( + configuration_name + ":" + key + " does not exist") + else: + self.logger.debug( + configuration_name + ":" + key + " is " + str(row[0])) + return_value = str(row[0]) + + db.close() + + return return_value def set_configuration_value(self, configuration_name, key, value): """Updates or creates the key under the given configuration @@ -85,19 +99,22 @@ class ConfigurationDB(object): if (value is None): return self.delete_configuration_value(configuration_name, key) - value = str(value) + with db_mutex: + value = str(value) - db = sqlite3.connect(ConfigurationDB.db_name) - cursor = db.cursor() + db = sqlite3.connect(ConfigurationDB.db_name) + cursor = db.cursor() - cursor.execute( - "delete from configuration where configuration_name=? and key=?", - (configuration_name, key)) + cursor.execute( + "delete from configuration where configuration_name=? and key=?", + (configuration_name, key)) - cursor.execute( - """insert into configuration(configuration_name, key, value) - values (?,?,?)""", (configuration_name, key, value)) + cursor.execute( + """insert into configuration(configuration_name, key, value) + values (?,?,?)""", (configuration_name, key, value)) - self.logger.debug(configuration_name + ":" + key + " set to " + value) + self.logger.debug( + configuration_name + ":" + key + " set to " + value) - db.commit() + db.commit() + db.close() diff --git a/storperf/db/job_db.py b/storperf/db/job_db.py index bec8d3f..f24ccf4 100644 --- a/storperf/db/job_db.py +++ b/storperf/db/job_db.py @@ -8,6 +8,7 @@ ############################################################################## from _sqlite3 import OperationalError +from threading import Lock import calendar import logging import sqlite3 @@ -17,140 +18,152 @@ import uuid import requests +db_mutex = Lock() + + class JobDB(object): - db_name = "StorPerf.db" + db_name = "StorPerfJob.db" def __init__(self): """ - Creates the StorPerf.db and jobs tables on demand + Creates the StorPerfJob.db and jobs tables on demand """ self.logger = logging.getLogger(__name__) self.logger.debug("Connecting to " + JobDB.db_name) self.job_id = None - db = sqlite3.connect(JobDB.db_name) - cursor = db.cursor() - try: - cursor.execute('''CREATE TABLE jobs - (job_id text, - workload text, - start text, - end text)''') - self.logger.debug("Created job table") - except OperationalError: - self.logger.debug("Job table exists") - - cursor.execute('SELECT * FROM jobs') + with db_mutex: + db = sqlite3.connect(JobDB.db_name) + cursor = db.cursor() + try: + cursor.execute('''CREATE TABLE jobs + (job_id text, + workload text, + start text, + end text)''') + self.logger.debug("Created job table") + except OperationalError: + self.logger.debug("Job table exists") + + cursor.execute('SELECT * FROM jobs') + db.commit() + db.close() def create_job_id(self): """ Returns a job id that is guaranteed to be unique in this StorPerf instance. """ - db = sqlite3.connect(JobDB.db_name) - cursor = db.cursor() + with db_mutex: + db = sqlite3.connect(JobDB.db_name) + cursor = db.cursor() - self.job_id = str(uuid.uuid4()) - row = cursor.execute( - "select * from jobs where job_id = ?", (self.job_id,)) - - while (row.fetchone() is not None): - self.logger.info("Duplicate job id found, regenerating") self.job_id = str(uuid.uuid4()) row = cursor.execute( "select * from jobs where job_id = ?", (self.job_id,)) - cursor.execute( - "insert into jobs(job_id) values (?)", (self.job_id,)) - self.logger.debug("Reserved job id " + self.job_id) - db.commit() + while (row.fetchone() is not None): + self.logger.info("Duplicate job id found, regenerating") + self.job_id = str(uuid.uuid4()) + row = cursor.execute( + "select * from jobs where job_id = ?", (self.job_id,)) + + cursor.execute( + "insert into jobs(job_id) values (?)", (self.job_id,)) + self.logger.debug("Reserved job id " + self.job_id) + db.commit() + db.close() def start_workload(self, workload_name): """ Records the start time for the given workload """ - if (self.job_id is None): - self.create_job_id() - - db = sqlite3.connect(JobDB.db_name) - cursor = db.cursor() - - now = str(calendar.timegm(time.gmtime())) + with db_mutex: + if (self.job_id is None): + self.create_job_id() - row = cursor.execute( - """select * from jobs - where job_id = ? - and workload = ?""", - (self.job_id, workload_name,)) + db = sqlite3.connect(JobDB.db_name) + cursor = db.cursor() - if (row.fetchone() is None): - cursor.execute( - """insert into jobs - (job_id, - workload, - start) - values (?, ?, ?)""", - (self.job_id, - workload_name, - now,)) - else: - self.logger.warn("Duplicate start time for workload " - + workload_name) - cursor.execute( - """update jobs set - job_id = ?, - start = ? - where workload = ?""", - (self.job_id, - now, - workload_name,)) + now = str(calendar.timegm(time.gmtime())) - db.commit() + row = cursor.execute( + """select * from jobs + where job_id = ? + and workload = ?""", + (self.job_id, workload_name,)) + + if (row.fetchone() is None): + cursor.execute( + """insert into jobs + (job_id, + workload, + start) + values (?, ?, ?)""", + (self.job_id, + workload_name, + now,)) + else: + self.logger.warn("Duplicate start time for workload " + + workload_name) + cursor.execute( + """update jobs set + job_id = ?, + start = ? + where workload = ?""", + (self.job_id, + now, + workload_name,)) + + db.commit() + db.close() def end_workload(self, workload_name): """ Records the end time for the given workload """ - if (self.job_id is None): - self.create_job_id() - - db = sqlite3.connect(JobDB.db_name) - cursor = db.cursor() - now = str(calendar.timegm(time.gmtime())) - - row = cursor.execute( - """select * from jobs - where job_id = ? - and workload = ?""", - (self.job_id, workload_name,)) - - if (row.fetchone() is None): - self.logger.warn("No start time recorded for workload " - + workload_name) - cursor.execute( - """insert into jobs - (job_id, - workload, - start, - end) - values (?, ?, ?, ?)""", - (self.job_id, - workload_name, - now, - now)) - else: - cursor.execute( - """update jobs set - job_id = ?, - end = ? - where workload = ?""", - (self.job_id, - now, - workload_name,)) + with db_mutex: + if (self.job_id is None): + self.create_job_id() - db.commit() + db = sqlite3.connect(JobDB.db_name) + cursor = db.cursor() + now = str(calendar.timegm(time.gmtime())) + + row = cursor.execute( + """select * from jobs + where job_id = ? + and workload = ?""", + (self.job_id, workload_name,)) + + if (row.fetchone() is None): + self.logger.warn("No start time recorded for workload " + + workload_name) + cursor.execute( + """insert into jobs + (job_id, + workload, + start, + end) + values (?, ?, ?, ?)""", + (self.job_id, + workload_name, + now, + now)) + else: + cursor.execute( + """update jobs set + job_id = ?, + end = ? + where workload = ?""", + (self.job_id, + now, + workload_name,)) + + db.commit() + db.close() def fetch_results(self, workload_prefix=""): if (workload_prefix is None): @@ -165,36 +178,39 @@ class JobDB(object): self.logger.debug("Workload like: " + workload_prefix) - db = sqlite3.connect(JobDB.db_name) - cursor = db.cursor() - cursor.execute("""select start, end, workload - from jobs where workload like ?""", - (workload_prefix,)) + with db_mutex: - while (True): - row = cursor.fetchone() - if (row is None): - break + db = sqlite3.connect(JobDB.db_name) + cursor = db.cursor() + cursor.execute("""select start, end, workload + from jobs where workload like ?""", + (workload_prefix,)) - start_time = str(row[0]) - end_time = str(row[1]) - workload = str(row[2]) + while (True): + row = cursor.fetchone() + if (row is None): + break - # for most of these stats, we just want the final one - # as that is cumulative average or whatever for the whole - # run + start_time = str(row[0]) + end_time = str(row[1]) + workload = str(row[2]) - self.logger.info("workload=" + workload + - "start=" + start_time + " end=" + end_time) + # for most of these stats, we just want the final one + # as that is cumulative average or whatever for the whole + # run - request = 'http://127.0.0.1:8000/render/?target=*.' + self.job_id + \ - '.' + workload + '.jobs.1.*.clat.mean&format=json&from=' + \ - start_time + "&until=" + end_time + self.logger.info("workload=" + workload + + "start=" + start_time + " end=" + end_time) - response = requests.get(request) + request = 'http://127.0.0.1:8000/render/?target=*.' + self.job_id + \ + '.' + workload + '.jobs.1.*.clat.mean&format=json&from=' + \ + start_time + "&until=" + end_time - if (response.status_code == 200): - data = response.json() - print data - else: - pass + response = requests.get(request) + + if (response.status_code == 200): + data = response.json() + print data + else: + pass + db.close() |