source: trunk/jobmond/jobmond.py @ 265

Last change on this file since 265 was 265, checked in by bastiaans, 16 years ago

jobmond/jobmond.py:

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