source: trunk/plugin/togap.py @ 125

Last change on this file since 125 was 125, checked in by bastiaans, 17 years ago

plugin/togap.py:

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