r5455 - in branches/tmckay/mint/python/mint: . plumage
tmckay at fedoraproject.org
tmckay at fedoraproject.org
Tue Sep 4 18:15:21 UTC 2012
Author: tmckay
Date: 2012-09-04 18:15:20 +0000 (Tue, 04 Sep 2012)
New Revision: 5455
Modified:
branches/tmckay/mint/python/mint/main.py
branches/tmckay/mint/python/mint/plumage/main.py
branches/tmckay/mint/python/mint/plumage/session.py
branches/tmckay/mint/python/mint/plumage/update.py
branches/tmckay/mint/python/mint/session.py
branches/tmckay/mint/python/mint/update.py
Log:
Move the package filter and set up the list of classes to bind in the main
application. Make the class list canonical, even if only a package list
was initially specified. Reformat some lines for length in plumage.
Modified: branches/tmckay/mint/python/mint/main.py
===================================================================
--- branches/tmckay/mint/python/mint/main.py 2012-08-31 17:58:10 UTC (rev 5454)
+++ branches/tmckay/mint/python/mint/main.py 2012-09-04 18:15:20 UTC (rev 5455)
@@ -2,6 +2,7 @@
from expire import ExpireThread
from model import MintModel
from session import MintSession
+from aviary.session import MintAviarySession
from update import UpdateThread
from vacuum import VacuumThread
from cumin.admin import *
@@ -17,6 +18,8 @@
self.database = MintDatabase(self, database_dsn)
self.admin = CuminAdmin(self)
self.update_thread = UpdateThread(self)
+ self.session = None
+ self.aviary_session = None
self.expire_enabled = True
self.expire_thread = ExpireThread(self)
@@ -30,15 +33,19 @@
# mechanisms, according to the sasl documentation
self.sasl_mech_list = None
- # List of classes to bind. Referenced by session and update thread.
- self.qmf_classes = set()
+ # List of classes to bind in QMF. Referenced by session and update thread.
+ self._qmf_classes = set()
- # If binding was not done by class, this is the final list of
- # packages that were bound
- self.qmf_packages = set()
+ # List of classes to gather via Aviary. Derived from self._qmf_classes
+ # in combination with the list of classes supported by Aviary.
+ self._aviary_classes = set()
+ # If self._qmf_classes is not set, this is the list of packages
+ # to bind. This will be translated into self.qmf_classes.
+ self._qmf_packages = set()
+
# Things we know should not be bound.
- self.qmf_package_filter = ["com.redhat.cumin", "com.redhat.cumin.grid"]
+ self._qmf_package_filter = ["com.redhat.cumin", "com.redhat.cumin.grid"]
# Aviary interface.
self.aviary_query_servers = ""
@@ -53,8 +60,11 @@
self.aviary_prefer_condor = True
def set_classes(self, classes):
- self.qmf_classes = classes
+ self._qmf_classes = classes
+ def get_bound_classes(self):
+ return self._aviary_classes.union(self._qmf_classes)
+
def check(self):
log.info("Checking %s", self)
@@ -71,19 +81,39 @@
log.info("Expiration is %s", state(self.expire_enabled))
log.info("Vacuum is %s", state(self.vacuum_enabled))
+ # If self._qmf_classes has not been set, then build classes
+ # here from the package list. We want to be canonical, ie
+ # always have things expressed in terms of classes
+ if not self._qmf_classes:
+ if not self._qmf_packages:
+ self._qmf_packages = self.model._packages
+ for pkg in self._qmf_packages:
+ if pkg._name not in self._qmf_package_filter:
+ self._qmf_classes = self._qmf_classes.union(set(pkg._classes))
+ else:
+ # Apply the package filter to the specified class list
+ black_list = set()
+ for cls in self._qmf_classes:
+ if cls._package._name in self._qmf_package_filter:
+ black_list.add(cls)
+ self._qmf_classes.difference_update(black_list)
+
# If aviary_query_servers or aviary_locator is defined
- # here, then we are going to get submissions from Aviary
+ # here, then we are going to get some classes from Aviary
# instead of QMF. Initialize a MintAviary object and
- # eliminate submissions from the set of things bound by
+ # eliminate classes from the set of things bound by
# the QMF interface.
- # For now, bind by class must be in effect or this option
- # cannot be supported (but bind by class is normal).
- if (self.aviary_query_servers or self.aviary_locator) \
- and self.model.com_redhat_grid.Submission in self.qmf_classes:
- self.qmf_classes.remove(self.model.com_redhat_grid.Submission)
+ if self.aviary_query_servers or self.aviary_locator:
+ supported_classes = [self.model.com_redhat_grid.Submission]
+ for c in supported_classes:
+ if c in self._qmf_classes:
+ self._qmf_classes.remove(c)
+ self._aviary_classes.add(c)
+ if self._aviary_classes:
+ self.aviary_session = MintAviarySession(self,
+ self._aviary_classes)
- self.session = MintSession(self, self.broker_uris)
- self.session.init()
+ self.session = MintSession(self, self.broker_uris, self._qmf_classes)
self.database.init()
self.update_thread.init()
@@ -92,14 +122,12 @@
def start(self):
log.info("Starting %s", self)
-
- # Scan the qmf class/package binding list
- # and do any necessary preprocessing
- self.session.init_qmf_classes()
self.update_thread.start()
self.session.start()
+ if self.aviary_session:
+ self.aviary_session.start()
if self.expire_enabled:
self.expire_thread.start()
@@ -119,6 +147,8 @@
self.vacuum_thread.stop()
self.session.stop()
+ if self.aviary_session:
+ self.aviary_session.stop()
log.info("Session stopped")
def __repr__(self):
Modified: branches/tmckay/mint/python/mint/plumage/main.py
===================================================================
--- branches/tmckay/mint/python/mint/plumage/main.py 2012-08-31 17:58:10 UTC (rev 5454)
+++ branches/tmckay/mint/python/mint/plumage/main.py 2012-09-04 18:15:20 UTC (rev 5455)
@@ -35,15 +35,18 @@
self.print_event_level = 0
- self.packages = set()
+ self._packages = set()
- self.classes = set()
+ self._classes = set()
- self.package_filter = ["com.redhat.cumin"]
+ self._package_filter = ["com.redhat.cumin"]
def set_classes(self, classes):
self.classes = classes
+ def get_bound_classes(self):
+ return self._classes
+
def check(self):
log.info("Checking %s", self)
@@ -63,11 +66,26 @@
self.database.init()
self.update_thread.init()
- self.session.init()
- # The package and class lists will be
- # processed here
- self.session.init_classes()
+ # If self._qmf_classes has not been set, then build classes
+ # here from the package list. We want to be canonical, ie
+ # always have things expressed in terms of classes
+ if not self._classes:
+ if not self._packages:
+ self._packages = self.model._packages
+ for pkg in self._packages:
+ if pkg._name not in self._package_filter:
+ self._classes = self._classes.union(set(pkg._classes))
+ else:
+ # Apply the package filter to the specified class list
+ black_list = set()
+ for cls in self._classes:
+ if cls._package._name in self._package_filter:
+ black_list.add(cls)
+ self._classes.difference_update(black_list)
+
+ self.session.init(self._classes)
+
if self.expire_enabled:
self.expire_thread.init()
Modified: branches/tmckay/mint/python/mint/plumage/session.py
===================================================================
--- branches/tmckay/mint/python/mint/plumage/session.py 2012-08-31 17:58:10 UTC (rev 5454)
+++ branches/tmckay/mint/python/mint/plumage/session.py 2012-09-04 18:15:20 UTC (rev 5455)
@@ -56,45 +56,24 @@
def check(self):
log.info("Checking %s", self)
- def init(self):
+ def init(self, classes):
log.info("Initializing %s", self)
-
- def init_classes(self):
if not imports_ok:
return
# Apply the package filter to the class list
- if len(self.app.classes):
- black_list = set()
- for cls in self.app.classes:
- if cls._package._name in self.app.package_filter:
- black_list.add(cls)
- else:
- try:
- loaders = ClassLoaders()
- # Here we grab the name of the method to use (should be in rosemary.xml package/class/loading_class)
- func = getattr(loaders, cls.loading_class, None)
- func(self, cls)
- except Exception, e:
- log.error("No loading function for class %s (%s). Be sure that the method exists and is defined in rosemary.xml" % (cls._name, str(e)))
+ loaders = ClassLoaders()
+ for cls in classes:
+ try:
+ # Here we grab the name of the method to use
+ # (should be in rosemary.xml package/class/loading_class)
+ func = getattr(loaders, cls.loading_class, None)
+ func(self, cls)
+ except Exception, e:
+ log.error("No loading function for class %s (%s). "\
+ "Be sure that the method exists and is "\
+ "defined in rosemary.xml" % (cls._name, str(e)))
- # Update our list, minus the black_list
- self.app.classes.difference_update(black_list)
-
- else:
- # Generate the package list from the model, minus
- # the package filter
- for pkg in self.app.model._packages:
- if pkg._name not in self.app.package_filter:
- self.app.packages.add(pkg)
- loaders = ClassLoaders()
- for cls in pkg._classes:
- try:
- # Here we grab the name of the method to use (should be in rosemary.xml package/class/loading_class)
- func = getattr(loaders, cls.loading_class, None)
- func(self, cls)
- except Exception, e:
- log.error("No loading function for class %s (%s). Be sure that the method exists and is defined in rosemary.xml" % (cls._name, str(e)))
def start(self):
log.info("Starting %s", self)
if not imports_ok:
@@ -109,21 +88,42 @@
t.stop()
def __repr__(self):
- return "%s(%s:%s)" % (self.__class__.__name__, self.server_host, self.server_port)
+ return "%s(%s:%s)" % (self.__class__.__name__,
+ self.server_host, self.server_port)
class ClassLoaders(object):
- ''' method for loading the com.redhat.grid.plumage.OSUtil data from plumage, name is found in rosemary.xml under package/class/loading_class '''
+ ''' method for loading the com.redhat.grid.plumage.OSUtil data from plumage,
+ name is found in rosemary.xml under package/class/loading_class '''
def OSUtilLoader(self, obj, cls):
- obj.threads.append(CatchUpPlumageOSUtilSessionThread(obj.app, obj.server_host, obj.server_port, cls))
- obj.threads.append(PlumageOSUtilSessionThread(obj.app, obj.server_host, obj.server_port, cls))
- obj.threads.append(CurrentPlumageOSUtilSessionThread(obj.app, obj.server_host, obj.server_port, cls))
+ obj.threads.append(CatchUpPlumageOSUtilSessionThread(obj.app,
+ obj.server_host,
+ obj.server_port,
+ cls))
+ obj.threads.append(PlumageOSUtilSessionThread(obj.app,
+ obj.server_host,
+ obj.server_port,
+ cls))
+ obj.threads.append(CurrentPlumageOSUtilSessionThread(obj.app,
+ obj.server_host,
+ obj.server_port,
+ cls))
- ''' method for loading the com.redhat.grid.plumage.Accountant data from plumage, name is found in rosemary.xml under package/class/loading_class '''
+ ''' method for loading the com.redhat.grid.plumage.Accountant data from plumage,
+ name is found in rosemary.xml under package/class/loading_class '''
def AccountantLoader(self, obj, cls):
log.debug("AccountLoader called")
- obj.threads.append(CatchupPlumageAccountantSessionThread(obj.app, obj.server_host, obj.server_port, cls))
- obj.threads.append(PlumageAccountantSessionThread(obj.app, obj.server_host, obj.server_port, cls))
- obj.threads.append(CurrentPlumageAccountantSessionThread(obj.app, obj.server_host, obj.server_port, cls))
+ obj.threads.append(CatchupPlumageAccountantSessionThread(obj.app,
+ obj.server_host,
+ obj.server_port,
+ cls))
+ obj.threads.append(PlumageAccountantSessionThread(obj.app,
+ obj.server_host,
+ obj.server_port,
+ cls))
+ obj.threads.append(CurrentPlumageAccountantSessionThread(obj.app,
+ obj.server_host,
+ obj.server_port,
+ cls))
class PlumageSessionThread(MintDaemonThread):
def __init__(self, app, server_host, server_port, cls):
@@ -209,11 +209,12 @@
self._init()
# We create objects here. Tag them with the right class,
- # probably specified to us from a config option (with a corresponding
- # query specification in the xml)
+ # probably specified to us from a config option
+ # (with a corresponding query specification in the xml)
(oldest, newest) = self.app.update_thread.get_first_and_last_sample_timestamp(self.cls)
if oldest is None:
- # if we have no oldest record (first run), start at "10 min ago" and start loading everything
+ # if we have no oldest record (first run),
+ # start at "10 min ago" and start loading everything
oldest = datetime.now() - timedelta(seconds=600)
oldest = oldest + UTC_DIFF
@@ -293,9 +294,14 @@
obj = ObjectUpdate(self.app.model, record, self.cls)
self.app.update_thread.enqueue(obj)
- log.info("CatchupPlumageAccountantThread--catch-up: catch-up run completed for records newer than %s and older than %s" % (newest, datetime.now() - timedelta(seconds=300) + UTC_DIFF))
+ log.info("CatchupPlumageAccountantThread--catch-up: "\
+ "catch-up run completed for records newer than %s "\
+ "and older than %s" % \
+ (newest,
+ datetime.now() - timedelta(seconds=300) + UTC_DIFF))
else:
- log.info("CatchUpPlumageSessionThread: Skipping catch-up, no records present (probably first-run)")
+ log.info("CatchUpPlumageSessionThread: "\
+ "Skipping catch-up, no records present (probably first-run)")
except Exception, e:
log.info("%s got exception %s, exiting" % (self.__class__.__name__, str(e)))
@@ -435,8 +441,13 @@
obj = ObjectUpdate(self.app.model, record, self.cls)
self.app.update_thread.enqueue(obj)
- log.info("CatchUpPlumageSessionThread--catch-up: catch-up run completed for records newer than %s and older than %s" % (newest, datetime.now() - timedelta(seconds=300) + UTC_DIFF))
+ log.info("CatchUpPlumageSessionThread--catch-up: "\
+ "catch-up run completed for records newer than %s "\
+ "and older than %s" % \
+ (newest,
+ datetime.now() - timedelta(seconds=300) + UTC_DIFF))
else:
- log.info("CatchUpPlumageSessionThread: Skipping catch-up, no records present (probably first-run)")
+ log.info("CatchUpPlumageSessionThread: "\
+ "Skipping catch-up, no records present (probably first-run)")
except Exception, e:
log.info("%s got exception %s, exiting" % (self.__class__.__name__, str(e)))
Modified: branches/tmckay/mint/python/mint/plumage/update.py
===================================================================
--- branches/tmckay/mint/python/mint/plumage/update.py 2012-08-31 17:58:10 UTC (rev 5454)
+++ branches/tmckay/mint/python/mint/plumage/update.py 2012-09-04 18:15:20 UTC (rev 5455)
@@ -52,17 +52,10 @@
else:
log.debug("Skipping persistent class " + str(cls))
- if len(self.app.classes):
- log.debug("Delete all objects by bound classes " +\
- str(self.app.classes))
- for cls in self.app.classes:
- loop_body(cls)
- else:
- log.debug("Delete all objects by bound packages " +\
- str(self.app.packages))
- for pkg in self.app.packages:
- for cls in pkg._classes:
- loop_body(cls)
+ classes = self.app.get_bound_classes()
+ log.debug("Delete all objects by bound classes " + str(classes))
+ for cls in classes:
+ loop_body(cls)
def init(self):
self.conn = self.app.database.get_connection()
Modified: branches/tmckay/mint/python/mint/session.py
===================================================================
--- branches/tmckay/mint/python/mint/session.py 2012-08-31 17:58:10 UTC (rev 5454)
+++ branches/tmckay/mint/python/mint/session.py 2012-09-04 18:15:20 UTC (rev 5455)
@@ -7,10 +7,11 @@
log = logging.getLogger("mint.session")
class MintSession(object):
- def __init__(self, app, broker_uris):
+ def __init__(self, app, broker_uris, qmf_classes):
self.app = app
self.broker_uris = broker_uris
+ self.qmf_classes = qmf_classes
self.qmf_session = None
self.qmf_brokers = list()
@@ -24,24 +25,6 @@
qmf_broker = self.qmf_session.addBroker(uri, mechanisms=mechs)
self.qmf_brokers.append(qmf_broker)
- def init(self):
- log.info("Initializing %s", self)
-
- def init_qmf_classes(self):
- # Apply the package filter to the class list
- if len(self.app.qmf_classes):
- black_list = set()
- for cls in self.app.qmf_classes:
- if cls._package._name in self.app.qmf_package_filter:
- black_list.add(cls)
- self.app.qmf_classes.difference_update(black_list)
- else:
- # Generate the package list from the model, minus
- # the package filter
- for pkg in self.app.model._packages:
- if pkg._name not in self.app.qmf_package_filter:
- self.app.qmf_packages.add(pkg)
-
def start(self):
log.info("Starting %s", self)
@@ -57,18 +40,11 @@
self.qmf_session.bindAgent("*")
log.info("Binding all agents")
- # Handle bind by class
- if len(self.app.qmf_classes):
- for cls in self.app.qmf_classes:
- pname = cls._package._name
- cname = cls._name
- self.qmf_session.bindClass(pname.lower(), cname.lower())
- log.info("Binding QMF class %s.%s" % (pname, cname))
- else:
- # Handle bind by package
- for pkg in self.app.qmf_packages:
- self.qmf_session.bindPackage(pkg._name.lower())
- log.info("Binding QMF package %s" % pkg._name)
+ for cls in self.qmf_classes:
+ pname = cls._package._name
+ cname = cls._name
+ self.qmf_session.bindClass(pname.lower(), cname.lower())
+ log.info("Binding QMF class %s.%s" % (pname, cname))
for uri in self.broker_uris:
self.add_broker(uri)
Modified: branches/tmckay/mint/python/mint/update.py
===================================================================
--- branches/tmckay/mint/python/mint/update.py 2012-08-31 17:58:10 UTC (rev 5454)
+++ branches/tmckay/mint/python/mint/update.py 2012-09-04 18:15:20 UTC (rev 5455)
@@ -45,19 +45,11 @@
else:
log.debug("Skipping persistent class " + str(cls))
- if len(self.app.qmf_classes):
- log.debug("Delete all objects by bound classes " +\
- str(self.app.qmf_classes))
- for cls in self.app.qmf_classes:
- loop_body(cls)
- else:
- # We bound all classes. Loop over model
- log.debug("Delete all objects by bound packages " +\
- str(self.app.qmf_packages))
- for pkg in self.app.qmf_packages:
- for cls in pkg._classes:
- loop_body(cls)
-
+ classes = self.app.get_bound_classes()
+ log.debug("Delete all objects by bound classes " + str(classes))
+ for cls in classes:
+ loop_body(cls)
+
def init(self):
self.conn = self.app.database.get_connection()
@@ -65,8 +57,8 @@
self.cursor.stats = self.stats
def enqueue(self, update):
+ update.app = self.app
self.updates.put(update)
-
self.stats.enqueued += 1
def run(self):
@@ -216,14 +208,13 @@
class Update(object):
def __init__(self, model):
self.model = model
+ self.app = None
def process(self, thread):
log.debug("Processing %s", self)
try:
- self.do_process(thread.cursor, thread.stats,
- thread.app.qmf_classes, thread.app.qmf_packages)
-
+ self.do_process(thread.cursor, thread.stats)
thread.conn.commit()
except UpdateDropped:
log.debug("Update dropped")
@@ -243,7 +234,7 @@
if thread.halt_on_error:
raise
- def do_process(self, cursor, stats, bound_classes, bound_packages):
+ def do_process(self, cursor, stats):
raise Exception("Not implemented")
def __repr__(self):
@@ -255,7 +246,7 @@
self.qmf_object = qmf_object
- def do_process(self, cursor, stats, bound_classes, bound_packages):
+ def do_process(self, cursor, stats):
cls = self.get_class()
agent_id = self.get_agent_id()
object_id = self.get_object_id()
@@ -654,7 +645,7 @@
def get_agent_id(self):
return make_agent_id(self.qmf_agent)
- def do_process(self, cursor, stats, bound_classes, bound_packages):
+ def do_process(self, cursor, stats):
agent_id = self.get_agent_id()
try:
@@ -671,8 +662,7 @@
stats.agents_updated += 1
- def delete_agent_objects(self, cursor, stats, agent,
- bound_classes, bound_packages):
+ def delete_agent_objects(self, cursor, stats, agent):
def loop_body(cls):
if cls._storage != "none" and cls.check_persistent() == "session":
@@ -681,18 +671,10 @@
#stats.objects_deleted_by_class[cls] += count
cursor.connection.commit()
- if len(bound_classes):
- log.debug("Delete agent objects by bound classes " +\
- str(bound_classes))
- for cls in bound_classes:
- loop_body(cls)
- else:
- # We bound all classes. Loop over model
- log.debug("Delete agent objects by bound packages " +\
- str(bound_packages))
- for pkg in bound_packages:
- for cls in pkg._classes:
- loop_body(cls)
+ classes = self.app.get_bound_classes()
+ log.debug("Delete agent objects by bound classes " + str(classes))
+ for cls in classes:
+ loop_body(cls)
def __repr__(self):
name = self.__class__.__name__
@@ -701,7 +683,7 @@
return "%s(%s)" % (name, agent_id)
class AgentDelete(AgentUpdate):
- def do_process(self, cursor, stats, bound_classes, bound_packages):
+ def do_process(self, cursor, stats):
agent_id = self.get_agent_id()
try:
@@ -713,7 +695,7 @@
stats.agents_deleted += 1
- self.delete_agent_objects(cursor, stats, agent, bound_classes, bound_packages)
+ self.delete_agent_objects(cursor, stats, agent)
class UpdateDropped(Exception):
pass
More information about the cumin-developers
mailing list