source: trunk/jobmond/jobmond.py @ 353

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

jobmond/jobmond.conf:

  • added GMETRIC_TARGET

jobmond/gmetric.py:

  • checkin of native python gmetric support, thanks to Nick Galbreath

jobmond/jobmond.py:

  • added handling of GMETRIC_TARGET
  • GMOND_CONF is now deprecated (GMETRIC_TARGET should be used) but we are still backwards compatible
  • Property svn:keywords set to Id
File size: 19.7 KB
Line 
1#!/usr/bin/env python
2#
3# This file is part of Jobmonarch
4#
5# Copyright (C) 2006  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 353 2007-05-03 10:13:52Z bastiaans $
22#
23
24import sys, getopt, ConfigParser
25
26import xml, xml.sax
27from xml.sax import saxutils, make_parser
28from xml.sax import make_parser
29from xml.sax.handler import feature_namespaces
30
31def usage():
32
33        print
34        print 'usage: jobmond [options]'
35        print 'options:'
36        print '      --config, -c      configuration file'
37        print '      --pidfile, -p     pid file'
38        print '      --help, -h        help'
39        print
40
41
42def processArgs( args ):
43
44        SHORT_L = 'c:'
45        LONG_L = 'config='
46
47        global PIDFILE
48        PIDFILE = None
49        config_filename = '/etc/jobmond.conf'
50
51        try:
52
53                opts, args = getopt.getopt( args, SHORT_L, LONG_L )
54
55        except getopt.GetoptError, detail:
56
57                print detail
58                usage()
59                sys.exit(1)
60
61        for opt, value in opts:
62
63                if opt in [ '--config', '-c' ]:
64               
65                        config_filename = value
66
67                if opt in [ '--pidfile', '-p' ]:
68
69                        PIDFILE = value
70               
71                if opt in [ '--help', '-h' ]:
72 
73                        usage()
74                        sys.exit(1)
75
76        return loadConfig( config_filename )
77
78def loadConfig( filename ):
79
80        def getlist( cfg_string ):
81
82                my_list = [ ]
83
84                for item_txt in cfg_string.split( ',' ):
85
86                        sep_char = None
87
88                        item_txt = item_txt.strip()
89
90                        for s_char in [ "'", '"' ]:
91
92                                if item_txt.find( s_char ) != -1:
93
94                                        if item_txt.count( s_char ) != 2:
95
96                                                print 'Missing quote: %s' %item_txt
97                                                sys.exit( 1 )
98
99                                        else:
100
101                                                sep_char = s_char
102                                                break
103
104                        if sep_char:
105
106                                item_txt = item_txt.split( sep_char )[1]
107
108                        my_list.append( item_txt )
109
110                return my_list
111
112        cfg = ConfigParser.ConfigParser()
113
114        cfg.read( filename )
115
116        global DEBUG_LEVEL, DAEMONIZE, BATCH_SERVER, BATCH_POLL_INTERVAL, GMOND_CONF, DETECT_TIME_DIFFS, BATCH_HOST_TRANSLATE, BATCH_API, QUEUE, GMETRIC_TARGET
117
118        DEBUG_LEVEL = cfg.getint( 'DEFAULT', 'DEBUG_LEVEL' )
119
120        DAEMONIZE = cfg.getboolean( 'DEFAULT', 'DAEMONIZE' )
121
122        try:
123
124                BATCH_SERVER = cfg.get( 'DEFAULT', 'BATCH_SERVER' )
125
126        except ConfigParser.NoOptionError:
127
128                # Backwards compatibility for old configs
129                #
130
131                BATCH_SERVER = cfg.get( 'DEFAULT', 'TORQUE_SERVER' )
132                api_guess = 'pbs'
133       
134        try:
135       
136                BATCH_POLL_INTERVAL = cfg.getint( 'DEFAULT', 'BATCH_POLL_INTERVAL' )
137
138        except ConfigParser.NoOptionError:
139
140                # Backwards compatibility for old configs
141                #
142
143                BATCH_POLL_INTERVAL = cfg.getint( 'DEFAULT', 'TORQUE_POLL_INTERVAL' )
144                api_guess = 'pbs'
145       
146        try:
147
148                GMOND_CONF = cfg.get( 'DEFAULT', 'GMOND_CONF' )
149
150        except ConfigParser.NoOptionError:
151
152                GMOND_CONF = None
153
154        DETECT_TIME_DIFFS = cfg.getboolean( 'DEFAULT', 'DETECT_TIME_DIFFS' )
155
156        BATCH_HOST_TRANSLATE = getlist( cfg.get( 'DEFAULT', 'BATCH_HOST_TRANSLATE' ) )
157
158        try:
159
160                BATCH_API = cfg.get( 'DEFAULT', 'BATCH_API' )
161
162        except ConfigParser.NoOptionError, detail:
163
164                if BATCH_SERVER and api_guess:
165                        BATCH_API = api_guess
166                else:
167                        debug_msg( 0, "fatal error: BATCH_API not set and can't make guess" )
168                        sys.exit( 1 )
169
170        try:
171
172                QUEUE = getlist( cfg.get( 'DEFAULT', 'QUEUE' ) )
173
174        except ConfigParser.NoOptionError, detail:
175
176                QUEUE = None
177
178        try:
179
180                GMETRIC_TARGET = cfg.get( 'DEFAULT', 'GMETRIC_TARGET' )
181
182        except ConfigParser.NoOptionError:
183
184                GMETRIC_TARGET = None
185
186                if not GMOND_CONF:
187
188                        debug_msg( 0, "fatal error: GMETRIC_TARGET or GMOND_CONF both not set!" )
189                        sys.exit( 1 )
190                else:
191
192                        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!" )
193
194        return True
195
196
197import time, os, socket, string, re
198
199METRIC_MAX_VAL_LEN = 900
200
201class DataProcessor:
202        """Class for processing of data"""
203
204        binary = '/usr/bin/gmetric'
205
206        def __init__( self, binary=None ):
207                """Remember alternate binary location if supplied"""
208
209                if binary:
210                        self.binary = binary
211
212                # Timeout for XML
213                #
214                # From ganglia's documentation:
215                #
216                # 'A metric will be deleted DMAX seconds after it is received, and
217                # DMAX=0 means eternal life.'
218
219                self.dmax = str( int( int( BATCH_POLL_INTERVAL ) * 2 ) )
220
221                if GMOND_CONF:
222                        try:
223                                gmond_file = GMOND_CONF
224
225                        except NameError:
226                                gmond_file = '/etc/gmond.conf'
227
228                        if not os.path.exists( gmond_file ):
229                                debug_msg( 0, 'fatal error: ' + gmond_file + ' does not exist' )
230                                sys.exit( 1 )
231
232                        incompatible = self.checkGmetricVersion()
233
234                        if incompatible:
235                                debug_msg( 0, 'Gmetric version not compatible, pls upgrade to at least 3.0.1' )
236                                sys.exit( 1 )
237
238        def checkGmetricVersion( self ):
239                """
240                Check version of gmetric is at least 3.0.1
241                for the syntax we use
242                """
243
244                global METRIC_MAX_VAL_LEN
245
246                incompatible    = 0
247
248                for line in os.popen( self.binary + ' --version' ).readlines():
249
250                        line = line.split( ' ' )
251
252                        if len( line ) == 2 and str(line).find( 'gmetric' ) != -1:
253                       
254                                gmetric_version = line[1].split( '\n' )[0]
255
256                                version_major = int( gmetric_version.split( '.' )[0] )
257                                version_minor = int( gmetric_version.split( '.' )[1] )
258                                version_patch = int( gmetric_version.split( '.' )[2] )
259
260                                incompatible = 0
261
262                                if version_major < 3:
263
264                                        incompatible = 1
265                               
266                                elif version_major == 3:
267
268                                        if version_minor == 0:
269
270                                                if version_patch < 1:
271                                               
272                                                        incompatible = 1
273
274                                                if version_patch < 3:
275
276                                                        METRIC_MAX_VAL_LEN = 900
277
278                                                elif version_patch >= 3:
279
280                                                        METRIC_MAX_VAL_LEN = 1400
281
282                return incompatible
283
284        def multicastGmetric( self, metricname, metricval, valtype='string' ):
285                """Call gmetric binary and multicast"""
286
287                cmd = self.binary
288
289                if GMETRIC_TARGET:
290
291                        from gmetric import Gmetric
292
293                if GMETRIC_TARGET:
294
295                        GMETRIC_TARGET_HOST     = GMETRIC_TARGET.split( ':' )[0]
296                        GMETRIC_TARGET_PORT     = GMETRIC_TARGET.split( ':' )[1]
297
298                        metric_debug            = "[gmetric] name: %s - val: %s - dmax: %s" %( str( metricname ), str( metricval ), str( self.dmax ) )
299
300                        debug_msg( 10, printTime() + ' ' + metric_debug)
301
302                        gm = Gmetric( GMETRIC_TARGET_HOST, GMETRIC_TARGET_PORT )
303
304                        gm.send( str( metricname ), str( metricval ), str( self.dmax ) )
305
306                else:
307                        try:
308                                cmd = cmd + ' -c' + GMOND_CONF
309
310                        except NameError:
311
312                                debug_msg( 10, 'Assuming /etc/gmond.conf for gmetric cmd (ommitting)' )
313
314                        cmd = cmd + ' -n' + str( metricname )+ ' -v"' + str( metricval )+ '" -t' + str( valtype ) + ' -d' + str( self.dmax )
315
316                        debug_msg( 10, printTime() + ' ' + cmd )
317
318                        os.system( cmd )
319
320class DataGatherer:
321
322        """Skeleton class for batch system DataGatherer"""
323
324        def printJobs( self, jobs ):
325                """Print a jobinfo overview"""
326
327                for name, attrs in self.jobs.items():
328
329                        print 'job %s' %(name)
330
331                        for name, val in attrs.items():
332
333                                print '\t%s = %s' %( name, val )
334
335        def printJob( self, jobs, job_id ):
336                """Print job with job_id from jobs"""
337
338                print 'job %s' %(job_id)
339
340                for name, val in jobs[ job_id ].items():
341
342                        print '\t%s = %s' %( name, val )
343
344        def daemon( self ):
345                """Run as daemon forever"""
346
347                # Fork the first child
348                #
349                pid = os.fork()
350                if pid > 0:
351                        sys.exit(0)  # end parent
352
353                # creates a session and sets the process group ID
354                #
355                os.setsid()
356
357                # Fork the second child
358                #
359                pid = os.fork()
360                if pid > 0:
361                        sys.exit(0)  # end parent
362
363                write_pidfile()
364
365                # Go to the root directory and set the umask
366                #
367                os.chdir('/')
368                os.umask(0)
369
370                sys.stdin.close()
371                sys.stdout.close()
372                sys.stderr.close()
373
374                os.open('/dev/null', os.O_RDWR)
375                os.dup2(0, 1)
376                os.dup2(0, 2)
377
378                self.run()
379
380        def run( self ):
381                """Main thread"""
382
383                while ( 1 ):
384               
385                        self.getJobData()
386                        self.submitJobData()
387                        time.sleep( BATCH_POLL_INTERVAL )       
388
389class SgeQstatXMLParser(xml.sax.handler.ContentHandler):
390
391        """Babu Sundaram's experimental SGE qstat XML parser"""
392
393        def __init__(self, qstatinxml):
394
395                self.qstatfile = qstatinxml
396                self.attribs = {}
397                self.value = ''
398                self.jobID = ''
399                self.currentJobInfo = ''
400                self.job_list = []
401                self.EOFFlag = 0
402                self.jobinfoCount = 0
403
404
405        def startElement(self, name, attrs):
406
407                if name == 'job_list':
408                        self.currentJobInfo = 'Status=' + attrs.get('state', None) + ' '
409                elif name == 'job_info':
410                        self.job_list = []
411                        self.jobinfoCount += 1
412
413        def characters(self, ch):
414
415                self.value = self.value + ch
416
417        def endElement(self, name):
418
419                if len(self.value.strip()) > 0 :
420
421                        self.currentJobInfo += name + '=' + self.value.strip() + ' '         
422                elif name != 'job_list':
423
424                        self.currentJobInfo += name + '=Unknown '
425
426                if name == 'JB_job_number':
427
428                        self.jobID = self.value.strip()
429                        self.job_list.append(self.jobID)         
430
431                if name == 'job_list':
432
433                        if self.attribs.has_key(self.jobID) == False:
434                                self.attribs[self.jobID] = self.currentJobInfo
435                        elif self.attribs.has_key(self.jobID) and self.attribs[self.jobID] != self.currentJobInfo:
436                                self.attribs[self.jobID] = self.currentJobInfo
437                        self.currentJobInfo = ''
438                        self.jobID = ''
439
440                elif name == 'job_info' and self.jobinfoCount == 2:
441
442                        deljobs = []
443                        for id in self.attribs:
444                                try:
445                                        self.job_list.index(str(id))
446                                except ValueError:
447                                        deljobs.append(id)
448                        for i in deljobs:
449                                del self.attribs[i]
450                        deljobs = []
451                        self.jobinfoCount = 0
452
453                self.value = ''
454
455class SgeDataGatherer(DataGatherer):
456
457        jobs = { }
458        SGE_QSTAT_XML_FILE      = '/tmp/.jobmonarch.sge.qstat'
459
460        def __init__( self ):
461                """Setup appropriate variables"""
462
463                self.jobs = { }
464                self.timeoffset = 0
465                self.dp = DataProcessor()
466                self.initSgeJobInfo()
467
468        def initSgeJobInfo( self ):
469                """This is outside the scope of DRMAA; Get the current jobs in SGE"""
470                """This is a hack because we cant get info about jobs beyond"""
471                """those in the current DRMAA session"""
472
473                self.qstatparser = SgeQstatXMLParser( self.SGE_QSTAT_XML_FILE )
474
475                # Obtain the qstat information from SGE in XML format
476                # This would change to DRMAA-specific calls from 6.0u9
477
478        def getJobData(self):
479                """Gather all data on current jobs in SGE"""
480
481                # Get the information about the current jobs in the SGE queue
482                info = os.popen("qstat -ext -xml").readlines()
483                f = open(self.SGE_QSTAT_XML_FILE,'w')
484                for lines in info:
485                        f.write(lines)
486                f.close()
487
488                # Parse the input
489                f = open(self.qstatparser.qstatfile, 'r')
490                xml.sax.parse(f, self.qstatparser)
491                f.close()
492
493                self.cur_time = time.time()
494
495                return self.qstatparser.attribs
496
497        def submitJobData(self):
498                """Submit job info list"""
499
500                self.dp.multicastGmetric( 'MONARCH-HEARTBEAT', str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
501                # Now let's spread the knowledge
502                #
503                metric_increment = 0
504                for jobid, jobattrs in self.qstatparser.attribs.items():
505
506                        self.dp.multicastGmetric( 'MONARCH-JOB-' + jobid + '-' + str(metric_increment), jobattrs)
507
508class PbsDataGatherer(DataGatherer):
509
510        """This is the DataGatherer for PBS and Torque"""
511
512        global PBSQuery
513
514        def __init__( self ):
515                """Setup appropriate variables"""
516
517                self.jobs = { }
518                self.timeoffset = 0
519                self.dp = DataProcessor()
520                self.initPbsQuery()
521
522        def initPbsQuery( self ):
523
524                self.pq = None
525                if( BATCH_SERVER ):
526                        self.pq = PBSQuery( BATCH_SERVER )
527                else:
528                        self.pq = PBSQuery()
529
530        def getAttr( self, attrs, name ):
531                """Return certain attribute from dictionary, if exists"""
532
533                if attrs.has_key( name ):
534                        return attrs[name]
535                else:
536                        return ''
537
538        def jobDataChanged( self, jobs, job_id, attrs ):
539                """Check if job with attrs and job_id in jobs has changed"""
540
541                if jobs.has_key( job_id ):
542                        oldData = jobs[ job_id ]       
543                else:
544                        return 1
545
546                for name, val in attrs.items():
547
548                        if oldData.has_key( name ):
549
550                                if oldData[ name ] != attrs[ name ]:
551
552                                        return 1
553
554                        else:
555                                return 1
556
557                return 0
558
559        def getJobData( self ):
560                """Gather all data on current jobs in Torque"""
561
562                #self.initPbsQuery()
563       
564                #print self.pq.getnodes()
565       
566                joblist = {}
567
568                while len(joblist) == 0:
569                        try:
570                                joblist = self.pq.getjobs()
571                        except PBSError, detail:
572                                debug_msg( 10, "Caught PBS unavaible, skipping until next polling interval: " + str( detail ) )
573                                return None
574
575                self.cur_time = time.time()
576
577                jobs_processed = [ ]
578
579                #self.printJobs( joblist )
580
581                for name, attrs in joblist.items():
582
583                        job_id = name.split( '.' )[0]
584
585                        jobs_processed.append( job_id )
586
587                        name = self.getAttr( attrs, 'Job_Name' )
588                        queue = self.getAttr( attrs, 'queue' )
589
590                        if QUEUE:
591
592                                if QUEUE != queue:
593
594                                        continue
595
596                        owner = self.getAttr( attrs, 'Job_Owner' ).split( '@' )[0]
597                        requested_time = self.getAttr( attrs, 'Resource_List.walltime' )
598                        requested_memory = self.getAttr( attrs, 'Resource_List.mem' )
599
600                        mynoderequest = self.getAttr( attrs, 'Resource_List.nodes' )
601
602                        ppn = ''
603
604                        if mynoderequest.find( ':' ) != -1 and mynoderequest.find( 'ppn' ) != -1:
605
606                                mynoderequest_fields = mynoderequest.split( ':' )
607
608                                for mynoderequest_field in mynoderequest_fields:
609
610                                        if mynoderequest_field.find( 'ppn' ) != -1:
611
612                                                ppn = mynoderequest_field.split( 'ppn=' )[1]
613
614                        status = self.getAttr( attrs, 'job_state' )
615
616                        queued_timestamp = self.getAttr( attrs, 'ctime' )
617
618                        if status == 'R':
619                                start_timestamp = self.getAttr( attrs, 'mtime' )
620                                nodes = self.getAttr( attrs, 'exec_host' ).split( '+' )
621
622                                nodeslist = [ ]
623
624                                for node in nodes:
625                                        host = node.split( '/' )[0]
626
627                                        if nodeslist.count( host ) == 0:
628
629                                                for translate_pattern in BATCH_HOST_TRANSLATE:
630
631                                                        if translate_pattern.find( '/' ) != -1:
632
633                                                                translate_orig = translate_pattern.split( '/' )[1]
634                                                                translate_new = translate_pattern.split( '/' )[2]
635
636                                                                host = re.sub( translate_orig, translate_new, host )
637                               
638                                                if not host in nodeslist:
639                               
640                                                        nodeslist.append( host )
641
642                                if DETECT_TIME_DIFFS:
643
644                                        # If a job start if later than our current date,
645                                        # that must mean the Torque server's time is later
646                                        # than our local time.
647                               
648                                        if int(start_timestamp) > int( int(self.cur_time) + int(self.timeoffset) ):
649
650                                                self.timeoffset = int( int(start_timestamp) - int(self.cur_time) )
651
652                        elif status == 'Q':
653                                start_timestamp = ''
654                                count_mynodes = 0
655                                numeric_node = 1
656
657                                for node in mynoderequest.split( '+' ):
658
659                                        nodepart = node.split( ':' )[0]
660
661                                        for letter in nodepart:
662
663                                                if letter not in string.digits:
664
665                                                        numeric_node = 0
666
667                                        if not numeric_node:
668                                                count_mynodes = count_mynodes + 1
669                                        else:
670                                                try:
671                                                        count_mynodes = count_mynodes + int( nodepart )
672                                                except ValueError, detail:
673                                                        debug_msg( 10, str( detail ) )
674                                                        debug_msg( 10, "Encountered weird node in Resources_List?!" )
675                                                        debug_msg( 10, 'nodepart = ' + str( nodepart ) )
676                                                        debug_msg( 10, 'job = ' + str( name ) )
677                                                        debug_msg( 10, 'attrs = ' + str( attrs ) )
678                                               
679                                nodeslist = str( count_mynodes )
680                        else:
681                                start_timestamp = ''
682                                nodeslist = ''
683
684                        myAttrs = { }
685                        myAttrs['name'] = str( name )
686                        myAttrs['queue'] = str( queue )
687                        myAttrs['owner'] = str( owner )
688                        myAttrs['requested_time'] = str( requested_time )
689                        myAttrs['requested_memory'] = str( requested_memory )
690                        myAttrs['ppn'] = str( ppn )
691                        myAttrs['status'] = str( status )
692                        myAttrs['start_timestamp'] = str( start_timestamp )
693                        myAttrs['queued_timestamp'] = str( queued_timestamp )
694                        myAttrs['reported'] = str( int( int( self.cur_time ) + int( self.timeoffset ) ) )
695                        myAttrs['nodes'] = nodeslist
696                        myAttrs['domain'] = string.join( socket.getfqdn().split( '.' )[1:], '.' )
697                        myAttrs['poll_interval'] = str( BATCH_POLL_INTERVAL )
698
699                        if self.jobDataChanged( self.jobs, job_id, myAttrs ) and myAttrs['status'] in [ 'R', 'Q' ]:
700                                self.jobs[ job_id ] = myAttrs
701
702                                #debug_msg( 10, printTime() + ' job %s state changed' %(job_id) )
703
704                for id, attrs in self.jobs.items():
705
706                        if id not in jobs_processed:
707
708                                # This one isn't there anymore; toedeledoki!
709                                #
710                                del self.jobs[ id ]
711
712        def submitJobData( self ):
713                """Submit job info list"""
714
715                self.dp.multicastGmetric( 'MONARCH-HEARTBEAT', str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
716
717                # Now let's spread the knowledge
718                #
719                for jobid, jobattrs in self.jobs.items():
720
721                        gmetric_val = self.compileGmetricVal( jobid, jobattrs )
722
723                        metric_increment = 0
724
725                        for val in gmetric_val:
726                                self.dp.multicastGmetric( 'MONARCH-JOB-' + jobid + '-' + str(metric_increment), val )
727                                metric_increment = metric_increment + 1
728
729        def compileGmetricVal( self, jobid, jobattrs ):
730                """Create a val string for gmetric of jobinfo"""
731
732                gval_lists = [ ]
733
734                mystr = None
735
736                val_list = { }
737
738                for val_name, val_value in jobattrs.items():
739
740                        val_list_names_len      = len( string.join( val_list.keys() ) ) + len(val_list.keys())
741                        val_list_vals_len       = len( string.join( val_list.values() ) ) + len(val_list.values())
742
743                        if val_name == 'nodes' and jobattrs['status'] == 'R':
744
745                                node_str = None
746
747                                for node in val_value:
748
749                                        if node_str:
750                                                node_str = node_str + ';' + node
751                                        else:
752                                                node_str = node
753
754                                        if (val_list_names_len + len(val_name) ) + (val_list_vals_len + len(node_str) ) > METRIC_MAX_VAL_LEN:
755
756                                                val_list[ val_name ] = node_str
757                                                gval_lists.append( val_list )
758                                                val_list = { }
759                                                node_str = None
760
761                                val_list[ val_name ] = node_str
762                                gval_lists.append( val_list )
763                                val_list = { }
764
765                        elif val_value != '':
766
767                                if (val_list_names_len + len(val_name) ) + (val_list_vals_len + len(str(val_value)) ) > METRIC_MAX_VAL_LEN:
768
769                                        gval_lists.append( val_list )
770                                        val_list = { }
771
772                                val_list[ val_name ] = val_value
773
774                if len(val_list) > 0:
775                        gval_lists.append( val_list )
776
777                str_list = [ ]
778
779                for val_list in gval_lists:
780
781                        my_val_str = None
782
783                        for val_name, val_value in val_list.items():
784
785                                if my_val_str:
786
787                                        my_val_str = my_val_str + ' ' + val_name + '=' + val_value
788                                else:
789                                        my_val_str = val_name + '=' + val_value
790
791                        str_list.append( my_val_str )
792
793                return str_list
794
795def printTime( ):
796        """Print current time/date in human readable format for log/debug"""
797
798        return time.strftime("%a, %d %b %Y %H:%M:%S")
799
800def debug_msg( level, msg ):
801        """Print msg if at or above current debug level"""
802
803        if (DEBUG_LEVEL >= level):
804                        sys.stderr.write( msg + '\n' )
805
806def write_pidfile():
807
808        # Write pidfile if PIDFILE exists
809        if PIDFILE:
810                pid = os.getpid()
811
812                pidfile = open(PIDFILE, 'w')
813                pidfile.write(str(pid))
814                pidfile.close()
815
816def main():
817        """Application start"""
818
819        global PBSQuery, PBSError
820
821        if not processArgs( sys.argv[1:] ):
822                sys.exit( 1 )
823
824        if BATCH_API == 'pbs':
825
826                try:
827                        from PBSQuery import PBSQuery, PBSError
828
829                except ImportError:
830
831                        debug_msg( 0, "fatal error: BATCH_API set to 'pbs' but python module 'pbs_python' is not installed" )
832                        sys.exit( 1 )
833
834                gather = PbsDataGatherer()
835
836        elif BATCH_API == 'sge':
837
838                gather = SgeDataGatherer()
839
840        else:
841                debug_msg( 0, "fatal error: unknown BATCH_API '" + BATCH_API + "' is not supported" )
842                sys.exit( 1 )
843
844        if DAEMONIZE:
845                gather.daemon()
846        else:
847                gather.run()
848
849# wh00t? someone started me! :)
850#
851if __name__ == '__main__':
852        main()
Note: See TracBrowser for help on using the repository browser.