source: trunk/jobmond/jobmond.py @ 224

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

jobarchived/examples:

  • added

jobmond/jobmond.py:

  • removed config option comments

jobarchived/jobarchived.py:

  • removed config option comments
  • added RRDTOOL option

jobmond/jobmond.conf:

  • added config option comments

jobarchived/jobarchived.conf:

  • added config option comments
  • added RRDTOOL option
File size: 12.6 KB
Line 
1#!/usr/bin/env python
2
3import sys, getopt, ConfigParser
4
5def processArgs( args ):
6
7        SHORT_L = 'c:'
8        LONG_L = 'config='
9
10        config_filename = None
11
12        try:
13
14                opts, args = getopt.getopt( args, SHORT_L, LONG_L )
15
16        except getopt.error, detail:
17
18                print detail
19                sys.exit(1)
20
21        for opt, value in opts:
22
23                if opt in [ '--config', '-c' ]:
24               
25                        config_filename = value
26
27        if not config_filename:
28
29                config_filename = '/etc/jobmond.conf'
30
31        return loadConfig( config_filename )
32
33def loadConfig( filename ):
34
35        def getlist( cfg_string ):
36
37                my_list = [ ]
38
39                for item_txt in cfg_string.split( ',' ):
40
41                        sep_char = None
42
43                        item_txt = item_txt.strip()
44
45                        for s_char in [ "'", '"' ]:
46
47                                if item_txt.find( s_char ) != -1:
48
49                                        if item_txt.count( s_char ) != 2:
50
51                                                print 'Missing quote: %s' %item_txt
52                                                sys.exit( 1 )
53
54                                        else:
55
56                                                sep_char = s_char
57                                                break
58
59                        if sep_char:
60
61                                item_txt = item_txt.split( sep_char )[1]
62
63                        my_list.append( item_txt )
64
65                return my_list
66
67        cfg = ConfigParser.ConfigParser()
68
69        cfg.read( filename )
70
71        global DEBUG_LEVEL, DAEMONIZE, TORQUE_SERVER, TORQUE_POLL_INTERVAL, GMOND_CONF, DETECT_TIME_DIFFS, BATCH_HOST_TRANSLATE
72
73        DEBUG_LEVEL = cfg.getint( 'DEFAULT', 'DEBUG_LEVEL' )
74
75        DAEMONIZE = cfg.getboolean( 'DEFAULT', 'DAEMONIZE' )
76
77        TORQUE_SERVER = cfg.get( 'DEFAULT', 'TORQUE_SERVER' )
78
79        TORQUE_POLL_INTERVAL = cfg.getint( 'DEFAULT', 'TORQUE_POLL_INTERVAL' )
80
81        GMOND_CONF = cfg.get( 'DEFAULT', 'GMOND_CONF' )
82
83        DETECT_TIME_DIFFS = cfg.getboolean( 'DEFAULT', 'DETECT_TIME_DIFFS' )
84
85        BATCH_HOST_TRANSLATE = getlist( cfg.get( 'DEFAULT', 'BATCH_HOST_TRANSLATE' ) )
86
87        return True
88
89from PBSQuery import PBSQuery
90
91import time, os, socket, string, re
92
93class DataProcessor:
94        """Class for processing of data"""
95
96        binary = '/usr/bin/gmetric'
97
98        def __init__( self, binary=None ):
99                """Remember alternate binary location if supplied"""
100
101                if binary:
102                        self.binary = binary
103
104                # Timeout for XML
105                #
106                # From ganglia's documentation:
107                #
108                # 'A metric will be deleted DMAX seconds after it is received, and
109                # DMAX=0 means eternal life.'
110
111                self.dmax = str( int( int( TORQUE_POLL_INTERVAL ) + 2 ) )
112
113                try:
114                        gmond_file = GMOND_CONF
115
116                except NameError:
117                        gmond_file = '/etc/gmond.conf'
118
119                if not os.path.exists( gmond_file ):
120                        debug_msg( 0, gmond_file + ' does not exist' )
121                        sys.exit( 1 )
122
123                incompatible = self.checkGmetricVersion()
124
125                if incompatible:
126                        debug_msg( 0, 'Gmetric version not compatible, pls upgrade to at least 3.0.1' )
127                        sys.exit( 1 )
128
129        def checkGmetricVersion( self ):
130                """
131                Check version of gmetric is at least 3.0.1
132                for the syntax we use
133                """
134
135                for line in os.popen( self.binary + ' --version' ).readlines():
136
137                        line = line.split( ' ' )
138
139                        if len( line ) == 2 and str(line).find( 'gmetric' ) != -1:
140                       
141                                gmetric_version = line[1].split( '\n' )[0]
142
143                                version_major = int( gmetric_version.split( '.' )[0] )
144                                version_minor = int( gmetric_version.split( '.' )[1] )
145                                version_patch = int( gmetric_version.split( '.' )[2] )
146
147                                incompatible = 0
148
149                                if version_major < 3:
150
151                                        incompatible = 1
152                               
153                                elif version_major == 3:
154
155                                        if version_minor == 0:
156
157                                                if version_patch < 1:
158                                               
159                                                        incompatible = 1
160
161                return incompatible
162
163        def multicastGmetric( self, metricname, metricval, valtype='string' ):
164                """Call gmetric binary and multicast"""
165
166                cmd = self.binary
167
168                try:
169                        cmd = cmd + ' -c' + GMOND_CONF
170                except NameError:
171                        debug_msg( 10, 'Assuming /etc/gmond.conf for gmetric cmd (ommitting)' )
172
173                cmd = cmd + ' -n' + str( metricname )+ ' -v"' + str( metricval )+ '" -t' + str( valtype ) + ' -d' + str( self.dmax )
174
175                debug_msg( 10, printTime() + ' ' + cmd )
176                os.system( cmd )
177
178class DataGatherer:
179
180        jobs = { }
181
182        def __init__( self ):
183                """Setup appropriate variables"""
184
185                self.jobs = { }
186                self.timeoffset = 0
187                self.dp = DataProcessor()
188                self.initPbsQuery()
189
190        def initPbsQuery( self ):
191
192                self.pq = None
193                if( TORQUE_SERVER ):
194                        self.pq = PBSQuery( TORQUE_SERVER )
195                else:
196                        self.pq = PBSQuery()
197
198        def getAttr( self, attrs, name ):
199                """Return certain attribute from dictionary, if exists"""
200
201                if attrs.has_key( name ):
202                        return attrs[name]
203                else:
204                        return ''
205
206        def jobDataChanged( self, jobs, job_id, attrs ):
207                """Check if job with attrs and job_id in jobs has changed"""
208
209                if jobs.has_key( job_id ):
210                        oldData = jobs[ job_id ]       
211                else:
212                        return 1
213
214                for name, val in attrs.items():
215
216                        if oldData.has_key( name ):
217
218                                if oldData[ name ] != attrs[ name ]:
219
220                                        return 1
221
222                        else:
223                                return 1
224
225                return 0
226
227        def getJobData( self, known_jobs ):
228                """Gather all data on current jobs in Torque"""
229
230                if len( known_jobs ) > 0:
231                        jobs = known_jobs
232                else:
233                        jobs = { }
234
235                #self.initPbsQuery()
236       
237                #print self.pq.getnodes()
238       
239                joblist = self.pq.getjobs()
240
241                self.cur_time = time.time()
242
243                jobs_processed = [ ]
244
245                #self.printJobs( joblist )
246
247                for name, attrs in joblist.items():
248
249                        job_id = name.split( '.' )[0]
250
251                        jobs_processed.append( job_id )
252
253                        name = self.getAttr( attrs, 'Job_Name' )
254                        queue = self.getAttr( attrs, 'queue' )
255                        owner = self.getAttr( attrs, 'Job_Owner' ).split( '@' )[0]
256                        requested_time = self.getAttr( attrs, 'Resource_List.walltime' )
257                        requested_memory = self.getAttr( attrs, 'Resource_List.mem' )
258
259                        mynoderequest = self.getAttr( attrs, 'Resource_List.nodes' )
260
261                        if mynoderequest.find( ':' ) != -1 and mynoderequest.find( 'ppn' ) != -1:
262                                ppn = mynoderequest.split( ':' )[1].split( 'ppn=' )[1]
263                        else:
264                                ppn = ''
265
266                        status = self.getAttr( attrs, 'job_state' )
267
268                        if status == 'R':
269                                start_timestamp = self.getAttr( attrs, 'mtime' )
270                                nodes = self.getAttr( attrs, 'exec_host' ).split( '+' )
271
272                                nodeslist = [ ]
273
274                                for node in nodes:
275                                        host = node.split( '/' )[0]
276
277                                        if nodeslist.count( host ) == 0:
278
279                                                for translate_pattern in BATCH_HOST_TRANSLATE:
280
281                                                        if translate_pattern.find( '/' ) != -1:
282
283                                                                translate_orig = translate_pattern.split( '/' )[1]
284                                                                translate_new = translate_pattern.split( '/' )[2]
285
286                                                                host = re.sub( translate_orig, translate_new, host )
287                               
288                                                if not host in nodeslist:
289                               
290                                                        nodeslist.append( host )
291
292                                if DETECT_TIME_DIFFS:
293
294                                        # If a job start if later than our current date,
295                                        # that must mean the Torque server's time is later
296                                        # than our local time.
297                               
298                                        if int(start_timestamp) > int( int(self.cur_time) + int(self.timeoffset) ):
299
300                                                self.timeoffset = int( int(start_timestamp) - int(self.cur_time) )
301
302                        elif status == 'Q':
303                                start_timestamp = ''
304                                count_mynodes = 0
305                                numeric_node = 1
306
307                                for node in mynoderequest.split( '+' ):
308
309                                        nodepart = node.split( ':' )[0]
310
311                                        for letter in nodepart:
312
313                                                if letter not in string.digits:
314
315                                                        numeric_node = 0
316
317                                        if not numeric_node:
318                                                count_mynodes = count_mynodes + 1
319                                        else:
320                                                count_mynodes = count_mynodes + int( nodepart )
321                                               
322                                nodeslist = count_mynodes
323                        else:
324                                start_timestamp = ''
325                                nodeslist = ''
326
327                        myAttrs = { }
328                        myAttrs['name'] = name
329                        myAttrs['queue'] = queue
330                        myAttrs['owner'] = owner
331                        myAttrs['requested_time'] = requested_time
332                        myAttrs['requested_memory'] = requested_memory
333                        myAttrs['ppn'] = ppn
334                        myAttrs['status'] = status
335                        myAttrs['start_timestamp'] = start_timestamp
336                        myAttrs['reported'] = str( int( int( self.cur_time ) + int( self.timeoffset ) ) )
337                        myAttrs['nodes'] = nodeslist
338                        myAttrs['domain'] = string.join( socket.getfqdn().split( '.' )[1:], '.' )
339                        myAttrs['poll_interval'] = TORQUE_POLL_INTERVAL
340
341                        if self.jobDataChanged( jobs, job_id, myAttrs ) and myAttrs['status'] in [ 'R', 'Q' ]:
342                                jobs[ job_id ] = myAttrs
343
344                                #debug_msg( 10, printTime() + ' job %s state changed' %(job_id) )
345
346                for id, attrs in jobs.items():
347
348                        if id not in jobs_processed:
349
350                                # This one isn't there anymore; toedeledoki!
351                                #
352                                del jobs[ id ]
353
354                return jobs
355
356        def submitJobData( self, jobs ):
357                """Submit job info list"""
358
359                self.dp.multicastGmetric( 'MONARCH-HEARTBEAT', str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
360
361                # Now let's spread the knowledge
362                #
363                for jobid, jobattrs in jobs.items():
364
365                        gmetric_val = self.compileGmetricVal( jobid, jobattrs )
366
367                        for val in gmetric_val:
368                                self.dp.multicastGmetric( 'MONARCH-JOB-' + jobid, val )
369
370        def makeNodeString( self, nodelist ):
371                """Make one big string of all hosts"""
372
373                node_str = None
374
375                for node in nodelist:
376                        if not node_str:
377                                node_str = node
378                        else:
379                                node_str = node_str + ';' + node
380
381                return node_str
382
383        def compileGmetricVal( self, jobid, jobattrs ):
384                """Create a val string for gmetric of jobinfo"""
385
386                appendList = [ ]
387                appendList.append( 'name=' + jobattrs['name'] )
388                appendList.append( 'queue=' + jobattrs['queue'] )
389                appendList.append( 'owner=' + jobattrs['owner'] )
390                appendList.append( 'requested_time=' + jobattrs['requested_time'] )
391
392                if jobattrs['requested_memory'] != '':
393                        appendList.append( 'requested_memory=' + jobattrs['requested_memory'] )
394
395                if jobattrs['ppn'] != '':
396                        appendList.append( 'ppn=' + jobattrs['ppn'] )
397
398                appendList.append( 'status=' + jobattrs['status'] )
399
400                if jobattrs['start_timestamp'] != '':
401                        appendList.append( 'start_timestamp=' + jobattrs['start_timestamp'] )
402
403                appendList.append( 'reported=' + jobattrs['reported'] )
404                appendList.append( 'poll_interval=' + str( jobattrs['poll_interval'] ) )
405                appendList.append( 'domain=' + jobattrs['domain'] )
406
407                if jobattrs['status'] == 'R':
408                        if len( jobattrs['nodes'] ) > 0:
409                                appendList.append( 'nodes=' + self.makeNodeString( jobattrs['nodes'] ) )
410                elif jobattrs['status'] == 'Q':
411                        appendList.append( 'nodes=' + str(jobattrs['nodes']) )
412
413                return self.makeAppendLists( appendList )
414
415        def makeAppendLists( self, append_list ):
416                """
417                Divide all values from append_list over strings with a maximum
418                size of 1400
419                """
420
421                app_lists = [ ]
422
423                mystr = None
424
425                for val in append_list:
426
427                        if not mystr:
428                                mystr = val
429                        else:
430                                if not self.checkValAppendMaxSize( mystr, val ):
431                                        mystr = mystr + ' ' + val
432                                else:
433                                        # Too big, new appenlist
434                                        app_lists.append( mystr )
435                                        mystr = val
436
437                app_lists.append( mystr )
438
439                return app_lists
440
441        def checkValAppendMaxSize( self, val, text ):
442                """Check if val + text size is not above 1400 (max msg size)"""
443
444                # Max frame size of a udp datagram is 1500 bytes
445                # removing misc header and gmetric stuff leaves about 1300 bytes
446                #
447                if len( val + text ) > 900:
448                        return 1
449                else:
450                        return 0
451
452        def printJobs( self, jobs ):
453                """Print a jobinfo overview"""
454
455                for name, attrs in self.jobs.items():
456
457                        print 'job %s' %(name)
458
459                        for name, val in attrs.items():
460
461                                print '\t%s = %s' %( name, val )
462
463        def printJob( self, jobs, job_id ):
464                """Print job with job_id from jobs"""
465
466                print 'job %s' %(job_id)
467
468                for name, val in jobs[ job_id ].items():
469
470                        print '\t%s = %s' %( name, val )
471
472        def daemon( self ):
473                """Run as daemon forever"""
474
475                # Fork the first child
476                #
477                pid = os.fork()
478                if pid > 0:
479                        sys.exit(0)  # end parent
480
481                # creates a session and sets the process group ID
482                #
483                os.setsid()
484
485                # Fork the second child
486                #
487                pid = os.fork()
488                if pid > 0:
489                        sys.exit(0)  # end parent
490
491                # Go to the root directory and set the umask
492                #
493                os.chdir('/')
494                os.umask(0)
495
496                sys.stdin.close()
497                sys.stdout.close()
498                sys.stderr.close()
499
500                os.open('/dev/null', 0)
501                os.dup(0)
502                os.dup(0)
503
504                self.run()
505
506        def run( self ):
507                """Main thread"""
508
509                while ( 1 ):
510               
511                        self.jobs = self.getJobData( self.jobs )
512                        self.submitJobData( self.jobs )
513                        time.sleep( TORQUE_POLL_INTERVAL )     
514
515def printTime( ):
516        """Print current time/date in human readable format for log/debug"""
517
518        return time.strftime("%a, %d %b %Y %H:%M:%S")
519
520def debug_msg( level, msg ):
521        """Print msg if at or above current debug level"""
522
523        if (DEBUG_LEVEL >= level):
524                        sys.stderr.write( msg + '\n' )
525
526def main():
527        """Application start"""
528
529        if not processArgs( sys.argv[1:] ):
530                sys.exit( 1 )
531
532        gather = DataGatherer()
533        if DAEMONIZE:
534                gather.daemon()
535        else:
536                gather.run()
537
538# w00t someone started me
539#
540if __name__ == '__main__':
541        main()
Note: See TracBrowser for help on using the repository browser.