[rhmessaging-commits] rhmessaging commits: r2900 - mgmt/trunk/mint/python/mint.

Author: nunofsantos Date: 2008-12-01 14:56:35 -0500 (Mon, 01 Dec 2008) New Revision: 2900 Added: mgmt/trunk/mint/python/mint/dbexpire.py Modified: mgmt/trunk/mint/python/mint/__init__.py mgmt/trunk/mint/python/mint/schema.py mgmt/trunk/mint/python/mint/schemaparser.py mgmt/trunk/mint/python/mint/sql.py Log: Add a thread to do periodic cleanup of expired/old records in the database; currently set to delete any stats records older then 24 hours, as well as Job records older than 24 hours Modified: mgmt/trunk/mint/python/mint/__init__.py =================================================================== --- mgmt/trunk/mint/python/mint/__init__.py 2008-12-01 16:49:16 UTC (rev 2899) +++ mgmt/trunk/mint/python/mint/__init__.py 2008-12-01 19:56:35 UTC (rev 2900) @@ -12,6 +12,7 @@ from mint.schema import * from mint import update from mint.cache import MintCache +from mint.dbexpire import DBExpireThread from qmf.console import ClassKey log = logging.getLogger("mint") @@ -324,6 +325,7 @@ self.__lock = RLock() self.dbConn = None + self.dbExpireThread = dbexpire.DBExpireThread(self) self.updateThread = update.ModelUpdateThread(self) self.mgmtSession = qmf.console.Session(self, manageConnections=True) self.outstandingMethodCalls = dict() @@ -350,9 +352,11 @@ def start(self): self.updateThread.start() + self.dbExpireThread.start() def stop(self): self.updateThread.stop() + self.dbExpireThread.stop() def getSession(self): return self.mgmtSession Added: mgmt/trunk/mint/python/mint/dbexpire.py =================================================================== --- mgmt/trunk/mint/python/mint/dbexpire.py (rev 0) +++ mgmt/trunk/mint/python/mint/dbexpire.py 2008-12-01 19:56:35 UTC (rev 2900) @@ -0,0 +1,69 @@ +import logging +import mint +import time +from threading import Thread +from mint.schema import * +from sql import * + +log = logging.getLogger("mint.dbexpire") + +class DBExpireThread(Thread): + def __init__(self, model, frequency=600, expiration=24*3600): + super(DBExpireThread, self).__init__() + + self.model = model + self.setDaemon(False) + self.stopRequested = False + self.frequency = frequency + self.expiration = expiration + + frequency_out, frequency_unit = self.convertTimeUnits(frequency) + expiration_out, expiration_unit = self.convertTimeUnits(expiration) + log.debug("Expiring database records older than %d %s, every %d %s" \ + % (expiration_out, expiration_unit, frequency_out, frequency_unit)) + + def run(self): + ops = [] + for cls in mint.schema.statsClasses: + ops.append(SqlExpire(eval(cls))) + ops.append(SqlExpire(Job)) + + attrs = dict() + attrs["expiration"] = self.expiration + + conn = self.model.dbConn.getConnection() + + while True: + try: + if self.stopRequested: + break + + cursor = conn.cursor() + rowcount = 0 + for op in ops: + rowcount += op.execute(cursor, attrs) + log.debug("%d records expired" % (rowcount)) + conn.commit() + + time.sleep(self.frequency) + except: + conn.rollback() + log.exception("DB cleanup failed") + + def stop(self): + self.stopRequested = True + + def convertTimeUnits(self, t): + if t / (24*3600) >= 1: + t_out = t / (24*3600) + t_unit = "days" + elif t / 3600 >= 1: + t_out = t / 3600 + t_unit = "hours" + elif t / 60 >= 1: + t_out = t / 60 + t_unit = "minutes" + else: + t_out = t + t_unit = "seconds" + return (t_out, t_unit) Property changes on: mgmt/trunk/mint/python/mint/dbexpire.py ___________________________________________________________________ Name: svn:eol-style + native Modified: mgmt/trunk/mint/python/mint/schema.py =================================================================== --- mgmt/trunk/mint/python/mint/schema.py 2008-12-01 16:49:16 UTC (rev 2899) +++ mgmt/trunk/mint/python/mint/schema.py 2008-12-01 19:56:35 UTC (rev 2900) @@ -1479,3 +1479,7 @@ Sysimage.sqlmeta.addJoin(SQLMultipleJoin('SysimageStats', joinMethodName='stats')) + +entityClasses = ['Slot', 'Job', 'Scheduler', 'Submitter', 'Negotiator', 'Collector', 'Master', 'Acl', 'Cluster', 'Store', 'Journal', 'System', 'Broker', 'Agent', 'Vhost', 'Queue', 'Exchange', 'Binding', 'ClientConnection', 'Link', 'Bridge', 'Session', 'Sysimage'] + +statsClasses = ['SlotStats', 'JobStats', 'SchedulerStats', 'SubmitterStats', 'NegotiatorStats', 'CollectorStats', 'MasterStats', 'AclStats', 'ClusterStats', 'StoreStats', 'JournalStats', 'SystemStats', 'BrokerStats', 'AgentStats', 'VhostStats', 'QueueStats', 'ExchangeStats', 'BindingStats', 'ClientConnectionStats', 'LinkStats', 'BridgeStats', 'SessionStats', 'SysimageStats'] Modified: mgmt/trunk/mint/python/mint/schemaparser.py =================================================================== --- mgmt/trunk/mint/python/mint/schemaparser.py 2008-12-01 16:49:16 UTC (rev 2899) +++ mgmt/trunk/mint/python/mint/schemaparser.py 2008-12-01 19:56:35 UTC (rev 2900) @@ -13,6 +13,8 @@ self.currentClass = "" self.pythonOutput = "" self.finalPythonOutput = "" + self.entityClasses = [] + self.statsClasses = [] self.groups = dict() # mapping between xml schema types and database column types # see xml/MintTypes.xml @@ -139,10 +141,12 @@ self.pythonOutput += "\nclass %s(SQLObject):\n" % (pythonName) self.generateLazyUpdate() if (stats): + self.statsClasses.append(str(pythonName)) self.generateTimestampAttrib("Update") self.generateForeignKeyAttrib(colPythonName[0].lower() + colPythonName[1:], keyPythonName) self.generateMultipleJoin(origPythonName, pythonName, "stats") else: + self.entityClasses.append(str(pythonName)) self.generateAttrib("qmfBrokerId", "StringCol", "length=1000") self.generateAttrib("qmfScopeId", "BigIntCol") self.generateAttrib("qmfObjectId", "BigIntCol") @@ -231,6 +235,8 @@ self.generateClassAttribs(cls["@name"], self.groups[clsGroup["@name"]][1]) self.endClass() self.pythonOutput += "\n\n" + self.finalPythonOutput += "\nentityClasses = %s\n" % (self.entityClasses) + self.finalPythonOutput += "\nstatsClasses = %s\n" % (self.statsClasses) outputFile.write(self.pythonOutput + self.finalPythonOutput) outputFile.close() Modified: mgmt/trunk/mint/python/mint/sql.py =================================================================== --- mgmt/trunk/mint/python/mint/sql.py 2008-12-01 16:49:16 UTC (rev 2899) +++ mgmt/trunk/mint/python/mint/sql.py 2008-12-01 19:56:35 UTC (rev 2900) @@ -50,6 +50,7 @@ self.time = clock() - start else: cursor.execute(self.text, values) + return cursor.rowcount except: log.warn("Text: %s", self.text) @@ -138,6 +139,28 @@ return sql +class SqlExpire(SqlOperation): + def __init__(self, cls): + super(SqlExpire, self).__init__("expire") + + self.cls = cls + + def generate(self): + table = self.cls.sqlmeta.table + if table.endswith("_stats"): + parent_table = table[0:table.find("_stats")] + sql = """ + delete from %s + where qmf_update_time < now() - interval '%%(expiration)s seconds' + and id not in (select stats_curr_id from %s) + """ % (table, parent_table) + else: + sql = """ + delete from %s + where qmf_create_time < now() - interval '%%(expiration)s seconds' + """ % (table) + return sql + class SqlGetBrokerRegistration(SqlOperation): def __init__(self): super(SqlGetBrokerRegistration, self).__init__("get_broker_reg")