source: trunk/plugin/togap.py @ 95

Last change on this file since 95 was 95, checked in by bastiaans, 18 years ago

plugin/togap.py:

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