[rhmessaging-commits] rhmessaging commits: r3253 - in mgmt/trunk: mint/instance/etc and 1 other directories.

Author: justi9 Date: 2009-04-02 15:49:59 -0400 (Thu, 02 Apr 2009) New Revision: 3253 Added: mgmt/trunk/mint/python/mint/database.py mgmt/trunk/mint/python/mint/expire.py mgmt/trunk/mint/python/mint/model.py mgmt/trunk/mint/python/mint/poll.py Removed: mgmt/trunk/mint/python/mint/dbexpire.py Modified: mgmt/trunk/cumin/python/cumin/broker.py mgmt/trunk/cumin/python/cumin/limits.py mgmt/trunk/cumin/python/cumin/managementserver.py mgmt/trunk/cumin/python/cumin/model.py mgmt/trunk/cumin/python/cumin/tools.py mgmt/trunk/mint/instance/etc/mint.conf mgmt/trunk/mint/python/mint/__init__.py mgmt/trunk/mint/python/mint/main.py mgmt/trunk/mint/python/mint/tools.py mgmt/trunk/mint/python/mint/update.py mgmt/trunk/mint/python/mint/util.py Log: Don't enable debug by default for the devel mint instance Modified: mgmt/trunk/cumin/python/cumin/broker.py =================================================================== --- mgmt/trunk/cumin/python/cumin/broker.py 2009-04-02 19:42:25 UTC (rev 3252) +++ mgmt/trunk/cumin/python/cumin/broker.py 2009-04-02 19:49:59 UTC (rev 3253) @@ -79,7 +79,7 @@ def render_content(self, session, data): scopeId = data["qmf_scope_id"] - dt = self.app.model.data.getLatestHeartbeat(scopeId) + dt = self.app.model.mint.model.getLatestHeartbeat(scopeId) if dt is None: return fmt_none() @@ -202,7 +202,7 @@ try: id = broker.qmfBrokerId - mbroker = self.app.model.data.mintBrokersById[id] + mbroker = self.app.model.mint.model.mintBrokersById[id] connected = mbroker.connected except KeyError: pass Modified: mgmt/trunk/cumin/python/cumin/limits.py =================================================================== --- mgmt/trunk/cumin/python/cumin/limits.py 2009-04-02 19:42:25 UTC (rev 3252) +++ mgmt/trunk/cumin/python/cumin/limits.py 2009-04-02 19:49:59 UTC (rev 3253) @@ -52,7 +52,7 @@ #TODO: this probably shouldn't be called until after the invoke completes def completion(): pass - negotiator.Reconfig(self.app.model.data, completion) + negotiator.Reconfig(self.app.model.mint.model, completion) def get_raw_limits(self, session, negotiator): action = self.app.model.negotiator.GetLimits Modified: mgmt/trunk/cumin/python/cumin/managementserver.py =================================================================== --- mgmt/trunk/cumin/python/cumin/managementserver.py 2009-04-02 19:42:25 UTC (rev 3252) +++ mgmt/trunk/cumin/python/cumin/managementserver.py 2009-04-02 19:49:59 UTC (rev 3253) @@ -41,7 +41,7 @@ url = data["url"] try: - server = self.app.model.data.mintBrokersByUrl[url] + server = self.app.model.mint.model.mintBrokersByUrl[url] if server.connected: html = "Connected" @@ -274,7 +274,7 @@ url = data["url"] try: - server = self.app.model.data.mintBrokersByUrl[url] + server = self.app.model.mint.model.mintBrokersByUrl[url] if server.connected: html = "Connected" Modified: mgmt/trunk/cumin/python/cumin/model.py =================================================================== --- mgmt/trunk/cumin/python/cumin/model.py 2009-04-02 19:42:25 UTC (rev 3252) +++ mgmt/trunk/cumin/python/cumin/model.py 2009-04-02 19:49:59 UTC (rev 3253) @@ -1,31 +1,37 @@ +import logging from datetime import datetime, timedelta +from mint import Mint, MintConfig +from struct import unpack, calcsize +from types import * +from wooly import * +from wooly.parameters import * +from wooly.widgets import * + from formats import * from job import * from parameters import * from pool import PoolSlotSet, PoolJobStats from slot import SlotStatSet -from struct import unpack, calcsize from system import SystemSlotSet from time import * -from types import * from util import * -from wooly import * -from wooly.parameters import * -from wooly.widgets import * -from mint.schema import * -import logging - log = logging.getLogger("cumin.model") class CuminModel(object): def __init__(self, app, data_uri): self.app = app - self.data = MintModel(data_uri) - self.data.updateObjects = False - self.data.expireObjects = False + opts = {"data": data_uri, "qmf": None} + config = MintConfig() + config.init(opts) + + self.mint = Mint(config) + self.mint.updateEnabled = False + self.mint.pollEnabled = True + self.mint.expireEnabled = False + self.classes = list() self.invocations = set() @@ -68,10 +74,10 @@ CuminSlot(self) def check(self): - self.data.check() + self.mint.check() def init(self): - self.data.init() + self.mint.init() self.frame = self.app.main_page.main @@ -79,10 +85,10 @@ cls.init() def start(self): - self.data.start() + self.mint.start() def stop(self): - self.data.stop() + self.mint.stop() def add_class(self, cls): self.classes.append(cls) @@ -306,7 +312,7 @@ def get_session_by_object(self, object): assert object - broker = self.model.data.mintBrokersById[object.qmfBrokerId] + broker = self.mint.model.mintBrokersById[object.qmfBrokerId] return broker.getAmqpSession() @@ -737,7 +743,7 @@ master = Master.select("System = '%s'" % system_name)[0] except IndexError: raise Exception("Master daemon not running") - master.Start(self.model.data, completion, args["subsystem"]) + master.Start(self.mint.model, completion, args["subsystem"]) class Stop(CuminAction): def do_invoke(self, object, args, completion): @@ -746,7 +752,7 @@ master = Master.select("System = '%s'" % system_name)[0] except IndexError: raise Exception("Master daemon not running") - #master.Stop(self.model.data, completion, args["subsystem"]) + #master.Stop(self.mint.model, completion, args["subsystem"]) class CuminBroker(RemoteClass): def __init__(self, model): @@ -816,7 +822,8 @@ connected = False if broker: try: - mbroker = self.model.data.mintBrokersById[broker.qmfBrokerId] + mbroker = self.mint.model.mintBrokersById \ + [broker.qmfBrokerId] connected = mbroker.connected except KeyError: pass @@ -897,7 +904,7 @@ else: authMechanism = "PLAIN" - broker.connect(self.model.data, completion, host, port, durable, + broker.connect(self.mint.model, completion, host, port, durable, authMechanism, username, password, transport) class AddQueue(CuminAction): @@ -1131,7 +1138,7 @@ return "Purge" def do_invoke(self, queue, args, completion): - queue.purge(self.model.data, completion, args["request"]) + queue.purge(self.mint.model, completion, args["request"]) class Remove(CuminAction): def show(self, session, queue): @@ -1174,7 +1181,9 @@ def do_invoke(self, queue, args, completion): broker = queue.vhost.broker - broker.queueMoveMessages(self.model.data, completion, args["srcQueue"], args["destQueue"], args["qty"]) + broker.queueMoveMessages(self.mint.model, completion, + args["srcQueue"], args["destQueue"], + args["qty"]) class CuminExchange(RemoteClass): def __init__(self, model): @@ -1351,7 +1360,7 @@ return frame.show_remove(session) def do_invoke(self, bridge, args, completion): - bridge.close(self.model.data, completion) + bridge.close(self.mint.model, completion) class CuminConnection(RemoteClass): def __init__(self, model): @@ -1421,7 +1430,7 @@ def do_invoke(self, conn, args, completion): session_ids = set() - for broker in self.model.data.mintBrokersByQmfBroker: + for broker in self.mint.model.mintBrokersByQmfBroker: session_ids.add(broker.getSessionId()) for sess in conn.sessions: @@ -1430,7 +1439,7 @@ ("Cannot close management connection %s" % \ conn.address) - conn.close(self.model.data, completion) + conn.close(self.mint.model, completion) class CuminSession(RemoteClass): def __init__(self, model): @@ -1472,38 +1481,38 @@ return "Close" def do_invoke(self, sess, args, completion): - for broker in self.model.data.mintBrokersByQmfBroker: + for broker in self.mint.model.mintBrokersByQmfBroker: if sess.name == broker.getSessionId(): raise Exception \ ("Cannot close management session %s" % sess.name) - sess.close(self.model.data, completion) + sess.close(self.mint.model, completion) class Detach(CuminAction): def get_title(self, session): return "Detach" def do_invoke(self, sess, args, completion): - for broker in self.model.data.mintBrokersByQmfBroker: + for broker in self.mint.model.mintBrokersByQmfBroker: if sess.name == broker.getSessionId(): raise Exception \ ("Cannot detach management session %s" % sess.name) - sess.detach(self.model.data, completion) + sess.detach(self.mint.model, completion) class ResetLifespan(CuminAction): def get_title(self, session): return "Reset Lifespan" def do_invoke(self, object, args, completion): - object.resetLifespan(self.model.data, completion) + object.resetLifespan(self.mint.model, completion) class SolicitAck(CuminAction): def get_title(self, session): return "Solicit Acknowledgment" def do_invoke(self, object, args, completion): - object.solicitAck(self.model.data, completion) + object.solicitAck(self.mint.model, completion) class CuminLink(RemoteClass): def __init__(self, model): @@ -1577,7 +1586,7 @@ excludes = args["excludes"] sync = args["sync"] - link.bridge(self.model.data, completion, + link.bridge(self.mint.model, completion, durable, src, dest, key, tag, excludes, False, False, dynamic, sync) @@ -1590,7 +1599,7 @@ return "Close" def do_invoke(self, link, args, completion): - link.close(self.model.data, completion) + link.close(self.mint.model, completion) class CuminBrokerStoreModule(RemoteClass): def __init__(self, model): @@ -2110,8 +2119,8 @@ def do_invoke(self, limit, negotiator, completion): Name = limit.id Max = limit.max - negotiator.SetLimit(self.model.data, completion, Name, Max) - #negotiator.SetLimit(self.model.data, completion, Name, str(Max)) + negotiator.SetLimit(self.mint.model, completion, Name, Max) + #negotiator.SetLimit(self.mint.model, completion, Name, str(Max)) class CuminJobGroup(CuminClass): def __init__(self, model): @@ -2418,7 +2427,7 @@ return self.got_data try: - job.GetAd(self.model.data, completion, None) + job.GetAd(self.mint.model, completion, None) except Exception, e: return self.job_ads @@ -2452,7 +2461,7 @@ try: data = None - job.Fetch(self.model.data, completion, file, start, end, data) + job.Fetch(self.mint.model, completion, file, start, end, data) # wait for up to 20 seconds for completion to be called wait(predicate, timeout=20) if not self.got_data: @@ -2475,7 +2484,7 @@ return "Hold" def do_invoke(self, job, reason, completion): - job.Hold(self.model.data, completion, reason) + job.Hold(self.mint.model, completion, reason) def get_enabled(self, session, job): is_held = JobStatusInfo.get_status_int("Held") == job.JobStatus @@ -2492,7 +2501,7 @@ return "Release" def do_invoke(self, job, reason, completion): - job.Release(self.model.data, completion, reason) + job.Release(self.mint.model, completion, reason) def get_enabled(self, session, job): is_held = JobStatusInfo.get_status_int("Held") == job.JobStatus @@ -2509,7 +2518,7 @@ return "Remove" def do_invoke(self, job, reason, completion): - job.Remove(self.model.data, completion, reason) + job.Remove(self.mint.model, completion, reason) def get_enabled(self, session, job): is_deleted = job.qmfDeleteTime is not None @@ -2525,7 +2534,7 @@ def do_invoke(self, job, args, completion): Name = args[0] Value = args[1] - job.SetAttribute(self.model.data, completion, Name, str(Value)) + job.SetAttribute(self.mint.model, completion, Name, str(Value)) class GetStartedAction(CuminAction): def get_xml_response(self, session, object, *args): @@ -2808,7 +2817,7 @@ return self.got_data try: - negotiator.GetLimits(self.model.data, completion, None) + negotiator.GetLimits(self.mint.model, completion, None) except Exception, e: return self.lim Modified: mgmt/trunk/cumin/python/cumin/tools.py =================================================================== --- mgmt/trunk/cumin/python/cumin/tools.py 2009-04-02 19:42:25 UTC (rev 3252) +++ mgmt/trunk/cumin/python/cumin/tools.py 2009-04-02 19:49:59 UTC (rev 3253) @@ -155,7 +155,15 @@ def init(self): super(CuminAdminTool, self).init() - self.database = MintDatabase(self.config.data) + config = MintConfig() + config.init({"data": self.config.data}) + + app = Mint(config) + app.updateEnabled = False + app.pollEnabled = False + app.expireEnabled = False + + self.database = MintDatabase(app) self.database.check() self.database.init() @@ -189,8 +197,6 @@ try: opts, args = command.parse(remaining) - - print opts, args, remaining except CommandException, e: print "Error: %s" % e.message e.command.print_help() Modified: mgmt/trunk/mint/instance/etc/mint.conf =================================================================== --- mgmt/trunk/mint/instance/etc/mint.conf 2009-04-02 19:42:25 UTC (rev 3252) +++ mgmt/trunk/mint/instance/etc/mint.conf 2009-04-02 19:49:59 UTC (rev 3253) @@ -1,3 +1,3 @@ [main] data: postgresql://cumin@localhost/cumin -debug: True +#debug: True Modified: mgmt/trunk/mint/python/mint/__init__.py =================================================================== --- mgmt/trunk/mint/python/mint/__init__.py 2009-04-02 19:42:25 UTC (rev 3252) +++ mgmt/trunk/mint/python/mint/__init__.py 2009-04-02 19:49:59 UTC (rev 3253) @@ -1,615 +1,2 @@ -import logging -import qmf.console -import pickle -import struct -import sys -import os -import types -import socket - -from parsley.config import Config, ConfigParameter -from psycopg2 import OperationalError -from qmf.console import ClassKey -from qpid.datatypes import UUID -from qpid.util import URL -from sqlobject import * -from threading import Lock, RLock -from time import sleep -from traceback import print_exc - -from mint import update -from mint.cache import MintCache -from mint.dbexpire import DBExpireThread -from mint.schema import * -from util import * - -log = logging.getLogger("mint") - -thisModule = __import__(__name__) - -for item in dir(schema): - cls = getattr(schema, item) - - try: - if issubclass(cls, SQLObject) and cls is not SQLObject: - setattr(thisModule, item, cls) - except TypeError: - pass - -Broker.sqlmeta.addColumn(ForeignKey("BrokerRegistration", - cascade="null", default=None, - name="registration")) - -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): - sqlhub.processConnection = connectionForURI(self.uri) - - def checkConnection(self): - conn = self.getConnection() - - try: - cursor = conn.cursor() - cursor.execute("select now()") - finally: - conn.close() - - def checkSchema(self): - pass - - def dropSchema(self): - conn = self.getConnection() - try: - cursor = conn.cursor() - - cursor.execute("drop schema public cascade") - - conn.commit() - finally: - conn.close() - - def __splitSQLStatements(self, text): - result = list() - unmatchedQuote = False - tmpStmt = "" - - for stmt in text.split(";"): - stmt = stmt.rstrip() - quotePos = stmt.find("'") - while quotePos > 0: - quotePos += 1 - if quotePos < len(stmt): - if stmt[quotePos] != "'": - unmatchedQuote = not unmatchedQuote - else: - # ignore 2 single quotes - quotePos += 1 - quotePos = stmt.find("'", quotePos) - - if len(stmt.lstrip()) > 0: - tmpStmt += stmt + ";" - if not unmatchedQuote: - # single quote has been matched/closed, generate statement - result.append(tmpStmt.lstrip()) - tmpStmt = "" - - if unmatchedQuote: - result.append(tmpStmt.lstrip()) - return result - - def createSchema(self, file_paths): - conn = self.getConnection() - - scripts = list() - - for path in file_paths: - file = open(path, "r") - try: - scripts.append((path, file.read())) - finally: - file.close() - - try: - cursor = conn.cursor() - - try: - cursor.execute("create schema public"); - except: - conn.rollback() - pass - - for path, text in scripts: - stmts = self.__splitSQLStatements(text) - 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() - - 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 - - name = StringCol(length=1000, default=None, unique=True, notNone=True) - password = StringCol(length=1000, default=None) - lastChallenged = TimestampCol(default=None) - lastLoggedIn = TimestampCol(default=None) - lastLoggedOut = TimestampCol(default=None) - roles = SQLRelatedJoin("Role", intermediateTable="subject_role_mapping", - createRelatedTable=False) - - def getByName(cls, name): - try: - return Subject.selectBy(name=name)[0] - except IndexError: - pass - - getByName = classmethod(getByName) - -class Role(SQLObject): - class sqlmeta: - lazyUpdate = True - - name = StringCol(length=1000, default=None, unique=True, notNone=True) - subjects = SQLRelatedJoin("Subject", - intermediateTable="subject_role_mapping", - createRelatedTable=False) - - def getByName(cls, name): - try: - return Role.selectBy(name=name)[0] - except IndexError: - pass - - getByName = classmethod(getByName) - -class SubjectRoleMapping(SQLObject): - class sqlmeta: - lazyUpdate = True - - subject = ForeignKey("Subject", notNull=True, cascade=True) - role = ForeignKey("Role", notNull=True, cascade=True) - unique = DatabaseIndex(subject, role, unique=True) - -class ObjectNotFound(Exception): - pass - -class MintInfo(SQLObject): - class sqlmeta: - lazyUpdate = True - - version = StringCol(length=1000, default="0.1", notNone=True) - -class CollectorRegistration(SQLObject): - class sqlmeta: - lazyUpdate = True - - name = StringCol(length=1000, default=None) - collectorId = StringCol(length=1000, default=None) - -class BrokerRegistration(SQLObject): - class sqlmeta: - lazyUpdate = True - - name = StringCol(length=1000, default=None, unique=True, notNone=True) - url = StringCol(length=1000, default=None) - broker = ForeignKey("Broker", cascade="null", default=None) - cluster = ForeignKey("BrokerCluster", cascade="null", default=None) - profile = ForeignKey("BrokerProfile", cascade="null", default=None) - - url_unique = DatabaseIndex(url, unique=True) - - def getBrokerId(self): - return self.mintBroker.qmfId - - def getDefaultVhost(self): - if self.broker: - try: - return Vhost.selectBy(broker=self.broker, name="/")[0] - except IndexError: - return None - -class BrokerGroup(SQLObject): - class sqlmeta: - lazyUpdate = True - - name = StringCol(length=1000, default=None, unique=True, notNone=True) - brokers = SQLRelatedJoin("Broker", - intermediateTable="broker_group_mapping", - createRelatedTable=False) - -class BrokerGroupMapping(SQLObject): - class sqlmeta: - lazyUpdate = True - - broker = ForeignKey("Broker", notNull=True, cascade=True) - brokerGroup = ForeignKey("BrokerGroup", notNull=True, cascade=True) - unique = DatabaseIndex(broker, brokerGroup, unique=True) - -class BrokerCluster(SQLObject): - class sqlmeta: - lazyUpdate = True - - name = StringCol(length=1000, default=None) - brokers = SQLMultipleJoin("BrokerRegistration", joinColumn="cluster_id") - -class BrokerProfile(SQLObject): - class sqlmeta: - lazyUpdate = True - - name = StringCol(length=1000, default=None) - brokers = SQLMultipleJoin("BrokerRegistration", joinColumn="profile_id") - properties = SQLMultipleJoin("ConfigProperty", joinColumn="profile_id") - -class MintBroker(object): - def __init__(self, url, qmfBroker): - self.url = url - self.qmfBroker = qmfBroker - - self.qmfId = None - self.databaseId = None - - self.objectDatabaseIds = MintCache() # database ids by qmf object id - self.orphans = dict() # updates by qmf object id - - self.connected = False - - def __repr__(self): - return "%s(%s)" % (self.__class__.__name__, self.url) - - def getAmqpSession(self): - return self.qmfBroker.getAmqpSession() - - def getAmqpSessionId(self): - return self.qmfBroker.getSessionId() - - def getFullUrl(self): - return self.qmfBroker.getFullUrl() - -class MintModel(qmf.console.Console): - staticInstance = None - - def __init__(self, dataUri, debug=False): - self.dataUri = dataUri - self.debug = debug - - self.updateObjects = True - - self.pollRegistrations = True - - self.expireObjects = True - self.expireFrequency = 600 - self.expireThreshold = 24 * 3600 - - assert MintModel.staticInstance is None - MintModel.staticInstance = self - - # Lookup tables used for recovering MintBroker objects, which have - # mint-specific accounting and wrap a qmf broker object - - self.mintBrokersByQmfBroker = dict() - self.mintBrokersById = dict() - self.mintBrokersByUrl = dict() - - # Agent heartbeats - # (broker bank, agent bank) => latest heartbeat timestamp - - self.heartbeatsByAgentId = dict() - - self.__lock = RLock() - - self.dbConn = None - self.qmfSession = None - - self.updateThread = update.ModelUpdateThread(self) - self.expireThread = dbexpire.DBExpireThread(self) - self.registrationThread = RegistrationThread(self) - - self.outstandingMethodCalls = dict() - - def lock(self): - self.__lock.acquire() - - def unlock(self): - self.__lock.release() - - def check(self): - try: - connectionForURI(self.dataUri) - 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): - log.info("Object updates are %s", - self.updateObjects and "enabled" or "disabled") - - log.info("QMF server polling is %s", - self.pollRegistrations and "enabled" or "disabled") - - log.info("Object expiration is %s", - self.expireObjects and "enabled" or "disabled") - - sqlhub.processConnection = self.dbConn = connectionForURI(self.dataUri) - - assert self.qmfSession is None - - self.qmfSession = qmf.console.Session \ - (self, manageConnections=True, rcvObjects=self.updateObjects) - - self.updateThread.init() - self.expireThread.init() - - def start(self): - self.updateThread.start() - - if self.expireObjects: - self.expireThread.start() - - if self.pollRegistrations: - self.registrationThread.start() - - def stop(self): - self.updateThread.stop() - - if self.expireObjects: - self.expireThread.stop() - - if self.pollRegistrations: - self.registrationThread.stop() - - def callMethod(self, brokerId, objId, classKey, methodName, callback, args): - self.lock() - try: - broker = self.mintBrokersById[brokerId] - finally: - self.unlock() - - seq = self.qmfSession._sendMethodRequest \ - (broker.qmfBroker, ClassKey(classKey), objId, methodName, args) - - if seq is not None: - self.lock() - try: - self.outstandingMethodCalls[seq] = callback - finally: - self.unlock() - return seq - - def addBroker(self, url): - log.info("Adding qmf broker at %s", url) - - self.lock() - try: - qbroker = self.qmfSession.addBroker(url) - mbroker = MintBroker(url, qbroker) - - # Can't add the by-id mapping here, as the id is not set yet; - # instead we add it when brokerConnected is called - - self.mintBrokersByQmfBroker[qbroker] = mbroker - self.mintBrokersByUrl[url] = mbroker - - return mbroker - finally: - self.unlock() - - def delBroker(self, mbroker): - assert isinstance(mbroker, MintBroker) - - log.info("Removing qmf broker at %s", mbroker.url) - - self.lock() - try: - self.qmfSession.delBroker(mbroker.qmfBroker) - - del self.mintBrokersByQmfBroker[mbroker.qmfBroker] - del self.mintBrokersByUrl[mbroker.url] - - if mbroker.qmfId: - del self.mintBrokersById[mbroker.qmfId] - finally: - self.unlock() - - def brokerConnected(self, qbroker): - """ Invoked when a connection is established to a broker """ - self.lock() - try: - mbroker = self.mintBrokersByQmfBroker[qbroker] - - assert mbroker.connected is False - - mbroker.connected = True - finally: - self.unlock() - - def brokerInfo(self, qbroker): - self.lock() - try: - id = str(qbroker.getBrokerId()) - mbroker = self.mintBrokersByQmfBroker[qbroker] - - mbroker.qmfId = id - self.mintBrokersById[id] = mbroker - finally: - self.unlock() - - def brokerDisconnected(self, qbroker): - """ Invoked when the connection to a broker is lost """ - self.lock() - try: - mbroker = self.mintBrokersByQmfBroker[qbroker] - - assert mbroker.connected is True - - mbroker.connected = False - finally: - self.unlock() - - def getMintBrokerByQmfBroker(self, qbroker): - self.lock() - try: - return self.mintBrokersByQmfBroker[qbroker] - finally: - self.unlock() - - def newPackage(self, name): - """ Invoked when a QMF package is discovered. """ - pass - - def newClass(self, kind, classKey): - """ Invoked when a new class is discovered. Session.getSchema can be - used to obtain details about the class.""" - pass - - def newAgent(self, agent): - """ Invoked when a QMF agent is discovered. """ - pass - - def delAgent(self, agent): - """ Invoked when a QMF agent disconects. """ - pass - - def objectProps(self, broker, record): - """ Invoked when an object is updated. """ - - if not self.updateObjects: - return - - mbroker = self.getMintBrokerByQmfBroker(broker) - - up = update.PropertyUpdate(self, mbroker, record) - - if record.getClassKey().getClassName() == "job": - up.priority = 1 - - self.updateThread.enqueue(up) - - def objectStats(self, broker, record): - """ Invoked when an object is updated. """ - - if not self.updateObjects: - return - - if record.getClassKey().getClassName() == "job": - return - - mbroker = self.getMintBrokerByQmfBroker(broker) - up = update.StatisticUpdate(self, mbroker, record) - - self.updateThread.enqueue(up) - - def event(self, broker, event): - """ Invoked when an event is raised. """ - pass - - def heartbeat(self, agent, timestamp): - key = (agent.getBrokerBank(), agent.getAgentBank()) - timestamp = timestamp / 1000000000 - self.heartbeatsByAgentId[key] = datetime.fromtimestamp(timestamp) - - def getLatestHeartbeat(self, scopeId): - scope = int(scopeId) - broker = (scope & 0x0000FFFFF0000000) >> 28 - agent = scope & 0x000000000FFFFFFF - key = (broker, agent) - - return self.heartbeatsByAgentId.get(key) - - def methodResponse(self, broker, seq, response): - mbroker = self.getMintBrokerByQmfBroker(broker) - up = update.MethodUpdate(self, mbroker, seq, response) - - self.updateThread.enqueue(up) - -class RegistrationThread(MintDaemonThread): - def run(self): - log.info("Polling for registration changes every 5 seconds") - - while True: - try: - if self.stopRequested: - break - - self.do_run() - except OperationalError, e: - log.exception(e) - if str(e).find("server closed the connection unexpectedly") >= 0: - sys.exit() - except Exception, e: - log.exception(e) - - def do_run(self): - regUrls = set() - - for reg in BrokerRegistration.select(): - if reg.url not in self.model.mintBrokersByUrl: - try: - self.model.addBroker(reg.url) - except socket.error, e: - log.info("Can't connect to broker at %s: %s", reg.url, e) - pass - - regUrls.add(reg.url) - - for mbroker in self.model.mintBrokersByQmfBroker.values(): - if mbroker.url not in regUrls: - self.model.delBroker(mbroker) - - sleep(5) +from main import * +from model import * Added: mgmt/trunk/mint/python/mint/database.py =================================================================== --- mgmt/trunk/mint/python/mint/database.py (rev 0) +++ mgmt/trunk/mint/python/mint/database.py 2009-04-02 19:49:59 UTC (rev 3253) @@ -0,0 +1,145 @@ +from sqlobject import connectionForURI, sqlhub + +from model import MintInfo, Role + +class MintDatabase(object): + def __init__(self, app): + self.app = app + + def getConnection(self): + return connectionForURI(self.app.config.data).getConnection() + + def check(self): + self.checkConnection() + + def init(self): + sqlhub.processConnection = connectionForURI(self.app.config.data) + + def checkConnection(self): + conn = self.getConnection() + + try: + cursor = conn.cursor() + cursor.execute("select now()") + finally: + conn.close() + + def checkSchema(self): + pass + + def dropSchema(self): + conn = self.getConnection() + + try: + cursor = conn.cursor() + + cursor.execute("drop schema public cascade") + + conn.commit() + finally: + conn.close() + + def __splitSQLStatements(self, text): + result = list() + unmatchedQuote = False + tmpStmt = "" + + for stmt in text.split(";"): + stmt = stmt.rstrip() + quotePos = stmt.find("'") + while quotePos > 0: + quotePos += 1 + if quotePos < len(stmt): + if stmt[quotePos] != "'": + unmatchedQuote = not unmatchedQuote + else: + # ignore 2 single quotes + quotePos += 1 + quotePos = stmt.find("'", quotePos) + + if len(stmt.lstrip()) > 0: + tmpStmt += stmt + ";" + if not unmatchedQuote: + # single quote has been matched/closed, generate statement + result.append(tmpStmt.lstrip()) + tmpStmt = "" + + if unmatchedQuote: + result.append(tmpStmt.lstrip()) + return result + + def createSchema(self, file_paths): + conn = self.getConnection() + + scripts = list() + + for path in file_paths: + file = open(path, "r") + try: + scripts.append((path, file.read())) + finally: + file.close() + + try: + cursor = conn.cursor() + + try: + cursor.execute("create schema public"); + except: + conn.rollback() + pass + + for path, text in scripts: + stmts = self.__splitSQLStatements(text) + 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() + + 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() Deleted: mgmt/trunk/mint/python/mint/dbexpire.py =================================================================== --- mgmt/trunk/mint/python/mint/dbexpire.py 2009-04-02 19:42:25 UTC (rev 3252) +++ mgmt/trunk/mint/python/mint/dbexpire.py 2009-04-02 19:49:59 UTC (rev 3253) @@ -1,58 +0,0 @@ -import logging -import mint -import time -from threading import Thread -from mint.schema import * -from sql import * -from util import * - -log = logging.getLogger("mint.dbexpire") - -class DBExpireThread(MintDaemonThread): - def __init__(self, model): - super(DBExpireThread, self).__init__(model) - - self.keepCurrStats = False - - self.ops = [] - self.attrs = dict() - - def init(self): - frequency = self.model.expireFrequency - threshold = self.model.expireThreshold - - for cls in mint.schema.statsClasses: - self.ops.append(SqlExpire(eval(cls), self.keepCurrStats)) - self.ops.append(SqlExpire(Job, self.keepCurrStats)) - - self.attrs["threshold"] = threshold - - frequency_out, frequency_unit = self.__convertTimeUnits(frequency) - threshold_out, threshold_unit = self.__convertTimeUnits(threshold) - log.debug("Expiring database records older than %d %s, every %d %s" % \ - (threshold_out, threshold_unit, frequency_out, frequency_unit)) - - def run(self): - frequency = self.model.expireFrequency - - while True: - if self.stopRequested: - break - up = mint.update.DBExpireUpdate(self.model) - self.model.updateThread.enqueue(up) - time.sleep(frequency) - - def __convertTimeUnits(self, t): - if t / (24*3600) >= 1: - t_out = t / (24*3600) - t_unit = "days" - elif t / 3600 >= 1: - t_out = t / 3600 - t_unit = "hours" - elif t / 60 >= 1: - t_out = t / 60 - t_unit = "minutes" - else: - t_out = t - t_unit = "seconds" - return (t_out, t_unit) Copied: mgmt/trunk/mint/python/mint/expire.py (from rev 3248, mgmt/trunk/mint/python/mint/dbexpire.py) =================================================================== --- mgmt/trunk/mint/python/mint/expire.py (rev 0) +++ mgmt/trunk/mint/python/mint/expire.py 2009-04-02 19:49:59 UTC (rev 3253) @@ -0,0 +1,58 @@ +import logging +import mint +import time +from threading import Thread +from mint.schema import * +from sql import * +from util import * + +log = logging.getLogger("mint.expire") + +class ExpireThread(MintDaemonThread): + def __init__(self, app): + super(ExpireThread, self).__init__(app) + + self.keepCurrStats = False + + self.ops = [] + self.attrs = dict() + + def init(self): + frequency = self.app.expireFrequency + threshold = self.app.expireThreshold + + for cls in mint.schema.statsClasses: + self.ops.append(SqlExpire(eval(cls), self.keepCurrStats)) + self.ops.append(SqlExpire(Job, self.keepCurrStats)) + + self.attrs["threshold"] = threshold + + frequency_out, frequency_unit = self.__convertTimeUnits(frequency) + threshold_out, threshold_unit = self.__convertTimeUnits(threshold) + log.debug("Expiring database records older than %d %s, every %d %s" % \ + (threshold_out, threshold_unit, frequency_out, frequency_unit)) + + def run(self): + frequency = self.app.expireFrequency + + while True: + if self.stopRequested: + break + up = mint.update.ExpireUpdate(self.app.model) + self.app.updateThread.enqueue(up) + time.sleep(frequency) + + def __convertTimeUnits(self, t): + if t / (24*3600) >= 1: + t_out = t / (24*3600) + t_unit = "days" + elif t / 3600 >= 1: + t_out = t / 3600 + t_unit = "hours" + elif t / 60 >= 1: + t_out = t / 60 + t_unit = "minutes" + else: + t_out = t + t_unit = "seconds" + return (t_out, t_unit) Property changes on: mgmt/trunk/mint/python/mint/expire.py ___________________________________________________________________ Name: svn:mergeinfo + Name: svn:eol-style + native Modified: mgmt/trunk/mint/python/mint/main.py =================================================================== --- mgmt/trunk/mint/python/mint/main.py 2009-04-02 19:42:25 UTC (rev 3252) +++ mgmt/trunk/mint/python/mint/main.py 2009-04-02 19:49:59 UTC (rev 3253) @@ -1,30 +1,77 @@ import sys import os +import logging from parsley.config import Config, ConfigParameter from parsley.loggingex import enable_logging -from mint import MintModel +from database import MintDatabase +from model import MintModel +from update import UpdateThread +from poll import PollThread +from expire import ExpireThread +log = logging.getLogger("mint.main") + class Mint(object): def __init__(self, config): self.config = config + self.database = MintDatabase(self) + self.model = MintModel(self) - self.model = MintModel(self.config.data) - self.model.expireFrequency = self.config.expire_frequency - self.model.expireThreshold = self.config.expire_threshold + self.updateEnabled = True + self.updateThread = UpdateThread(self) + self.pollEnabled = False + self.pollThread = PollThread(self) + + self.expireEnabled = True + self.expireFrequency = self.config.expire_frequency + self.expireThreshold = self.config.expire_threshold + self.expireThread = ExpireThread(self) + def check(self): + self.database.check() self.model.check() def init(self): + self.database.init() self.model.init() + def state(cond): + return cond and "enabled" or "disabled" + + log.info("Updates are %s", state(self.updateEnabled)) + log.info("Polling is %s", state(self.pollEnabled)) + log.info("Expiration is %s", state(self.expireEnabled)) + + self.updateThread.init() + self.pollThread.init() + self.expireThread.init() + def start(self): self.model.start() + if self.updateEnabled: + self.updateThread.start() + + if self.pollEnabled: + self.pollThread.start() + + if self.expireEnabled: + self.expireThread.start() + def stop(self): self.model.stop() + if self.updateEnabled: + self.updateThread.stop() + + if self.pollEnabled: + self.pollThread.stop() + + if self.expireEnabled: + self.expireThread.stop() + class MintConfig(Config): def __init__(self): super(MintConfig, self).__init__() @@ -35,6 +82,9 @@ param = ConfigParameter(self, "data", str) param.default = "postgresql://mint@localhost/mint" + param = ConfigParameter(self, "qmf", str) + param.default = "amqp://localhost" + param = ConfigParameter(self, "log-file", str) param.default = os.path.join(self.home, "log", "mint.log") @@ -50,13 +100,15 @@ param = ConfigParameter(self, "expire-threshold", int) param.default = 24 * 3600 # 1 day - def init(self, opts): + def init(self, opts=None): super(MintConfig, self).init() self.load_file(os.path.join(self.home, "etc", "mint.conf")) self.load_file(os.path.join(os.path.expanduser("~"), ".mint.conf")) - self.load_dict(opts) + if opts: + self.load_dict(opts) + if self.debug: enable_logging("mint", "debug", sys.stderr) else: Added: mgmt/trunk/mint/python/mint/model.py =================================================================== --- mgmt/trunk/mint/python/mint/model.py (rev 0) +++ mgmt/trunk/mint/python/mint/model.py 2009-04-02 19:49:59 UTC (rev 3253) @@ -0,0 +1,397 @@ +import logging +import qmf.console +import pickle +import struct +import sys +import os +import types +import socket + +from qmf.console import ClassKey +from qpid.datatypes import UUID +from qpid.util import URL +from sqlobject import * +from threading import Lock, RLock +from time import sleep +from traceback import print_exc + +from mint import update +from mint.cache import MintCache +from mint.schema import * +from util import * + +log = logging.getLogger("mint.model") + +thisModule = __import__(__name__) + +for item in dir(mint.schema): + cls = getattr(mint.schema, item) + + try: + if issubclass(cls, SQLObject) and cls is not SQLObject: + setattr(thisModule, item, cls) + except TypeError: + pass + +Broker.sqlmeta.addColumn(ForeignKey("BrokerRegistration", + cascade="null", default=None, + name="registration")) + +class Subject(SQLObject): + class sqlmeta: + lazyUpdate = True + + name = StringCol(length=1000, default=None, unique=True, notNone=True) + password = StringCol(length=1000, default=None) + lastChallenged = TimestampCol(default=None) + lastLoggedIn = TimestampCol(default=None) + lastLoggedOut = TimestampCol(default=None) + roles = SQLRelatedJoin("Role", intermediateTable="subject_role_mapping", + createRelatedTable=False) + + def getByName(cls, name): + try: + return Subject.selectBy(name=name)[0] + except IndexError: + pass + + getByName = classmethod(getByName) + +class Role(SQLObject): + class sqlmeta: + lazyUpdate = True + + name = StringCol(length=1000, default=None, unique=True, notNone=True) + subjects = SQLRelatedJoin("Subject", + intermediateTable="subject_role_mapping", + createRelatedTable=False) + + def getByName(cls, name): + try: + return Role.selectBy(name=name)[0] + except IndexError: + pass + + getByName = classmethod(getByName) + +class SubjectRoleMapping(SQLObject): + class sqlmeta: + lazyUpdate = True + + subject = ForeignKey("Subject", notNull=True, cascade=True) + role = ForeignKey("Role", notNull=True, cascade=True) + unique = DatabaseIndex(subject, role, unique=True) + +class ObjectNotFound(Exception): + pass + +class MintInfo(SQLObject): + class sqlmeta: + lazyUpdate = True + + version = StringCol(length=1000, default="0.1", notNone=True) + +class CollectorRegistration(SQLObject): + class sqlmeta: + lazyUpdate = True + + name = StringCol(length=1000, default=None) + collectorId = StringCol(length=1000, default=None) + +class BrokerRegistration(SQLObject): + class sqlmeta: + lazyUpdate = True + + name = StringCol(length=1000, default=None, unique=True, notNone=True) + url = StringCol(length=1000, default=None) + broker = ForeignKey("Broker", cascade="null", default=None) + cluster = ForeignKey("BrokerCluster", cascade="null", default=None) + profile = ForeignKey("BrokerProfile", cascade="null", default=None) + + url_unique = DatabaseIndex(url, unique=True) + + def getBrokerId(self): + return self.mintBroker.qmfId + + def getDefaultVhost(self): + if self.broker: + try: + return Vhost.selectBy(broker=self.broker, name="/")[0] + except IndexError: + return None + +class BrokerGroup(SQLObject): + class sqlmeta: + lazyUpdate = True + + name = StringCol(length=1000, default=None, unique=True, notNone=True) + brokers = SQLRelatedJoin("Broker", + intermediateTable="broker_group_mapping", + createRelatedTable=False) + +class BrokerGroupMapping(SQLObject): + class sqlmeta: + lazyUpdate = True + + broker = ForeignKey("Broker", notNull=True, cascade=True) + brokerGroup = ForeignKey("BrokerGroup", notNull=True, cascade=True) + unique = DatabaseIndex(broker, brokerGroup, unique=True) + +class BrokerCluster(SQLObject): + class sqlmeta: + lazyUpdate = True + + name = StringCol(length=1000, default=None) + brokers = SQLMultipleJoin("BrokerRegistration", joinColumn="cluster_id") + +class BrokerProfile(SQLObject): + class sqlmeta: + lazyUpdate = True + + name = StringCol(length=1000, default=None) + brokers = SQLMultipleJoin("BrokerRegistration", joinColumn="profile_id") + properties = SQLMultipleJoin("ConfigProperty", joinColumn="profile_id") + +class MintBroker(object): + def __init__(self, url, qmfBroker): + self.url = url + self.qmfBroker = qmfBroker + + self.qmfId = None + self.databaseId = None + + self.objectDatabaseIds = MintCache() # database ids by qmf object id + self.orphans = dict() # updates by qmf object id + + self.connected = False + + def __repr__(self): + return "%s(%s)" % (self.__class__.__name__, self.url) + + def getAmqpSession(self): + return self.qmfBroker.getAmqpSession() + + def getAmqpSessionId(self): + return self.qmfBroker.getSessionId() + + def getFullUrl(self): + return self.qmfBroker.getFullUrl() + +class MintModel(qmf.console.Console): + staticInstance = None + + def __init__(self, app): + assert MintModel.staticInstance is None + MintModel.staticInstance = self + + self.app = app + + # Lookup tables used for recovering MintBroker objects, which have + # mint-specific accounting and wrap a qmf broker object + + self.mintBrokersByQmfBroker = dict() + self.mintBrokersById = dict() + self.mintBrokersByUrl = dict() + + # Agent heartbeats + # (broker bank, agent bank) => latest heartbeat timestamp + + self.heartbeatsByAgentId = dict() + + self.__lock = RLock() + + self.dbConn = None + self.qmfSession = None + + self.outstandingMethodCalls = dict() + + def lock(self): + self.__lock.acquire() + + def unlock(self): + self.__lock.release() + + def check(self): + pass + + def init(self): + assert self.qmfSession is None + + self.qmfSession = qmf.console.Session \ + (self, manageConnections=True, rcvObjects=self.app.updateEnabled) + + def start(self): + pass + + def stop(self): + for mbroker in self.mintBrokersById.values(): + self.delBroker(mbroker) + + def callMethod(self, brokerId, objId, classKey, methodName, callback, args): + self.lock() + try: + broker = self.mintBrokersById[brokerId] + finally: + self.unlock() + + seq = self.qmfSession._sendMethodRequest \ + (broker.qmfBroker, ClassKey(classKey), objId, methodName, args) + + if seq is not None: + self.lock() + try: + self.outstandingMethodCalls[seq] = callback + finally: + self.unlock() + return seq + + def addBroker(self, url): + log.info("Adding qmf broker at %s", url) + + self.lock() + try: + qbroker = self.qmfSession.addBroker(url) + mbroker = MintBroker(url, qbroker) + + # Can't add the by-id mapping here, as the id is not set yet; + # instead we add it when brokerConnected is called + + self.mintBrokersByQmfBroker[qbroker] = mbroker + self.mintBrokersByUrl[url] = mbroker + + return mbroker + finally: + self.unlock() + + def delBroker(self, mbroker): + assert isinstance(mbroker, MintBroker) + + log.info("Removing qmf broker at %s", mbroker.url) + + self.lock() + try: + self.qmfSession.delBroker(mbroker.qmfBroker) + + del self.mintBrokersByQmfBroker[mbroker.qmfBroker] + del self.mintBrokersByUrl[mbroker.url] + + if mbroker.qmfId: + del self.mintBrokersById[mbroker.qmfId] + finally: + self.unlock() + + def brokerConnected(self, qbroker): + """ Invoked when a connection is established to a broker """ + self.lock() + try: + mbroker = self.mintBrokersByQmfBroker[qbroker] + + assert mbroker.connected is False + + mbroker.connected = True + finally: + self.unlock() + + def brokerInfo(self, qbroker): + self.lock() + try: + id = str(qbroker.getBrokerId()) + mbroker = self.mintBrokersByQmfBroker[qbroker] + + mbroker.qmfId = id + self.mintBrokersById[id] = mbroker + finally: + self.unlock() + + def brokerDisconnected(self, qbroker): + """ Invoked when the connection to a broker is lost """ + self.lock() + try: + mbroker = self.mintBrokersByQmfBroker[qbroker] + + assert mbroker.connected is True + + mbroker.connected = False + finally: + self.unlock() + + def getMintBrokerByQmfBroker(self, qbroker): + self.lock() + try: + return self.mintBrokersByQmfBroker[qbroker] + finally: + self.unlock() + + def newPackage(self, name): + """ Invoked when a QMF package is discovered. """ + pass + + def newClass(self, kind, classKey): + """ Invoked when a new class is discovered. Session.getSchema can be + used to obtain details about the class.""" + pass + + def newAgent(self, agent): + """ Invoked when a QMF agent is discovered. """ + pass + + def delAgent(self, agent): + """ Invoked when a QMF agent disconects. """ + pass + + def objectProps(self, broker, record): + """ Invoked when an object is updated. """ + + if not self.app.updateThread.isAlive(): + return + + mbroker = self.getMintBrokerByQmfBroker(broker) + + up = update.PropertyUpdate(self, mbroker, record) + + if record.getClassKey().getClassName() == "job": + up.priority = 1 + + self.app.updateThread.enqueue(up) + + def objectStats(self, broker, record): + """ Invoked when an object is updated. """ + + if not self.app.updateThread.isAlive(): + return + + if record.getClassKey().getClassName() == "job": + return + + mbroker = self.getMintBrokerByQmfBroker(broker) + up = update.StatisticUpdate(self, mbroker, record) + + self.app.updateThread.enqueue(up) + + def event(self, broker, event): + """ Invoked when an event is raised. """ + pass + + def heartbeat(self, agent, timestamp): + key = (agent.getBrokerBank(), agent.getAgentBank()) + timestamp = timestamp / 1000000000 + self.heartbeatsByAgentId[key] = datetime.fromtimestamp(timestamp) + + def getLatestHeartbeat(self, scopeId): + scope = int(scopeId) + broker = (scope & 0x0000FFFFF0000000) >> 28 + agent = scope & 0x000000000FFFFFFF + key = (broker, agent) + + return self.heartbeatsByAgentId.get(key) + + def methodResponse(self, broker, seq, response): + # XXX don't do this via the update thread? + + if not self.updateThread.isAlive(): + return + + mbroker = self.getMintBrokerByQmfBroker(broker) + up = update.MethodUpdate(self, mbroker, seq, response) + + self.app.updateThread.enqueue(up) Added: mgmt/trunk/mint/python/mint/poll.py =================================================================== --- mgmt/trunk/mint/python/mint/poll.py (rev 0) +++ mgmt/trunk/mint/python/mint/poll.py 2009-04-02 19:49:59 UTC (rev 3253) @@ -0,0 +1,46 @@ +import logging +from time import sleep +from psycopg2 import OperationalError +from mint.model import BrokerRegistration + +from util import MintDaemonThread + +log = logging.getLogger("mint.poll") + +class PollThread(MintDaemonThread): + interval = 5 + + def run(self): + log.info("Polling for changes every %i seconds", self.interval) + + while True: + try: + if self.stopRequested: + break + + self.do_run() + except OperationalError, e: + log.exception(e) + if str(e).find("server closed the connection unexpectedly") >= 0: + sys.exit() # XXX This seems like the wrong thing to do + except Exception, e: + log.exception(e) + + sleep(self.interval) + + def do_run(self): + regUrls = set() + + for reg in BrokerRegistration.select(): + if reg.url not in self.app.model.mintBrokersByUrl: + try: + self.app.model.addBroker(reg.url) + except socket.error, e: + log.info("Can't connect to broker at %s: %s", reg.url, e) + + regUrls.add(reg.url) + + for mbroker in self.app.model.mintBrokersByQmfBroker.values(): + if mbroker.url not in regUrls: + self.app.model.delBroker(mbroker) + Modified: mgmt/trunk/mint/python/mint/tools.py =================================================================== --- mgmt/trunk/mint/python/mint/tools.py 2009-04-02 19:42:25 UTC (rev 3252) +++ mgmt/trunk/mint/python/mint/tools.py 2009-04-02 19:49:59 UTC (rev 3253) @@ -89,6 +89,9 @@ def do_run(self, opts, args): app = Mint(self.config) + + app.pollEnabled = True + app.check() app.init() app.start() @@ -109,24 +112,17 @@ def do_run(self, opts, args): app = Mint(self.config) - app.model.pollRegistrations = False - app.check() app.init() app.start() - added = list() - try: for arg in args[1:]: - added.append(app.model.addBroker(arg)) + app.model.addBroker(arg) while True: sleep(2) finally: - for broker in added: - app.model.delBroker(broker) - app.stop() class MintBenchTool(BaseMintTool): @@ -142,14 +138,10 @@ def do_run(self, opts, args): app = Mint(self.config) - app.model.pollRegistrations = False - app.check() app.init() app.start() - added = list() - head = "%6s %6s %6s %6s %6s %6s %6s %6s %6s" % \ ("enqs", "deqs", "depth", "drop", "defer", "prop", "stat", "meth", "exp") @@ -158,7 +150,7 @@ try: for arg in args[1:]: try: - added.append(app.model.addBroker(arg)) + app.model.addBroker(arg) except socket.error, e: print "Warning: Failed connecting to broker at '%s'" % arg @@ -182,7 +174,7 @@ sleep(1) - ut = app.model.updateThread + ut = app.updateThread enq = ut.enqueueCount deq = ut.dequeueCount @@ -228,9 +220,5 @@ finally: #from threading import enumerate #for item in enumerate(): - # print item - for broker in added: - app.model.delBroker(broker) - app.stop() Modified: mgmt/trunk/mint/python/mint/update.py =================================================================== --- mgmt/trunk/mint/python/mint/update.py 2009-04-02 19:42:25 UTC (rev 3252) +++ mgmt/trunk/mint/python/mint/update.py 2009-04-02 19:49:59 UTC (rev 3253) @@ -17,10 +17,14 @@ log = logging.getLogger("mint.update") -class ModelUpdateThread(MintDaemonThread): - def __init__(self, model): - super(ModelUpdateThread, self).__init__(model) +class UpdateThread(MintDaemonThread): + """ + Only the update thread writes to the database + """ + def __init__(self, app): + super(UpdateThread, self).__init__(app) + self.updates = UpdateQueue(slotCount=2) self.enqueueCount = 0 @@ -36,7 +40,7 @@ self.conn = None def init(self): - self.conn = self.model.dbConn.getConnection() + self.conn = self.app.database.getConnection() def enqueue(self, update): try: @@ -71,7 +75,7 @@ continue self.process_update(update) - + def process_update(self, update): try: update.process(self) @@ -93,13 +97,13 @@ def commit(self): self.conn.commit() - for broker in self.model.mintBrokersByQmfBroker.values(): + for broker in self.app.model.mintBrokersByQmfBroker.values(): broker.objectDatabaseIds.commit() def rollback(self): self.conn.rollback() - for broker in self.model.mintBrokersByQmfBroker.values(): + for broker in self.app.model.mintBrokersByQmfBroker.values(): broker.objectDatabaseIds.rollback() class ReferenceException(Exception): @@ -214,10 +218,15 @@ self.processTimestamp("qmfUpdateTime", timestamps[0], attrs) - if cls is Job and attrs["qmfUpdateTime"] < datetime.now() - timedelta(seconds=thread.model.expireThread.threshold): - # drop updates for Jobs that are older then the expiration threshold, - # since they would be deleted anyway in the next run of the db expiration thread + delta = timedelta(seconds=thread.app.expireThreshold) + + if cls is Job and attrs["qmfUpdateTime"] < datetime.now() - delta: + # drop updates for Jobs that are older then the expiration + # threshold, since they would be deleted anyway in the next run + # of the db expiration thread + log.debug("Property update is older than expiration threshold; skipping it") + thread.dropCount += 1 return @@ -384,18 +393,18 @@ thread.methUpdateCount += 1 -class DBExpireUpdate(ModelUpdate): +class ExpireUpdate(ModelUpdate): def __init__(self, model): - super(DBExpireUpdate, self).__init__(model, None, None) + super(ExpireUpdate, self).__init__(model, None, None) def process(self, thread): cursor = thread.cursor() - attrs = self.model.expireThread.attrs + attrs = thread.app.expireThread.attrs total = 0 thread.commit() - for op in self.model.expireThread.ops: + for op in thread.app.expireThread.ops: log.debug("Running expire op %s", op) count = op.execute(cursor, attrs) Modified: mgmt/trunk/mint/python/mint/util.py =================================================================== --- mgmt/trunk/mint/python/mint/util.py 2009-04-02 19:42:25 UTC (rev 3252) +++ mgmt/trunk/mint/python/mint/util.py 2009-04-02 19:49:59 UTC (rev 3253) @@ -5,15 +5,18 @@ log = logging.getLogger("mint") class MintDaemonThread(Thread): - def __init__(self, model): + def __init__(self, app): super(MintDaemonThread, self).__init__() - self.model = model + self.app = app self.stopRequested = False self.setDaemon(True) + def init(self): + pass + def stop(self): assert self.stopRequested is False self.stopRequested = True