source: trunk/jobmond/jobmond.py @ 409

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

web/addons/job_monarch/version.php:

  • SVN version

web/addons/job_monarch/libtoga.php:

  • need $start in my graph.php

jobmond/jobmond.py:

  • added number of running and queued jobs metric reporting

web/addons/job_monarch/graph.php:

  • added 'job_report' graph, shows running and queued jobs

web/addons/job_monarch/overview.php:

  • included IMG to job_report graph

web/addons/job_monarch/index.php:

  • added range menu
  • fixed metric_menu selection now properly instantly refreshes graphs

web/addons/job_monarch/templates/overview.tpl:

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