source: trunk/jobmond/jobmond.py @ 220

Last change on this file since 220 was 220, checked in by bastiaans, 18 years ago

jobmond/jobmond.py:

  • bugfix: allow BATCH_HOST_TRANSLATE to be empty
File size: 13.3 KB
RevLine 
[23]1#!/usr/bin/env python
2
[212]3import sys, getopt, ConfigParser
[26]4
[212]5def processArgs( args ):
[26]6
[212]7        SHORT_L = 'c:'
8        LONG_L = 'config='
[165]9
[212]10        config_filename = None
[61]11
[212]12        try:
[68]13
[212]14                opts, args = getopt.getopt( args, SHORT_L, LONG_L )
[185]15
[212]16        except getopt.error, detail:
17
18                print detail
19                sys.exit(1)
20
21        for opt, value in opts:
22
23                if opt in [ '--config', '-c' ]:
24               
25                        config_filename = value
26
27        if not config_filename:
28
29                config_filename = '/etc/jobmond.conf'
30
31        return loadConfig( config_filename )
32
33def loadConfig( filename ):
34
[215]35        def getlist( cfg_string ):
36
37                my_list = [ ]
38
39                for item_txt in cfg_string.split( ',' ):
40
41                        sep_char = None
42
43                        item_txt = item_txt.strip()
44
45                        for s_char in [ "'", '"' ]:
46
47                                if item_txt.find( s_char ) != -1:
48
49                                        if item_txt.count( s_char ) != 2:
50
51                                                print 'Missing quote: %s' %item_txt
52                                                sys.exit( 1 )
53
54                                        else:
55
56                                                sep_char = s_char
57                                                break
58
59                        if sep_char:
60
61                                item_txt = item_txt.split( sep_char )[1]
62
63                        my_list.append( item_txt )
64
65                return my_list
66
[212]67        cfg = ConfigParser.ConfigParser()
68
69        cfg.read( filename )
70
[215]71        global DEBUG_LEVEL, DAEMONIZE, TORQUE_SERVER, TORQUE_POLL_INTERVAL, GMOND_CONF, DETECT_TIME_DIFFS, BATCH_HOST_TRANSLATE
[212]72
73        # Specify debugging level here;
74        #
75        # 10 = gemtric cmd's
76        DEBUG_LEVEL = cfg.getint( 'DEFAULT', 'DEBUG_LEVEL' )
77
78        # Wether or not to run as a daemon in background
79        #
80        DAEMONIZE = cfg.getboolean( 'DEFAULT', 'DAEMONIZE' )
81
82        # Which Torque server to monitor
83        #
84        TORQUE_SERVER = cfg.get( 'DEFAULT', 'TORQUE_SERVER' )
85
86        # How many seconds interval for polling of jobs
87        #
88        # this will effect directly how accurate the
89        # end time of a job can be determined
90        #
91        TORQUE_POLL_INTERVAL = cfg.getint( 'DEFAULT', 'TORQUE_POLL_INTERVAL' )
92
93        # Alternate location of gmond.conf
94        #
95        # Default: /etc/gmond.conf
96        #
97        GMOND_CONF = cfg.get( 'DEFAULT', 'GMOND_CONF' )
98
99        # Wether or not to detect differences in
100        # time from Torque server and local time.
101        #
102        # Ideally both machines (if not the same)
103        # should have the same time (via ntp or whatever)
104        #
105        DETECT_TIME_DIFFS = cfg.getboolean( 'DEFAULT', 'DETECT_TIME_DIFFS' )
106
[215]107        # Regexp style hostname translation
108        #
109        # Usefull if your Torque hostnames are not the same as your
110        # Ganglia hostnames (different network interfaces)
111        #
112        # Syntax: /orig/new/
113        #
114        BATCH_HOST_TRANSLATE = getlist( cfg.get( 'DEFAULT', 'BATCH_HOST_TRANSLATE' ) )
115
[212]116        return True
117
[23]118from PBSQuery import PBSQuery
119
[215]120import time, os, socket, string, re
[212]121
[61]122class DataProcessor:
[68]123        """Class for processing of data"""
[61]124
125        binary = '/usr/bin/gmetric'
126
127        def __init__( self, binary=None ):
[68]128                """Remember alternate binary location if supplied"""
[61]129
130                if binary:
131                        self.binary = binary
132
[80]133                # Timeout for XML
134                #
135                # From ganglia's documentation:
136                #
137                # 'A metric will be deleted DMAX seconds after it is received, and
138                # DMAX=0 means eternal life.'
[61]139
[194]140                self.dmax = str( int( int( TORQUE_POLL_INTERVAL ) + 2 ) )
[80]141
[68]142                try:
143                        gmond_file = GMOND_CONF
144
145                except NameError:
146                        gmond_file = '/etc/gmond.conf'
147
148                if not os.path.exists( gmond_file ):
149                        debug_msg( 0, gmond_file + ' does not exist' )
150                        sys.exit( 1 )
151
[69]152                incompatible = self.checkGmetricVersion()
[61]153
[65]154                if incompatible:
155                        debug_msg( 0, 'Gmetric version not compatible, pls upgrade to at least 3.0.1' )
156                        sys.exit( 1 )
157
158        def checkGmetricVersion( self ):
[68]159                """
160                Check version of gmetric is at least 3.0.1
161                for the syntax we use
162                """
[65]163
164                for line in os.popen( self.binary + ' --version' ).readlines():
165
166                        line = line.split( ' ' )
167
[69]168                        if len( line ) == 2 and str(line).find( 'gmetric' ) != -1:
[65]169                       
[69]170                                gmetric_version = line[1].split( '\n' )[0]
[65]171
[69]172                                version_major = int( gmetric_version.split( '.' )[0] )
173                                version_minor = int( gmetric_version.split( '.' )[1] )
174                                version_patch = int( gmetric_version.split( '.' )[2] )
[65]175
176                                incompatible = 0
177
178                                if version_major < 3:
179
180                                        incompatible = 1
181                               
182                                elif version_major == 3:
183
184                                        if version_minor == 0:
185
186                                                if version_patch < 1:
187                                               
[91]188                                                        incompatible = 1
[65]189
190                return incompatible
191
[75]192        def multicastGmetric( self, metricname, metricval, valtype='string' ):
[68]193                """Call gmetric binary and multicast"""
[65]194
195                cmd = self.binary
196
[61]197                try:
198                        cmd = cmd + ' -c' + GMOND_CONF
199                except NameError:
[64]200                        debug_msg( 10, 'Assuming /etc/gmond.conf for gmetric cmd (ommitting)' )
[61]201
[168]202                cmd = cmd + ' -n' + str( metricname )+ ' -v"' + str( metricval )+ '" -t' + str( valtype ) + ' -d' + str( self.dmax )
[61]203
[101]204                debug_msg( 10, printTime() + ' ' + cmd )
[69]205                os.system( cmd )
[61]206
[174]207class DataGatherer:
[23]208
[61]209        jobs = { }
210
[23]211        def __init__( self ):
[68]212                """Setup appropriate variables"""
[23]213
[26]214                self.jobs = { }
[185]215                self.timeoffset = 0
[61]216                self.dp = DataProcessor()
[91]217                self.initPbsQuery()
[23]218
[91]219        def initPbsQuery( self ):
220
221                self.pq = None
[165]222                if( TORQUE_SERVER ):
223                        self.pq = PBSQuery( TORQUE_SERVER )
[174]224                else:
[165]225                        self.pq = PBSQuery()
[91]226
[26]227        def getAttr( self, attrs, name ):
[68]228                """Return certain attribute from dictionary, if exists"""
[26]229
230                if attrs.has_key( name ):
231                        return attrs[name]
232                else:
233                        return ''
234
235        def jobDataChanged( self, jobs, job_id, attrs ):
[68]236                """Check if job with attrs and job_id in jobs has changed"""
[26]237
238                if jobs.has_key( job_id ):
239                        oldData = jobs[ job_id ]       
240                else:
241                        return 1
242
243                for name, val in attrs.items():
244
245                        if oldData.has_key( name ):
246
247                                if oldData[ name ] != attrs[ name ]:
248
249                                        return 1
250
251                        else:
252                                return 1
253
254                return 0
255
[65]256        def getJobData( self, known_jobs ):
[68]257                """Gather all data on current jobs in Torque"""
[26]258
[65]259                if len( known_jobs ) > 0:
260                        jobs = known_jobs
261                else:
262                        jobs = { }
[26]263
[101]264                #self.initPbsQuery()
[125]265       
266                #print self.pq.getnodes()
267       
[26]268                joblist = self.pq.getjobs()
269
[69]270                self.cur_time = time.time()
[68]271
[26]272                jobs_processed = [ ]
273
[125]274                #self.printJobs( joblist )
275
[26]276                for name, attrs in joblist.items():
277
278                        job_id = name.split( '.' )[0]
279
280                        jobs_processed.append( job_id )
[61]281
[26]282                        name = self.getAttr( attrs, 'Job_Name' )
283                        queue = self.getAttr( attrs, 'queue' )
284                        owner = self.getAttr( attrs, 'Job_Owner' ).split( '@' )[0]
285                        requested_time = self.getAttr( attrs, 'Resource_List.walltime' )
286                        requested_memory = self.getAttr( attrs, 'Resource_List.mem' )
[95]287
[26]288                        mynoderequest = self.getAttr( attrs, 'Resource_List.nodes' )
[95]289
[26]290                        if mynoderequest.find( ':' ) != -1 and mynoderequest.find( 'ppn' ) != -1:
291                                ppn = mynoderequest.split( ':' )[1].split( 'ppn=' )[1]
292                        else:
293                                ppn = ''
[95]294
[26]295                        status = self.getAttr( attrs, 'job_state' )
[25]296
[95]297                        if status == 'R':
298                                start_timestamp = self.getAttr( attrs, 'mtime' )
299                                nodes = self.getAttr( attrs, 'exec_host' ).split( '+' )
[133]300
301                                nodeslist = [ ]
302
303                                for node in nodes:
304                                        host = node.split( '/' )[0]
305
306                                        if nodeslist.count( host ) == 0:
[215]307
308                                                for translate_pattern in BATCH_HOST_TRANSLATE:
309
[220]310                                                        if translate_pattern.find( '/' ) != -1:
[215]311
[220]312                                                                translate_orig = translate_pattern.split( '/' )[1]
313                                                                translate_new = translate_pattern.split( '/' )[2]
314
315                                                                host = re.sub( translate_orig, translate_new, host )
[216]316                               
[217]317                                                if not host in nodeslist:
[216]318                               
319                                                        nodeslist.append( host )
[133]320
[185]321                                if DETECT_TIME_DIFFS:
322
323                                        # If a job start if later than our current date,
324                                        # that must mean the Torque server's time is later
325                                        # than our local time.
326                               
327                                        if int(start_timestamp) > int( int(self.cur_time) + int(self.timeoffset) ):
328
329                                                self.timeoffset = int( int(start_timestamp) - int(self.cur_time) )
330
[133]331                        elif status == 'Q':
[95]332                                start_timestamp = ''
[133]333                                count_mynodes = 0
334                                numeric_node = 1
[95]335
[133]336                                for node in mynoderequest.split( '+' ):
[67]337
[133]338                                        nodepart = node.split( ':' )[0]
[67]339
[133]340                                        for letter in nodepart:
[67]341
[133]342                                                if letter not in string.digits:
343
344                                                        numeric_node = 0
345
346                                        if not numeric_node:
347                                                count_mynodes = count_mynodes + 1
348                                        else:
349                                                count_mynodes = count_mynodes + int( nodepart )
350                                               
[134]351                                nodeslist = count_mynodes
[172]352                        else:
353                                start_timestamp = ''
[173]354                                nodeslist = ''
[133]355
[26]356                        myAttrs = { }
357                        myAttrs['name'] = name
358                        myAttrs['queue'] = queue
359                        myAttrs['owner'] = owner
360                        myAttrs['requested_time'] = requested_time
361                        myAttrs['requested_memory'] = requested_memory
362                        myAttrs['ppn'] = ppn
363                        myAttrs['status'] = status
364                        myAttrs['start_timestamp'] = start_timestamp
[185]365                        myAttrs['reported'] = str( int( int( self.cur_time ) + int( self.timeoffset ) ) )
[67]366                        myAttrs['nodes'] = nodeslist
367                        myAttrs['domain'] = string.join( socket.getfqdn().split( '.' )[1:], '.' )
[80]368                        myAttrs['poll_interval'] = TORQUE_POLL_INTERVAL
[26]369
[184]370                        if self.jobDataChanged( jobs, job_id, myAttrs ) and myAttrs['status'] in [ 'R', 'Q' ]:
[26]371                                jobs[ job_id ] = myAttrs
[61]372
[101]373                                #debug_msg( 10, printTime() + ' job %s state changed' %(job_id) )
[26]374
[76]375                for id, attrs in jobs.items():
376
377                        if id not in jobs_processed:
378
379                                # This one isn't there anymore; toedeledoki!
380                                #
381                                del jobs[ id ]
382
[65]383                return jobs
384
385        def submitJobData( self, jobs ):
386                """Submit job info list"""
387
[219]388                self.dp.multicastGmetric( 'MONARCH-HEARTBEAT', str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
[69]389
[61]390                # Now let's spread the knowledge
391                #
392                for jobid, jobattrs in jobs.items():
393
[95]394                        gmetric_val = self.compileGmetricVal( jobid, jobattrs )
[61]395
[95]396                        for val in gmetric_val:
[219]397                                self.dp.multicastGmetric( 'MONARCH-JOB-' + jobid, val )
[61]398
[67]399        def makeNodeString( self, nodelist ):
[68]400                """Make one big string of all hosts"""
[67]401
402                node_str = None
403
404                for node in nodelist:
405                        if not node_str:
406                                node_str = node
407                        else:
408                                node_str = node_str + ';' + node
409
410                return node_str
411
[65]412        def compileGmetricVal( self, jobid, jobattrs ):
413                """Create a val string for gmetric of jobinfo"""
[61]414
[80]415                appendList = [ ]
416                appendList.append( 'name=' + jobattrs['name'] )
417                appendList.append( 'queue=' + jobattrs['queue'] )
418                appendList.append( 'owner=' + jobattrs['owner'] )
419                appendList.append( 'requested_time=' + jobattrs['requested_time'] )
[95]420
421                if jobattrs['requested_memory'] != '':
422                        appendList.append( 'requested_memory=' + jobattrs['requested_memory'] )
423
424                if jobattrs['ppn'] != '':
425                        appendList.append( 'ppn=' + jobattrs['ppn'] )
426
[80]427                appendList.append( 'status=' + jobattrs['status'] )
[95]428
429                if jobattrs['start_timestamp'] != '':
430                        appendList.append( 'start_timestamp=' + jobattrs['start_timestamp'] )
431
[80]432                appendList.append( 'reported=' + jobattrs['reported'] )
[85]433                appendList.append( 'poll_interval=' + str( jobattrs['poll_interval'] ) )
[80]434                appendList.append( 'domain=' + jobattrs['domain'] )
[26]435
[134]436                if jobattrs['status'] == 'R':
437                        if len( jobattrs['nodes'] ) > 0:
438                                appendList.append( 'nodes=' + self.makeNodeString( jobattrs['nodes'] ) )
439                elif jobattrs['status'] == 'Q':
440                        appendList.append( 'nodes=' + str(jobattrs['nodes']) )
[95]441
[65]442                return self.makeAppendLists( appendList )
443
444        def makeAppendLists( self, append_list ):
[68]445                """
446                Divide all values from append_list over strings with a maximum
447                size of 1400
448                """
[65]449
450                app_lists = [ ]
451
452                mystr = None
453
454                for val in append_list:
455
456                        if not mystr:
457                                mystr = val
458                        else:
459                                if not self.checkValAppendMaxSize( mystr, val ):
460                                        mystr = mystr + ' ' + val
461                                else:
462                                        # Too big, new appenlist
463                                        app_lists.append( mystr )
464                                        mystr = val
465
466                app_lists.append( mystr )
467
468                return app_lists
469
470        def checkValAppendMaxSize( self, val, text ):
471                """Check if val + text size is not above 1400 (max msg size)"""
472
[69]473                # Max frame size of a udp datagram is 1500 bytes
[216]474                # removing misc header and gmetric stuff leaves about 1300 bytes
[69]475                #
[216]476                if len( val + text ) > 900:
[65]477                        return 1
478                else:
479                        return 0
480
[61]481        def printJobs( self, jobs ):
[65]482                """Print a jobinfo overview"""
483
[26]484                for name, attrs in self.jobs.items():
485
486                        print 'job %s' %(name)
487
488                        for name, val in attrs.items():
489
490                                print '\t%s = %s' %( name, val )
491
[61]492        def printJob( self, jobs, job_id ):
[65]493                """Print job with job_id from jobs"""
[26]494
495                print 'job %s' %(job_id)
496
[65]497                for name, val in jobs[ job_id ].items():
[26]498
499                        print '\t%s = %s' %( name, val )
500
501        def daemon( self ):
[65]502                """Run as daemon forever"""
[26]503
504                # Fork the first child
505                #
506                pid = os.fork()
507                if pid > 0:
[212]508                        sys.exit(0)  # end parent
[26]509
510                # creates a session and sets the process group ID
511                #
512                os.setsid()
513
514                # Fork the second child
515                #
516                pid = os.fork()
517                if pid > 0:
[212]518                        sys.exit(0)  # end parent
[26]519
520                # Go to the root directory and set the umask
521                #
522                os.chdir('/')
523                os.umask(0)
524
525                sys.stdin.close()
526                sys.stdout.close()
527                sys.stderr.close()
528
529                os.open('/dev/null', 0)
530                os.dup(0)
531                os.dup(0)
532
533                self.run()
534
535        def run( self ):
[65]536                """Main thread"""
[26]537
538                while ( 1 ):
539               
[65]540                        self.jobs = self.getJobData( self.jobs )
541                        self.submitJobData( self.jobs )
[64]542                        time.sleep( TORQUE_POLL_INTERVAL )     
[26]543
544def printTime( ):
[65]545        """Print current time/date in human readable format for log/debug"""
[26]546
547        return time.strftime("%a, %d %b %Y %H:%M:%S")
548
549def debug_msg( level, msg ):
[65]550        """Print msg if at or above current debug level"""
[26]551
552        if (DEBUG_LEVEL >= level):
553                        sys.stderr.write( msg + '\n' )
554
[23]555def main():
[65]556        """Application start"""
[23]557
[212]558        if not processArgs( sys.argv[1:] ):
559                sys.exit( 1 )
560
[174]561        gather = DataGatherer()
[26]562        if DAEMONIZE:
563                gather.daemon()
564        else:
565                gather.run()
[23]566
[65]567# w00t someone started me
568#
[23]569if __name__ == '__main__':
570        main()
Note: See TracBrowser for help on using the repository browser.