source: trunk/plugin/togap.py @ 69

Last change on this file since 69 was 69, checked in by bastiaans, 19 years ago

plugin/togap.py:

  • Misc bugfixes
File size: 8.8 KB
RevLine 
[23]1#!/usr/bin/env python
2
[26]3# Specify debugging level here;
4#
[69]5DEBUG_LEVEL = 0
[26]6
7# Wether or not to run as a daemon in background
8#
9DAEMONIZE = 0
10
[64]11# How many seconds interval for polling of jobs
[61]12#
[64]13# this will effect directly how accurate the
14# end time of a job can be determined
15#
16TORQUE_POLL_INTERVAL = 10
[61]17
[68]18# Alternate location of gmond.conf
19#
20# Default: /etc/gmond.conf
21#
22#GMOND_CONF = '/etc/gmond.conf'
23
[23]24from PBSQuery import PBSQuery
[26]25import sys
26import time
[65]27import os
[67]28import socket
[65]29import string
[23]30
[61]31class DataProcessor:
[68]32        """Class for processing of data"""
[61]33
34        binary = '/usr/bin/gmetric'
35
36        def __init__( self, binary=None ):
[68]37                """Remember alternate binary location if supplied"""
[61]38
39                if binary:
40                        self.binary = binary
41
[65]42                self.dmax = TORQUE_POLL_INTERVAL
[61]43
[68]44                try:
45                        gmond_file = GMOND_CONF
46
47                except NameError:
48                        gmond_file = '/etc/gmond.conf'
49
50                if not os.path.exists( gmond_file ):
51                        debug_msg( 0, gmond_file + ' does not exist' )
52                        sys.exit( 1 )
53
[69]54                incompatible = self.checkGmetricVersion()
[61]55
[65]56                if incompatible:
57                        debug_msg( 0, 'Gmetric version not compatible, pls upgrade to at least 3.0.1' )
58                        sys.exit( 1 )
59
60        def checkGmetricVersion( self ):
[68]61                """
62                Check version of gmetric is at least 3.0.1
63                for the syntax we use
64                """
[65]65
66                for line in os.popen( self.binary + ' --version' ).readlines():
67
68                        line = line.split( ' ' )
69
[69]70                        if len( line ) == 2 and str(line).find( 'gmetric' ) != -1:
[65]71                       
[69]72                                gmetric_version = line[1].split( '\n' )[0]
[65]73
[69]74                                version_major = int( gmetric_version.split( '.' )[0] )
75                                version_minor = int( gmetric_version.split( '.' )[1] )
76                                version_patch = int( gmetric_version.split( '.' )[2] )
[65]77
78                                incompatible = 0
79
80                                if version_major < 3:
81
82                                        incompatible = 1
83                               
84                                elif version_major == 3:
85
86                                        if version_minor == 0:
87
88                                                if version_patch < 1:
89                                               
90                                                        incompatbiel = 1
91
92                return incompatible
93
[66]94        def multicastGmetric( self, metricname, metricval, valtype='string', tmax='15' ):
[68]95                """Call gmetric binary and multicast"""
[65]96
97                cmd = self.binary
98
[61]99                try:
100                        cmd = cmd + ' -c' + GMOND_CONF
101                except NameError:
[64]102                        debug_msg( 10, 'Assuming /etc/gmond.conf for gmetric cmd (ommitting)' )
[61]103
[66]104                cmd = cmd + ' -n' + metricname + ' -v"' + metricval + '" -t' + valtype + ' -x' + tmax + ' -d' + str( self.dmax )
[61]105
106                print cmd
[69]107                os.system( cmd )
[61]108
[23]109class PBSDataGatherer:
110
[61]111        jobs = { }
112
[23]113        def __init__( self ):
[68]114                """Setup appropriate variables"""
[23]115
116                self.pq = PBSQuery()
[26]117                self.jobs = { }
[61]118                self.dp = DataProcessor()
[23]119
[26]120        def getAttr( self, attrs, name ):
[68]121                """Return certain attribute from dictionary, if exists"""
[26]122
123                if attrs.has_key( name ):
124                        return attrs[name]
125                else:
126                        return ''
127
128        def jobDataChanged( self, jobs, job_id, attrs ):
[68]129                """Check if job with attrs and job_id in jobs has changed"""
[26]130
131                if jobs.has_key( job_id ):
132                        oldData = jobs[ job_id ]       
133                else:
134                        return 1
135
136                for name, val in attrs.items():
137
138                        if oldData.has_key( name ):
139
140                                if oldData[ name ] != attrs[ name ]:
141
142                                        return 1
143
144                        else:
145                                return 1
146
147                return 0
148
[65]149        def getJobData( self, known_jobs ):
[68]150                """Gather all data on current jobs in Torque"""
[26]151
[65]152                if len( known_jobs ) > 0:
153                        jobs = known_jobs
154                else:
155                        jobs = { }
[26]156
157                joblist = self.pq.getjobs()
158
[69]159                self.cur_time = time.time()
[68]160
[26]161                jobs_processed = [ ]
162
163                for name, attrs in joblist.items():
164
165                        job_id = name.split( '.' )[0]
166
167                        jobs_processed.append( job_id )
[61]168
[26]169                        name = self.getAttr( attrs, 'Job_Name' )
170                        queue = self.getAttr( attrs, 'queue' )
171                        owner = self.getAttr( attrs, 'Job_Owner' ).split( '@' )[0]
172                        requested_time = self.getAttr( attrs, 'Resource_List.walltime' )
173                        requested_memory = self.getAttr( attrs, 'Resource_List.mem' )
174                        mynoderequest = self.getAttr( attrs, 'Resource_List.nodes' )
175                        if mynoderequest.find( ':' ) != -1 and mynoderequest.find( 'ppn' ) != -1:
176                                ppn = mynoderequest.split( ':' )[1].split( 'ppn=' )[1]
177                        else:
178                                ppn = ''
179                        status = self.getAttr( attrs, 'job_state' )
180                        start_timestamp = self.getAttr( attrs, 'mtime' )
[25]181
[67]182                        nodes = self.getAttr( attrs, 'exec_host' ).split( '+' )
183                        nodeslist = [ ]
184
185                        for node in nodes:
186                                host = node.split( '/' )[0]
187
188                                if nodeslist.count( host ) == 0:
189                                        nodeslist.append( node.split( '/' )[0] )
190
[26]191                        myAttrs = { }
192                        myAttrs['name'] = name
193                        myAttrs['queue'] = queue
194                        myAttrs['owner'] = owner
195                        myAttrs['requested_time'] = requested_time
196                        myAttrs['requested_memory'] = requested_memory
197                        myAttrs['ppn'] = ppn
198                        myAttrs['status'] = status
199                        myAttrs['start_timestamp'] = start_timestamp
[69]200                        myAttrs['reported_timestamp'] = str( int( self.cur_time ) )
[67]201                        myAttrs['nodes'] = nodeslist
202                        myAttrs['domain'] = string.join( socket.getfqdn().split( '.' )[1:], '.' )
[26]203
204                        if self.jobDataChanged( jobs, job_id, myAttrs ):
205                                jobs[ job_id ] = myAttrs
[61]206
[69]207                                #self.printJob( jobs, job_id )
[61]208
[26]209                                debug_msg( 10, printTime() + ' job %s state changed' %(job_id) )
210
[65]211                return jobs
212
213        def submitJobData( self, jobs ):
214                """Submit job info list"""
215
[69]216                self.dp.multicastGmetric( 'TOGA-HEARTBEAT', str( int( self.cur_time ) ) )
217
[61]218                # Now let's spread the knowledge
219                #
220                for jobid, jobattrs in jobs.items():
221
[65]222                        gmetric_val = self.compileGmetricVal( jobid, jobattrs )
[61]223
[65]224                        for val in gmetric_val:
225                                self.dp.multicastGmetric( 'TOGA-JOB-' + jobid, val )
[61]226
[67]227        def makeNodeString( self, nodelist ):
[68]228                """Make one big string of all hosts"""
[67]229
230                node_str = None
231
232                for node in nodelist:
233                        if not node_str:
234                                node_str = node
235                        else:
236                                node_str = node_str + ';' + node
237
238                return node_str
239
[65]240        def compileGmetricVal( self, jobid, jobattrs ):
241                """Create a val string for gmetric of jobinfo"""
[61]242
[65]243                name_str = 'name=' + jobattrs['name']
244                queue_str = 'queue=' + jobattrs['queue']
245                owner_str = 'owner=' + jobattrs['owner']
246                rtime_str = 'rtime=' + jobattrs['requested_time']
247                rmem_str = 'rmem=' + jobattrs['requested_memory']
248                ppn_str = 'ppn=' + jobattrs['ppn']
249                status_str = 'status=' + jobattrs['status']
250                stime_str = 'stime=' + jobattrs['start_timestamp']
[68]251                reported_str = 'reported=' + jobattrs['reported_timestamp']
[67]252                domain_str = 'domain=' + jobattrs['domain']
253                node_str = 'nodes=' + self.makeNodeString( jobattrs['nodes'] )
[26]254
[68]255                appendList = [ name_str, queue_str, owner_str, rtime_str, rmem_str, ppn_str, status_str, stime_str, reported_str, domain_str, node_str ]
[65]256
257                return self.makeAppendLists( appendList )
258
259        def makeAppendLists( self, append_list ):
[68]260                """
261                Divide all values from append_list over strings with a maximum
262                size of 1400
263                """
[65]264
265                app_lists = [ ]
266
267                mystr = None
268
269                for val in append_list:
270
271                        if not mystr:
272                                mystr = val
273                        else:
274                                if not self.checkValAppendMaxSize( mystr, val ):
275                                        mystr = mystr + ' ' + val
276                                else:
277                                        # Too big, new appenlist
278                                        app_lists.append( mystr )
279                                        mystr = val
280
281                app_lists.append( mystr )
282
283                return app_lists
284
285        def checkValAppendMaxSize( self, val, text ):
286                """Check if val + text size is not above 1400 (max msg size)"""
287
[69]288                # Max frame size of a udp datagram is 1500 bytes
289                # removing misc header and gmetric stuff leaves about 1400 bytes
290                #
[65]291                if len( val + text ) > 1400:
292                        return 1
293                else:
294                        return 0
295
[61]296        def printJobs( self, jobs ):
[65]297                """Print a jobinfo overview"""
298
[26]299                for name, attrs in self.jobs.items():
300
301                        print 'job %s' %(name)
302
303                        for name, val in attrs.items():
304
305                                print '\t%s = %s' %( name, val )
306
[61]307        def printJob( self, jobs, job_id ):
[65]308                """Print job with job_id from jobs"""
[26]309
310                print 'job %s' %(job_id)
311
[65]312                for name, val in jobs[ job_id ].items():
[26]313
314                        print '\t%s = %s' %( name, val )
315
316        def daemon( self ):
[65]317                """Run as daemon forever"""
[26]318
319                # Fork the first child
320                #
321                pid = os.fork()
322                if pid > 0:
323                        sys.exit(0)  # end parrent
324
325                # creates a session and sets the process group ID
326                #
327                os.setsid()
328
329                # Fork the second child
330                #
331                pid = os.fork()
332                if pid > 0:
333                        sys.exit(0)  # end parrent
334
335                # Go to the root directory and set the umask
336                #
337                os.chdir('/')
338                os.umask(0)
339
340                sys.stdin.close()
341                sys.stdout.close()
342                sys.stderr.close()
343
344                os.open('/dev/null', 0)
345                os.dup(0)
346                os.dup(0)
347
348                self.run()
349
350        def run( self ):
[65]351                """Main thread"""
[26]352
353                while ( 1 ):
354               
[65]355                        self.jobs = self.getJobData( self.jobs )
356                        self.submitJobData( self.jobs )
[64]357                        time.sleep( TORQUE_POLL_INTERVAL )     
[26]358
359def printTime( ):
[65]360        """Print current time/date in human readable format for log/debug"""
[26]361
362        return time.strftime("%a, %d %b %Y %H:%M:%S")
363
364def debug_msg( level, msg ):
[65]365        """Print msg if at or above current debug level"""
[26]366
367        if (DEBUG_LEVEL >= level):
368                        sys.stderr.write( msg + '\n' )
369
[23]370def main():
[65]371        """Application start"""
[23]372
373        gather = PBSDataGatherer()
[26]374        if DAEMONIZE:
375                gather.daemon()
376        else:
377                gather.run()
[23]378
[65]379# w00t someone started me
380#
[23]381if __name__ == '__main__':
382        main()
Note: See TracBrowser for help on using the repository browser.