source: branches/0.4/jobmond/jobmond.py @ 701

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