Author: justi9 Date: 2007-11-13 14:45:55 -0500 (Tue, 13 Nov 2007) New Revision: 1303 Added: mgmt/mint/python/mint/schema.sql Removed: mgmt/mint/python/mint/updater.py Modified: mgmt/mint/bin/mint-test mgmt/mint/python/mint/__init__.py mgmt/mint/python/mint/schema.py Log: Updates mint-test to use Nuno's new model stuff. Moves model.py's Model to MintModel in mint/__init__.py. Modified: mgmt/mint/bin/mint-test =================================================================== --- mgmt/mint/bin/mint-test 2007-11-13 18:36:01 UTC (rev 1302) +++ mgmt/mint/bin/mint-test 2007-11-13 19:45:55 UTC (rev 1303) @@ -17,18 +17,12 @@ usage() from mint import * -from mint.updater import * from qpid.management import ManagedBroker def do_main(dburi, brokerhost, brokerport): - model = MintModel(dburi) - model.init() - - broker = ManagedBroker(host=brokerhost, port=brokerport) + model = MintModel() + model.addManagedBroker(brokerhost, brokerport) - updater = MintUpdater(model, broker) - updater.start() - def main(): if len(sys.argv) != 3: usage() Modified: mgmt/mint/python/mint/__init__.py =================================================================== --- mgmt/mint/python/mint/__init__.py 2007-11-13 18:36:01 UTC (rev 1302) +++ mgmt/mint/python/mint/__init__.py 2007-11-13 19:45:55 UTC (rev 1303) @@ -1,3 +1,5 @@ +from qpid.management import ManagedBroker +from datetime import * from sqlobject import * from schema import * @@ -2,7 +4,179 @@ -class MintModel(object): - def __init__(self, dburi): - self.dburi = dburi +class MintModel: + currentMethodId = None + outstandingMethodCalls = None + managedBrokers = None + + def __init__(self): + self.currentMethodId = 1 + self.outstandingMethodCalls = dict() + self.managedBrokers = dict() + + def getQueueByOriginalId(self, id, create=False): + queue = None + try: + queue = MgmtQueue.selectBy(idOriginal=id)[:1][0] + except IndexError: + if (create): queue = MgmtQueue(idOriginal=id) + return queue - def init(self): - pass + def getQueueByName(self, name, vhost, create=False): + queue = None + try: + queue = MgmtQueue.selectBy(name=name, mgmtVhost=vhost)[:1][0] + except IndexError: + if (create): queue = MgmtQueue(name=name, mgmtVhost=vhost) + return queue + + def getVhostByName(self, name, broker, create=False): + vhost = None + try: + vhost = MgmtVhost.selectBy(name=name, mgmtBroker=broker)[:1][0] + except IndexError: + if (create): vhost = MgmtVhost(name=name, mgmtBroker=broker) + return vhost + + def getVhostByOriginalId(self, id, create=False): + vhost = None + try: + vhost = MgmtVhost.selectBy(idOriginal=id)[:1][0] + except IndexError: + if (create): vhost = MgmtVhost(idOriginal=id) + return vhost + + def getBrokerByPort(self, port, system, create=False): + broker = None + try: + broker = MgmtBroker.selectBy(port=port, mgmtSystem=system)[:1][0] + except IndexError: + if (create): broker = MgmtBroker(port=port, mgmtSystem=system) + return broker + + def getBrokerByOriginalId(self, id, create=False): + broker = None + try: + broker = MgmtBroker.selectBy(idOriginal=id)[:1][0] + except IndexError: + if (create): broker = MgmtBroker(idOriginal=id) + return broker + + def getSystemByOriginalId(self, id, create=False): + system = None + try: + system = MgmtSystem.selectBy(idOriginal=id)[:1][0] + except IndexError: + if (create): system = MgmtSystem(idOriginal=id) + return system + + def sanitizeDict(self, d): + for k in d.iterkeys(): + if (k.endswith("Id")): + d[self.convertKey(k)] = d.pop(k) + elif (k == "id"): + d[self.convertKey(k)] = d.pop(k) + for k in d.iterkeys(): + if (k.endswith("Ref")): + d[self.convertKey(k)] = d.pop(k) + return d + + def convertKey(self, k): + if (k == "id"): + return k + "Original" + if (k.endswith("Id")): + return k + "ent" + elif (k.endswith("Ref")): + oldK = k + k = k[0].upper() + k[1:] + return "mgmt" + k.replace("Ref", "ID") + + def configCallback(self, broker, objectName, list, timestamps): + print "\nCONFIG---------------------------------------------------" + print "broker=" + broker + print objectName + print list + result = None + d = self.sanitizeDict(dict(list)) + d["managedBroker"] = self.managedBrokers[broker] + print d + + d["recTime"] = datetime.fromtimestamp(timestamps[0]/1000000000) + d["creationTime"] = datetime.fromtimestamp(timestamps[1]/1000000000) + if (objectName == "queue"): + print "* QUEUE" + queue = self.getQueueByName(d["name"], self.getVhostByOriginalId(d.pop(self.convertKey("vhostRef"))), True) + queue.set(**d) + print "queue id = %d" % (queue.id) + result = queue + elif (objectName == "vhost"): + print "* VHOST" + vhost = self.getVhostByName(d["name"], self.getBrokerByOriginalId(d.pop(self.convertKey("brokerRef"))), True) + vhost.set(**d) + print "vhost id = %d" % (vhost.id) + result = vhost + elif (objectName == "broker"): + print "* BROKER" + d.pop(self.convertKey("systemRef")) + broker = self.getBrokerByPort(d["port"], self.getSystemByOriginalId("0"), True) + broker.set(**d) + broker.sync() + print "broker id = %d" % (broker.id) + result = broker + print "END CONFIG---------------------------------------------------\n" + return result + + def instCallback(self, broker, objectName, list, timestamps): + print "\nINST---------------------------------------------------" + print "broker=" + broker + print objectName + print list + result = None + d = self.sanitizeDict(dict(list)) + if (objectName == "queue"): + print "* QUEUE" + queue = self.getQueueByOriginalId(d[self.convertKey("id")]) + d["mgmtQueue"] = queue.id + d["recTime"] = datetime.fromtimestamp(timestamps[0]/1000000000) + queueStats = MgmtQueueStats() + queueStats.set(**d) + d = dict() + if (timestamps[2] != 0): + d["deletionTime"] = datetime.fromtimestamp(timestamps[2]/1000000000) + d["mgmtQueueStats"] = queueStats + queue.set(**d) + print queue.id + result = queueStats + elif (objectName == "vhost"): + print "* VHOST" + elif (objectName == "broker"): + print "* BROKER" + print "END INST---------------------------------------------------\n" + return result + + def methodCallback(self, broker, methodId, errorNo, errorText, args): + print "\nMETHOD---------------------------------------------------" + print "broker=" + broker + print "MethodId=%d" % (methodId) + print "Error: %d %s" % (errorNo, errorText) + print args + method = self.outstandingMethodCalls.pop(methodId) + method(errorText, args) + print "END METHOD---------------------------------------------------\n" + + def addManagedBroker(self, host, port): + broker = ManagedBroker(host=host, port=port) + label = "%s:%d" % (host, port) + self.managedBrokers[label] = broker + broker.configListener(label, self.configCallback) + broker.instrumentationListener (label, self.instCallback) + broker.methodListener (label, self.methodCallback) + broker.start() + return label + + def registerCallback(self, callback): + self.currentMethodId += 1 + methodId = self.currentMethodId + self.outstandingMethodCalls[methodId] = callback + return methodId + + def allSystems(self): + return MgmtSystem.select() Modified: mgmt/mint/python/mint/schema.py =================================================================== --- mgmt/mint/python/mint/schema.py 2007-11-13 18:36:01 UTC (rev 1302) +++ mgmt/mint/python/mint/schema.py 2007-11-13 19:45:55 UTC (rev 1303) @@ -1,127 +1,241 @@ from sqlobject import * -class MgmtServer(SQLObject): +class MgmtSystemStats(SQLObject): class sqlmeta: fromDatabase = True - def joinCluster(): - pass - def leaveCluster(): - pass +class MgmtSystem(SQLObject): + schemaId = 1 + schemaName = "system" + managedBroker = None + class sqlmeta: + fromDatabase = True + _SO_class_Mgmt_system_stats = None +class MgmtBrokerStats(SQLObject): + class sqlmeta: + fromDatabase = True -class MgmtServerStats(SQLObject): +class MgmtBroker(SQLObject): + schemaId = 2 + schemaName = "broker" + managedBroker = None class sqlmeta: fromDatabase = True + def joinCluster(self, model, callbackMethod, clusterName): + actualArgs = dict() + actualArgs["clusterName"] = clusterName + methodId = model.registerCallback(callbackMethod) + self.managedBroker.method(methodId, self.idOriginal, self.schemaName, "joinCluster", args=actualArgs, packageName="qpid") + def leaveCluster(self, model, callbackMethod): + actualArgs = dict() + methodId = model.registerCallback(callbackMethod) + self.managedBroker.method(methodId, self.idOriginal, self.schemaName, "leaveCluster", args=actualArgs, packageName="qpid") + def echo(self, model, callbackMethod, sequence, body): + actualArgs = dict() + actualArgs["sequence"] = sequence + actualArgs["body"] = body + methodId = model.registerCallback(callbackMethod) + self.managedBroker.method(methodId, self.idOriginal, self.schemaName, "echo", args=actualArgs, packageName="qpid") + def crash(self, model, callbackMethod): + """Temporary test method to crash the broker""" + actualArgs = dict() + methodId = model.registerCallback(callbackMethod) + self.managedBroker.method(methodId, self.idOriginal, self.schemaName, "crash", args=actualArgs, packageName="qpid") + _SO_class_Mgmt_broker_stats = None + _SO_class_Mgmt_system = None +class MgmtVhostStats(SQLObject): + class sqlmeta: + fromDatabase = True + +MgmtSystem.sqlmeta.addJoin(MultipleJoin('MgmtBroker', joinMethodName='allBrokers')) + class MgmtVhost(SQLObject): + schemaId = 3 + schemaName = "vhost" + managedBroker = None class sqlmeta: fromDatabase = True + _SO_class_Mgmt_vhost_stats = None + _SO_class_Mgmt_broker = None -class MgmtVhostStats(SQLObject): +class MgmtQueueStats(SQLObject): class sqlmeta: fromDatabase = True +MgmtBroker.sqlmeta.addJoin(MultipleJoin('MgmtVhost', joinMethodName='allVhosts')) + class MgmtQueue(SQLObject): + schemaId = 4 + schemaName = "queue" + managedBroker = None class sqlmeta: fromDatabase = True - def purge(): + def purge(self, model, callbackMethod): """Discard all messages on queue""" - pass - - def increaseDiskSize(): + actualArgs = dict() + methodId = model.registerCallback(callbackMethod) + self.managedBroker.method(methodId, self.idOriginal, self.schemaName, "purge", args=actualArgs, packageName="qpid") + def increaseDiskSize(self, model, callbackMethod, pages): """Increase number of disk pages allocated for this queue""" - pass + actualArgs = dict() + actualArgs["pages"] = pages + methodId = model.registerCallback(callbackMethod) + self.managedBroker.method(methodId, self.idOriginal, self.schemaName, "increaseDiskSize", args=actualArgs, packageName="qpid") + _SO_class_Mgmt_queue_stats = None + _SO_class_Mgmt_vhost = None - -class MgmtQueueStats(SQLObject): +class MgmtExchangeStats(SQLObject): class sqlmeta: fromDatabase = True +MgmtVhost.sqlmeta.addJoin(MultipleJoin('MgmtQueue', joinMethodName='allQueues')) + class MgmtExchange(SQLObject): + schemaId = 5 + schemaName = "exchange" + managedBroker = None class sqlmeta: fromDatabase = True + _SO_class_Mgmt_exchange_stats = None + _SO_class_Mgmt_vhost = None -class MgmtExchangeStats(SQLObject): +class MgmtBindingStats(SQLObject): class sqlmeta: fromDatabase = True +MgmtVhost.sqlmeta.addJoin(MultipleJoin('MgmtExchange', joinMethodName='allExchanges')) + class MgmtBinding(SQLObject): + schemaId = 6 + schemaName = "binding" + managedBroker = None class sqlmeta: fromDatabase = True + _SO_class_Mgmt_binding_stats = None + _SO_class_Mgmt_queue = None + _SO_class_Mgmt_exchange = None -class MgmtBindingStats(SQLObject): +class MgmtClientStats(SQLObject): class sqlmeta: fromDatabase = True +MgmtQueue.sqlmeta.addJoin(MultipleJoin('MgmtBinding', joinMethodName='allBindings')) +MgmtExchange.sqlmeta.addJoin(MultipleJoin('MgmtBinding', joinMethodName='allBindings')) + class MgmtClient(SQLObject): + schemaId = 7 + schemaName = "client" + managedBroker = None class sqlmeta: fromDatabase = True - def close(): - pass + def close(self, model, callbackMethod): + actualArgs = dict() + methodId = model.registerCallback(callbackMethod) + self.managedBroker.method(methodId, self.idOriginal, self.schemaName, "close", args=actualArgs, packageName="qpid") + def detach(self, model, callbackMethod): + actualArgs = dict() + methodId = model.registerCallback(callbackMethod) + self.managedBroker.method(methodId, self.idOriginal, self.schemaName, "detach", args=actualArgs, packageName="qpid") + _SO_class_Mgmt_client_stats = None + _SO_class_Mgmt_vhost = None - def detach(): - pass - - -class MgmtClientStats(SQLObject): +class MgmtSessionStats(SQLObject): class sqlmeta: fromDatabase = True +MgmtVhost.sqlmeta.addJoin(MultipleJoin('MgmtClient', joinMethodName='allClients')) + class MgmtSession(SQLObject): + schemaId = 8 + schemaName = "session" + managedBroker = None class sqlmeta: fromDatabase = True - def solicitAck(): - pass + def solicitAck(self, model, callbackMethod): + actualArgs = dict() + methodId = model.registerCallback(callbackMethod) + self.managedBroker.method(methodId, self.idOriginal, self.schemaName, "solicitAck", args=actualArgs, packageName="qpid") + def detach(self, model, callbackMethod): + actualArgs = dict() + methodId = model.registerCallback(callbackMethod) + self.managedBroker.method(methodId, self.idOriginal, self.schemaName, "detach", args=actualArgs, packageName="qpid") + def resetLifespan(self, model, callbackMethod): + actualArgs = dict() + methodId = model.registerCallback(callbackMethod) + self.managedBroker.method(methodId, self.idOriginal, self.schemaName, "resetLifespan", args=actualArgs, packageName="qpid") + def close(self, model, callbackMethod): + actualArgs = dict() + methodId = model.registerCallback(callbackMethod) + self.managedBroker.method(methodId, self.idOriginal, self.schemaName, "close", args=actualArgs, packageName="qpid") + _SO_class_Mgmt_session_stats = None + _SO_class_Mgmt_vhost = None + _SO_class_Mgmt_client = None - def detach(): - pass - - def resetLifespan(): - pass - - def close(): - pass - - -class MgmtSessionStats(SQLObject): +class MgmtDestinationStats(SQLObject): class sqlmeta: fromDatabase = True +MgmtVhost.sqlmeta.addJoin(MultipleJoin('MgmtSession', joinMethodName='allSessions')) +MgmtClient.sqlmeta.addJoin(MultipleJoin('MgmtSession', joinMethodName='allSessions')) + class MgmtDestination(SQLObject): + schemaId = 9 + schemaName = "destination" + managedBroker = None class sqlmeta: fromDatabase = True - def throttle(): + def throttle(self, model, callbackMethod, strength): """Apply extra rate limiting to destination: 0 = Normal, 10 = Maximum""" - pass + actualArgs = dict() + actualArgs["strength"] = strength + methodId = model.registerCallback(callbackMethod) + self.managedBroker.method(methodId, self.idOriginal, self.schemaName, "throttle", args=actualArgs, packageName="qpid") + def stop(self, model, callbackMethod): + actualArgs = dict() + methodId = model.registerCallback(callbackMethod) + self.managedBroker.method(methodId, self.idOriginal, self.schemaName, "stop", args=actualArgs, packageName="qpid") + def start(self, model, callbackMethod): + actualArgs = dict() + methodId = model.registerCallback(callbackMethod) + self.managedBroker.method(methodId, self.idOriginal, self.schemaName, "start", args=actualArgs, packageName="qpid") + _SO_class_Mgmt_destination_stats = None + _SO_class_Mgmt_session = None - def stop(): - pass - - def start(): - pass - - -class MgmtDestinationStats(SQLObject): +class MgmtProducerStats(SQLObject): class sqlmeta: fromDatabase = True +MgmtSession.sqlmeta.addJoin(MultipleJoin('MgmtDestination', joinMethodName='allDestinations')) + class MgmtProducer(SQLObject): + schemaId = 10 + schemaName = "producer" + managedBroker = None class sqlmeta: fromDatabase = True + _SO_class_Mgmt_producer_stats = None + _SO_class_Mgmt_destination = None + _SO_class_Mgmt_exchange = None -class MgmtProducerStats(SQLObject): +class MgmtConsumerStats(SQLObject): class sqlmeta: fromDatabase = True +MgmtDestination.sqlmeta.addJoin(MultipleJoin('MgmtProducer', joinMethodName='allProducers')) +MgmtExchange.sqlmeta.addJoin(MultipleJoin('MgmtProducer', joinMethodName='allProducers')) + class MgmtConsumer(SQLObject): + schemaId = 11 + schemaName = "consumer" + managedBroker = None class sqlmeta: fromDatabase = True - def close(): - pass - - -class MgmtConsumerStats(SQLObject): - class sqlmeta: - fromDatabase = True - + def close(self, model, callbackMethod): + actualArgs = dict() + methodId = model.registerCallback(callbackMethod) + self.managedBroker.method(methodId, self.idOriginal, self.schemaName, "close", args=actualArgs, packageName="qpid") + _SO_class_Mgmt_consumer_stats = None + _SO_class_Mgmt_destination = None + _SO_class_Mgmt_queue = None Added: mgmt/mint/python/mint/schema.sql =================================================================== --- mgmt/mint/python/mint/schema.sql (rev 0) +++ mgmt/mint/python/mint/schema.sql 2007-11-13 19:45:55 UTC (rev 1303) @@ -0,0 +1,332 @@ + +CREATE TABLE mgmt_system ( + id BIGSERIAL PRIMARY KEY, + id_original BIGINT, + mgmt_system_stats_id BIGINT, + rec_time TIMESTAMP, + creation_time TIMESTAMP, + deletion_time TIMESTAMP, + sys_ident VARCHAR(1000) +); + +CREATE INDEX mgmt_system_sys_ident_index ON mgmt_system(sys_ident); + +CREATE TABLE mgmt_system_stats ( + id BIGSERIAL PRIMARY KEY, + id_original BIGINT, + mgmt_system_id BIGINT REFERENCES mgmt_system , + rec_time TIMESTAMP +); + +ALTER TABLE mgmt_system ADD FOREIGN KEY (mgmt_system_stats_id) REFERENCES mgmt_system_stats; + +CREATE TABLE mgmt_broker ( + id BIGSERIAL PRIMARY KEY, + id_original BIGINT, + mgmt_broker_stats_id BIGINT, + rec_time TIMESTAMP, + creation_time TIMESTAMP, + deletion_time TIMESTAMP, + mgmt_system_id BIGINT REFERENCES mgmt_system, + port INT2 , + worker_threads INT2 , + max_conns INT2 , + conn_backlog INT2 , + staging_threshold INT4 , + store_lib VARCHAR(1000) , + async_store BOOLEAN , + mgmt_pub_interval INT2 , + initial_disk_page_size INT4 , + initial_pages_per_queue INT4 , + cluster_name VARCHAR(1000) , + version VARCHAR(1000) +); + +CREATE INDEX mgmt_broker_port_index ON mgmt_broker(port); + +CREATE TABLE mgmt_broker_stats ( + id BIGSERIAL PRIMARY KEY, + id_original BIGINT, + mgmt_broker_id BIGINT REFERENCES mgmt_broker , + rec_time TIMESTAMP +); + +ALTER TABLE mgmt_broker ADD FOREIGN KEY (mgmt_broker_stats_id) REFERENCES mgmt_broker_stats; + +CREATE TABLE mgmt_vhost ( + id BIGSERIAL PRIMARY KEY, + id_original BIGINT, + mgmt_vhost_stats_id BIGINT, + rec_time TIMESTAMP, + creation_time TIMESTAMP, + deletion_time TIMESTAMP, + mgmt_broker_id BIGINT REFERENCES mgmt_broker, + name VARCHAR(1000) +); + +CREATE INDEX mgmt_vhost_name_index ON mgmt_vhost(name); + +CREATE TABLE mgmt_vhost_stats ( + id BIGSERIAL PRIMARY KEY, + id_original BIGINT, + mgmt_vhost_id BIGINT REFERENCES mgmt_vhost , + rec_time TIMESTAMP +); + +ALTER TABLE mgmt_vhost ADD FOREIGN KEY (mgmt_vhost_stats_id) REFERENCES mgmt_vhost_stats; + +CREATE TABLE mgmt_queue ( + id BIGSERIAL PRIMARY KEY, + id_original BIGINT, + mgmt_queue_stats_id BIGINT, + rec_time TIMESTAMP, + creation_time TIMESTAMP, + deletion_time TIMESTAMP, + mgmt_vhost_id BIGINT REFERENCES mgmt_vhost, + name VARCHAR(1000) , + durable BOOLEAN , + auto_delete BOOLEAN , + exclusive BOOLEAN , + page_memory_limit INT4 +); + +CREATE INDEX mgmt_queue_name_index ON mgmt_queue(name); + +CREATE TABLE mgmt_queue_stats ( + id BIGSERIAL PRIMARY KEY, + id_original BIGINT, + mgmt_queue_id BIGINT REFERENCES mgmt_queue , + rec_time TIMESTAMP, + disk_page_size INT4 , + disk_pages INT4 , + disk_available_size INT4 , + msg_total_enqueues INT8 , + msg_total_dequeues INT8 , + msg_txn_enqueues INT8 , + msg_txn_dequeues INT8 , + msg_persist_enqueues INT8 , + msg_persist_dequeues INT8 , + msg_depth INT4 , + msg_depth_high INT4 , + msg_depth_low INT4 , + byte_total_enqueues INT8 , + byte_total_dequeues INT8 , + byte_txn_enqueues INT8 , + byte_txn_dequeues INT8 , + byte_persist_enqueues INT8 , + byte_persist_dequeues INT8 , + byte_depth INT4 , + byte_depth_high INT4 , + byte_depth_low INT4 , + enqueue_txn_starts INT8 , + enqueue_txn_commits INT8 , + enqueue_txn_rejects INT8 , + enqueue_txn_count INT4 , + enqueue_txn_count_high INT4 , + enqueue_txn_count_low INT4 , + dequeue_txn_starts INT8 , + dequeue_txn_commits INT8 , + dequeue_txn_rejects INT8 , + dequeue_txn_count INT4 , + dequeue_txn_count_high INT4 , + dequeue_txn_count_low INT4 , + consumers INT4 , + consumers_high INT4 , + consumers_low INT4 , + bindings INT4 , + bindings_high INT4 , + bindings_low INT4 , + unacked_messages INT4 , + unacked_messages_high INT4 , + unacked_messages_low INT4 +); + +ALTER TABLE mgmt_queue ADD FOREIGN KEY (mgmt_queue_stats_id) REFERENCES mgmt_queue_stats; + +CREATE TABLE mgmt_exchange ( + id BIGSERIAL PRIMARY KEY, + id_original BIGINT, + mgmt_exchange_stats_id BIGINT, + rec_time TIMESTAMP, + creation_time TIMESTAMP, + deletion_time TIMESTAMP, + mgmt_vhost_id BIGINT REFERENCES mgmt_vhost, + name VARCHAR(1000) , + type VARCHAR(1000) +); + +CREATE INDEX mgmt_exchange_name_index ON mgmt_exchange(name); + +CREATE TABLE mgmt_exchange_stats ( + id BIGSERIAL PRIMARY KEY, + id_original BIGINT, + mgmt_exchange_id BIGINT REFERENCES mgmt_exchange , + rec_time TIMESTAMP, + producers INT4 , + producers_high INT4 , + producers_low INT4 , + bindings INT4 , + bindings_high INT4 , + bindings_low INT4 , + msg_receives INT8 , + msg_drops INT8 , + msg_routes INT8 , + byte_receives INT8 , + byte_drops INT8 , + byte_routes INT8 +); + +ALTER TABLE mgmt_exchange ADD FOREIGN KEY (mgmt_exchange_stats_id) REFERENCES mgmt_exchange_stats; + +CREATE TABLE mgmt_binding ( + id BIGSERIAL PRIMARY KEY, + id_original BIGINT, + mgmt_binding_stats_id BIGINT, + rec_time TIMESTAMP, + creation_time TIMESTAMP, + deletion_time TIMESTAMP, + mgmt_queue_id BIGINT REFERENCES mgmt_queue, + mgmt_exchange_id BIGINT REFERENCES mgmt_exchange, + binding_key VARCHAR(1000) +); + +CREATE TABLE mgmt_binding_stats ( + id BIGSERIAL PRIMARY KEY, + id_original BIGINT, + mgmt_binding_id BIGINT REFERENCES mgmt_binding , + rec_time TIMESTAMP, + msg_matched INT8 +); + +ALTER TABLE mgmt_binding ADD FOREIGN KEY (mgmt_binding_stats_id) REFERENCES mgmt_binding_stats; + +CREATE TABLE mgmt_client ( + id BIGSERIAL PRIMARY KEY, + id_original BIGINT, + mgmt_client_stats_id BIGINT, + rec_time TIMESTAMP, + creation_time TIMESTAMP, + deletion_time TIMESTAMP, + mgmt_vhost_id BIGINT REFERENCES mgmt_vhost, + ip_addr INT4 , + port INT2 +); + +CREATE INDEX mgmt_client_ip_addr_index ON mgmt_client(ip_addr); + +CREATE INDEX mgmt_client_port_index ON mgmt_client(port); + +CREATE TABLE mgmt_client_stats ( + id BIGSERIAL PRIMARY KEY, + id_original BIGINT, + mgmt_client_id BIGINT REFERENCES mgmt_client , + rec_time TIMESTAMP, + auth_identity VARCHAR(1000) , + msgs_produced INT8 , + msgs_consumed INT8 , + bytes_produced INT8 , + bytes_consumed INT8 +); + +ALTER TABLE mgmt_client ADD FOREIGN KEY (mgmt_client_stats_id) REFERENCES mgmt_client_stats; + +CREATE TABLE mgmt_session ( + id BIGSERIAL PRIMARY KEY, + id_original BIGINT, + mgmt_session_stats_id BIGINT, + rec_time TIMESTAMP, + creation_time TIMESTAMP, + deletion_time TIMESTAMP, + mgmt_vhost_id BIGINT REFERENCES mgmt_vhost, + name VARCHAR(1000) , + mgmt_client_id BIGINT REFERENCES mgmt_client, + detached_lifespan INT4 +); + +CREATE INDEX mgmt_session_name_index ON mgmt_session(name); + +CREATE TABLE mgmt_session_stats ( + id BIGSERIAL PRIMARY KEY, + id_original BIGINT, + mgmt_session_id BIGINT REFERENCES mgmt_session , + rec_time TIMESTAMP, + attached BOOLEAN , + remaining_lifespan INT4 , + frames_outstanding INT4 +); + +ALTER TABLE mgmt_session ADD FOREIGN KEY (mgmt_session_stats_id) REFERENCES mgmt_session_stats; + +CREATE TABLE mgmt_destination ( + id BIGSERIAL PRIMARY KEY, + id_original BIGINT, + mgmt_destination_stats_id BIGINT, + rec_time TIMESTAMP, + creation_time TIMESTAMP, + deletion_time TIMESTAMP, + mgmt_session_id BIGINT REFERENCES mgmt_session, + name VARCHAR(1000) +); + +CREATE INDEX mgmt_destination_name_index ON mgmt_destination(name); + +CREATE TABLE mgmt_destination_stats ( + id BIGSERIAL PRIMARY KEY, + id_original BIGINT, + mgmt_destination_id BIGINT REFERENCES mgmt_destination , + rec_time TIMESTAMP, + flow_mode INT2 , + max_msg_credits INT4 , + max_byte_credits INT4 , + msg_credits INT4 , + byte_credits INT4 +); + +ALTER TABLE mgmt_destination ADD FOREIGN KEY (mgmt_destination_stats_id) REFERENCES mgmt_destination_stats; + +CREATE TABLE mgmt_producer ( + id BIGSERIAL PRIMARY KEY, + id_original BIGINT, + mgmt_producer_stats_id BIGINT, + rec_time TIMESTAMP, + creation_time TIMESTAMP, + deletion_time TIMESTAMP, + mgmt_destination_id BIGINT REFERENCES mgmt_destination, + mgmt_exchange_id BIGINT REFERENCES mgmt_exchange +); + +CREATE TABLE mgmt_producer_stats ( + id BIGSERIAL PRIMARY KEY, + id_original BIGINT, + mgmt_producer_id BIGINT REFERENCES mgmt_producer , + rec_time TIMESTAMP, + msgs_produced INT8 , + bytes_produced INT8 +); + +ALTER TABLE mgmt_producer ADD FOREIGN KEY (mgmt_producer_stats_id) REFERENCES mgmt_producer_stats; + +CREATE TABLE mgmt_consumer ( + id BIGSERIAL PRIMARY KEY, + id_original BIGINT, + mgmt_consumer_stats_id BIGINT, + rec_time TIMESTAMP, + creation_time TIMESTAMP, + deletion_time TIMESTAMP, + mgmt_destination_id BIGINT REFERENCES mgmt_destination, + mgmt_queue_id BIGINT REFERENCES mgmt_queue +); + +CREATE TABLE mgmt_consumer_stats ( + id BIGSERIAL PRIMARY KEY, + id_original BIGINT, + mgmt_consumer_id BIGINT REFERENCES mgmt_consumer , + rec_time TIMESTAMP, + msgs_consumed INT8 , + bytes_consumed INT8 , + unacked_messages INT4 , + unacked_messages_high INT4 , + unacked_messages_low INT4 +); + +ALTER TABLE mgmt_consumer ADD FOREIGN KEY (mgmt_consumer_stats_id) REFERENCES mgmt_consumer_stats; Deleted: mgmt/mint/python/mint/updater.py =================================================================== --- mgmt/mint/python/mint/updater.py 2007-11-13 18:36:01 UTC (rev 1302) +++ mgmt/mint/python/mint/updater.py 2007-11-13 19:45:55 UTC (rev 1303) @@ -1,57 +0,0 @@ -from time import sleep -from datetime import datetime - -from schema import * - -class MintUpdater(object): - def __init__(self, model, broker): - self.model = model - self.broker = broker - - self.broker.configListener("XXXcontext", configCallback) - self.broker.instrumentationListener("XXXcontext", instCallback) - - def start(self): - self.broker.start() - - while True: - sleep(1) - -def getQueueByName(name, create=False): - try: - queues = MgmtQueue.selectBy(name=name)[:1] - queue = queues[0] - except IndexError: - if (create): queue = MgmtQueue() - return queue - -def configCallback(broker, oid, list, timestamps): - print "broker=" + broker - if oid == 4: - print list - d = dict(list) - queue = getQueueByName(d["name"], True) - queue.set(**d) - recOn = datetime.fromtimestamp(timestamps[0]/1000000000) - createdOn = datetime.fromtimestamp(timestamps[1]/1000000000) - queue.set(recTime=recOn,creationTime=createdOn) - print queue.id - print " -> " + d["name"] - return queue - -def instCallback(broker, oid, list, timestamps): - print "broker=" + broker - if oid == 4: - print list - d = dict(list) - queue = getQueueByName(d.pop("name")) - d["mgmtQueue"] = queue.id - recOn = datetime.fromtimestamp(timestamps[0]/1000000000) - queueStats = MgmtQueueStats() - queueStats.set(recTime=recOn) - queueStats.set(**d) - if (timestamps[2] != 0): - deletedOn = datetime.fromtimestamp(timestamps[2]/1000000000) - queue.set(deletionTime=deletedOn) - print queue.id - return queueStats