Context Navigation

source: trunk/jobmond/jobmond.py @ 514

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

jobmond/jobmond.py:

added report time to down/offline nodes
added domain name to down/offline nodes

web/addons/job_monarch/libtoga.php:

incorporate down/offline node status in ClusterImage?

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

File size: 33.6 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 514 2008-03-08 20:16:32Z 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
[514]	492	domain = fqdn_parts( socket.getfqdn() )[1]
	493
[512]	494	downed_nodes = list()
	495	offline_nodes = list()
	496
	497	l = ['state']
	498
	499	for name, node in self.pq.getnodes().items():
	500
	501	if ( node[ 'state' ].find( "down" ) != -1 ):
	502
	503	downed_nodes.append( name )
	504
	505	if ( node[ 'state' ].find( "offline" ) != -1 ):
	506
	507	offline_nodes.append( name )
	508
[514]	509	downnodeslist = do_nodelist( downed_nodes )
	510	offlinenodeslist = do_nodelist( offline_nodes )
[512]	511
[514]	512	down_str = 'nodes=%s domain=%s reported=%s' %( string.join( downnodeslist, ';' ), domain, str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
	513	offl_str = 'nodes=%s domain=%s reported=%s' %( string.join( offlinenodeslist, ';' ), domain, str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
	514	self.dp.multicastGmetric( 'MONARCH-DOWN' , down_str )
	515	self.dp.multicastGmetric( 'MONARCH-OFFLINE', offl_str )
	516
[507]	517	# Now let's spread the knowledge
	518	#
	519	for jobid, jobattrs in self.jobs.items():
	520
	521	# Make gmetric values for each job: respect max gmetric value length
	522	#
	523	gmetric_val = self.compileGmetricVal( jobid, jobattrs )
	524	metric_increment = 0
	525
	526	# If we have more job info than max gmetric value length allows, split it up
	527	# amongst multiple metrics
	528	#
	529	for val in gmetric_val:
	530
	531	self.dp.multicastGmetric( 'MONARCH-JOB-' + jobid + '-' + str(metric_increment), val )
	532
	533	# Increase follow number if this jobinfo is split up amongst more than 1 gmetric
	534	#
	535	metric_increment = metric_increment + 1
	536
	537	def compileGmetricVal( self, jobid, jobattrs ):
	538
	539	"""Create a val string for gmetric of jobinfo"""
	540
	541	gval_lists = [ ]
	542	val_list = { }
	543
	544	for val_name, val_value in jobattrs.items():
	545
	546	# These are our own metric names, i.e.: status, start_timestamp, etc
	547	#
	548	val_list_names_len = len( string.join( val_list.keys() ) ) + len(val_list.keys())
	549
	550	# These are their corresponding values
	551	#
	552	val_list_vals_len = len( string.join( val_list.values() ) ) + len(val_list.values())
	553
	554	if val_name == 'nodes' and jobattrs['status'] == 'R':
	555
	556	node_str = None
	557
	558	for node in val_value:
	559
	560	if node_str:
	561
	562	node_str = node_str + ';' + node
	563	else:
	564	node_str = node
	565
	566	# Make sure if we add this new info, that the total metric's value length does not exceed METRIC_MAX_VAL_LEN
	567	#
	568	if (val_list_names_len + len(val_name) ) + (val_list_vals_len + len(node_str) ) > METRIC_MAX_VAL_LEN:
	569
	570	# It's too big, we need to make a new gmetric for the additional info
	571	#
	572	val_list[ val_name ] = node_str
	573
	574	gval_lists.append( val_list )
	575
	576	val_list = { }
	577	node_str = None
	578
	579	val_list[ val_name ] = node_str
	580
	581	gval_lists.append( val_list )
	582
	583	val_list = { }
	584
	585	elif val_value != '':
	586
	587	# Make sure if we add this new info, that the total metric's value length does not exceed METRIC_MAX_VAL_LEN
	588	#
	589	if (val_list_names_len + len(val_name) ) + (val_list_vals_len + len(str(val_value)) ) > METRIC_MAX_VAL_LEN:
	590
	591	# It's too big, we need to make a new gmetric for the additional info
	592	#
	593	gval_lists.append( val_list )
	594
	595	val_list = { }
	596
	597	val_list[ val_name ] = val_value
	598
	599	if len( val_list ) > 0:
	600
	601	gval_lists.append( val_list )
	602
	603	str_list = [ ]
	604
	605	# Now append the value names and values together, i.e.: stop_timestamp=value, etc
	606	#
	607	for val_list in gval_lists:
	608
	609	my_val_str = None
	610
	611	for val_name, val_value in val_list.items():
	612
	613	if my_val_str:
	614
	615	my_val_str = my_val_str + ' ' + val_name + '=' + val_value
	616	else:
	617	my_val_str = val_name + '=' + val_value
	618
	619	str_list.append( my_val_str )
	620
	621	return str_list
	622
[256]	623	def daemon( self ):
[355]	624
[318]	625	"""Run as daemon forever"""
[256]	626
[318]	627	# Fork the first child
	628	#
	629	pid = os.fork()
	630	if pid > 0:
	631	sys.exit(0) # end parent
[256]	632
[318]	633	# creates a session and sets the process group ID
	634	#
	635	os.setsid()
	636
	637	# Fork the second child
	638	#
	639	pid = os.fork()
	640	if pid > 0:
	641	sys.exit(0) # end parent
	642
	643	write_pidfile()
	644
	645	# Go to the root directory and set the umask
	646	#
	647	os.chdir('/')
	648	os.umask(0)
	649
	650	sys.stdin.close()
	651	sys.stdout.close()
	652	sys.stderr.close()
	653
	654	os.open('/dev/null', os.O_RDWR)
	655	os.dup2(0, 1)
	656	os.dup2(0, 2)
	657
	658	self.run()
	659
[256]	660	def run( self ):
[355]	661
[318]	662	"""Main thread"""
[256]	663
[318]	664	while ( 1 ):
	665
[348]	666	self.getJobData()
	667	self.submitJobData()
[318]	668	time.sleep( BATCH_POLL_INTERVAL )
[256]	669
[507]	670	# SGE code by Dave Love <fx@gnu.org>. Tested with SGE 6.0u8 and 6.0u11.
	671	# Probably needs modification for SGE 6.1. See also the fixmes.
[256]	672
[507]	673	class NoJobs (Exception):
	674	"""Exception raised by empty job list in qstat output."""
	675	pass
[256]	676
[507]	677	class SgeQstatXMLParser(xml.sax.handler.ContentHandler):
	678	"""SAX handler for XML output from Sun Grid Engine's `qstat'."""
[318]	679
[507]	680	def __init__(self):
	681	self.value = ""
	682	self.joblist = []
	683	self.job = {}
	684	self.queue = ""
	685	self.in_joblist = False
	686	self.lrequest = False
	687	xml.sax.handler.ContentHandler.__init__(self)
[318]	688
[507]	689	# The structure of the output is as follows. Unfortunately
	690	# it's voluminous, and probably doesn't scale to large
	691	# clusters/queues.
[318]	692
[507]	693	# <detailed_job_info xmlns:xsd="http://www.w3.org/2001/XMLSchema">
	694	# <djob_info>
	695	# <qmaster_response> <!-- job -->
	696	# ...
	697	# <JB_ja_template>
	698	# <ulong_sublist>
	699	# ... <!-- start_time, state ... -->
	700	# </ulong_sublist>
	701	# </JB_ja_template>
	702	# <JB_ja_tasks>
	703	# <ulong_sublist>
	704	# ... <!-- task info
	705	# </ulong_sublist>
	706	# ...
	707	# </JB_ja_tasks>
	708	# ...
	709	# </qmaster_response>
	710	# </djob_info>
	711	# <messages>
	712	# ...
[318]	713
[507]	714	# NB. We might treat each task as a separate job, like
	715	# straight qstat output, but the web interface expects jobs to
	716	# be identified by integers, not, say, <job number>.<task>.
[318]	717
[507]	718	# So, I lied. If the job list is empty, we get invalid XML
	719	# like this, which we need to defend against:
[318]	720
[507]	721	# <unknown_jobs xmlns:xsd="http://www.w3.org/2001/XMLSchema">
	722	# <>
	723	# <ST_name>*</ST_name>
	724	# </>
	725	# </unknown_jobs>
[318]	726
[507]	727	def startElement(self, name, attrs):
	728	self.value = ""
	729	if name == "djob_info": # job list
	730	self.in_joblist = True
	731	elif name == "qmaster_response" and self.in_joblist: # job
	732	self.job = {"job_state": "U", "slots": 0,
	733	"nodes": [], "queued_timestamp": "",
	734	"queued_timestamp": "", "queue": "",
	735	"ppn": "0", "RN_max": 0,
	736	# fixme in endElement
	737	"requested_memory": 0, "requested_time": 0
	738	}
	739	self.joblist.append(self.job)
	740	elif name == "qstat_l_requests": # resource request
	741	self.lrequest = True
	742	elif name == "unknown_jobs":
	743	raise NoJobs
[318]	744
[507]	745	def characters(self, ch):
	746	self.value += ch
[318]	747
[507]	748	def endElement(self, name):
	749	"""Snarf job elements contents into job dictionary.
	750	Translate keys if appropriate."""
[318]	751
[507]	752	name_trans = {
	753	"JB_job_number": "number",
	754	"JB_job_name": "name", "JB_owner": "owner",
	755	"queue_name": "queue", "JAT_start_time": "start_timestamp",
	756	"JB_submission_time": "queued_timestamp"
	757	}
	758	value = self.value
[318]	759
[507]	760	if name == "djob_info":
	761	self.in_joblist = False
	762	self.job = {}
	763	elif name == "JAT_master_queue":
	764	self.job["queue"] = value.split("@")[0]
	765	elif name == "JG_qhostname":
	766	if not (value in self.job["nodes"]):
	767	self.job["nodes"].append(value)
	768	elif name == "JG_slots": # slots in use
	769	self.job["slots"] += int(value)
	770	elif name == "RN_max": # requested slots (tasks or parallel)
	771	self.job["RN_max"] = max (self.job["RN_max"],
	772	int(value))
	773	elif name == "JAT_state": # job state (bitwise or)
	774	value = int (value)
	775	# Status values from sge_jobL.h
	776	#define JIDLE 0x00000000
	777	#define JHELD 0x00000010
	778	#define JMIGRATING 0x00000020
	779	#define JQUEUED 0x00000040
	780	#define JRUNNING 0x00000080
	781	#define JSUSPENDED 0x00000100
	782	#define JTRANSFERING 0x00000200
	783	#define JDELETED 0x00000400
	784	#define JWAITING 0x00000800
	785	#define JEXITING 0x00001000
	786	#define JWRITTEN 0x00002000
	787	#define JSUSPENDED_ON_THRESHOLD 0x00010000
	788	#define JFINISHED 0x00010000
	789	if value & 0x80:
	790	self.job["status"] = "R"
	791	elif value & 0x40:
	792	self.job["status"] = "Q"
	793	else:
	794	self.job["status"] = "O" # `other'
	795	elif name == "CE_name" and self.lrequest and self.value in \
	796	("h_cpu", "s_cpu", "cpu", "h_core", "s_core"):
	797	# We're in a container for an interesting resource
	798	# request; record which type.
	799	self.lrequest = self.value
	800	elif name == "CE_doubleval" and self.lrequest:
	801	# if we're in a container for an interesting
	802	# resource request, use the maxmimum of the hard
	803	# and soft requests to record the requested CPU
	804	# or core. Fixme: I'm not sure if this logic is
	805	# right.
	806	if self.lrequest in ("h_core", "s_core"):
	807	self.job["requested_memory"] = \
	808	max (float (value),
	809	self.job["requested_memory"])
	810	# Fixme: Check what cpu means, c.f [hs]_cpu.
	811	elif self.lrequest in ("h_cpu", "s_cpu", "cpu"):
	812	self.job["requested_time"] = \
	813	max (float (value),
	814	self.job["requested_time"])
	815	elif name == "qstat_l_requests":
	816	self.lrequest = False
	817	elif self.job and self.in_joblist:
	818	if name in name_trans:
	819	name = name_trans[name]
	820	self.job[name] = value
[318]	821
[507]	822	# Abstracted from PBS original.
	823	# Fixme: Is it worth (or appropriate for PBS) sorting the result?
	824	def do_nodelist (nodes):
	825	"""Translate node list as appropriate."""
	826	nodeslist = [ ]
	827	my_domain = fqdn_parts(socket.getfqdn())[1]
	828	for node in nodes:
	829	host = node.split( '/' )[0] # not relevant for SGE
	830	h, host_domain = fqdn_parts(host)
	831	if host_domain == my_domain:
	832	host = h
	833	if nodeslist.count( host ) == 0:
	834	for translate_pattern in BATCH_HOST_TRANSLATE:
	835	if translate_pattern.find( '/' ) != -1:
	836	translate_orig = \
	837	translate_pattern.split( '/' )[1]
	838	translate_new = \
	839	translate_pattern.split( '/' )[2]
	840	host = re.sub( translate_orig,
	841	translate_new, host )
	842	if not host in nodeslist:
	843	nodeslist.append( host )
	844	return nodeslist
[318]	845
	846	class SgeDataGatherer(DataGatherer):
	847
[507]	848	jobs = {}
[61]	849
[318]	850	def __init__( self ):
[507]	851	self.jobs = {}
[318]	852	self.timeoffset = 0
	853	self.dp = DataProcessor()
	854
[507]	855	def getJobData( self ):
[318]	856	"""Gather all data on current jobs in SGE"""
	857
[507]	858	import popen2
[318]	859
[507]	860	self.cur_time = 0
	861	queues = ""
	862	if QUEUE: # only for specific queues
	863	# Fixme: assumes queue names don't contain single
	864	# quote or comma. Don't know what the SGE rules are.
	865	queues = " -q '" + string.join (QUEUE, ",") + "'"
	866	# Note the comment in SgeQstatXMLParser about scaling with
	867	# this method of getting data. I haven't found better one.
	868	# Output with args `-xml -ext -f -r' is easier to parse
	869	# in some ways, harder in others, but it doesn't provide
	870	# the submission time, at least.
	871	piping = popen2.Popen3("qstat -u '' -j '' -xml" + queues,
	872	True)
	873	qstatparser = SgeQstatXMLParser()
	874	parse_err = 0
	875	try:
	876	xml.sax.parse(piping.fromchild, qstatparser)
	877	except NoJobs:
	878	pass
	879	except:
	880	parse_err = 1
	881	if piping.wait():
	882	debug_msg(10,
	883	"qstat error, skipping until next polling interval: "
	884	+ piping.childerr.readline())
	885	return None
	886	elif parse_err:
	887	debug_msg(10, "Bad XML output from qstat"())
	888	exit (1)
	889	for f in piping.fromchild, piping.tochild, piping.childerr:
	890	f.close()
[318]	891	self.cur_time = time.time()
[507]	892	jobs_processed = []
	893	for job in qstatparser.joblist:
	894	job_id = job["number"]
	895	if job["status"] in [ 'Q', 'R' ]:
	896	jobs_processed.append(job_id)
	897	if job["status"] == "R":
	898	job["nodes"] = do_nodelist (job["nodes"])
	899	# Fixme: Is this right?
	900	job["ppn"] = float(job["slots"]) / \
	901	len(job["nodes"])
	902	if DETECT_TIME_DIFFS:
	903	# If a job start is later than our
	904	# current date, that must mean
	905	# the SGE server's time is later
	906	# than our local time.
	907	start_timestamp = \
	908	int (job["start_timestamp"])
	909	if start_timestamp > \
	910	int(self.cur_time) + \
	911	int(self.timeoffset):
[318]	912
[507]	913	self.timeoffset = \
	914	start_timestamp - \
	915	int(self.cur_time)
	916	else:
	917	# fixme: Note sure what this should be:
	918	job["ppn"] = job["RN_max"]
	919	job["nodes"] = "1"
[318]	920
[507]	921	myAttrs = {}
	922	for attr in ["name", "queue", "owner",
	923	"requested_time", "status",
	924	"requested_memory", "ppn",
	925	"start_timestamp", "queued_timestamp"]:
	926	myAttrs[attr] = str(job[attr])
	927	myAttrs["nodes"] = job["nodes"]
	928	myAttrs["reported"] = str(int(self.cur_time) + \
	929	int(self.timeoffset))
	930	myAttrs["domain"] = fqdn_parts(socket.getfqdn())[1]
	931	myAttrs["poll_interval"] = str(BATCH_POLL_INTERVAL)
[318]	932
[507]	933	if self.jobDataChanged(self.jobs, job_id, myAttrs) \
	934	and myAttrs["status"] in ["R", "Q"]:
	935	self.jobs[job_id] = myAttrs
	936	for id, attrs in self.jobs.items():
	937	if id not in jobs_processed:
	938	del self.jobs[id]
[318]	939
[355]	940	class PbsDataGatherer( DataGatherer ):
[318]	941
	942	"""This is the DataGatherer for PBS and Torque"""
	943
[256]	944	global PBSQuery
	945
[23]	946	def __init__( self ):
[354]	947
[68]	948	"""Setup appropriate variables"""
[23]	949
[354]	950	self.jobs = { }
	951	self.timeoffset = 0
	952	self.dp = DataProcessor()
	953
[91]	954	self.initPbsQuery()
[23]	955
[91]	956	def initPbsQuery( self ):
	957
[354]	958	self.pq = None
	959
[256]	960	if( BATCH_SERVER ):
[354]	961
	962	self.pq = PBSQuery( BATCH_SERVER )
[174]	963	else:
[354]	964	self.pq = PBSQuery()
[91]	965
[348]	966	def getJobData( self ):
[354]	967
[68]	968	"""Gather all data on current jobs in Torque"""
[26]	969
[354]	970	joblist = {}
[359]	971	self.cur_time = 0
[349]	972
[359]	973	try:
	974	joblist = self.pq.getjobs()
	975	self.cur_time = time.time()
[354]	976
[359]	977	except PBSError, detail:
[354]	978
[359]	979	debug_msg( 10, "Caught PBS unavailable, skipping until next polling interval: " + str( detail ) )
	980	return None
[354]	981
	982	jobs_processed = [ ]
[26]	983
	984	for name, attrs in joblist.items():
[508]	985	display_queue = 1
[354]	986	job_id = name.split( '.' )[0]
[26]	987
[354]	988	name = self.getAttr( attrs, 'Job_Name' )
	989	queue = self.getAttr( attrs, 'queue' )
[317]	990
	991	if QUEUE:
[508]	992	for q in QUEUE:
	993	if q == queue:
	994	display_queue = 1
	995	break
	996	else:
	997	display_queue = 0
	998	continue
	999	if display_queue == 0:
	1000	continue
[317]	1001
	1002
[354]	1003	owner = self.getAttr( attrs, 'Job_Owner' ).split( '@' )[0]
	1004	requested_time = self.getAttr( attrs, 'Resource_List.walltime' )
	1005	requested_memory = self.getAttr( attrs, 'Resource_List.mem' )
[95]	1006
[354]	1007	mynoderequest = self.getAttr( attrs, 'Resource_List.nodes' )
[95]	1008
[354]	1009	ppn = ''
[281]	1010
[26]	1011	if mynoderequest.find( ':' ) != -1 and mynoderequest.find( 'ppn' ) != -1:
[95]	1012
[354]	1013	mynoderequest_fields = mynoderequest.split( ':' )
[281]	1014
	1015	for mynoderequest_field in mynoderequest_fields:
	1016
	1017	if mynoderequest_field.find( 'ppn' ) != -1:
	1018
[354]	1019	ppn = mynoderequest_field.split( 'ppn=' )[1]
[281]	1020
[354]	1021	status = self.getAttr( attrs, 'job_state' )
[25]	1022
[450]	1023	if status in [ 'Q', 'R' ]:
	1024
	1025	jobs_processed.append( job_id )
	1026
[354]	1027	queued_timestamp = self.getAttr( attrs, 'ctime' )
[243]	1028
[95]	1029	if status == 'R':
[133]	1030
[354]	1031	start_timestamp = self.getAttr( attrs, 'mtime' )
	1032	nodes = self.getAttr( attrs, 'exec_host' ).split( '+' )
[133]	1033
[507]	1034	nodeslist = do_nodelist( nodes )
[354]	1035
[185]	1036	if DETECT_TIME_DIFFS:
	1037
	1038	# If a job start if later than our current date,
	1039	# that must mean the Torque server's time is later
	1040	# than our local time.
	1041
[354]	1042	if int( start_timestamp ) > int( int( self.cur_time ) + int( self.timeoffset ) ):
[185]	1043
[354]	1044	self.timeoffset = int( int(start_timestamp) - int(self.cur_time) )
[185]	1045
[133]	1046	elif status == 'Q':
[95]	1047
[451]	1048	# 'mynodequest' can be a string in the following syntax according to the
	1049	# Torque Administator's manual:
	1050	#
	1051	# {<node_count> \| <hostname>}[:ppn=<ppn>][:<property>[:<property>]...][+ ...]
	1052	# {<node_count> \| <hostname>}[:ppn=<ppn>][:<property>[:<property>]...][+ ...]
	1053	# etc
	1054	#
	1055
	1056	#
	1057	# For now we only count the amount of nodes request and ignore properties
	1058	#
	1059
[354]	1060	start_timestamp = ''
	1061	count_mynodes = 0
	1062
[133]	1063	for node in mynoderequest.split( '+' ):
[67]	1064
[451]	1065	# Just grab the {node_count\|hostname} part and ignore properties
	1066	#
[354]	1067	nodepart = node.split( ':' )[0]
[67]	1068
[451]	1069	# Let's assume a node_count value
	1070	#
	1071	numeric_node = 1
	1072
	1073	# Chop the value up into characters
	1074	#
[133]	1075	for letter in nodepart:
[67]	1076
[451]	1077	# If this char is not a digit (0-9), this must be a hostname
	1078	#
[133]	1079	if letter not in string.digits:
	1080
[354]	1081	numeric_node = 0
[133]	1082
[451]	1083	# If this is a hostname, just count this as one (1) node
	1084	#
[133]	1085	if not numeric_node:
[354]	1086
	1087	count_mynodes = count_mynodes + 1
[133]	1088	else:
[451]	1089
	1090	# If this a number, it must be the node_count
	1091	# and increase our count with it's value
	1092	#
[327]	1093	try:
[354]	1094	count_mynodes = count_mynodes + int( nodepart )
	1095
[327]	1096	except ValueError, detail:
[354]	1097
[451]	1098	# When we arrive here I must be bugged or very confused
	1099	# THIS SHOULD NOT HAPPEN!
	1100	#
[327]	1101	debug_msg( 10, str( detail ) )
	1102	debug_msg( 10, "Encountered weird node in Resources_List?!" )
	1103	debug_msg( 10, 'nodepart = ' + str( nodepart ) )
	1104	debug_msg( 10, 'job = ' + str( name ) )
	1105	debug_msg( 10, 'attrs = ' + str( attrs ) )
[133]	1106
[354]	1107	nodeslist = str( count_mynodes )
[172]	1108	else:
[354]	1109	start_timestamp = ''
	1110	nodeslist = ''
[133]	1111
[354]	1112	myAttrs = { }
[26]	1113
[471]	1114	myAttrs[ 'name' ] = str( name )
[354]	1115	myAttrs[ 'queue' ] = str( queue )
	1116	myAttrs[ 'owner' ] = str( owner )
	1117	myAttrs[ 'requested_time' ] = str( requested_time )
	1118	myAttrs[ 'requested_memory' ] = str( requested_memory )
	1119	myAttrs[ 'ppn' ] = str( ppn )
	1120	myAttrs[ 'status' ] = str( status )
	1121	myAttrs[ 'start_timestamp' ] = str( start_timestamp )
	1122	myAttrs[ 'queued_timestamp' ] = str( queued_timestamp )
	1123	myAttrs[ 'reported' ] = str( int( int( self.cur_time ) + int( self.timeoffset ) ) )
	1124	myAttrs[ 'nodes' ] = nodeslist
[507]	1125	myAttrs[ 'domain' ] = fqdn_parts( socket.getfqdn() )[1]
[354]	1126	myAttrs[ 'poll_interval' ] = str( BATCH_POLL_INTERVAL )
	1127
[348]	1128	if self.jobDataChanged( self.jobs, job_id, myAttrs ) and myAttrs['status'] in [ 'R', 'Q' ]:
[61]	1129
[354]	1130	self.jobs[ job_id ] = myAttrs
[26]	1131
[348]	1132	for id, attrs in self.jobs.items():
[76]	1133
	1134	if id not in jobs_processed:
	1135
	1136	# This one isn't there anymore; toedeledoki!
	1137	#
[348]	1138	del self.jobs[ id ]
[76]	1139
[363]	1140	#
	1141	# Gmetric by Nick Galbreath - nickg(a.t)modp(d.o.t)com
	1142	# Version 1.0 - 21-April2-2007
	1143	# http://code.google.com/p/embeddedgmetric/
	1144	#
	1145	# Modified by: Ramon Bastiaans
	1146	# For the Job Monarch Project, see: https://subtrac.sara.nl/oss/jobmonarch/
	1147	#
	1148	# added: DEFAULT_TYPE for Gmetric's
	1149	# added: checkHostProtocol to determine if target is multicast or not
	1150	# changed: allow default for Gmetric constructor
	1151	# changed: allow defaults for all send() values except dmax
	1152	#
	1153
[362]	1154	GMETRIC_DEFAULT_TYPE = 'string'
	1155	GMETRIC_DEFAULT_HOST = '127.0.0.1'
	1156	GMETRIC_DEFAULT_PORT = '8649'
[431]	1157	GMETRIC_DEFAULT_UNITS = ''
[362]	1158
	1159	class Gmetric:
	1160
	1161	global GMETRIC_DEFAULT_HOST, GMETRIC_DEFAULT_PORT
	1162
	1163	slope = { 'zero' : 0, 'positive' : 1, 'negative' : 2, 'both' : 3, 'unspecified' : 4 }
	1164	type = ( '', 'string', 'uint16', 'int16', 'uint32', 'int32', 'float', 'double', 'timestamp' )
	1165	protocol = ( 'udp', 'multicast' )
	1166
	1167	def __init__( self, host=GMETRIC_DEFAULT_HOST, port=GMETRIC_DEFAULT_PORT ):
	1168
	1169	global GMETRIC_DEFAULT_TYPE
	1170
	1171	self.prot = self.checkHostProtocol( host )
	1172	self.msg = xdrlib.Packer()
	1173	self.socket = socket.socket( socket.AF_INET, socket.SOCK_DGRAM )
	1174
	1175	if self.prot not in self.protocol:
	1176
	1177	raise ValueError( "Protocol must be one of: " + str( self.protocol ) )
	1178
	1179	if self.prot == 'multicast':
	1180
[471]	1181	# Set multicast options
	1182	#
[362]	1183	self.socket.setsockopt( socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 20 )
	1184
	1185	self.hostport = ( host, int( port ) )
	1186	self.slopestr = 'both'
	1187	self.tmax = 60
	1188
	1189	def checkHostProtocol( self, ip ):
	1190
[471]	1191	"""Detect if a ip adress is a multicast address"""
	1192
[362]	1193	MULTICAST_ADDRESS_MIN = ( "224", "0", "0", "0" )
	1194	MULTICAST_ADDRESS_MAX = ( "239", "255", "255", "255" )
	1195
	1196	ip_fields = ip.split( '.' )
	1197
	1198	if ip_fields >= MULTICAST_ADDRESS_MIN and ip_fields <= MULTICAST_ADDRESS_MAX:
	1199
	1200	return 'multicast'
	1201	else:
	1202	return 'udp'
	1203
[431]	1204	def send( self, name, value, dmax, typestr = '', units = '' ):
[362]	1205
[409]	1206	if len( units ) == 0:
[431]	1207	units = GMETRIC_DEFAULT_UNITS
[471]	1208
[431]	1209	if len( typestr ) == 0:
	1210	typestr = GMETRIC_DEFAULT_TYPE
[362]	1211
[424]	1212	msg = self.makexdr( name, value, typestr, units, self.slopestr, self.tmax, dmax )
[409]	1213
[362]	1214	return self.socket.sendto( msg, self.hostport )
	1215
	1216	def makexdr( self, name, value, typestr, unitstr, slopestr, tmax, dmax ):
	1217
	1218	if slopestr not in self.slope:
	1219
	1220	raise ValueError( "Slope must be one of: " + str( self.slope.keys() ) )
	1221
	1222	if typestr not in self.type:
	1223
	1224	raise ValueError( "Type must be one of: " + str( self.type ) )
	1225
	1226	if len( name ) == 0:
	1227
	1228	raise ValueError( "Name must be non-empty" )
	1229
	1230	self.msg.reset()
	1231	self.msg.pack_int( 0 )
	1232	self.msg.pack_string( typestr )
	1233	self.msg.pack_string( name )
	1234	self.msg.pack_string( str( value ) )
	1235	self.msg.pack_string( unitstr )
	1236	self.msg.pack_int( self.slope[ slopestr ] )
	1237	self.msg.pack_uint( int( tmax ) )
	1238	self.msg.pack_uint( int( dmax ) )
	1239
	1240	return self.msg.get_buffer()
	1241
[26]	1242	def printTime( ):
[354]	1243
[65]	1244	"""Print current time/date in human readable format for log/debug"""
[26]	1245
	1246	return time.strftime("%a, %d %b %Y %H:%M:%S")
	1247
	1248	def debug_msg( level, msg ):
[354]	1249
[65]	1250	"""Print msg if at or above current debug level"""
[26]	1251
[377]	1252	global DAEMONIZE, DEBUG_LEVEL, SYSLOG_LEVEL
	1253
[373]	1254	if (not DAEMONIZE and DEBUG_LEVEL >= level):
	1255	sys.stderr.write( msg + '\n' )
[26]	1256
[373]	1257	if (DAEMONIZE and USE_SYSLOG and SYSLOG_LEVEL >= level):
	1258	syslog.syslog( msg )
	1259
[307]	1260	def write_pidfile():
	1261
[471]	1262	# Write pidfile if PIDFILE is set
	1263	#
[307]	1264	if PIDFILE:
	1265
[354]	1266	pid = os.getpid()
	1267
[471]	1268	pidfile = open( PIDFILE, 'w' )
[354]	1269
	1270	pidfile.write( str( pid ) )
[307]	1271	pidfile.close()
	1272
[23]	1273	def main():
[354]	1274
[65]	1275	"""Application start"""
[23]	1276
[352]	1277	global PBSQuery, PBSError
[373]	1278	global SYSLOG_FACILITY, USE_SYSLOG, BATCH_API, DAEMONIZE
[256]	1279
[212]	1280	if not processArgs( sys.argv[1:] ):
[354]	1281
[212]	1282	sys.exit( 1 )
	1283
[471]	1284	# Load appropriate DataGatherer depending on which BATCH_API is set
	1285	# and any required modules for the Gatherer
	1286	#
[256]	1287	if BATCH_API == 'pbs':
	1288
	1289	try:
[282]	1290	from PBSQuery import PBSQuery, PBSError
[256]	1291
	1292	except ImportError:
	1293
[373]	1294	debug_msg( 0, "FATAL ERROR: BATCH_API set to 'pbs' but python module 'pbs_python' is not installed" )
[256]	1295	sys.exit( 1 )
	1296
	1297	gather = PbsDataGatherer()
	1298
	1299	elif BATCH_API == 'sge':
	1300
[507]	1301	# Tested with SGE 6.0u11.
	1302	# debug_msg( 0, "FATAL ERROR: BATCH_API 'sge' implementation is currently broken, check future releases" )
[368]	1303
[507]	1304	# sys.exit( 1 )
[368]	1305
[347]	1306	gather = SgeDataGatherer()
[256]	1307
	1308	else:
[373]	1309	debug_msg( 0, "FATAL ERROR: unknown BATCH_API '" + BATCH_API + "' is not supported" )
[354]	1310
[256]	1311	sys.exit( 1 )
	1312
[373]	1313	if( DAEMONIZE and USE_SYSLOG ):
	1314
	1315	syslog.openlog( 'jobmond', syslog.LOG_NOWAIT, SYSLOG_FACILITY )
	1316
	1317
[26]	1318	if DAEMONIZE:
[354]	1319
[26]	1320	gather.daemon()
	1321	else:
	1322	gather.run()
[23]	1323
[256]	1324	# wh00t? someone started me! :)
[65]	1325	#
[23]	1326	if __name__ == '__main__':
	1327	main()

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

Download in other formats: