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

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