Context Navigation

source: trunk/jobmond/jobmond.py @ 345

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

jobmond/jobmond.py:

fixed bug #21 thanks to Peter Kruse

Property svn:keywords set to Id

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

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

Download in other formats: