source: trunk/jobmond/jobmond.py @ 225

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

ALL:

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