Context Navigation

source: trunk/plugin/togap.py @ 125

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

plugin/togap.py:

Daemon & real cluster

File size: 9.6 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
[64]	12	# How many seconds interval for polling of jobs
[61]	13	#
[64]	14	# this will effect directly how accurate the
	15	# end time of a job can be determined
	16	#
	17	TORQUE_POLL_INTERVAL = 10
[61]	18
[68]	19	# Alternate location of gmond.conf
	20	#
	21	# Default: /etc/gmond.conf
	22	#
	23	#GMOND_CONF = '/etc/gmond.conf'
	24
[23]	25	from PBSQuery import PBSQuery
[26]	26	import sys
	27	import time
[65]	28	import os
[67]	29	import socket
[65]	30	import string
[23]	31
[61]	32	class DataProcessor:
[68]	33	"""Class for processing of data"""
[61]	34
	35	binary = '/usr/bin/gmetric'
	36
	37	def __init__( self, binary=None ):
[68]	38	"""Remember alternate binary location if supplied"""
[61]	39
	40	if binary:
	41	self.binary = binary
	42
[80]	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.'
[61]	49
[88]	50	self.dmax = str( int( TORQUE_POLL_INTERVAL ) )
[80]	51
[68]	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
[69]	62	incompatible = self.checkGmetricVersion()
[61]	63
[65]	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 ):
[68]	69	"""
	70	Check version of gmetric is at least 3.0.1
	71	for the syntax we use
	72	"""
[65]	73
	74	for line in os.popen( self.binary + ' --version' ).readlines():
	75
	76	line = line.split( ' ' )
	77
[69]	78	if len( line ) == 2 and str(line).find( 'gmetric' ) != -1:
[65]	79
[69]	80	gmetric_version = line[1].split( '\n' )[0]
[65]	81
[69]	82	version_major = int( gmetric_version.split( '.' )[0] )
	83	version_minor = int( gmetric_version.split( '.' )[1] )
	84	version_patch = int( gmetric_version.split( '.' )[2] )
[65]	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
[91]	98	incompatible = 1
[65]	99
	100	return incompatible
	101
[75]	102	def multicastGmetric( self, metricname, metricval, valtype='string' ):
[68]	103	"""Call gmetric binary and multicast"""
[65]	104
	105	cmd = self.binary
	106
[61]	107	try:
	108	cmd = cmd + ' -c' + GMOND_CONF
	109	except NameError:
[64]	110	debug_msg( 10, 'Assuming /etc/gmond.conf for gmetric cmd (ommitting)' )
[61]	111
[80]	112	cmd = cmd + ' -n' + str( metricname )+ ' -v"' + str( metricval )+ '" -t' + str( valtype ) + ' -d' + str( self.dmax )
[61]	113
[101]	114	debug_msg( 10, printTime() + ' ' + cmd )
[69]	115	os.system( cmd )
[61]	116
[23]	117	class PBSDataGatherer:
	118
[61]	119	jobs = { }
	120
[23]	121	def __init__( self ):
[68]	122	"""Setup appropriate variables"""
[23]	123
[26]	124	self.jobs = { }
[61]	125	self.dp = DataProcessor()
[91]	126	self.initPbsQuery()
[23]	127
[91]	128	def initPbsQuery( self ):
	129
	130	self.pq = None
[125]	131	self.pq = PBSQuery( 'login.irc.sara.nl' )
[91]	132
[26]	133	def getAttr( self, attrs, name ):
[68]	134	"""Return certain attribute from dictionary, if exists"""
[26]	135
	136	if attrs.has_key( name ):
	137	return attrs[name]
	138	else:
	139	return ''
	140
	141	def jobDataChanged( self, jobs, job_id, attrs ):
[68]	142	"""Check if job with attrs and job_id in jobs has changed"""
[26]	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
[65]	162	def getJobData( self, known_jobs ):
[68]	163	"""Gather all data on current jobs in Torque"""
[26]	164
[65]	165	if len( known_jobs ) > 0:
	166	jobs = known_jobs
	167	else:
	168	jobs = { }
[26]	169
[101]	170	#self.initPbsQuery()
[125]	171
	172	#print self.pq.getnodes()
	173
[26]	174	joblist = self.pq.getjobs()
	175
[69]	176	self.cur_time = time.time()
[68]	177
[26]	178	jobs_processed = [ ]
	179
[125]	180	#self.printJobs( joblist )
	181
[26]	182	for name, attrs in joblist.items():
	183
	184	job_id = name.split( '.' )[0]
	185
	186	jobs_processed.append( job_id )
[61]	187
[26]	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' )
[95]	193
[26]	194	mynoderequest = self.getAttr( attrs, 'Resource_List.nodes' )
[95]	195
[26]	196	if mynoderequest.find( ':' ) != -1 and mynoderequest.find( 'ppn' ) != -1:
	197	ppn = mynoderequest.split( ':' )[1].split( 'ppn=' )[1]
	198	else:
	199	ppn = ''
[95]	200
[26]	201	status = self.getAttr( attrs, 'job_state' )
[25]	202
[95]	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
[67]	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
[26]	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
[75]	227	myAttrs['reported'] = str( int( self.cur_time ) )
[67]	228	myAttrs['nodes'] = nodeslist
	229	myAttrs['domain'] = string.join( socket.getfqdn().split( '.' )[1:], '.' )
[80]	230	myAttrs['poll_interval'] = TORQUE_POLL_INTERVAL
[26]	231
	232	if self.jobDataChanged( jobs, job_id, myAttrs ):
	233	jobs[ job_id ] = myAttrs
[61]	234
[101]	235	#debug_msg( 10, printTime() + ' job %s state changed' %(job_id) )
[26]	236
[76]	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
[65]	245	return jobs
	246
	247	def submitJobData( self, jobs ):
	248	"""Submit job info list"""
	249
[69]	250	self.dp.multicastGmetric( 'TOGA-HEARTBEAT', str( int( self.cur_time ) ) )
	251
[61]	252	# Now let's spread the knowledge
	253	#
	254	for jobid, jobattrs in jobs.items():
	255
[95]	256	gmetric_val = self.compileGmetricVal( jobid, jobattrs )
[61]	257
[95]	258	for val in gmetric_val:
	259	self.dp.multicastGmetric( 'TOGA-JOB-' + jobid, val )
[61]	260
[67]	261	def makeNodeString( self, nodelist ):
[68]	262	"""Make one big string of all hosts"""
[67]	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
[65]	274	def compileGmetricVal( self, jobid, jobattrs ):
	275	"""Create a val string for gmetric of jobinfo"""
[61]	276
[80]	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'] )
[95]	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
[80]	289	appendList.append( 'status=' + jobattrs['status'] )
[95]	290
	291	if jobattrs['start_timestamp'] != '':
	292	appendList.append( 'start_timestamp=' + jobattrs['start_timestamp'] )
	293
[80]	294	appendList.append( 'reported=' + jobattrs['reported'] )
[85]	295	appendList.append( 'poll_interval=' + str( jobattrs['poll_interval'] ) )
[80]	296	appendList.append( 'domain=' + jobattrs['domain'] )
[26]	297
[95]	298	if len( jobattrs['nodes'] ) > 0:
	299	appendList.append( 'nodes=' + self.makeNodeString( jobattrs['nodes'] ) )
	300
[65]	301	return self.makeAppendLists( appendList )
	302
	303	def makeAppendLists( self, append_list ):
[68]	304	"""
	305	Divide all values from append_list over strings with a maximum
	306	size of 1400
	307	"""
[65]	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
[69]	332	# Max frame size of a udp datagram is 1500 bytes
	333	# removing misc header and gmetric stuff leaves about 1400 bytes
	334	#
[65]	335	if len( val + text ) > 1400:
	336	return 1
	337	else:
	338	return 0
	339
[61]	340	def printJobs( self, jobs ):
[65]	341	"""Print a jobinfo overview"""
	342
[26]	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
[61]	351	def printJob( self, jobs, job_id ):
[65]	352	"""Print job with job_id from jobs"""
[26]	353
	354	print 'job %s' %(job_id)
	355
[65]	356	for name, val in jobs[ job_id ].items():
[26]	357
	358	print '\t%s = %s' %( name, val )
	359
	360	def daemon( self ):
[65]	361	"""Run as daemon forever"""
[26]	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 ):
[65]	395	"""Main thread"""
[26]	396
	397	while ( 1 ):
	398
[65]	399	self.jobs = self.getJobData( self.jobs )
	400	self.submitJobData( self.jobs )
[64]	401	time.sleep( TORQUE_POLL_INTERVAL )
[26]	402
	403	def printTime( ):
[65]	404	"""Print current time/date in human readable format for log/debug"""
[26]	405
	406	return time.strftime("%a, %d %b %Y %H:%M:%S")
	407
	408	def debug_msg( level, msg ):
[65]	409	"""Print msg if at or above current debug level"""
[26]	410
	411	if (DEBUG_LEVEL >= level):
	412	sys.stderr.write( msg + '\n' )
	413
[23]	414	def main():
[65]	415	"""Application start"""
[23]	416
	417	gather = PBSDataGatherer()
[26]	418	if DAEMONIZE:
	419	gather.daemon()
	420	else:
	421	gather.run()
[23]	422
[65]	423	# w00t someone started me
	424	#
[23]	425	if __name__ == '__main__':
	426	main()

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

Download in other formats: