source: trunk/jobmond/jobmond.py @ 317

Last change on this file since 317 was 317, checked in by bastiaans, 17 years ago

jobmond/jobmond.conf:

  • added QUEUE option

jobmond/jobmond.py:

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