source: trunk/jobmond/jobmond.py @ 354

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

jobmond/jobmond.py:

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