source: trunk/jobmond/jobmond.py @ 363

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