Changeset 65 for trunk/plugin

Timestamp:

04/14/05 12:10:51 (18 years ago)

Author:

bastiaans

Message:

plugin/togap.py:

• Added version check of gmetric since we only understand 3.0.1 syntax atm
• Disabled stop_timestamp transmitting should be determined at toga _server_ and not here
• Will check if gmetrics to be send > 1400 bytes and chop it up in pieces if needed

File:

• trunk/plugin/togap.py (modified) (12 diffs)

trunk/plugin/togap.py

@@ -19 +19 @@
 import sys
 import time
+import os
+import string
 
 class DataProcessor:
@@ -29 +31 @@
                         self.binary = binary
 
-        def multicastGmetric( self, metricname, metricval, tmax ):
-
-                cmd = binary
+                self.dmax = TORQUE_POLL_INTERVAL
+
+                #incompatible = self.checkGmetricVersion
+                incompatible = 0
+
+                if incompatible:
+                        debug_msg( 0, 'Gmetric version not compatible, pls upgrade to at least 3.0.1' )
+                        sys.exit( 1 )
+
+        def checkGmetricVersion( self ):
+
+                for line in os.popen( self.binary + ' --version' ).readlines():
+
+                        line = line.split( ' ' )
+
+                        if len( line ) == 2 and line.find( 'gmetric' ) != -1:
+                        
+                                gmetric_version = line[1]
+
+                                version_major = int( gemtric_version.split( '.' )[0] )
+                                version_minor = int( gemtric_version.split( '.' )[1] )
+                                version_patch = int( gemtric_version.split( '.' )[2] )
+
+                                incompatible = 0
+
+                                if version_major < 3:
+
+                                        incompatible = 1
+                                
+                                elif version_major == 3:
+
+                                        if version_minor == 0:
+
+                                                if version_patch < 1:
+                                                
+                                                        incompatbiel = 1
+
+                return incompatible
+
+        def multicastGmetric( self, metricname, metricval, tmax='15' ):
+
+                cmd = self.binary
 
                 try:
@@ -38 +79 @@
                         debug_msg( 10, 'Assuming /etc/gmond.conf for gmetric cmd (ommitting)' )
 
-                cmd = cmd + ' -n' + metricname + ' -v' + metricval + ' -t' + tmax
+                cmd = cmd + ' -n' + metricname + ' -v"' + metricval + '" -t' + tmax + ' -d' + str( self.dmax )
 
                 print cmd
@@ -80 +121 @@
                 return 0
 
-        def getJobData( self ):
-
-                jobs = self.jobs[:]
+        def getJobData( self, known_jobs ):
+
+                if len( known_jobs ) > 0:
+                        jobs = known_jobs
+                else:
+                        jobs = { }
 
                 joblist = self.pq.getjobs()
@@ -106 +150 @@
                         status = self.getAttr( attrs, 'job_state' )
                         start_timestamp = self.getAttr( attrs, 'mtime' )
-                        stop_timestamp = ''
+                        #stop_timestamp = ''
 
                         myAttrs = { }
@@ -117 +161 @@
                         myAttrs['status'] = status
                         myAttrs['start_timestamp'] = start_timestamp
-                        myAttrs['stop_timestamp'] = stop_timestamp
+                        #myAttrs['stop_timestamp'] = stop_timestamp
 
                         if self.jobDataChanged( jobs, job_id, myAttrs ):
@@ -126 +170 @@
                                 debug_msg( 10, printTime() + ' job %s state changed' %(job_id) )
 
-                for id, attrs in jobs.items():
-
-                        # This job was there in the last run, and not anymore
-                        # it must have finished
-
-                        if id not in jobs_processed and attrs['stop_timestamp'] == '':
-
-                                jobs[ id ]['status'] = 'F'
-                                jobs[ id ]['stop_timestamp'] = time.time()
-                                debug_msg( 10, printTime() + ' job %s finished' %(id) )
-                                self.printJob( jobs, id )
+                #for id, attrs in jobs.items():
+
+                #       # This job was there in the last run, and not anymore
+                #       # it must have finished
+
+                #       if id not in jobs_processed and attrs['stop_timestamp'] == '':
+
+                #               jobs[ id ]['status'] = 'F'
+                #               jobs[ id ]['stop_timestamp'] = time.time()
+                #               debug_msg( 10, printTime() + ' job %s finished' %(id) )
+                #               self.printJob( jobs, id )
+
+                return jobs
+
+        def submitJobData( self, jobs ):
+                """Submit job info list"""
+
+                time_now = time.time()
+
+                self.dp.multicastGmetric( 'TOGA-HEARTBEAT', str( time_now ) )
 
                 # Now let's spread the knowledge
@@ -142 +195 @@
                 for jobid, jobattrs in jobs.items():
 
-                        if ARCHIVE_MODE:
-
-                                if self.jobDataChanged( self.jobs, jobid, jobattrs ):
-
-                                        self.dp.togaSubmitJob( jobid, jobattrs )
-
-                        self.dp.multicastGmetric( jobid, jobattrs )
-                                        
-                self.jobs = jobs
+                        gmetric_val = self.compileGmetricVal( jobid, jobattrs )
+
+                        for val in gmetric_val:
+                                self.dp.multicastGmetric( 'TOGA-JOB-' + jobid, val )
+
+        def compileGmetricVal( self, jobid, jobattrs ):
+                """Create a val string for gmetric of jobinfo"""
+
+                name_str = 'name=' + jobattrs['name']
+                queue_str = 'queue=' + jobattrs['queue']
+                owner_str = 'owner=' + jobattrs['owner']
+                rtime_str = 'rtime=' + jobattrs['requested_time']
+                rmem_str = 'rmem=' + jobattrs['requested_memory']
+                ppn_str = 'ppn=' + jobattrs['ppn']
+                status_str = 'status=' + jobattrs['status']
+                stime_str = 'stime=' + jobattrs['start_timestamp']
+
+                appendList = [ name_str, queue_str, owner_str, rtime_str, rmem_str, ppn_str, status_str, stime_str ]
+
+                return self.makeAppendLists( appendList )
+
+        def makeAppendLists( self, append_list ):
+
+                app_lists = [ ]
+
+                mystr = None
+
+                for val in append_list:
+
+                        if not mystr:
+                                mystr = val
+                        else:
+                                if not self.checkValAppendMaxSize( mystr, val ):
+                                        mystr = mystr + ' ' + val
+                                else:
+                                        # Too big, new appenlist
+                                        app_lists.append( mystr )
+                                        mystr = val
+
+                app_lists.append( mystr )
+
+                return app_lists
+
+        def checkValAppendMaxSize( self, val, text ):
+                """Check if val + text size is not above 1400 (max msg size)"""
+
+                if len( val + text ) > 1400:
+                        return 1
+                else:
+                        return 0
 
         def printJobs( self, jobs ):
-        
+                """Print a jobinfo overview"""
+
                 for name, attrs in self.jobs.items():
 
@@ -163 +258 @@
 
         def printJob( self, jobs, job_id ):
+                """Print job with job_id from jobs"""
 
                 print 'job %s' %(job_id)
 
-                for name, val in self.jobs[ job_id ].items():
+                for name, val in jobs[ job_id ].items():
 
                         print '\t%s = %s' %( name, val )
 
         def daemon( self ):
-                "Run as daemon forever"
-
-                self.DAEMON = 1
+                """Run as daemon forever"""
 
                 # Fork the first child
@@ -207 +301 @@
 
         def run( self ):
-                "Main thread"
+                """Main thread"""
 
                 while ( 1 ):
                 
-                        self.getJobData()
+                        self.jobs = self.getJobData( self.jobs )
+                        self.submitJobData( self.jobs )
                         time.sleep( TORQUE_POLL_INTERVAL )      
 
 def printTime( ):
+        """Print current time/date in human readable format for log/debug"""
 
         return time.strftime("%a, %d %b %Y %H:%M:%S")
 
 def debug_msg( level, msg ):
+        """Print msg if at or above current debug level"""
 
         if (DEBUG_LEVEL >= level):
@@ -224 +321 @@
 
 def main():
+        """Application start"""
 
         gather = PBSDataGatherer()
@@ -231 +329 @@
                 gather.run()
 
+# w00t someone started me
+#
 if __name__ == '__main__':
         main()