source: trunk/plugin/togap.py @ 67

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

plugin/togap.py:

  • Added domain name parsing
  • Added nodelist parsing
File size: 7.7 KB
Line 
1#!/usr/bin/env python
2
3# Specify debugging level here;
4#
5DEBUG_LEVEL = 10
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
18from PBSQuery import PBSQuery
19import sys
20import time
21import os
22import socket
23import string
24
25class DataProcessor:
26
27        binary = '/usr/bin/gmetric'
28
29        def __init__( self, binary=None ):
30
31                if binary:
32                        self.binary = binary
33
34                self.dmax = TORQUE_POLL_INTERVAL
35
36                #incompatible = self.checkGmetricVersion
37                incompatible = 0
38
39                if incompatible:
40                        debug_msg( 0, 'Gmetric version not compatible, pls upgrade to at least 3.0.1' )
41                        sys.exit( 1 )
42
43        def checkGmetricVersion( self ):
44
45                for line in os.popen( self.binary + ' --version' ).readlines():
46
47                        line = line.split( ' ' )
48
49                        if len( line ) == 2 and line.find( 'gmetric' ) != -1:
50                       
51                                gmetric_version = line[1]
52
53                                version_major = int( gemtric_version.split( '.' )[0] )
54                                version_minor = int( gemtric_version.split( '.' )[1] )
55                                version_patch = int( gemtric_version.split( '.' )[2] )
56
57                                incompatible = 0
58
59                                if version_major < 3:
60
61                                        incompatible = 1
62                               
63                                elif version_major == 3:
64
65                                        if version_minor == 0:
66
67                                                if version_patch < 1:
68                                               
69                                                        incompatbiel = 1
70
71                return incompatible
72
73        def multicastGmetric( self, metricname, metricval, valtype='string', tmax='15' ):
74
75                cmd = self.binary
76
77                try:
78                        cmd = cmd + ' -c' + GMOND_CONF
79                except NameError:
80                        debug_msg( 10, 'Assuming /etc/gmond.conf for gmetric cmd (ommitting)' )
81
82                cmd = cmd + ' -n' + metricname + ' -v"' + metricval + '" -t' + valtype + ' -x' + tmax + ' -d' + str( self.dmax )
83
84                print cmd
85                #os.system( cmd )
86
87class PBSDataGatherer:
88
89        jobs = { }
90
91        def __init__( self ):
92
93                self.pq = PBSQuery()
94                self.jobs = { }
95                self.dp = DataProcessor()
96
97        def getAttr( self, attrs, name ):
98
99                if attrs.has_key( name ):
100                        return attrs[name]
101                else:
102                        return ''
103
104        def jobDataChanged( self, jobs, job_id, attrs ):
105
106                if jobs.has_key( job_id ):
107                        oldData = jobs[ job_id ]       
108                else:
109                        return 1
110
111                for name, val in attrs.items():
112
113                        if oldData.has_key( name ):
114
115                                if oldData[ name ] != attrs[ name ]:
116
117                                        return 1
118
119                        else:
120                                return 1
121
122                return 0
123
124        def getJobData( self, known_jobs ):
125
126                if len( known_jobs ) > 0:
127                        jobs = known_jobs
128                else:
129                        jobs = { }
130
131                joblist = self.pq.getjobs()
132
133                jobs_processed = [ ]
134
135                for name, attrs in joblist.items():
136
137                        print attrs
138
139                        job_id = name.split( '.' )[0]
140
141                        jobs_processed.append( job_id )
142
143                        name = self.getAttr( attrs, 'Job_Name' )
144                        queue = self.getAttr( attrs, 'queue' )
145                        owner = self.getAttr( attrs, 'Job_Owner' ).split( '@' )[0]
146                        requested_time = self.getAttr( attrs, 'Resource_List.walltime' )
147                        requested_memory = self.getAttr( attrs, 'Resource_List.mem' )
148                        mynoderequest = self.getAttr( attrs, 'Resource_List.nodes' )
149                        if mynoderequest.find( ':' ) != -1 and mynoderequest.find( 'ppn' ) != -1:
150                                ppn = mynoderequest.split( ':' )[1].split( 'ppn=' )[1]
151                        else:
152                                ppn = ''
153                        status = self.getAttr( attrs, 'job_state' )
154                        start_timestamp = self.getAttr( attrs, 'mtime' )
155
156                        nodes = self.getAttr( attrs, 'exec_host' ).split( '+' )
157                        nodeslist = [ ]
158
159                        for node in nodes:
160                                host = node.split( '/' )[0]
161
162                                if nodeslist.count( host ) == 0:
163                                        nodeslist.append( node.split( '/' )[0] )
164
165                        myAttrs = { }
166                        myAttrs['name'] = name
167                        myAttrs['queue'] = queue
168                        myAttrs['owner'] = owner
169                        myAttrs['requested_time'] = requested_time
170                        myAttrs['requested_memory'] = requested_memory
171                        myAttrs['ppn'] = ppn
172                        myAttrs['status'] = status
173                        myAttrs['start_timestamp'] = start_timestamp
174                        myAttrs['nodes'] = nodeslist
175                        myAttrs['domain'] = string.join( socket.getfqdn().split( '.' )[1:], '.' )
176
177                        if self.jobDataChanged( jobs, job_id, myAttrs ):
178                                jobs[ job_id ] = myAttrs
179
180                                self.printJob( jobs, job_id )
181
182                                debug_msg( 10, printTime() + ' job %s state changed' %(job_id) )
183
184                return jobs
185
186        def submitJobData( self, jobs ):
187                """Submit job info list"""
188
189                time_now = time.time()
190
191                self.dp.multicastGmetric( 'TOGA-HEARTBEAT', str( time_now ), 'float' )
192
193                # Now let's spread the knowledge
194                #
195                for jobid, jobattrs in jobs.items():
196
197                        gmetric_val = self.compileGmetricVal( jobid, jobattrs )
198
199                        for val in gmetric_val:
200                                self.dp.multicastGmetric( 'TOGA-JOB-' + jobid, val )
201
202        def makeNodeString( self, nodelist ):
203
204                node_str = None
205
206                for node in nodelist:
207                        if not node_str:
208                                node_str = node
209                        else:
210                                node_str = node_str + ';' + node
211
212                return node_str
213
214        def compileGmetricVal( self, jobid, jobattrs ):
215                """Create a val string for gmetric of jobinfo"""
216
217                name_str = 'name=' + jobattrs['name']
218                queue_str = 'queue=' + jobattrs['queue']
219                owner_str = 'owner=' + jobattrs['owner']
220                rtime_str = 'rtime=' + jobattrs['requested_time']
221                rmem_str = 'rmem=' + jobattrs['requested_memory']
222                ppn_str = 'ppn=' + jobattrs['ppn']
223                status_str = 'status=' + jobattrs['status']
224                stime_str = 'stime=' + jobattrs['start_timestamp']
225                domain_str = 'domain=' + jobattrs['domain']
226                node_str = 'nodes=' + self.makeNodeString( jobattrs['nodes'] )
227
228                appendList = [ name_str, queue_str, owner_str, rtime_str, rmem_str, ppn_str, status_str, stime_str, domain_str, node_str ]
229
230                return self.makeAppendLists( appendList )
231
232        def makeAppendLists( self, append_list ):
233
234                app_lists = [ ]
235
236                mystr = None
237
238                for val in append_list:
239
240                        if not mystr:
241                                mystr = val
242                        else:
243                                if not self.checkValAppendMaxSize( mystr, val ):
244                                        mystr = mystr + ' ' + val
245                                else:
246                                        # Too big, new appenlist
247                                        app_lists.append( mystr )
248                                        mystr = val
249
250                app_lists.append( mystr )
251
252                return app_lists
253
254        def checkValAppendMaxSize( self, val, text ):
255                """Check if val + text size is not above 1400 (max msg size)"""
256
257                if len( val + text ) > 1400:
258                        return 1
259                else:
260                        return 0
261
262        def printJobs( self, jobs ):
263                """Print a jobinfo overview"""
264
265                for name, attrs in self.jobs.items():
266
267                        print 'job %s' %(name)
268
269                        for name, val in attrs.items():
270
271                                print '\t%s = %s' %( name, val )
272
273        def printJob( self, jobs, job_id ):
274                """Print job with job_id from jobs"""
275
276                print 'job %s' %(job_id)
277
278                for name, val in jobs[ job_id ].items():
279
280                        print '\t%s = %s' %( name, val )
281
282        def daemon( self ):
283                """Run as daemon forever"""
284
285                # Fork the first child
286                #
287                pid = os.fork()
288                if pid > 0:
289                        sys.exit(0)  # end parrent
290
291                # creates a session and sets the process group ID
292                #
293                os.setsid()
294
295                # Fork the second child
296                #
297                pid = os.fork()
298                if pid > 0:
299                        sys.exit(0)  # end parrent
300
301                # Go to the root directory and set the umask
302                #
303                os.chdir('/')
304                os.umask(0)
305
306                sys.stdin.close()
307                sys.stdout.close()
308                sys.stderr.close()
309
310                os.open('/dev/null', 0)
311                os.dup(0)
312                os.dup(0)
313
314                self.run()
315
316        def run( self ):
317                """Main thread"""
318
319                while ( 1 ):
320               
321                        self.jobs = self.getJobData( self.jobs )
322                        self.submitJobData( self.jobs )
323                        time.sleep( TORQUE_POLL_INTERVAL )     
324
325def printTime( ):
326        """Print current time/date in human readable format for log/debug"""
327
328        return time.strftime("%a, %d %b %Y %H:%M:%S")
329
330def debug_msg( level, msg ):
331        """Print msg if at or above current debug level"""
332
333        if (DEBUG_LEVEL >= level):
334                        sys.stderr.write( msg + '\n' )
335
336def main():
337        """Application start"""
338
339        gather = PBSDataGatherer()
340        if DAEMONIZE:
341                gather.daemon()
342        else:
343                gather.run()
344
345# w00t someone started me
346#
347if __name__ == '__main__':
348        main()
Note: See TracBrowser for help on using the repository browser.