source: trunk/jobmond/jobmond.py @ 374

Last change on this file since 374 was 373, checked in by bastiaans, 15 years ago

jobmond/jobmond.py:

  • added syslog support with backwards compatible config
  • Property svn:keywords set to Id
File size: 23.5 KB
Line 
1#!/usr/bin/env python
2#
3# This file is part of Jobmonarch
4#
5# Copyright (C) 2006-2007  Ramon Bastiaans
6#
7# Jobmonarch is free software; you can redistribute it and/or modify
8# it under the terms of the GNU General Public License as published by
9# the Free Software Foundation; either version 2 of the License, or
10# (at your option) any later version.
11#
12# Jobmonarch is distributed in the hope that it will be useful,
13# but WITHOUT ANY WARRANTY; without even the implied warranty of
14# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
15# GNU General Public License for more details.
16#
17# You should have received a copy of the GNU General Public License
18# along with this program; if not, write to the Free Software
19# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
20#
21# SVN $Id: jobmond.py 373 2007-06-13 13:38:52Z bastiaans $
22#
23
24import sys, getopt, ConfigParser
25import time, os, socket, string, re
26import xdrlib, socket, syslog
27import xml, xml.sax
28from xml.sax import saxutils, make_parser
29from xml.sax import make_parser
30from xml.sax.handler import feature_namespaces
31
32def usage():
33
34        print
35        print 'usage: jobmond [options]'
36        print 'options:'
37        print '      --config, -c      configuration file'
38        print '      --pidfile, -p     pid file'
39        print '      --help, -h        help'
40        print
41
42def processArgs( args ):
43
44        SHORT_L         = 'hc:'
45        LONG_L          = [ 'help', 'config=' ]
46
47        global PIDFILE
48        PIDFILE         = None
49
50        config_filename = '/etc/jobmond.conf'
51
52        try:
53
54                opts, args      = getopt.getopt( args, SHORT_L, LONG_L )
55
56        except getopt.GetoptError, detail:
57
58                print detail
59                usage()
60                sys.exit( 1 )
61
62        for opt, value in opts:
63
64                if opt in [ '--config', '-c' ]:
65               
66                        config_filename = value
67
68                if opt in [ '--pidfile', '-p' ]:
69
70                        PIDFILE         = value
71               
72                if opt in [ '--help', '-h' ]:
73 
74                        usage()
75                        sys.exit( 0 )
76
77        return loadConfig( config_filename )
78
79def loadConfig( filename ):
80
81        def getlist( cfg_string ):
82
83                my_list = [ ]
84
85                for item_txt in cfg_string.split( ',' ):
86
87                        sep_char = None
88
89                        item_txt = item_txt.strip()
90
91                        for s_char in [ "'", '"' ]:
92
93                                if item_txt.find( s_char ) != -1:
94
95                                        if item_txt.count( s_char ) != 2:
96
97                                                print 'Missing quote: %s' %item_txt
98                                                sys.exit( 1 )
99
100                                        else:
101
102                                                sep_char = s_char
103                                                break
104
105                        if sep_char:
106
107                                item_txt = item_txt.split( sep_char )[1]
108
109                        my_list.append( item_txt )
110
111                return my_list
112
113        cfg             = ConfigParser.ConfigParser()
114
115        cfg.read( filename )
116
117        global DEBUG_LEVEL, DAEMONIZE, BATCH_SERVER, BATCH_POLL_INTERVAL
118        global GMOND_CONF, DETECT_TIME_DIFFS, BATCH_HOST_TRANSLATE
119        global BATCH_API, QUEUE, GMETRIC_TARGET, USE_SYSLOG
120        global SYSLOG_LEVEL, SYSLOG_FACILITY
121
122        DEBUG_LEVEL     = cfg.getint( 'DEFAULT', 'DEBUG_LEVEL' )
123
124        DAEMONIZE       = cfg.getboolean( 'DEFAULT', 'DAEMONIZE' )
125
126        try:
127                USE_SYSLOG      = cfg.getboolean( 'DEFAULT', 'USE_SYSLOG' )
128
129        except ConfigParser.NoOptionError:
130
131                USE_SYSLOG      = True
132
133                debug_msg( 0, 'ERROR: no option USE_SYSLOG found: assuming yes' )
134
135        if USE_SYSLOG:
136
137                try:
138                        SYSLOG_LEVEL    = cfg.getint( 'DEFAULT', 'SYSLOG_LEVEL' )
139
140                except ConfigParser.NoOptionError:
141
142                        debug_msg( 0, 'ERROR: no option SYSLOG_LEVEL found: assuming level 0' )
143                        SYSLOG_LEVEL    = 0
144
145                try:
146
147                        SYSLOG_FACILITY = eval( 'syslog.LOG_' + cfg.get( 'DEFAULT', 'SYSLOG_FACILITY' ) )
148
149                except AttributeError, detail:
150
151                        SYSLOG_FACILITY = syslog.LOG_DAEMON
152
153                        debug_msg( 0, 'ERROR: no option SYSLOG_FACILITY found: assuming facility DAEMON' )
154
155        try:
156
157                BATCH_SERVER            = cfg.get( 'DEFAULT', 'BATCH_SERVER' )
158
159        except ConfigParser.NoOptionError:
160
161                # Backwards compatibility for old configs
162                #
163
164                BATCH_SERVER            = cfg.get( 'DEFAULT', 'TORQUE_SERVER' )
165                api_guess               = 'pbs'
166       
167        try:
168       
169                BATCH_POLL_INTERVAL     = cfg.getint( 'DEFAULT', 'BATCH_POLL_INTERVAL' )
170
171        except ConfigParser.NoOptionError:
172
173                # Backwards compatibility for old configs
174                #
175
176                BATCH_POLL_INTERVAL     = cfg.getint( 'DEFAULT', 'TORQUE_POLL_INTERVAL' )
177                api_guess               = 'pbs'
178       
179        try:
180
181                GMOND_CONF              = cfg.get( 'DEFAULT', 'GMOND_CONF' )
182
183        except ConfigParser.NoOptionError:
184
185                GMOND_CONF              = None
186
187        DETECT_TIME_DIFFS       = cfg.getboolean( 'DEFAULT', 'DETECT_TIME_DIFFS' )
188
189        BATCH_HOST_TRANSLATE    = getlist( cfg.get( 'DEFAULT', 'BATCH_HOST_TRANSLATE' ) )
190
191        try:
192
193                BATCH_API       = cfg.get( 'DEFAULT', 'BATCH_API' )
194
195        except ConfigParser.NoOptionError, detail:
196
197                if BATCH_SERVER and api_guess:
198
199                        BATCH_API       = api_guess
200                else:
201                        debug_msg( 0, "FATAL ERROR: BATCH_API not set and can't make guess" )
202                        sys.exit( 1 )
203
204        try:
205
206                QUEUE           = getlist( cfg.get( 'DEFAULT', 'QUEUE' ) )
207
208        except ConfigParser.NoOptionError, detail:
209
210                QUEUE           = None
211
212        try:
213
214                GMETRIC_TARGET  = cfg.get( 'DEFAULT', 'GMETRIC_TARGET' )
215
216        except ConfigParser.NoOptionError:
217
218                GMETRIC_TARGET  = None
219
220                if not GMOND_CONF:
221
222                        debug_msg( 0, "FATAL ERROR: GMETRIC_TARGET and GMOND_CONF both not set! Set at least one!" )
223                        sys.exit( 1 )
224                else:
225
226                        debug_msg( 0, "ERROR: GMETRIC_TARGET not set: internel Gmetric handling aborted. Failing back to DEPRECATED use of gmond.conf/gmetric binary. This will slow down jobmond significantly!" )
227
228        return True
229
230METRIC_MAX_VAL_LEN = 900
231
232class DataProcessor:
233
234        """Class for processing of data"""
235
236        binary = '/usr/bin/gmetric'
237
238        def __init__( self, binary=None ):
239
240                """Remember alternate binary location if supplied"""
241
242                if binary:
243                        self.binary = binary
244
245                # Timeout for XML
246                #
247                # From ganglia's documentation:
248                #
249                # 'A metric will be deleted DMAX seconds after it is received, and
250                # DMAX=0 means eternal life.'
251
252                self.dmax = str( int( int( BATCH_POLL_INTERVAL ) * 2 ) )
253
254                if GMOND_CONF:
255
256                        try:
257                                gmond_file = GMOND_CONF
258
259                        except NameError:
260                                gmond_file = '/etc/gmond.conf'
261
262                        if not os.path.exists( gmond_file ):
263                                debug_msg( 0, 'FATAL ERROR: ' + gmond_file + ' does not exist' )
264                                sys.exit( 1 )
265
266                        incompatible = self.checkGmetricVersion()
267
268                        if incompatible:
269
270                                debug_msg( 0, 'Gmetric version not compatible, please upgrade to at least 3.0.1' )
271                                sys.exit( 1 )
272
273        def checkGmetricVersion( self ):
274
275                """
276                Check version of gmetric is at least 3.0.1
277                for the syntax we use
278                """
279
280                global METRIC_MAX_VAL_LEN
281
282                incompatible    = 0
283
284                gfp             = os.popen( self.binary + ' --version' )
285                lines           = gfp.readlines()
286
287                gfp.close()
288
289                for line in lines:
290
291                        line = line.split( ' ' )
292
293                        if len( line ) == 2 and str( line ).find( 'gmetric' ) != -1:
294                       
295                                gmetric_version = line[1].split( '\n' )[0]
296
297                                version_major   = int( gmetric_version.split( '.' )[0] )
298                                version_minor   = int( gmetric_version.split( '.' )[1] )
299                                version_patch   = int( gmetric_version.split( '.' )[2] )
300
301                                incompatible    = 0
302
303                                if version_major < 3:
304
305                                        incompatible = 1
306                               
307                                elif version_major == 3:
308
309                                        if version_minor == 0:
310
311                                                if version_patch < 1:
312                                               
313                                                        incompatible = 1
314
315                                                if version_patch < 3:
316
317                                                        METRIC_MAX_VAL_LEN = 900
318
319                                                elif version_patch >= 3:
320
321                                                        METRIC_MAX_VAL_LEN = 1400
322
323                return incompatible
324
325        def multicastGmetric( self, metricname, metricval, valtype='string' ):
326
327                """Call gmetric binary and multicast"""
328
329                cmd = self.binary
330
331                if GMETRIC_TARGET:
332
333                        from gmetric import Gmetric
334
335                if GMETRIC_TARGET:
336
337                        GMETRIC_TARGET_HOST     = GMETRIC_TARGET.split( ':' )[0]
338                        GMETRIC_TARGET_PORT     = GMETRIC_TARGET.split( ':' )[1]
339
340                        metric_debug            = "[gmetric] name: %s - val: %s - dmax: %s" %( str( metricname ), str( metricval ), str( self.dmax ) )
341
342                        debug_msg( 10, printTime() + ' ' + metric_debug)
343
344                        gm = Gmetric( GMETRIC_TARGET_HOST, GMETRIC_TARGET_PORT )
345
346                        gm.send( str( metricname ), str( metricval ), str( self.dmax ) )
347
348                else:
349                        try:
350                                cmd = cmd + ' -c' + GMOND_CONF
351
352                        except NameError:
353
354                                debug_msg( 10, 'Assuming /etc/gmond.conf for gmetric cmd (ommitting)' )
355
356                        cmd = cmd + ' -n' + str( metricname )+ ' -v"' + str( metricval )+ '" -t' + str( valtype ) + ' -d' + str( self.dmax )
357
358                        debug_msg( 10, printTime() + ' ' + cmd )
359
360                        os.system( cmd )
361
362class DataGatherer:
363
364        """Skeleton class for batch system DataGatherer"""
365
366        def printJobs( self, jobs ):
367
368                """Print a jobinfo overview"""
369
370                for name, attrs in self.jobs.items():
371
372                        print 'job %s' %(name)
373
374                        for name, val in attrs.items():
375
376                                print '\t%s = %s' %( name, val )
377
378        def printJob( self, jobs, job_id ):
379
380                """Print job with job_id from jobs"""
381
382                print 'job %s' %(job_id)
383
384                for name, val in jobs[ job_id ].items():
385
386                        print '\t%s = %s' %( name, val )
387
388        def daemon( self ):
389
390                """Run as daemon forever"""
391
392                # Fork the first child
393                #
394                pid = os.fork()
395                if pid > 0:
396                        sys.exit(0)  # end parent
397
398                # creates a session and sets the process group ID
399                #
400                os.setsid()
401
402                # Fork the second child
403                #
404                pid = os.fork()
405                if pid > 0:
406                        sys.exit(0)  # end parent
407
408                write_pidfile()
409
410                # Go to the root directory and set the umask
411                #
412                os.chdir('/')
413                os.umask(0)
414
415                sys.stdin.close()
416                sys.stdout.close()
417                sys.stderr.close()
418
419                os.open('/dev/null', os.O_RDWR)
420                os.dup2(0, 1)
421                os.dup2(0, 2)
422
423                self.run()
424
425        def run( self ):
426
427                """Main thread"""
428
429                while ( 1 ):
430               
431                        self.getJobData()
432                        self.submitJobData()
433                        time.sleep( BATCH_POLL_INTERVAL )       
434
435class SgeQstatXMLParser(xml.sax.handler.ContentHandler):
436
437        """Babu Sundaram's experimental SGE qstat XML parser"""
438
439        def __init__(self, qstatinxml):
440
441                self.qstatfile = qstatinxml
442                self.attribs = {}
443                self.value = ''
444                self.jobID = ''
445                self.currentJobInfo = ''
446                self.job_list = []
447                self.EOFFlag = 0
448                self.jobinfoCount = 0
449
450
451        def startElement(self, name, attrs):
452
453                if name == 'job_list':
454                        self.currentJobInfo = 'Status=' + attrs.get('state', None) + ' '
455                elif name == 'job_info':
456                        self.job_list = []
457                        self.jobinfoCount += 1
458
459        def characters(self, ch):
460
461                self.value = self.value + ch
462
463        def endElement(self, name):
464
465                if len(self.value.strip()) > 0 :
466
467                        self.currentJobInfo += name + '=' + self.value.strip() + ' '         
468                elif name != 'job_list':
469
470                        self.currentJobInfo += name + '=Unknown '
471
472                if name == 'JB_job_number':
473
474                        self.jobID = self.value.strip()
475                        self.job_list.append(self.jobID)         
476
477                if name == 'job_list':
478
479                        if self.attribs.has_key(self.jobID) == False:
480                                self.attribs[self.jobID] = self.currentJobInfo
481                        elif self.attribs.has_key(self.jobID) and self.attribs[self.jobID] != self.currentJobInfo:
482                                self.attribs[self.jobID] = self.currentJobInfo
483                        self.currentJobInfo = ''
484                        self.jobID = ''
485
486                elif name == 'job_info' and self.jobinfoCount == 2:
487
488                        deljobs = []
489                        for id in self.attribs:
490                                try:
491                                        self.job_list.index(str(id))
492                                except ValueError:
493                                        deljobs.append(id)
494                        for i in deljobs:
495                                del self.attribs[i]
496                        deljobs = []
497                        self.jobinfoCount = 0
498
499                self.value = ''
500
501class SgeDataGatherer(DataGatherer):
502
503        jobs = { }
504        SGE_QSTAT_XML_FILE      = '/tmp/.jobmonarch.sge.qstat'
505
506        def __init__( self ):
507                """Setup appropriate variables"""
508
509                self.jobs = { }
510                self.timeoffset = 0
511                self.dp = DataProcessor()
512                self.initSgeJobInfo()
513
514        def initSgeJobInfo( self ):
515                """This is outside the scope of DRMAA; Get the current jobs in SGE"""
516                """This is a hack because we cant get info about jobs beyond"""
517                """those in the current DRMAA session"""
518
519                self.qstatparser = SgeQstatXMLParser( self.SGE_QSTAT_XML_FILE )
520
521                # Obtain the qstat information from SGE in XML format
522                # This would change to DRMAA-specific calls from 6.0u9
523
524        def getJobData(self):
525                """Gather all data on current jobs in SGE"""
526
527                # Get the information about the current jobs in the SGE queue
528                info = os.popen("qstat -ext -xml").readlines()
529                f = open(self.SGE_QSTAT_XML_FILE,'w')
530                for lines in info:
531                        f.write(lines)
532                f.close()
533
534                # Parse the input
535                f = open(self.qstatparser.qstatfile, 'r')
536                xml.sax.parse(f, self.qstatparser)
537                f.close()
538
539                self.cur_time = time.time()
540
541                return self.qstatparser.attribs
542
543        def submitJobData(self):
544                """Submit job info list"""
545
546                self.dp.multicastGmetric( 'MONARCH-HEARTBEAT', str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
547                # Now let's spread the knowledge
548                #
549                metric_increment = 0
550                for jobid, jobattrs in self.qstatparser.attribs.items():
551
552                        self.dp.multicastGmetric( 'MONARCH-JOB-' + jobid + '-' + str(metric_increment), jobattrs)
553
554class PbsDataGatherer( DataGatherer ):
555
556        """This is the DataGatherer for PBS and Torque"""
557
558        global PBSQuery
559
560        def __init__( self ):
561
562                """Setup appropriate variables"""
563
564                self.jobs       = { }
565                self.timeoffset = 0
566                self.dp         = DataProcessor()
567
568                self.initPbsQuery()
569
570        def initPbsQuery( self ):
571
572                self.pq         = None
573
574                if( BATCH_SERVER ):
575
576                        self.pq         = PBSQuery( BATCH_SERVER )
577                else:
578                        self.pq         = PBSQuery()
579
580        def getAttr( self, attrs, name ):
581
582                """Return certain attribute from dictionary, if exists"""
583
584                if attrs.has_key( name ):
585
586                        return attrs[ name ]
587                else:
588                        return ''
589
590        def jobDataChanged( self, jobs, job_id, attrs ):
591
592                """Check if job with attrs and job_id in jobs has changed"""
593
594                if jobs.has_key( job_id ):
595
596                        oldData = jobs[ job_id ]       
597                else:
598                        return 1
599
600                for name, val in attrs.items():
601
602                        if oldData.has_key( name ):
603
604                                if oldData[ name ] != attrs[ name ]:
605
606                                        return 1
607
608                        else:
609                                return 1
610
611                return 0
612
613        def getJobData( self ):
614
615                """Gather all data on current jobs in Torque"""
616
617                joblist         = {}
618                self.cur_time   = 0
619
620                try:
621                        joblist         = self.pq.getjobs()
622                        self.cur_time   = time.time()
623
624                except PBSError, detail:
625
626                        debug_msg( 10, "Caught PBS unavailable, skipping until next polling interval: " + str( detail ) )
627                        return None
628
629                jobs_processed  = [ ]
630
631                for name, attrs in joblist.items():
632
633                        job_id                  = name.split( '.' )[0]
634
635                        jobs_processed.append( job_id )
636
637                        name                    = self.getAttr( attrs, 'Job_Name' )
638                        queue                   = self.getAttr( attrs, 'queue' )
639
640                        if QUEUE:
641
642                                if QUEUE != queue:
643
644                                        continue
645
646                        owner                   = self.getAttr( attrs, 'Job_Owner' ).split( '@' )[0]
647                        requested_time          = self.getAttr( attrs, 'Resource_List.walltime' )
648                        requested_memory        = self.getAttr( attrs, 'Resource_List.mem' )
649
650                        mynoderequest           = self.getAttr( attrs, 'Resource_List.nodes' )
651
652                        ppn                     = ''
653
654                        if mynoderequest.find( ':' ) != -1 and mynoderequest.find( 'ppn' ) != -1:
655
656                                mynoderequest_fields    = mynoderequest.split( ':' )
657
658                                for mynoderequest_field in mynoderequest_fields:
659
660                                        if mynoderequest_field.find( 'ppn' ) != -1:
661
662                                                ppn     = mynoderequest_field.split( 'ppn=' )[1]
663
664                        status                  = self.getAttr( attrs, 'job_state' )
665
666                        queued_timestamp        = self.getAttr( attrs, 'ctime' )
667
668                        if status == 'R':
669
670                                start_timestamp         = self.getAttr( attrs, 'mtime' )
671                                nodes                   = self.getAttr( attrs, 'exec_host' ).split( '+' )
672
673                                nodeslist               = [ ]
674
675                                for node in nodes:
676
677                                        host            = node.split( '/' )[0]
678
679                                        if nodeslist.count( host ) == 0:
680
681                                                for translate_pattern in BATCH_HOST_TRANSLATE:
682
683                                                        if translate_pattern.find( '/' ) != -1:
684
685                                                                translate_orig  = translate_pattern.split( '/' )[1]
686                                                                translate_new   = translate_pattern.split( '/' )[2]
687
688                                                                host            = re.sub( translate_orig, translate_new, host )
689                               
690                                                if not host in nodeslist:
691                               
692                                                        nodeslist.append( host )
693
694                                if DETECT_TIME_DIFFS:
695
696                                        # If a job start if later than our current date,
697                                        # that must mean the Torque server's time is later
698                                        # than our local time.
699                               
700                                        if int( start_timestamp ) > int( int( self.cur_time ) + int( self.timeoffset ) ):
701
702                                                self.timeoffset = int( int(start_timestamp) - int(self.cur_time) )
703
704                        elif status == 'Q':
705
706                                start_timestamp         = ''
707                                count_mynodes           = 0
708                                numeric_node            = 1
709
710                                for node in mynoderequest.split( '+' ):
711
712                                        nodepart        = node.split( ':' )[0]
713
714                                        for letter in nodepart:
715
716                                                if letter not in string.digits:
717
718                                                        numeric_node    = 0
719
720                                        if not numeric_node:
721
722                                                count_mynodes   = count_mynodes + 1
723                                        else:
724                                                try:
725                                                        count_mynodes   = count_mynodes + int( nodepart )
726
727                                                except ValueError, detail:
728
729                                                        debug_msg( 10, str( detail ) )
730                                                        debug_msg( 10, "Encountered weird node in Resources_List?!" )
731                                                        debug_msg( 10, 'nodepart = ' + str( nodepart ) )
732                                                        debug_msg( 10, 'job = ' + str( name ) )
733                                                        debug_msg( 10, 'attrs = ' + str( attrs ) )
734                                               
735                                nodeslist       = str( count_mynodes )
736                        else:
737                                start_timestamp = ''
738                                nodeslist       = ''
739
740                        myAttrs                         = { }
741
742                        myAttrs[ 'name' ]                       = str( name )
743                        myAttrs[ 'queue' ]              = str( queue )
744                        myAttrs[ 'owner' ]              = str( owner )
745                        myAttrs[ 'requested_time' ]     = str( requested_time )
746                        myAttrs[ 'requested_memory' ]   = str( requested_memory )
747                        myAttrs[ 'ppn' ]                = str( ppn )
748                        myAttrs[ 'status' ]             = str( status )
749                        myAttrs[ 'start_timestamp' ]    = str( start_timestamp )
750                        myAttrs[ 'queued_timestamp' ]   = str( queued_timestamp )
751                        myAttrs[ 'reported' ]           = str( int( int( self.cur_time ) + int( self.timeoffset ) ) )
752                        myAttrs[ 'nodes' ]              = nodeslist
753                        myAttrs[ 'domain' ]             = string.join( socket.getfqdn().split( '.' )[1:], '.' )
754                        myAttrs[ 'poll_interval' ]      = str( BATCH_POLL_INTERVAL )
755
756                        if self.jobDataChanged( self.jobs, job_id, myAttrs ) and myAttrs['status'] in [ 'R', 'Q' ]:
757
758                                self.jobs[ job_id ]     = myAttrs
759
760                for id, attrs in self.jobs.items():
761
762                        if id not in jobs_processed:
763
764                                # This one isn't there anymore; toedeledoki!
765                                #
766                                del self.jobs[ id ]
767
768        def submitJobData( self ):
769
770                """Submit job info list"""
771
772                self.dp.multicastGmetric( 'MONARCH-HEARTBEAT', str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
773
774                # Now let's spread the knowledge
775                #
776                for jobid, jobattrs in self.jobs.items():
777
778                        gmetric_val             = self.compileGmetricVal( jobid, jobattrs )
779                        metric_increment        = 0
780
781                        for val in gmetric_val:
782
783                                self.dp.multicastGmetric( 'MONARCH-JOB-' + jobid + '-' + str(metric_increment), val )
784
785                                metric_increment        = metric_increment + 1
786
787        def compileGmetricVal( self, jobid, jobattrs ):
788
789                """Create a val string for gmetric of jobinfo"""
790
791                gval_lists      = [ ]
792                mystr           = None
793                val_list        = { }
794
795                for val_name, val_value in jobattrs.items():
796
797                        val_list_names_len      = len( string.join( val_list.keys() ) ) + len(val_list.keys())
798                        val_list_vals_len       = len( string.join( val_list.values() ) ) + len(val_list.values())
799
800                        if val_name == 'nodes' and jobattrs['status'] == 'R':
801
802                                node_str = None
803
804                                for node in val_value:
805
806                                        if node_str:
807
808                                                node_str = node_str + ';' + node
809                                        else:
810                                                node_str = node
811
812                                        if (val_list_names_len + len(val_name) ) + (val_list_vals_len + len(node_str) ) > METRIC_MAX_VAL_LEN:
813
814                                                val_list[ val_name ]    = node_str
815
816                                                gval_lists.append( val_list )
817
818                                                val_list                = { }
819                                                node_str                = None
820
821                                val_list[ val_name ]    = node_str
822
823                                gval_lists.append( val_list )
824
825                                val_list                = { }
826
827                        elif val_value != '':
828
829                                if (val_list_names_len + len(val_name) ) + (val_list_vals_len + len(str(val_value)) ) > METRIC_MAX_VAL_LEN:
830
831                                        gval_lists.append( val_list )
832
833                                        val_list                = { }
834
835                                val_list[ val_name ]    = val_value
836
837                if len( val_list ) > 0:
838
839                        gval_lists.append( val_list )
840
841                str_list        = [ ]
842
843                for val_list in gval_lists:
844
845                        my_val_str      = None
846
847                        for val_name, val_value in val_list.items():
848
849                                if my_val_str:
850
851                                        my_val_str = my_val_str + ' ' + val_name + '=' + val_value
852                                else:
853                                        my_val_str = val_name + '=' + val_value
854
855                        str_list.append( my_val_str )
856
857                return str_list
858
859#
860# Gmetric by Nick Galbreath - nickg(a.t)modp(d.o.t)com
861# Version 1.0 - 21-April2-2007
862# http://code.google.com/p/embeddedgmetric/
863#
864# Modified by: Ramon Bastiaans
865# For the Job Monarch Project, see: https://subtrac.sara.nl/oss/jobmonarch/
866#
867# added: DEFAULT_TYPE for Gmetric's
868# added: checkHostProtocol to determine if target is multicast or not
869# changed: allow default for Gmetric constructor
870# changed: allow defaults for all send() values except dmax
871#
872
873GMETRIC_DEFAULT_TYPE    = 'string'
874GMETRIC_DEFAULT_HOST    = '127.0.0.1'
875GMETRIC_DEFAULT_PORT    = '8649'
876
877class Gmetric:
878
879        global GMETRIC_DEFAULT_HOST, GMETRIC_DEFAULT_PORT
880
881        slope           = { 'zero' : 0, 'positive' : 1, 'negative' : 2, 'both' : 3, 'unspecified' : 4 }
882        type            = ( '', 'string', 'uint16', 'int16', 'uint32', 'int32', 'float', 'double', 'timestamp' )
883        protocol        = ( 'udp', 'multicast' )
884
885        def __init__( self, host=GMETRIC_DEFAULT_HOST, port=GMETRIC_DEFAULT_PORT ):
886               
887                global GMETRIC_DEFAULT_TYPE
888
889                self.prot       = self.checkHostProtocol( host )
890                self.msg        = xdrlib.Packer()
891                self.socket     = socket.socket( socket.AF_INET, socket.SOCK_DGRAM )
892
893                if self.prot not in self.protocol:
894
895                        raise ValueError( "Protocol must be one of: " + str( self.protocol ) )
896
897                if self.prot == 'multicast':
898
899                        self.socket.setsockopt( socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 20 )
900
901                self.hostport   = ( host, int( port ) )
902                self.type       = GMETRIC_DEFAULT_TYPE
903                self.unitstr    = ''
904                self.slopestr   = 'both'
905                self.tmax       = 60
906
907        def checkHostProtocol( self, ip ):
908
909                MULTICAST_ADDRESS_MIN   = ( "224", "0", "0", "0" )
910                MULTICAST_ADDRESS_MAX   = ( "239", "255", "255", "255" )
911
912                ip_fields               = ip.split( '.' )
913
914                if ip_fields >= MULTICAST_ADDRESS_MIN and ip_fields <= MULTICAST_ADDRESS_MAX:
915
916                        return 'multicast'
917                else:
918                        return 'udp'
919
920        def send( self, name, value, dmax ):
921
922                msg             = self.makexdr( name, value, self.type, self.unitstr, self.slopestr, self.tmax, dmax )
923
924                return self.socket.sendto( msg, self.hostport )
925
926        def makexdr( self, name, value, typestr, unitstr, slopestr, tmax, dmax ):
927
928                if slopestr not in self.slope:
929
930                        raise ValueError( "Slope must be one of: " + str( self.slope.keys() ) )
931
932                if typestr not in self.type:
933
934                        raise ValueError( "Type must be one of: " + str( self.type ) )
935
936                if len( name ) == 0:
937
938                        raise ValueError( "Name must be non-empty" )
939
940                self.msg.reset()
941                self.msg.pack_int( 0 )
942                self.msg.pack_string( typestr )
943                self.msg.pack_string( name )
944                self.msg.pack_string( str( value ) )
945                self.msg.pack_string( unitstr )
946                self.msg.pack_int( self.slope[ slopestr ] )
947                self.msg.pack_uint( int( tmax ) )
948                self.msg.pack_uint( int( dmax ) )
949
950                return self.msg.get_buffer()
951
952def printTime( ):
953
954        """Print current time/date in human readable format for log/debug"""
955
956        return time.strftime("%a, %d %b %Y %H:%M:%S")
957
958def debug_msg( level, msg ):
959
960        """Print msg if at or above current debug level"""
961
962        if (not DAEMONIZE and DEBUG_LEVEL >= level):
963                sys.stderr.write( msg + '\n' )
964
965        if (DAEMONIZE and USE_SYSLOG and SYSLOG_LEVEL >= level):
966                syslog.syslog( msg )
967
968def write_pidfile():
969
970        # Write pidfile if PIDFILE exists
971        if PIDFILE:
972
973                pid     = os.getpid()
974
975                pidfile = open(PIDFILE, 'w')
976
977                pidfile.write( str( pid ) )
978                pidfile.close()
979
980def main():
981
982        """Application start"""
983
984        global PBSQuery, PBSError
985        global SYSLOG_FACILITY, USE_SYSLOG, BATCH_API, DAEMONIZE
986
987        if not processArgs( sys.argv[1:] ):
988
989                sys.exit( 1 )
990
991        if BATCH_API == 'pbs':
992
993                try:
994                        from PBSQuery import PBSQuery, PBSError
995
996                except ImportError:
997
998                        debug_msg( 0, "FATAL ERROR: BATCH_API set to 'pbs' but python module 'pbs_python' is not installed" )
999                        sys.exit( 1 )
1000
1001                gather = PbsDataGatherer()
1002
1003        elif BATCH_API == 'sge':
1004
1005                debug_msg( 0, "FATAL ERROR: BATCH_API 'sge' implementation is currently broken, check future releases" )
1006
1007                sys.exit( 1 )
1008
1009                gather = SgeDataGatherer()
1010
1011        else:
1012                debug_msg( 0, "FATAL ERROR: unknown BATCH_API '" + BATCH_API + "' is not supported" )
1013
1014                sys.exit( 1 )
1015
1016        if( DAEMONIZE and USE_SYSLOG ):
1017
1018                syslog.openlog( 'jobmond', syslog.LOG_NOWAIT, SYSLOG_FACILITY )
1019
1020
1021        if DAEMONIZE:
1022
1023                gather.daemon()
1024        else:
1025                gather.run()
1026
1027# wh00t? someone started me! :)
1028#
1029if __name__ == '__main__':
1030        main()
Note: See TracBrowser for help on using the repository browser.