Context Navigation

source: trunk/plugin/togap.py @ 95

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

plugin/togap.py:

Made sure no empty variabeles are transmitted

File size: 9.5 KB

Rev	Line
[23]	1	#!/usr/bin/env python
	2
[26]	3	# Specify debugging level here;
	4	#
[69]	5	DEBUG_LEVEL = 0
[26]	6
	7	# Wether or not to run as a daemon in background
	8	#
	9	DAEMONIZE = 0
	10
[64]	11	# How many seconds interval for polling of jobs
[61]	12	#
[64]	13	# this will effect directly how accurate the
	14	# end time of a job can be determined
	15	#
	16	TORQUE_POLL_INTERVAL = 10
[61]	17
[68]	18	# Alternate location of gmond.conf
	19	#
	20	# Default: /etc/gmond.conf
	21	#
	22	#GMOND_CONF = '/etc/gmond.conf'
	23
[23]	24	from PBSQuery import PBSQuery
[26]	25	import sys
	26	import time
[65]	27	import os
[67]	28	import socket
[65]	29	import string
[23]	30
[61]	31	class DataProcessor:
[68]	32	"""Class for processing of data"""
[61]	33
	34	binary = '/usr/bin/gmetric'
	35
	36	def __init__( self, binary=None ):
[68]	37	"""Remember alternate binary location if supplied"""
[61]	38
	39	if binary:
	40	self.binary = binary
	41
[80]	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.'
[61]	48
[88]	49	self.dmax = str( int( TORQUE_POLL_INTERVAL ) )
[80]	50
[68]	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
[69]	61	incompatible = self.checkGmetricVersion()
[61]	62
[65]	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]	68	"""
	69	Check version of gmetric is at least 3.0.1
	70	for the syntax we use
	71	"""
[65]	72
	73	for line in os.popen( self.binary + ' --version' ).readlines():
	74
	75	line = line.split( ' ' )
	76
[69]	77	if len( line ) == 2 and str(line).find( 'gmetric' ) != -1:
[65]	78
[69]	79	gmetric_version = line[1].split( '\n' )[0]
[65]	80
[69]	81	version_major = int( gmetric_version.split( '.' )[0] )
	82	version_minor = int( gmetric_version.split( '.' )[1] )
	83	version_patch = int( gmetric_version.split( '.' )[2] )
[65]	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
[91]	97	incompatible = 1
[65]	98
	99	return incompatible
	100
[75]	101	def multicastGmetric( self, metricname, metricval, valtype='string' ):
[68]	102	"""Call gmetric binary and multicast"""
[65]	103
	104	cmd = self.binary
	105
[61]	106	try:
	107	cmd = cmd + ' -c' + GMOND_CONF
	108	except NameError:
[64]	109	debug_msg( 10, 'Assuming /etc/gmond.conf for gmetric cmd (ommitting)' )
[61]	110
[80]	111	cmd = cmd + ' -n' + str( metricname )+ ' -v"' + str( metricval )+ '" -t' + str( valtype ) + ' -d' + str( self.dmax )
[61]	112
	113	print cmd
[69]	114	os.system( cmd )
[61]	115
[23]	116	class PBSDataGatherer:
	117
[61]	118	jobs = { }
	119
[23]	120	def __init__( self ):
[68]	121	"""Setup appropriate variables"""
[23]	122
[26]	123	self.jobs = { }
[61]	124	self.dp = DataProcessor()
[91]	125	self.initPbsQuery()
[23]	126
[91]	127	def initPbsQuery( self ):
	128
	129	self.pq = None
	130	self.pq = PBSQuery()
	131
[26]	132	def getAttr( self, attrs, name ):
[68]	133	"""Return certain attribute from dictionary, if exists"""
[26]	134
	135	if attrs.has_key( name ):
	136	return attrs[name]
	137	else:
	138	return ''
	139
	140	def jobDataChanged( self, jobs, job_id, attrs ):
[68]	141	"""Check if job with attrs and job_id in jobs has changed"""
[26]	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
[65]	161	def getJobData( self, known_jobs ):
[68]	162	"""Gather all data on current jobs in Torque"""
[26]	163
[65]	164	if len( known_jobs ) > 0:
	165	jobs = known_jobs
	166	else:
	167	jobs = { }
[26]	168
[91]	169	self.initPbsQuery()
	170
[26]	171	joblist = self.pq.getjobs()
	172
[69]	173	self.cur_time = time.time()
[68]	174
[26]	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 )
[61]	182
[26]	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' )
[95]	188
[26]	189	mynoderequest = self.getAttr( attrs, 'Resource_List.nodes' )
[95]	190
[26]	191	if mynoderequest.find( ':' ) != -1 and mynoderequest.find( 'ppn' ) != -1:
	192	ppn = mynoderequest.split( ':' )[1].split( 'ppn=' )[1]
	193	else:
	194	ppn = ''
[95]	195
[26]	196	status = self.getAttr( attrs, 'job_state' )
[25]	197
[95]	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
[67]	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
[26]	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
[75]	222	myAttrs['reported'] = str( int( self.cur_time ) )
[67]	223	myAttrs['nodes'] = nodeslist
	224	myAttrs['domain'] = string.join( socket.getfqdn().split( '.' )[1:], '.' )
[80]	225	myAttrs['poll_interval'] = TORQUE_POLL_INTERVAL
[26]	226
	227	if self.jobDataChanged( jobs, job_id, myAttrs ):
	228	jobs[ job_id ] = myAttrs
[61]	229
[26]	230	debug_msg( 10, printTime() + ' job %s state changed' %(job_id) )
	231
[76]	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
[65]	240	return jobs
	241
	242	def submitJobData( self, jobs ):
	243	"""Submit job info list"""
	244
[69]	245	self.dp.multicastGmetric( 'TOGA-HEARTBEAT', str( int( self.cur_time ) ) )
	246
[61]	247	# Now let's spread the knowledge
	248	#
	249	for jobid, jobattrs in jobs.items():
	250
[95]	251	gmetric_val = self.compileGmetricVal( jobid, jobattrs )
[61]	252
[95]	253	for val in gmetric_val:
	254	self.dp.multicastGmetric( 'TOGA-JOB-' + jobid, val )
[61]	255
[67]	256	def makeNodeString( self, nodelist ):
[68]	257	"""Make one big string of all hosts"""
[67]	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
[65]	269	def compileGmetricVal( self, jobid, jobattrs ):
	270	"""Create a val string for gmetric of jobinfo"""
[61]	271
[80]	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'] )
[95]	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
[80]	284	appendList.append( 'status=' + jobattrs['status'] )
[95]	285
	286	if jobattrs['start_timestamp'] != '':
	287	appendList.append( 'start_timestamp=' + jobattrs['start_timestamp'] )
	288
[80]	289	appendList.append( 'reported=' + jobattrs['reported'] )
[85]	290	appendList.append( 'poll_interval=' + str( jobattrs['poll_interval'] ) )
[80]	291	appendList.append( 'domain=' + jobattrs['domain'] )
[26]	292
[95]	293	if len( jobattrs['nodes'] ) > 0:
	294	appendList.append( 'nodes=' + self.makeNodeString( jobattrs['nodes'] ) )
	295
[65]	296	return self.makeAppendLists( appendList )
	297
	298	def makeAppendLists( self, append_list ):
[68]	299	"""
	300	Divide all values from append_list over strings with a maximum
	301	size of 1400
	302	"""
[65]	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
[69]	327	# Max frame size of a udp datagram is 1500 bytes
	328	# removing misc header and gmetric stuff leaves about 1400 bytes
	329	#
[65]	330	if len( val + text ) > 1400:
	331	return 1
	332	else:
	333	return 0
	334
[61]	335	def printJobs( self, jobs ):
[65]	336	"""Print a jobinfo overview"""
	337
[26]	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
[61]	346	def printJob( self, jobs, job_id ):
[65]	347	"""Print job with job_id from jobs"""
[26]	348
	349	print 'job %s' %(job_id)
	350
[65]	351	for name, val in jobs[ job_id ].items():
[26]	352
	353	print '\t%s = %s' %( name, val )
	354
	355	def daemon( self ):
[65]	356	"""Run as daemon forever"""
[26]	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 ):
[65]	390	"""Main thread"""
[26]	391
	392	while ( 1 ):
	393
[65]	394	self.jobs = self.getJobData( self.jobs )
	395	self.submitJobData( self.jobs )
[64]	396	time.sleep( TORQUE_POLL_INTERVAL )
[26]	397
	398	def printTime( ):
[65]	399	"""Print current time/date in human readable format for log/debug"""
[26]	400
	401	return time.strftime("%a, %d %b %Y %H:%M:%S")
	402
	403	def debug_msg( level, msg ):
[65]	404	"""Print msg if at or above current debug level"""
[26]	405
	406	if (DEBUG_LEVEL >= level):
	407	sys.stderr.write( msg + '\n' )
	408
[23]	409	def main():
[65]	410	"""Application start"""
[23]	411
	412	gather = PBSDataGatherer()
[26]	413	if DAEMONIZE:
	414	gather.daemon()
	415	else:
	416	gather.run()
[23]	417
[65]	418	# w00t someone started me
	419	#
[23]	420	if __name__ == '__main__':
	421	main()

Note: See TracBrowser for help on using the repository browser.

Download in other formats: