source: trunk/jobmond/jobmond.py @ 359

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

jobmond/jobmond.py:

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