[rhmessaging-commits] rhmessaging commits: r2805 - in mgmt/trunk: cumin/python/wooly and 3 other directories.

Author: justi9 Date: 2008-11-14 15:12:00 -0500 (Fri, 14 Nov 2008) New Revision: 2805 Modified: mgmt/trunk/cumin/python/cumin/__init__.py mgmt/trunk/cumin/python/cumin/broker.py mgmt/trunk/cumin/python/cumin/model.py mgmt/trunk/cumin/python/cumin/page.py mgmt/trunk/cumin/python/cumin/page.strings mgmt/trunk/cumin/python/cumin/tools.py mgmt/trunk/cumin/python/wooly/forms.py 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/update.py mgmt/trunk/mint/sql/schema.sql mgmt/trunk/parsley/python/parsley/command.py Log: Nuno's adaptation of mint to the qmfconsole api, which replaces the older "management.py" api. Modified: mgmt/trunk/cumin/python/cumin/__init__.py =================================================================== --- mgmt/trunk/cumin/python/cumin/__init__.py 2008-11-14 19:30:06 UTC (rev 2804) +++ mgmt/trunk/cumin/python/cumin/__init__.py 2008-11-14 20:12:00 UTC (rev 2805) @@ -18,6 +18,7 @@ from action import ActionPage from user import LoginPage, UserSession from datetime import timedelta +from qpid.util import URL from wooly import Session @@ -121,8 +122,7 @@ def do_run(self): while True: for reg in BrokerRegistration.select(): - if reg.broker is None or reg.broker.managedBroker not in \ - self.model.data.connections: + if reg.broker is None or reg.getBrokerId() not in self.model.data.managedBrokers: attempts = self.attempts.get(reg, 0) attempts += 1 self.attempts[reg] = attempts @@ -133,7 +133,6 @@ reg.connect(self.model.data) elif attempts % 100 == 0: reg.connect(self.model.data) - self.event.wait(10) class CuminServer(WebServer): Modified: mgmt/trunk/cumin/python/cumin/broker.py =================================================================== --- mgmt/trunk/cumin/python/cumin/broker.py 2008-11-14 19:30:06 UTC (rev 2804) +++ mgmt/trunk/cumin/python/cumin/broker.py 2008-11-14 20:12:00 UTC (rev 2805) @@ -658,14 +658,9 @@ if addr: name = names[i] - host, port = parse_broker_addr(addr) + url = "amqp://%s:%i" % parse_broker_addr(addr) - args = { - "name": name, - "host": host, - "port": port - } - + args = {"name": name, "url": url} reg = action.invoke(None, args) if len(groups) > i: @@ -701,13 +696,13 @@ errs = aerrs.setdefault(i, list()) errs.append("The address field is empty; it is required") else: - host, port = parse_broker_addr(addr) + #host, port = parse_broker_addr(addr) count = BrokerRegistration.selectBy \ - (host=host, port=port).count() + (url=addr).count() if count: errs = aerrs.setdefault(i, list()) - errs.append("The broker at %s:%i " % (host, port) + + errs.append("The broker at %s " % (url) + "is already registered") return not len(nerrs) and not len(aerrs) Modified: mgmt/trunk/cumin/python/cumin/model.py =================================================================== --- mgmt/trunk/cumin/python/cumin/model.py 2008-11-14 19:30:06 UTC (rev 2804) +++ mgmt/trunk/cumin/python/cumin/model.py 2008-11-14 20:12:00 UTC (rev 2805) @@ -1668,14 +1668,10 @@ super(CuminBrokerRegistration, self).__init__ \ (model, "broker_registration", BrokerRegistration) - prop = CuminProperty(self, "host") - prop.title = "Host" + prop = CuminProperty(self, "url") + prop.title = "URL" prop.summary = True - prop = CuminProperty(self, "port") - prop.title = "Port" - prop.summary = True - action = self.Add(self, "add") action.title = "Add" action.navigable = False Modified: mgmt/trunk/cumin/python/cumin/page.py =================================================================== --- mgmt/trunk/cumin/python/cumin/page.py 2008-11-14 19:30:06 UTC (rev 2804) +++ mgmt/trunk/cumin/python/cumin/page.py 2008-11-14 20:12:00 UTC (rev 2805) @@ -209,6 +209,9 @@ heading = self.Heading(app, "heading") self.add_child(heading) + self.add_tab(self.OverviewTab(app, "over")) + self.add_tab(self.AccountTab(app, "acct")) + def render_change_password_href(self, session): branch = session.branch() self.frame.change_password.show(branch) @@ -218,6 +221,22 @@ def render_title(self, session): return "MRG Management" + class OverviewTab(Widget): + def render_title(self, session): + return "Overview" + + def render_content(self, session): + pass + + class AccountTab(ActionSet): + def render_title(self, session): + return "Your Account" + + def render_change_password_href(self, session): + branch = session.branch() + self.frame.change_password.show(branch) + return branch.marshal() + class MessagingView(TabbedModeSet): def __init__(self, app, name): super(MessagingView, self).__init__(app, name) Modified: mgmt/trunk/cumin/python/cumin/page.strings =================================================================== --- mgmt/trunk/cumin/python/cumin/page.strings 2008-11-14 19:30:06 UTC (rev 2804) +++ mgmt/trunk/cumin/python/cumin/page.strings 2008-11-14 20:12:00 UTC (rev 2805) @@ -180,11 +180,15 @@ <div class="oblock"> {heading} - <ul class="actions"> - <a class="nav" href="{change_password_href}">Change Password</a> - </ul> + <ul class="TabbedModeSet tabs">{tabs}</ul> + <div class="TabbedModeSet mode">{mode}</div> </div> +[AccountTab.html] +<ul class="actions"> + <a class="nav" href="{change_password_href}">Change Password</a> +</ul> + [MessagingView.html] <script type="text/javascript"> <![CDATA[ Modified: mgmt/trunk/cumin/python/cumin/tools.py =================================================================== --- mgmt/trunk/cumin/python/cumin/tools.py 2008-11-14 19:30:06 UTC (rev 2804) +++ mgmt/trunk/cumin/python/cumin/tools.py 2008-11-14 20:12:00 UTC (rev 2805) @@ -31,9 +31,6 @@ self.config = CuminConfig() - opt = CommandOption(self, "help", "h") - opt.description = "Print this message" - opt = CommandOption(self, "data") opt.argument = "URI" opt.description = "Connect to database at URI" @@ -84,6 +81,20 @@ opt = CommandOption(command, "force") opt.description = "Don't complain and just do it" + command = self.AddBroker(self, "add-broker") + command.arguments = ("NAME", "URL") + command.description = "Add a new broker called NAME at URL" + + command = self.RemoveBroker(self, "remove-broker") + command.arguments = ("URL",) + command.description = "Remove broker called NAME; requires --force" + + opt = CommandOption(command, "force") + opt.description = "Don't complain and just do it" + + command = self.ListBrokers(self, "list-brokers") + command.description = "List existing broker registrations" + command = self.AddUser(self, "add-user") command.arguments = ("NAME",) command.description = "Add a new user called NAME" @@ -179,6 +190,46 @@ def run(self, opts, args): self.parent.database.checkSchema() + class AddBroker(Command): + def run(self, opts, args): + try: + name, url = args[1:] + except IndexError: + raise CommandException(self, "NAME and URL are required") + + for reg in BrokerRegistration.selectBy(name=name): + print "Error: a broker called '%s' already exists" % name + sys.exit(1) + + for reg in BrokerRegistration.selectBy(url=url): + print "Error: a broker at %s already exists" % url + sys.exit(1) + + reg = BrokerRegistration(name=name, url=url) + reg.syncUpdate() + + class RemoveBroker(Command): + def run(self, opts, args): + try: + name = args[1] + except IndexError: + raise CommandException(self, "NAME is required") + + for reg in BrokerRegistration.selectBy(name=name): + break + + if reg: + reg.destroySelf() + reg.syncUpdate() + else: + raise CommandException \ + (self, "Broker '%s' is unknown", reg.name) + + class ListBrokers(Command): + def run(self, opts, args): + for reg in BrokerRegistration.select(): + print reg.name, reg.url + class AddUser(Command): def run(self, opts, args): try: Modified: mgmt/trunk/cumin/python/wooly/forms.py =================================================================== --- mgmt/trunk/cumin/python/wooly/forms.py 2008-11-14 19:30:06 UTC (rev 2804) +++ mgmt/trunk/cumin/python/wooly/forms.py 2008-11-14 20:12:00 UTC (rev 2805) @@ -184,6 +184,9 @@ class PasswordInput(StringInput): pass +# XXX Why does this have a boolean param? Shouldn't the name suggest +# that somehow? Would some folks want a hidden input with a different +# param? I think this needs to take a param class HiddenInput(ScalarInput): def __init__(self, app, name): super(HiddenInput, self).__init__(app, name, None) Modified: mgmt/trunk/mint/python/mint/__init__.py =================================================================== --- mgmt/trunk/mint/python/mint/__init__.py 2008-11-14 19:30:06 UTC (rev 2804) +++ mgmt/trunk/mint/python/mint/__init__.py 2008-11-14 20:12:00 UTC (rev 2805) @@ -1,14 +1,12 @@ -import sys, os, socket, qpid, logging -from qpid.datatypes import uuid4 -from qpid.connection import Connection as QpidConnection -from qpid.util import connect -from qpid.management import managementChannel, managementClient -from datetime import * -from sqlobject import * +import logging +import qpid.qmfconsole +import struct from threading import Lock, RLock -from traceback import print_exc, print_stack - -from mint import schema +from sqlobject import * +from traceback import print_exc +from qpid.util import URL +from qpid.datatypes import UUID +from mint.schema import * from mint import update log = logging.getLogger("mint") @@ -28,95 +26,118 @@ cascade="null", default=None, name="registration")) -class MintInfo(SQLObject): - class sqlmeta: - lazyUpdate = True +class MintDatabase(object): + def __init__(self, uri): + self.uri = uri - version = StringCol(length=1000, default="0.1", notNone=True) + def getConnection(self): + return connectionForURI(self.uri).getConnection() -class BrokerRegistration(SQLObject): - class sqlmeta: - lazyUpdate = True + def check(self): + self.checkConnection() - name = StringCol(length=1000, default=None, unique=True, notNone=True) - host = StringCol(length=1000, default=None, notNone=True) - port = IntCol(default=None, notNone=True) - broker = ForeignKey("Broker", cascade="null", default=None) - groups = SQLRelatedJoin("BrokerGroup", - intermediateTable="broker_group_mapping", - createRelatedTable=False) - cluster = ForeignKey("BrokerCluster", cascade="null", default=None) - profile = ForeignKey("BrokerProfile", cascade="null", default=None) + def init(self): + sqlhub.processConnection = connectionForURI(self.uri) - host_port_unique = index.DatabaseIndex(host, port, unique=True) + def checkConnection(self): + conn = self.getConnection() - def connect(self, model): - log.info("Connecting to broker '%s' at %s:%i" % \ - (self.name, self.host, self.port or 5672)) + try: + cursor = conn.cursor() + cursor.execute("select now()") + finally: + conn.close() - conn = BrokerConnection(model, self.host, self.port or 5672) + def checkSchema(self): + pass + def dropSchema(self): + conn = self.getConnection() try: - conn.open() - log.info("Connection succeeded") - except: - log.info("Connection failed: " + str(conn.exception)) + cursor = conn.cursor() + + cursor.execute("drop schema public cascade") - def getDefaultVhost(self): - if self.broker: + conn.commit() + finally: + conn.close() + + def createSchema(self, file_paths): + conn = self.getConnection() + + scripts = list() + + for path in file_paths: + file = open(path, "r") try: - return Vhost.selectBy(broker=self.broker, name="/")[0] - except IndexError: + scripts.append((path, file.read())) + finally: + file.close() + + try: + cursor = conn.cursor() + + try: + cursor.execute("create schema public"); + except: + conn.rollback() pass -class BrokerGroup(SQLObject): - class sqlmeta: - lazyUpdate = True + for path, text in scripts: + stmts = text.split(";") + count = 0 - name = StringCol(length=1000, default=None, unique=True, notNone=True) - brokers = SQLRelatedJoin("BrokerRegistration", - intermediateTable="broker_group_mapping", - createRelatedTable=False) + for stmt in stmts: + stmt = stmt.strip() -class BrokerGroupMapping(SQLObject): - class sqlmeta: - lazyUpdate = True + if stmt: + try: + cursor.execute(stmt) + except Exception, e: + print "Failed executing statement:" + print stmt - brokerRegistration = ForeignKey("BrokerRegistration", notNull=True, - cascade=True) - brokerGroup = ForeignKey("BrokerGroup", notNull=True, cascade=True) - unique = index.DatabaseIndex(brokerRegistration, brokerGroup, unique=True) + raise e -class BrokerCluster(SQLObject): - class sqlmeta: - lazyUpdate = True + count += 1 - name = StringCol(length=1000, default=None) - brokers = SQLMultipleJoin("BrokerRegistration", joinColumn="cluster_id") + print "Executed %i statements from file '%s'" % (count, path) -class BrokerProfile(SQLObject): - class sqlmeta: - lazyUpdate = True + conn.commit() - name = StringCol(length=1000, default=None) - brokers = SQLMultipleJoin("BrokerRegistration", joinColumn="profile_id") - properties = SQLMultipleJoin("ConfigProperty", joinColumn="profile_id") + info = MintInfo(version="0.1") + info.sync() -class ConfigProperty(SQLObject): - class sqlmeta: - lazyUpdate = True + # Standard roles - name = StringCol(length=1000, default=None) - value = StringCol(length=1000, default=None) - type = StringCol(length=1, default="s") + user = Role(name="user") + user.sync() -class CollectorRegistration(SQLObject): - class sqlmeta: - lazyUpdate = True + admin = Role(name="admin") + admin.sync() + finally: + conn.close() - name = StringCol(length=1000, default=None) - collectorId = StringCol(length=1000, default=None) + def checkSchema(self): + conn = self.getConnection() + try: + cursor = conn.cursor() + + try: + cursor.execute("select version from mint_info"); + except Exception, e: + print "No schema present" + return + + for rec in cursor: + print "OK (version %s)" % rec[0] + return; + + print "No schema present" + finally: + conn.close() + class Subject(SQLObject): class sqlmeta: lazyUpdate = True @@ -165,139 +186,101 @@ class ObjectNotFound(Exception): pass -# Not thread safe -class BrokerConnection(object): - def __init__(self, model, host, port): - self.model = model - self.host = host - self.port = port - self.id = "%s:%i" % (host, port) - self.objectsById = dict() +class MintInfo(SQLObject): + class sqlmeta: + lazyUpdate = True - # Set upon receiving a broker info control - self.sessionId = None - self.brokerId = None + version = StringCol(length=1000, default="0.1", notNone=True) - # state in (None, "opening", "opened", "closing", "closed") - self.state = None - self.exception = None +class CollectorRegistration(SQLObject): + class sqlmeta: + lazyUpdate = True - self.mconn = None - self.mclient = None - self.mchan = None + name = StringCol(length=1000, default=None) + collectorId = StringCol(length=1000, default=None) - def getObject(self, cls, id): - compositeId = "%s:%s" % (id.first, id.second) +class BrokerRegistration(SQLObject): + class sqlmeta: + lazyUpdate = True - try: - obj = self.objectsById[compositeId] - except KeyError: - try: - obj = cls.selectBy(sourceScopeId=id.first, sourceObjectId=id.second, managedBroker=self.id)[0] - self.objectsById[compositeId] = obj - except IndexError: - raise ObjectNotFound() + name = StringCol(length=1000, default=None, unique=True, notNone=True) + url = StringCol(length=1000, default=None) + qmfBroker = None + broker = ForeignKey("Broker", cascade="null", default=None) + groups = SQLRelatedJoin("BrokerGroup", + intermediateTable="broker_group_mapping", + createRelatedTable=False) + cluster = ForeignKey("BrokerCluster", cascade="null", default=None) + profile = ForeignKey("BrokerProfile", cascade="null", default=None) - return obj + url_unique = index.DatabaseIndex(url, unique=True) - def isOpen(self): - return self.state == "opened" - - def open(self): - assert self.mconn is None - assert self.mclient is None - assert self.mchan is None - - self.state = "opening" - + def connect(self, model): + log.info("Connecting to broker '%s' at %s" % (self.name, self.url)) try: - spec = qpid.spec.load(self.model.specPath) - sock = connect(self.host, self.port) + self.qmfBroker = model.getSession().addBroker(self.url) + log.info("Connection succeeded") except Exception, e: - self.exception = e - raise e + log.info("Connection failed: %s ", e.message) + print_exc() - self.mconn = QpidConnection(sock, spec) - self.mclient = managementClient(spec, - self.model.controlCallback, - self.model.propsCallback, - self.model.statsCallback, - self.model.methodCallback, - self.model.closeCallback) - self.mclient.schemaListener(self.model.schemaCallback) + def getBrokerId(self): + if self.qmfBroker is not None: + return str(self.qmfBroker.getBrokerId()) + else: + return None - try: - self.mconn.start() - self.mchan = self.mclient.addChannel \ - (self.mconn.session(str(uuid4())), self) + def getDefaultVhost(self): + if self.broker: + try: + return Vhost.selectBy(broker=self.broker, name="/")[0] + except IndexError: + return None - self.state = "opened" - except Exception, e: - self.exception = e - raise e +class BrokerGroup(SQLObject): + class sqlmeta: + lazyUpdate = True - self.model.lock() - try: - self.model.connections[self.id] = self - finally: - self.model.unlock() + name = StringCol(length=1000, default=None, unique=True, notNone=True) + brokers = SQLRelatedJoin("BrokerRegistration", + intermediateTable="broker_group_mapping", + createRelatedTable=False) - def getSessionId(self): - if not self.isOpen(): - raise Exception("Connection not open") +class BrokerGroupMapping(SQLObject): + class sqlmeta: + lazyUpdate = True - return self.mchan.sessionId + brokerRegistration = ForeignKey("BrokerRegistration", notNull=True, + cascade=True) + brokerGroup = ForeignKey("BrokerGroup", notNull=True, cascade=True) + unique = index.DatabaseIndex(brokerRegistration, brokerGroup, unique=True) - def callMethod(self, objId, className, methodName, callback, args): - if not self.isOpen(): - raise Exception("Connection not open") +class BrokerCluster(SQLObject): + class sqlmeta: + lazyUpdate = True - self.model.lock() - try: - self.model.currentMethodId += 1 - seq = self.model.currentMethodId - self.model.outstandingMethodCalls[seq] = callback - finally: - self.model.unlock() + name = StringCol(length=1000, default=None) + brokers = SQLMultipleJoin("BrokerRegistration", joinColumn="cluster_id") - self.mclient.callMethod(self.mchan, seq, objId, - className, methodName, args) +class BrokerProfile(SQLObject): + class sqlmeta: + lazyUpdate = True - def close(self): - self.state = "closing" + name = StringCol(length=1000, default=None) + brokers = SQLMultipleJoin("BrokerRegistration", joinColumn="profile_id") + properties = SQLMultipleJoin("ConfigProperty", joinColumn="profile_id") - self.model.lock() - try: - del self.model.connections[self.id] - finally: - self.model.unlock() - - try: - self.mclient.removeChannel(self.mchan) - - self.state = "closed" - except Exception, e: - self.exception = e - raise e - - self.mconn.close() - # XXX What else do I need to try to shutdown here? - - self.mconn = None - self.mclient = None - self.mchan = None - -class MintModel: + +class MintModel(qpid.qmfconsole.Console): staticInstance = None - def __init__(self, dataUri, specPath, debug=False): + def __init__(self, dataUri, specPath="", debug=False): self.dataUri = dataUri - self.specPath = specPath self.debug = debug - self.currentMethodId = 1 - self.outstandingMethodCalls = dict() - self.connections = dict() + assert MintModel.staticInstance is None + MintModel.staticInstance = self + self.connCloseListener = None self.__lock = RLock() @@ -306,15 +289,14 @@ self.orphanObjectMap = dict() self.updateThread = update.ModelUpdateThread(self) + self.mgmtSession = qpid.qmfconsole.Session(self) + self.outstandingMethodCalls = dict() + self.managedBrokers = dict() - assert MintModel.staticInstance is None - MintModel.staticInstance = self - if self.debug: log.setLevel(logging.DEBUG) def lock(self): - #print_stack() self.__lock.acquire() def unlock(self): @@ -326,12 +308,10 @@ except Exception, e: if hasattr(e, "message") and e.message.find("does not exist"): print "Database not found; run cumin-database-init" - raise e def init(self): - conn = connectionForURI(self.dataUri) - sqlhub.processConnection = conn + sqlhub.processConnection = connectionForURI(self.dataUri) def start(self): self.updateThread.start() @@ -342,156 +322,99 @@ def setCloseListener(self, connCloseListener): self.connCloseListener = connCloseListener - def schemaCallback(self, conn, classInfo, props, stats, methods, events): - up = update.SchemaUpdate(conn, classInfo, props, stats, methods, events) - self.updateThread.enqueue(up) + def getObject(self, cls, id, broker): + obj = None + if isinstance(id, qpid.qmfconsole.ObjectId): + compositeId = "%s:%s" % (id.first, id.second) + try: + obj = cls.selectBy(sourceScopeId=id.first, sourceObjectId=id.second)[0] + except IndexError: + raise ObjectNotFound() + elif cls.__name__.endswith("Pool"): + try: + obj = cls.selectBy(sourceId=id)[0] + except IndexError: + raise ObjectNotFound() + + return obj - def propsCallback(self, conn, classInfo, props, timestamps): - up = update.PropertyUpdate(conn, classInfo, props, timestamps) - self.updateThread.enqueue(up) + def getSession(self): + return self.mgmtSession - def statsCallback(self, conn, classInfo, stats, timestamps): - up = update.StatisticUpdate(conn, classInfo, stats, timestamps) - self.updateThread.enqueue(up) - - def methodCallback(self, conn, methodId, errorId, errorText, args): - up = update.MethodUpdate(conn, methodId, errorId, errorText, args) - self.updateThread.enqueue(up) - - def closeCallback(self, conn, data): - up = update.CloseUpdate(conn, data) - self.updateThread.enqueue(up) - - def controlCallback(self, conn, type, data): - up = update.ControlUpdate(conn, type, data) - self.updateThread.enqueue(up) - - def registerCallback(self, callback): + def callMethod(self, managedBroker, objId, classKey, methodName, callback, args): self.lock() try: - self.currentMethodId += 1 - methodId = self.currentMethodId - self.outstandingMethodCalls[methodId] = callback - return methodId + broker = self.managedBrokers[managedBroker] finally: self.unlock() - def getConnectionByRegistration(self, reg): + pname, cname, hash = classKey.split(", ") + seq = self.mgmtSession._sendMethodRequest(broker, (pname, cname, hash), objId, methodName, args) + + if seq is not None: + self.lock() + try: + self.outstandingMethodCalls[seq] = callback + finally: + self.unlock() + return seq + + def brokerConnected(self, broker): + """ Invoked when a connection is established to a broker """ self.lock() try: - return self.connections.get("%s:%i" % (reg.host, reg.port)) + self.managedBrokers[str(broker.getBrokerId())] = broker finally: self.unlock() -class MintDatabase(object): - def __init__(self, uri): - self.uri = uri - - def getConnection(self): - return connectionForURI(self.uri).getConnection() - - def check(self): - self.checkConnection() - - def init(self): - conn = connectionForURI(self.uri) - sqlhub.processConnection = conn - - def checkConnection(self): - conn = self.getConnection() - + def brokerDisconnected(self, broker): + """ Invoked when the connection to a broker is lost """ + self.lock() try: - cursor = conn.cursor() - cursor.execute("select now()") + del self.managedBrokers[str(broker.getBrokerId())] finally: - conn.close() + self.unlock() + if (self.connCloseListener != None): + self.connCloseListener(broker) - def checkSchema(self): + def newPackage(self, name): + """ Invoked when a QMF package is discovered. """ pass - def dropSchema(self): - conn = self.getConnection() - try: - cursor = conn.cursor() - - cursor.execute("drop schema public cascade") + def newClass(self, kind, classKey): + """ Invoked when a new class is discovered. Session.getSchema can be + used to obtain details about the class.""" + pass - conn.commit() - finally: - conn.close() + def newAgent(self, agent): + """ Invoked when a QMF agent is discovered. """ + pass - def createSchema(self, file_paths): - conn = self.getConnection() + def delAgent(self, agent): + """ Invoked when a QMF agent disconects. """ + pass - scripts = list() + def objectProps(self, broker, record): + """ Invoked when an object is updated. """ + self.updateThread.enqueue(update.PropertyUpdate(broker, record)) - for path in file_paths: - file = open(path, "r") - try: - scripts.append((path, file.read())) - finally: - file.close() + def objectStats(self, broker, record): + """ Invoked when an object is updated. """ + self.updateThread.enqueue(update.StatisticUpdate(broker, record)) - try: - cursor = conn.cursor() + def event(self, broker, event): + """ Invoked when an event is raised. """ + pass - try: - cursor.execute("create schema public"); - except: - conn.rollback() - pass + def heartbeat(self, agent, timestamp): + pass - for path, text in scripts: - stmts = text.split(";") - count = 0 - - for stmt in stmts: - stmt = stmt.strip() - - if stmt: - try: - cursor.execute(stmt) - except Exception, e: - print "Failed executing statement:" - print stmt - - raise e - - count += 1 - - print "Executed %i statements from file '%s'" % (count, path) - - conn.commit() - - info = MintInfo(version="0.1") - info.sync() - - # Standard roles - - user = Role(name="user") - user.sync() - - admin = Role(name="admin") - admin.sync() - finally: - conn.close() - - def checkSchema(self): - conn = self.getConnection() - + def brokerInfo(self, broker): + self.lock() try: - cursor = conn.cursor() - - try: - cursor.execute("select version from mint_info"); - except Exception, e: - print "No schema present" - return - - for rec in cursor: - print "OK (version %s)" % rec[0] - return; - - print "No schema present" + self.managedBrokers[str(broker.getBrokerId())] = broker finally: - conn.close() + self.unlock() + + def methodResponse(self, broker, seq, response): + self.updateThread.enqueue(update.MethodUpdate(broker, seq, response)) Modified: mgmt/trunk/mint/python/mint/schema.py =================================================================== --- mgmt/trunk/mint/python/mint/schema.py 2008-11-14 19:30:06 UTC (rev 2804) +++ mgmt/trunk/mint/python/mint/schema.py 2008-11-14 20:12:00 UTC (rev 2805) @@ -1,7 +1,7 @@ import mint from sqlobject import * from datetime import datetime -from qpid.management import objectId +from qpid.qmfconsole import ObjectId class Pool(SQLObject): class sqlmeta: @@ -16,6 +16,7 @@ recTime = TimestampCol(default=None) sourceScopeId = BigIntCol(default=None) sourceObjectId = BigIntCol(default=None) + qmfClassKey = BLOBCol(default=None) creationTime = TimestampCol(default=None) deletionTime = TimestampCol(default=None) managedBroker = StringCol(length=1000, default=None) @@ -135,6 +136,7 @@ recTime = TimestampCol(default=None) sourceScopeId = BigIntCol(default=None) sourceObjectId = BigIntCol(default=None) + qmfClassKey = BLOBCol(default=None) creationTime = TimestampCol(default=None) deletionTime = TimestampCol(default=None) managedBroker = StringCol(length=1000, default=None) @@ -172,61 +174,49 @@ def GetAd(self, model, callback, JobAd): - actualArgs = dict() - actualArgs["JobAd"] = JobAd - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "GetAd", + actualArgs = list() + actualArgs.append(JobAd) + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "GetAd", callback, args=actualArgs) def SetAttribute(self, model, callback, Name, Value): - actualArgs = dict() - actualArgs["Name"] = Name - actualArgs["Value"] = Value - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "SetAttribute", + actualArgs = list() + actualArgs.append(Name) + actualArgs.append(Value) + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "SetAttribute", callback, args=actualArgs) def Hold(self, model, callback, Reason): - actualArgs = dict() - actualArgs["Reason"] = Reason - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "Hold", + actualArgs = list() + actualArgs.append(Reason) + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "Hold", callback, args=actualArgs) def Release(self, model, callback, Reason): - actualArgs = dict() - actualArgs["Reason"] = Reason - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "Release", + actualArgs = list() + actualArgs.append(Reason) + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "Release", callback, args=actualArgs) def Remove(self, model, callback, Reason): - actualArgs = dict() - actualArgs["Reason"] = Reason - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "Remove", + actualArgs = list() + actualArgs.append(Reason) + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "Remove", callback, args=actualArgs) def Fetch(self, model, callback, File, Start, End, Data): - actualArgs = dict() - actualArgs["File"] = File - actualArgs["Start"] = Start - actualArgs["End"] = End - actualArgs["Data"] = Data - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "Fetch", + actualArgs = list() + actualArgs.append(File) + actualArgs.append(Start) + actualArgs.append(End) + actualArgs.append(Data) + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "Fetch", callback, args=actualArgs) class JobStats(SQLObject): @@ -247,6 +237,7 @@ recTime = TimestampCol(default=None) sourceScopeId = BigIntCol(default=None) sourceObjectId = BigIntCol(default=None) + qmfClassKey = BLOBCol(default=None) creationTime = TimestampCol(default=None) deletionTime = TimestampCol(default=None) managedBroker = StringCol(length=1000, default=None) @@ -298,6 +289,7 @@ recTime = TimestampCol(default=None) sourceScopeId = BigIntCol(default=None) sourceObjectId = BigIntCol(default=None) + qmfClassKey = BLOBCol(default=None) creationTime = TimestampCol(default=None) deletionTime = TimestampCol(default=None) managedBroker = StringCol(length=1000, default=None) @@ -332,6 +324,7 @@ recTime = TimestampCol(default=None) sourceScopeId = BigIntCol(default=None) sourceObjectId = BigIntCol(default=None) + qmfClassKey = BLOBCol(default=None) creationTime = TimestampCol(default=None) deletionTime = TimestampCol(default=None) managedBroker = StringCol(length=1000, default=None) @@ -350,30 +343,24 @@ def GetLimits(self, model, callback, Limits): - actualArgs = dict() - actualArgs["Limits"] = Limits - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "GetLimits", + actualArgs = list() + actualArgs.append(Limits) + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "GetLimits", callback, args=actualArgs) def SetLimit(self, model, callback, Name, Max): - actualArgs = dict() - actualArgs["Name"] = Name - actualArgs["Max"] = Max - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "SetLimit", + actualArgs = list() + actualArgs.append(Name) + actualArgs.append(Max) + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "SetLimit", callback, args=actualArgs) def Reconfig(self, model, callback): - actualArgs = dict() - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "Reconfig", + actualArgs = list() + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "Reconfig", callback, args=actualArgs) class NegotiatorStats(SQLObject): @@ -401,6 +388,7 @@ recTime = TimestampCol(default=None) sourceScopeId = BigIntCol(default=None) sourceObjectId = BigIntCol(default=None) + qmfClassKey = BLOBCol(default=None) creationTime = TimestampCol(default=None) deletionTime = TimestampCol(default=None) managedBroker = StringCol(length=1000, default=None) @@ -433,6 +421,7 @@ recTime = TimestampCol(default=None) sourceScopeId = BigIntCol(default=None) sourceObjectId = BigIntCol(default=None) + qmfClassKey = BLOBCol(default=None) creationTime = TimestampCol(default=None) deletionTime = TimestampCol(default=None) managedBroker = StringCol(length=1000, default=None) @@ -453,21 +442,17 @@ def Start(self, model, callback, Subsystem): - actualArgs = dict() - actualArgs["Subsystem"] = Subsystem - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "Start", + actualArgs = list() + actualArgs.append(Subsystem) + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "Start", callback, args=actualArgs) def Stop(self, model, callback, Subsystem): - actualArgs = dict() - actualArgs["Subsystem"] = Subsystem - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "Stop", + actualArgs = list() + actualArgs.append(Subsystem) + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "Stop", callback, args=actualArgs) class MasterStats(SQLObject): @@ -495,6 +480,7 @@ recTime = TimestampCol(default=None) sourceScopeId = BigIntCol(default=None) sourceObjectId = BigIntCol(default=None) + qmfClassKey = BLOBCol(default=None) creationTime = TimestampCol(default=None) deletionTime = TimestampCol(default=None) managedBroker = StringCol(length=1000, default=None) @@ -510,11 +496,9 @@ def reloadACLFile(self, model, callback): """Reload the ACL file""" - actualArgs = dict() - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "reloadACLFile", + actualArgs = list() + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "reloadACLFile", callback, args=actualArgs) class AclStats(SQLObject): @@ -536,6 +520,7 @@ recTime = TimestampCol(default=None) sourceScopeId = BigIntCol(default=None) sourceObjectId = BigIntCol(default=None) + qmfClassKey = BLOBCol(default=None) creationTime = TimestampCol(default=None) deletionTime = TimestampCol(default=None) managedBroker = StringCol(length=1000, default=None) @@ -552,19 +537,15 @@ def stopClusterNode(self, model, callback): - actualArgs = dict() - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "stopClusterNode", + actualArgs = list() + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "stopClusterNode", callback, args=actualArgs) def stopFullCluster(self, model, callback): - actualArgs = dict() - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "stopFullCluster", + actualArgs = list() + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "stopFullCluster", callback, args=actualArgs) class ClusterStats(SQLObject): @@ -585,6 +566,7 @@ recTime = TimestampCol(default=None) sourceScopeId = BigIntCol(default=None) sourceObjectId = BigIntCol(default=None) + qmfClassKey = BLOBCol(default=None) creationTime = TimestampCol(default=None) deletionTime = TimestampCol(default=None) managedBroker = StringCol(length=1000, default=None) @@ -631,6 +613,7 @@ recTime = TimestampCol(default=None) sourceScopeId = BigIntCol(default=None) sourceObjectId = BigIntCol(default=None) + qmfClassKey = BLOBCol(default=None) creationTime = TimestampCol(default=None) deletionTime = TimestampCol(default=None) managedBroker = StringCol(length=1000, default=None) @@ -652,12 +635,10 @@ def expand(self, model, callback, by): """Increase number of files allocated for this journal""" - actualArgs = dict() - actualArgs["by"] = by - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "expand", + actualArgs = list() + actualArgs.append(by) + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "expand", callback, args=actualArgs) class JournalStats(SQLObject): @@ -706,6 +687,7 @@ recTime = TimestampCol(default=None) sourceScopeId = BigIntCol(default=None) sourceObjectId = BigIntCol(default=None) + qmfClassKey = BLOBCol(default=None) creationTime = TimestampCol(default=None) deletionTime = TimestampCol(default=None) managedBroker = StringCol(length=1000, default=None) @@ -738,6 +720,7 @@ recTime = TimestampCol(default=None) sourceScopeId = BigIntCol(default=None) sourceObjectId = BigIntCol(default=None) + qmfClassKey = BLOBCol(default=None) creationTime = TimestampCol(default=None) deletionTime = TimestampCol(default=None) managedBroker = StringCol(length=1000, default=None) @@ -757,41 +740,35 @@ def echo(self, model, callback, sequence, body): """Request a response to test the path to the management broker""" - actualArgs = dict() - actualArgs["sequence"] = sequence - actualArgs["body"] = body - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "echo", + actualArgs = list() + actualArgs.append(sequence) + actualArgs.append(body) + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "echo", callback, args=actualArgs) def connect(self, model, callback, host, port, durable, authMechanism, username, password, transport): """Establish a connection to another broker""" - actualArgs = dict() - actualArgs["host"] = host - actualArgs["port"] = port - actualArgs["durable"] = durable - actualArgs["authMechanism"] = authMechanism - actualArgs["username"] = username - actualArgs["password"] = password - actualArgs["transport"] = transport - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "connect", + actualArgs = list() + actualArgs.append(host) + actualArgs.append(port) + actualArgs.append(durable) + actualArgs.append(authMechanism) + actualArgs.append(username) + actualArgs.append(password) + actualArgs.append(transport) + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "connect", callback, args=actualArgs) def queueMoveMessages(self, model, callback, srcQueue, destQueue, qty): """Move messages from one queue to another""" - actualArgs = dict() - actualArgs["srcQueue"] = srcQueue - actualArgs["destQueue"] = destQueue - actualArgs["qty"] = qty - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "queueMoveMessages", + actualArgs = list() + actualArgs.append(srcQueue) + actualArgs.append(destQueue) + actualArgs.append(qty) + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "queueMoveMessages", callback, args=actualArgs) class BrokerStats(SQLObject): @@ -812,6 +789,7 @@ recTime = TimestampCol(default=None) sourceScopeId = BigIntCol(default=None) sourceObjectId = BigIntCol(default=None) + qmfClassKey = BLOBCol(default=None) creationTime = TimestampCol(default=None) deletionTime = TimestampCol(default=None) managedBroker = StringCol(length=1000, default=None) @@ -844,6 +822,7 @@ recTime = TimestampCol(default=None) sourceScopeId = BigIntCol(default=None) sourceObjectId = BigIntCol(default=None) + qmfClassKey = BLOBCol(default=None) creationTime = TimestampCol(default=None) deletionTime = TimestampCol(default=None) managedBroker = StringCol(length=1000, default=None) @@ -873,6 +852,7 @@ recTime = TimestampCol(default=None) sourceScopeId = BigIntCol(default=None) sourceObjectId = BigIntCol(default=None) + qmfClassKey = BLOBCol(default=None) creationTime = TimestampCol(default=None) deletionTime = TimestampCol(default=None) managedBroker = StringCol(length=1000, default=None) @@ -889,12 +869,10 @@ def purge(self, model, callback, request): """Discard all or some messages on a queue""" - actualArgs = dict() - actualArgs["request"] = request - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "purge", + actualArgs = list() + actualArgs.append(request) + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "purge", callback, args=actualArgs) class QueueStats(SQLObject): @@ -942,6 +920,7 @@ recTime = TimestampCol(default=None) sourceScopeId = BigIntCol(default=None) sourceObjectId = BigIntCol(default=None) + qmfClassKey = BLOBCol(default=None) creationTime = TimestampCol(default=None) deletionTime = TimestampCol(default=None) managedBroker = StringCol(length=1000, default=None) @@ -985,6 +964,7 @@ recTime = TimestampCol(default=None) sourceScopeId = BigIntCol(default=None) sourceObjectId = BigIntCol(default=None) + qmfClassKey = BLOBCol(default=None) creationTime = TimestampCol(default=None) deletionTime = TimestampCol(default=None) managedBroker = StringCol(length=1000, default=None) @@ -1017,6 +997,7 @@ recTime = TimestampCol(default=None) sourceScopeId = BigIntCol(default=None) sourceObjectId = BigIntCol(default=None) + qmfClassKey = BLOBCol(default=None) creationTime = TimestampCol(default=None) deletionTime = TimestampCol(default=None) managedBroker = StringCol(length=1000, default=None) @@ -1032,11 +1013,9 @@ def close(self, model, callback): - actualArgs = dict() - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "close", + actualArgs = list() + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "close", callback, args=actualArgs) class ClientConnectionStats(SQLObject): @@ -1062,6 +1041,7 @@ recTime = TimestampCol(default=None) sourceScopeId = BigIntCol(default=None) sourceObjectId = BigIntCol(default=None) + qmfClassKey = BLOBCol(default=None) creationTime = TimestampCol(default=None) deletionTime = TimestampCol(default=None) managedBroker = StringCol(length=1000, default=None) @@ -1076,29 +1056,25 @@ def close(self, model, callback): - actualArgs = dict() - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "close", + actualArgs = list() + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "close", callback, args=actualArgs) def bridge(self, model, callback, durable, src, dest, key, tag, excludes, srcIsQueue, srcIsLocal, dynamic): """Bridge messages over the link""" - actualArgs = dict() - actualArgs["durable"] = durable - actualArgs["src"] = src - actualArgs["dest"] = dest - actualArgs["key"] = key - actualArgs["tag"] = tag - actualArgs["excludes"] = excludes - actualArgs["srcIsQueue"] = srcIsQueue - actualArgs["srcIsLocal"] = srcIsLocal - actualArgs["dynamic"] = dynamic - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "bridge", + actualArgs = list() + actualArgs.append(durable) + actualArgs.append(src) + actualArgs.append(dest) + actualArgs.append(key) + actualArgs.append(tag) + actualArgs.append(excludes) + actualArgs.append(srcIsQueue) + actualArgs.append(srcIsLocal) + actualArgs.append(dynamic) + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "bridge", callback, args=actualArgs) class LinkStats(SQLObject): @@ -1121,6 +1097,7 @@ recTime = TimestampCol(default=None) sourceScopeId = BigIntCol(default=None) sourceObjectId = BigIntCol(default=None) + qmfClassKey = BLOBCol(default=None) creationTime = TimestampCol(default=None) deletionTime = TimestampCol(default=None) managedBroker = StringCol(length=1000, default=None) @@ -1141,11 +1118,9 @@ def close(self, model, callback): - actualArgs = dict() - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "close", + actualArgs = list() + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "close", callback, args=actualArgs) class BridgeStats(SQLObject): @@ -1166,6 +1141,7 @@ recTime = TimestampCol(default=None) sourceScopeId = BigIntCol(default=None) sourceObjectId = BigIntCol(default=None) + qmfClassKey = BLOBCol(default=None) creationTime = TimestampCol(default=None) deletionTime = TimestampCol(default=None) managedBroker = StringCol(length=1000, default=None) @@ -1182,35 +1158,27 @@ def solicitAck(self, model, callback): - actualArgs = dict() - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "solicitAck", + actualArgs = list() + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "solicitAck", callback, args=actualArgs) def detach(self, model, callback): - actualArgs = dict() - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "detach", + actualArgs = list() + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "detach", callback, args=actualArgs) def resetLifespan(self, model, callback): - actualArgs = dict() - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "resetLifespan", + actualArgs = list() + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "resetLifespan", callback, args=actualArgs) def close(self, model, callback): - actualArgs = dict() - conn = model.connections[self.managedBroker] - classInfo = self.classInfos[self.managedBroker] - originalId = objectId(None, self.sourceScopeId, self.sourceObjectId) - conn.callMethod(originalId, classInfo, "close", + actualArgs = list() + originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId) + model.callMethod(self.managedBroker, originalId, self.qmfClassKey, "close", callback, args=actualArgs) class SessionStats(SQLObject): Modified: mgmt/trunk/mint/python/mint/schemaparser.py =================================================================== --- mgmt/trunk/mint/python/mint/schemaparser.py 2008-11-14 19:30:06 UTC (rev 2804) +++ mgmt/trunk/mint/python/mint/schemaparser.py 2008-11-14 20:12:00 UTC (rev 2805) @@ -139,6 +139,7 @@ else: self.generateAttrib("sourceScopeId", "BigIntCol") self.generateAttrib("sourceObjectId", "BigIntCol") + self.generateAttrib("qmfClassKey", "BLOBCol") self.generateTimestampAttrib("creation") self.generateTimestampAttrib("deletion") self.generateAttrib("managedBroker", "StringCol", "length=1000") @@ -154,10 +155,11 @@ else: comment = "" formalArgs = ", " - actualArgs = " actualArgs = dict()\n" + actualArgs = " actualArgs = list()\n" for arg in elem.query["arg"]: formalArgs += "%s, " % (arg["@name"]) - actualArgs += " actualArgs[\"%s\"] = %s\n" % (arg["@name"], arg["@name"]) + actualArgs += " actualArgs.append(%s)\n" % (arg["@name"]) + if (formalArgs != ", "): formalArgs = formalArgs[:-2] else: @@ -165,10 +167,8 @@ self.pythonOutput += "\n def %s(self, model, callback%s):\n" % (elem["@name"], formalArgs) self.pythonOutput += comment self.pythonOutput += actualArgs - self.pythonOutput += " conn = model.connections[self.managedBroker]\n" - self.pythonOutput += " classInfo = self.classInfos[self.managedBroker]\n" - self.pythonOutput += " originalId = objectId(None, self.sourceScopeId, self.sourceObjectId)\n" - self.pythonOutput += " conn.callMethod(originalId, classInfo, \"%s\",\n" % elem["@name"] + self.pythonOutput += " originalId = ObjectId(None, self.sourceScopeId, self.sourceObjectId)\n" + self.pythonOutput += " model.callMethod(self.managedBroker, originalId, self.qmfClassKey, \"%s\",\n" % elem["@name"] self.pythonOutput += " callback, args=actualArgs)\n" def endClass(self): @@ -183,7 +183,7 @@ self.pythonOutput += "import mint\n" self.pythonOutput += "from sqlobject import *\n" self.pythonOutput += "from datetime import datetime\n" - self.pythonOutput += "from qpid.management import objectId\n\n" + self.pythonOutput += "from qpid.qmfconsole import ObjectId\n\n" self.pythonOutput += "class Pool(SQLObject):\n" self.pythonOutput += " class sqlmeta:\n" self.pythonOutput += " lazyUpdate = True\n" Modified: mgmt/trunk/mint/python/mint/update.py =================================================================== --- mgmt/trunk/mint/python/mint/update.py 2008-11-14 19:30:06 UTC (rev 2804) +++ mgmt/trunk/mint/python/mint/update.py 2008-11-14 20:12:00 UTC (rev 2805) @@ -1,16 +1,13 @@ import logging - +import datetime +import types +import struct from Queue import Queue as ConcurrentQueue, Full, Empty from threading import Thread -from datetime import datetime -from qpid.management import managementClient -from struct import unpack -from schema import schemaReservedWordsMap -from sqlobject import DateTimeCol, TimestampCol, Col +from traceback import print_exc +from qpid.datatypes import UUID +from mint.schema import * -import types -import mint - log = logging.getLogger("mint.update") def time_unwarp(t): @@ -33,14 +30,10 @@ self.updates.put(update) except Full: log.exception("Queue is full") + pass def run(self): while True: - #size = self.updates.qsize() - # - #if size > 1: - # log.debug("Queue depth is %i", self.updates.qsize()) - try: update = self.updates.get(True, 1) except Empty: @@ -54,303 +47,202 @@ update.process(self.model) except: log.exception("Update failed") + pass def stop(self): self.stopRequested = True -class UnknownClassException(Exception): - pass +class ModelUpdate(object): + def __init__(self, broker, obj): + self.broker = broker + self.qmfObj = obj -def unmarshalClassInfo(classInfo): - package = classInfo[0] - name = classInfo[1].capitalize() + def getStatsClass(self, cls): + return getattr(mint, cls.__name__ + "Stats") - if name == "Connection": - name = "ClientConnection" + def process(self): + pass - cls = getattr(mint, name) + def processAttributes(self, attrs, cls, model): + # translate keys into their string representation + for key in attrs.keys(): + attrs[key.__repr__()] = attrs.pop(key) - if cls is None: - raise UnknownClassException("Class '%s' is unknown" % name) + orphan = False + for name in attrs.keys(): + rename = "" + orig_name = name + if name in mint.schema.schemaReservedWordsMap: + rename = mint.schema.schemaReservedWordsMap.get(name) + attrs[rename] = attrs.pop(name) + name = rename - return package, cls + if len(name) > 3 and name.endswith("Ref"): + # Navigate to referenced objects + clsname = name[0].upper() + name[1:-3] + id = attrs.pop(name) -def processAttributes(conn, attrs, cls, model, updateObj): - attrs.pop("id") + othercls = getattr(mint, clsname, None) - if "connectionRef" in attrs: - attrs["clientConnectionRef"] = attrs.pop("connectionRef") + if othercls: + attrname = name[0:-3] - orphan = False - for name in attrs.keys(): - rename = schemaReservedWordsMap.get(name) + try: + attrs[attrname] = model.getObject(othercls, id, self.broker) + except KeyError: + log.info("Referenced object %s '%s' not found by key '%s'" % (clsname, id, attrname)) + except mint.ObjectNotFound: + if not orphan: + log.info("Referenced object %s '%s' not found, deferring creation of orphan object" % (clsname, id)) + # store object in orphan map, will be picked up later when parent info is received + if (clsname, id.first, id.second) not in model.orphanObjectMap: + model.orphanObjectMap[(clsname, id.first, id.second)] = set() + model.orphanObjectMap[(clsname, id.first, id.second)].add(self) + orphan = True + else: + log.error("Class '%s' not found" % clsname) + elif not hasattr(cls, orig_name): + # Remove attrs that we don't have in our schema + log.debug("Class '%s' has no field '%s'" % (cls.__name__, name)) + del attrs[name] + #XXX FIX -- TODO when converting to new API, will lookup attribute type in schema representation + elif name in ("DaemonStartTime", "EnteredCurrentActivity", "EnteredCurrentState", "JobStart", + "LastBenchmark", "LastFetchWorkCompleted", "LastFetchWorkSpawned", "LastPeriodicCheckpoint", + "MyCurrentTime", "QDate", "JobQueueBirthdate", "MonitorSelfTime") \ + and (type(attrs[name]) is types.LongType or type(attrs[name]) is types.IntType or attrs[name] == 0): + attrs[name] = datetime.fromtimestamp(attrs[name]/1000000000) + elif name.endswith("Time") and type(attrs[name]) is types.IntType and attrs[name] == 0: + attrs[name] = datetime.fromtimestamp(attrs[name]) + #XXX FIX -- TODO when converting to new API, will lookup attribute type in schema representation + elif isinstance(attrs[name], UUID): + # convert UUIDs into their string representation, to be handled by sqlobject + attrs[name] = str(attrs[name]) + if orphan: + return None + else: + return attrs - if rename: - attrs[rename] = attrs.pop(name) - name = rename - if len(name) > 3 and name.endswith("Ref"): - # Navigate to referenced objects +class PropertyUpdate(ModelUpdate): + def __init__(self, broker, obj): + ModelUpdate.__init__(self, broker, obj) - clsname = name[0].upper() + name[1:-3] - id = attrs.pop(name) - - othercls = getattr(mint, clsname, None) - - if othercls: - attrname = name[0:-3] - - try: - attrs[attrname] = conn.getObject(othercls, id) - except KeyError: - log.info("Referenced object %s '%s' not found by key '%s'" % (clsname, id, attrname)) - except mint.ObjectNotFound: - if not orphan: - log.info("Referenced object %s '%s' not found, deferring creation of orphan object" % (clsname, id)) - # store object in orphan map, will be picked up later when parent info is received - if (clsname, id.first, id.second) not in model.orphanObjectMap: - model.orphanObjectMap[(clsname, id.first, id.second)] = set() - model.orphanObjectMap[(clsname, id.first, id.second)].add(updateObj) - orphan = True - else: - log.error("Class '%s' not found" % clsname) - elif not hasattr(cls, name): - # Remove attrs that we don't have in our schema - log.debug("Class '%s' has no field '%s'" % (cls.__name__, name)) - del attrs[name] - #XXX FIX -- TODO when converting to new API, will lookup attribute type in schema representation - elif name in ("DaemonStartTime", "EnteredCurrentActivity", "EnteredCurrentState", "JobStart", - "LastBenchmark", "LastFetchWorkCompleted", "LastFetchWorkSpawned", "LastPeriodicCheckpoint", - "MyCurrentTime", "QDate", "JobQueueBirthdate", "MonitorSelfTime") \ - and (type(attrs[name]) is types.LongType or type(attrs[name]) is types.IntType or attrs[name] == 0): - attrs[name] = datetime.fromtimestamp(attrs[name]/1000000000) - elif name.endswith("Time") and type(attrs[name]) is types.IntType and attrs[name] == 0: - attrs[name] = datetime.fromtimestamp(attrs[name]) - #XXX FIX -- TODO when converting to new API, will lookup attribute type in schema representation - if orphan: - return None - else: - return attrs - -def getStatsClass(cls): - return getattr(mint, cls.__name__ + "Stats") - -class SchemaUpdate(object): - def __init__(self, conn, classInfo, props, stats, methods, events): - self.conn = conn - self.classInfo = classInfo - self.props = props - self.stats = stats - self.methods = methods - self.events = events - def process(self, model): - cls = "%s.%s" % (self.classInfo[0], self.classInfo[1]) - log.info("Processing %-8s %-16s %-16s" % ("schema", self.conn.id, cls)) - try: - pkg, cls = unmarshalClassInfo(self.classInfo) - cls.classInfos[self.conn.id] = self.classInfo - except UnknownClassException, e: - log.warn(e) - return + cls = self.qmfObj.getSchema().getKey()[1] + if cls in mint.schema.schemaReservedWordsMap: + cls = mint.schema.schemaReservedWordsMap.get(cls) + cls = eval(cls[0].upper()+cls[1:]) + attrs = dict(self.qmfObj.getProperties()) + timestamps = self.qmfObj.getTimestamps() + id = self.qmfObj.getObjectId() - # XXX do more schema checking + if self.processAttributes(attrs, cls, model) == None: + # object is orphan, a parent dependency was not found; + # insertion in db is deferred until parent info is received + return -class PropertyUpdate(object): - def __init__(self, conn, classInfo, props, timestamps): - self.conn = conn - self.classInfo = classInfo - self.props = props - self.timestamps = timestamps + obj = None + try: + obj = model.getObject(cls, id, self.broker) + except mint.ObjectNotFound: + obj = cls() + log.debug("%s(%i) created", cls.__name__, obj.id) + attrs["sourceScopeId"] = id.first + attrs["sourceObjectId"] = id.second + pkg, cls, hash = self.qmfObj.getClassKey() + attrs["qmfClassKey"] = "%s, %s, %s" % (pkg, cls, hash) + attrs["managedBroker"] = str(self.broker.getBrokerId()) + except Exception, e: + print e - def process(self, model): - cls = "%s.%s" % (self.classInfo[0], self.classInfo[1]) - args = ("props", self.conn.id, cls, len(self.props)) - log.info("Processing %-8s %-16s %-16s %3i" % args) + attrs["recTime"] = datetime.fromtimestamp(timestamps[0]/1000000000) + attrs["creationTime"] = datetime.fromtimestamp(timestamps[1]/1000000000) + if timestamps[2] != 0: + attrs["deletionTime"] = datetime.fromtimestamp(timestamps[2]/1000000000) + log.debug("%s(%i) marked deleted", cls, obj.id) - try: - pkg, cls = unmarshalClassInfo(self.classInfo) - except UnknownClassException, e: - log.warn(e) - return + obj.set(**attrs) + obj.syncUpdate() - attrs = dict(self.props) + if (cls, id.first, id.second) in model.orphanObjectMap: + # this object is the parent of orphan objects in the map, re-enqueue for insertion + orphanObjects = model.orphanObjectMap.pop((cls, id.first, id.second)) + for orphanObj in orphanObjects: + model.updateThread.enqueue(orphanObj) + log.info("Inserted %d orphan objects whose creation had been deferred" % (len(orphanObjects))) - id = attrs["id"] - if processAttributes(self.conn, attrs, cls, model, self) == None: - # object is orphan, a parent dependency was not found; - # insertion in db is deferred until parent info is received - return + if cls == "broker": + if str(self.broker.getBrokerId()) in model.managedBrokers: + model.managedBrokers[str(self.broker.getBrokerId())].broker = self.broker + reg = mint.BrokerRegistration.selectBy(url=self.broker.getFullUrl())[0] + reg.broker = obj + reg.syncUpdate() - # XXX move these down to the try/except + except: + print_exc() - attrs["managedBroker"] = self.conn.id + attrs["managedBroker"] = str(self.broker.getBrokerId()) attrs["recTime"] = time_unwarp(datetime.fromtimestamp \ - (self.timestamps[0]/1000000000)) + (timestamps[0]/1000000000)) attrs["creationTime"] = datetime.fromtimestamp \ - (self.timestamps[1]/1000000000) + (timestamps[1]/1000000000) - if self.timestamps[2] != 0: - attrs["deletionTime"] = datetime.fromtimestamp \ - (self.timestamps[2]/1000000000) - try: - obj = self.conn.getObject(cls, id) - except mint.ObjectNotFound: - obj = cls() +class StatisticUpdate(ModelUpdate): + def __init__(self, broker, obj): + ModelUpdate.__init__(self, broker, obj) - log.debug("%s(%i) created", cls.__name__, obj.id) - - attrs["sourceScopeId"] = id.first - attrs["sourceObjectId"] = id.second - - #hash = "%08x%08x%08x%08x" % unpack("!LLLL", self.classInfo[2]) - #classInfo = (self.classInfo[0], self.classInfo[1], hash) - #obj.sourceClassInfo = ",".join(classInfo) - - obj.set(**attrs) - obj.syncUpdate() - - if obj.deletionTime: - log.debug("%s(%i) marked deleted", cls.__name__, obj.id) - - if (cls.__name__, id.first, id.second) in model.orphanObjectMap: - # this object is the parent of orphan objects in the map, re-enqueue for insertion - orphanObjects = model.orphanObjectMap.pop((cls.__name__, id.first, id.second)) - for orphanObj in orphanObjects: - model.updateThread.enqueue(orphanObj) - log.info("Inserted %d orphan objects whose creation had been deferred" % (len(orphanObjects))) - - # XXX refactor this to take advantage of the get/create logic - # above - if isinstance(obj, mint.Broker) and obj.managedBroker: - host, port = obj.managedBroker.split(":") - port = int(port) - - if not obj.registration: - try: - reg = mint.BrokerRegistration.selectBy(host=host, port=port)[0] - except IndexError: - reg = None - - if reg: - reg.broker = obj - obj.registration = reg - - reg.syncUpdate() - obj.syncUpdate() - -class StatisticUpdate(object): - def __init__(self, conn, classInfo, stats, timestamps): - self.conn = conn - self.classInfo = classInfo - self.stats = stats - self.timestamps = timestamps - def process(self, model): - cls = "%s.%s" % (self.classInfo[0], self.classInfo[1]) - args = ("stats", self.conn.id, cls, len(self.stats)) - log.info("Processing %-8s %-16s %-16s %3i" % args) - try: - pkg, cls = unmarshalClassInfo(self.classInfo) - except UnknownClassException, e: - log.warn(e) - return + cls = self.qmfObj.getSchema().getKey()[1] + if cls in mint.schema.schemaReservedWordsMap: + cls = mint.schema.schemaReservedWordsMap.get(cls) + cls = eval(cls[0].upper()+cls[1:]) + attrs = dict(self.qmfObj.getStatistics()) + timestamps = self.qmfObj.getTimestamps() + id = self.qmfObj.getObjectId() - attrs = dict(self.stats) + obj = None + try: + obj = model.getObject(cls, id, self.broker) + except mint.ObjectNotFound: + # handle this + raise mint.ObjectNotFound - id = attrs["id"] - obj = self.conn.getObject(cls, id) - statscls = getStatsClass(cls) - if processAttributes(self.conn, attrs, statscls, model, self) == None: - # object is orphan, a parent dependency was not found; - # insertion in db is deferred until parent info is received - return + statscls = self.getStatsClass(cls) + if self.processAttributes(attrs, statscls, model) == None: + # object is orphan, a parent dependency was not found; + # insertion in db is deferred until parent info is received + return - attrs["recTime"] = time_unwarp(datetime.fromtimestamp \ - (self.timestamps[0]/1000000000)) - # Set the stat->obj reference - attrs[cls.__name__[0].lower() + cls.__name__[1:]] = obj + attrs["recTime"] = time_unwarp(datetime.fromtimestamp \ + (timestamps[0]/1000000000)) - statsobj = statscls() - statsobj.set(**attrs) - statsobj.syncUpdate() + # Set the stat->obj reference + attrs[cls.__name__[0].lower() + cls.__name__[1:]] = obj + statsobj = statscls() + statsobj.set(**attrs) + statsobj.syncUpdate() - # XXX not sure if this should happen here. makes more sense in - # prop update - if self.timestamps[2] != 0: - obj.deletionTime = datetime.fromtimestamp(self.timestamps[2]/1000000000) + obj.statsPrev = obj.statsCurr + obj.statsCurr = statsobj + obj.syncUpdate() - obj.statsPrev = obj.statsCurr - obj.statsCurr = statsobj - obj.syncUpdate() + except: + print_exc() -class MethodUpdate(object): - def __init__(self, conn, methodId, errorCode, errorText, args): - self.conn = conn - self.methodId = methodId - self.errorCode = errorCode - self.errorText = errorText - self.args = args - def process(self, model): - logArgs = ("method", self.conn.id, self.methodId, self.errorCode, - self.errorText) - log.info("Processing %-8s %-16s %-12s %-12s %s" % logArgs) +class MethodUpdate(ModelUpdate): + def __init__(self, broker, seq, response): + ModelUpdate.__init__(self, broker, response) + self.seq = seq - model.lock() - try: - method = model.outstandingMethodCalls.pop(self.methodId) - method(self.errorText, self.args) - finally: - model.unlock() - -class CloseUpdate(object): - def __init__(self, conn, data): - self.conn = conn - self.data = data - def process(self, model): - log.info("Processing %-8s %-16s" % ("close", self.conn.id)) - model.lock() try: - del model.connections[self.conn.id] - - if model.connCloseListener: - model.connCloseListener(self.conn, self.data) + methodCallback = model.outstandingMethodCalls.pop(self.seq) + methodCallback(self.qmfObj.text, self.qmfObj.outArgs) finally: model.unlock() - -class ControlUpdate(object): - __types = { - managementClient.CTRL_BROKER_INFO: "broker_info", - managementClient.CTRL_SCHEMA_LOADED: "schema_loaded", - managementClient.CTRL_USER: "user", - managementClient.CTRL_HEARTBEAT: "heartbeat" - } - - def __init__(self, conn, typeCode, data): - self.conn = conn - self.typeCode = typeCode - self.data = data - - def process(self, model): - type = self.__types.get(self.typeCode, "[unknown]") - args = ("control", self.conn.id, type, self.data) - log.info("Processing %-8s %-16s %-16s %s" % args) - - if self.typeCode == managementClient.CTRL_BROKER_INFO: - uuid = "%08x-%04x-%04x-%04x-%04x%08x" % unpack \ - ("!LHHHHL", self.data.brokerId) - - log.info("Broker ID from %s is '%s'" % (self.conn.id, uuid)) - log.info("Session ID from %s is '%s'" % \ - (self.conn.id, self.data.sessionId)) - - self.conn.brokerId = uuid - self.conn.sessionId = self.data.sessionId - Modified: mgmt/trunk/mint/sql/schema.sql =================================================================== --- mgmt/trunk/mint/sql/schema.sql 2008-11-14 19:30:06 UTC (rev 2804) +++ mgmt/trunk/mint/sql/schema.sql 2008-11-14 20:12:00 UTC (rev 2805) @@ -23,13 +23,12 @@ CREATE TABLE broker_registration ( id SERIAL PRIMARY KEY, name VARCHAR(1000) NOT NULL UNIQUE, - host VARCHAR(1000) NOT NULL, - port INT NOT NULL, + url VARCHAR(1000), broker_id INT, cluster_id INT, profile_id INT ); -CREATE UNIQUE INDEX broker_registration_host_port_unique ON broker_registration (host, port); +CREATE UNIQUE INDEX broker_registration_url_unique ON broker_registration (url); CREATE TABLE collector_registration ( id SERIAL PRIMARY KEY, @@ -37,13 +36,6 @@ collector_id VARCHAR(1000) ); -CREATE TABLE config_property ( - id SERIAL PRIMARY KEY, - name VARCHAR(1000), - value VARCHAR(1000), - type VARCHAR(1) -); - CREATE TABLE mint_info ( id SERIAL PRIMARY KEY, version VARCHAR(1000) NOT NULL @@ -75,6 +67,7 @@ rec_time TIMESTAMP, source_scope_id BIGINT, source_object_id BIGINT, + qmf_class_key BYTEA, creation_time TIMESTAMP, deletion_time TIMESTAMP, managed_broker VARCHAR(1000), @@ -99,6 +92,7 @@ rec_time TIMESTAMP, source_scope_id BIGINT, source_object_id BIGINT, + qmf_class_key BYTEA, creation_time TIMESTAMP, deletion_time TIMESTAMP, managed_broker VARCHAR(1000), @@ -123,6 +117,7 @@ rec_time TIMESTAMP, source_scope_id BIGINT, source_object_id BIGINT, + qmf_class_key BYTEA, creation_time TIMESTAMP, deletion_time TIMESTAMP, managed_broker VARCHAR(1000), @@ -147,6 +142,7 @@ rec_time TIMESTAMP, source_scope_id BIGINT, source_object_id BIGINT, + qmf_class_key BYTEA, creation_time TIMESTAMP, deletion_time TIMESTAMP, managed_broker VARCHAR(1000), @@ -176,6 +172,7 @@ rec_time TIMESTAMP, source_scope_id BIGINT, source_object_id BIGINT, + qmf_class_key BYTEA, creation_time TIMESTAMP, deletion_time TIMESTAMP, managed_broker VARCHAR(1000), @@ -204,6 +201,7 @@ rec_time TIMESTAMP, source_scope_id BIGINT, source_object_id BIGINT, + qmf_class_key BYTEA, creation_time TIMESTAMP, deletion_time TIMESTAMP, managed_broker VARCHAR(1000), @@ -233,6 +231,7 @@ rec_time TIMESTAMP, source_scope_id BIGINT, source_object_id BIGINT, + qmf_class_key BYTEA, creation_time TIMESTAMP, deletion_time TIMESTAMP, managed_broker VARCHAR(1000), @@ -258,6 +257,7 @@ rec_time TIMESTAMP, source_scope_id BIGINT, source_object_id BIGINT, + qmf_class_key BYTEA, creation_time TIMESTAMP, deletion_time TIMESTAMP, managed_broker VARCHAR(1000), @@ -282,6 +282,7 @@ rec_time TIMESTAMP, source_scope_id BIGINT, source_object_id BIGINT, + qmf_class_key BYTEA, creation_time TIMESTAMP, deletion_time TIMESTAMP, managed_broker VARCHAR(1000), @@ -317,6 +318,7 @@ rec_time TIMESTAMP, source_scope_id BIGINT, source_object_id BIGINT, + qmf_class_key BYTEA, creation_time TIMESTAMP, deletion_time TIMESTAMP, managed_broker VARCHAR(1000), @@ -363,6 +365,7 @@ rec_time TIMESTAMP, source_scope_id BIGINT, source_object_id BIGINT, + qmf_class_key BYTEA, creation_time TIMESTAMP, deletion_time TIMESTAMP, managed_broker VARCHAR(1000), @@ -420,6 +423,7 @@ rec_time TIMESTAMP, source_scope_id BIGINT, source_object_id BIGINT, + qmf_class_key BYTEA, creation_time TIMESTAMP, deletion_time TIMESTAMP, managed_broker VARCHAR(1000), @@ -445,6 +449,7 @@ rec_time TIMESTAMP, source_scope_id BIGINT, source_object_id BIGINT, + qmf_class_key BYTEA, creation_time TIMESTAMP, deletion_time TIMESTAMP, managed_broker VARCHAR(1000), @@ -479,6 +484,7 @@ rec_time TIMESTAMP, source_scope_id BIGINT, source_object_id BIGINT, + qmf_class_key BYTEA, creation_time TIMESTAMP, deletion_time TIMESTAMP, managed_broker VARCHAR(1000), @@ -516,6 +522,7 @@ rec_time TIMESTAMP, source_scope_id BIGINT, source_object_id BIGINT, + qmf_class_key BYTEA, creation_time TIMESTAMP, deletion_time TIMESTAMP, managed_broker VARCHAR(1000), @@ -567,6 +574,7 @@ rec_time TIMESTAMP, source_scope_id BIGINT, source_object_id BIGINT, + qmf_class_key BYTEA, creation_time TIMESTAMP, deletion_time TIMESTAMP, managed_broker VARCHAR(1000), @@ -608,6 +616,7 @@ rec_time TIMESTAMP, source_scope_id BIGINT, source_object_id BIGINT, + qmf_class_key BYTEA, creation_time TIMESTAMP, deletion_time TIMESTAMP, managed_broker VARCHAR(1000), @@ -638,6 +647,7 @@ rec_time TIMESTAMP, source_scope_id BIGINT, source_object_id BIGINT, + qmf_class_key BYTEA, creation_time TIMESTAMP, deletion_time TIMESTAMP, managed_broker VARCHAR(1000), @@ -747,6 +757,7 @@ rec_time TIMESTAMP, source_scope_id BIGINT, source_object_id BIGINT, + qmf_class_key BYTEA, creation_time TIMESTAMP, deletion_time TIMESTAMP, managed_broker VARCHAR(1000), @@ -785,6 +796,7 @@ rec_time TIMESTAMP, source_scope_id BIGINT, source_object_id BIGINT, + qmf_class_key BYTEA, creation_time TIMESTAMP, deletion_time TIMESTAMP, managed_broker VARCHAR(1000), @@ -811,6 +823,7 @@ rec_time TIMESTAMP, source_scope_id BIGINT, source_object_id BIGINT, + qmf_class_key BYTEA, creation_time TIMESTAMP, deletion_time TIMESTAMP, managed_broker VARCHAR(1000), @@ -835,6 +848,7 @@ rec_time TIMESTAMP, source_scope_id BIGINT, source_object_id BIGINT, + qmf_class_key BYTEA, creation_time TIMESTAMP, deletion_time TIMESTAMP, managed_broker VARCHAR(1000), Modified: mgmt/trunk/parsley/python/parsley/command.py =================================================================== --- mgmt/trunk/parsley/python/parsley/command.py 2008-11-14 19:30:06 UTC (rev 2804) +++ mgmt/trunk/parsley/python/parsley/command.py 2008-11-14 20:12:00 UTC (rev 2805) @@ -12,6 +12,9 @@ self.commands = list() self.commands_by_name = dict() + opt = CommandOption(self, "help", "h") + opt.description = "Print this message" + if self.parent: self.parent.commands.append(self) self.parent.commands_by_name[self.name] = self