source: trunk/jobmond/jobmond.py @ 661

Last change on this file since 661 was 661, checked in by ramonb, 12 years ago
  • also report Waiting jobs
  • renamed some metrics to make more sense
  • more PBS info
  • Property svn:executable set to *
  • Property svn:keywords set to Id
File size: 49.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
[622]6# Copyright (C) 2007, 2009  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 661 2012-09-04 09:23:16Z ramonb $
[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
[622]28from collections import deque
[660]29from types import *
[318]30
[500]31VERSION='0.3.1'
[307]32
[471]33def usage( ver ):
34
[659]35    print 'jobmond %s' %VERSION
[471]36
[659]37    if ver:
38        return 0
[471]39
[659]40    print
41    print 'Purpose:'
42    print '  The Job Monitoring Daemon (jobmond) reports batch jobs information and statistics'
43    print '  to Ganglia, which can be viewed with Job Monarch web frontend'
44    print
45    print 'Usage:   jobmond [OPTIONS]'
46    print
47    print '  -c, --config=FILE  The configuration file to use (default: /etc/jobmond.conf)'
48    print '  -p, --pidfile=FILE Use pid file to store the process id'
49    print '  -h, --help     Print help and exit'
50    print '  -v, --version          Print version and exit'
51    print
[307]52
[212]53def processArgs( args ):
[26]54
[659]55    SHORT_L     = 'p:hvc:'
56    LONG_L      = [ 'help', 'config=', 'pidfile=', 'version' ]
[165]57
[659]58    global PIDFILE
59    PIDFILE     = None
[61]60
[659]61    config_filename = '/etc/jobmond.conf'
[354]62
[659]63    try:
[68]64
[659]65        opts, args  = getopt.getopt( args, SHORT_L, LONG_L )
[185]66
[659]67    except getopt.GetoptError, detail:
[212]68
[659]69        print detail
70        usage()
71        sys.exit( 1 )
[212]72
[659]73    for opt, value in opts:
[212]74
[659]75        if opt in [ '--config', '-c' ]:
76       
77            config_filename = value
[212]78
[659]79        if opt in [ '--pidfile', '-p' ]:
[212]80
[659]81            PIDFILE     = value
82       
83        if opt in [ '--help', '-h' ]:
[307]84 
[659]85            usage( False )
86            sys.exit( 0 )
[212]87
[659]88        if opt in [ '--version', '-v' ]:
[471]89
[659]90            usage( True )
91            sys.exit( 0 )
[471]92
[659]93    return loadConfig( config_filename )
[212]94
[622]95# Fixme:  This doesn't DTRT with commented-out bits of the file.  E.g.
96# it picked up a commented-out `mcast_join' and tried to use a
97# multicast channel when it shouldn't have done.
[520]98class GangliaConfigParser:
99
[659]100    def __init__( self, config_file ):
[520]101
[659]102        self.config_file    = config_file
[520]103
[659]104        if not os.path.exists( self.config_file ):
[520]105
[659]106            debug_msg( 0, "FATAL ERROR: gmond config '" + self.config_file + "' not found!" )
107            sys.exit( 1 )
[520]108
[659]109    def removeQuotes( self, value ):
[520]110
[659]111        clean_value = value
112        clean_value = clean_value.replace( "'", "" )
113        clean_value = clean_value.replace( '"', '' )
114        clean_value = clean_value.strip()
[520]115
[659]116        return clean_value
[520]117
[659]118    def getVal( self, section, valname ):
[520]119
[659]120        cfg_fp      = open( self.config_file )
121        section_start   = False
122        section_found   = False
123        value       = None
[520]124
[659]125        for line in cfg_fp.readlines():
[520]126
[659]127            if line.find( section ) != -1:
[520]128
[659]129                section_found   = True
[520]130
[659]131            if line.find( '{' ) != -1 and section_found:
[520]132
[659]133                section_start   = True
[520]134
[659]135            if line.find( '}' ) != -1 and section_found:
[520]136
[659]137                section_start   = False
138                section_found   = False
[520]139
[659]140            if line.find( valname ) != -1 and section_start:
[520]141
[659]142                value       = string.join( line.split( '=' )[1:], '' ).strip()
[520]143
[659]144        cfg_fp.close()
[520]145
[659]146        return value
[520]147
[659]148    def getInt( self, section, valname ):
[520]149
[659]150        value   = self.getVal( section, valname )
[520]151
[659]152        if not value:
153            return False
[520]154
[659]155        value   = self.removeQuotes( value )
[520]156
[659]157        return int( value )
[520]158
[659]159    def getStr( self, section, valname ):
[520]160
[659]161        value   = self.getVal( section, valname )
[520]162
[659]163        if not value:
164            return False
[520]165
[659]166        value   = self.removeQuotes( value )
[520]167
[659]168        return str( value )
[520]169
170def findGmetric():
171
[659]172    for dir in os.path.expandvars( '$PATH' ).split( ':' ):
[520]173
[659]174        guess   = '%s/%s' %( dir, 'gmetric' )
[520]175
[659]176        if os.path.exists( guess ):
[520]177
[659]178            return guess
[520]179
[659]180    return False
[520]181
[212]182def loadConfig( filename ):
183
[660]184    def getlist( cfg_string ):
[215]185
[660]186        my_list = [ ]
[215]187
[660]188        for item_txt in cfg_string.split( ',' ):
[215]189
[660]190                sep_char = None
[215]191
[660]192                item_txt = item_txt.strip()
[215]193
[660]194                for s_char in [ "'", '"' ]:
[215]195
[660]196                        if item_txt.find( s_char ) != -1:
[215]197
[660]198                                if item_txt.count( s_char ) != 2:
[215]199
[660]200                                        print 'Missing quote: %s' %item_txt
201                                        sys.exit( 1 )
[215]202
[660]203                                else:
[215]204
[660]205                                        sep_char = s_char
206                                        break
[215]207
[660]208                if sep_char:
[215]209
[660]210                        item_txt = item_txt.split( sep_char )[1]
[215]211
[660]212                my_list.append( item_txt )
[215]213
[660]214        return my_list
[215]215
[659]216    cfg     = ConfigParser.ConfigParser()
[212]217
[659]218    cfg.read( filename )
[212]219
[659]220    global DEBUG_LEVEL, DAEMONIZE, BATCH_SERVER, BATCH_POLL_INTERVAL
221    global GMOND_CONF, DETECT_TIME_DIFFS, BATCH_HOST_TRANSLATE
222    global BATCH_API, QUEUE, GMETRIC_TARGET, USE_SYSLOG
223    global SYSLOG_LEVEL, SYSLOG_FACILITY, GMETRIC_BINARY
[212]224
[659]225    DEBUG_LEVEL = cfg.getint( 'DEFAULT', 'DEBUG_LEVEL' )
[212]226
[659]227    DAEMONIZE   = cfg.getboolean( 'DEFAULT', 'DAEMONIZE' )
[212]228
[659]229    SYSLOG_LEVEL    = -1
230    SYSLOG_FACILITY = None
[377]231
[659]232    try:
233        USE_SYSLOG  = cfg.getboolean( 'DEFAULT', 'USE_SYSLOG' )
[212]234
[659]235    except ConfigParser.NoOptionError:
[373]236
[659]237        USE_SYSLOG  = True
[373]238
[659]239        debug_msg( 0, 'ERROR: no option USE_SYSLOG found: assuming yes' )
[373]240
[659]241    if USE_SYSLOG:
[373]242
[659]243        try:
244            SYSLOG_LEVEL    = cfg.getint( 'DEFAULT', 'SYSLOG_LEVEL' )
[373]245
[659]246        except ConfigParser.NoOptionError:
[373]247
[659]248            debug_msg( 0, 'ERROR: no option SYSLOG_LEVEL found: assuming level 0' )
249            SYSLOG_LEVEL    = 0
[373]250
[659]251        try:
[373]252
[659]253            SYSLOG_FACILITY = eval( 'syslog.LOG_' + cfg.get( 'DEFAULT', 'SYSLOG_FACILITY' ) )
[373]254
[659]255        except ConfigParser.NoOptionError:
[373]256
[659]257            SYSLOG_FACILITY = syslog.LOG_DAEMON
[373]258
[659]259            debug_msg( 0, 'ERROR: no option SYSLOG_FACILITY found: assuming facility DAEMON' )
[373]260
[659]261    try:
[373]262
[659]263        BATCH_SERVER        = cfg.get( 'DEFAULT', 'BATCH_SERVER' )
[212]264
[659]265    except ConfigParser.NoOptionError:
[265]266
[659]267        # Backwards compatibility for old configs
268        #
[265]269
[659]270        BATCH_SERVER        = cfg.get( 'DEFAULT', 'TORQUE_SERVER' )
271        api_guess       = 'pbs'
272   
273    try:
274   
275        BATCH_POLL_INTERVAL = cfg.getint( 'DEFAULT', 'BATCH_POLL_INTERVAL' )
[265]276
[659]277    except ConfigParser.NoOptionError:
[265]278
[659]279        # Backwards compatibility for old configs
280        #
[265]281
[659]282        BATCH_POLL_INTERVAL = cfg.getint( 'DEFAULT', 'TORQUE_POLL_INTERVAL' )
283        api_guess       = 'pbs'
284   
285    try:
[212]286
[659]287        GMOND_CONF      = cfg.get( 'DEFAULT', 'GMOND_CONF' )
[353]288
[659]289    except ConfigParser.NoOptionError:
[353]290
[661]291        # Not specified: assume /etc/ganglia/gmond.conf
[659]292        #
[661]293        GMOND_CONF      = '/etc/ganglia/gmond.conf'
[353]294
[659]295    ganglia_cfg     = GangliaConfigParser( GMOND_CONF )
[449]296
[659]297    # Let's try to find the GMETRIC_TARGET ourselves first from GMOND_CONF
298    #
299    gmetric_dest_ip     = ganglia_cfg.getStr( 'udp_send_channel', 'mcast_join' )
[449]300
[659]301    if not gmetric_dest_ip:
[449]302
[659]303        # Maybe unicast target then
304        #
305        gmetric_dest_ip     = ganglia_cfg.getStr( 'udp_send_channel', 'host' )
[449]306
[660]307        gmetric_dest_port   = ganglia_cfg.getStr( 'udp_send_channel', 'port' )
[520]308
[659]309    if gmetric_dest_ip and gmetric_dest_port:
[520]310
[659]311        GMETRIC_TARGET  = '%s:%s' %( gmetric_dest_ip, gmetric_dest_port )
312    else:
[520]313
[659]314        debug_msg( 0, "WARNING: Can't parse udp_send_channel from: '%s'" %GMOND_CONF )
[520]315
[659]316        # Couldn't figure it out: let's see if it's in our jobmond.conf
317        #
318        try:
[520]319
[659]320            GMETRIC_TARGET  = cfg.get( 'DEFAULT', 'GMETRIC_TARGET' )
[520]321
[659]322        # Guess not: now just give up
323        #
324        except ConfigParser.NoOptionError:
[520]325
[659]326            GMETRIC_TARGET  = None
[520]327
[659]328            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!" )
[520]329
[659]330    gmetric_bin = findGmetric()
[520]331
[659]332    if gmetric_bin:
[520]333
[659]334        GMETRIC_BINARY      = gmetric_bin
335    else:
336        debug_msg( 0, "WARNING: Can't find gmetric binary anywhere in $PATH" )
[520]337
[659]338        try:
[520]339
[659]340            GMETRIC_BINARY      = cfg.get( 'DEFAULT', 'GMETRIC_BINARY' )
[520]341
[659]342        except ConfigParser.NoOptionError:
[520]343
[659]344            debug_msg( 0, "FATAL ERROR: GMETRIC_BINARY not set and not in $PATH" )
345            sys.exit( 1 )
[520]346
[659]347    DETECT_TIME_DIFFS   = cfg.getboolean( 'DEFAULT', 'DETECT_TIME_DIFFS' )
[212]348
[659]349    BATCH_HOST_TRANSLATE    = getlist( cfg.get( 'DEFAULT', 'BATCH_HOST_TRANSLATE' ) )
[215]350
[659]351    try:
[256]352
[659]353        BATCH_API   = cfg.get( 'DEFAULT', 'BATCH_API' )
[266]354
[659]355    except ConfigParser.NoOptionError, detail:
[266]356
[659]357        if BATCH_SERVER and api_guess:
[354]358
[659]359            BATCH_API   = api_guess
360        else:
361            debug_msg( 0, "FATAL ERROR: BATCH_API not set and can't make guess" )
362            sys.exit( 1 )
[317]363
[659]364    try:
[317]365
[659]366        QUEUE       = getlist( cfg.get( 'DEFAULT', 'QUEUE' ) )
[317]367
[659]368    except ConfigParser.NoOptionError, detail:
[317]369
[659]370        QUEUE       = None
[353]371
[659]372    return True
[212]373
[507]374def fqdn_parts (fqdn):
[520]375
[659]376    """Return pair of host and domain for fully-qualified domain name arg."""
[520]377
[659]378    parts = fqdn.split (".")
[520]379
[659]380    return (parts[0], string.join(parts[1:], "."))
[507]381
[253]382METRIC_MAX_VAL_LEN = 900
383
[61]384class DataProcessor:
[355]385
[659]386    """Class for processing of data"""
[61]387
[659]388    binary = None
[61]389
[659]390    def __init__( self, binary=None ):
[355]391
[659]392        """Remember alternate binary location if supplied"""
[61]393
[659]394        global GMETRIC_BINARY
[449]395
[659]396        if binary:
397            self.binary = binary
[61]398
[659]399        if not self.binary:
400            self.binary = GMETRIC_BINARY
[449]401
[659]402        # Timeout for XML
403        #
404        # From ganglia's documentation:
405        #
406        # 'A metric will be deleted DMAX seconds after it is received, and
407            # DMAX=0 means eternal life.'
[61]408
[659]409        self.dmax = str( int( int( BATCH_POLL_INTERVAL ) * 2 ) )
[80]410
[659]411        if GMOND_CONF:
[354]412
[659]413            incompatible = self.checkGmetricVersion()
[61]414
[659]415            if incompatible:
[355]416
[659]417                debug_msg( 0, 'Gmetric version not compatible, please upgrade to at least 3.0.1' )
418                sys.exit( 1 )
[65]419
[659]420    def checkGmetricVersion( self ):
[355]421
[659]422        """
423        Check version of gmetric is at least 3.0.1
424        for the syntax we use
425        """
[65]426
[659]427        global METRIC_MAX_VAL_LEN
[255]428
[659]429        incompatible    = 0
[341]430
[659]431        gfp     = os.popen( self.binary + ' --version' )
432        lines       = gfp.readlines()
[65]433
[659]434        gfp.close()
[355]435
[659]436        for line in lines:
[355]437
[659]438            line = line.split( ' ' )
[65]439
[659]440            if len( line ) == 2 and str( line ).find( 'gmetric' ) != -1:
441           
442                gmetric_version = line[1].split( '\n' )[0]
[65]443
[659]444                version_major   = int( gmetric_version.split( '.' )[0] )
445                version_minor   = int( gmetric_version.split( '.' )[1] )
446                version_patch   = int( gmetric_version.split( '.' )[2] )
[65]447
[659]448                incompatible    = 0
[65]449
[659]450                if version_major < 3:
[65]451
[659]452                    incompatible = 1
453               
454                elif version_major == 3:
[65]455
[659]456                    if version_minor == 0:
[65]457
[659]458                        if version_patch < 1:
459                       
460                            incompatible = 1
[65]461
[659]462                        # Gmetric 3.0.1 >< 3.0.3 had a bug in the max metric length
463                        #
464                        if version_patch < 3:
[255]465
[659]466                            METRIC_MAX_VAL_LEN = 900
[255]467
[659]468                        elif version_patch >= 3:
[255]469
[659]470                            METRIC_MAX_VAL_LEN = 1400
[255]471
[659]472        return incompatible
[65]473
[659]474    def multicastGmetric( self, metricname, metricval, valtype='string', units='' ):
[355]475
[659]476        """Call gmetric binary and multicast"""
[65]477
[659]478        cmd = self.binary
[65]479
[659]480        if GMETRIC_TARGET:
[61]481
[659]482            GMETRIC_TARGET_HOST = GMETRIC_TARGET.split( ':' )[0]
483            GMETRIC_TARGET_PORT = GMETRIC_TARGET.split( ':' )[1]
[353]484
[659]485            metric_debug        = "[gmetric] name: %s - val: %s - dmax: %s" %( str( metricname ), str( metricval ), str( self.dmax ) )
[353]486
[659]487            debug_msg( 10, printTime() + ' ' + metric_debug)
[353]488
[659]489            gm = Gmetric( GMETRIC_TARGET_HOST, GMETRIC_TARGET_PORT )
[353]490
[659]491            gm.send( str( metricname ), str( metricval ), str( self.dmax ), valtype, units )
[353]492
[659]493        else:
494            try:
495                cmd = cmd + ' -c' + GMOND_CONF
[353]496
[659]497            except NameError:
[353]498
[659]499                debug_msg( 10, 'Assuming /etc/gmond.conf for gmetric cmd (omitting)' )
[353]500
[659]501            cmd = cmd + ' -n' + str( metricname )+ ' -v"' + str( metricval )+ '" -t' + str( valtype ) + ' -d' + str( self.dmax )
[353]502
[659]503            if len( units ) > 0:
[409]504
[659]505                cmd = cmd + ' -u"' + units + '"'
[409]506
[659]507            debug_msg( 10, printTime() + ' ' + cmd )
[353]508
[659]509            os.system( cmd )
[353]510
[318]511class DataGatherer:
[23]512
[659]513    """Skeleton class for batch system DataGatherer"""
[256]514
[659]515    def printJobs( self, jobs ):
[355]516
[659]517        """Print a jobinfo overview"""
[318]518
[659]519        for name, attrs in self.jobs.items():
[318]520
[659]521            print 'job %s' %(name)
[318]522
[659]523            for name, val in attrs.items():
[318]524
[659]525                print '\t%s = %s' %( name, val )
[318]526
[659]527    def printJob( self, jobs, job_id ):
[355]528
[659]529        """Print job with job_id from jobs"""
[318]530
[659]531        print 'job %s' %(job_id)
[318]532
[659]533        for name, val in jobs[ job_id ].items():
[318]534
[659]535            print '\t%s = %s' %( name, val )
[318]536
[660]537    def getAttr( self, d, name ):
[507]538
[659]539        """Return certain attribute from dictionary, if exists"""
[507]540
[660]541        if d.has_key( name ):
[507]542
[660]543            if type( d[ name ] ) == ListType:
[507]544
[660]545                return string.join( d[ name ], ' ' )
546
547            return d[ name ]
548       
549        return ''
550
[659]551    def jobDataChanged( self, jobs, job_id, attrs ):
[507]552
[659]553        """Check if job with attrs and job_id in jobs has changed"""
[507]554
[659]555        if jobs.has_key( job_id ):
[507]556
[659]557            oldData = jobs[ job_id ]   
558        else:
559            return 1
[507]560
[659]561        for name, val in attrs.items():
[507]562
[659]563            if oldData.has_key( name ):
[507]564
[659]565                if oldData[ name ] != attrs[ name ]:
[507]566
[659]567                    return 1
[507]568
[659]569            else:
570                return 1
[507]571
[659]572        return 0
[507]573
[659]574    def submitJobData( self ):
[507]575
[659]576        """Submit job info list"""
[507]577
[659]578        global BATCH_API
[512]579
[659]580        self.dp.multicastGmetric( 'MONARCH-HEARTBEAT', str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
[507]581
[659]582        running_jobs    = 0
583        queued_jobs = 0
[507]584
[659]585        # Count how many running/queued jobs we found
[507]586                #
[659]587        for jobid, jobattrs in self.jobs.items():
[507]588
[659]589            if jobattrs[ 'status' ] == 'Q':
[507]590
[659]591                queued_jobs += 1
[507]592
[659]593            elif jobattrs[ 'status' ] == 'R':
[507]594
[659]595                running_jobs += 1
[507]596
[659]597        # Report running/queued jobs as seperate metric for a nice RRD graph
[507]598                #
[659]599        self.dp.multicastGmetric( 'MONARCH-RJ', str( running_jobs ), 'uint32', 'jobs' )
600        self.dp.multicastGmetric( 'MONARCH-QJ', str( queued_jobs ), 'uint32', 'jobs' )
[507]601
[659]602        # Report down/offline nodes in batch (PBS only ATM)
603        #
604        if BATCH_API == 'pbs':
[512]605
[659]606            domain      = fqdn_parts( socket.getfqdn() )[1]
[514]607
[659]608            downed_nodes    = list()
609            offline_nodes   = list()
610       
611            l       = ['state']
612       
613            for name, node in self.pq.getnodes().items():
[512]614
[660]615                if 'down' in node[ 'state' ]:
[512]616
[659]617                    downed_nodes.append( name )
[512]618
[660]619                if 'offline' in node[ 'state' ]:
[512]620
[659]621                    offline_nodes.append( name )
[512]622
[659]623            downnodeslist       = do_nodelist( downed_nodes )
624            offlinenodeslist    = do_nodelist( offline_nodes )
[512]625
[659]626            down_str    = 'nodes=%s domain=%s reported=%s' %( string.join( downnodeslist, ';' ), domain, str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
627            offl_str    = 'nodes=%s domain=%s reported=%s' %( string.join( offlinenodeslist, ';' ), domain, str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
628            self.dp.multicastGmetric( 'MONARCH-DOWN'   , down_str )
629            self.dp.multicastGmetric( 'MONARCH-OFFLINE', offl_str )
[514]630
[659]631        # Now let's spread the knowledge
632        #
633        for jobid, jobattrs in self.jobs.items():
[507]634
[659]635            # Make gmetric values for each job: respect max gmetric value length
636            #
[660]637            gmetrics     = self.compileGmetricVal( jobid, jobattrs )
[507]638
[660]639            for g_name, g_val in gmetrics.items():
[507]640
[660]641                self.dp.multicastGmetric( g_name, g_val )
[507]642
[659]643    def compileGmetricVal( self, jobid, jobattrs ):
[507]644
[660]645        """Create gmetric name/value pairs of jobinfo"""
[507]646
[660]647        gmetrics = { }
[507]648
[659]649        for val_name, val_value in jobattrs.items():
[507]650
[660]651            gmetric_sequence = 0
[507]652
[660]653            if len( val_value ) > METRIC_MAX_VAL_LEN:
[507]654
[660]655                while len( val_value ) > METRIC_MAX_VAL_LEN:
[507]656
[660]657                    gmetric_value   = val_value[:METRIC_MAX_VAL_LEN]
658                    val_value       = val_value[METRIC_MAX_VAL_LEN:]
[507]659
[660]660                    gmetric_name    = 'MONARCHJOB$%s$%s$%s' %( jobid, string.upper(val_name), gmetric_sequence )
[507]661
[660]662                    gmetrics[ gmetric_name ] = gmetric_value
[507]663
[660]664                    gmetric_sequence = gmetric_sequence + 1
665            else:
666                gmetric_value   = val_value
[507]667
[660]668                gmetric_name    = 'MONARCH$%s$%s$%s' %( jobid, string.upper(val_name), gmetric_sequence )
[507]669
[660]670                gmetrics[ gmetric_name ] = gmetric_value
[507]671
[660]672        return gmetrics
[507]673
[660]674    def daemon( self ):
[507]675
[660]676        """Run as daemon forever"""
[507]677
[660]678        # Fork the first child
679        #
680        pid = os.fork()
681        if pid > 0:
682                sys.exit(0)  # end parent
[507]683
[660]684        # creates a session and sets the process group ID
685        #
686        os.setsid()
[507]687
[660]688        # Fork the second child
689        #
690        pid = os.fork()
691        if pid > 0:
692                sys.exit(0)  # end parent
[507]693
[659]694        write_pidfile()
[318]695
[660]696        # Go to the root directory and set the umask
697        #
698        os.chdir('/')
699        os.umask(0)
[318]700
[660]701        sys.stdin.close()
702        sys.stdout.close()
703        sys.stderr.close()
[318]704
[660]705        os.open('/dev/null', os.O_RDWR)
706        os.dup2(0, 1)
707        os.dup2(0, 2)
[318]708
[660]709        self.run()
[318]710
[660]711    def run( self ):
[355]712
[660]713        """Main thread"""
[256]714
[660]715        while ( 1 ):
[659]716       
717            self.getJobData()
718            self.submitJobData()
719            time.sleep( BATCH_POLL_INTERVAL )   
[256]720
[622]721# SGE code by Dave Love <fx@gnu.org>.  Tested with SGE 6.0u8 and 6.0u11.  May
722# work with SGE 6.1 (else should be easily fixable), but definitely doesn't
723# with 6.2.  See also the fixmes.
[256]724
[507]725class NoJobs (Exception):
[659]726    """Exception raised by empty job list in qstat output."""
727    pass
[256]728
[507]729class SgeQstatXMLParser(xml.sax.handler.ContentHandler):
[659]730    """SAX handler for XML output from Sun Grid Engine's `qstat'."""
[318]731
[659]732    def __init__(self):
733        self.value = ""
734        self.joblist = []
735        self.job = {}
736        self.queue = ""
737        self.in_joblist = False
738        self.lrequest = False
739        self.eltq = deque()
740        xml.sax.handler.ContentHandler.__init__(self)
[318]741
[659]742    # The structure of the output is as follows (for SGE 6.0).  It's
743    # similar for 6.1, but radically different for 6.2, and is
744    # undocumented generally.  Unfortunately it's voluminous, and probably
745    # doesn't scale to large clusters/queues.
[318]746
[659]747    # <detailed_job_info  xmlns:xsd="http://www.w3.org/2001/XMLSchema">
748    #   <djob_info>
749    #     <qmaster_response>  <!-- job -->
750    #       ...
751    #       <JB_ja_template> 
752    #         <ulong_sublist>
753    #         ...             <!-- start_time, state ... -->
754    #         </ulong_sublist>
755    #       </JB_ja_template> 
756    #       <JB_ja_tasks>
757    #         <ulong_sublist>
758    #           ...           <!-- task info
759    #         </ulong_sublist>
760    #         ...
761    #       </JB_ja_tasks>
762    #       ...
763    #     </qmaster_response>
764    #   </djob_info>
765    #   <messages>
766    #   ...
[318]767
[659]768    # NB.  We might treat each task as a separate job, like
769    # straight qstat output, but the web interface expects jobs to
770    # be identified by integers, not, say, <job number>.<task>.
[318]771
[659]772    # So, I lied.  If the job list is empty, we get invalid XML
773    # like this, which we need to defend against:
[318]774
[659]775    # <unknown_jobs  xmlns:xsd="http://www.w3.org/2001/XMLSchema">
776    #   <>
777    #     <ST_name>*</ST_name>
778    #   </>
779    # </unknown_jobs>
[318]780
[659]781    def startElement(self, name, attrs):
782        self.value = ""
783        if name == "djob_info": # job list
784            self.in_joblist = True
785        # The job container is "qmaster_response" in SGE 6.0
786        # and 6.1, but "element" in 6.2.  This is only the very
787        # start of what's necessary for 6.2, though (sigh).
788        elif (name == "qmaster_response" or name == "element") \
789                and self.eltq[-1] == "djob_info": # job
790            self.job = {"job_state": "U", "slots": 0,
791                    "nodes": [], "queued_timestamp": "",
792                    "queued_timestamp": "", "queue": "",
793                    "ppn": "0", "RN_max": 0,
794                    # fixme in endElement
795                    "requested_memory": 0, "requested_time": 0
796                    }
797            self.joblist.append(self.job)
798        elif name == "qstat_l_requests": # resource request
799            self.lrequest = True
800        elif name == "unknown_jobs":
801            raise NoJobs
802        self.eltq.append (name)
[318]803
[659]804    def characters(self, ch):
805        self.value += ch
[318]806
[659]807    def endElement(self, name): 
808        """Snarf job elements contents into job dictionary.
809           Translate keys if appropriate."""
[318]810
[659]811        name_trans = {
812          "JB_job_number": "number",
813          "JB_job_name": "name", "JB_owner": "owner",
814          "queue_name": "queue", "JAT_start_time": "start_timestamp",
815          "JB_submission_time": "queued_timestamp"
816          }
817        value = self.value
818        self.eltq.pop ()
[318]819
[659]820        if name == "djob_info":
821            self.in_joblist = False
822            self.job = {}
823        elif name == "JAT_master_queue":
824            self.job["queue"] = value.split("@")[0]
825        elif name == "JG_qhostname":
826            if not (value in self.job["nodes"]):
827                self.job["nodes"].append(value)
828        elif name == "JG_slots": # slots in use
829            self.job["slots"] += int(value)
830        elif name == "RN_max": # requested slots (tasks or parallel)
831            self.job["RN_max"] = max (self.job["RN_max"],
832                          int(value))
833        elif name == "JAT_state": # job state (bitwise or)
834            value = int (value)
835            # Status values from sge_jobL.h
836            #define JIDLE                   0x00000000
837            #define JHELD                   0x00000010
838            #define JMIGRATING              0x00000020
839            #define JQUEUED                 0x00000040
840            #define JRUNNING                0x00000080
841            #define JSUSPENDED              0x00000100
842            #define JTRANSFERING            0x00000200
843            #define JDELETED                0x00000400
844            #define JWAITING                0x00000800
845            #define JEXITING                0x00001000
846            #define JWRITTEN                0x00002000
847            #define JSUSPENDED_ON_THRESHOLD 0x00010000
848            #define JFINISHED               0x00010000
849            if value & 0x80:
850                self.job["status"] = "R"
851            elif value & 0x40:
852                self.job["status"] = "Q"
853            else:
854                self.job["status"] = "O" # `other'
855        elif name == "CE_name" and self.lrequest and self.value in \
856                ("h_cpu", "s_cpu", "cpu", "h_core", "s_core"):
857            # We're in a container for an interesting resource
858            # request; record which type.
859            self.lrequest = self.value
860        elif name == "CE_doubleval" and self.lrequest:
861            # if we're in a container for an interesting
862            # resource request, use the maxmimum of the hard
863            # and soft requests to record the requested CPU
864            # or core.  Fixme:  I'm not sure if this logic is
865            # right.
866            if self.lrequest in ("h_core", "s_core"):
867                self.job["requested_memory"] = \
868                    max (float (value),
869                     self.job["requested_memory"])
870            # Fixme:  Check what cpu means, c.f [hs]_cpu.
871            elif self.lrequest in ("h_cpu", "s_cpu", "cpu"):
872                self.job["requested_time"] = \
873                    max (float (value),
874                     self.job["requested_time"])
875        elif name == "qstat_l_requests":
876            self.lrequest = False
877        elif self.job and self.in_joblist:
878            if name in name_trans:
879                name = name_trans[name]
880                self.job[name] = value
[318]881
[507]882# Abstracted from PBS original.
[520]883#
884def do_nodelist( nodes ):
885
[659]886    """Translate node list as appropriate."""
[520]887
[659]888    nodeslist       = [ ]
889    my_domain       = fqdn_parts( socket.getfqdn() )[1]
[520]890
[659]891    for node in nodes:
[520]892
[659]893        host        = node.split( '/' )[0] # not relevant for SGE
894        h, host_domain  = fqdn_parts(host)
[520]895
[659]896        if host_domain == my_domain:
[520]897
[659]898            host    = h
[520]899
[659]900        if nodeslist.count( host ) == 0:
[520]901
[659]902            for translate_pattern in BATCH_HOST_TRANSLATE:
[520]903
[659]904                if translate_pattern.find( '/' ) != -1:
[520]905
[659]906                    translate_orig  = \
907                        translate_pattern.split( '/' )[1]
908                    translate_new   = \
909                        translate_pattern.split( '/' )[2]
910                    host = re.sub( translate_orig,
911                               translate_new, host )
912            if not host in nodeslist:
913                nodeslist.append( host )
914    return nodeslist
[318]915
916class SgeDataGatherer(DataGatherer):
917
[659]918    jobs = {}
[61]919
[659]920    def __init__( self ):
921        self.jobs = {}
922        self.timeoffset = 0
923        self.dp = DataProcessor()
[318]924
[659]925    def getJobData( self ):
926        """Gather all data on current jobs in SGE"""
[318]927
[659]928        import popen2
[318]929
[659]930        self.cur_time = 0
931        queues = ""
932        if QUEUE:   # only for specific queues
933            # Fixme:  assumes queue names don't contain single
934            # quote or comma.  Don't know what the SGE rules are.
935            queues = " -q '" + string.join (QUEUE, ",") + "'"
936        # Note the comment in SgeQstatXMLParser about scaling with
937        # this method of getting data.  I haven't found better one.
938        # Output with args `-xml -ext -f -r' is easier to parse
939        # in some ways, harder in others, but it doesn't provide
940        # the submission time (at least SGE 6.0).  The pipeline
941        # into sed corrects bogus XML observed with a configuration
942        # of SGE 6.0u8, which otherwise causes the parsing to hang.
943        piping = popen2.Popen3("qstat -u '*' -j '*' -xml | \
[622]944sed -e 's/reported usage>/reported_usage>/g' -e 's;<\/*JATASK:.*>;;'" \
[659]945                           + queues, True)
946        qstatparser = SgeQstatXMLParser()
947        parse_err = 0
948        try:
949            xml.sax.parse(piping.fromchild, qstatparser)
950        except NoJobs:
951            pass
952        except:
953            parse_err = 1
954            if piping.wait():
[660]955                debug_msg(10,
[659]956                  "qstat error, skipping until next polling interval: "
957                  + piping.childerr.readline())
[660]958                return None
959            elif parse_err:
960                debug_msg(10, "Bad XML output from qstat"())
961                exit (1)
[659]962        for f in piping.fromchild, piping.tochild, piping.childerr:
963            f.close()
964        self.cur_time = time.time()
965        jobs_processed = []
966        for job in qstatparser.joblist:
967            job_id = job["number"]
968            if job["status"] in [ 'Q', 'R' ]:
969                jobs_processed.append(job_id)
970            if job["status"] == "R":
971                job["nodes"] = do_nodelist (job["nodes"])
972                # Fixme: why is job["nodes"] sometimes null?
973                try:
974                    # Fixme: Is this sensible?  The
975                    # PBS-type PPN isn't something you use
976                    # with SGE.
977                    job["ppn"] = float(job["slots"]) / \
978                        len(job["nodes"])
979                except:
980                    job["ppn"] = 0
981                if DETECT_TIME_DIFFS:
982                    # If a job start is later than our
983                    # current date, that must mean
984                    # the SGE server's time is later
985                    # than our local time.
986                    start_timestamp = \
987                        int (job["start_timestamp"])
988                    if start_timestamp > \
989                            int(self.cur_time) + \
990                            int(self.timeoffset):
[318]991
[659]992                        self.timeoffset = \
993                            start_timestamp - \
994                            int(self.cur_time)
995            else:
996                # fixme: Note sure what this should be:
997                job["ppn"] = job["RN_max"]
998                job["nodes"] = "1"
[318]999
[659]1000            myAttrs = {}
1001            for attr in ["name", "queue", "owner",
1002                     "requested_time", "status",
1003                     "requested_memory", "ppn",
1004                     "start_timestamp", "queued_timestamp"]:
1005                myAttrs[attr] = str(job[attr])
1006            myAttrs["nodes"] = job["nodes"]
1007            myAttrs["reported"] = str(int(self.cur_time) + \
1008                          int(self.timeoffset))
1009            myAttrs["domain"] = fqdn_parts(socket.getfqdn())[1]
1010            myAttrs["poll_interval"] = str(BATCH_POLL_INTERVAL)
[318]1011
[659]1012            if self.jobDataChanged(self.jobs, job_id, myAttrs) \
1013                    and myAttrs["status"] in ["R", "Q"]:
1014                self.jobs[job_id] = myAttrs
1015        for id, attrs in self.jobs.items():
1016            if id not in jobs_processed:
1017                del self.jobs[id]
[318]1018
[524]1019# LSF code by Mahmoud Hanafi <hanafim@users.sourceforge.nt>
1020# Requres LSFObject http://sourceforge.net/projects/lsfobject
1021#
1022class LsfDataGatherer(DataGatherer):
[525]1023
[524]1024        """This is the DataGatherer for LSf"""
1025
1026        global lsfObject
1027
1028        def __init__( self ):
[525]1029
[524]1030                self.jobs = { }
1031                self.timeoffset = 0
1032                self.dp = DataProcessor()
1033                self.initLsfQuery()
1034
[525]1035        def _countDuplicatesInList( self, dupedList ):
1036
[660]1037            countDupes  = { }
[525]1038
[660]1039            for item in dupedList:
[525]1040
[660]1041                if not countDupes.has_key( item ):
[525]1042
[660]1043                    countDupes[ item ]  = 1
1044                else:
1045                    countDupes[ item ]  = countDupes[ item ] + 1
[525]1046
[660]1047            dupeCountList   = [ ]
[525]1048
[660]1049            for item, count in countDupes.items():
[525]1050
[660]1051                dupeCountList.append( ( item, count ) )
[525]1052
[660]1053            return dupeCountList
[524]1054#
1055#lst = ['I1','I2','I1','I3','I4','I4','I7','I7','I7','I7','I7']
1056#print _countDuplicatesInList(lst)
1057#[('I1', 2), ('I3', 1), ('I2', 1), ('I4', 2), ('I7', 5)]
1058########################
1059
1060        def initLsfQuery( self ):
1061                self.pq = None
1062                self.pq = lsfObject.jobInfoEntObject()
1063
1064        def getJobData( self, known_jobs="" ):
1065                """Gather all data on current jobs in LSF"""
1066                if len( known_jobs ) > 0:
1067                        jobs = known_jobs
1068                else:
1069                        jobs = { }
1070                joblist = {}
1071                joblist = self.pq.getJobInfo()
1072                nodelist = ''
1073
1074                self.cur_time = time.time()
1075
1076                jobs_processed = [ ]
1077
1078                for name, attrs in joblist.items():
1079                        job_id = str(name)
1080                        jobs_processed.append( job_id )
1081                        name = self.getAttr( attrs, 'jobName' )
1082                        queue = self.getAttr( self.getAttr( attrs, 'submit') , 'queue' )
1083                        owner = self.getAttr( attrs, 'user' )
1084
1085### THIS IS THE rLimit List index values
1086#define LSF_RLIMIT_CPU      0            /* cpu time in milliseconds */
1087#define LSF_RLIMIT_FSIZE    1            /* maximum file size */
1088#define LSF_RLIMIT_DATA     2            /* data size */
1089#define LSF_RLIMIT_STACK    3            /* stack size */
1090#define LSF_RLIMIT_CORE     4            /* core file size */
1091#define LSF_RLIMIT_RSS      5            /* resident set size */
1092#define LSF_RLIMIT_NOFILE   6            /* open files */
1093#define LSF_RLIMIT_OPEN_MAX 7            /* (from HP-UX) */
1094#define LSF_RLIMIT_VMEM     8            /* maximum swap mem */
1095#define LSF_RLIMIT_SWAP     8
1096#define LSF_RLIMIT_RUN      9            /* max wall-clock time limit */
1097#define LSF_RLIMIT_PROCESS  10           /* process number limit */
1098#define LSF_RLIMIT_THREAD   11           /* thread number limit (introduced in LSF6.0) */
1099#define LSF_RLIM_NLIMITS    12           /* number of resource limits */
1100
1101                        requested_time = self.getAttr( self.getAttr( attrs, 'submit') , 'rLimits' )[9]
1102                        if requested_time == -1: 
1103                                requested_time = ""
1104                        requested_memory = self.getAttr( self.getAttr( attrs, 'submit') , 'rLimits' )[8]
1105                        if requested_memory == -1: 
1106                                requested_memory = ""
1107# This tries to get proc per node. We don't support this right now
1108                        ppn = 0 #self.getAttr( self.getAttr( attrs, 'SubmitList') , 'numProessors' )
1109                        requested_cpus = self.getAttr( self.getAttr( attrs, 'submit') , 'numProcessors' )
1110                        if requested_cpus == None or requested_cpus == "":
1111                                requested_cpus = 1
1112
[660]1113                        if QUEUE:
1114                            for q in QUEUE:
1115                                if q == queue:
1116                                    display_queue = 1
1117                                    break
1118                                else:
1119                                    display_queue = 0
1120                                    continue
1121                        if display_queue == 0:
1122                            continue
[524]1123
1124                        runState = self.getAttr( attrs, 'status' )
1125                        if runState == 4:
1126                                status = 'R'
1127                        else:
1128                                status = 'Q'
1129                        queued_timestamp = self.getAttr( attrs, 'submitTime' )
1130
1131                        if status == 'R':
1132                                start_timestamp = self.getAttr( attrs, 'startTime' )
1133                                nodesCpu =  dict(self._countDuplicatesInList(self.getAttr( attrs, 'exHosts' )))
1134                                nodelist = nodesCpu.keys()
1135
1136                                if DETECT_TIME_DIFFS:
1137
1138                                        # If a job start if later than our current date,
1139                                        # that must mean the Torque server's time is later
1140                                        # than our local time.
1141
1142                                        if int(start_timestamp) > int( int(self.cur_time) + int(self.timeoffset) ):
1143
1144                                                self.timeoffset = int( int(start_timestamp) - int(self.cur_time) )
1145
1146                        elif status == 'Q':
1147                                start_timestamp = ''
1148                                count_mynodes = 0
1149                                numeric_node = 1
1150                                nodelist = ''
1151
1152                        myAttrs = { }
1153                        if name == "":
1154                                myAttrs['name'] = "none"
1155                        else:
1156                                myAttrs['name'] = name
1157
[659]1158                        myAttrs[ 'owner' ]      = owner
1159                        myAttrs[ 'requested_time' ] = str(requested_time)
1160                        myAttrs[ 'requested_memory' ]   = str(requested_memory)
1161                        myAttrs[ 'requested_cpus' ] = str(requested_cpus)
1162                        myAttrs[ 'ppn' ]        = str( ppn )
1163                        myAttrs[ 'status' ]     = status
1164                        myAttrs[ 'start_timestamp' ]    = str(start_timestamp)
1165                        myAttrs[ 'queue' ]      = str(queue)
1166                        myAttrs[ 'queued_timestamp' ]   = str(queued_timestamp)
1167                        myAttrs[ 'reported' ]       = str( int( int( self.cur_time ) + int( self.timeoffset ) ) )
1168                        myAttrs[ 'nodes' ]      = do_nodelist( nodelist )
[660]1169                        myAttrs[ 'domain' ]     = fqdn_parts( socket.getfqdn() )[1]
[659]1170                        myAttrs[ 'poll_interval' ]  = str(BATCH_POLL_INTERVAL)
[524]1171
1172                        if self.jobDataChanged( jobs, job_id, myAttrs ) and myAttrs['status'] in [ 'R', 'Q' ]:
1173                                jobs[ job_id ] = myAttrs
1174
[525]1175                                debug_msg( 10, printTime() + ' job %s state changed' %(job_id) )
[524]1176
1177                for id, attrs in jobs.items():
1178                        if id not in jobs_processed:
[525]1179                                # This one isn't there anymore
[524]1180                                #
1181                                del jobs[ id ]
1182
[660]1183                self.jobs = jobs
[524]1184
[355]1185class PbsDataGatherer( DataGatherer ):
[318]1186
[659]1187    """This is the DataGatherer for PBS and Torque"""
[318]1188
[659]1189    global PBSQuery, PBSError
[256]1190
[659]1191    def __init__( self ):
[354]1192
[659]1193        """Setup appropriate variables"""
[23]1194
[659]1195        self.jobs   = { }
1196        self.timeoffset = 0
1197        self.dp     = DataProcessor()
[354]1198
[659]1199        self.initPbsQuery()
[23]1200
[659]1201    def initPbsQuery( self ):
[91]1202
[659]1203        self.pq     = None
[354]1204
[659]1205        if( BATCH_SERVER ):
[354]1206
[659]1207            self.pq     = PBSQuery( BATCH_SERVER )
1208        else:
1209            self.pq     = PBSQuery()
[91]1210
[659]1211    def getJobData( self ):
[354]1212
[659]1213        """Gather all data on current jobs in Torque"""
[26]1214
[659]1215        joblist     = {}
1216        self.cur_time   = 0
[349]1217
[659]1218        try:
1219            joblist     = self.pq.getjobs()
1220            self.cur_time   = time.time()
[354]1221
[659]1222        except PBSError, detail:
[354]1223
[659]1224            debug_msg( 10, "Caught PBS unavailable, skipping until next polling interval: " + str( detail ) )
1225            return None
[354]1226
[659]1227        jobs_processed  = [ ]
[26]1228
[659]1229        for name, attrs in joblist.items():
1230            display_queue       = 1
1231            job_id          = name.split( '.' )[0]
[26]1232
[661]1233            owner           = self.getAttr( attrs, 'Job_Owner' )
[659]1234            name            = self.getAttr( attrs, 'Job_Name' )
1235            queue           = self.getAttr( attrs, 'queue' )
[661]1236            nodect          = self.getAttr( attrs['nodes'], 'nodect' )
1237            exec_group      = self.getAttr( attrs, 'egroup' )
[317]1238
[660]1239            requested_time      = self.getAttr( attrs['Resource_List'], 'walltime' )
1240            requested_memory    = self.getAttr( attrs['Resource_List'], 'mem' )
[95]1241
[661]1242
1243            requested_nodes     = ''
[660]1244            mynoderequest       = self.getAttr( attrs['Resource_List'], 'nodes' )
[95]1245
[659]1246            ppn         = ''
[661]1247            attributes  = ''
[281]1248
[661]1249            if mynoderequest.find( ':' ) != -1:
[95]1250
[659]1251                mynoderequest_fields    = mynoderequest.split( ':' )
[281]1252
[659]1253                for mynoderequest_field in mynoderequest_fields:
[281]1254
[661]1255                    if mynoderequest_field.isdigit():
1256
1257                        continue #TODO add requested_nodes if is hostname(s)
1258
[659]1259                    if mynoderequest_field.find( 'ppn' ) != -1:
[281]1260
[659]1261                        ppn = mynoderequest_field.split( 'ppn=' )[1]
[281]1262
[661]1263                    else:
1264
1265                        if attributes == '':
1266
1267                            attributes = '%s' %mynoderequest_field
1268                        else:
1269                            attributes = '%s:%s' %( attributes, mynoderequest_field )
1270
[659]1271            status          = self.getAttr( attrs, 'job_state' )
[25]1272
[661]1273            if status in [ 'Q', 'R', 'W' ]:
[450]1274
[659]1275                jobs_processed.append( job_id )
[450]1276
[661]1277            create_timestamp    = self.getAttr( attrs, 'ctime' )
1278            running_nodes       = ''
1279            exec_nodestr        = '' 
[243]1280
[659]1281            if status == 'R':
[133]1282
[661]1283                #start_timestamp     = self.getAttr( attrs, 'etime' )
1284                start_timestamp     = self.getAttr( attrs, 'start_time' )
1285                exec_nodestr        = self.getAttr( attrs, 'exec_host' )
[133]1286
[661]1287                nodes           = exec_nodestr.split( '+' )
[659]1288                nodeslist       = do_nodelist( nodes )
[661]1289                running_nodes   = string.join( nodeslist, ' ' )
[354]1290
[659]1291                if DETECT_TIME_DIFFS:
[185]1292
[659]1293                    # If a job start if later than our current date,
1294                    # that must mean the Torque server's time is later
1295                    # than our local time.
1296               
1297                    if int( start_timestamp ) > int( int( self.cur_time ) + int( self.timeoffset ) ):
[185]1298
[659]1299                        self.timeoffset = int( int(start_timestamp) - int(self.cur_time) )
[185]1300
[659]1301            elif status == 'Q':
[95]1302
[659]1303                # 'mynodequest' can be a string in the following syntax according to the
1304                # Torque Administator's manual:
1305                #
1306                # {<node_count> | <hostname>}[:ppn=<ppn>][:<property>[:<property>]...][+ ...]
1307                # {<node_count> | <hostname>}[:ppn=<ppn>][:<property>[:<property>]...][+ ...]
1308                # etc
1309                #
[451]1310
[659]1311                #
1312                # For now we only count the amount of nodes request and ignore properties
1313                #
[451]1314
[659]1315                start_timestamp     = ''
1316                count_mynodes       = 0
[354]1317
[661]1318                queued_timestamp    = self.getAttr( attrs, 'qtime' )
1319
[659]1320                for node in mynoderequest.split( '+' ):
[67]1321
[659]1322                    # Just grab the {node_count|hostname} part and ignore properties
1323                    #
1324                    nodepart    = node.split( ':' )[0]
[67]1325
[659]1326                    # Let's assume a node_count value
1327                    #
1328                    numeric_node    = 1
[451]1329
[659]1330                    # Chop the value up into characters
1331                    #
1332                    for letter in nodepart:
[67]1333
[659]1334                        # If this char is not a digit (0-9), this must be a hostname
1335                        #
1336                        if letter not in string.digits:
[133]1337
[659]1338                            numeric_node    = 0
[133]1339
[659]1340                    # If this is a hostname, just count this as one (1) node
1341                    #
1342                    if not numeric_node:
[354]1343
[659]1344                        count_mynodes   = count_mynodes + 1
1345                    else:
[451]1346
[659]1347                        # If this a number, it must be the node_count
1348                        # and increase our count with it's value
1349                        #
1350                        try:
1351                            count_mynodes   = count_mynodes + int( nodepart )
[354]1352
[659]1353                        except ValueError, detail:
[354]1354
[659]1355                            # When we arrive here I must be bugged or very confused
1356                            # THIS SHOULD NOT HAPPEN!
1357                            #
1358                            debug_msg( 10, str( detail ) )
1359                            debug_msg( 10, "Encountered weird node in Resources_List?!" )
1360                            debug_msg( 10, 'nodepart = ' + str( nodepart ) )
1361                            debug_msg( 10, 'job = ' + str( name ) )
1362                            debug_msg( 10, 'attrs = ' + str( attrs ) )
1363                       
1364                nodeslist   = str( count_mynodes )
1365            else:
1366                start_timestamp = ''
1367                nodeslist   = ''
[133]1368
[659]1369            myAttrs             = { }
[26]1370
[661]1371            myAttrs[ 'name' ]              = str( name )
1372            myAttrs[ 'status' ]            = str( status )
1373            myAttrs[ 'queue' ]             = str( queue )
1374            myAttrs[ 'owner' ]             = str( owner )
1375            myAttrs[ 'nodect' ]            = str( nodect )
1376            myAttrs[ 'exec.group' ]        = str( exec_group )
1377            myAttrs[ 'exec.hostnames' ]    = str( running_nodes )
1378            myAttrs[ 'exec.nodestr' ]      = str( exec_nodestr )
1379            myAttrs[ 'req.walltime' ]      = str( requested_time )
1380            myAttrs[ 'req.memory' ]        = str( requested_memory )
1381            myAttrs[ 'req.nodes' ]         = str( requested_nodes )
1382            myAttrs[ 'req.ppn' ]           = str( ppn )
1383            myAttrs[ 'req.attributes' ]    = str( attributes )
1384            myAttrs[ 'timestamp.running' ] = str( start_timestamp )
1385            myAttrs[ 'timestamp.created' ] = str( create_timestamp )
1386            myAttrs[ 'timestamp.queued' ]  = str( queued_timestamp )
[354]1387
[661]1388            if self.jobDataChanged( self.jobs, job_id, myAttrs ) and myAttrs['status'] in [ 'R', 'Q', 'W' ]:
[61]1389
[659]1390                self.jobs[ job_id ] = myAttrs
[26]1391
[659]1392        for id, attrs in self.jobs.items():
[76]1393
[659]1394            if id not in jobs_processed:
[76]1395
[659]1396                # This one isn't there anymore; toedeledoki!
1397                #
1398                del self.jobs[ id ]
[76]1399
[363]1400#
1401# Gmetric by Nick Galbreath - nickg(a.t)modp(d.o.t)com
1402# Version 1.0 - 21-April2-2007
1403# http://code.google.com/p/embeddedgmetric/
1404#
1405# Modified by: Ramon Bastiaans
1406# For the Job Monarch Project, see: https://subtrac.sara.nl/oss/jobmonarch/
1407#
1408# added: DEFAULT_TYPE for Gmetric's
1409# added: checkHostProtocol to determine if target is multicast or not
1410# changed: allow default for Gmetric constructor
1411# changed: allow defaults for all send() values except dmax
1412#
1413
[362]1414GMETRIC_DEFAULT_TYPE    = 'string'
1415GMETRIC_DEFAULT_HOST    = '127.0.0.1'
1416GMETRIC_DEFAULT_PORT    = '8649'
[659]1417GMETRIC_DEFAULT_UNITS   = ''
[362]1418
1419class Gmetric:
1420
[659]1421    global GMETRIC_DEFAULT_HOST, GMETRIC_DEFAULT_PORT
[362]1422
[659]1423    slope           = { 'zero' : 0, 'positive' : 1, 'negative' : 2, 'both' : 3, 'unspecified' : 4 }
1424    type            = ( '', 'string', 'uint16', 'int16', 'uint32', 'int32', 'float', 'double', 'timestamp' )
1425    protocol        = ( 'udp', 'multicast' )
[362]1426
[659]1427    def __init__( self, host=GMETRIC_DEFAULT_HOST, port=GMETRIC_DEFAULT_PORT ):
[362]1428               
[659]1429        global GMETRIC_DEFAULT_TYPE
[362]1430
[659]1431        self.prot       = self.checkHostProtocol( host )
1432        self.msg        = xdrlib.Packer()
1433        self.socket     = socket.socket( socket.AF_INET, socket.SOCK_DGRAM )
[362]1434
[659]1435        if self.prot not in self.protocol:
[362]1436
[659]1437            raise ValueError( "Protocol must be one of: " + str( self.protocol ) )
[362]1438
[659]1439        if self.prot == 'multicast':
[362]1440
[659]1441            # Set multicast options
1442            #
1443            self.socket.setsockopt( socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 20 )
[362]1444
[659]1445        self.hostport   = ( host, int( port ) )
1446        self.slopestr   = 'both'
1447        self.tmax       = 60
[362]1448
[659]1449    def checkHostProtocol( self, ip ):
[362]1450
[659]1451        """Detect if a ip adress is a multicast address"""
[471]1452
[659]1453        MULTICAST_ADDRESS_MIN   = ( "224", "0", "0", "0" )
1454        MULTICAST_ADDRESS_MAX   = ( "239", "255", "255", "255" )
[362]1455
[659]1456        ip_fields               = ip.split( '.' )
[362]1457
[659]1458        if ip_fields >= MULTICAST_ADDRESS_MIN and ip_fields <= MULTICAST_ADDRESS_MAX:
[362]1459
[659]1460            return 'multicast'
1461        else:
1462            return 'udp'
[362]1463
[659]1464    def send( self, name, value, dmax, typestr = '', units = '' ):
[362]1465
[659]1466        if len( units ) == 0:
1467            units       = GMETRIC_DEFAULT_UNITS
[471]1468
[659]1469        if len( typestr ) == 0:
1470            typestr     = GMETRIC_DEFAULT_TYPE
[362]1471
[659]1472        msg             = self.makexdr( name, value, typestr, units, self.slopestr, self.tmax, dmax )
[409]1473
[659]1474        return self.socket.sendto( msg, self.hostport )
[362]1475
[659]1476    def makexdr( self, name, value, typestr, unitstr, slopestr, tmax, dmax ):
[362]1477
[659]1478        if slopestr not in self.slope:
[362]1479
[659]1480            raise ValueError( "Slope must be one of: " + str( self.slope.keys() ) )
[362]1481
[659]1482        if typestr not in self.type:
[362]1483
[659]1484            raise ValueError( "Type must be one of: " + str( self.type ) )
[362]1485
[659]1486        if len( name ) == 0:
[362]1487
[659]1488            raise ValueError( "Name must be non-empty" )
[362]1489
[659]1490        self.msg.reset()
1491        self.msg.pack_int( 0 )
1492        self.msg.pack_string( typestr )
1493        self.msg.pack_string( name )
1494        self.msg.pack_string( str( value ) )
1495        self.msg.pack_string( unitstr )
1496        self.msg.pack_int( self.slope[ slopestr ] )
1497        self.msg.pack_uint( int( tmax ) )
1498        self.msg.pack_uint( int( dmax ) )
[362]1499
[659]1500        return self.msg.get_buffer()
[362]1501
[26]1502def printTime( ):
[354]1503
[659]1504    """Print current time/date in human readable format for log/debug"""
[26]1505
[659]1506    return time.strftime("%a, %d %b %Y %H:%M:%S")
[26]1507
1508def debug_msg( level, msg ):
[354]1509
[659]1510    """Print msg if at or above current debug level"""
[26]1511
[659]1512    global DAEMONIZE, DEBUG_LEVEL, SYSLOG_LEVEL
[377]1513
[660]1514    if (not DAEMONIZE and DEBUG_LEVEL >= level):
[659]1515        sys.stderr.write( msg + '\n' )
[26]1516
[659]1517    if (DAEMONIZE and USE_SYSLOG and SYSLOG_LEVEL >= level):
1518        syslog.syslog( msg )
[373]1519
[307]1520def write_pidfile():
1521
[659]1522    # Write pidfile if PIDFILE is set
1523    #
1524    if PIDFILE:
[307]1525
[659]1526        pid = os.getpid()
[354]1527
[659]1528        pidfile = open( PIDFILE, 'w' )
[354]1529
[659]1530        pidfile.write( str( pid ) )
1531        pidfile.close()
[307]1532
[23]1533def main():
[354]1534
[659]1535    """Application start"""
[23]1536
[659]1537    global PBSQuery, PBSError, lsfObject
1538    global SYSLOG_FACILITY, USE_SYSLOG, BATCH_API, DAEMONIZE
[256]1539
[659]1540    if not processArgs( sys.argv[1:] ):
[354]1541
[659]1542        sys.exit( 1 )
[212]1543
[659]1544    # Load appropriate DataGatherer depending on which BATCH_API is set
1545    # and any required modules for the Gatherer
1546    #
1547    if BATCH_API == 'pbs':
[256]1548
[659]1549        try:
1550            from PBSQuery import PBSQuery, PBSError
[256]1551
[659]1552        except ImportError:
[256]1553
[659]1554            debug_msg( 0, "FATAL ERROR: BATCH_API set to 'pbs' but python module 'pbs_python' is not installed" )
1555            sys.exit( 1 )
[256]1556
[659]1557        gather = PbsDataGatherer()
[256]1558
[659]1559    elif BATCH_API == 'sge':
[256]1560
[659]1561        # Tested with SGE 6.0u11.
1562        #
1563        gather = SgeDataGatherer()
[368]1564
[659]1565    elif BATCH_API == 'lsf':
[368]1566
[659]1567        try:
1568            from lsfObject import lsfObject
1569        except:
1570            debug_msg(0, "fatal error: BATCH_API set to 'lsf' but python module is not found or installed")
1571            sys.exit( 1)
[256]1572
[659]1573        gather = LsfDataGatherer()
[524]1574
[659]1575    else:
1576        debug_msg( 0, "FATAL ERROR: unknown BATCH_API '" + BATCH_API + "' is not supported" )
[354]1577
[659]1578        sys.exit( 1 )
[256]1579
[659]1580    if( DAEMONIZE and USE_SYSLOG ):
[373]1581
[659]1582        syslog.openlog( 'jobmond', syslog.LOG_NOWAIT, SYSLOG_FACILITY )
[373]1583
[659]1584    if DAEMONIZE:
[354]1585
[659]1586        gather.daemon()
1587    else:
1588        gather.run()
[23]1589
[256]1590# wh00t? someone started me! :)
[65]1591#
[23]1592if __name__ == '__main__':
[659]1593    main()
Note: See TracBrowser for help on using the repository browser.