source: trunk/jobmond/jobmond.py @ 349

Last change on this file since 349 was 349, checked in by bastiaans, 17 years ago

jobmond/jobmond.py:

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