Added ThreadedStatistics plugin class. Migrated DBStats to use ThreadedStatistics.

git-svn-id: https://svn.mcs.anl.gov/repos/bcfg/trunk/bcfg2@5889 ce84e21b-d406-0410-9b95-82705330c041

2 files changed, 38 insertions, 7 deletions

diff --git a/src/lib/Server/Plugin.py b/src/lib/Server/Plugin.py
index f3b97b336..ff87c15df 100644
--- a/src/lib/Server/Plugin.py
+++ b/src/lib/Server/Plugin.py
@@ -7,6 +7,8 @@ import lxml.etree
 import os
 import posixpath
 import re
+import Queue
+import threading
 
 from lxml.etree import XML, XMLSyntaxError
 
@@ -141,6 +143,35 @@ class Statistics(object):
     def process_statistics(self, client, xdata):
         pass
 
+class ThreadedStatistics(Statistics,
+                         threading.Thread):
+    '''Threaded statistics handling capability'''
+    def __init__(self, core, datastore):
+        Statistics.__init__(self)
+        threading.Thread.__init__(self)
+        # Event from the core signaling an exit
+        self.terminate = core.terminate
+        self.work_queue = Queue.Queue()
+        self.start()
+
+    def run(self):
+        while not (self.terminate.isSet() and self.work_queue.empty()):
+            try:
+                (xdata, client) = self.work_queue.get(block=True, timeout=5)
+            except Queue.Empty:
+                continue
+            except Exception, e:
+                logger.error("ThreadedStatistics: %s" % e)
+                continue
+            self.handle_statistic(xdata, client)
+
+    def process_statistics(self, metadata, data):
+        self.work_queue.put((metadata, copy.deepcopy(data)))
+
+    def handle_statistics(self, metadata, data):
+        '''Handle stats here'''
+        pass
+
 class PullSource(object):
     def GetExtra(self, client):
         return []
diff --git a/src/lib/Server/Plugins/DBStats.py b/src/lib/Server/Plugins/DBStats.py
index 49048fad9..1b8052c2d 100644
--- a/src/lib/Server/Plugins/DBStats.py
+++ b/src/lib/Server/Plugins/DBStats.py
@@ -13,14 +13,14 @@ from Bcfg2.Server.Reports.updatefix import update_database
 logger = logging.getLogger('Bcfg2.Plugins.DBStats')
 
 class DBStats(Bcfg2.Server.Plugin.Plugin,
-              Bcfg2.Server.Plugin.Statistics,
+              Bcfg2.Server.Plugin.ThreadedStatistics,
               Bcfg2.Server.Plugin.PullSource):
     name = 'DBStats'
     __version__ = '$Id$'
 
     def __init__(self, core, datastore):
         Bcfg2.Server.Plugin.Plugin.__init__(self, core, datastore)
-        Bcfg2.Server.Plugin.Statistics.__init__(self)
+        Bcfg2.Server.Plugin.ThreadedStatistics.__init__(self, core, datastore)
         Bcfg2.Server.Plugin.PullSource.__init__(self)
         self.cpath = "%s/Metadata/clients.xml" % datastore
         self.core = core
@@ -31,13 +31,13 @@ class DBStats(Bcfg2.Server.Plugin.Plugin,
             logger.debug(str(inst))
             logger.debug(str(type(inst)))
 
-    def process_statistics(self, mdata, xdata):
-        newstats = xdata.find("Statistics")
+    def handle_statistic(self, metadata, data):
+        newstats = data.find("Statistics")
         newstats.set('time', time.asctime(time.localtime()))
         # ick
-        xdata = lxml.etree.tostring(newstats)
-        ndx = lxml.etree.XML(xdata)
-        e = lxml.etree.Element('Node', name=mdata.hostname)
+        data = lxml.etree.tostring(newstats)
+        ndx = lxml.etree.XML(data)
+        e = lxml.etree.Element('Node', name=metadata.hostname)
         e.append(ndx)
         container = lxml.etree.Element("ConfigStatistics")
         container.append(e)