source: trunk/plugin/togap.py @ 175

Last change on this file since 175 was 174, checked in by bastiaans, 19 years ago

plugin/togap.py:

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