Context Navigation

source: trunk/jobmond/jobmond.py @ 317

Last change on this file since 317 was 317, checked in by bastiaans, 17 years ago

jobmond/jobmond.conf:

added QUEUE option

jobmond/jobmond.py:

implemented QUEUE option (thanks to Michael Jeanson)

Property svn:keywords set to Id

File size: 15.5 KB

Rev	Line
[23]	1	#!/usr/bin/env python
[225]	2	#
	3	# This file is part of Jobmonarch
	4	#
	5	# Copyright (C) 2006 Ramon Bastiaans
	6	#
	7	# Jobmonarch is free software; you can redistribute it and/or modify
	8	# it under the terms of the GNU General Public License as published by
	9	# the Free Software Foundation; either version 2 of the License, or
	10	# (at your option) any later version.
	11	#
	12	# Jobmonarch is distributed in the hope that it will be useful,
	13	# but WITHOUT ANY WARRANTY; without even the implied warranty of
	14	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	15	# GNU General Public License for more details.
	16	#
	17	# You should have received a copy of the GNU General Public License
	18	# along with this program; if not, write to the Free Software
	19	# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
	20	#
[228]	21	# SVN $Id: jobmond.py 317 2007-04-18 13:49:27Z bastiaans $
[227]	22	#
[23]	23
[212]	24	import sys, getopt, ConfigParser
[26]	25
[307]	26	def usage():
	27
	28	print
	29	print 'usage: jobmond [options]'
	30	print 'options:'
	31	print ' --config, -c configuration file'
	32	print ' --pidfile, -p pid file'
	33	print ' --help, -h help'
	34	print
	35
	36
[212]	37	def processArgs( args ):
[26]	38
[212]	39	SHORT_L = 'c:'
	40	LONG_L = 'config='
[165]	41
[307]	42	global PIDFILE
	43	PIDFILE = None
	44	config_filename = '/etc/jobmond.conf'
[61]	45
[212]	46	try:
[68]	47
[212]	48	opts, args = getopt.getopt( args, SHORT_L, LONG_L )
[185]	49
[307]	50	except getopt.GetoptError, detail:
[212]	51
	52	print detail
[307]	53	usage()
[212]	54	sys.exit(1)
	55
	56	for opt, value in opts:
	57
	58	if opt in [ '--config', '-c' ]:
	59
	60	config_filename = value
	61
[307]	62	if opt in [ '--pidfile', '-p' ]:
[212]	63
[307]	64	PIDFILE = value
	65
	66	if opt in [ '--help', '-h' ]:
	67
	68	usage()
	69	sys.exit(1)
[212]	70
	71	return loadConfig( config_filename )
	72
	73	def loadConfig( filename ):
	74
[215]	75	def getlist( cfg_string ):
	76
	77	my_list = [ ]
	78
	79	for item_txt in cfg_string.split( ',' ):
	80
	81	sep_char = None
	82
	83	item_txt = item_txt.strip()
	84
	85	for s_char in [ "'", '"' ]:
	86
	87	if item_txt.find( s_char ) != -1:
	88
	89	if item_txt.count( s_char ) != 2:
	90
	91	print 'Missing quote: %s' %item_txt
	92	sys.exit( 1 )
	93
	94	else:
	95
	96	sep_char = s_char
	97	break
	98
	99	if sep_char:
	100
	101	item_txt = item_txt.split( sep_char )[1]
	102
	103	my_list.append( item_txt )
	104
	105	return my_list
	106
[212]	107	cfg = ConfigParser.ConfigParser()
	108
	109	cfg.read( filename )
	110
[317]	111	global DEBUG_LEVEL, DAEMONIZE, BATCH_SERVER, BATCH_POLL_INTERVAL, GMOND_CONF, DETECT_TIME_DIFFS, BATCH_HOST_TRANSLATE, BATCH_API, QUEUE
[212]	112
	113	DEBUG_LEVEL = cfg.getint( 'DEFAULT', 'DEBUG_LEVEL' )
	114
	115	DAEMONIZE = cfg.getboolean( 'DEFAULT', 'DAEMONIZE' )
	116
[265]	117	try:
[212]	118
[265]	119	BATCH_SERVER = cfg.get( 'DEFAULT', 'BATCH_SERVER' )
[212]	120
[265]	121	except ConfigParser.NoOptionError:
	122
	123	# Backwards compatibility for old configs
	124	#
	125
	126	BATCH_SERVER = cfg.get( 'DEFAULT', 'TORQUE_SERVER' )
[266]	127	api_guess = 'pbs'
[265]	128
	129	try:
	130
	131	BATCH_POLL_INTERVAL = cfg.getint( 'DEFAULT', 'BATCH_POLL_INTERVAL' )
	132
	133	except ConfigParser.NoOptionError:
	134
	135	# Backwards compatibility for old configs
	136	#
	137
	138	BATCH_POLL_INTERVAL = cfg.getint( 'DEFAULT', 'TORQUE_POLL_INTERVAL' )
[266]	139	api_guess = 'pbs'
	140
[212]	141	GMOND_CONF = cfg.get( 'DEFAULT', 'GMOND_CONF' )
	142
	143	DETECT_TIME_DIFFS = cfg.getboolean( 'DEFAULT', 'DETECT_TIME_DIFFS' )
	144
[215]	145	BATCH_HOST_TRANSLATE = getlist( cfg.get( 'DEFAULT', 'BATCH_HOST_TRANSLATE' ) )
	146
[266]	147	try:
[256]	148
[266]	149	BATCH_API = cfg.get( 'DEFAULT', 'BATCH_API' )
	150
	151	except ConfigParser.NoOptionError, detail:
	152
	153	if BATCH_SERVER and api_guess:
	154	BATCH_API = api_guess
	155	else:
	156	debug_msg( 0, "fatal error: BATCH_API not set and can't make guess" )
	157	sys.exit( 1 )
[317]	158
	159	try:
	160
	161	QUEUE = cfg.getlist( 'DEFAULT', 'QUEUE' )
	162
	163	except ConfigParser.NoOptionError, detail:
	164
	165	QUEUE = None
[266]	166
[212]	167	return True
	168
[23]	169
[215]	170	import time, os, socket, string, re
[212]	171
[253]	172	METRIC_MAX_VAL_LEN = 900
	173
[61]	174	class DataProcessor:
[68]	175	"""Class for processing of data"""
[61]	176
	177	binary = '/usr/bin/gmetric'
	178
	179	def __init__( self, binary=None ):
[68]	180	"""Remember alternate binary location if supplied"""
[61]	181
	182	if binary:
	183	self.binary = binary
	184
[80]	185	# Timeout for XML
	186	#
	187	# From ganglia's documentation:
	188	#
	189	# 'A metric will be deleted DMAX seconds after it is received, and
	190	# DMAX=0 means eternal life.'
[61]	191
[256]	192	self.dmax = str( int( int( BATCH_POLL_INTERVAL ) * 2 ) )
[80]	193
[68]	194	try:
	195	gmond_file = GMOND_CONF
	196
	197	except NameError:
	198	gmond_file = '/etc/gmond.conf'
	199
	200	if not os.path.exists( gmond_file ):
	201	debug_msg( 0, gmond_file + ' does not exist' )
	202	sys.exit( 1 )
	203
[69]	204	incompatible = self.checkGmetricVersion()
[61]	205
[65]	206	if incompatible:
	207	debug_msg( 0, 'Gmetric version not compatible, pls upgrade to at least 3.0.1' )
	208	sys.exit( 1 )
	209
	210	def checkGmetricVersion( self ):
[68]	211	"""
	212	Check version of gmetric is at least 3.0.1
	213	for the syntax we use
	214	"""
[65]	215
[255]	216	global METRIC_MAX_VAL_LEN
	217
[65]	218	for line in os.popen( self.binary + ' --version' ).readlines():
	219
	220	line = line.split( ' ' )
	221
[69]	222	if len( line ) == 2 and str(line).find( 'gmetric' ) != -1:
[65]	223
[69]	224	gmetric_version = line[1].split( '\n' )[0]
[65]	225
[69]	226	version_major = int( gmetric_version.split( '.' )[0] )
	227	version_minor = int( gmetric_version.split( '.' )[1] )
	228	version_patch = int( gmetric_version.split( '.' )[2] )
[65]	229
	230	incompatible = 0
	231
	232	if version_major < 3:
	233
	234	incompatible = 1
	235
	236	elif version_major == 3:
	237
	238	if version_minor == 0:
	239
	240	if version_patch < 1:
	241
[91]	242	incompatible = 1
[65]	243
[255]	244	if version_patch < 3:
	245
	246	METRIC_MAX_VAL_LEN = 900
	247
	248	elif version_patch >= 3:
	249
	250	METRIC_MAX_VAL_LEN = 1400
	251
[65]	252	return incompatible
	253
[75]	254	def multicastGmetric( self, metricname, metricval, valtype='string' ):
[68]	255	"""Call gmetric binary and multicast"""
[65]	256
	257	cmd = self.binary
	258
[61]	259	try:
	260	cmd = cmd + ' -c' + GMOND_CONF
	261	except NameError:
[64]	262	debug_msg( 10, 'Assuming /etc/gmond.conf for gmetric cmd (ommitting)' )
[61]	263
[168]	264	cmd = cmd + ' -n' + str( metricname )+ ' -v"' + str( metricval )+ '" -t' + str( valtype ) + ' -d' + str( self.dmax )
[61]	265
[101]	266	debug_msg( 10, printTime() + ' ' + cmd )
[69]	267	os.system( cmd )
[61]	268
[256]	269	class SgeDataGatherer:
[23]	270
[256]	271	"""Placeholder for Babu Sundaram's SGE implementation"""
	272
	273	def daemon( self ):
	274
	275	pass
	276
	277	def run( self ):
	278
	279	pass
	280
	281	class PbsDataGatherer:
	282
	283	"""This is the DataGatherer for PBS and Torque"""
	284
[61]	285	jobs = { }
	286
[256]	287	global PBSQuery
	288
[23]	289	def __init__( self ):
[68]	290	"""Setup appropriate variables"""
[23]	291
[26]	292	self.jobs = { }
[185]	293	self.timeoffset = 0
[61]	294	self.dp = DataProcessor()
[91]	295	self.initPbsQuery()
[23]	296
[91]	297	def initPbsQuery( self ):
	298
	299	self.pq = None
[256]	300	if( BATCH_SERVER ):
	301	self.pq = PBSQuery( BATCH_SERVER )
[174]	302	else:
[165]	303	self.pq = PBSQuery()
[91]	304
[26]	305	def getAttr( self, attrs, name ):
[68]	306	"""Return certain attribute from dictionary, if exists"""
[26]	307
	308	if attrs.has_key( name ):
	309	return attrs[name]
	310	else:
	311	return ''
	312
	313	def jobDataChanged( self, jobs, job_id, attrs ):
[68]	314	"""Check if job with attrs and job_id in jobs has changed"""
[26]	315
	316	if jobs.has_key( job_id ):
	317	oldData = jobs[ job_id ]
	318	else:
	319	return 1
	320
	321	for name, val in attrs.items():
	322
	323	if oldData.has_key( name ):
	324
	325	if oldData[ name ] != attrs[ name ]:
	326
	327	return 1
	328
	329	else:
	330	return 1
	331
	332	return 0
	333
[65]	334	def getJobData( self, known_jobs ):
[68]	335	"""Gather all data on current jobs in Torque"""
[26]	336
[65]	337	if len( known_jobs ) > 0:
	338	jobs = known_jobs
	339	else:
	340	jobs = { }
[26]	341
[101]	342	#self.initPbsQuery()
[125]	343
	344	#print self.pq.getnodes()
	345
[282]	346	joblist = {}
	347	while len(joblist) == 0:
	348	try:
	349	joblist = self.pq.getjobs()
	350	except PBSError:
	351	time.sleep( TORQUE_POLL_INTERVAL )
[26]	352
[69]	353	self.cur_time = time.time()
[68]	354
[26]	355	jobs_processed = [ ]
	356
[125]	357	#self.printJobs( joblist )
	358
[26]	359	for name, attrs in joblist.items():
	360
	361	job_id = name.split( '.' )[0]
	362
	363	jobs_processed.append( job_id )
[61]	364
[26]	365	name = self.getAttr( attrs, 'Job_Name' )
	366	queue = self.getAttr( attrs, 'queue' )
[317]	367
	368	if QUEUE:
	369
	370	if QUEUE != queue:
	371
	372	continue
	373
[26]	374	owner = self.getAttr( attrs, 'Job_Owner' ).split( '@' )[0]
	375	requested_time = self.getAttr( attrs, 'Resource_List.walltime' )
	376	requested_memory = self.getAttr( attrs, 'Resource_List.mem' )
[95]	377
[26]	378	mynoderequest = self.getAttr( attrs, 'Resource_List.nodes' )
[95]	379
[281]	380	ppn = ''
	381
[26]	382	if mynoderequest.find( ':' ) != -1 and mynoderequest.find( 'ppn' ) != -1:
[95]	383
[281]	384	mynoderequest_fields = mynoderequest.split( ':' )
	385
	386	for mynoderequest_field in mynoderequest_fields:
	387
	388	if mynoderequest_field.find( 'ppn' ) != -1:
	389
	390	ppn = mynoderequest_field.split( 'ppn=' )[1]
	391
[26]	392	status = self.getAttr( attrs, 'job_state' )
[25]	393
[243]	394	queued_timestamp = self.getAttr( attrs, 'ctime' )
	395
[95]	396	if status == 'R':
	397	start_timestamp = self.getAttr( attrs, 'mtime' )
	398	nodes = self.getAttr( attrs, 'exec_host' ).split( '+' )
[133]	399
	400	nodeslist = [ ]
	401
	402	for node in nodes:
	403	host = node.split( '/' )[0]
	404
	405	if nodeslist.count( host ) == 0:
[215]	406
	407	for translate_pattern in BATCH_HOST_TRANSLATE:
	408
[220]	409	if translate_pattern.find( '/' ) != -1:
[215]	410
[220]	411	translate_orig = translate_pattern.split( '/' )[1]
	412	translate_new = translate_pattern.split( '/' )[2]
	413
	414	host = re.sub( translate_orig, translate_new, host )
[216]	415
[217]	416	if not host in nodeslist:
[216]	417
	418	nodeslist.append( host )
[133]	419
[185]	420	if DETECT_TIME_DIFFS:
	421
	422	# If a job start if later than our current date,
	423	# that must mean the Torque server's time is later
	424	# than our local time.
	425
	426	if int(start_timestamp) > int( int(self.cur_time) + int(self.timeoffset) ):
	427
	428	self.timeoffset = int( int(start_timestamp) - int(self.cur_time) )
	429
[133]	430	elif status == 'Q':
[95]	431	start_timestamp = ''
[133]	432	count_mynodes = 0
	433	numeric_node = 1
[95]	434
[133]	435	for node in mynoderequest.split( '+' ):
[67]	436
[133]	437	nodepart = node.split( ':' )[0]
[67]	438
[133]	439	for letter in nodepart:
[67]	440
[133]	441	if letter not in string.digits:
	442
	443	numeric_node = 0
	444
	445	if not numeric_node:
	446	count_mynodes = count_mynodes + 1
	447	else:
	448	count_mynodes = count_mynodes + int( nodepart )
	449
[254]	450	nodeslist = str( count_mynodes )
[172]	451	else:
	452	start_timestamp = ''
[173]	453	nodeslist = ''
[133]	454
[26]	455	myAttrs = { }
[253]	456	myAttrs['name'] = str( name )
	457	myAttrs['queue'] = str( queue )
	458	myAttrs['owner'] = str( owner )
	459	myAttrs['requested_time'] = str( requested_time )
	460	myAttrs['requested_memory'] = str( requested_memory )
	461	myAttrs['ppn'] = str( ppn )
	462	myAttrs['status'] = str( status )
	463	myAttrs['start_timestamp'] = str( start_timestamp )
	464	myAttrs['queued_timestamp'] = str( queued_timestamp )
[185]	465	myAttrs['reported'] = str( int( int( self.cur_time ) + int( self.timeoffset ) ) )
[67]	466	myAttrs['nodes'] = nodeslist
	467	myAttrs['domain'] = string.join( socket.getfqdn().split( '.' )[1:], '.' )
[256]	468	myAttrs['poll_interval'] = str( BATCH_POLL_INTERVAL )
[26]	469
[184]	470	if self.jobDataChanged( jobs, job_id, myAttrs ) and myAttrs['status'] in [ 'R', 'Q' ]:
[26]	471	jobs[ job_id ] = myAttrs
[61]	472
[101]	473	#debug_msg( 10, printTime() + ' job %s state changed' %(job_id) )
[26]	474
[76]	475	for id, attrs in jobs.items():
	476
	477	if id not in jobs_processed:
	478
	479	# This one isn't there anymore; toedeledoki!
	480	#
	481	del jobs[ id ]
	482
[65]	483	return jobs
	484
	485	def submitJobData( self, jobs ):
	486	"""Submit job info list"""
	487
[219]	488	self.dp.multicastGmetric( 'MONARCH-HEARTBEAT', str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
[69]	489
[61]	490	# Now let's spread the knowledge
	491	#
	492	for jobid, jobattrs in jobs.items():
	493
[95]	494	gmetric_val = self.compileGmetricVal( jobid, jobattrs )
[61]	495
[253]	496	metric_increment = 0
	497
[95]	498	for val in gmetric_val:
[253]	499	self.dp.multicastGmetric( 'MONARCH-JOB-' + jobid + '-' + str(metric_increment), val )
	500	metric_increment = metric_increment + 1
[61]	501
[253]	502	def compileGmetricVal( self, jobid, jobattrs ):
	503	"""Create a val string for gmetric of jobinfo"""
[67]	504
[253]	505	gval_lists = [ ]
[67]	506
[253]	507	mystr = None
[67]	508
[253]	509	val_list = { }
[67]	510
[253]	511	for val_name, val_value in jobattrs.items():
[61]	512
[253]	513	val_list_names_len = len( string.join( val_list.keys() ) ) + len(val_list.keys())
	514	val_list_vals_len = len( string.join( val_list.values() ) ) + len(val_list.values())
[95]	515
[254]	516	if val_name == 'nodes' and jobattrs['status'] == 'R':
[95]	517
[253]	518	node_str = None
[95]	519
[253]	520	for node in val_value:
[65]	521
[253]	522	if node_str:
	523	node_str = node_str + ';' + node
	524	else:
	525	node_str = node
[65]	526
[253]	527	if (val_list_names_len + len(val_name) ) + (val_list_vals_len + len(node_str) ) > METRIC_MAX_VAL_LEN:
[65]	528
[253]	529	val_list[ val_name ] = node_str
	530	gval_lists.append( val_list )
	531	val_list = { }
	532	node_str = None
[65]	533
[253]	534	val_list[ val_name ] = node_str
	535	gval_lists.append( val_list )
	536	val_list = { }
[65]	537
[254]	538	elif val_value != '':
	539
	540	if (val_list_names_len + len(val_name) ) + (val_list_vals_len + len(str(val_value)) ) > METRIC_MAX_VAL_LEN:
	541
	542	gval_lists.append( val_list )
	543	val_list = { }
	544
	545	val_list[ val_name ] = val_value
	546
	547	if len(val_list) > 0:
	548	gval_lists.append( val_list )
	549
[253]	550	str_list = [ ]
[65]	551
[253]	552	for val_list in gval_lists:
[65]	553
[253]	554	my_val_str = None
[65]	555
[253]	556	for val_name, val_value in val_list.items():
[65]	557
[253]	558	if my_val_str:
[65]	559
[253]	560	my_val_str = my_val_str + ' ' + val_name + '=' + val_value
	561	else:
	562	my_val_str = val_name + '=' + val_value
	563
	564	str_list.append( my_val_str )
	565
	566	return str_list
	567
[61]	568	def printJobs( self, jobs ):
[65]	569	"""Print a jobinfo overview"""
	570
[26]	571	for name, attrs in self.jobs.items():
	572
	573	print 'job %s' %(name)
	574
	575	for name, val in attrs.items():
	576
	577	print '\t%s = %s' %( name, val )
	578
[61]	579	def printJob( self, jobs, job_id ):
[65]	580	"""Print job with job_id from jobs"""
[26]	581
	582	print 'job %s' %(job_id)
	583
[65]	584	for name, val in jobs[ job_id ].items():
[26]	585
	586	print '\t%s = %s' %( name, val )
	587
	588	def daemon( self ):
[65]	589	"""Run as daemon forever"""
[26]	590
	591	# Fork the first child
	592	#
	593	pid = os.fork()
	594	if pid > 0:
[212]	595	sys.exit(0) # end parent
[26]	596
	597	# creates a session and sets the process group ID
	598	#
	599	os.setsid()
	600
	601	# Fork the second child
	602	#
	603	pid = os.fork()
	604	if pid > 0:
[212]	605	sys.exit(0) # end parent
[26]	606
[307]	607	write_pidfile()
	608
[26]	609	# Go to the root directory and set the umask
	610	#
	611	os.chdir('/')
	612	os.umask(0)
	613
	614	sys.stdin.close()
	615	sys.stdout.close()
	616	sys.stderr.close()
	617
[272]	618	os.open('/dev/null', os.O_RDWR)
	619	os.dup2(0, 1)
	620	os.dup2(0, 2)
[26]	621
	622	self.run()
	623
	624	def run( self ):
[65]	625	"""Main thread"""
[26]	626
	627	while ( 1 ):
	628
[65]	629	self.jobs = self.getJobData( self.jobs )
	630	self.submitJobData( self.jobs )
[256]	631	time.sleep( BATCH_POLL_INTERVAL )
[26]	632
	633	def printTime( ):
[65]	634	"""Print current time/date in human readable format for log/debug"""
[26]	635
	636	return time.strftime("%a, %d %b %Y %H:%M:%S")
	637
	638	def debug_msg( level, msg ):
[65]	639	"""Print msg if at or above current debug level"""
[26]	640
	641	if (DEBUG_LEVEL >= level):
	642	sys.stderr.write( msg + '\n' )
	643
[307]	644	def write_pidfile():
	645
	646	# Write pidfile if PIDFILE exists
	647	if PIDFILE:
	648	pid = os.getpid()
	649
	650	pidfile = open(PIDFILE, 'w')
	651	pidfile.write(str(pid))
	652	pidfile.close()
	653
[23]	654	def main():
[65]	655	"""Application start"""
[23]	656
[256]	657	global PBSQuery
	658
[212]	659	if not processArgs( sys.argv[1:] ):
	660	sys.exit( 1 )
	661
[256]	662	if BATCH_API == 'pbs':
	663
	664	try:
[282]	665	from PBSQuery import PBSQuery, PBSError
[256]	666
	667	except ImportError:
	668
	669	debug_msg( 0, "fatal error: BATCH_API set to 'pbs' but python module 'pbs_python' is not installed" )
	670	sys.exit( 1 )
	671
	672	gather = PbsDataGatherer()
	673
	674	elif BATCH_API == 'sge':
	675
	676	pass
	677	# import Babu's code here
	678	#
	679	#try:
	680	# import sge_drmaa
	681	#
	682	#except ImportError:
	683	#
[257]	684	# debug_msg( 0, "fatal error: BATCH_API set to 'sge' but python module 'sge_drmaa' is not installed' )
[256]	685	# sys.exit( 1 )
	686	#
	687	#gather = SgeDataGatherer()
	688
	689	else:
	690	debug_msg( 0, "fatal error: unknown BATCH_API '" + BATCH_API + "' is not supported" )
	691	sys.exit( 1 )
	692
[26]	693	if DAEMONIZE:
	694	gather.daemon()
	695	else:
	696	gather.run()
[23]	697
[256]	698	# wh00t? someone started me! :)
[65]	699	#
[23]	700	if __name__ == '__main__':
	701	main()

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

Download in other formats: