source: trunk/plugin/togap.py @ 68

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

plugin/togap.py:

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