source: trunk/jobmond/jobmond.py @ 579

Last change on this file since 579 was 579, checked in by ramonb, 12 years ago

jobmond/jobmond.py:

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