source: trunk/jobmond/jobmond.py @ 244

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

jobmond/jobmond.py:

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