Context Navigation

source: trunk/jobmond/jobmond.py @ 512

Last change on this file since 512 was 512, checked in by bastiaans, 16 years ago

web/addons/job_monarch/templates/overview.tpl,
web/addons/job_monarch/overview.php,
jobmond/jobmond.py:

added Craig West's support for offline/down PBS nodes

Property svn:executable set to *
Property svn:keywords set to Id

File size: 33.2 KB

Rev	Line
[23]	1	#!/usr/bin/env python
[225]	2	#
	3	# This file is part of Jobmonarch
	4	#
[363]	5	# Copyright (C) 2006-2007 Ramon Bastiaans
[507]	6	# Copyright (C) 2007 Dave Love (SGE code)
[225]	7	#
	8	# Jobmonarch is free software; you can redistribute it and/or modify
	9	# it under the terms of the GNU General Public License as published by
	10	# the Free Software Foundation; either version 2 of the License, or
	11	# (at your option) any later version.
	12	#
	13	# Jobmonarch is distributed in the hope that it will be useful,
	14	# but WITHOUT ANY WARRANTY; without even the implied warranty of
	15	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	16	# GNU General Public License for more details.
	17	#
	18	# You should have received a copy of the GNU General Public License
	19	# along with this program; if not, write to the Free Software
	20	# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
	21	#
[228]	22	# SVN $Id: jobmond.py 512 2008-03-07 16:58:33Z bastiaans $
[227]	23	#
[23]	24
[471]	25	import sys, getopt, ConfigParser, time, os, socket, string, re
	26	import xdrlib, socket, syslog, xml, xml.sax
[318]	27	from xml.sax.handler import feature_namespaces
	28
[500]	29	VERSION='0.3.1'
[307]	30
[471]	31	def usage( ver ):
	32
	33	print 'jobmond %s' %VERSION
	34
	35	if ver:
	36	return 0
	37
[307]	38	print
[471]	39	print 'Purpose:'
	40	print ' The Job Monitoring Daemon (jobmond) reports batch jobs information and statistics'
	41	print ' to Ganglia, which can be viewed with Job Monarch web frontend'
[307]	42	print
[471]	43	print 'Usage: jobmond [OPTIONS]'
	44	print
	45	print ' -c, --config=FILE The configuration file to use (default: /etc/jobmond.conf)'
	46	print ' -p, --pidfile=FILE Use pid file to store the process id'
	47	print ' -h, --help Print help and exit'
	48	print ' -v, --version Print version and exit'
	49	print
[307]	50
[212]	51	def processArgs( args ):
[26]	52
[471]	53	SHORT_L = 'p:hvc:'
	54	LONG_L = [ 'help', 'config=', 'pidfile=', 'version' ]
[165]	55
[307]	56	global PIDFILE
[354]	57	PIDFILE = None
[61]	58
[354]	59	config_filename = '/etc/jobmond.conf'
	60
[212]	61	try:
[68]	62
[354]	63	opts, args = getopt.getopt( args, SHORT_L, LONG_L )
[185]	64
[307]	65	except getopt.GetoptError, detail:
[212]	66
	67	print detail
[307]	68	usage()
[354]	69	sys.exit( 1 )
[212]	70
	71	for opt, value in opts:
	72
	73	if opt in [ '--config', '-c' ]:
	74
[354]	75	config_filename = value
[212]	76
[307]	77	if opt in [ '--pidfile', '-p' ]:
[212]	78
[354]	79	PIDFILE = value
[307]	80
	81	if opt in [ '--help', '-h' ]:
	82
[471]	83	usage( False )
[354]	84	sys.exit( 0 )
[212]	85
[476]	86	if opt in [ '--version', '-v' ]:
[471]	87
[476]	88	usage( True )
[471]	89	sys.exit( 0 )
	90
[212]	91	return loadConfig( config_filename )
	92
	93	def loadConfig( filename ):
	94
[215]	95	def getlist( cfg_string ):
	96
	97	my_list = [ ]
	98
	99	for item_txt in cfg_string.split( ',' ):
	100
	101	sep_char = None
	102
	103	item_txt = item_txt.strip()
	104
	105	for s_char in [ "'", '"' ]:
	106
	107	if item_txt.find( s_char ) != -1:
	108
	109	if item_txt.count( s_char ) != 2:
	110
	111	print 'Missing quote: %s' %item_txt
	112	sys.exit( 1 )
	113
	114	else:
	115
	116	sep_char = s_char
	117	break
	118
	119	if sep_char:
	120
	121	item_txt = item_txt.split( sep_char )[1]
	122
	123	my_list.append( item_txt )
	124
	125	return my_list
	126
[354]	127	cfg = ConfigParser.ConfigParser()
[212]	128
	129	cfg.read( filename )
	130
[354]	131	global DEBUG_LEVEL, DAEMONIZE, BATCH_SERVER, BATCH_POLL_INTERVAL
	132	global GMOND_CONF, DETECT_TIME_DIFFS, BATCH_HOST_TRANSLATE
[373]	133	global BATCH_API, QUEUE, GMETRIC_TARGET, USE_SYSLOG
[449]	134	global SYSLOG_LEVEL, SYSLOG_FACILITY, GMETRIC_BINARY
[212]	135
[354]	136	DEBUG_LEVEL = cfg.getint( 'DEFAULT', 'DEBUG_LEVEL' )
[212]	137
[354]	138	DAEMONIZE = cfg.getboolean( 'DEFAULT', 'DAEMONIZE' )
[212]	139
[377]	140	SYSLOG_LEVEL = -1
	141	SYSLOG_FACILITY = None
	142
[265]	143	try:
[373]	144	USE_SYSLOG = cfg.getboolean( 'DEFAULT', 'USE_SYSLOG' )
[212]	145
[373]	146	except ConfigParser.NoOptionError:
	147
	148	USE_SYSLOG = True
	149
	150	debug_msg( 0, 'ERROR: no option USE_SYSLOG found: assuming yes' )
	151
[377]	152
[449]	153
[373]	154	if USE_SYSLOG:
	155
	156	try:
	157	SYSLOG_LEVEL = cfg.getint( 'DEFAULT', 'SYSLOG_LEVEL' )
	158
	159	except ConfigParser.NoOptionError:
	160
	161	debug_msg( 0, 'ERROR: no option SYSLOG_LEVEL found: assuming level 0' )
	162	SYSLOG_LEVEL = 0
	163
	164	try:
	165
	166	SYSLOG_FACILITY = eval( 'syslog.LOG_' + cfg.get( 'DEFAULT', 'SYSLOG_FACILITY' ) )
	167
[377]	168	except ConfigParser.NoOptionError:
[373]	169
	170	SYSLOG_FACILITY = syslog.LOG_DAEMON
	171
	172	debug_msg( 0, 'ERROR: no option SYSLOG_FACILITY found: assuming facility DAEMON' )
	173
	174	try:
	175
[354]	176	BATCH_SERVER = cfg.get( 'DEFAULT', 'BATCH_SERVER' )
[212]	177
[265]	178	except ConfigParser.NoOptionError:
	179
	180	# Backwards compatibility for old configs
	181	#
	182
[354]	183	BATCH_SERVER = cfg.get( 'DEFAULT', 'TORQUE_SERVER' )
	184	api_guess = 'pbs'
[265]	185
	186	try:
	187
[354]	188	BATCH_POLL_INTERVAL = cfg.getint( 'DEFAULT', 'BATCH_POLL_INTERVAL' )
[265]	189
	190	except ConfigParser.NoOptionError:
	191
	192	# Backwards compatibility for old configs
	193	#
	194
[354]	195	BATCH_POLL_INTERVAL = cfg.getint( 'DEFAULT', 'TORQUE_POLL_INTERVAL' )
	196	api_guess = 'pbs'
[353]	197
	198	try:
[212]	199
[354]	200	GMOND_CONF = cfg.get( 'DEFAULT', 'GMOND_CONF' )
[353]	201
	202	except ConfigParser.NoOptionError:
	203
[354]	204	GMOND_CONF = None
[353]	205
[449]	206	try:
	207
	208	GMETRIC_BINARY = cfg.get( 'DEFAULT', 'GMETRIC_BINARY' )
	209
	210	except ConfigParser.NoOptionError:
	211
	212	GMETRIC_BINARY = '/usr/bin/gmetric'
	213
[354]	214	DETECT_TIME_DIFFS = cfg.getboolean( 'DEFAULT', 'DETECT_TIME_DIFFS' )
[212]	215
[354]	216	BATCH_HOST_TRANSLATE = getlist( cfg.get( 'DEFAULT', 'BATCH_HOST_TRANSLATE' ) )
[215]	217
[266]	218	try:
[256]	219
[354]	220	BATCH_API = cfg.get( 'DEFAULT', 'BATCH_API' )
[266]	221
	222	except ConfigParser.NoOptionError, detail:
	223
	224	if BATCH_SERVER and api_guess:
[354]	225
	226	BATCH_API = api_guess
[266]	227	else:
[373]	228	debug_msg( 0, "FATAL ERROR: BATCH_API not set and can't make guess" )
[266]	229	sys.exit( 1 )
[317]	230
	231	try:
	232
[354]	233	QUEUE = getlist( cfg.get( 'DEFAULT', 'QUEUE' ) )
[317]	234
	235	except ConfigParser.NoOptionError, detail:
	236
[354]	237	QUEUE = None
[353]	238
	239	try:
	240
[354]	241	GMETRIC_TARGET = cfg.get( 'DEFAULT', 'GMETRIC_TARGET' )
[353]	242
	243	except ConfigParser.NoOptionError:
	244
[354]	245	GMETRIC_TARGET = None
[353]	246
	247	if not GMOND_CONF:
	248
[373]	249	debug_msg( 0, "FATAL ERROR: GMETRIC_TARGET and GMOND_CONF both not set! Set at least one!" )
[353]	250	sys.exit( 1 )
	251	else:
	252
[507]	253	debug_msg( 0, "ERROR: GMETRIC_TARGET not set: internal Gmetric handling aborted. Failing back to DEPRECATED use of gmond.conf/gmetric binary. This will slow down jobmond significantly!" )
[353]	254
[212]	255	return True
	256
[507]	257	def fqdn_parts (fqdn):
	258	"""Return pair of host and domain for fully-qualified domain name arg."""
	259	parts = fqdn.split (".")
	260	return (parts[0], string.join(parts[1:], "."))
	261
[253]	262	METRIC_MAX_VAL_LEN = 900
	263
[61]	264	class DataProcessor:
[355]	265
[68]	266	"""Class for processing of data"""
[61]	267
[449]	268	binary = None
[61]	269
	270	def __init__( self, binary=None ):
[355]	271
[68]	272	"""Remember alternate binary location if supplied"""
[61]	273
[449]	274	global GMETRIC_BINARY
	275
[61]	276	if binary:
	277	self.binary = binary
	278
[449]	279	if not self.binary:
	280	self.binary = GMETRIC_BINARY
	281
[80]	282	# Timeout for XML
	283	#
	284	# From ganglia's documentation:
	285	#
	286	# 'A metric will be deleted DMAX seconds after it is received, and
	287	# DMAX=0 means eternal life.'
[61]	288
[256]	289	self.dmax = str( int( int( BATCH_POLL_INTERVAL ) * 2 ) )
[80]	290
[353]	291	if GMOND_CONF:
[354]	292
[353]	293	try:
	294	gmond_file = GMOND_CONF
[68]	295
[353]	296	except NameError:
	297	gmond_file = '/etc/gmond.conf'
[68]	298
[353]	299	if not os.path.exists( gmond_file ):
[373]	300	debug_msg( 0, 'FATAL ERROR: ' + gmond_file + ' does not exist' )
[353]	301	sys.exit( 1 )
[68]	302
[353]	303	incompatible = self.checkGmetricVersion()
[61]	304
[353]	305	if incompatible:
[355]	306
	307	debug_msg( 0, 'Gmetric version not compatible, please upgrade to at least 3.0.1' )
[353]	308	sys.exit( 1 )
[65]	309
	310	def checkGmetricVersion( self ):
[355]	311
[68]	312	"""
	313	Check version of gmetric is at least 3.0.1
	314	for the syntax we use
	315	"""
[65]	316
[255]	317	global METRIC_MAX_VAL_LEN
	318
[341]	319	incompatible = 0
	320
[355]	321	gfp = os.popen( self.binary + ' --version' )
	322	lines = gfp.readlines()
[65]	323
[355]	324	gfp.close()
	325
	326	for line in lines:
	327
[65]	328	line = line.split( ' ' )
	329
[355]	330	if len( line ) == 2 and str( line ).find( 'gmetric' ) != -1:
[65]	331
[355]	332	gmetric_version = line[1].split( '\n' )[0]
[65]	333
[355]	334	version_major = int( gmetric_version.split( '.' )[0] )
	335	version_minor = int( gmetric_version.split( '.' )[1] )
	336	version_patch = int( gmetric_version.split( '.' )[2] )
[65]	337
[355]	338	incompatible = 0
[65]	339
	340	if version_major < 3:
	341
	342	incompatible = 1
	343
	344	elif version_major == 3:
	345
	346	if version_minor == 0:
	347
	348	if version_patch < 1:
	349
[91]	350	incompatible = 1
[65]	351
[471]	352	# Gmetric 3.0.1 >< 3.0.3 had a bug in the max metric length
	353	#
[255]	354	if version_patch < 3:
	355
	356	METRIC_MAX_VAL_LEN = 900
	357
	358	elif version_patch >= 3:
	359
	360	METRIC_MAX_VAL_LEN = 1400
	361
[65]	362	return incompatible
	363
[409]	364	def multicastGmetric( self, metricname, metricval, valtype='string', units='' ):
[355]	365
[68]	366	"""Call gmetric binary and multicast"""
[65]	367
	368	cmd = self.binary
	369
[353]	370	if GMETRIC_TARGET:
[61]	371
[353]	372	GMETRIC_TARGET_HOST = GMETRIC_TARGET.split( ':' )[0]
	373	GMETRIC_TARGET_PORT = GMETRIC_TARGET.split( ':' )[1]
	374
	375	metric_debug = "[gmetric] name: %s - val: %s - dmax: %s" %( str( metricname ), str( metricval ), str( self.dmax ) )
	376
	377	debug_msg( 10, printTime() + ' ' + metric_debug)
	378
	379	gm = Gmetric( GMETRIC_TARGET_HOST, GMETRIC_TARGET_PORT )
	380
[425]	381	gm.send( str( metricname ), str( metricval ), str( self.dmax ), valtype, units )
[353]	382
	383	else:
	384	try:
	385	cmd = cmd + ' -c' + GMOND_CONF
	386
	387	except NameError:
	388
[507]	389	debug_msg( 10, 'Assuming /etc/gmond.conf for gmetric cmd (omitting)' )
[353]	390
	391	cmd = cmd + ' -n' + str( metricname )+ ' -v"' + str( metricval )+ '" -t' + str( valtype ) + ' -d' + str( self.dmax )
	392
[409]	393	if len( units ) > 0:
	394
	395	cmd = cmd + ' -u"' + units + '"'
	396
[353]	397	debug_msg( 10, printTime() + ' ' + cmd )
	398
	399	os.system( cmd )
	400
[318]	401	class DataGatherer:
[23]	402
[318]	403	"""Skeleton class for batch system DataGatherer"""
[256]	404
[318]	405	def printJobs( self, jobs ):
[355]	406
[318]	407	"""Print a jobinfo overview"""
	408
	409	for name, attrs in self.jobs.items():
	410
	411	print 'job %s' %(name)
	412
	413	for name, val in attrs.items():
	414
	415	print '\t%s = %s' %( name, val )
	416
	417	def printJob( self, jobs, job_id ):
[355]	418
[318]	419	"""Print job with job_id from jobs"""
	420
	421	print 'job %s' %(job_id)
	422
	423	for name, val in jobs[ job_id ].items():
	424
	425	print '\t%s = %s' %( name, val )
	426
[507]	427	def getAttr( self, attrs, name ):
	428
	429	"""Return certain attribute from dictionary, if exists"""
	430
	431	if attrs.has_key( name ):
	432
	433	return attrs[ name ]
	434	else:
	435	return ''
	436
	437	def jobDataChanged( self, jobs, job_id, attrs ):
	438
	439	"""Check if job with attrs and job_id in jobs has changed"""
	440
	441	if jobs.has_key( job_id ):
	442
	443	oldData = jobs[ job_id ]
	444	else:
	445	return 1
	446
	447	for name, val in attrs.items():
	448
	449	if oldData.has_key( name ):
	450
	451	if oldData[ name ] != attrs[ name ]:
	452
	453	return 1
	454
	455	else:
	456	return 1
	457
	458	return 0
	459
	460	def submitJobData( self ):
	461
	462	"""Submit job info list"""
	463
[512]	464	global BATCH_API
	465
[507]	466	self.dp.multicastGmetric( 'MONARCH-HEARTBEAT', str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
	467
	468	running_jobs = 0
	469	queued_jobs = 0
	470
	471	# Count how many running/queued jobs we found
	472	#
	473	for jobid, jobattrs in self.jobs.items():
	474
	475	if jobattrs[ 'status' ] == 'Q':
	476
	477	queued_jobs += 1
	478
	479	elif jobattrs[ 'status' ] == 'R':
	480
	481	running_jobs += 1
	482
	483	# Report running/queued jobs as seperate metric for a nice RRD graph
	484	#
	485	self.dp.multicastGmetric( 'MONARCH-RJ', str( running_jobs ), 'uint32', 'jobs' )
	486	self.dp.multicastGmetric( 'MONARCH-QJ', str( queued_jobs ), 'uint32', 'jobs' )
	487
[512]	488	# Report down/offline nodes in batch (PBS only ATM)
	489	#
	490	if BATCH_API == 'pbs':
	491
	492	downed_nodes = list()
	493	offline_nodes = list()
	494
	495	l = ['state']
	496
	497	for name, node in self.pq.getnodes().items():
	498
	499	if ( node[ 'state' ].find( "down" ) != -1 ):
	500
	501	downed_nodes.append( name )
	502
	503	if ( node[ 'state' ].find( "offline" ) != -1 ):
	504
	505	offline_nodes.append( name )
	506
	507	self.dp.multicastGmetric( 'MONARCH-DOWN' , str( downed_nodes ), 'uint32', 'jobs' )
	508	self.dp.multicastGmetric( 'MONARCH-OFFLINE', str( offline_nodes ), 'uint32', 'jobs' )
	509
[507]	510	# Now let's spread the knowledge
	511	#
	512	for jobid, jobattrs in self.jobs.items():
	513
	514	# Make gmetric values for each job: respect max gmetric value length
	515	#
	516	gmetric_val = self.compileGmetricVal( jobid, jobattrs )
	517	metric_increment = 0
	518
	519	# If we have more job info than max gmetric value length allows, split it up
	520	# amongst multiple metrics
	521	#
	522	for val in gmetric_val:
	523
	524	self.dp.multicastGmetric( 'MONARCH-JOB-' + jobid + '-' + str(metric_increment), val )
	525
	526	# Increase follow number if this jobinfo is split up amongst more than 1 gmetric
	527	#
	528	metric_increment = metric_increment + 1
	529
	530	def compileGmetricVal( self, jobid, jobattrs ):
	531
	532	"""Create a val string for gmetric of jobinfo"""
	533
	534	gval_lists = [ ]
	535	val_list = { }
	536
	537	for val_name, val_value in jobattrs.items():
	538
	539	# These are our own metric names, i.e.: status, start_timestamp, etc
	540	#
	541	val_list_names_len = len( string.join( val_list.keys() ) ) + len(val_list.keys())
	542
	543	# These are their corresponding values
	544	#
	545	val_list_vals_len = len( string.join( val_list.values() ) ) + len(val_list.values())
	546
	547	if val_name == 'nodes' and jobattrs['status'] == 'R':
	548
	549	node_str = None
	550
	551	for node in val_value:
	552
	553	if node_str:
	554
	555	node_str = node_str + ';' + node
	556	else:
	557	node_str = node
	558
	559	# Make sure if we add this new info, that the total metric's value length does not exceed METRIC_MAX_VAL_LEN
	560	#
	561	if (val_list_names_len + len(val_name) ) + (val_list_vals_len + len(node_str) ) > METRIC_MAX_VAL_LEN:
	562
	563	# It's too big, we need to make a new gmetric for the additional info
	564	#
	565	val_list[ val_name ] = node_str
	566
	567	gval_lists.append( val_list )
	568
	569	val_list = { }
	570	node_str = None
	571
	572	val_list[ val_name ] = node_str
	573
	574	gval_lists.append( val_list )
	575
	576	val_list = { }
	577
	578	elif val_value != '':
	579
	580	# Make sure if we add this new info, that the total metric's value length does not exceed METRIC_MAX_VAL_LEN
	581	#
	582	if (val_list_names_len + len(val_name) ) + (val_list_vals_len + len(str(val_value)) ) > METRIC_MAX_VAL_LEN:
	583
	584	# It's too big, we need to make a new gmetric for the additional info
	585	#
	586	gval_lists.append( val_list )
	587
	588	val_list = { }
	589
	590	val_list[ val_name ] = val_value
	591
	592	if len( val_list ) > 0:
	593
	594	gval_lists.append( val_list )
	595
	596	str_list = [ ]
	597
	598	# Now append the value names and values together, i.e.: stop_timestamp=value, etc
	599	#
	600	for val_list in gval_lists:
	601
	602	my_val_str = None
	603
	604	for val_name, val_value in val_list.items():
	605
	606	if my_val_str:
	607
	608	my_val_str = my_val_str + ' ' + val_name + '=' + val_value
	609	else:
	610	my_val_str = val_name + '=' + val_value
	611
	612	str_list.append( my_val_str )
	613
	614	return str_list
	615
[256]	616	def daemon( self ):
[355]	617
[318]	618	"""Run as daemon forever"""
[256]	619
[318]	620	# Fork the first child
	621	#
	622	pid = os.fork()
	623	if pid > 0:
	624	sys.exit(0) # end parent
[256]	625
[318]	626	# creates a session and sets the process group ID
	627	#
	628	os.setsid()
	629
	630	# Fork the second child
	631	#
	632	pid = os.fork()
	633	if pid > 0:
	634	sys.exit(0) # end parent
	635
	636	write_pidfile()
	637
	638	# Go to the root directory and set the umask
	639	#
	640	os.chdir('/')
	641	os.umask(0)
	642
	643	sys.stdin.close()
	644	sys.stdout.close()
	645	sys.stderr.close()
	646
	647	os.open('/dev/null', os.O_RDWR)
	648	os.dup2(0, 1)
	649	os.dup2(0, 2)
	650
	651	self.run()
	652
[256]	653	def run( self ):
[355]	654
[318]	655	"""Main thread"""
[256]	656
[318]	657	while ( 1 ):
	658
[348]	659	self.getJobData()
	660	self.submitJobData()
[318]	661	time.sleep( BATCH_POLL_INTERVAL )
[256]	662
[507]	663	# SGE code by Dave Love <fx@gnu.org>. Tested with SGE 6.0u8 and 6.0u11.
	664	# Probably needs modification for SGE 6.1. See also the fixmes.
[256]	665
[507]	666	class NoJobs (Exception):
	667	"""Exception raised by empty job list in qstat output."""
	668	pass
[256]	669
[507]	670	class SgeQstatXMLParser(xml.sax.handler.ContentHandler):
	671	"""SAX handler for XML output from Sun Grid Engine's `qstat'."""
[318]	672
[507]	673	def __init__(self):
	674	self.value = ""
	675	self.joblist = []
	676	self.job = {}
	677	self.queue = ""
	678	self.in_joblist = False
	679	self.lrequest = False
	680	xml.sax.handler.ContentHandler.__init__(self)
[318]	681
[507]	682	# The structure of the output is as follows. Unfortunately
	683	# it's voluminous, and probably doesn't scale to large
	684	# clusters/queues.
[318]	685
[507]	686	# <detailed_job_info xmlns:xsd="http://www.w3.org/2001/XMLSchema">
	687	# <djob_info>
	688	# <qmaster_response> <!-- job -->
	689	# ...
	690	# <JB_ja_template>
	691	# <ulong_sublist>
	692	# ... <!-- start_time, state ... -->
	693	# </ulong_sublist>
	694	# </JB_ja_template>
	695	# <JB_ja_tasks>
	696	# <ulong_sublist>
	697	# ... <!-- task info
	698	# </ulong_sublist>
	699	# ...
	700	# </JB_ja_tasks>
	701	# ...
	702	# </qmaster_response>
	703	# </djob_info>
	704	# <messages>
	705	# ...
[318]	706
[507]	707	# NB. We might treat each task as a separate job, like
	708	# straight qstat output, but the web interface expects jobs to
	709	# be identified by integers, not, say, <job number>.<task>.
[318]	710
[507]	711	# So, I lied. If the job list is empty, we get invalid XML
	712	# like this, which we need to defend against:
[318]	713
[507]	714	# <unknown_jobs xmlns:xsd="http://www.w3.org/2001/XMLSchema">
	715	# <>
	716	# <ST_name>*</ST_name>
	717	# </>
	718	# </unknown_jobs>
[318]	719
[507]	720	def startElement(self, name, attrs):
	721	self.value = ""
	722	if name == "djob_info": # job list
	723	self.in_joblist = True
	724	elif name == "qmaster_response" and self.in_joblist: # job
	725	self.job = {"job_state": "U", "slots": 0,
	726	"nodes": [], "queued_timestamp": "",
	727	"queued_timestamp": "", "queue": "",
	728	"ppn": "0", "RN_max": 0,
	729	# fixme in endElement
	730	"requested_memory": 0, "requested_time": 0
	731	}
	732	self.joblist.append(self.job)
	733	elif name == "qstat_l_requests": # resource request
	734	self.lrequest = True
	735	elif name == "unknown_jobs":
	736	raise NoJobs
[318]	737
[507]	738	def characters(self, ch):
	739	self.value += ch
[318]	740
[507]	741	def endElement(self, name):
	742	"""Snarf job elements contents into job dictionary.
	743	Translate keys if appropriate."""
[318]	744
[507]	745	name_trans = {
	746	"JB_job_number": "number",
	747	"JB_job_name": "name", "JB_owner": "owner",
	748	"queue_name": "queue", "JAT_start_time": "start_timestamp",
	749	"JB_submission_time": "queued_timestamp"
	750	}
	751	value = self.value
[318]	752
[507]	753	if name == "djob_info":
	754	self.in_joblist = False
	755	self.job = {}
	756	elif name == "JAT_master_queue":
	757	self.job["queue"] = value.split("@")[0]
	758	elif name == "JG_qhostname":
	759	if not (value in self.job["nodes"]):
	760	self.job["nodes"].append(value)
	761	elif name == "JG_slots": # slots in use
	762	self.job["slots"] += int(value)
	763	elif name == "RN_max": # requested slots (tasks or parallel)
	764	self.job["RN_max"] = max (self.job["RN_max"],
	765	int(value))
	766	elif name == "JAT_state": # job state (bitwise or)
	767	value = int (value)
	768	# Status values from sge_jobL.h
	769	#define JIDLE 0x00000000
	770	#define JHELD 0x00000010
	771	#define JMIGRATING 0x00000020
	772	#define JQUEUED 0x00000040
	773	#define JRUNNING 0x00000080
	774	#define JSUSPENDED 0x00000100
	775	#define JTRANSFERING 0x00000200
	776	#define JDELETED 0x00000400
	777	#define JWAITING 0x00000800
	778	#define JEXITING 0x00001000
	779	#define JWRITTEN 0x00002000
	780	#define JSUSPENDED_ON_THRESHOLD 0x00010000
	781	#define JFINISHED 0x00010000
	782	if value & 0x80:
	783	self.job["status"] = "R"
	784	elif value & 0x40:
	785	self.job["status"] = "Q"
	786	else:
	787	self.job["status"] = "O" # `other'
	788	elif name == "CE_name" and self.lrequest and self.value in \
	789	("h_cpu", "s_cpu", "cpu", "h_core", "s_core"):
	790	# We're in a container for an interesting resource
	791	# request; record which type.
	792	self.lrequest = self.value
	793	elif name == "CE_doubleval" and self.lrequest:
	794	# if we're in a container for an interesting
	795	# resource request, use the maxmimum of the hard
	796	# and soft requests to record the requested CPU
	797	# or core. Fixme: I'm not sure if this logic is
	798	# right.
	799	if self.lrequest in ("h_core", "s_core"):
	800	self.job["requested_memory"] = \
	801	max (float (value),
	802	self.job["requested_memory"])
	803	# Fixme: Check what cpu means, c.f [hs]_cpu.
	804	elif self.lrequest in ("h_cpu", "s_cpu", "cpu"):
	805	self.job["requested_time"] = \
	806	max (float (value),
	807	self.job["requested_time"])
	808	elif name == "qstat_l_requests":
	809	self.lrequest = False
	810	elif self.job and self.in_joblist:
	811	if name in name_trans:
	812	name = name_trans[name]
	813	self.job[name] = value
[318]	814
[507]	815	# Abstracted from PBS original.
	816	# Fixme: Is it worth (or appropriate for PBS) sorting the result?
	817	def do_nodelist (nodes):
	818	"""Translate node list as appropriate."""
	819	nodeslist = [ ]
	820	my_domain = fqdn_parts(socket.getfqdn())[1]
	821	for node in nodes:
	822	host = node.split( '/' )[0] # not relevant for SGE
	823	h, host_domain = fqdn_parts(host)
	824	if host_domain == my_domain:
	825	host = h
	826	if nodeslist.count( host ) == 0:
	827	for translate_pattern in BATCH_HOST_TRANSLATE:
	828	if translate_pattern.find( '/' ) != -1:
	829	translate_orig = \
	830	translate_pattern.split( '/' )[1]
	831	translate_new = \
	832	translate_pattern.split( '/' )[2]
	833	host = re.sub( translate_orig,
	834	translate_new, host )
	835	if not host in nodeslist:
	836	nodeslist.append( host )
	837	return nodeslist
[318]	838
	839	class SgeDataGatherer(DataGatherer):
	840
[507]	841	jobs = {}
[61]	842
[318]	843	def __init__( self ):
[507]	844	self.jobs = {}
[318]	845	self.timeoffset = 0
	846	self.dp = DataProcessor()
	847
[507]	848	def getJobData( self ):
[318]	849	"""Gather all data on current jobs in SGE"""
	850
[507]	851	import popen2
[318]	852
[507]	853	self.cur_time = 0
	854	queues = ""
	855	if QUEUE: # only for specific queues
	856	# Fixme: assumes queue names don't contain single
	857	# quote or comma. Don't know what the SGE rules are.
	858	queues = " -q '" + string.join (QUEUE, ",") + "'"
	859	# Note the comment in SgeQstatXMLParser about scaling with
	860	# this method of getting data. I haven't found better one.
	861	# Output with args `-xml -ext -f -r' is easier to parse
	862	# in some ways, harder in others, but it doesn't provide
	863	# the submission time, at least.
	864	piping = popen2.Popen3("qstat -u '' -j '' -xml" + queues,
	865	True)
	866	qstatparser = SgeQstatXMLParser()
	867	parse_err = 0
	868	try:
	869	xml.sax.parse(piping.fromchild, qstatparser)
	870	except NoJobs:
	871	pass
	872	except:
	873	parse_err = 1
	874	if piping.wait():
	875	debug_msg(10,
	876	"qstat error, skipping until next polling interval: "
	877	+ piping.childerr.readline())
	878	return None
	879	elif parse_err:
	880	debug_msg(10, "Bad XML output from qstat"())
	881	exit (1)
	882	for f in piping.fromchild, piping.tochild, piping.childerr:
	883	f.close()
[318]	884	self.cur_time = time.time()
[507]	885	jobs_processed = []
	886	for job in qstatparser.joblist:
	887	job_id = job["number"]
	888	if job["status"] in [ 'Q', 'R' ]:
	889	jobs_processed.append(job_id)
	890	if job["status"] == "R":
	891	job["nodes"] = do_nodelist (job["nodes"])
	892	# Fixme: Is this right?
	893	job["ppn"] = float(job["slots"]) / \
	894	len(job["nodes"])
	895	if DETECT_TIME_DIFFS:
	896	# If a job start is later than our
	897	# current date, that must mean
	898	# the SGE server's time is later
	899	# than our local time.
	900	start_timestamp = \
	901	int (job["start_timestamp"])
	902	if start_timestamp > \
	903	int(self.cur_time) + \
	904	int(self.timeoffset):
[318]	905
[507]	906	self.timeoffset = \
	907	start_timestamp - \
	908	int(self.cur_time)
	909	else:
	910	# fixme: Note sure what this should be:
	911	job["ppn"] = job["RN_max"]
	912	job["nodes"] = "1"
[318]	913
[507]	914	myAttrs = {}
	915	for attr in ["name", "queue", "owner",
	916	"requested_time", "status",
	917	"requested_memory", "ppn",
	918	"start_timestamp", "queued_timestamp"]:
	919	myAttrs[attr] = str(job[attr])
	920	myAttrs["nodes"] = job["nodes"]
	921	myAttrs["reported"] = str(int(self.cur_time) + \
	922	int(self.timeoffset))
	923	myAttrs["domain"] = fqdn_parts(socket.getfqdn())[1]
	924	myAttrs["poll_interval"] = str(BATCH_POLL_INTERVAL)
[318]	925
[507]	926	if self.jobDataChanged(self.jobs, job_id, myAttrs) \
	927	and myAttrs["status"] in ["R", "Q"]:
	928	self.jobs[job_id] = myAttrs
	929	for id, attrs in self.jobs.items():
	930	if id not in jobs_processed:
	931	del self.jobs[id]
[318]	932
[355]	933	class PbsDataGatherer( DataGatherer ):
[318]	934
	935	"""This is the DataGatherer for PBS and Torque"""
	936
[256]	937	global PBSQuery
	938
[23]	939	def __init__( self ):
[354]	940
[68]	941	"""Setup appropriate variables"""
[23]	942
[354]	943	self.jobs = { }
	944	self.timeoffset = 0
	945	self.dp = DataProcessor()
	946
[91]	947	self.initPbsQuery()
[23]	948
[91]	949	def initPbsQuery( self ):
	950
[354]	951	self.pq = None
	952
[256]	953	if( BATCH_SERVER ):
[354]	954
	955	self.pq = PBSQuery( BATCH_SERVER )
[174]	956	else:
[354]	957	self.pq = PBSQuery()
[91]	958
[348]	959	def getJobData( self ):
[354]	960
[68]	961	"""Gather all data on current jobs in Torque"""
[26]	962
[354]	963	joblist = {}
[359]	964	self.cur_time = 0
[349]	965
[359]	966	try:
	967	joblist = self.pq.getjobs()
	968	self.cur_time = time.time()
[354]	969
[359]	970	except PBSError, detail:
[354]	971
[359]	972	debug_msg( 10, "Caught PBS unavailable, skipping until next polling interval: " + str( detail ) )
	973	return None
[354]	974
	975	jobs_processed = [ ]
[26]	976
	977	for name, attrs in joblist.items():
[508]	978	display_queue = 1
[354]	979	job_id = name.split( '.' )[0]
[26]	980
[354]	981	name = self.getAttr( attrs, 'Job_Name' )
	982	queue = self.getAttr( attrs, 'queue' )
[317]	983
	984	if QUEUE:
[508]	985	for q in QUEUE:
	986	if q == queue:
	987	display_queue = 1
	988	break
	989	else:
	990	display_queue = 0
	991	continue
	992	if display_queue == 0:
	993	continue
[317]	994
	995
[354]	996	owner = self.getAttr( attrs, 'Job_Owner' ).split( '@' )[0]
	997	requested_time = self.getAttr( attrs, 'Resource_List.walltime' )
	998	requested_memory = self.getAttr( attrs, 'Resource_List.mem' )
[95]	999
[354]	1000	mynoderequest = self.getAttr( attrs, 'Resource_List.nodes' )
[95]	1001
[354]	1002	ppn = ''
[281]	1003
[26]	1004	if mynoderequest.find( ':' ) != -1 and mynoderequest.find( 'ppn' ) != -1:
[95]	1005
[354]	1006	mynoderequest_fields = mynoderequest.split( ':' )
[281]	1007
	1008	for mynoderequest_field in mynoderequest_fields:
	1009
	1010	if mynoderequest_field.find( 'ppn' ) != -1:
	1011
[354]	1012	ppn = mynoderequest_field.split( 'ppn=' )[1]
[281]	1013
[354]	1014	status = self.getAttr( attrs, 'job_state' )
[25]	1015
[450]	1016	if status in [ 'Q', 'R' ]:
	1017
	1018	jobs_processed.append( job_id )
	1019
[354]	1020	queued_timestamp = self.getAttr( attrs, 'ctime' )
[243]	1021
[95]	1022	if status == 'R':
[133]	1023
[354]	1024	start_timestamp = self.getAttr( attrs, 'mtime' )
	1025	nodes = self.getAttr( attrs, 'exec_host' ).split( '+' )
[133]	1026
[507]	1027	nodeslist = do_nodelist( nodes )
[354]	1028
[185]	1029	if DETECT_TIME_DIFFS:
	1030
	1031	# If a job start if later than our current date,
	1032	# that must mean the Torque server's time is later
	1033	# than our local time.
	1034
[354]	1035	if int( start_timestamp ) > int( int( self.cur_time ) + int( self.timeoffset ) ):
[185]	1036
[354]	1037	self.timeoffset = int( int(start_timestamp) - int(self.cur_time) )
[185]	1038
[133]	1039	elif status == 'Q':
[95]	1040
[451]	1041	# 'mynodequest' can be a string in the following syntax according to the
	1042	# Torque Administator's manual:
	1043	#
	1044	# {<node_count> \| <hostname>}[:ppn=<ppn>][:<property>[:<property>]...][+ ...]
	1045	# {<node_count> \| <hostname>}[:ppn=<ppn>][:<property>[:<property>]...][+ ...]
	1046	# etc
	1047	#
	1048
	1049	#
	1050	# For now we only count the amount of nodes request and ignore properties
	1051	#
	1052
[354]	1053	start_timestamp = ''
	1054	count_mynodes = 0
	1055
[133]	1056	for node in mynoderequest.split( '+' ):
[67]	1057
[451]	1058	# Just grab the {node_count\|hostname} part and ignore properties
	1059	#
[354]	1060	nodepart = node.split( ':' )[0]
[67]	1061
[451]	1062	# Let's assume a node_count value
	1063	#
	1064	numeric_node = 1
	1065
	1066	# Chop the value up into characters
	1067	#
[133]	1068	for letter in nodepart:
[67]	1069
[451]	1070	# If this char is not a digit (0-9), this must be a hostname
	1071	#
[133]	1072	if letter not in string.digits:
	1073
[354]	1074	numeric_node = 0
[133]	1075
[451]	1076	# If this is a hostname, just count this as one (1) node
	1077	#
[133]	1078	if not numeric_node:
[354]	1079
	1080	count_mynodes = count_mynodes + 1
[133]	1081	else:
[451]	1082
	1083	# If this a number, it must be the node_count
	1084	# and increase our count with it's value
	1085	#
[327]	1086	try:
[354]	1087	count_mynodes = count_mynodes + int( nodepart )
	1088
[327]	1089	except ValueError, detail:
[354]	1090
[451]	1091	# When we arrive here I must be bugged or very confused
	1092	# THIS SHOULD NOT HAPPEN!
	1093	#
[327]	1094	debug_msg( 10, str( detail ) )
	1095	debug_msg( 10, "Encountered weird node in Resources_List?!" )
	1096	debug_msg( 10, 'nodepart = ' + str( nodepart ) )
	1097	debug_msg( 10, 'job = ' + str( name ) )
	1098	debug_msg( 10, 'attrs = ' + str( attrs ) )
[133]	1099
[354]	1100	nodeslist = str( count_mynodes )
[172]	1101	else:
[354]	1102	start_timestamp = ''
	1103	nodeslist = ''
[133]	1104
[354]	1105	myAttrs = { }
[26]	1106
[471]	1107	myAttrs[ 'name' ] = str( name )
[354]	1108	myAttrs[ 'queue' ] = str( queue )
	1109	myAttrs[ 'owner' ] = str( owner )
	1110	myAttrs[ 'requested_time' ] = str( requested_time )
	1111	myAttrs[ 'requested_memory' ] = str( requested_memory )
	1112	myAttrs[ 'ppn' ] = str( ppn )
	1113	myAttrs[ 'status' ] = str( status )
	1114	myAttrs[ 'start_timestamp' ] = str( start_timestamp )
	1115	myAttrs[ 'queued_timestamp' ] = str( queued_timestamp )
	1116	myAttrs[ 'reported' ] = str( int( int( self.cur_time ) + int( self.timeoffset ) ) )
	1117	myAttrs[ 'nodes' ] = nodeslist
[507]	1118	myAttrs[ 'domain' ] = fqdn_parts( socket.getfqdn() )[1]
[354]	1119	myAttrs[ 'poll_interval' ] = str( BATCH_POLL_INTERVAL )
	1120
[348]	1121	if self.jobDataChanged( self.jobs, job_id, myAttrs ) and myAttrs['status'] in [ 'R', 'Q' ]:
[61]	1122
[354]	1123	self.jobs[ job_id ] = myAttrs
[26]	1124
[348]	1125	for id, attrs in self.jobs.items():
[76]	1126
	1127	if id not in jobs_processed:
	1128
	1129	# This one isn't there anymore; toedeledoki!
	1130	#
[348]	1131	del self.jobs[ id ]
[76]	1132
[363]	1133	#
	1134	# Gmetric by Nick Galbreath - nickg(a.t)modp(d.o.t)com
	1135	# Version 1.0 - 21-April2-2007
	1136	# http://code.google.com/p/embeddedgmetric/
	1137	#
	1138	# Modified by: Ramon Bastiaans
	1139	# For the Job Monarch Project, see: https://subtrac.sara.nl/oss/jobmonarch/
	1140	#
	1141	# added: DEFAULT_TYPE for Gmetric's
	1142	# added: checkHostProtocol to determine if target is multicast or not
	1143	# changed: allow default for Gmetric constructor
	1144	# changed: allow defaults for all send() values except dmax
	1145	#
	1146
[362]	1147	GMETRIC_DEFAULT_TYPE = 'string'
	1148	GMETRIC_DEFAULT_HOST = '127.0.0.1'
	1149	GMETRIC_DEFAULT_PORT = '8649'
[431]	1150	GMETRIC_DEFAULT_UNITS = ''
[362]	1151
	1152	class Gmetric:
	1153
	1154	global GMETRIC_DEFAULT_HOST, GMETRIC_DEFAULT_PORT
	1155
	1156	slope = { 'zero' : 0, 'positive' : 1, 'negative' : 2, 'both' : 3, 'unspecified' : 4 }
	1157	type = ( '', 'string', 'uint16', 'int16', 'uint32', 'int32', 'float', 'double', 'timestamp' )
	1158	protocol = ( 'udp', 'multicast' )
	1159
	1160	def __init__( self, host=GMETRIC_DEFAULT_HOST, port=GMETRIC_DEFAULT_PORT ):
	1161
	1162	global GMETRIC_DEFAULT_TYPE
	1163
	1164	self.prot = self.checkHostProtocol( host )
	1165	self.msg = xdrlib.Packer()
	1166	self.socket = socket.socket( socket.AF_INET, socket.SOCK_DGRAM )
	1167
	1168	if self.prot not in self.protocol:
	1169
	1170	raise ValueError( "Protocol must be one of: " + str( self.protocol ) )
	1171
	1172	if self.prot == 'multicast':
	1173
[471]	1174	# Set multicast options
	1175	#
[362]	1176	self.socket.setsockopt( socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 20 )
	1177
	1178	self.hostport = ( host, int( port ) )
	1179	self.slopestr = 'both'
	1180	self.tmax = 60
	1181
	1182	def checkHostProtocol( self, ip ):
	1183
[471]	1184	"""Detect if a ip adress is a multicast address"""
	1185
[362]	1186	MULTICAST_ADDRESS_MIN = ( "224", "0", "0", "0" )
	1187	MULTICAST_ADDRESS_MAX = ( "239", "255", "255", "255" )
	1188
	1189	ip_fields = ip.split( '.' )
	1190
	1191	if ip_fields >= MULTICAST_ADDRESS_MIN and ip_fields <= MULTICAST_ADDRESS_MAX:
	1192
	1193	return 'multicast'
	1194	else:
	1195	return 'udp'
	1196
[431]	1197	def send( self, name, value, dmax, typestr = '', units = '' ):
[362]	1198
[409]	1199	if len( units ) == 0:
[431]	1200	units = GMETRIC_DEFAULT_UNITS
[471]	1201
[431]	1202	if len( typestr ) == 0:
	1203	typestr = GMETRIC_DEFAULT_TYPE
[362]	1204
[424]	1205	msg = self.makexdr( name, value, typestr, units, self.slopestr, self.tmax, dmax )
[409]	1206
[362]	1207	return self.socket.sendto( msg, self.hostport )
	1208
	1209	def makexdr( self, name, value, typestr, unitstr, slopestr, tmax, dmax ):
	1210
	1211	if slopestr not in self.slope:
	1212
	1213	raise ValueError( "Slope must be one of: " + str( self.slope.keys() ) )
	1214
	1215	if typestr not in self.type:
	1216
	1217	raise ValueError( "Type must be one of: " + str( self.type ) )
	1218
	1219	if len( name ) == 0:
	1220
	1221	raise ValueError( "Name must be non-empty" )
	1222
	1223	self.msg.reset()
	1224	self.msg.pack_int( 0 )
	1225	self.msg.pack_string( typestr )
	1226	self.msg.pack_string( name )
	1227	self.msg.pack_string( str( value ) )
	1228	self.msg.pack_string( unitstr )
	1229	self.msg.pack_int( self.slope[ slopestr ] )
	1230	self.msg.pack_uint( int( tmax ) )
	1231	self.msg.pack_uint( int( dmax ) )
	1232
	1233	return self.msg.get_buffer()
	1234
[26]	1235	def printTime( ):
[354]	1236
[65]	1237	"""Print current time/date in human readable format for log/debug"""
[26]	1238
	1239	return time.strftime("%a, %d %b %Y %H:%M:%S")
	1240
	1241	def debug_msg( level, msg ):
[354]	1242
[65]	1243	"""Print msg if at or above current debug level"""
[26]	1244
[377]	1245	global DAEMONIZE, DEBUG_LEVEL, SYSLOG_LEVEL
	1246
[373]	1247	if (not DAEMONIZE and DEBUG_LEVEL >= level):
	1248	sys.stderr.write( msg + '\n' )
[26]	1249
[373]	1250	if (DAEMONIZE and USE_SYSLOG and SYSLOG_LEVEL >= level):
	1251	syslog.syslog( msg )
	1252
[307]	1253	def write_pidfile():
	1254
[471]	1255	# Write pidfile if PIDFILE is set
	1256	#
[307]	1257	if PIDFILE:
	1258
[354]	1259	pid = os.getpid()
	1260
[471]	1261	pidfile = open( PIDFILE, 'w' )
[354]	1262
	1263	pidfile.write( str( pid ) )
[307]	1264	pidfile.close()
	1265
[23]	1266	def main():
[354]	1267
[65]	1268	"""Application start"""
[23]	1269
[352]	1270	global PBSQuery, PBSError
[373]	1271	global SYSLOG_FACILITY, USE_SYSLOG, BATCH_API, DAEMONIZE
[256]	1272
[212]	1273	if not processArgs( sys.argv[1:] ):
[354]	1274
[212]	1275	sys.exit( 1 )
	1276
[471]	1277	# Load appropriate DataGatherer depending on which BATCH_API is set
	1278	# and any required modules for the Gatherer
	1279	#
[256]	1280	if BATCH_API == 'pbs':
	1281
	1282	try:
[282]	1283	from PBSQuery import PBSQuery, PBSError
[256]	1284
	1285	except ImportError:
	1286
[373]	1287	debug_msg( 0, "FATAL ERROR: BATCH_API set to 'pbs' but python module 'pbs_python' is not installed" )
[256]	1288	sys.exit( 1 )
	1289
	1290	gather = PbsDataGatherer()
	1291
	1292	elif BATCH_API == 'sge':
	1293
[507]	1294	# Tested with SGE 6.0u11.
	1295	# debug_msg( 0, "FATAL ERROR: BATCH_API 'sge' implementation is currently broken, check future releases" )
[368]	1296
[507]	1297	# sys.exit( 1 )
[368]	1298
[347]	1299	gather = SgeDataGatherer()
[256]	1300
	1301	else:
[373]	1302	debug_msg( 0, "FATAL ERROR: unknown BATCH_API '" + BATCH_API + "' is not supported" )
[354]	1303
[256]	1304	sys.exit( 1 )
	1305
[373]	1306	if( DAEMONIZE and USE_SYSLOG ):
	1307
	1308	syslog.openlog( 'jobmond', syslog.LOG_NOWAIT, SYSLOG_FACILITY )
	1309
	1310
[26]	1311	if DAEMONIZE:
[354]	1312
[26]	1313	gather.daemon()
	1314	else:
	1315	gather.run()
[23]	1316
[256]	1317	# wh00t? someone started me! :)
[65]	1318	#
[23]	1319	if __name__ == '__main__':
	1320	main()

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

Download in other formats: