source: trunk/jobmond/jobmond.py @ 256

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

jobmond/jobmond.conf, jobmond/examples/lisa_production.conf, jobmond/examples/lisa_test.conf:

  • changed: TORQUE_SERVER is now BATCH_SERVER
  • changed: TORQUE_POLL_INTERVAL is now BATCH_POLL_INTERVAL
  • added: BATCH_API to select which api to use for batch system polling

jobmond/jobmond.py:

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