source: trunk/jobmond/jobmond.py @ 295

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

jobmond/jobmond.py:

  • applied patch that prevents jobmond from exiting when Torque server is unavailable. Thanks to Peter Kruse <pk@…>
  • Property svn:keywords set to Id
File size: 14.8 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 282 2006-11-29 00:22:03Z 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                api_guess = 'pbs'
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' )
120                api_guess = 'pbs'
121               
122        GMOND_CONF = cfg.get( 'DEFAULT', 'GMOND_CONF' )
123
124        DETECT_TIME_DIFFS = cfg.getboolean( 'DEFAULT', 'DETECT_TIME_DIFFS' )
125
126        BATCH_HOST_TRANSLATE = getlist( cfg.get( 'DEFAULT', 'BATCH_HOST_TRANSLATE' ) )
127
128        try:
129
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       
140        return True
141
142
143import time, os, socket, string, re
144
145METRIC_MAX_VAL_LEN = 900
146
147class DataProcessor:
148        """Class for processing of data"""
149
150        binary = '/usr/bin/gmetric'
151
152        def __init__( self, binary=None ):
153                """Remember alternate binary location if supplied"""
154
155                if binary:
156                        self.binary = binary
157
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.'
164
165                self.dmax = str( int( int( BATCH_POLL_INTERVAL ) * 2 ) )
166
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
177                incompatible = self.checkGmetricVersion()
178
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 ):
184                """
185                Check version of gmetric is at least 3.0.1
186                for the syntax we use
187                """
188
189                global METRIC_MAX_VAL_LEN
190
191                for line in os.popen( self.binary + ' --version' ).readlines():
192
193                        line = line.split( ' ' )
194
195                        if len( line ) == 2 and str(line).find( 'gmetric' ) != -1:
196                       
197                                gmetric_version = line[1].split( '\n' )[0]
198
199                                version_major = int( gmetric_version.split( '.' )[0] )
200                                version_minor = int( gmetric_version.split( '.' )[1] )
201                                version_patch = int( gmetric_version.split( '.' )[2] )
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                                               
215                                                        incompatible = 1
216
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
225                return incompatible
226
227        def multicastGmetric( self, metricname, metricval, valtype='string' ):
228                """Call gmetric binary and multicast"""
229
230                cmd = self.binary
231
232                try:
233                        cmd = cmd + ' -c' + GMOND_CONF
234                except NameError:
235                        debug_msg( 10, 'Assuming /etc/gmond.conf for gmetric cmd (ommitting)' )
236
237                cmd = cmd + ' -n' + str( metricname )+ ' -v"' + str( metricval )+ '" -t' + str( valtype ) + ' -d' + str( self.dmax )
238
239                debug_msg( 10, printTime() + ' ' + cmd )
240                os.system( cmd )
241
242class SgeDataGatherer:
243
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
258        jobs = { }
259
260        global PBSQuery
261
262        def __init__( self ):
263                """Setup appropriate variables"""
264
265                self.jobs = { }
266                self.timeoffset = 0
267                self.dp = DataProcessor()
268                self.initPbsQuery()
269
270        def initPbsQuery( self ):
271
272                self.pq = None
273                if( BATCH_SERVER ):
274                        self.pq = PBSQuery( BATCH_SERVER )
275                else:
276                        self.pq = PBSQuery()
277
278        def getAttr( self, attrs, name ):
279                """Return certain attribute from dictionary, if exists"""
280
281                if attrs.has_key( name ):
282                        return attrs[name]
283                else:
284                        return ''
285
286        def jobDataChanged( self, jobs, job_id, attrs ):
287                """Check if job with attrs and job_id in jobs has changed"""
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
307        def getJobData( self, known_jobs ):
308                """Gather all data on current jobs in Torque"""
309
310                if len( known_jobs ) > 0:
311                        jobs = known_jobs
312                else:
313                        jobs = { }
314
315                #self.initPbsQuery()
316       
317                #print self.pq.getnodes()
318       
319                joblist = {}
320                while len(joblist) == 0:
321                        try:
322                                joblist = self.pq.getjobs()
323                        except PBSError:
324                                time.sleep( TORQUE_POLL_INTERVAL )
325
326                self.cur_time = time.time()
327
328                jobs_processed = [ ]
329
330                #self.printJobs( joblist )
331
332                for name, attrs in joblist.items():
333
334                        job_id = name.split( '.' )[0]
335
336                        jobs_processed.append( job_id )
337
338                        name = self.getAttr( attrs, 'Job_Name' )
339                        queue = self.getAttr( attrs, 'queue' )
340                        owner = self.getAttr( attrs, 'Job_Owner' ).split( '@' )[0]
341                        requested_time = self.getAttr( attrs, 'Resource_List.walltime' )
342                        requested_memory = self.getAttr( attrs, 'Resource_List.mem' )
343
344                        mynoderequest = self.getAttr( attrs, 'Resource_List.nodes' )
345
346                        ppn = ''
347
348                        if mynoderequest.find( ':' ) != -1 and mynoderequest.find( 'ppn' ) != -1:
349
350                                mynoderequest_fields = mynoderequest.split( ':' )
351
352                                for mynoderequest_field in mynoderequest_fields:
353
354                                        if mynoderequest_field.find( 'ppn' ) != -1:
355
356                                                ppn = mynoderequest_field.split( 'ppn=' )[1]
357
358                        status = self.getAttr( attrs, 'job_state' )
359
360                        queued_timestamp = self.getAttr( attrs, 'ctime' )
361
362                        if status == 'R':
363                                start_timestamp = self.getAttr( attrs, 'mtime' )
364                                nodes = self.getAttr( attrs, 'exec_host' ).split( '+' )
365
366                                nodeslist = [ ]
367
368                                for node in nodes:
369                                        host = node.split( '/' )[0]
370
371                                        if nodeslist.count( host ) == 0:
372
373                                                for translate_pattern in BATCH_HOST_TRANSLATE:
374
375                                                        if translate_pattern.find( '/' ) != -1:
376
377                                                                translate_orig = translate_pattern.split( '/' )[1]
378                                                                translate_new = translate_pattern.split( '/' )[2]
379
380                                                                host = re.sub( translate_orig, translate_new, host )
381                               
382                                                if not host in nodeslist:
383                               
384                                                        nodeslist.append( host )
385
386                                if DETECT_TIME_DIFFS:
387
388                                        # If a job start if later than our current date,
389                                        # that must mean the Torque server's time is later
390                                        # than our local time.
391                               
392                                        if int(start_timestamp) > int( int(self.cur_time) + int(self.timeoffset) ):
393
394                                                self.timeoffset = int( int(start_timestamp) - int(self.cur_time) )
395
396                        elif status == 'Q':
397                                start_timestamp = ''
398                                count_mynodes = 0
399                                numeric_node = 1
400
401                                for node in mynoderequest.split( '+' ):
402
403                                        nodepart = node.split( ':' )[0]
404
405                                        for letter in nodepart:
406
407                                                if letter not in string.digits:
408
409                                                        numeric_node = 0
410
411                                        if not numeric_node:
412                                                count_mynodes = count_mynodes + 1
413                                        else:
414                                                count_mynodes = count_mynodes + int( nodepart )
415                                               
416                                nodeslist = str( count_mynodes )
417                        else:
418                                start_timestamp = ''
419                                nodeslist = ''
420
421                        myAttrs = { }
422                        myAttrs['name'] = str( name )
423                        myAttrs['queue'] = str( queue )
424                        myAttrs['owner'] = str( owner )
425                        myAttrs['requested_time'] = str( requested_time )
426                        myAttrs['requested_memory'] = str( requested_memory )
427                        myAttrs['ppn'] = str( ppn )
428                        myAttrs['status'] = str( status )
429                        myAttrs['start_timestamp'] = str( start_timestamp )
430                        myAttrs['queued_timestamp'] = str( queued_timestamp )
431                        myAttrs['reported'] = str( int( int( self.cur_time ) + int( self.timeoffset ) ) )
432                        myAttrs['nodes'] = nodeslist
433                        myAttrs['domain'] = string.join( socket.getfqdn().split( '.' )[1:], '.' )
434                        myAttrs['poll_interval'] = str( BATCH_POLL_INTERVAL )
435
436                        if self.jobDataChanged( jobs, job_id, myAttrs ) and myAttrs['status'] in [ 'R', 'Q' ]:
437                                jobs[ job_id ] = myAttrs
438
439                                #debug_msg( 10, printTime() + ' job %s state changed' %(job_id) )
440
441                for id, attrs in jobs.items():
442
443                        if id not in jobs_processed:
444
445                                # This one isn't there anymore; toedeledoki!
446                                #
447                                del jobs[ id ]
448
449                return jobs
450
451        def submitJobData( self, jobs ):
452                """Submit job info list"""
453
454                self.dp.multicastGmetric( 'MONARCH-HEARTBEAT', str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
455
456                # Now let's spread the knowledge
457                #
458                for jobid, jobattrs in jobs.items():
459
460                        gmetric_val = self.compileGmetricVal( jobid, jobattrs )
461
462                        metric_increment = 0
463
464                        for val in gmetric_val:
465                                self.dp.multicastGmetric( 'MONARCH-JOB-' + jobid + '-' + str(metric_increment), val )
466                                metric_increment = metric_increment + 1
467
468        def compileGmetricVal( self, jobid, jobattrs ):
469                """Create a val string for gmetric of jobinfo"""
470
471                gval_lists = [ ]
472
473                mystr = None
474
475                val_list = { }
476
477                for val_name, val_value in jobattrs.items():
478
479                        val_list_names_len      = len( string.join( val_list.keys() ) ) + len(val_list.keys())
480                        val_list_vals_len       = len( string.join( val_list.values() ) ) + len(val_list.values())
481
482                        if val_name == 'nodes' and jobattrs['status'] == 'R':
483
484                                node_str = None
485
486                                for node in val_value:
487
488                                        if node_str:
489                                                node_str = node_str + ';' + node
490                                        else:
491                                                node_str = node
492
493                                        if (val_list_names_len + len(val_name) ) + (val_list_vals_len + len(node_str) ) > METRIC_MAX_VAL_LEN:
494
495                                                val_list[ val_name ] = node_str
496                                                gval_lists.append( val_list )
497                                                val_list = { }
498                                                node_str = None
499
500                                val_list[ val_name ] = node_str
501                                gval_lists.append( val_list )
502                                val_list = { }
503
504                        elif val_value != '':
505
506                                if (val_list_names_len + len(val_name) ) + (val_list_vals_len + len(str(val_value)) ) > METRIC_MAX_VAL_LEN:
507
508                                        gval_lists.append( val_list )
509                                        val_list = { }
510
511                                val_list[ val_name ] = val_value
512
513                if len(val_list) > 0:
514                        gval_lists.append( val_list )
515
516                str_list = [ ]
517
518                for val_list in gval_lists:
519
520                        my_val_str = None
521
522                        for val_name, val_value in val_list.items():
523
524                                if my_val_str:
525
526                                        my_val_str = my_val_str + ' ' + val_name + '=' + val_value
527                                else:
528                                        my_val_str = val_name + '=' + val_value
529
530                        str_list.append( my_val_str )
531
532                return str_list
533
534        def printJobs( self, jobs ):
535                """Print a jobinfo overview"""
536
537                for name, attrs in self.jobs.items():
538
539                        print 'job %s' %(name)
540
541                        for name, val in attrs.items():
542
543                                print '\t%s = %s' %( name, val )
544
545        def printJob( self, jobs, job_id ):
546                """Print job with job_id from jobs"""
547
548                print 'job %s' %(job_id)
549
550                for name, val in jobs[ job_id ].items():
551
552                        print '\t%s = %s' %( name, val )
553
554        def daemon( self ):
555                """Run as daemon forever"""
556
557                # Fork the first child
558                #
559                pid = os.fork()
560                if pid > 0:
561                        sys.exit(0)  # end parent
562
563                # creates a session and sets the process group ID
564                #
565                os.setsid()
566
567                # Fork the second child
568                #
569                pid = os.fork()
570                if pid > 0:
571                        sys.exit(0)  # end parent
572
573                # Go to the root directory and set the umask
574                #
575                os.chdir('/')
576                os.umask(0)
577
578                sys.stdin.close()
579                sys.stdout.close()
580                sys.stderr.close()
581
582                os.open('/dev/null', os.O_RDWR)
583                os.dup2(0, 1)
584                os.dup2(0, 2)
585
586                self.run()
587
588        def run( self ):
589                """Main thread"""
590
591                while ( 1 ):
592               
593                        self.jobs = self.getJobData( self.jobs )
594                        self.submitJobData( self.jobs )
595                        time.sleep( BATCH_POLL_INTERVAL )       
596
597def printTime( ):
598        """Print current time/date in human readable format for log/debug"""
599
600        return time.strftime("%a, %d %b %Y %H:%M:%S")
601
602def debug_msg( level, msg ):
603        """Print msg if at or above current debug level"""
604
605        if (DEBUG_LEVEL >= level):
606                        sys.stderr.write( msg + '\n' )
607
608def main():
609        """Application start"""
610
611        global PBSQuery
612
613        if not processArgs( sys.argv[1:] ):
614                sys.exit( 1 )
615
616        if BATCH_API == 'pbs':
617
618                try:
619                        from PBSQuery import PBSQuery, PBSError
620
621                except ImportError:
622
623                        debug_msg( 0, "fatal error: BATCH_API set to 'pbs' but python module 'pbs_python' is not installed" )
624                        sys.exit( 1 )
625
626                gather = PbsDataGatherer()
627
628        elif BATCH_API == 'sge':
629
630                pass
631                # import Babu's code here
632                #
633                #try:
634                #       import sge_drmaa
635                #
636                #except ImportError:
637                #
638                #       debug_msg( 0, "fatal error: BATCH_API set to 'sge' but python module 'sge_drmaa' is not installed' )
639                #       sys.exit( 1 )
640                #
641                #gather = SgeDataGatherer()
642
643        else:
644                debug_msg( 0, "fatal error: unknown BATCH_API '" + BATCH_API + "' is not supported" )
645                sys.exit( 1 )
646
647        if DAEMONIZE:
648                gather.daemon()
649        else:
650                gather.run()
651
652# wh00t? someone started me! :)
653#
654if __name__ == '__main__':
655        main()
Note: See TracBrowser for help on using the repository browser.