source: trunk/jobmond/jobmond.py @ 508

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

jobmond/jobmond.py:

  • fix to queue support by Craig West
  • Property svn:executable set to *
  • Property svn:keywords set to Id
File size: 32.6 KB
RevLine 
[23]1#!/usr/bin/env python
[225]2#
3# This file is part of Jobmonarch
4#
[363]5# Copyright (C) 2006-2007  Ramon Bastiaans
[507]6# Copyright (C) 2007  Dave Love  (SGE code)
[225]7#
8# Jobmonarch is free software; you can redistribute it and/or modify
9# it under the terms of the GNU General Public License as published by
10# the Free Software Foundation; either version 2 of the License, or
11# (at your option) any later version.
12#
13# Jobmonarch is distributed in the hope that it will be useful,
14# but WITHOUT ANY WARRANTY; without even the implied warranty of
15# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
16# GNU General Public License for more details.
17#
18# You should have received a copy of the GNU General Public License
19# along with this program; if not, write to the Free Software
20# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
21#
[228]22# SVN $Id: jobmond.py 508 2008-03-07 16:43:22Z bastiaans $
[227]23#
[23]24
[471]25import sys, getopt, ConfigParser, time, os, socket, string, re
26import xdrlib, socket, syslog, xml, xml.sax
[318]27from xml.sax.handler import feature_namespaces
28
[500]29VERSION='0.3.1'
[307]30
[471]31def usage( ver ):
32
33        print 'jobmond %s' %VERSION
34
35        if ver:
36                return 0
37
[307]38        print
[471]39        print 'Purpose:'
40        print '  The Job Monitoring Daemon (jobmond) reports batch jobs information and statistics'
41        print '  to Ganglia, which can be viewed with Job Monarch web frontend'
[307]42        print
[471]43        print 'Usage:   jobmond [OPTIONS]'
44        print
45        print '  -c, --config=FILE      The configuration file to use (default: /etc/jobmond.conf)'
46        print '  -p, --pidfile=FILE     Use pid file to store the process id'
47        print '  -h, --help             Print help and exit'
48        print '  -v, --version          Print version and exit'
49        print
[307]50
[212]51def processArgs( args ):
[26]52
[471]53        SHORT_L         = 'p:hvc:'
54        LONG_L          = [ 'help', 'config=', 'pidfile=', 'version' ]
[165]55
[307]56        global PIDFILE
[354]57        PIDFILE         = None
[61]58
[354]59        config_filename = '/etc/jobmond.conf'
60
[212]61        try:
[68]62
[354]63                opts, args      = getopt.getopt( args, SHORT_L, LONG_L )
[185]64
[307]65        except getopt.GetoptError, detail:
[212]66
67                print detail
[307]68                usage()
[354]69                sys.exit( 1 )
[212]70
71        for opt, value in opts:
72
73                if opt in [ '--config', '-c' ]:
74               
[354]75                        config_filename = value
[212]76
[307]77                if opt in [ '--pidfile', '-p' ]:
[212]78
[354]79                        PIDFILE         = value
[307]80               
81                if opt in [ '--help', '-h' ]:
82 
[471]83                        usage( False )
[354]84                        sys.exit( 0 )
[212]85
[476]86                if opt in [ '--version', '-v' ]:
[471]87
[476]88                        usage( True )
[471]89                        sys.exit( 0 )
90
[212]91        return loadConfig( config_filename )
92
93def loadConfig( filename ):
94
[215]95        def getlist( cfg_string ):
96
97                my_list = [ ]
98
99                for item_txt in cfg_string.split( ',' ):
100
101                        sep_char = None
102
103                        item_txt = item_txt.strip()
104
105                        for s_char in [ "'", '"' ]:
106
107                                if item_txt.find( s_char ) != -1:
108
109                                        if item_txt.count( s_char ) != 2:
110
111                                                print 'Missing quote: %s' %item_txt
112                                                sys.exit( 1 )
113
114                                        else:
115
116                                                sep_char = s_char
117                                                break
118
119                        if sep_char:
120
121                                item_txt = item_txt.split( sep_char )[1]
122
123                        my_list.append( item_txt )
124
125                return my_list
126
[354]127        cfg             = ConfigParser.ConfigParser()
[212]128
129        cfg.read( filename )
130
[354]131        global DEBUG_LEVEL, DAEMONIZE, BATCH_SERVER, BATCH_POLL_INTERVAL
132        global GMOND_CONF, DETECT_TIME_DIFFS, BATCH_HOST_TRANSLATE
[373]133        global BATCH_API, QUEUE, GMETRIC_TARGET, USE_SYSLOG
[449]134        global SYSLOG_LEVEL, SYSLOG_FACILITY, GMETRIC_BINARY
[212]135
[354]136        DEBUG_LEVEL     = cfg.getint( 'DEFAULT', 'DEBUG_LEVEL' )
[212]137
[354]138        DAEMONIZE       = cfg.getboolean( 'DEFAULT', 'DAEMONIZE' )
[212]139
[377]140        SYSLOG_LEVEL    = -1
141        SYSLOG_FACILITY = None
142
[265]143        try:
[373]144                USE_SYSLOG      = cfg.getboolean( 'DEFAULT', 'USE_SYSLOG' )
[212]145
[373]146        except ConfigParser.NoOptionError:
147
148                USE_SYSLOG      = True
149
150                debug_msg( 0, 'ERROR: no option USE_SYSLOG found: assuming yes' )
151
[377]152
[449]153
[373]154        if USE_SYSLOG:
155
156                try:
157                        SYSLOG_LEVEL    = cfg.getint( 'DEFAULT', 'SYSLOG_LEVEL' )
158
159                except ConfigParser.NoOptionError:
160
161                        debug_msg( 0, 'ERROR: no option SYSLOG_LEVEL found: assuming level 0' )
162                        SYSLOG_LEVEL    = 0
163
164                try:
165
166                        SYSLOG_FACILITY = eval( 'syslog.LOG_' + cfg.get( 'DEFAULT', 'SYSLOG_FACILITY' ) )
167
[377]168                except ConfigParser.NoOptionError:
[373]169
170                        SYSLOG_FACILITY = syslog.LOG_DAEMON
171
172                        debug_msg( 0, 'ERROR: no option SYSLOG_FACILITY found: assuming facility DAEMON' )
173
174        try:
175
[354]176                BATCH_SERVER            = cfg.get( 'DEFAULT', 'BATCH_SERVER' )
[212]177
[265]178        except ConfigParser.NoOptionError:
179
180                # Backwards compatibility for old configs
181                #
182
[354]183                BATCH_SERVER            = cfg.get( 'DEFAULT', 'TORQUE_SERVER' )
184                api_guess               = 'pbs'
[265]185       
186        try:
187       
[354]188                BATCH_POLL_INTERVAL     = cfg.getint( 'DEFAULT', 'BATCH_POLL_INTERVAL' )
[265]189
190        except ConfigParser.NoOptionError:
191
192                # Backwards compatibility for old configs
193                #
194
[354]195                BATCH_POLL_INTERVAL     = cfg.getint( 'DEFAULT', 'TORQUE_POLL_INTERVAL' )
196                api_guess               = 'pbs'
[353]197       
198        try:
[212]199
[354]200                GMOND_CONF              = cfg.get( 'DEFAULT', 'GMOND_CONF' )
[353]201
202        except ConfigParser.NoOptionError:
203
[354]204                GMOND_CONF              = None
[353]205
[449]206        try:
207
208                GMETRIC_BINARY          = cfg.get( 'DEFAULT', 'GMETRIC_BINARY' )
209
210        except ConfigParser.NoOptionError:
211
212                GMETRIC_BINARY          = '/usr/bin/gmetric'
213
[354]214        DETECT_TIME_DIFFS       = cfg.getboolean( 'DEFAULT', 'DETECT_TIME_DIFFS' )
[212]215
[354]216        BATCH_HOST_TRANSLATE    = getlist( cfg.get( 'DEFAULT', 'BATCH_HOST_TRANSLATE' ) )
[215]217
[266]218        try:
[256]219
[354]220                BATCH_API       = cfg.get( 'DEFAULT', 'BATCH_API' )
[266]221
222        except ConfigParser.NoOptionError, detail:
223
224                if BATCH_SERVER and api_guess:
[354]225
226                        BATCH_API       = api_guess
[266]227                else:
[373]228                        debug_msg( 0, "FATAL ERROR: BATCH_API not set and can't make guess" )
[266]229                        sys.exit( 1 )
[317]230
231        try:
232
[354]233                QUEUE           = getlist( cfg.get( 'DEFAULT', 'QUEUE' ) )
[317]234
235        except ConfigParser.NoOptionError, detail:
236
[354]237                QUEUE           = None
[353]238
239        try:
240
[354]241                GMETRIC_TARGET  = cfg.get( 'DEFAULT', 'GMETRIC_TARGET' )
[353]242
243        except ConfigParser.NoOptionError:
244
[354]245                GMETRIC_TARGET  = None
[353]246
247                if not GMOND_CONF:
248
[373]249                        debug_msg( 0, "FATAL ERROR: GMETRIC_TARGET and GMOND_CONF both not set! Set at least one!" )
[353]250                        sys.exit( 1 )
251                else:
252
[507]253                        debug_msg( 0, "ERROR: GMETRIC_TARGET not set: internal Gmetric handling aborted. Failing back to DEPRECATED use of gmond.conf/gmetric binary. This will slow down jobmond significantly!" )
[353]254
[212]255        return True
256
[507]257def fqdn_parts (fqdn):
258        """Return pair of host and domain for fully-qualified domain name arg."""
259        parts = fqdn.split (".")
260        return (parts[0], string.join(parts[1:], "."))
261
[253]262METRIC_MAX_VAL_LEN = 900
263
[61]264class DataProcessor:
[355]265
[68]266        """Class for processing of data"""
[61]267
[449]268        binary = None
[61]269
270        def __init__( self, binary=None ):
[355]271
[68]272                """Remember alternate binary location if supplied"""
[61]273
[449]274                global GMETRIC_BINARY
275
[61]276                if binary:
277                        self.binary = binary
278
[449]279                if not self.binary:
280                        self.binary = GMETRIC_BINARY
281
[80]282                # Timeout for XML
283                #
284                # From ganglia's documentation:
285                #
286                # 'A metric will be deleted DMAX seconds after it is received, and
287                # DMAX=0 means eternal life.'
[61]288
[256]289                self.dmax = str( int( int( BATCH_POLL_INTERVAL ) * 2 ) )
[80]290
[353]291                if GMOND_CONF:
[354]292
[353]293                        try:
294                                gmond_file = GMOND_CONF
[68]295
[353]296                        except NameError:
297                                gmond_file = '/etc/gmond.conf'
[68]298
[353]299                        if not os.path.exists( gmond_file ):
[373]300                                debug_msg( 0, 'FATAL ERROR: ' + gmond_file + ' does not exist' )
[353]301                                sys.exit( 1 )
[68]302
[353]303                        incompatible = self.checkGmetricVersion()
[61]304
[353]305                        if incompatible:
[355]306
307                                debug_msg( 0, 'Gmetric version not compatible, please upgrade to at least 3.0.1' )
[353]308                                sys.exit( 1 )
[65]309
310        def checkGmetricVersion( self ):
[355]311
[68]312                """
313                Check version of gmetric is at least 3.0.1
314                for the syntax we use
315                """
[65]316
[255]317                global METRIC_MAX_VAL_LEN
318
[341]319                incompatible    = 0
320
[355]321                gfp             = os.popen( self.binary + ' --version' )
322                lines           = gfp.readlines()
[65]323
[355]324                gfp.close()
325
326                for line in lines:
327
[65]328                        line = line.split( ' ' )
329
[355]330                        if len( line ) == 2 and str( line ).find( 'gmetric' ) != -1:
[65]331                       
[355]332                                gmetric_version = line[1].split( '\n' )[0]
[65]333
[355]334                                version_major   = int( gmetric_version.split( '.' )[0] )
335                                version_minor   = int( gmetric_version.split( '.' )[1] )
336                                version_patch   = int( gmetric_version.split( '.' )[2] )
[65]337
[355]338                                incompatible    = 0
[65]339
340                                if version_major < 3:
341
342                                        incompatible = 1
343                               
344                                elif version_major == 3:
345
346                                        if version_minor == 0:
347
348                                                if version_patch < 1:
349                                               
[91]350                                                        incompatible = 1
[65]351
[471]352                                                # Gmetric 3.0.1 >< 3.0.3 had a bug in the max metric length
353                                                #
[255]354                                                if version_patch < 3:
355
356                                                        METRIC_MAX_VAL_LEN = 900
357
358                                                elif version_patch >= 3:
359
360                                                        METRIC_MAX_VAL_LEN = 1400
361
[65]362                return incompatible
363
[409]364        def multicastGmetric( self, metricname, metricval, valtype='string', units='' ):
[355]365
[68]366                """Call gmetric binary and multicast"""
[65]367
368                cmd = self.binary
369
[353]370                if GMETRIC_TARGET:
[61]371
[353]372                        GMETRIC_TARGET_HOST     = GMETRIC_TARGET.split( ':' )[0]
373                        GMETRIC_TARGET_PORT     = GMETRIC_TARGET.split( ':' )[1]
374
375                        metric_debug            = "[gmetric] name: %s - val: %s - dmax: %s" %( str( metricname ), str( metricval ), str( self.dmax ) )
376
377                        debug_msg( 10, printTime() + ' ' + metric_debug)
378
379                        gm = Gmetric( GMETRIC_TARGET_HOST, GMETRIC_TARGET_PORT )
380
[425]381                        gm.send( str( metricname ), str( metricval ), str( self.dmax ), valtype, units )
[353]382
383                else:
384                        try:
385                                cmd = cmd + ' -c' + GMOND_CONF
386
387                        except NameError:
388
[507]389                                debug_msg( 10, 'Assuming /etc/gmond.conf for gmetric cmd (omitting)' )
[353]390
391                        cmd = cmd + ' -n' + str( metricname )+ ' -v"' + str( metricval )+ '" -t' + str( valtype ) + ' -d' + str( self.dmax )
392
[409]393                        if len( units ) > 0:
394
395                                cmd = cmd + ' -u"' + units + '"'
396
[353]397                        debug_msg( 10, printTime() + ' ' + cmd )
398
399                        os.system( cmd )
400
[318]401class DataGatherer:
[23]402
[318]403        """Skeleton class for batch system DataGatherer"""
[256]404
[318]405        def printJobs( self, jobs ):
[355]406
[318]407                """Print a jobinfo overview"""
408
409                for name, attrs in self.jobs.items():
410
411                        print 'job %s' %(name)
412
413                        for name, val in attrs.items():
414
415                                print '\t%s = %s' %( name, val )
416
417        def printJob( self, jobs, job_id ):
[355]418
[318]419                """Print job with job_id from jobs"""
420
421                print 'job %s' %(job_id)
422
423                for name, val in jobs[ job_id ].items():
424
425                        print '\t%s = %s' %( name, val )
426
[507]427        def getAttr( self, attrs, name ):
428
429                """Return certain attribute from dictionary, if exists"""
430
431                if attrs.has_key( name ):
432
433                        return attrs[ name ]
434                else:
435                        return ''
436
437        def jobDataChanged( self, jobs, job_id, attrs ):
438
439                """Check if job with attrs and job_id in jobs has changed"""
440
441                if jobs.has_key( job_id ):
442
443                        oldData = jobs[ job_id ]       
444                else:
445                        return 1
446
447                for name, val in attrs.items():
448
449                        if oldData.has_key( name ):
450
451                                if oldData[ name ] != attrs[ name ]:
452
453                                        return 1
454
455                        else:
456                                return 1
457
458                return 0
459
460        def submitJobData( self ):
461
462                """Submit job info list"""
463
464                self.dp.multicastGmetric( 'MONARCH-HEARTBEAT', str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
465
466                running_jobs    = 0
467                queued_jobs     = 0
468
469                # Count how many running/queued jobs we found
470                #
471                for jobid, jobattrs in self.jobs.items():
472
473                        if jobattrs[ 'status' ] == 'Q':
474
475                                queued_jobs += 1
476
477                        elif jobattrs[ 'status' ] == 'R':
478
479                                running_jobs += 1
480
481                # Report running/queued jobs as seperate metric for a nice RRD graph
482                #
483                self.dp.multicastGmetric( 'MONARCH-RJ', str( running_jobs ), 'uint32', 'jobs' )
484                self.dp.multicastGmetric( 'MONARCH-QJ', str( queued_jobs ), 'uint32', 'jobs' )
485
486                # Now let's spread the knowledge
487                #
488                for jobid, jobattrs in self.jobs.items():
489
490                        # Make gmetric values for each job: respect max gmetric value length
491                        #
492                        gmetric_val             = self.compileGmetricVal( jobid, jobattrs )
493                        metric_increment        = 0
494
495                        # If we have more job info than max gmetric value length allows, split it up
496                        # amongst multiple metrics
497                        #
498                        for val in gmetric_val:
499
500                                self.dp.multicastGmetric( 'MONARCH-JOB-' + jobid + '-' + str(metric_increment), val )
501
502                                # Increase follow number if this jobinfo is split up amongst more than 1 gmetric
503                                #
504                                metric_increment        = metric_increment + 1
505
506        def compileGmetricVal( self, jobid, jobattrs ):
507
508                """Create a val string for gmetric of jobinfo"""
509
510                gval_lists      = [ ]
511                val_list        = { }
512
513                for val_name, val_value in jobattrs.items():
514
515                        # These are our own metric names, i.e.: status, start_timestamp, etc
516                        #
517                        val_list_names_len      = len( string.join( val_list.keys() ) ) + len(val_list.keys())
518
519                        # These are their corresponding values
520                        #
521                        val_list_vals_len       = len( string.join( val_list.values() ) ) + len(val_list.values())
522
523                        if val_name == 'nodes' and jobattrs['status'] == 'R':
524
525                                node_str = None
526
527                                for node in val_value:
528
529                                        if node_str:
530
531                                                node_str = node_str + ';' + node
532                                        else:
533                                                node_str = node
534
535                                        # Make sure if we add this new info, that the total metric's value length does not exceed METRIC_MAX_VAL_LEN
536                                        #
537                                        if (val_list_names_len + len(val_name) ) + (val_list_vals_len + len(node_str) ) > METRIC_MAX_VAL_LEN:
538
539                                                # It's too big, we need to make a new gmetric for the additional info
540                                                #
541                                                val_list[ val_name ]    = node_str
542
543                                                gval_lists.append( val_list )
544
545                                                val_list                = { }
546                                                node_str                = None
547
548                                val_list[ val_name ]    = node_str
549
550                                gval_lists.append( val_list )
551
552                                val_list                = { }
553
554                        elif val_value != '':
555
556                                # Make sure if we add this new info, that the total metric's value length does not exceed METRIC_MAX_VAL_LEN
557                                #
558                                if (val_list_names_len + len(val_name) ) + (val_list_vals_len + len(str(val_value)) ) > METRIC_MAX_VAL_LEN:
559
560                                        # It's too big, we need to make a new gmetric for the additional info
561                                        #
562                                        gval_lists.append( val_list )
563
564                                        val_list                = { }
565
566                                val_list[ val_name ]    = val_value
567
568                if len( val_list ) > 0:
569
570                        gval_lists.append( val_list )
571
572                str_list        = [ ]
573
574                # Now append the value names and values together, i.e.: stop_timestamp=value, etc
575                #
576                for val_list in gval_lists:
577
578                        my_val_str      = None
579
580                        for val_name, val_value in val_list.items():
581
582                                if my_val_str:
583
584                                        my_val_str = my_val_str + ' ' + val_name + '=' + val_value
585                                else:
586                                        my_val_str = val_name + '=' + val_value
587
588                        str_list.append( my_val_str )
589
590                return str_list
591
[256]592        def daemon( self ):
[355]593
[318]594                """Run as daemon forever"""
[256]595
[318]596                # Fork the first child
597                #
598                pid = os.fork()
599                if pid > 0:
600                        sys.exit(0)  # end parent
[256]601
[318]602                # creates a session and sets the process group ID
603                #
604                os.setsid()
605
606                # Fork the second child
607                #
608                pid = os.fork()
609                if pid > 0:
610                        sys.exit(0)  # end parent
611
612                write_pidfile()
613
614                # Go to the root directory and set the umask
615                #
616                os.chdir('/')
617                os.umask(0)
618
619                sys.stdin.close()
620                sys.stdout.close()
621                sys.stderr.close()
622
623                os.open('/dev/null', os.O_RDWR)
624                os.dup2(0, 1)
625                os.dup2(0, 2)
626
627                self.run()
628
[256]629        def run( self ):
[355]630
[318]631                """Main thread"""
[256]632
[318]633                while ( 1 ):
634               
[348]635                        self.getJobData()
636                        self.submitJobData()
[318]637                        time.sleep( BATCH_POLL_INTERVAL )       
[256]638
[507]639# SGE code by Dave Love <fx@gnu.org>.  Tested with SGE 6.0u8 and 6.0u11.
640# Probably needs modification for SGE 6.1.  See also the fixmes.
[256]641
[507]642class NoJobs (Exception):
643        """Exception raised by empty job list in qstat output."""
644        pass
[256]645
[507]646class SgeQstatXMLParser(xml.sax.handler.ContentHandler):
647        """SAX handler for XML output from Sun Grid Engine's `qstat'."""
[318]648
[507]649        def __init__(self):
650                self.value = ""
651                self.joblist = []
652                self.job = {}
653                self.queue = ""
654                self.in_joblist = False
655                self.lrequest = False
656                xml.sax.handler.ContentHandler.__init__(self)
[318]657
[507]658        # The structure of the output is as follows.  Unfortunately
659        # it's voluminous, and probably doesn't scale to large
660        # clusters/queues.
[318]661
[507]662        # <detailed_job_info  xmlns:xsd="http://www.w3.org/2001/XMLSchema">
663        #   <djob_info>
664        #     <qmaster_response>  <!-- job -->
665        #       ...
666        #       <JB_ja_template> 
667        #         <ulong_sublist>
668        #         ...             <!-- start_time, state ... -->
669        #         </ulong_sublist>
670        #       </JB_ja_template> 
671        #       <JB_ja_tasks>
672        #         <ulong_sublist>
673        #           ...           <!-- task info
674        #         </ulong_sublist>
675        #         ...
676        #       </JB_ja_tasks>
677        #       ...
678        #     </qmaster_response>
679        #   </djob_info>
680        #   <messages>
681        #   ...
[318]682
[507]683        # NB.  We might treat each task as a separate job, like
684        # straight qstat output, but the web interface expects jobs to
685        # be identified by integers, not, say, <job number>.<task>.
[318]686
[507]687        # So, I lied.  If the job list is empty, we get invalid XML
688        # like this, which we need to defend against:
[318]689
[507]690        # <unknown_jobs  xmlns:xsd="http://www.w3.org/2001/XMLSchema">
691        #   <>
692        #     <ST_name>*</ST_name>
693        #   </>
694        # </unknown_jobs>
[318]695
[507]696        def startElement(self, name, attrs):
697                self.value = ""
698                if name == "djob_info": # job list
699                        self.in_joblist = True
700                elif name == "qmaster_response" and self.in_joblist: # job
701                        self.job = {"job_state": "U", "slots": 0,
702                                    "nodes": [], "queued_timestamp": "",
703                                    "queued_timestamp": "", "queue": "",
704                                    "ppn": "0", "RN_max": 0,
705                                    # fixme in endElement
706                                    "requested_memory": 0, "requested_time": 0
707                                    }
708                        self.joblist.append(self.job)
709                elif name == "qstat_l_requests": # resource request
710                        self.lrequest = True
711                elif name == "unknown_jobs":
712                        raise NoJobs
[318]713
[507]714        def characters(self, ch):
715                self.value += ch
[318]716
[507]717        def endElement(self, name): 
718                """Snarf job elements contents into job dictionary.
719                   Translate keys if appropriate."""
[318]720
[507]721                name_trans = {
722                  "JB_job_number": "number",
723                  "JB_job_name": "name", "JB_owner": "owner",
724                  "queue_name": "queue", "JAT_start_time": "start_timestamp",
725                  "JB_submission_time": "queued_timestamp"
726                  }
727                value = self.value
[318]728
[507]729                if name == "djob_info":
730                        self.in_joblist = False
731                        self.job = {}
732                elif name == "JAT_master_queue":
733                        self.job["queue"] = value.split("@")[0]
734                elif name == "JG_qhostname":
735                        if not (value in self.job["nodes"]):
736                                self.job["nodes"].append(value)
737                elif name == "JG_slots": # slots in use
738                        self.job["slots"] += int(value)
739                elif name == "RN_max": # requested slots (tasks or parallel)
740                        self.job["RN_max"] = max (self.job["RN_max"],
741                                                  int(value))
742                elif name == "JAT_state": # job state (bitwise or)
743                        value = int (value)
744                        # Status values from sge_jobL.h
745                        #define JIDLE                   0x00000000
746                        #define JHELD                   0x00000010
747                        #define JMIGRATING              0x00000020
748                        #define JQUEUED                 0x00000040
749                        #define JRUNNING                0x00000080
750                        #define JSUSPENDED              0x00000100
751                        #define JTRANSFERING            0x00000200
752                        #define JDELETED                0x00000400
753                        #define JWAITING                0x00000800
754                        #define JEXITING                0x00001000
755                        #define JWRITTEN                0x00002000
756                        #define JSUSPENDED_ON_THRESHOLD 0x00010000
757                        #define JFINISHED               0x00010000
758                        if value & 0x80:
759                                self.job["status"] = "R"
760                        elif value & 0x40:
761                                self.job["status"] = "Q"
762                        else:
763                                self.job["status"] = "O" # `other'
764                elif name == "CE_name" and self.lrequest and self.value in \
765                            ("h_cpu", "s_cpu", "cpu", "h_core", "s_core"):
766                        # We're in a container for an interesting resource
767                        # request; record which type.
768                        self.lrequest = self.value
769                elif name == "CE_doubleval" and self.lrequest:
770                        # if we're in a container for an interesting
771                        # resource request, use the maxmimum of the hard
772                        # and soft requests to record the requested CPU
773                        # or core.  Fixme:  I'm not sure if this logic is
774                        # right.
775                        if self.lrequest in ("h_core", "s_core"):
776                                self.job["requested_memory"] = \
777                                    max (float (value),
778                                         self.job["requested_memory"])
779                        # Fixme:  Check what cpu means, c.f [hs]_cpu.
780                        elif self.lrequest in ("h_cpu", "s_cpu", "cpu"):
781                                self.job["requested_time"] = \
782                                    max (float (value),
783                                         self.job["requested_time"])
784                elif name == "qstat_l_requests":
785                        self.lrequest = False
786                elif self.job and self.in_joblist:
787                        if name in name_trans:
788                                name = name_trans[name]
789                                self.job[name] = value
[318]790
[507]791# Abstracted from PBS original.
792# Fixme:  Is it worth (or appropriate for PBS) sorting the result?
793def do_nodelist (nodes):
794        """Translate node list as appropriate."""
795        nodeslist               = [ ]
796        my_domain = fqdn_parts(socket.getfqdn())[1]
797        for node in nodes:
798                host            = node.split( '/' )[0] # not relevant for SGE
799                h, host_domain  = fqdn_parts(host)
800                if host_domain == my_domain:
801                        host    = h
802                if nodeslist.count( host ) == 0:
803                        for translate_pattern in BATCH_HOST_TRANSLATE:
804                                if translate_pattern.find( '/' ) != -1:
805                                        translate_orig  = \
806                                            translate_pattern.split( '/' )[1]
807                                        translate_new   = \
808                                            translate_pattern.split( '/' )[2]
809                                        host = re.sub( translate_orig,
810                                                       translate_new, host )
811                        if not host in nodeslist:
812                                nodeslist.append( host )
813        return nodeslist
[318]814
815class SgeDataGatherer(DataGatherer):
816
[507]817        jobs = {}
[61]818
[318]819        def __init__( self ):
[507]820                self.jobs = {}
[318]821                self.timeoffset = 0
822                self.dp = DataProcessor()
823
[507]824        def getJobData( self ):
[318]825                """Gather all data on current jobs in SGE"""
826
[507]827                import popen2
[318]828
[507]829                self.cur_time = 0
830                queues = ""
831                if QUEUE:       # only for specific queues
832                        # Fixme:  assumes queue names don't contain single
833                        # quote or comma.  Don't know what the SGE rules are.
834                        queues = " -q '" + string.join (QUEUE, ",") + "'"
835                # Note the comment in SgeQstatXMLParser about scaling with
836                # this method of getting data.  I haven't found better one.
837                # Output with args `-xml -ext -f -r' is easier to parse
838                # in some ways, harder in others, but it doesn't provide
839                # the submission time, at least.
840                piping = popen2.Popen3("qstat -u '*' -j '*' -xml" + queues,
841                                       True)
842                qstatparser = SgeQstatXMLParser()
843                parse_err = 0
844                try:
845                        xml.sax.parse(piping.fromchild, qstatparser)
846                except NoJobs:
847                        pass
848                except:
849                        parse_err = 1
850                if piping.wait():
851                        debug_msg(10,
852                                  "qstat error, skipping until next polling interval: "
853                                  + piping.childerr.readline())
854                        return None
855                elif parse_err:
856                        debug_msg(10, "Bad XML output from qstat"())
857                        exit (1)
858                for f in piping.fromchild, piping.tochild, piping.childerr:
859                        f.close()
[318]860                self.cur_time = time.time()
[507]861                jobs_processed = []
862                for job in qstatparser.joblist:
863                        job_id = job["number"]
864                        if job["status"] in [ 'Q', 'R' ]:
865                                jobs_processed.append(job_id)
866                        if job["status"] == "R":
867                                job["nodes"] = do_nodelist (job["nodes"])
868                                # Fixme: Is this right?
869                                job["ppn"] = float(job["slots"]) / \
870                                    len(job["nodes"])
871                                if DETECT_TIME_DIFFS:
872                                        # If a job start is later than our
873                                        # current date, that must mean
874                                        # the SGE server's time is later
875                                        # than our local time.
876                                        start_timestamp = \
877                                            int (job["start_timestamp"])
878                                        if start_timestamp > \
879                                                    int(self.cur_time) + \
880                                                    int(self.timeoffset):
[318]881
[507]882                                                self.timeoffset = \
883                                                    start_timestamp - \
884                                                    int(self.cur_time)
885                        else:
886                                # fixme: Note sure what this should be:
887                                job["ppn"] = job["RN_max"]
888                                job["nodes"] = "1"
[318]889
[507]890                        myAttrs = {}
891                        for attr in ["name", "queue", "owner",
892                                     "requested_time", "status",
893                                     "requested_memory", "ppn",
894                                     "start_timestamp", "queued_timestamp"]:
895                                myAttrs[attr] = str(job[attr])
896                        myAttrs["nodes"] = job["nodes"]
897                        myAttrs["reported"] = str(int(self.cur_time) + \
898                                                  int(self.timeoffset))
899                        myAttrs["domain"] = fqdn_parts(socket.getfqdn())[1]
900                        myAttrs["poll_interval"] = str(BATCH_POLL_INTERVAL)
[318]901
[507]902                        if self.jobDataChanged(self.jobs, job_id, myAttrs) \
903                                    and myAttrs["status"] in ["R", "Q"]:
904                                self.jobs[job_id] = myAttrs
905                for id, attrs in self.jobs.items():
906                        if id not in jobs_processed:
907                                del self.jobs[id]
[318]908
[355]909class PbsDataGatherer( DataGatherer ):
[318]910
911        """This is the DataGatherer for PBS and Torque"""
912
[256]913        global PBSQuery
914
[23]915        def __init__( self ):
[354]916
[68]917                """Setup appropriate variables"""
[23]918
[354]919                self.jobs       = { }
920                self.timeoffset = 0
921                self.dp         = DataProcessor()
922
[91]923                self.initPbsQuery()
[23]924
[91]925        def initPbsQuery( self ):
926
[354]927                self.pq         = None
928
[256]929                if( BATCH_SERVER ):
[354]930
931                        self.pq         = PBSQuery( BATCH_SERVER )
[174]932                else:
[354]933                        self.pq         = PBSQuery()
[91]934
[348]935        def getJobData( self ):
[354]936
[68]937                """Gather all data on current jobs in Torque"""
[26]938
[354]939                joblist         = {}
[359]940                self.cur_time   = 0
[349]941
[359]942                try:
943                        joblist         = self.pq.getjobs()
944                        self.cur_time   = time.time()
[354]945
[359]946                except PBSError, detail:
[354]947
[359]948                        debug_msg( 10, "Caught PBS unavailable, skipping until next polling interval: " + str( detail ) )
949                        return None
[354]950
951                jobs_processed  = [ ]
[26]952
953                for name, attrs in joblist.items():
[508]954                        display_queue           = 1
[354]955                        job_id                  = name.split( '.' )[0]
[26]956
[354]957                        name                    = self.getAttr( attrs, 'Job_Name' )
958                        queue                   = self.getAttr( attrs, 'queue' )
[317]959
960                        if QUEUE:
[508]961                                for q in QUEUE:
962                                        if q == queue:
963                                                display_queue = 1
964                                                break
965                                        else:
966                                                display_queue = 0
967                                                continue
968                        if display_queue == 0:
969                                continue
[317]970
971
[354]972                        owner                   = self.getAttr( attrs, 'Job_Owner' ).split( '@' )[0]
973                        requested_time          = self.getAttr( attrs, 'Resource_List.walltime' )
974                        requested_memory        = self.getAttr( attrs, 'Resource_List.mem' )
[95]975
[354]976                        mynoderequest           = self.getAttr( attrs, 'Resource_List.nodes' )
[95]977
[354]978                        ppn                     = ''
[281]979
[26]980                        if mynoderequest.find( ':' ) != -1 and mynoderequest.find( 'ppn' ) != -1:
[95]981
[354]982                                mynoderequest_fields    = mynoderequest.split( ':' )
[281]983
984                                for mynoderequest_field in mynoderequest_fields:
985
986                                        if mynoderequest_field.find( 'ppn' ) != -1:
987
[354]988                                                ppn     = mynoderequest_field.split( 'ppn=' )[1]
[281]989
[354]990                        status                  = self.getAttr( attrs, 'job_state' )
[25]991
[450]992                        if status in [ 'Q', 'R' ]:
993
994                                jobs_processed.append( job_id )
995
[354]996                        queued_timestamp        = self.getAttr( attrs, 'ctime' )
[243]997
[95]998                        if status == 'R':
[133]999
[354]1000                                start_timestamp         = self.getAttr( attrs, 'mtime' )
1001                                nodes                   = self.getAttr( attrs, 'exec_host' ).split( '+' )
[133]1002
[507]1003                                nodeslist               = do_nodelist( nodes )
[354]1004
[185]1005                                if DETECT_TIME_DIFFS:
1006
1007                                        # If a job start if later than our current date,
1008                                        # that must mean the Torque server's time is later
1009                                        # than our local time.
1010                               
[354]1011                                        if int( start_timestamp ) > int( int( self.cur_time ) + int( self.timeoffset ) ):
[185]1012
[354]1013                                                self.timeoffset = int( int(start_timestamp) - int(self.cur_time) )
[185]1014
[133]1015                        elif status == 'Q':
[95]1016
[451]1017                                # 'mynodequest' can be a string in the following syntax according to the
1018                                # Torque Administator's manual:
1019                                #
1020                                # {<node_count> | <hostname>}[:ppn=<ppn>][:<property>[:<property>]...][+ ...]
1021                                # {<node_count> | <hostname>}[:ppn=<ppn>][:<property>[:<property>]...][+ ...]
1022                                # etc
1023                                #
1024
1025                                #
1026                                # For now we only count the amount of nodes request and ignore properties
1027                                #
1028
[354]1029                                start_timestamp         = ''
1030                                count_mynodes           = 0
1031
[133]1032                                for node in mynoderequest.split( '+' ):
[67]1033
[451]1034                                        # Just grab the {node_count|hostname} part and ignore properties
1035                                        #
[354]1036                                        nodepart        = node.split( ':' )[0]
[67]1037
[451]1038                                        # Let's assume a node_count value
1039                                        #
1040                                        numeric_node    = 1
1041
1042                                        # Chop the value up into characters
1043                                        #
[133]1044                                        for letter in nodepart:
[67]1045
[451]1046                                                # If this char is not a digit (0-9), this must be a hostname
1047                                                #
[133]1048                                                if letter not in string.digits:
1049
[354]1050                                                        numeric_node    = 0
[133]1051
[451]1052                                        # If this is a hostname, just count this as one (1) node
1053                                        #
[133]1054                                        if not numeric_node:
[354]1055
1056                                                count_mynodes   = count_mynodes + 1
[133]1057                                        else:
[451]1058
1059                                                # If this a number, it must be the node_count
1060                                                # and increase our count with it's value
1061                                                #
[327]1062                                                try:
[354]1063                                                        count_mynodes   = count_mynodes + int( nodepart )
1064
[327]1065                                                except ValueError, detail:
[354]1066
[451]1067                                                        # When we arrive here I must be bugged or very confused
1068                                                        # THIS SHOULD NOT HAPPEN!
1069                                                        #
[327]1070                                                        debug_msg( 10, str( detail ) )
1071                                                        debug_msg( 10, "Encountered weird node in Resources_List?!" )
1072                                                        debug_msg( 10, 'nodepart = ' + str( nodepart ) )
1073                                                        debug_msg( 10, 'job = ' + str( name ) )
1074                                                        debug_msg( 10, 'attrs = ' + str( attrs ) )
[133]1075                                               
[354]1076                                nodeslist       = str( count_mynodes )
[172]1077                        else:
[354]1078                                start_timestamp = ''
1079                                nodeslist       = ''
[133]1080
[354]1081                        myAttrs                         = { }
[26]1082
[471]1083                        myAttrs[ 'name' ]               = str( name )
[354]1084                        myAttrs[ 'queue' ]              = str( queue )
1085                        myAttrs[ 'owner' ]              = str( owner )
1086                        myAttrs[ 'requested_time' ]     = str( requested_time )
1087                        myAttrs[ 'requested_memory' ]   = str( requested_memory )
1088                        myAttrs[ 'ppn' ]                = str( ppn )
1089                        myAttrs[ 'status' ]             = str( status )
1090                        myAttrs[ 'start_timestamp' ]    = str( start_timestamp )
1091                        myAttrs[ 'queued_timestamp' ]   = str( queued_timestamp )
1092                        myAttrs[ 'reported' ]           = str( int( int( self.cur_time ) + int( self.timeoffset ) ) )
1093                        myAttrs[ 'nodes' ]              = nodeslist
[507]1094                        myAttrs[ 'domain' ]             = fqdn_parts( socket.getfqdn() )[1]
[354]1095                        myAttrs[ 'poll_interval' ]      = str( BATCH_POLL_INTERVAL )
1096
[348]1097                        if self.jobDataChanged( self.jobs, job_id, myAttrs ) and myAttrs['status'] in [ 'R', 'Q' ]:
[61]1098
[354]1099                                self.jobs[ job_id ]     = myAttrs
[26]1100
[348]1101                for id, attrs in self.jobs.items():
[76]1102
1103                        if id not in jobs_processed:
1104
1105                                # This one isn't there anymore; toedeledoki!
1106                                #
[348]1107                                del self.jobs[ id ]
[76]1108
[363]1109#
1110# Gmetric by Nick Galbreath - nickg(a.t)modp(d.o.t)com
1111# Version 1.0 - 21-April2-2007
1112# http://code.google.com/p/embeddedgmetric/
1113#
1114# Modified by: Ramon Bastiaans
1115# For the Job Monarch Project, see: https://subtrac.sara.nl/oss/jobmonarch/
1116#
1117# added: DEFAULT_TYPE for Gmetric's
1118# added: checkHostProtocol to determine if target is multicast or not
1119# changed: allow default for Gmetric constructor
1120# changed: allow defaults for all send() values except dmax
1121#
1122
[362]1123GMETRIC_DEFAULT_TYPE    = 'string'
1124GMETRIC_DEFAULT_HOST    = '127.0.0.1'
1125GMETRIC_DEFAULT_PORT    = '8649'
[431]1126GMETRIC_DEFAULT_UNITS   = ''
[362]1127
1128class Gmetric:
1129
1130        global GMETRIC_DEFAULT_HOST, GMETRIC_DEFAULT_PORT
1131
1132        slope           = { 'zero' : 0, 'positive' : 1, 'negative' : 2, 'both' : 3, 'unspecified' : 4 }
1133        type            = ( '', 'string', 'uint16', 'int16', 'uint32', 'int32', 'float', 'double', 'timestamp' )
1134        protocol        = ( 'udp', 'multicast' )
1135
1136        def __init__( self, host=GMETRIC_DEFAULT_HOST, port=GMETRIC_DEFAULT_PORT ):
1137               
1138                global GMETRIC_DEFAULT_TYPE
1139
1140                self.prot       = self.checkHostProtocol( host )
1141                self.msg        = xdrlib.Packer()
1142                self.socket     = socket.socket( socket.AF_INET, socket.SOCK_DGRAM )
1143
1144                if self.prot not in self.protocol:
1145
1146                        raise ValueError( "Protocol must be one of: " + str( self.protocol ) )
1147
1148                if self.prot == 'multicast':
1149
[471]1150                        # Set multicast options
1151                        #
[362]1152                        self.socket.setsockopt( socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 20 )
1153
1154                self.hostport   = ( host, int( port ) )
1155                self.slopestr   = 'both'
1156                self.tmax       = 60
1157
1158        def checkHostProtocol( self, ip ):
1159
[471]1160                """Detect if a ip adress is a multicast address"""
1161
[362]1162                MULTICAST_ADDRESS_MIN   = ( "224", "0", "0", "0" )
1163                MULTICAST_ADDRESS_MAX   = ( "239", "255", "255", "255" )
1164
1165                ip_fields               = ip.split( '.' )
1166
1167                if ip_fields >= MULTICAST_ADDRESS_MIN and ip_fields <= MULTICAST_ADDRESS_MAX:
1168
1169                        return 'multicast'
1170                else:
1171                        return 'udp'
1172
[431]1173        def send( self, name, value, dmax, typestr = '', units = '' ):
[362]1174
[409]1175                if len( units ) == 0:
[431]1176                        units           = GMETRIC_DEFAULT_UNITS
[471]1177
[431]1178                if len( typestr ) == 0:
1179                        typestr         = GMETRIC_DEFAULT_TYPE
[362]1180
[424]1181                msg             = self.makexdr( name, value, typestr, units, self.slopestr, self.tmax, dmax )
[409]1182
[362]1183                return self.socket.sendto( msg, self.hostport )
1184
1185        def makexdr( self, name, value, typestr, unitstr, slopestr, tmax, dmax ):
1186
1187                if slopestr not in self.slope:
1188
1189                        raise ValueError( "Slope must be one of: " + str( self.slope.keys() ) )
1190
1191                if typestr not in self.type:
1192
1193                        raise ValueError( "Type must be one of: " + str( self.type ) )
1194
1195                if len( name ) == 0:
1196
1197                        raise ValueError( "Name must be non-empty" )
1198
1199                self.msg.reset()
1200                self.msg.pack_int( 0 )
1201                self.msg.pack_string( typestr )
1202                self.msg.pack_string( name )
1203                self.msg.pack_string( str( value ) )
1204                self.msg.pack_string( unitstr )
1205                self.msg.pack_int( self.slope[ slopestr ] )
1206                self.msg.pack_uint( int( tmax ) )
1207                self.msg.pack_uint( int( dmax ) )
1208
1209                return self.msg.get_buffer()
1210
[26]1211def printTime( ):
[354]1212
[65]1213        """Print current time/date in human readable format for log/debug"""
[26]1214
1215        return time.strftime("%a, %d %b %Y %H:%M:%S")
1216
1217def debug_msg( level, msg ):
[354]1218
[65]1219        """Print msg if at or above current debug level"""
[26]1220
[377]1221        global DAEMONIZE, DEBUG_LEVEL, SYSLOG_LEVEL
1222
[373]1223        if (not DAEMONIZE and DEBUG_LEVEL >= level):
1224                sys.stderr.write( msg + '\n' )
[26]1225
[373]1226        if (DAEMONIZE and USE_SYSLOG and SYSLOG_LEVEL >= level):
1227                syslog.syslog( msg )
1228
[307]1229def write_pidfile():
1230
[471]1231        # Write pidfile if PIDFILE is set
1232        #
[307]1233        if PIDFILE:
1234
[354]1235                pid     = os.getpid()
1236
[471]1237                pidfile = open( PIDFILE, 'w' )
[354]1238
1239                pidfile.write( str( pid ) )
[307]1240                pidfile.close()
1241
[23]1242def main():
[354]1243
[65]1244        """Application start"""
[23]1245
[352]1246        global PBSQuery, PBSError
[373]1247        global SYSLOG_FACILITY, USE_SYSLOG, BATCH_API, DAEMONIZE
[256]1248
[212]1249        if not processArgs( sys.argv[1:] ):
[354]1250
[212]1251                sys.exit( 1 )
1252
[471]1253        # Load appropriate DataGatherer depending on which BATCH_API is set
1254        # and any required modules for the Gatherer
1255        #
[256]1256        if BATCH_API == 'pbs':
1257
1258                try:
[282]1259                        from PBSQuery import PBSQuery, PBSError
[256]1260
1261                except ImportError:
1262
[373]1263                        debug_msg( 0, "FATAL ERROR: BATCH_API set to 'pbs' but python module 'pbs_python' is not installed" )
[256]1264                        sys.exit( 1 )
1265
1266                gather = PbsDataGatherer()
1267
1268        elif BATCH_API == 'sge':
1269
[507]1270                # Tested with SGE 6.0u11.
1271#               debug_msg( 0, "FATAL ERROR: BATCH_API 'sge' implementation is currently broken, check future releases" )
[368]1272
[507]1273#               sys.exit( 1 )
[368]1274
[347]1275                gather = SgeDataGatherer()
[256]1276
1277        else:
[373]1278                debug_msg( 0, "FATAL ERROR: unknown BATCH_API '" + BATCH_API + "' is not supported" )
[354]1279
[256]1280                sys.exit( 1 )
1281
[373]1282        if( DAEMONIZE and USE_SYSLOG ):
1283
1284                syslog.openlog( 'jobmond', syslog.LOG_NOWAIT, SYSLOG_FACILITY )
1285
1286
[26]1287        if DAEMONIZE:
[354]1288
[26]1289                gather.daemon()
1290        else:
1291                gather.run()
[23]1292
[256]1293# wh00t? someone started me! :)
[65]1294#
[23]1295if __name__ == '__main__':
1296        main()
Note: See TracBrowser for help on using the repository browser.