Context Navigation

source: trunk/jobmond/jobmond.py @ 355

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

jobmond/jobmond.py:

close gmetric pipe after checking version
more code cleanup

Property svn:keywords set to Id

File size: 19.7 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 355 2007-05-03 13:09:25Z bastiaans $
[227]	22	#
[23]	23
[212]	24	import sys, getopt, ConfigParser
[354]	25	import time, os, socket, string, re
[318]	26	import xml, xml.sax
	27	from xml.sax import saxutils, make_parser
	28	from xml.sax import make_parser
	29	from xml.sax.handler import feature_namespaces
	30
[307]	31	def usage():
	32
	33	print
	34	print 'usage: jobmond [options]'
	35	print 'options:'
	36	print ' --config, -c configuration file'
	37	print ' --pidfile, -p pid file'
	38	print ' --help, -h help'
	39	print
	40
[212]	41	def processArgs( args ):
[26]	42
[354]	43	SHORT_L = 'hc:'
	44	LONG_L = [ 'help', 'config=' ]
[165]	45
[307]	46	global PIDFILE
[354]	47	PIDFILE = None
[61]	48
[354]	49	config_filename = '/etc/jobmond.conf'
	50
[212]	51	try:
[68]	52
[354]	53	opts, args = getopt.getopt( args, SHORT_L, LONG_L )
[185]	54
[307]	55	except getopt.GetoptError, detail:
[212]	56
	57	print detail
[307]	58	usage()
[354]	59	sys.exit( 1 )
[212]	60
	61	for opt, value in opts:
	62
	63	if opt in [ '--config', '-c' ]:
	64
[354]	65	config_filename = value
[212]	66
[307]	67	if opt in [ '--pidfile', '-p' ]:
[212]	68
[354]	69	PIDFILE = value
[307]	70
	71	if opt in [ '--help', '-h' ]:
	72
	73	usage()
[354]	74	sys.exit( 0 )
[212]	75
	76	return loadConfig( config_filename )
	77
	78	def loadConfig( filename ):
	79
[215]	80	def getlist( cfg_string ):
	81
	82	my_list = [ ]
	83
	84	for item_txt in cfg_string.split( ',' ):
	85
	86	sep_char = None
	87
	88	item_txt = item_txt.strip()
	89
	90	for s_char in [ "'", '"' ]:
	91
	92	if item_txt.find( s_char ) != -1:
	93
	94	if item_txt.count( s_char ) != 2:
	95
	96	print 'Missing quote: %s' %item_txt
	97	sys.exit( 1 )
	98
	99	else:
	100
	101	sep_char = s_char
	102	break
	103
	104	if sep_char:
	105
	106	item_txt = item_txt.split( sep_char )[1]
	107
	108	my_list.append( item_txt )
	109
	110	return my_list
	111
[354]	112	cfg = ConfigParser.ConfigParser()
[212]	113
	114	cfg.read( filename )
	115
[354]	116	global DEBUG_LEVEL, DAEMONIZE, BATCH_SERVER, BATCH_POLL_INTERVAL
	117	global GMOND_CONF, DETECT_TIME_DIFFS, BATCH_HOST_TRANSLATE
	118	global BATCH_API, QUEUE, GMETRIC_TARGET
[212]	119
[354]	120	DEBUG_LEVEL = cfg.getint( 'DEFAULT', 'DEBUG_LEVEL' )
[212]	121
[354]	122	DAEMONIZE = cfg.getboolean( 'DEFAULT', 'DAEMONIZE' )
[212]	123
[265]	124	try:
[212]	125
[354]	126	BATCH_SERVER = cfg.get( 'DEFAULT', 'BATCH_SERVER' )
[212]	127
[265]	128	except ConfigParser.NoOptionError:
	129
	130	# Backwards compatibility for old configs
	131	#
	132
[354]	133	BATCH_SERVER = cfg.get( 'DEFAULT', 'TORQUE_SERVER' )
	134	api_guess = 'pbs'
[265]	135
	136	try:
	137
[354]	138	BATCH_POLL_INTERVAL = cfg.getint( 'DEFAULT', 'BATCH_POLL_INTERVAL' )
[265]	139
	140	except ConfigParser.NoOptionError:
	141
	142	# Backwards compatibility for old configs
	143	#
	144
[354]	145	BATCH_POLL_INTERVAL = cfg.getint( 'DEFAULT', 'TORQUE_POLL_INTERVAL' )
	146	api_guess = 'pbs'
[353]	147
	148	try:
[212]	149
[354]	150	GMOND_CONF = cfg.get( 'DEFAULT', 'GMOND_CONF' )
[353]	151
	152	except ConfigParser.NoOptionError:
	153
[354]	154	GMOND_CONF = None
[353]	155
[354]	156	DETECT_TIME_DIFFS = cfg.getboolean( 'DEFAULT', 'DETECT_TIME_DIFFS' )
[212]	157
[354]	158	BATCH_HOST_TRANSLATE = getlist( cfg.get( 'DEFAULT', 'BATCH_HOST_TRANSLATE' ) )
[215]	159
[266]	160	try:
[256]	161
[354]	162	BATCH_API = cfg.get( 'DEFAULT', 'BATCH_API' )
[266]	163
	164	except ConfigParser.NoOptionError, detail:
	165
	166	if BATCH_SERVER and api_guess:
[354]	167
	168	BATCH_API = api_guess
[266]	169	else:
	170	debug_msg( 0, "fatal error: BATCH_API not set and can't make guess" )
	171	sys.exit( 1 )
[317]	172
	173	try:
	174
[354]	175	QUEUE = getlist( cfg.get( 'DEFAULT', 'QUEUE' ) )
[317]	176
	177	except ConfigParser.NoOptionError, detail:
	178
[354]	179	QUEUE = None
[353]	180
	181	try:
	182
[354]	183	GMETRIC_TARGET = cfg.get( 'DEFAULT', 'GMETRIC_TARGET' )
[353]	184
	185	except ConfigParser.NoOptionError:
	186
[354]	187	GMETRIC_TARGET = None
[353]	188
	189	if not GMOND_CONF:
	190
	191	debug_msg( 0, "fatal error: GMETRIC_TARGET or GMOND_CONF both not set!" )
	192	sys.exit( 1 )
	193	else:
	194
	195	debug_msg( 0, "error: GMETRIC_TARGET not set: internel Gmetric handling aborted. Failing back to DEPRECATED use of gmond.conf/gmetric binary. This will slow down jobmond significantly!" )
	196
[212]	197	return True
	198
[253]	199	METRIC_MAX_VAL_LEN = 900
	200
[61]	201	class DataProcessor:
[355]	202
[68]	203	"""Class for processing of data"""
[61]	204
	205	binary = '/usr/bin/gmetric'
	206
	207	def __init__( self, binary=None ):
[355]	208
[68]	209	"""Remember alternate binary location if supplied"""
[61]	210
	211	if binary:
	212	self.binary = binary
	213
[80]	214	# Timeout for XML
	215	#
	216	# From ganglia's documentation:
	217	#
	218	# 'A metric will be deleted DMAX seconds after it is received, and
	219	# DMAX=0 means eternal life.'
[61]	220
[256]	221	self.dmax = str( int( int( BATCH_POLL_INTERVAL ) * 2 ) )
[80]	222
[353]	223	if GMOND_CONF:
[354]	224
[353]	225	try:
	226	gmond_file = GMOND_CONF
[68]	227
[353]	228	except NameError:
	229	gmond_file = '/etc/gmond.conf'
[68]	230
[353]	231	if not os.path.exists( gmond_file ):
	232	debug_msg( 0, 'fatal error: ' + gmond_file + ' does not exist' )
	233	sys.exit( 1 )
[68]	234
[353]	235	incompatible = self.checkGmetricVersion()
[61]	236
[353]	237	if incompatible:
[355]	238
	239	debug_msg( 0, 'Gmetric version not compatible, please upgrade to at least 3.0.1' )
[353]	240	sys.exit( 1 )
[65]	241
	242	def checkGmetricVersion( self ):
[355]	243
[68]	244	"""
	245	Check version of gmetric is at least 3.0.1
	246	for the syntax we use
	247	"""
[65]	248
[255]	249	global METRIC_MAX_VAL_LEN
	250
[341]	251	incompatible = 0
	252
[355]	253	gfp = os.popen( self.binary + ' --version' )
	254	lines = gfp.readlines()
[65]	255
[355]	256	gfp.close()
	257
	258	for line in lines:
	259
[65]	260	line = line.split( ' ' )
	261
[355]	262	if len( line ) == 2 and str( line ).find( 'gmetric' ) != -1:
[65]	263
[355]	264	gmetric_version = line[1].split( '\n' )[0]
[65]	265
[355]	266	version_major = int( gmetric_version.split( '.' )[0] )
	267	version_minor = int( gmetric_version.split( '.' )[1] )
	268	version_patch = int( gmetric_version.split( '.' )[2] )
[65]	269
[355]	270	incompatible = 0
[65]	271
	272	if version_major < 3:
	273
	274	incompatible = 1
	275
	276	elif version_major == 3:
	277
	278	if version_minor == 0:
	279
	280	if version_patch < 1:
	281
[91]	282	incompatible = 1
[65]	283
[255]	284	if version_patch < 3:
	285
	286	METRIC_MAX_VAL_LEN = 900
	287
	288	elif version_patch >= 3:
	289
	290	METRIC_MAX_VAL_LEN = 1400
	291
[65]	292	return incompatible
	293
[75]	294	def multicastGmetric( self, metricname, metricval, valtype='string' ):
[355]	295
[68]	296	"""Call gmetric binary and multicast"""
[65]	297
	298	cmd = self.binary
	299
[353]	300	if GMETRIC_TARGET:
[61]	301
[353]	302	from gmetric import Gmetric
[61]	303
[353]	304	if GMETRIC_TARGET:
[61]	305
[353]	306	GMETRIC_TARGET_HOST = GMETRIC_TARGET.split( ':' )[0]
	307	GMETRIC_TARGET_PORT = GMETRIC_TARGET.split( ':' )[1]
	308
	309	metric_debug = "[gmetric] name: %s - val: %s - dmax: %s" %( str( metricname ), str( metricval ), str( self.dmax ) )
	310
	311	debug_msg( 10, printTime() + ' ' + metric_debug)
	312
	313	gm = Gmetric( GMETRIC_TARGET_HOST, GMETRIC_TARGET_PORT )
	314
	315	gm.send( str( metricname ), str( metricval ), str( self.dmax ) )
	316
	317	else:
	318	try:
	319	cmd = cmd + ' -c' + GMOND_CONF
	320
	321	except NameError:
	322
	323	debug_msg( 10, 'Assuming /etc/gmond.conf for gmetric cmd (ommitting)' )
	324
	325	cmd = cmd + ' -n' + str( metricname )+ ' -v"' + str( metricval )+ '" -t' + str( valtype ) + ' -d' + str( self.dmax )
	326
	327	debug_msg( 10, printTime() + ' ' + cmd )
	328
	329	os.system( cmd )
	330
[318]	331	class DataGatherer:
[23]	332
[318]	333	"""Skeleton class for batch system DataGatherer"""
[256]	334
[318]	335	def printJobs( self, jobs ):
[355]	336
[318]	337	"""Print a jobinfo overview"""
	338
	339	for name, attrs in self.jobs.items():
	340
	341	print 'job %s' %(name)
	342
	343	for name, val in attrs.items():
	344
	345	print '\t%s = %s' %( name, val )
	346
	347	def printJob( self, jobs, job_id ):
[355]	348
[318]	349	"""Print job with job_id from jobs"""
	350
	351	print 'job %s' %(job_id)
	352
	353	for name, val in jobs[ job_id ].items():
	354
	355	print '\t%s = %s' %( name, val )
	356
[256]	357	def daemon( self ):
[355]	358
[318]	359	"""Run as daemon forever"""
[256]	360
[318]	361	# Fork the first child
	362	#
	363	pid = os.fork()
	364	if pid > 0:
	365	sys.exit(0) # end parent
[256]	366
[318]	367	# creates a session and sets the process group ID
	368	#
	369	os.setsid()
	370
	371	# Fork the second child
	372	#
	373	pid = os.fork()
	374	if pid > 0:
	375	sys.exit(0) # end parent
	376
	377	write_pidfile()
	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', os.O_RDWR)
	389	os.dup2(0, 1)
	390	os.dup2(0, 2)
	391
	392	self.run()
	393
[256]	394	def run( self ):
[355]	395
[318]	396	"""Main thread"""
[256]	397
[318]	398	while ( 1 ):
	399
[348]	400	self.getJobData()
	401	self.submitJobData()
[318]	402	time.sleep( BATCH_POLL_INTERVAL )
[256]	403
[318]	404	class SgeQstatXMLParser(xml.sax.handler.ContentHandler):
[256]	405
[318]	406	"""Babu Sundaram's experimental SGE qstat XML parser"""
[256]	407
[318]	408	def __init__(self, qstatinxml):
	409
	410	self.qstatfile = qstatinxml
	411	self.attribs = {}
	412	self.value = ''
	413	self.jobID = ''
	414	self.currentJobInfo = ''
	415	self.job_list = []
	416	self.EOFFlag = 0
	417	self.jobinfoCount = 0
	418
	419
	420	def startElement(self, name, attrs):
	421
	422	if name == 'job_list':
	423	self.currentJobInfo = 'Status=' + attrs.get('state', None) + ' '
	424	elif name == 'job_info':
	425	self.job_list = []
	426	self.jobinfoCount += 1
	427
	428	def characters(self, ch):
	429
	430	self.value = self.value + ch
	431
	432	def endElement(self, name):
	433
	434	if len(self.value.strip()) > 0 :
	435
	436	self.currentJobInfo += name + '=' + self.value.strip() + ' '
	437	elif name != 'job_list':
	438
	439	self.currentJobInfo += name + '=Unknown '
	440
	441	if name == 'JB_job_number':
	442
	443	self.jobID = self.value.strip()
	444	self.job_list.append(self.jobID)
	445
	446	if name == 'job_list':
	447
	448	if self.attribs.has_key(self.jobID) == False:
	449	self.attribs[self.jobID] = self.currentJobInfo
	450	elif self.attribs.has_key(self.jobID) and self.attribs[self.jobID] != self.currentJobInfo:
	451	self.attribs[self.jobID] = self.currentJobInfo
	452	self.currentJobInfo = ''
	453	self.jobID = ''
	454
	455	elif name == 'job_info' and self.jobinfoCount == 2:
	456
	457	deljobs = []
	458	for id in self.attribs:
	459	try:
	460	self.job_list.index(str(id))
	461	except ValueError:
	462	deljobs.append(id)
	463	for i in deljobs:
	464	del self.attribs[i]
	465	deljobs = []
	466	self.jobinfoCount = 0
	467
	468	self.value = ''
	469
	470	class SgeDataGatherer(DataGatherer):
	471
[61]	472	jobs = { }
[347]	473	SGE_QSTAT_XML_FILE = '/tmp/.jobmonarch.sge.qstat'
[61]	474
[318]	475	def __init__( self ):
	476	"""Setup appropriate variables"""
	477
	478	self.jobs = { }
	479	self.timeoffset = 0
	480	self.dp = DataProcessor()
	481	self.initSgeJobInfo()
	482
	483	def initSgeJobInfo( self ):
	484	"""This is outside the scope of DRMAA; Get the current jobs in SGE"""
	485	"""This is a hack because we cant get info about jobs beyond"""
	486	"""those in the current DRMAA session"""
	487
[347]	488	self.qstatparser = SgeQstatXMLParser( self.SGE_QSTAT_XML_FILE )
[318]	489
	490	# Obtain the qstat information from SGE in XML format
	491	# This would change to DRMAA-specific calls from 6.0u9
	492
	493	def getJobData(self):
	494	"""Gather all data on current jobs in SGE"""
	495
	496	# Get the information about the current jobs in the SGE queue
	497	info = os.popen("qstat -ext -xml").readlines()
[347]	498	f = open(self.SGE_QSTAT_XML_FILE,'w')
[318]	499	for lines in info:
	500	f.write(lines)
	501	f.close()
	502
	503	# Parse the input
	504	f = open(self.qstatparser.qstatfile, 'r')
	505	xml.sax.parse(f, self.qstatparser)
	506	f.close()
	507
	508	self.cur_time = time.time()
	509
	510	return self.qstatparser.attribs
	511
	512	def submitJobData(self):
	513	"""Submit job info list"""
	514
	515	self.dp.multicastGmetric( 'MONARCH-HEARTBEAT', str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
	516	# Now let's spread the knowledge
	517	#
	518	metric_increment = 0
	519	for jobid, jobattrs in self.qstatparser.attribs.items():
	520
	521	self.dp.multicastGmetric( 'MONARCH-JOB-' + jobid + '-' + str(metric_increment), jobattrs)
	522
[355]	523	class PbsDataGatherer( DataGatherer ):
[318]	524
	525	"""This is the DataGatherer for PBS and Torque"""
	526
[256]	527	global PBSQuery
	528
[23]	529	def __init__( self ):
[354]	530
[68]	531	"""Setup appropriate variables"""
[23]	532
[354]	533	self.jobs = { }
	534	self.timeoffset = 0
	535	self.dp = DataProcessor()
	536
[91]	537	self.initPbsQuery()
[23]	538
[91]	539	def initPbsQuery( self ):
	540
[354]	541	self.pq = None
	542
[256]	543	if( BATCH_SERVER ):
[354]	544
	545	self.pq = PBSQuery( BATCH_SERVER )
[174]	546	else:
[354]	547	self.pq = PBSQuery()
[91]	548
[26]	549	def getAttr( self, attrs, name ):
[354]	550
[68]	551	"""Return certain attribute from dictionary, if exists"""
[26]	552
	553	if attrs.has_key( name ):
[354]	554
	555	return attrs[ name ]
[26]	556	else:
	557	return ''
	558
	559	def jobDataChanged( self, jobs, job_id, attrs ):
[354]	560
[68]	561	"""Check if job with attrs and job_id in jobs has changed"""
[26]	562
	563	if jobs.has_key( job_id ):
[354]	564
[26]	565	oldData = jobs[ job_id ]
	566	else:
	567	return 1
	568
	569	for name, val in attrs.items():
	570
	571	if oldData.has_key( name ):
	572
	573	if oldData[ name ] != attrs[ name ]:
	574
	575	return 1
	576
	577	else:
	578	return 1
	579
	580	return 0
	581
[348]	582	def getJobData( self ):
[354]	583
[68]	584	"""Gather all data on current jobs in Torque"""
[26]	585
[354]	586	joblist = {}
[349]	587
[354]	588	while len( joblist ) == 0:
	589
[282]	590	try:
	591	joblist = self.pq.getjobs()
[354]	592
[349]	593	except PBSError, detail:
[354]	594
	595	debug_msg( 10, "Caught PBS unavailable, skipping until next polling interval: " + str( detail ) )
[349]	596	return None
[26]	597
[354]	598	self.cur_time = time.time()
[68]	599
[354]	600	jobs_processed = [ ]
[26]	601
	602	for name, attrs in joblist.items():
	603
[354]	604	job_id = name.split( '.' )[0]
[26]	605
	606	jobs_processed.append( job_id )
[61]	607
[354]	608	name = self.getAttr( attrs, 'Job_Name' )
	609	queue = self.getAttr( attrs, 'queue' )
[317]	610
	611	if QUEUE:
	612
	613	if QUEUE != queue:
	614
	615	continue
	616
[354]	617	owner = self.getAttr( attrs, 'Job_Owner' ).split( '@' )[0]
	618	requested_time = self.getAttr( attrs, 'Resource_List.walltime' )
	619	requested_memory = self.getAttr( attrs, 'Resource_List.mem' )
[95]	620
[354]	621	mynoderequest = self.getAttr( attrs, 'Resource_List.nodes' )
[95]	622
[354]	623	ppn = ''
[281]	624
[26]	625	if mynoderequest.find( ':' ) != -1 and mynoderequest.find( 'ppn' ) != -1:
[95]	626
[354]	627	mynoderequest_fields = mynoderequest.split( ':' )
[281]	628
	629	for mynoderequest_field in mynoderequest_fields:
	630
	631	if mynoderequest_field.find( 'ppn' ) != -1:
	632
[354]	633	ppn = mynoderequest_field.split( 'ppn=' )[1]
[281]	634
[354]	635	status = self.getAttr( attrs, 'job_state' )
[25]	636
[354]	637	queued_timestamp = self.getAttr( attrs, 'ctime' )
[243]	638
[95]	639	if status == 'R':
[133]	640
[354]	641	start_timestamp = self.getAttr( attrs, 'mtime' )
	642	nodes = self.getAttr( attrs, 'exec_host' ).split( '+' )
[133]	643
[354]	644	nodeslist = [ ]
	645
[133]	646	for node in nodes:
	647
[354]	648	host = node.split( '/' )[0]
	649
[133]	650	if nodeslist.count( host ) == 0:
[215]	651
	652	for translate_pattern in BATCH_HOST_TRANSLATE:
	653
[220]	654	if translate_pattern.find( '/' ) != -1:
[215]	655
[354]	656	translate_orig = translate_pattern.split( '/' )[1]
	657	translate_new = translate_pattern.split( '/' )[2]
[220]	658
[354]	659	host = re.sub( translate_orig, translate_new, host )
[216]	660
[217]	661	if not host in nodeslist:
[216]	662
	663	nodeslist.append( host )
[133]	664
[185]	665	if DETECT_TIME_DIFFS:
	666
	667	# If a job start if later than our current date,
	668	# that must mean the Torque server's time is later
	669	# than our local time.
	670
[354]	671	if int( start_timestamp ) > int( int( self.cur_time ) + int( self.timeoffset ) ):
[185]	672
[354]	673	self.timeoffset = int( int(start_timestamp) - int(self.cur_time) )
[185]	674
[133]	675	elif status == 'Q':
[95]	676
[354]	677	start_timestamp = ''
	678	count_mynodes = 0
	679	numeric_node = 1
	680
[133]	681	for node in mynoderequest.split( '+' ):
[67]	682
[354]	683	nodepart = node.split( ':' )[0]
[67]	684
[133]	685	for letter in nodepart:
[67]	686
[133]	687	if letter not in string.digits:
	688
[354]	689	numeric_node = 0
[133]	690
	691	if not numeric_node:
[354]	692
	693	count_mynodes = count_mynodes + 1
[133]	694	else:
[327]	695	try:
[354]	696	count_mynodes = count_mynodes + int( nodepart )
	697
[327]	698	except ValueError, detail:
[354]	699
[327]	700	debug_msg( 10, str( detail ) )
	701	debug_msg( 10, "Encountered weird node in Resources_List?!" )
	702	debug_msg( 10, 'nodepart = ' + str( nodepart ) )
	703	debug_msg( 10, 'job = ' + str( name ) )
	704	debug_msg( 10, 'attrs = ' + str( attrs ) )
[133]	705
[354]	706	nodeslist = str( count_mynodes )
[172]	707	else:
[354]	708	start_timestamp = ''
	709	nodeslist = ''
[133]	710
[354]	711	myAttrs = { }
[26]	712
[354]	713	myAttrs[ 'name' ] = str( name )
	714	myAttrs[ 'queue' ] = str( queue )
	715	myAttrs[ 'owner' ] = str( owner )
	716	myAttrs[ 'requested_time' ] = str( requested_time )
	717	myAttrs[ 'requested_memory' ] = str( requested_memory )
	718	myAttrs[ 'ppn' ] = str( ppn )
	719	myAttrs[ 'status' ] = str( status )
	720	myAttrs[ 'start_timestamp' ] = str( start_timestamp )
	721	myAttrs[ 'queued_timestamp' ] = str( queued_timestamp )
	722	myAttrs[ 'reported' ] = str( int( int( self.cur_time ) + int( self.timeoffset ) ) )
	723	myAttrs[ 'nodes' ] = nodeslist
	724	myAttrs[ 'domain' ] = string.join( socket.getfqdn().split( '.' )[1:], '.' )
	725	myAttrs[ 'poll_interval' ] = str( BATCH_POLL_INTERVAL )
	726
[348]	727	if self.jobDataChanged( self.jobs, job_id, myAttrs ) and myAttrs['status'] in [ 'R', 'Q' ]:
[61]	728
[354]	729	self.jobs[ job_id ] = myAttrs
[26]	730
[348]	731	for id, attrs in self.jobs.items():
[76]	732
	733	if id not in jobs_processed:
	734
	735	# This one isn't there anymore; toedeledoki!
	736	#
[348]	737	del self.jobs[ id ]
[76]	738
[348]	739	def submitJobData( self ):
[354]	740
[65]	741	"""Submit job info list"""
	742
[219]	743	self.dp.multicastGmetric( 'MONARCH-HEARTBEAT', str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
[69]	744
[61]	745	# Now let's spread the knowledge
	746	#
[348]	747	for jobid, jobattrs in self.jobs.items():
[61]	748
[354]	749	gmetric_val = self.compileGmetricVal( jobid, jobattrs )
	750	metric_increment = 0
[61]	751
[354]	752	for val in gmetric_val:
[253]	753
	754	self.dp.multicastGmetric( 'MONARCH-JOB-' + jobid + '-' + str(metric_increment), val )
[61]	755
[354]	756	metric_increment = metric_increment + 1
	757
[253]	758	def compileGmetricVal( self, jobid, jobattrs ):
[354]	759
[253]	760	"""Create a val string for gmetric of jobinfo"""
[67]	761
[354]	762	gval_lists = [ ]
	763	mystr = None
	764	val_list = { }
[67]	765
[253]	766	for val_name, val_value in jobattrs.items():
[61]	767
[253]	768	val_list_names_len = len( string.join( val_list.keys() ) ) + len(val_list.keys())
	769	val_list_vals_len = len( string.join( val_list.values() ) ) + len(val_list.values())
[95]	770
[254]	771	if val_name == 'nodes' and jobattrs['status'] == 'R':
[95]	772
[253]	773	node_str = None
[95]	774
[253]	775	for node in val_value:
[65]	776
[253]	777	if node_str:
[354]	778
[253]	779	node_str = node_str + ';' + node
	780	else:
	781	node_str = node
[65]	782
[253]	783	if (val_list_names_len + len(val_name) ) + (val_list_vals_len + len(node_str) ) > METRIC_MAX_VAL_LEN:
[65]	784
[354]	785	val_list[ val_name ] = node_str
	786
[253]	787	gval_lists.append( val_list )
[65]	788
[354]	789	val_list = { }
	790	node_str = None
	791
	792	val_list[ val_name ] = node_str
	793
[253]	794	gval_lists.append( val_list )
[65]	795
[354]	796	val_list = { }
	797
[254]	798	elif val_value != '':
	799
	800	if (val_list_names_len + len(val_name) ) + (val_list_vals_len + len(str(val_value)) ) > METRIC_MAX_VAL_LEN:
	801
	802	gval_lists.append( val_list )
	803
[354]	804	val_list = { }
[254]	805
[354]	806	val_list[ val_name ] = val_value
	807
	808	if len( val_list ) > 0:
	809
[254]	810	gval_lists.append( val_list )
	811
[354]	812	str_list = [ ]
[65]	813
[253]	814	for val_list in gval_lists:
[65]	815
[354]	816	my_val_str = None
[65]	817
[253]	818	for val_name, val_value in val_list.items():
[65]	819
[253]	820	if my_val_str:
[65]	821
[253]	822	my_val_str = my_val_str + ' ' + val_name + '=' + val_value
	823	else:
	824	my_val_str = val_name + '=' + val_value
	825
	826	str_list.append( my_val_str )
	827
	828	return str_list
	829
[26]	830	def printTime( ):
[354]	831
[65]	832	"""Print current time/date in human readable format for log/debug"""
[26]	833
	834	return time.strftime("%a, %d %b %Y %H:%M:%S")
	835
	836	def debug_msg( level, msg ):
[354]	837
[65]	838	"""Print msg if at or above current debug level"""
[26]	839
	840	if (DEBUG_LEVEL >= level):
	841	sys.stderr.write( msg + '\n' )
	842
[307]	843	def write_pidfile():
	844
	845	# Write pidfile if PIDFILE exists
	846	if PIDFILE:
	847
[354]	848	pid = os.getpid()
	849
	850	pidfile = open(PIDFILE, 'w')
	851
	852	pidfile.write( str( pid ) )
[307]	853	pidfile.close()
	854
[23]	855	def main():
[354]	856
[65]	857	"""Application start"""
[23]	858
[352]	859	global PBSQuery, PBSError
[256]	860
[212]	861	if not processArgs( sys.argv[1:] ):
[354]	862
[212]	863	sys.exit( 1 )
	864
[256]	865	if BATCH_API == 'pbs':
	866
	867	try:
[282]	868	from PBSQuery import PBSQuery, PBSError
[256]	869
	870	except ImportError:
	871
	872	debug_msg( 0, "fatal error: BATCH_API set to 'pbs' but python module 'pbs_python' is not installed" )
	873	sys.exit( 1 )
	874
	875	gather = PbsDataGatherer()
	876
	877	elif BATCH_API == 'sge':
	878
[347]	879	gather = SgeDataGatherer()
[256]	880
	881	else:
	882	debug_msg( 0, "fatal error: unknown BATCH_API '" + BATCH_API + "' is not supported" )
[354]	883
[256]	884	sys.exit( 1 )
	885
[26]	886	if DAEMONIZE:
[354]	887
[26]	888	gather.daemon()
	889	else:
	890	gather.run()
[23]	891
[256]	892	# wh00t? someone started me! :)
[65]	893	#
[23]	894	if __name__ == '__main__':
	895	main()

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

Download in other formats: