Context Navigation

source: trunk/plugin/togap.py @ 184

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

daemon/togad.py:

Delete jobs from memory after written to dbase

plugin/togap.py:

Only accept job attrs changes for jobs in Q or R status

File size: 10.3 KB

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

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

Download in other formats: