source: trunk/jobmond/jobmond.py @ 453

Last change on this file since 453 was 451, checked in by bastiaans, 16 years ago

jobmond/jobmond.py:

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