source: trunk/jobmond/jobmond.py @ 266

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

jobmond/jobmond.py:

  • try to guess api 'pbs' for old configs

web/addons/job_monarch/index.php:

  • changed makeHeader() to check for page_call. this will prevent multiple Jobarchive link/images in the header, by multiple calls

web/addons/job_monarch/overview.php:

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