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