Context Navigation

source: trunk/jobmond/jobmond.py @ 282

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

jobmond/jobmond.py:

applied patch that prevents jobmond from exiting when Torque server is unavailable. Thanks to Peter Kruse <pk@…>

Property svn:keywords set to Id

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

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

Download in other formats: