r5427 - in trunk: cumin/model/plumage mint/python/mint/plumage

croberts at fedoraproject.org croberts at fedoraproject.org
Thu Jul 5 19:28:01 UTC 2012


Author: croberts
Date: 2012-07-05 19:28:00 +0000 (Thu, 05 Jul 2012)
New Revision: 5427

Modified:
   trunk/cumin/model/plumage/plumage.xml
   trunk/cumin/model/plumage/rosemary.xml
   trunk/mint/python/mint/plumage/session.py
Log:
Adding code to support loading the plumage Accountant samples data into the cumin database.

Modified: trunk/cumin/model/plumage/plumage.xml
===================================================================
--- trunk/cumin/model/plumage/plumage.xml	2012-07-03 18:23:06 UTC (rev 5426)
+++ trunk/cumin/model/plumage/plumage.xml	2012-07-05 19:28:00 UTC (rev 5427)
@@ -17,4 +17,18 @@
     <statistic name="availcpu" type="uint32" uint="uint32" />
     <statistic name="totalcpu" type="uint32" uint="uint32" />
   </class>
+  
+  <class name="Accountant">
+    <property name="host" type="sstr" />
+    <statistic name="user" type="sstr" />
+    <statistic name="agroup" type="sstr" optional="y" />
+    <statistic name="ts" type="absTime" unit="uint32" />    
+    <statistic name="prio" type="float" optional="y" />
+    <statistic name="wresused" type="uint32" uint="uint32" optional="y" />
+    <statistic name="resused" type="uint32" uint="uint32" optional="y" />
+    <statistic name="cquota" type="float" optional="y" />
+    <statistic name="equota" type="float" optional="y" />
+    <statistic name="squota" type="float" optional="y" />
+    <statistic name="wausage" type="uint32" uint="uint32" optional="y" />
+  </class>  
 </schema>
\ No newline at end of file

Modified: trunk/cumin/model/plumage/rosemary.xml
===================================================================
--- trunk/cumin/model/plumage/rosemary.xml	2012-07-03 18:23:06 UTC (rev 5426)
+++ trunk/cumin/model/plumage/rosemary.xml	2012-07-05 19:28:00 UTC (rev 5427)
@@ -68,8 +68,51 @@
 		<statistic name="totalcpu">
 			<title>Total</title>
 		</statistic>
+    </class>
+    <class name="Accountant">
+        <loading_class>
+        	<name>AccountantLoader</name>
+        </loading_class>
+        
+		<source>
+			<database>condor_stats</database>
+			<collection>samples.accountant</collection>
+		</source>
+        
+        <statistic name="user" />
+        <statistic name="ts" timestamp="y" />
+        
+        <statistic name="agroup">
+        	<title>Accounting group</title>
+        </statistic>        	
+        
+		<statistic name="prio">
+			<title>Priority</title>
+		</statistic>
 
-    </class>
+		<statistic name="wresused">
+			<title>Weighted resources used</title>
+		</statistic>
+		
+		<statistic name="resused">
+			<title>Resources used</title>
+		</statistic>
+		
+		<statistic name="cquota">
+			<title>Configured quota</title>
+		</statistic>
+		
+		<statistic name="equota">
+			<title>Effective quota</title>
+		</statistic>
+		
+		<statistic name="squota">
+			<title>Subtree quota</title>
+		</statistic>
+		
+		<statistic name="wausage">
+			<title>Weighted accumulated usage</title>
+		</statistic>
+	</class>    
    </package>
-
 </model>

Modified: trunk/mint/python/mint/plumage/session.py
===================================================================
--- trunk/mint/python/mint/plumage/session.py	2012-07-03 18:23:06 UTC (rev 5426)
+++ trunk/mint/python/mint/plumage/session.py	2012-07-05 19:28:00 UTC (rev 5427)
@@ -43,7 +43,7 @@
                 """)
 
 
-class OSUtil(object):
+class RecordObject(object):
     pass
 
 class PlumageSession(object):
@@ -114,22 +114,17 @@
 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 '''
     def OSUtilLoader(self, obj, cls):
-        obj.threads.append(CatchUpPlumageSessionThread(
-            obj.app,
-            obj.server_host,
-            obj.server_port,
-            cls))
-        obj.threads.append(PlumageSessionThread(
-                obj.app,
-                obj.server_host,
-                obj.server_port,
-                cls))
-        obj.threads.append(CurrentPlumageSessionThread(
-                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 '''        
+    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))
+
 class PlumageSessionThread(MintDaemonThread):
     def __init__(self, app, server_host, server_port, cls):
         super(PlumageSessionThread, self).__init__(app)
@@ -174,9 +169,139 @@
                     sleep(1)
                     if self.stop_requested:
                         break
+    
 
+class PlumageAccountantSessionThread(PlumageSessionThread):
+    def fillAccountantStats(self, time, temptable, name):
+        record = RecordObject()
+        record.host= "%s:%s" % (self.connection.host, self.connection.port)
+        record.user = name
+        record.ts = time - UTC_DIFF
+        
+        userentry = self.collection.find({'ts':time, 'n':name})[0]
+        record.agroup = self.getEntry(userentry, "ag")
+        record.prio = self.getEntry(userentry, "prio")
+        record.wresused = self.getEntry(userentry, "wru")
+        record.resused = self.getEntry(userentry, "ru")
+        record.cquota = self.getEntry(userentry, "cq")
+        record.equota = self.getEntry(userentry, "eq")
+        record.squota = self.getEntry(userentry, "sq")
+        record.wausage = self.getEntry(userentry, "au")
+        
+        return record    
+    
+    def getEntry(self, item, value):
+        result = None
+        try:
+            result = item[value]
+        except:
+            result = None
+        return result
+    
+    def run(self):
+        while True:
+            self._check_connection()
+            if self.stop_requested:
+                return
+    
+            try:
+                # Make sure we have a database
+                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)
+                (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
+                    oldest = datetime.now() - timedelta(seconds=600)
+                oldest = oldest + UTC_DIFF 
+    
+                log.info("PlumageAccountantSessionThread--history:  Loading records older than %s" % oldest)
+                times = sorted(self.collection.find({"ts": {'$lt': oldest}}).distinct('ts'), reverse=True)
+                sample_times = map(lambda i: times[i],filter(lambda i: i%5 == 0,range(len(times))))
+    
+                for time in sample_times:
+                    names = self.collection.find({'ts':time}).distinct('n')
+                    for name in names:
+                        record = self.fillAccountantStats(time, "history", name)    
+                        obj = ObjectUpdate(self.app.model, record, self.cls)
+                        self.app.update_thread.enqueue(obj)
+    
+                log.info("PlumageAccountantSessionThread--history:  run completed")
+    
+            except Exception, e:
+                log.info("%s got exception %s, exiting" % (self.__class__.__name__, str(e)))
+            #wake up once a day just in case there has been additional historical items added
+            sleep(86400)
+
+class CurrentPlumageAccountantSessionThread(PlumageAccountantSessionThread):
+    def run(self):
+        while True:
+            self._check_connection()
+            if self.stop_requested:
+                break
+            
+            # Make sure we have a db
+            try:
+                self._init()
+
+                (oldest, newest) = self.app.update_thread.get_first_and_last_sample_timestamp(self.cls)
+                if newest is not None:            
+                    most_recent = max((datetime.now() - timedelta(seconds=600) + UTC_DIFF), (newest + UTC_DIFF))
+                else:
+                    most_recent = datetime.now() - timedelta(seconds=600) + UTC_DIFF
+
+                log.info("CurrentPlumageAccountantThread--current:  Loading records newer than %s" % most_recent)
+
+                times = sorted(self.collection.find({"ts": {'$gt': most_recent}}).distinct('ts'), reverse=True)
+                sample_times = map(lambda i: times[i],filter(lambda i: i%5 == 0,range(len(times))))
+
+                for time in sample_times:
+                    names = self.collection.find({'ts':time}).distinct('n')
+                    for name in names:
+                        record = self.fillAccountantStats(time, "current", name)    
+                        obj = ObjectUpdate(self.app.model, record, self.cls)
+                        self.app.update_thread.enqueue(obj)
+
+                log.info("CurrentPlumageAccountantThread--current:  pass completed")
+            except Exception, e:
+                log.info("%s got exception %s, sleeping" % (self.__class__.__name__, str(e)))
+            sleep(600)
+
+class CatchupPlumageAccountantSessionThread(PlumageAccountantSessionThread):
+    def run(self):
+        self._check_connection()
+        if self.stop_requested:
+            return
+
+        try:
+            # Make sure we have a db
+            self._init()
+
+            (oldest, newest) = self.app.update_thread.get_first_and_last_sample_timestamp(self.cls)
+            if newest is not None:
+                log.info("CatchupPlumageAccountantThread: Starting for records newer than %s" % newest)
+
+                times = sorted(self.collection.find({"ts": {'$gt': newest + UTC_DIFF, '$lt': datetime.now() - timedelta(seconds=600) + UTC_DIFF}}).distinct('ts'), reverse=True)
+                sample_times = map(lambda i: times[i],filter(lambda i: i%5 == 0,range(len(times))))
+
+                for time in sample_times:
+                    names = self.collection.find({'ts':time}).distinct('n')
+                    for name in names:
+                        record = self.fillAccountantStats(time, "catchup", name)    
+                        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))
+            else:
+                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)))
+
+class PlumageOSUtilSessionThread(PlumageSessionThread):    
     def fillOSUtilStats(self, time, temptable):
-        record = OSUtil()
+        record = RecordObject()
         record.host= "%s:%s" % (self.connection.host, self.connection.port)
         record.total = len(self.collection.find({'ts':time}).distinct('n'))
         record.used = len(self.collection.find({'ts':time,'st':{'$nin':['Unclaimed','Owner']}}).distinct('n'))
@@ -213,9 +338,7 @@
         except:
             return 0
         return itemcount
-            
         
-        
     def run(self):
         while True:
             self._check_connection()
@@ -252,7 +375,7 @@
             #wake up once a day just in case there has been additional historical items added
             sleep(86400)
           
-class CurrentPlumageSessionThread(PlumageSessionThread):
+class CurrentPlumageOSUtilSessionThread(PlumageOSUtilSessionThread):
     def run(self):
         while True:
             self._check_connection()
@@ -290,7 +413,7 @@
    database and then make a pass to load all records from that time forward (up to 5 min ago...those get
    picked-up by the currency thread.
 '''            
-class CatchUpPlumageSessionThread(PlumageSessionThread):
+class CatchUpPlumageOSUtilSessionThread(PlumageOSUtilSessionThread):
     def run(self):
         self._check_connection()
         if self.stop_requested:



More information about the cumin-developers mailing list