source: branches/1.1/jobmond/jobmond.py @ 941

Last change on this file since 941 was 927, checked in by ramonb, 11 years ago

jobmond/jobmond.py:

  • prevent NoOptionError? Exception to occur when BATCH_HOST_TRANSLATE is not set
  • closes #172
  • Property svn:executable set to *
  • Property svn:keywords set to Id
File size: 64.7 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 927 2013-07-25 12:02:52Z ramonb $
[227]23#
[23]24
[694]25# vi :set ts=4
26
[471]27import sys, getopt, ConfigParser, time, os, socket, string, re
[837]28import xdrlib, socket, syslog, xml, xml.sax, shlex, os.path, pwd
[318]29from xml.sax.handler import feature_namespaces
[623]30from collections import deque
[699]31from glob import glob
[318]32
[913]33VERSION='__VERSION__'
[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
[704]60    global PIDFILE, JOBMOND_CONF
[701]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
[851]72        usage( False )
[691]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
[707]356        udp_send_channels = [ ] # IP:PORT
357
358        if not self.conf_dict.has_key( 'udp_send_channel' ):
359            return None
360
361        for u in self.conf_dict[ 'udp_send_channel' ]:
362
363            if u.has_key( 'mcast_join' ):
364
365                ip = u['mcast_join'][0]
366
367            elif u.has_key( 'host' ):
368
369                ip = u['host'][0]
370
371            port = u['port'][0]
372
373            udp_send_channels.append( ( ip, port ) )
374
375        if len( udp_send_channels ) == 0:
376            return None
377
378        return udp_send_channels
379
[699]380    def getSectionLastOption( self, section, option ):
381
382        """
383        Get last option set in a config section that could be set multiple times in multiple (include) files.
384
385        i.e.: getSectionLastOption( 'globals', 'send_metadata_interval' )
386        """
387
388        self.checkConfDict()
389        value = None
390
391        if not self.conf_dict.has_key( section ):
392
393            return None
394
395        # Could be set multiple times in multiple (include) files: get last one set
396        for c in self.conf_dict[ section ]:
397
398                if c.has_key( option ):
399
[705]400                    value = c[ option ][0]
[699]401
[705]402        return value
[699]403
404    def getClusterName( self ):
405
406        return self.getSectionLastOption( 'cluster', 'name' )
407
408    def getVal( self, section, option ):
409
410        return self.getSectionLastOption( section, option )
411
[691]412    def getInt( self, section, valname ):
[520]413
[691]414        value    = self.getVal( section, valname )
[520]415
[691]416        if not value:
[699]417            return None
[520]418
[691]419        return int( value )
[520]420
[691]421    def getStr( self, section, valname ):
[520]422
[691]423        value    = self.getVal( section, valname )
[520]424
[691]425        if not value:
[699]426            return None
[520]427
[691]428        return str( value )
[520]429
430def findGmetric():
431
[691]432    for dir in os.path.expandvars( '$PATH' ).split( ':' ):
[520]433
[691]434        guess    = '%s/%s' %( dir, 'gmetric' )
[520]435
[691]436        if os.path.exists( guess ):
[520]437
[691]438            return guess
[520]439
[691]440    return False
[520]441
[212]442def loadConfig( filename ):
443
[691]444    def getlist( cfg_string ):
[215]445
[691]446        my_list = [ ]
[215]447
[691]448        for item_txt in cfg_string.split( ',' ):
[215]449
[691]450            sep_char = None
[215]451
[691]452            item_txt = item_txt.strip()
[215]453
[691]454            for s_char in [ "'", '"' ]:
[215]455
[691]456                if item_txt.find( s_char ) != -1:
[215]457
[691]458                    if item_txt.count( s_char ) != 2:
[215]459
[691]460                        print 'Missing quote: %s' %item_txt
461                        sys.exit( 1 )
[215]462
[691]463                    else:
[215]464
[691]465                        sep_char = s_char
466                        break
[215]467
[691]468            if sep_char:
[215]469
[691]470                item_txt = item_txt.split( sep_char )[1]
[215]471
[691]472            my_list.append( item_txt )
[215]473
[691]474        return my_list
[215]475
[784]476    if not os.path.isfile( JOBMOND_CONF ):
477
478        print "Is not a file or does not exist: '%s'" %JOBMOND_CONF
479        sys.exit( 1 )
480
481    try:
482        f = open( JOBMOND_CONF, 'r' )
483    except IOError, detail:
484        print "Cannot read config file: '%s'" %JOBMOND_CONF
485        sys.exit( 1 )
486    else:
487        f.close()
488
[691]489    cfg        = ConfigParser.ConfigParser()
[212]490
[691]491    cfg.read( filename )
[212]492
[691]493    global DEBUG_LEVEL, DAEMONIZE, BATCH_SERVER, BATCH_POLL_INTERVAL
494    global GMOND_CONF, DETECT_TIME_DIFFS, BATCH_HOST_TRANSLATE
495    global BATCH_API, QUEUE, GMETRIC_TARGET, USE_SYSLOG
496    global SYSLOG_LEVEL, SYSLOG_FACILITY, GMETRIC_BINARY
[707]497    global METRIC_MAX_VAL_LEN, GMOND_UDP_SEND_CHANNELS
[212]498
[701]499    DEBUG_LEVEL = cfg.getint( 'DEFAULT', 'DEBUG_LEVEL' )
[212]500
[701]501    DAEMONIZE   = cfg.getboolean( 'DEFAULT', 'DAEMONIZE' )
[212]502
[691]503    SYSLOG_LEVEL    = -1
[701]504    SYSLOG_FACILITY = None
[377]505
[691]506    try:
[701]507        USE_SYSLOG  = cfg.getboolean( 'DEFAULT', 'USE_SYSLOG' )
[212]508
[691]509    except ConfigParser.NoOptionError:
[373]510
[701]511        USE_SYSLOG  = True
[373]512
[691]513        debug_msg( 0, 'ERROR: no option USE_SYSLOG found: assuming yes' )
[373]514
[691]515    if USE_SYSLOG:
[373]516
[691]517        try:
[701]518            SYSLOG_LEVEL = cfg.getint( 'DEFAULT', 'SYSLOG_LEVEL' )
[373]519
[691]520        except ConfigParser.NoOptionError:
[373]521
[691]522            debug_msg( 0, 'ERROR: no option SYSLOG_LEVEL found: assuming level 0' )
[701]523            SYSLOG_LEVEL = 0
[373]524
[691]525        try:
[373]526
[691]527            SYSLOG_FACILITY = eval( 'syslog.LOG_' + cfg.get( 'DEFAULT', 'SYSLOG_FACILITY' ) )
[373]528
[691]529        except ConfigParser.NoOptionError:
[373]530
[691]531            SYSLOG_FACILITY = syslog.LOG_DAEMON
[373]532
[691]533            debug_msg( 0, 'ERROR: no option SYSLOG_FACILITY found: assuming facility DAEMON' )
[373]534
[691]535    try:
[373]536
[701]537        BATCH_SERVER = cfg.get( 'DEFAULT', 'BATCH_SERVER' )
[212]538
[691]539    except ConfigParser.NoOptionError:
[265]540
[864]541        # Not required for all API's: only pbs api allows remote connections
542        BATCH_SERVER = None
[265]543
[691]544    try:
545   
[701]546        BATCH_POLL_INTERVAL = cfg.getint( 'DEFAULT', 'BATCH_POLL_INTERVAL' )
[265]547
[691]548    except ConfigParser.NoOptionError:
[265]549
[691]550        # Backwards compatibility for old configs
551        #
[265]552
[701]553        BATCH_POLL_INTERVAL = cfg.getint( 'DEFAULT', 'TORQUE_POLL_INTERVAL' )
554        api_guess           = 'pbs'
[691]555   
556    try:
[212]557
[701]558        GMOND_CONF          = cfg.get( 'DEFAULT', 'GMOND_CONF' )
[353]559
[691]560    except ConfigParser.NoOptionError:
[353]561
[703]562        # Not specified: assume /etc/ganglia/gmond.conf
[691]563        #
[703]564        GMOND_CONF          = '/etc/ganglia/gmond.conf'
[353]565
[707]566    ganglia_cfg             = GangliaConfigParser( GMOND_CONF )
567    GMETRIC_TARGET          = None
[449]568
[707]569    GMOND_UDP_SEND_CHANNELS = ganglia_cfg.getUdpSendChannels()
[449]570
[707]571    if not GMOND_UDP_SEND_CHANNELS:
[449]572
[701]573        debug_msg( 0, "WARNING: Can't parse udp_send_channel from: '%s' - Trying: %s" %( GMOND_CONF, JOBMOND_CONF ) )
[520]574
[691]575        # Couldn't figure it out: let's see if it's in our jobmond.conf
576        #
577        try:
[520]578
[691]579            GMETRIC_TARGET    = cfg.get( 'DEFAULT', 'GMETRIC_TARGET' )
[520]580
[691]581        # Guess not: now just give up
[701]582       
[691]583        except ConfigParser.NoOptionError:
[520]584
[691]585            GMETRIC_TARGET    = None
[520]586
[691]587            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]588
[701]589            gmetric_bin    = findGmetric()
[520]590
[701]591            if gmetric_bin:
[520]592
[701]593                GMETRIC_BINARY     = gmetric_bin
594            else:
595                debug_msg( 0, "WARNING: Can't find gmetric binary anywhere in $PATH" )
[520]596
[701]597                try:
[520]598
[701]599                    GMETRIC_BINARY = cfg.get( 'DEFAULT', 'GMETRIC_BINARY' )
[520]600
[701]601                except ConfigParser.NoOptionError:
[520]602
[786]603                    print "FATAL ERROR: GMETRIC_BINARY not set and not in $PATH"
[701]604                    sys.exit( 1 )
[520]605
[701]606    #TODO: is this really still needed or should be automatic
[691]607    DETECT_TIME_DIFFS    = cfg.getboolean( 'DEFAULT', 'DETECT_TIME_DIFFS' )
[212]608
[927]609    try:
610        BATCH_HOST_TRANSLATE = getlist( cfg.get( 'DEFAULT', 'BATCH_HOST_TRANSLATE' ) )
[215]611
[927]612    except ConfigParser.NoOptionError:
613
614        BATCH_HOST_TRANSLATE = [ ]
615        pass
616
[691]617    try:
[256]618
[691]619        BATCH_API    = cfg.get( 'DEFAULT', 'BATCH_API' )
[266]620
[691]621    except ConfigParser.NoOptionError, detail:
[266]622
[864]623        print "FATAL ERROR: BATCH_API not set"
624        sys.exit( 1 )
[354]625
[691]626    try:
[317]627
[691]628        QUEUE        = getlist( cfg.get( 'DEFAULT', 'QUEUE' ) )
[317]629
[691]630    except ConfigParser.NoOptionError, detail:
[317]631
[691]632        QUEUE        = None
[353]633
[701]634    METRIC_MAX_VAL_LEN = ganglia_cfg.getInt( 'globals', 'max_udp_msg_len' )
635
[691]636    return True
[212]637
[507]638def fqdn_parts (fqdn):
[520]639
[691]640    """Return pair of host and domain for fully-qualified domain name arg."""
[520]641
[691]642    parts = fqdn.split (".")
[520]643
[691]644    return (parts[0], string.join(parts[1:], "."))
[507]645
[61]646class DataProcessor:
[355]647
[691]648    """Class for processing of data"""
[61]649
[691]650    binary = None
[61]651
[691]652    def __init__( self, binary=None ):
[355]653
[691]654        """Remember alternate binary location if supplied"""
[61]655
[691]656        global GMETRIC_BINARY, GMOND_CONF
[449]657
[691]658        if binary:
659            self.binary = binary
[61]660
[707]661        if not self.binary and not GMETRIC_TARGET and not GMOND_UDP_SEND_CHANNELS:
[691]662            self.binary = GMETRIC_BINARY
[449]663
[691]664        # Timeout for XML
665        #
666        # From ganglia's documentation:
667        #
668        # 'A metric will be deleted DMAX seconds after it is received, and
669        # DMAX=0 means eternal life.'
[61]670
[691]671        self.dmax = str( int( int( BATCH_POLL_INTERVAL ) * 2 ) )
[80]672
[707]673        if GMOND_CONF and not GMETRIC_TARGET and not GMOND_UDP_SEND_CHANNELS:
[354]674
[691]675            incompatible = self.checkGmetricVersion()
[61]676
[691]677            if incompatible:
[355]678
[903]679                print 'Ganglia/Gmetric version not compatible, please upgrade to at least 3.3.8'
[691]680                sys.exit( 1 )
[65]681
[691]682    def checkGmetricVersion( self ):
[355]683
[691]684        """
[903]685        Check version of gmetric is at least 3.3.8
[691]686        for the syntax we use
687        """
[65]688
[691]689        global METRIC_MAX_VAL_LEN, GMETRIC_TARGET
[255]690
[691]691        incompatible    = 0
[341]692
[691]693        gfp        = os.popen( self.binary + ' --version' )
[692]694        lines      = gfp.readlines()
[65]695
[691]696        gfp.close()
[355]697
[691]698        for line in lines:
[355]699
[691]700            line = line.split( ' ' )
[65]701
[691]702            if len( line ) == 2 and str( line ).find( 'gmetric' ) != -1:
703           
704                gmetric_version    = line[1].split( '\n' )[0]
[65]705
[691]706                version_major    = int( gmetric_version.split( '.' )[0] )
707                version_minor    = int( gmetric_version.split( '.' )[1] )
708                version_patch    = int( gmetric_version.split( '.' )[2] )
[65]709
[691]710                incompatible    = 0
[65]711
[691]712                if version_major < 3:
[65]713
[691]714                    incompatible = 1
715               
716                elif version_major == 3:
[65]717
[903]718                    if version_minor < 3:
[65]719
[692]720                        incompatible = 1
[65]721
[903]722                    elif version_patch < 8:
723
724                        incompatible = 1
725
[691]726        return incompatible
[65]727
[691]728    def multicastGmetric( self, metricname, metricval, valtype='string', units='' ):
[355]729
[691]730        """Call gmetric binary and multicast"""
[65]731
[691]732        cmd = self.binary
[65]733
[707]734        if GMOND_UDP_SEND_CHANNELS:
[61]735
[707]736            for c_ip, c_port  in GMOND_UDP_SEND_CHANNELS:
737
738                metric_debug        = "[gmetric %s:%s] name: %s - val: %s - dmax: %s" %( str(c_ip), str(c_port), str( metricname ), str( metricval ), str( self.dmax ) )
739
740                debug_msg( 10, printTime() + ' ' + metric_debug)
741
742                gm = Gmetric( c_ip, c_port )
743
744                gm.send( str( metricname ), str( metricval ), str( self.dmax ), valtype, units )
745
746        elif GMETRIC_TARGET:
747
[691]748            GMETRIC_TARGET_HOST    = GMETRIC_TARGET.split( ':' )[0]
749            GMETRIC_TARGET_PORT    = GMETRIC_TARGET.split( ':' )[1]
[353]750
[691]751            metric_debug        = "[gmetric] name: %s - val: %s - dmax: %s" %( str( metricname ), str( metricval ), str( self.dmax ) )
[353]752
[691]753            debug_msg( 10, printTime() + ' ' + metric_debug)
[353]754
[691]755            gm = Gmetric( GMETRIC_TARGET_HOST, GMETRIC_TARGET_PORT )
[353]756
[691]757            gm.send( str( metricname ), str( metricval ), str( self.dmax ), valtype, units )
[353]758
[691]759        else:
760            try:
761                cmd = cmd + ' -c' + GMOND_CONF
[353]762
[691]763            except NameError:
[353]764
[705]765                debug_msg( 10, 'Assuming /etc/ganglia/gmond.conf for gmetric cmd' )
[353]766
[691]767            cmd = cmd + ' -n' + str( metricname )+ ' -v"' + str( metricval )+ '" -t' + str( valtype ) + ' -d' + str( self.dmax )
[353]768
[691]769            if len( units ) > 0:
[409]770
[691]771                cmd = cmd + ' -u"' + units + '"'
[409]772
[691]773            debug_msg( 10, printTime() + ' ' + cmd )
[353]774
[691]775            os.system( cmd )
[353]776
[318]777class DataGatherer:
[23]778
[691]779    """Skeleton class for batch system DataGatherer"""
[256]780
[691]781    def printJobs( self, jobs ):
[355]782
[691]783        """Print a jobinfo overview"""
[318]784
[691]785        for name, attrs in self.jobs.items():
[318]786
[691]787            print 'job %s' %(name)
[318]788
[691]789            for name, val in attrs.items():
[318]790
[691]791                print '\t%s = %s' %( name, val )
[318]792
[691]793    def printJob( self, jobs, job_id ):
[355]794
[691]795        """Print job with job_id from jobs"""
[318]796
[691]797        print 'job %s' %(job_id)
[318]798
[691]799        for name, val in jobs[ job_id ].items():
[318]800
[691]801            print '\t%s = %s' %( name, val )
[318]802
[691]803    def getAttr( self, attrs, name ):
[507]804
[691]805        """Return certain attribute from dictionary, if exists"""
[507]806
[691]807        if attrs.has_key( name ):
[507]808
[691]809            return attrs[ name ]
810        else:
811            return ''
[507]812
[691]813    def jobDataChanged( self, jobs, job_id, attrs ):
[507]814
[691]815        """Check if job with attrs and job_id in jobs has changed"""
[507]816
[691]817        if jobs.has_key( job_id ):
[507]818
[691]819            oldData = jobs[ job_id ]   
820        else:
821            return 1
[507]822
[691]823        for name, val in attrs.items():
[507]824
[691]825            if oldData.has_key( name ):
[507]826
[691]827                if oldData[ name ] != attrs[ name ]:
[507]828
[691]829                    return 1
[507]830
[691]831            else:
832                return 1
[507]833
[691]834        return 0
[507]835
[691]836    def submitJobData( self ):
[507]837
[691]838        """Submit job info list"""
[507]839
[691]840        global BATCH_API
[512]841
[728]842        self.dp.multicastGmetric( 'zplugin_monarch_heartbeat', str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
[507]843
[724]844        running_jobs = 0
845        queued_jobs  = 0
[507]846
[691]847        # Count how many running/queued jobs we found
848        #
849        for jobid, jobattrs in self.jobs.items():
[507]850
[691]851            if jobattrs[ 'status' ] == 'Q':
[507]852
[691]853                queued_jobs += 1
[507]854
[691]855            elif jobattrs[ 'status' ] == 'R':
[507]856
[691]857                running_jobs += 1
[507]858
[691]859        # Report running/queued jobs as seperate metric for a nice RRD graph
860        #
[728]861        self.dp.multicastGmetric( 'zplugin_monarch_rj', str( running_jobs ), 'uint32', 'jobs' )
862        self.dp.multicastGmetric( 'zplugin_monarch_qj', str( queued_jobs ), 'uint32', 'jobs' )
[507]863
[691]864        # Report down/offline nodes in batch (PBS only ATM)
865        #
[866]866        if BATCH_API in [ 'pbs', 'slurm' ]:
[512]867
[691]868            domain        = fqdn_parts( socket.getfqdn() )[1]
[514]869
[728]870            downed_nodes  = list()
871            offline_nodes = list()
[691]872       
873            l        = ['state']
[512]874
[790]875            nodelist = self.getNodeData()
876
877            for name, node in nodelist.items():
878
[691]879                if ( node[ 'state' ].find( "down" ) != -1 ):
[512]880
[691]881                    downed_nodes.append( name )
[512]882
[691]883                if ( node[ 'state' ].find( "offline" ) != -1 ):
[512]884
[691]885                    offline_nodes.append( name )
[512]886
[728]887            downnodeslist    = do_nodelist( downed_nodes )
888            offlinenodeslist = do_nodelist( offline_nodes )
[512]889
[691]890            down_str    = 'nodes=%s domain=%s reported=%s' %( string.join( downnodeslist, ';' ), domain, str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
891            offl_str    = 'nodes=%s domain=%s reported=%s' %( string.join( offlinenodeslist, ';' ), domain, str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
[728]892            self.dp.multicastGmetric( 'zplugin_monarch_down'   , down_str )
893            self.dp.multicastGmetric( 'zplugin_monarch_offline', offl_str )
[514]894
[691]895        # Now let's spread the knowledge
896        #
897        for jobid, jobattrs in self.jobs.items():
[507]898
[691]899            # Make gmetric values for each job: respect max gmetric value length
900            #
901            gmetric_val        = self.compileGmetricVal( jobid, jobattrs )
902            metric_increment    = 0
[507]903
[691]904            # If we have more job info than max gmetric value length allows, split it up
905            # amongst multiple metrics
906            #
907            for val in gmetric_val:
[507]908
[728]909                metric_name = 'zplugin_monarch_job_%s_%s' %( str(metric_increment) , str( jobid ) )
910                self.dp.multicastGmetric( metric_name, val )
[507]911
[691]912                # Increase follow number if this jobinfo is split up amongst more than 1 gmetric
913                #
914                metric_increment    = metric_increment + 1
[507]915
[691]916    def compileGmetricVal( self, jobid, jobattrs ):
[507]917
[691]918        """Create a val string for gmetric of jobinfo"""
[507]919
[691]920        gval_lists    = [ ]
921        val_list    = { }
[507]922
[691]923        for val_name, val_value in jobattrs.items():
[507]924
[691]925            # These are our own metric names, i.e.: status, start_timestamp, etc
926            #
927            val_list_names_len    = len( string.join( val_list.keys() ) ) + len(val_list.keys())
[507]928
[691]929            # These are their corresponding values
930            #
931            val_list_vals_len    = len( string.join( val_list.values() ) ) + len(val_list.values())
[507]932
[691]933            if val_name == 'nodes' and jobattrs['status'] == 'R':
[507]934
[691]935                node_str = None
[507]936
[691]937                for node in val_value:
[507]938
[691]939                    if node_str:
[507]940
[691]941                        node_str = node_str + ';' + node
942                    else:
943                        node_str = node
[507]944
[691]945                    # Make sure if we add this new info, that the total metric's value length does not exceed METRIC_MAX_VAL_LEN
946                    #
947                    if (val_list_names_len + len(val_name) ) + (val_list_vals_len + len(node_str) ) > METRIC_MAX_VAL_LEN:
[507]948
[691]949                        # It's too big, we need to make a new gmetric for the additional info
950                        #
951                        val_list[ val_name ]    = node_str
[507]952
[691]953                        gval_lists.append( val_list )
[507]954
[691]955                        val_list        = { }
956                        node_str        = None
[507]957
[691]958                val_list[ val_name ]    = node_str
[507]959
[691]960                gval_lists.append( val_list )
[507]961
[691]962                val_list        = { }
[507]963
[691]964            elif val_value != '':
[507]965
[691]966                # Make sure if we add this new info, that the total metric's value length does not exceed METRIC_MAX_VAL_LEN
967                #
968                if (val_list_names_len + len(val_name) ) + (val_list_vals_len + len(str(val_value)) ) > METRIC_MAX_VAL_LEN:
[507]969
[691]970                    # It's too big, we need to make a new gmetric for the additional info
971                    #
972                    gval_lists.append( val_list )
[507]973
[691]974                    val_list        = { }
[507]975
[691]976                val_list[ val_name ]    = val_value
[507]977
[691]978        if len( val_list ) > 0:
[507]979
[691]980            gval_lists.append( val_list )
[507]981
[691]982        str_list    = [ ]
[507]983
[691]984        # Now append the value names and values together, i.e.: stop_timestamp=value, etc
985        #
986        for val_list in gval_lists:
[507]987
[691]988            my_val_str    = None
[507]989
[691]990            for val_name, val_value in val_list.items():
[507]991
[691]992                if type(val_value) == list:
[579]993
[691]994                    val_value    = val_value.join( ',' )
[579]995
[691]996                if my_val_str:
[507]997
[691]998                    try:
999                        # fixme: It's getting
1000                        # ('nodes', None) items
1001                        my_val_str = my_val_str + ' ' + val_name + '=' + val_value
1002                    except:
1003                        pass
[623]1004
[691]1005                else:
1006                    my_val_str = val_name + '=' + val_value
[507]1007
[691]1008            str_list.append( my_val_str )
[507]1009
[691]1010        return str_list
[507]1011
[691]1012    def daemon( self ):
[355]1013
[691]1014        """Run as daemon forever"""
[256]1015
[691]1016        # Fork the first child
1017        #
1018        pid = os.fork()
1019        if pid > 0:
1020            sys.exit(0)  # end parent
[256]1021
[691]1022        # creates a session and sets the process group ID
1023        #
1024        os.setsid()
[318]1025
[691]1026        # Fork the second child
1027        #
1028        pid = os.fork()
1029        if pid > 0:
1030            sys.exit(0)  # end parent
[318]1031
[691]1032        write_pidfile()
[318]1033
[691]1034        # Go to the root directory and set the umask
1035        #
1036        os.chdir('/')
1037        os.umask(0)
[318]1038
[691]1039        sys.stdin.close()
1040        sys.stdout.close()
1041        sys.stderr.close()
[318]1042
[691]1043        os.open('/dev/null', os.O_RDWR)
1044        os.dup2(0, 1)
1045        os.dup2(0, 2)
[318]1046
[691]1047        self.run()
[318]1048
[691]1049    def run( self ):
[355]1050
[691]1051        """Main thread"""
[256]1052
[691]1053        while ( 1 ):
1054       
1055            self.getJobData()
1056            self.submitJobData()
1057            time.sleep( BATCH_POLL_INTERVAL )   
[256]1058
[623]1059# SGE code by Dave Love <fx@gnu.org>.  Tested with SGE 6.0u8 and 6.0u11.  May
1060# work with SGE 6.1 (else should be easily fixable), but definitely doesn't
1061# with 6.2.  See also the fixmes.
[256]1062
[507]1063class NoJobs (Exception):
[691]1064    """Exception raised by empty job list in qstat output."""
1065    pass
[256]1066
[507]1067class SgeQstatXMLParser(xml.sax.handler.ContentHandler):
[691]1068    """SAX handler for XML output from Sun Grid Engine's `qstat'."""
[318]1069
[691]1070    def __init__(self):
1071        self.value = ""
1072        self.joblist = []
1073        self.job = {}
1074        self.queue = ""
1075        self.in_joblist = False
1076        self.lrequest = False
1077        self.eltq = deque()
1078        xml.sax.handler.ContentHandler.__init__(self)
[318]1079
[691]1080    # The structure of the output is as follows (for SGE 6.0).  It's
1081    # similar for 6.1, but radically different for 6.2, and is
1082    # undocumented generally.  Unfortunately it's voluminous, and probably
1083    # doesn't scale to large clusters/queues.
[318]1084
[691]1085    # <detailed_job_info  xmlns:xsd="http://www.w3.org/2001/XMLSchema">
1086    #   <djob_info>
1087    #     <qmaster_response>  <!-- job -->
1088    #       ...
1089    #       <JB_ja_template> 
1090    #     <ulong_sublist>
1091    #     ...         <!-- start_time, state ... -->
1092    #     </ulong_sublist>
1093    #       </JB_ja_template> 
1094    #       <JB_ja_tasks>
1095    #     <ulong_sublist>
1096    #       ...       <!-- task info
1097    #     </ulong_sublist>
1098    #     ...
1099    #       </JB_ja_tasks>
1100    #       ...
1101    #     </qmaster_response>
1102    #   </djob_info>
1103    #   <messages>
1104    #   ...
[318]1105
[691]1106    # NB.  We might treat each task as a separate job, like
1107    # straight qstat output, but the web interface expects jobs to
1108    # be identified by integers, not, say, <job number>.<task>.
[318]1109
[691]1110    # So, I lied.  If the job list is empty, we get invalid XML
1111    # like this, which we need to defend against:
[318]1112
[691]1113    # <unknown_jobs  xmlns:xsd="http://www.w3.org/2001/XMLSchema">
1114    #   <>
1115    #     <ST_name>*</ST_name>
1116    #   </>
1117    # </unknown_jobs>
[318]1118
[691]1119    def startElement(self, name, attrs):
1120        self.value = ""
1121        if name == "djob_info":    # job list
1122            self.in_joblist = True
1123        # The job container is "qmaster_response" in SGE 6.0
1124        # and 6.1, but "element" in 6.2.  This is only the very
1125        # start of what's necessary for 6.2, though (sigh).
1126        elif (name == "qmaster_response" or name == "element") \
1127                and self.eltq[-1] == "djob_info": # job
1128            self.job = {"job_state": "U", "slots": 0,
1129                    "nodes": [], "queued_timestamp": "",
1130                    "queued_timestamp": "", "queue": "",
1131                    "ppn": "0", "RN_max": 0,
1132                    # fixme in endElement
1133                    "requested_memory": 0, "requested_time": 0
1134                    }
1135            self.joblist.append(self.job)
1136        elif name == "qstat_l_requests": # resource request
1137            self.lrequest = True
1138        elif name == "unknown_jobs":
1139            raise NoJobs
1140        self.eltq.append (name)
[318]1141
[691]1142    def characters(self, ch):
1143        self.value += ch
[318]1144
[691]1145    def endElement(self, name): 
1146        """Snarf job elements contents into job dictionary.
1147           Translate keys if appropriate."""
[318]1148
[691]1149        name_trans = {
1150          "JB_job_number": "number",
1151          "JB_job_name": "name", "JB_owner": "owner",
1152          "queue_name": "queue", "JAT_start_time": "start_timestamp",
1153          "JB_submission_time": "queued_timestamp"
1154          }
1155        value = self.value
1156        self.eltq.pop ()
[318]1157
[691]1158        if name == "djob_info":
1159            self.in_joblist = False
1160            self.job = {}
1161        elif name == "JAT_master_queue":
1162            self.job["queue"] = value.split("@")[0]
1163        elif name == "JG_qhostname":
1164            if not (value in self.job["nodes"]):
1165                self.job["nodes"].append(value)
1166        elif name == "JG_slots": # slots in use
1167            self.job["slots"] += int(value)
1168        elif name == "RN_max": # requested slots (tasks or parallel)
1169            self.job["RN_max"] = max (self.job["RN_max"],
1170                          int(value))
1171        elif name == "JAT_state": # job state (bitwise or)
1172            value = int (value)
1173            # Status values from sge_jobL.h
1174            #define JIDLE           0x00000000
1175            #define JHELD           0x00000010
1176            #define JMIGRATING          0x00000020
1177            #define JQUEUED         0x00000040
1178            #define JRUNNING        0x00000080
1179            #define JSUSPENDED          0x00000100
1180            #define JTRANSFERING        0x00000200
1181            #define JDELETED        0x00000400
1182            #define JWAITING        0x00000800
1183            #define JEXITING        0x00001000
1184            #define JWRITTEN        0x00002000
1185            #define JSUSPENDED_ON_THRESHOLD 0x00010000
1186            #define JFINISHED           0x00010000
1187            if value & 0x80:
1188                self.job["status"] = "R"
1189            elif value & 0x40:
1190                self.job["status"] = "Q"
1191            else:
1192                self.job["status"] = "O" # `other'
1193        elif name == "CE_name" and self.lrequest and self.value in \
1194                ("h_cpu", "s_cpu", "cpu", "h_core", "s_core"):
1195            # We're in a container for an interesting resource
1196            # request; record which type.
1197            self.lrequest = self.value
1198        elif name == "CE_doubleval" and self.lrequest:
1199            # if we're in a container for an interesting
1200            # resource request, use the maxmimum of the hard
1201            # and soft requests to record the requested CPU
1202            # or core.  Fixme:  I'm not sure if this logic is
1203            # right.
1204            if self.lrequest in ("h_core", "s_core"):
1205                self.job["requested_memory"] = \
1206                    max (float (value),
1207                     self.job["requested_memory"])
1208            # Fixme:  Check what cpu means, c.f [hs]_cpu.
1209            elif self.lrequest in ("h_cpu", "s_cpu", "cpu"):
1210                self.job["requested_time"] = \
1211                    max (float (value),
1212                     self.job["requested_time"])
1213        elif name == "qstat_l_requests":
1214            self.lrequest = False
1215        elif self.job and self.in_joblist:
1216            if name in name_trans:
1217                name = name_trans[name]
1218                self.job[name] = value
[318]1219
[507]1220# Abstracted from PBS original.
1221# Fixme:  Is it worth (or appropriate for PBS) sorting the result?
[520]1222#
1223def do_nodelist( nodes ):
1224
[691]1225    """Translate node list as appropriate."""
[520]1226
[691]1227    nodeslist        = [ ]
1228    my_domain        = fqdn_parts( socket.getfqdn() )[1]
[520]1229
[691]1230    for node in nodes:
[520]1231
[691]1232        host        = node.split( '/' )[0] # not relevant for SGE
1233        h, host_domain    = fqdn_parts(host)
[520]1234
[691]1235        if host_domain == my_domain:
[520]1236
[691]1237            host    = h
[520]1238
[691]1239        if nodeslist.count( host ) == 0:
[520]1240
[691]1241            for translate_pattern in BATCH_HOST_TRANSLATE:
[520]1242
[691]1243                if translate_pattern.find( '/' ) != -1:
[520]1244
[691]1245                    translate_orig    = \
1246                        translate_pattern.split( '/' )[1]
1247                    translate_new    = \
1248                        translate_pattern.split( '/' )[2]
1249                    host = re.sub( translate_orig,
1250                               translate_new, host )
1251            if not host in nodeslist:
1252                nodeslist.append( host )
1253    return nodeslist
[318]1254
[837]1255class SLURMDataGatherer( DataGatherer ):
1256
1257    global pyslurm
1258
1259    """This is the DataGatherer for SLURM"""
1260
1261    def __init__( self ):
1262
1263        """Setup appropriate variables"""
1264
1265        self.jobs       = { }
1266        self.timeoffset = 0
1267        self.dp         = DataProcessor()
1268
1269    def getNodeData( self ):
1270
[866]1271        slurm_type  = pyslurm.node()
[837]1272
[866]1273        slurm_nodes = slurm_type.get()
[837]1274
[866]1275        nodedict    = { }
1276
1277        for node, attrs in slurm_nodes.items():
1278
1279            ( num_state, name_state ) = attrs['node_state'] 
1280
1281            if name_state == 'DOWN':
1282
1283                nodedict[ node ] = { 'state' : 'down' }
1284
1285            elif name_state == 'DRAIN':
1286
1287                nodedict[ node ] = { 'state' : 'offline' }
1288
[837]1289        return nodedict
1290
1291    def getJobData( self ):
1292
1293        """Gather all data on current jobs"""
1294
1295        joblist            = {}
1296
1297        self.cur_time  = time.time()
1298
1299        slurm_type = pyslurm.job()
1300        joblist    = slurm_type.get()
1301
1302        jobs_processed    = [ ]
1303
1304        for name, attrs in joblist.items():
1305            display_queue = 1
1306            job_id        = name
1307
1308            name          = self.getAttr( attrs, 'name' )
1309            queue         = self.getAttr( attrs, 'partition' )
1310
1311            if QUEUE:
1312                for q in QUEUE:
1313                    if q == queue:
1314                        display_queue = 1
1315                        break
1316                    else:
1317                        display_queue = 0
1318                        continue
1319            if display_queue == 0:
1320                continue
1321
1322            owner_uid        = attrs[ 'user_id' ]
1323            ( owner, owner_pw, owner_uid, owner_gid, owner_gecos, owner_dir, owner_shell ) = pwd.getpwuid( owner_uid )
1324
1325            requested_time   = self.getAttr( attrs, 'time_limit' )
[852]1326            min_memory       = self.getAttr( attrs, 'pn_min_memory' )
[837]1327
[852]1328            if min_memory == 0:
1329
1330                requested_memory = ''
1331
1332            else:
1333                requested_memory = min_memory
1334
[853]1335            min_cpus = self.getAttr( attrs, 'pn_min_cpus' )
[837]1336
[853]1337            if min_cpus == 0:
1338
1339                ppn = ''
1340
1341            else:
1342                ppn = min_cpus
1343
[837]1344            ( something, status_long ) = self.getAttr( attrs, 'job_state' )
1345
1346            status = 'Q'
1347
1348            if status_long == 'RUNNING':
1349
1350                status = 'R'
1351
1352            elif status_long == 'COMPLETED':
1353
1354                continue
1355
1356            jobs_processed.append( job_id )
1357
1358            queued_timestamp = self.getAttr( attrs, 'submit_time' )
1359
1360            start_timestamp = ''
1361            nodeslist       = ''
1362
1363            if status == 'R':
1364
1365                start_timestamp = self.getAttr( attrs, 'start_time' )
[851]1366                nodes           = attrs[ 'nodes' ]
[837]1367
[851]1368                if not nodes:
[837]1369
[851]1370                    # This should not happen
1371
1372                    # Something wrong: running but 'nodes' returned empty by pyslurm
1373                    # Possible pyslurm bug: abort/quit/warning
1374
1375                    err_msg = 'FATAL ERROR: job %s running but nodes returned empty: pyslurm bugged?' %job_id
1376
1377                    print err_msg
1378                    debug_msg( 0, err_msg )
1379                    sys.exit(1)
1380
1381                my_nodelist = [ ]
1382
1383                slurm_hostlist  = pyslurm.hostlist()
1384                slurm_hostlist.create( nodes )
1385                slurm_hostlist.uniq()
1386
1387                while slurm_hostlist.count() > 0:
1388
1389                    my_nodelist.append( slurm_hostlist.pop() )
1390
1391                slurm_hostlist.destroy()
1392
1393                del slurm_hostlist
1394
1395                nodeslist       = do_nodelist( my_nodelist )
1396
[837]1397                if DETECT_TIME_DIFFS:
1398
1399                    # If a job start if later than our current date,
1400                    # that must mean the Torque server's time is later
1401                    # than our local time.
1402               
1403                    if int( start_timestamp ) > int( int( self.cur_time ) + int( self.timeoffset ) ):
1404
1405                        self.timeoffset = int( int(start_timestamp) - int(self.cur_time) )
1406
1407            elif status == 'Q':
1408
1409                nodeslist       = str( attrs[ 'num_nodes' ] )
1410
1411            else:
1412                start_timestamp = ''
1413                nodeslist       = ''
1414
1415            myAttrs                = { }
1416
1417            myAttrs[ 'name' ]             = str( name )
1418            myAttrs[ 'queue' ]            = str( queue )
1419            myAttrs[ 'owner' ]            = str( owner )
1420            myAttrs[ 'requested_time' ]   = str( requested_time )
1421            myAttrs[ 'requested_memory' ] = str( requested_memory )
1422            myAttrs[ 'ppn' ]              = str( ppn )
1423            myAttrs[ 'status' ]           = str( status )
1424            myAttrs[ 'start_timestamp' ]  = str( start_timestamp )
1425            myAttrs[ 'queued_timestamp' ] = str( queued_timestamp )
1426            myAttrs[ 'reported' ]         = str( int( int( self.cur_time ) + int( self.timeoffset ) ) )
1427            myAttrs[ 'nodes' ]            = nodeslist
1428            myAttrs[ 'domain' ]           = fqdn_parts( socket.getfqdn() )[1]
1429            myAttrs[ 'poll_interval' ]    = str( BATCH_POLL_INTERVAL )
1430
1431            if self.jobDataChanged( self.jobs, job_id, myAttrs ) and myAttrs['status'] in [ 'R', 'Q' ]:
1432
1433                self.jobs[ job_id ] = myAttrs
1434
1435        for id, attrs in self.jobs.items():
1436
1437            if id not in jobs_processed:
1438
1439                # This one isn't there anymore; toedeledoki!
1440                #
1441                del self.jobs[ id ]
1442
[318]1443class SgeDataGatherer(DataGatherer):
1444
[691]1445    jobs = {}
[61]1446
[691]1447    def __init__( self ):
1448        self.jobs = {}
1449        self.timeoffset = 0
1450        self.dp = DataProcessor()
[318]1451
[691]1452    def getJobData( self ):
1453        """Gather all data on current jobs in SGE"""
[318]1454
[691]1455        import popen2
[318]1456
[691]1457        self.cur_time = 0
1458        queues = ""
1459        if QUEUE:    # only for specific queues
1460            # Fixme:  assumes queue names don't contain single
1461            # quote or comma.  Don't know what the SGE rules are.
1462            queues = " -q '" + string.join (QUEUE, ",") + "'"
1463        # Note the comment in SgeQstatXMLParser about scaling with
1464        # this method of getting data.  I haven't found better one.
1465        # Output with args `-xml -ext -f -r' is easier to parse
1466        # in some ways, harder in others, but it doesn't provide
1467        # the submission time (at least SGE 6.0).  The pipeline
1468        # into sed corrects bogus XML observed with a configuration
1469        # of SGE 6.0u8, which otherwise causes the parsing to hang.
1470        piping = popen2.Popen3("qstat -u '*' -j '*' -xml | \
[623]1471sed -e 's/reported usage>/reported_usage>/g' -e 's;<\/*JATASK:.*>;;'" \
[691]1472                           + queues, True)
1473        qstatparser = SgeQstatXMLParser()
1474        parse_err = 0
1475        try:
1476            xml.sax.parse(piping.fromchild, qstatparser)
1477        except NoJobs:
1478            pass
1479        except:
1480            parse_err = 1
[704]1481        if piping.wait():
1482            debug_msg(10, "qstat error, skipping until next polling interval: " + piping.childerr.readline())
[691]1483            return None
1484        elif parse_err:
1485            debug_msg(10, "Bad XML output from qstat"())
1486            exit (1)
1487        for f in piping.fromchild, piping.tochild, piping.childerr:
1488            f.close()
1489        self.cur_time = time.time()
1490        jobs_processed = []
1491        for job in qstatparser.joblist:
1492            job_id = job["number"]
1493            if job["status"] in [ 'Q', 'R' ]:
1494                jobs_processed.append(job_id)
1495            if job["status"] == "R":
1496                job["nodes"] = do_nodelist (job["nodes"])
1497                # Fixme: why is job["nodes"] sometimes null?
1498                try:
1499                    # Fixme: Is this sensible?  The
1500                    # PBS-type PPN isn't something you use
1501                    # with SGE.
[704]1502                    job["ppn"] = float(job["slots"]) / len(job["nodes"])
[691]1503                except:
1504                    job["ppn"] = 0
1505                if DETECT_TIME_DIFFS:
1506                    # If a job start is later than our
1507                    # current date, that must mean
1508                    # the SGE server's time is later
1509                    # than our local time.
[704]1510                    start_timestamp = int (job["start_timestamp"])
1511                    if start_timestamp > int(self.cur_time) + int(self.timeoffset):
[318]1512
[704]1513                        self.timeoffset    = start_timestamp - int(self.cur_time)
[691]1514            else:
1515                # fixme: Note sure what this should be:
1516                job["ppn"] = job["RN_max"]
1517                job["nodes"] = "1"
[318]1518
[691]1519            myAttrs = {}
1520            for attr in ["name", "queue", "owner",
1521                     "requested_time", "status",
1522                     "requested_memory", "ppn",
1523                     "start_timestamp", "queued_timestamp"]:
1524                myAttrs[attr] = str(job[attr])
1525            myAttrs["nodes"] = job["nodes"]
[704]1526            myAttrs["reported"] = str(int(self.cur_time) + int(self.timeoffset))
[691]1527            myAttrs["domain"] = fqdn_parts(socket.getfqdn())[1]
1528            myAttrs["poll_interval"] = str(BATCH_POLL_INTERVAL)
[318]1529
[704]1530            if self.jobDataChanged(self.jobs, job_id, myAttrs) and myAttrs["status"] in ["R", "Q"]:
[691]1531                self.jobs[job_id] = myAttrs
1532        for id, attrs in self.jobs.items():
1533            if id not in jobs_processed:
1534                del self.jobs[id]
[318]1535
[524]1536# LSF code by Mahmoud Hanafi <hanafim@users.sourceforge.nt>
1537# Requres LSFObject http://sourceforge.net/projects/lsfobject
1538#
1539class LsfDataGatherer(DataGatherer):
[525]1540
[691]1541    """This is the DataGatherer for LSf"""
[524]1542
[691]1543    global lsfObject
[524]1544
[691]1545    def __init__( self ):
[525]1546
[691]1547        self.jobs = { }
1548        self.timeoffset = 0
1549        self.dp = DataProcessor()
1550        self.initLsfQuery()
[524]1551
[691]1552    def _countDuplicatesInList( self, dupedList ):
[525]1553
[691]1554        countDupes    = { }
[525]1555
[691]1556        for item in dupedList:
[525]1557
[691]1558            if not countDupes.has_key( item ):
[525]1559
[691]1560                countDupes[ item ]    = 1
1561            else:
1562                countDupes[ item ]    = countDupes[ item ] + 1
[525]1563
[691]1564        dupeCountList    = [ ]
[525]1565
[691]1566        for item, count in countDupes.items():
[525]1567
[691]1568            dupeCountList.append( ( item, count ) )
[525]1569
[691]1570        return dupeCountList
[524]1571#
1572#lst = ['I1','I2','I1','I3','I4','I4','I7','I7','I7','I7','I7']
1573#print _countDuplicatesInList(lst)
1574#[('I1', 2), ('I3', 1), ('I2', 1), ('I4', 2), ('I7', 5)]
1575########################
1576
[691]1577    def initLsfQuery( self ):
1578        self.pq = None
1579        self.pq = lsfObject.jobInfoEntObject()
[524]1580
[691]1581    def getJobData( self, known_jobs="" ):
1582        """Gather all data on current jobs in LSF"""
1583        if len( known_jobs ) > 0:
1584            jobs = known_jobs
1585        else:
1586            jobs = { }
1587        joblist = {}
1588        joblist = self.pq.getJobInfo()
1589        nodelist = ''
[524]1590
[691]1591        self.cur_time = time.time()
[524]1592
[691]1593        jobs_processed = [ ]
[524]1594
[691]1595        for name, attrs in joblist.items():
1596            job_id = str(name)
1597            jobs_processed.append( job_id )
1598            name = self.getAttr( attrs, 'jobName' )
1599            queue = self.getAttr( self.getAttr( attrs, 'submit') , 'queue' )
1600            owner = self.getAttr( attrs, 'user' )
[524]1601
1602### THIS IS THE rLimit List index values
[691]1603#define LSF_RLIMIT_CPU      0        /* cpu time in milliseconds */
1604#define LSF_RLIMIT_FSIZE    1        /* maximum file size */
1605#define LSF_RLIMIT_DATA     2        /* data size */
1606#define LSF_RLIMIT_STACK    3        /* stack size */
1607#define LSF_RLIMIT_CORE     4        /* core file size */
1608#define LSF_RLIMIT_RSS      5        /* resident set size */
1609#define LSF_RLIMIT_NOFILE   6        /* open files */
1610#define LSF_RLIMIT_OPEN_MAX 7        /* (from HP-UX) */
1611#define LSF_RLIMIT_VMEM     8        /* maximum swap mem */
[524]1612#define LSF_RLIMIT_SWAP     8
[691]1613#define LSF_RLIMIT_RUN      9        /* max wall-clock time limit */
1614#define LSF_RLIMIT_PROCESS  10       /* process number limit */
1615#define LSF_RLIMIT_THREAD   11       /* thread number limit (introduced in LSF6.0) */
1616#define LSF_RLIM_NLIMITS    12       /* number of resource limits */
[524]1617
[691]1618            requested_time = self.getAttr( self.getAttr( attrs, 'submit') , 'rLimits' )[9]
1619            if requested_time == -1: 
1620                requested_time = ""
1621            requested_memory = self.getAttr( self.getAttr( attrs, 'submit') , 'rLimits' )[8]
1622            if requested_memory == -1: 
1623                requested_memory = ""
[524]1624# This tries to get proc per node. We don't support this right now
[691]1625            ppn = 0 #self.getAttr( self.getAttr( attrs, 'SubmitList') , 'numProessors' )
1626            requested_cpus = self.getAttr( self.getAttr( attrs, 'submit') , 'numProcessors' )
1627            if requested_cpus == None or requested_cpus == "":
1628                requested_cpus = 1
[524]1629
[691]1630            if QUEUE:
1631                for q in QUEUE:
1632                    if q == queue:
1633                        display_queue = 1
1634                        break
1635                    else:
1636                        display_queue = 0
1637                        continue
1638            if display_queue == 0:
1639                continue
[524]1640
[691]1641            runState = self.getAttr( attrs, 'status' )
1642            if runState == 4:
1643                status = 'R'
1644            else:
1645                status = 'Q'
1646            queued_timestamp = self.getAttr( attrs, 'submitTime' )
[524]1647
[691]1648            if status == 'R':
1649                start_timestamp = self.getAttr( attrs, 'startTime' )
1650                nodesCpu =  dict(self._countDuplicatesInList(self.getAttr( attrs, 'exHosts' )))
1651                nodelist = nodesCpu.keys()
[524]1652
[691]1653                if DETECT_TIME_DIFFS:
[524]1654
[691]1655                    # If a job start if later than our current date,
1656                    # that must mean the Torque server's time is later
1657                    # than our local time.
[524]1658
[691]1659                    if int(start_timestamp) > int( int(self.cur_time) + int(self.timeoffset) ):
[524]1660
[691]1661                        self.timeoffset = int( int(start_timestamp) - int(self.cur_time) )
[524]1662
[691]1663            elif status == 'Q':
1664                start_timestamp = ''
1665                count_mynodes = 0
1666                numeric_node = 1
1667                nodelist = ''
[524]1668
[691]1669            myAttrs = { }
1670            if name == "":
1671                myAttrs['name'] = "none"
1672            else:
1673                myAttrs['name'] = name
[524]1674
[691]1675            myAttrs[ 'owner' ]        = owner
1676            myAttrs[ 'requested_time' ]    = str(requested_time)
1677            myAttrs[ 'requested_memory' ]    = str(requested_memory)
1678            myAttrs[ 'requested_cpus' ]    = str(requested_cpus)
1679            myAttrs[ 'ppn' ]        = str( ppn )
1680            myAttrs[ 'status' ]        = status
1681            myAttrs[ 'start_timestamp' ]    = str(start_timestamp)
1682            myAttrs[ 'queue' ]        = str(queue)
1683            myAttrs[ 'queued_timestamp' ]    = str(queued_timestamp)
1684            myAttrs[ 'reported' ]        = str( int( int( self.cur_time ) + int( self.timeoffset ) ) )
1685            myAttrs[ 'nodes' ]        = do_nodelist( nodelist )
1686            myAttrs[ 'domain' ]        = fqdn_parts( socket.getfqdn() )[1]
1687            myAttrs[ 'poll_interval' ]    = str(BATCH_POLL_INTERVAL)
[524]1688
[691]1689            if self.jobDataChanged( jobs, job_id, myAttrs ) and myAttrs['status'] in [ 'R', 'Q' ]:
1690                jobs[ job_id ] = myAttrs
[524]1691
[691]1692                debug_msg( 10, printTime() + ' job %s state changed' %(job_id) )
[524]1693
[691]1694        for id, attrs in jobs.items():
1695            if id not in jobs_processed:
1696                # This one isn't there anymore
1697                #
1698                del jobs[ id ]
1699        self.jobs=jobs
[524]1700
1701
[355]1702class PbsDataGatherer( DataGatherer ):
[318]1703
[691]1704    """This is the DataGatherer for PBS and Torque"""
[318]1705
[691]1706    global PBSQuery, PBSError
[256]1707
[691]1708    def __init__( self ):
[354]1709
[691]1710        """Setup appropriate variables"""
[23]1711
[785]1712        self.jobs       = { }
1713        self.timeoffset = 0
1714        self.dp         = DataProcessor()
[354]1715
[691]1716        self.initPbsQuery()
[23]1717
[691]1718    def initPbsQuery( self ):
[91]1719
[785]1720        self.pq = None
[354]1721
[788]1722        try:
[354]1723
[788]1724            if( BATCH_SERVER ):
[91]1725
[788]1726                self.pq = PBSQuery( BATCH_SERVER )
1727            else:
1728                self.pq = PBSQuery()
1729
1730        except PBSError, details:
1731            print 'Cannot connect to pbs server'
1732            print details
1733            sys.exit( 1 )
1734
[691]1735        try:
[791]1736            # TODO: actually use new data structure
[691]1737            self.pq.old_data_structure()
[656]1738
[691]1739        except AttributeError:
[656]1740
[691]1741            # pbs_query is older
1742            #
1743            pass
[656]1744
[790]1745    def getNodeData( self ):
1746
1747        nodedict = { }
1748
1749        try:
1750            nodedict = self.pq.getnodes()
1751
1752        except PBSError, detail:
1753
1754            debug_msg( 10, "PBS server unavailable, skipping until next polling interval: " + str( detail ) )
1755
1756        return nodedict
1757
[691]1758    def getJobData( self ):
[354]1759
[691]1760        """Gather all data on current jobs in Torque"""
[26]1761
[785]1762        joblist            = {}
1763        self.cur_time      = 0
[349]1764
[691]1765        try:
1766            joblist        = self.pq.getjobs()
[785]1767            self.cur_time  = time.time()
[354]1768
[691]1769        except PBSError, detail:
[354]1770
[790]1771            debug_msg( 10, "PBS server unavailable, skipping until next polling interval: " + str( detail ) )
[691]1772            return None
[354]1773
[691]1774        jobs_processed    = [ ]
[26]1775
[691]1776        for name, attrs in joblist.items():
[785]1777            display_queue = 1
1778            job_id        = name.split( '.' )[0]
[26]1779
[785]1780            name          = self.getAttr( attrs, 'Job_Name' )
1781            queue         = self.getAttr( attrs, 'queue' )
[317]1782
[691]1783            if QUEUE:
1784                for q in QUEUE:
1785                    if q == queue:
1786                        display_queue = 1
1787                        break
1788                    else:
1789                        display_queue = 0
1790                        continue
1791            if display_queue == 0:
1792                continue
[317]1793
1794
[691]1795            owner            = self.getAttr( attrs, 'Job_Owner' ).split( '@' )[0]
[785]1796            requested_time   = self.getAttr( attrs, 'Resource_List.walltime' )
1797            requested_memory = self.getAttr( attrs, 'Resource_List.mem' )
[95]1798
[785]1799            mynoderequest    = self.getAttr( attrs, 'Resource_List.nodes' )
[95]1800
[785]1801            ppn = ''
[281]1802
[691]1803            if mynoderequest.find( ':' ) != -1 and mynoderequest.find( 'ppn' ) != -1:
[95]1804
[785]1805                mynoderequest_fields = mynoderequest.split( ':' )
[281]1806
[691]1807                for mynoderequest_field in mynoderequest_fields:
[281]1808
[691]1809                    if mynoderequest_field.find( 'ppn' ) != -1:
[281]1810
[785]1811                        ppn = mynoderequest_field.split( 'ppn=' )[1]
[281]1812
[785]1813            status = self.getAttr( attrs, 'job_state' )
[25]1814
[691]1815            if status in [ 'Q', 'R' ]:
[450]1816
[691]1817                jobs_processed.append( job_id )
[450]1818
[785]1819            queued_timestamp = self.getAttr( attrs, 'ctime' )
[243]1820
[691]1821            if status == 'R':
[133]1822
[785]1823                start_timestamp = self.getAttr( attrs, 'mtime' )
1824                nodes           = self.getAttr( attrs, 'exec_host' ).split( '+' )
[133]1825
[785]1826                nodeslist       = do_nodelist( nodes )
[354]1827
[691]1828                if DETECT_TIME_DIFFS:
[185]1829
[691]1830                    # If a job start if later than our current date,
1831                    # that must mean the Torque server's time is later
1832                    # than our local time.
1833               
1834                    if int( start_timestamp ) > int( int( self.cur_time ) + int( self.timeoffset ) ):
[185]1835
[785]1836                        self.timeoffset = int( int(start_timestamp) - int(self.cur_time) )
[185]1837
[691]1838            elif status == 'Q':
[95]1839
[691]1840                # 'mynodequest' can be a string in the following syntax according to the
1841                # Torque Administator's manual:
1842                #
1843                # {<node_count> | <hostname>}[:ppn=<ppn>][:<property>[:<property>]...][+ ...]
1844                # {<node_count> | <hostname>}[:ppn=<ppn>][:<property>[:<property>]...][+ ...]
1845                # etc
1846                #
[451]1847
[691]1848                #
1849                # For now we only count the amount of nodes request and ignore properties
1850                #
[451]1851
[785]1852                start_timestamp = ''
1853                count_mynodes   = 0
[354]1854
[691]1855                for node in mynoderequest.split( '+' ):
[67]1856
[691]1857                    # Just grab the {node_count|hostname} part and ignore properties
1858                    #
[785]1859                    nodepart     = node.split( ':' )[0]
[67]1860
[691]1861                    # Let's assume a node_count value
1862                    #
[785]1863                    numeric_node = 1
[451]1864
[691]1865                    # Chop the value up into characters
1866                    #
1867                    for letter in nodepart:
[67]1868
[691]1869                        # If this char is not a digit (0-9), this must be a hostname
1870                        #
1871                        if letter not in string.digits:
[133]1872
[785]1873                            numeric_node = 0
[133]1874
[691]1875                    # If this is a hostname, just count this as one (1) node
1876                    #
1877                    if not numeric_node:
[354]1878
[785]1879                        count_mynodes = count_mynodes + 1
[691]1880                    else:
[451]1881
[691]1882                        # If this a number, it must be the node_count
1883                        # and increase our count with it's value
1884                        #
1885                        try:
[785]1886                            count_mynodes = count_mynodes + int( nodepart )
[354]1887
[691]1888                        except ValueError, detail:
[354]1889
[691]1890                            # When we arrive here I must be bugged or very confused
1891                            # THIS SHOULD NOT HAPPEN!
1892                            #
1893                            debug_msg( 10, str( detail ) )
1894                            debug_msg( 10, "Encountered weird node in Resources_List?!" )
1895                            debug_msg( 10, 'nodepart = ' + str( nodepart ) )
1896                            debug_msg( 10, 'job = ' + str( name ) )
1897                            debug_msg( 10, 'attrs = ' + str( attrs ) )
1898                       
[785]1899                nodeslist       = str( count_mynodes )
[691]1900            else:
[785]1901                start_timestamp = ''
1902                nodeslist       = ''
[133]1903
[691]1904            myAttrs                = { }
[26]1905
[785]1906            myAttrs[ 'name' ]             = str( name )
1907            myAttrs[ 'queue' ]            = str( queue )
1908            myAttrs[ 'owner' ]            = str( owner )
1909            myAttrs[ 'requested_time' ]   = str( requested_time )
1910            myAttrs[ 'requested_memory' ] = str( requested_memory )
1911            myAttrs[ 'ppn' ]              = str( ppn )
1912            myAttrs[ 'status' ]           = str( status )
1913            myAttrs[ 'start_timestamp' ]  = str( start_timestamp )
1914            myAttrs[ 'queued_timestamp' ] = str( queued_timestamp )
1915            myAttrs[ 'reported' ]         = str( int( int( self.cur_time ) + int( self.timeoffset ) ) )
1916            myAttrs[ 'nodes' ]            = nodeslist
1917            myAttrs[ 'domain' ]           = fqdn_parts( socket.getfqdn() )[1]
[691]1918            myAttrs[ 'poll_interval' ]    = str( BATCH_POLL_INTERVAL )
[354]1919
[691]1920            if self.jobDataChanged( self.jobs, job_id, myAttrs ) and myAttrs['status'] in [ 'R', 'Q' ]:
[61]1921
[785]1922                self.jobs[ job_id ] = myAttrs
[26]1923
[691]1924        for id, attrs in self.jobs.items():
[76]1925
[691]1926            if id not in jobs_processed:
[76]1927
[691]1928                # This one isn't there anymore; toedeledoki!
1929                #
1930                del self.jobs[ id ]
[76]1931
[362]1932GMETRIC_DEFAULT_TYPE    = 'string'
1933GMETRIC_DEFAULT_HOST    = '127.0.0.1'
1934GMETRIC_DEFAULT_PORT    = '8649'
[700]1935GMETRIC_DEFAULT_UNITS   = ''
[362]1936
1937class Gmetric:
1938
[691]1939    global GMETRIC_DEFAULT_HOST, GMETRIC_DEFAULT_PORT
[362]1940
[700]1941    slope           = { 'zero' : 0, 'positive' : 1, 'negative' : 2, 'both' : 3, 'unspecified' : 4 }
1942    type            = ( '', 'string', 'uint16', 'int16', 'uint32', 'int32', 'float', 'double', 'timestamp' )
1943    protocol        = ( 'udp', 'multicast' )
[362]1944
[691]1945    def __init__( self, host=GMETRIC_DEFAULT_HOST, port=GMETRIC_DEFAULT_PORT ):
[700]1946               
[691]1947        global GMETRIC_DEFAULT_TYPE
[362]1948
[691]1949        self.prot       = self.checkHostProtocol( host )
[700]1950        self.data_msg   = xdrlib.Packer()
1951        self.meta_msg   = xdrlib.Packer()
[691]1952        self.socket     = socket.socket( socket.AF_INET, socket.SOCK_DGRAM )
[362]1953
[691]1954        if self.prot not in self.protocol:
[362]1955
[691]1956            raise ValueError( "Protocol must be one of: " + str( self.protocol ) )
[362]1957
[691]1958        if self.prot == 'multicast':
[362]1959
[691]1960            # Set multicast options
1961            #
1962            self.socket.setsockopt( socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 20 )
[362]1963
[691]1964        self.hostport   = ( host, int( port ) )
1965        self.slopestr   = 'both'
1966        self.tmax       = 60
[362]1967
[691]1968    def checkHostProtocol( self, ip ):
[362]1969
[691]1970        """Detect if a ip adress is a multicast address"""
[471]1971
[691]1972        MULTICAST_ADDRESS_MIN   = ( "224", "0", "0", "0" )
1973        MULTICAST_ADDRESS_MAX   = ( "239", "255", "255", "255" )
[362]1974
[700]1975        ip_fields               = ip.split( '.' )
[362]1976
[691]1977        if ip_fields >= MULTICAST_ADDRESS_MIN and ip_fields <= MULTICAST_ADDRESS_MAX:
[362]1978
[691]1979            return 'multicast'
1980        else:
1981            return 'udp'
[362]1982
[691]1983    def send( self, name, value, dmax, typestr = '', units = '' ):
[362]1984
[691]1985        if len( units ) == 0:
[700]1986            units       = GMETRIC_DEFAULT_UNITS
[471]1987
[691]1988        if len( typestr ) == 0:
[700]1989            typestr     = GMETRIC_DEFAULT_TYPE
[362]1990
[700]1991        (meta_msg, data_msg) = self.makexdr( name, value, typestr, units, self.slopestr, self.tmax, dmax )
[409]1992
[700]1993        meta_rt = self.socket.sendto( meta_msg, self.hostport )
1994        data_rt = self.socket.sendto( data_msg, self.hostport )
[362]1995
[700]1996        return ( meta_rt, data_rt )
[362]1997
[700]1998    def makexdr( self, name, value, typestr, unitstr, slopestr, tmax, dmax, group=None, spoof=None ):
1999
2000        hostname = "unset"
2001
[691]2002        if slopestr not in self.slope:
[362]2003
[691]2004            raise ValueError( "Slope must be one of: " + str( self.slope.keys() ) )
[362]2005
[691]2006        if typestr not in self.type:
[362]2007
[691]2008            raise ValueError( "Type must be one of: " + str( self.type ) )
[362]2009
[691]2010        if len( name ) == 0:
[362]2011
[691]2012            raise ValueError( "Name must be non-empty" )
[362]2013
[700]2014        self.meta_msg.reset()
2015        self.meta_msg.pack_int( 128 )
[362]2016
[700]2017        if not spoof:
2018            self.meta_msg.pack_string( hostname )
2019        else:
2020            self.meta_msg.pack_string( spoof )
[362]2021
[700]2022        self.meta_msg.pack_string( name )
2023
2024        if not spoof:
2025            self.meta_msg.pack_int( 0 )
2026        else:
2027            self.meta_msg.pack_int( 1 )
2028           
2029        self.meta_msg.pack_string( typestr )
2030        self.meta_msg.pack_string( name )
2031        self.meta_msg.pack_string( unitstr )
2032        self.meta_msg.pack_int( self.slope[ slopestr ] )
2033        self.meta_msg.pack_uint( int( tmax ) )
2034        self.meta_msg.pack_uint( int( dmax ) )
2035
2036        if not group:
2037            self.meta_msg.pack_int( 0 )
2038        else:
2039            self.meta_msg.pack_int( 1 )
2040            self.meta_msg.pack_string( "GROUP" )
2041            self.meta_msg.pack_string( group )
2042
2043        self.data_msg.reset()
2044        self.data_msg.pack_int( 128+5 )
2045
2046        if not spoof:
2047            self.data_msg.pack_string( hostname )
2048        else:
2049            self.data_msg.pack_string( spoof )
2050
2051        self.data_msg.pack_string( name )
2052
2053        if not spoof:
2054            self.data_msg.pack_int( 0 )
2055        else:
2056            self.data_msg.pack_int( 1 )
2057
2058        self.data_msg.pack_string( "%s" )
2059        self.data_msg.pack_string( str( value ) )
2060
2061        return ( self.meta_msg.get_buffer(), self.data_msg.get_buffer() )
2062
[26]2063def printTime( ):
[354]2064
[691]2065    """Print current time/date in human readable format for log/debug"""
[26]2066
[691]2067    return time.strftime("%a, %d %b %Y %H:%M:%S")
[26]2068
2069def debug_msg( level, msg ):
[354]2070
[691]2071    """Print msg if at or above current debug level"""
[26]2072
[691]2073    global DAEMONIZE, DEBUG_LEVEL, SYSLOG_LEVEL
[377]2074
[691]2075    if (not DAEMONIZE and DEBUG_LEVEL >= level):
2076        sys.stderr.write( msg + '\n' )
[26]2077
[691]2078    if (DAEMONIZE and USE_SYSLOG and SYSLOG_LEVEL >= level):
2079        syslog.syslog( msg )
[373]2080
[307]2081def write_pidfile():
2082
[691]2083    # Write pidfile if PIDFILE is set
2084    #
2085    if PIDFILE:
[307]2086
[691]2087        pid    = os.getpid()
[354]2088
[691]2089        pidfile    = open( PIDFILE, 'w' )
[354]2090
[691]2091        pidfile.write( str( pid ) )
2092        pidfile.close()
[307]2093
[23]2094def main():
[354]2095
[691]2096    """Application start"""
[23]2097
[837]2098    global PBSQuery, PBSError, lsfObject, pyslurm
[854]2099    global SYSLOG_FACILITY, USE_SYSLOG, BATCH_API, DAEMONIZE, BATCH_SERVER
[256]2100
[691]2101    if not processArgs( sys.argv[1:] ):
[354]2102
[691]2103        sys.exit( 1 )
[212]2104
[691]2105    # Load appropriate DataGatherer depending on which BATCH_API is set
2106    # and any required modules for the Gatherer
2107    #
2108    if BATCH_API == 'pbs':
[256]2109
[691]2110        try:
2111            from PBSQuery import PBSQuery, PBSError
[256]2112
[787]2113        except ImportError, details:
[256]2114
[854]2115            print "FATAL ERROR: BATCH_API set to 'pbs' but python module 'pbs_python' is not found or installed"
[787]2116            print details
[691]2117            sys.exit( 1 )
[256]2118
[691]2119        gather = PbsDataGatherer()
[256]2120
[691]2121    elif BATCH_API == 'sge':
[256]2122
[854]2123        if BATCH_SERVER != 'localhost':
2124
2125            # Print and log, but continue execution
2126            err_msg = "WARNING: BATCH_API 'sge' ignores BATCH_SERVER (can only be 'localhost')"
2127            print err_msg
2128            debug_msg( 0, err_msg )
2129
[691]2130        # Tested with SGE 6.0u11.
2131        #
2132        gather = SgeDataGatherer()
[368]2133
[691]2134    elif BATCH_API == 'lsf':
[368]2135
[854]2136        if BATCH_SERVER != 'localhost':
2137
2138            # Print and log, but continue execution
2139            err_msg = "WARNING: BATCH_API 'lsf' ignores BATCH_SERVER (can only be 'localhost')"
2140            print err_msg
2141            debug_msg( 0, err_msg )
2142
[691]2143        try:
2144            from lsfObject import lsfObject
2145        except:
[854]2146            print "FATAL ERROR: BATCH_API set to 'lsf' but python module 'lsfObject' is not found or installed"
2147            sys.exit( 1 )
[256]2148
[691]2149        gather = LsfDataGatherer()
[524]2150
[837]2151    elif BATCH_API == 'slurm':
2152
[854]2153        if BATCH_SERVER != 'localhost':
2154
2155            # Print and log, but continue execution
2156            err_msg = "WARNING: BATCH_API 'slurm' ignores BATCH_SERVER (can only be 'localhost')"
2157            print err_msg
2158            debug_msg( 0, err_msg )
2159
[837]2160        try:
2161            import pyslurm
2162        except:
2163            print "FATAL ERROR: BATCH_API set to 'slurm' but python module is not found or installed"
[854]2164            sys.exit( 1 )
[837]2165
2166        gather = SLURMDataGatherer()
2167
[691]2168    else:
[786]2169        print "FATAL ERROR: unknown BATCH_API '" + BATCH_API + "' is not supported"
[354]2170
[691]2171        sys.exit( 1 )
[256]2172
[691]2173    if( DAEMONIZE and USE_SYSLOG ):
[373]2174
[691]2175        syslog.openlog( 'jobmond', syslog.LOG_NOWAIT, SYSLOG_FACILITY )
[373]2176
[691]2177    if DAEMONIZE:
[354]2178
[691]2179        gather.daemon()
2180    else:
2181        gather.run()
[23]2182
[256]2183# wh00t? someone started me! :)
[65]2184#
[23]2185if __name__ == '__main__':
[691]2186    main()
Note: See TracBrowser for help on using the repository browser.