source: trunk/jobmond/jobmond.py @ 253

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

CHANGELOG:

  • lost commit

jobmond/jobmond.py:

  • changed job metric fragmenting. jobinfo fragmented over multiple metrics is now actually WORKING. and there are no more bugs on very big jobs (with lots of nodes)

web/addons/job_monarch/libtoga.php:

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