Context Navigation

jobarchived.py @ 224

Last change on this file since 224 was 224, checked in by bastiaans, 18 years ago

jobarchived/examples:

added

jobmond/jobmond.py:

removed config option comments

jobarchived/jobarchived.py:

removed config option comments
added RRDTOOL option

jobmond/jobmond.conf:

added config option comments

jobarchived/jobarchived.conf:

added config option comments
added RRDTOOL option

File size: 35.5 KB

Rev	Line
[3]	1	#!/usr/bin/env python
	2
[214]	3	import getopt, syslog, ConfigParser, sys
[3]	4
[214]	5	def processArgs( args ):
[6]	6
[214]	7	SHORT_L = 'c:'
	8	LONG_L = 'config='
[169]	9
[214]	10	config_filename = None
[169]	11
[214]	12	try:
[169]	13
[214]	14	opts, args = getopt.getopt( args, SHORT_L, LONG_L )
[9]	15
[214]	16	except getopt.error, detail:
[60]	17
[214]	18	print detail
	19	sys.exit(1)
[9]	20
[214]	21	for opt, value in opts:
[60]	22
[214]	23	if opt in [ '--config', '-c' ]:
[13]	24
[214]	25	config_filename = value
[198]	26
[214]	27	if not config_filename:
[60]	28
[214]	29	config_filename = '/etc/jobarchived.conf'
[22]	30
[214]	31	try:
	32	return loadConfig( config_filename )
[13]	33
[214]	34	except ConfigParser.NoOptionError, detail:
	35
	36	print detail
	37	sys.exit( 1 )
	38
	39	def loadConfig( filename ):
	40
	41	def getlist( cfg_string ):
	42
	43	my_list = [ ]
	44
	45	for item_txt in cfg_string.split( ',' ):
	46
	47	sep_char = None
	48
	49	item_txt = item_txt.strip()
	50
	51	for s_char in [ "'", '"' ]:
	52
	53	if item_txt.find( s_char ) != -1:
	54
	55	if item_txt.count( s_char ) != 2:
	56
	57	print 'Missing quote: %s' %item_txt
	58	sys.exit( 1 )
	59
	60	else:
	61
	62	sep_char = s_char
	63	break
	64
	65	if sep_char:
	66
	67	item_txt = item_txt.split( sep_char )[1]
	68
	69	my_list.append( item_txt )
	70
	71	return my_list
	72
	73	cfg = ConfigParser.ConfigParser()
	74
	75	cfg.read( filename )
	76
[224]	77	global DEBUG_LEVEL, USE_SYSLOG, SYSLOG_LEVEL, SYSLOG_FACILITY, GMETAD_CONF, ARCHIVE_XMLSOURCE, ARCHIVE_DATASOURCES, ARCHIVE_PATH, ARCHIVE_HOURS_PER_RRD, ARCHIVE_EXCLUDE_METRICS, JOB_SQL_DBASE, DAEMONIZE, RRDTOOL
[214]	78
	79	ARCHIVE_PATH = cfg.get( 'DEFAULT', 'ARCHIVE_PATH' )
	80
	81	ARCHIVE_HOURS_PER_RRD = cfg.getint( 'DEFAULT', 'ARCHIVE_HOURS_PER_RRD' )
	82
	83	DEBUG_LEVEL = cfg.getint( 'DEFAULT', 'DEBUG_LEVEL' )
	84
	85	USE_SYSLOG = cfg.getboolean( 'DEFAULT', 'USE_SYSLOG' )
	86
	87	SYSLOG_LEVEL = cfg.getint( 'DEFAULT', 'SYSLOG_LEVEL' )
	88
	89	try:
	90
	91	SYSLOG_FACILITY = eval( 'syslog.LOG_' + cfg.get( 'DEFAULT', 'SYSLOG_FACILITY' ) )
	92
	93	except AttributeError, detail:
	94
	95	print 'Unknown syslog facility'
	96	sys.exit( 1 )
	97
	98	GMETAD_CONF = cfg.get( 'DEFAULT', 'GMETAD_CONF' )
	99
	100	ARCHIVE_XMLSOURCE = cfg.get( 'DEFAULT', 'ARCHIVE_XMLSOURCE' )
	101
	102	ARCHIVE_DATASOURCES = getlist( cfg.get( 'DEFAULT', 'ARCHIVE_DATASOURCES' ) )
	103
	104	ARCHIVE_EXCLUDE_METRICS = getlist( cfg.get( 'DEFAULT', 'ARCHIVE_EXCLUDE_METRICS' ) )
	105
	106	JOB_SQL_DBASE = cfg.get( 'DEFAULT', 'JOB_SQL_DBASE' )
	107
	108	DAEMONIZE = cfg.getboolean( 'DEFAULT', 'DAEMONIZE' )
	109
[224]	110	RRDTOOL = cfg.get( 'DEFAULT', 'RRDTOOL' )
	111
[214]	112	return True
	113
[17]	114	# What XML data types not to store
[13]	115	#
[17]	116	UNSUPPORTED_ARCHIVE_TYPES = [ 'string' ]
[9]	117
[47]	118	# Maximum time (in seconds) a parsethread may run
	119	#
	120	PARSE_TIMEOUT = 60
	121
	122	# Maximum time (in seconds) a storethread may run
	123	#
	124	STORE_TIMEOUT = 360
	125
[8]	126	"""
[224]	127	The Job Archiving Daemon
[8]	128	"""
	129
[214]	130	from types import *
	131
	132	import DBClass
	133	import xml.sax, xml.sax.handler, socket, string, os, os.path, time, thread, threading, random, re
	134
[84]	135	class DataSQLStore:
	136
	137	db_vars = None
	138	dbc = None
	139
	140	def __init__( self, hostname, database ):
	141
	142	self.db_vars = DBClass.InitVars(DataBaseName=database,
	143	User='root',
	144	Host=hostname,
	145	Password='',
	146	Dictionary='true')
	147
	148	try:
	149	self.dbc = DBClass.DB(self.db_vars)
	150	except DBClass.DBError, details:
[169]	151	debug_msg( 0, 'FATAL ERROR: Unable to connect to database!: ' +str(details) )
[84]	152	sys.exit(1)
	153
	154	def setDatabase(self, statement):
	155	ret = self.doDatabase('set', statement)
	156	return ret
	157
	158	def getDatabase(self, statement):
	159	ret = self.doDatabase('get', statement)
	160	return ret
	161
	162	def doDatabase(self, type, statement):
	163
	164	debug_msg( 6, 'doDatabase(): %s: %s' %(type, statement) )
	165	try:
	166	if type == 'set':
	167	result = self.dbc.Set( statement )
	168	self.dbc.Commit()
	169	elif type == 'get':
	170	result = self.dbc.Get( statement )
	171
	172	except DBClass.DBError, detail:
	173	operation = statement.split(' ')[0]
[169]	174	debug_msg( 0, 'FATAL ERROR: ' +operation+ ' on database failed while doing ['+statement+'] full msg: '+str(detail) )
[84]	175	sys.exit(1)
	176
	177	debug_msg( 6, 'doDatabase(): result: %s' %(result) )
	178	return result
	179
[191]	180	def getJobNodeId( self, job_id, node_id ):
	181
	182	id = self.getDatabase( "SELECT job_id,node_id FROM job_nodes WHERE job_id = '%s' AND node_id = '%s'" %(job_id, node_id) )
	183	if len( id ) > 0:
	184
	185	if len( id[0] ) > 0 and id[0] != '':
	186
	187	return 1
	188
	189	return 0
	190
[84]	191	def getNodeId( self, hostname ):
	192
[89]	193	id = self.getDatabase( "SELECT node_id FROM nodes WHERE node_hostname = '%s'" %hostname )
[84]	194
[89]	195	if len( id ) > 0:
[84]	196
[89]	197	id = id[0][0]
	198
[84]	199	return id
	200	else:
	201	return None
	202
	203	def getNodeIds( self, hostnames ):
	204
	205	ids = [ ]
	206
	207	for node in hostnames:
	208
	209	id = self.getNodeId( node )
	210
	211	if id:
	212	ids.append( id )
	213
	214	return ids
	215
	216	def getJobId( self, jobid ):
	217
	218	id = self.getDatabase( "SELECT job_id FROM jobs WHERE job_id = '%s'" %jobid )
	219
	220	if id:
[89]	221	id = id[0][0]
[84]	222
	223	return id
	224	else:
	225	return None
	226
	227	def addJob( self, job_id, jobattrs ):
	228
	229	if not self.getJobId( job_id ):
	230
	231	self.mutateJob( 'insert', job_id, jobattrs )
	232	else:
	233	self.mutateJob( 'update', job_id, jobattrs )
	234
	235	def mutateJob( self, action, job_id, jobattrs ):
	236
	237	job_values = [ 'name', 'queue', 'owner', 'requested_time', 'requested_memory', 'ppn', 'status', 'start_timestamp', 'stop_timestamp' ]
	238
	239	insert_col_str = 'job_id'
	240	insert_val_str = "'%s'" %job_id
	241	update_str = None
	242
	243	debug_msg( 6, 'mutateJob(): %s %s' %(action,job_id))
	244
[99]	245	ids = [ ]
	246
[84]	247	for valname, value in jobattrs.items():
	248
[96]	249	if valname in job_values and value != '':
[84]	250
	251	column_name = 'job_' + valname
	252
	253	if action == 'insert':
	254
	255	if not insert_col_str:
	256	insert_col_str = column_name
	257	else:
	258	insert_col_str = insert_col_str + ',' + column_name
	259
	260	if not insert_val_str:
	261	insert_val_str = value
	262	else:
	263	insert_val_str = insert_val_str + ",'%s'" %value
	264
	265	elif action == 'update':
	266
	267	if not update_str:
	268	update_str = "%s='%s'" %(column_name, value)
	269	else:
	270	update_str = update_str + ",%s='%s'" %(column_name, value)
	271
[90]	272	elif valname == 'nodes' and value:
[84]	273
[191]	274	node_valid = 1
[190]	275
	276	if len(value) == 1:
	277
[191]	278	if jobattrs['status'] == 'Q':
[190]	279
[191]	280	node_valid = 0
[190]	281
[191]	282	else:
[190]	283
[191]	284	node_valid = 0
[190]	285
[191]	286	for node_char in str(value[0]):
[190]	287
[191]	288	if string.find( string.digits, node_char ) != -1 and not node_valid:
[190]	289
[191]	290	node_valid = 1
[84]	291
[191]	292	if node_valid:
	293
	294	ids = self.addNodes( value, jobattrs['domain'] )
	295
[84]	296	if action == 'insert':
	297
	298	self.setDatabase( "INSERT INTO jobs ( %s ) VALUES ( %s )" %( insert_col_str, insert_val_str ) )
[86]	299
[84]	300	elif action == 'update':
	301
[89]	302	self.setDatabase( "UPDATE jobs SET %s WHERE job_id=%s" %(update_str, job_id) )
[84]	303
[191]	304	if len( ids ) > 0:
	305	self.addJobNodes( job_id, ids )
[190]	306
[154]	307	def addNodes( self, hostnames, domain ):
[84]	308
[98]	309	ids = [ ]
	310
[84]	311	for node in hostnames:
	312
[154]	313	node = '%s.%s' %( node, domain )
[84]	314	id = self.getNodeId( node )
	315
	316	if not id:
	317	self.setDatabase( "INSERT INTO nodes ( node_hostname ) VALUES ( '%s' )" %node )
[98]	318	id = self.getNodeId( node )
[84]	319
[98]	320	ids.append( id )
	321
	322	return ids
	323
[86]	324	def addJobNodes( self, jobid, nodes ):
	325
	326	for node in nodes:
	327
[191]	328	if not self.getJobNodeId( jobid, node ):
	329
	330	self.addJobNode( jobid, node )
	331
[84]	332	def addJobNode( self, jobid, nodeid ):
	333
	334	self.setDatabase( "INSERT INTO job_nodes (job_id,node_id) VALUES ( %s,%s )" %(jobid, nodeid) )
	335
	336	def storeJobInfo( self, jobid, jobattrs ):
	337
	338	self.addJob( jobid, jobattrs )
	339
[37]	340	class RRDMutator:
[63]	341	"""A class for performing RRD mutations"""
[37]	342
[224]	343	binary = RRDTOOL
[37]	344
	345	def __init__( self, binary=None ):
[63]	346	"""Set alternate binary if supplied"""
[37]	347
	348	if binary:
	349	self.binary = binary
	350
	351	def create( self, filename, args ):
[63]	352	"""Create a new rrd with args"""
	353
[40]	354	return self.perform( 'create', '"' + filename + '"', args )
[37]	355
	356	def update( self, filename, args ):
[63]	357	"""Update a rrd with args"""
	358
[40]	359	return self.perform( 'update', '"' + filename + '"', args )
[37]	360
[42]	361	def grabLastUpdate( self, filename ):
[63]	362	"""Determine the last update time of filename rrd"""
[42]	363
	364	last_update = 0
	365
[53]	366	debug_msg( 8, self.binary + ' info "' + filename + '"' )
	367
[42]	368	for line in os.popen( self.binary + ' info "' + filename + '"' ).readlines():
	369
	370	if line.find( 'last_update') != -1:
	371
	372	last_update = line.split( ' = ' )[1]
	373
	374	if last_update:
	375	return last_update
	376	else:
	377	return 0
	378
[40]	379	def perform( self, action, filename, args ):
[63]	380	"""Perform action on rrd filename with args"""
[37]	381
	382	arg_string = None
	383
[40]	384	if type( args ) is not ListType:
	385	debug_msg( 8, 'Arguments needs to be of type List' )
	386	return 1
	387
[37]	388	for arg in args:
	389
	390	if not arg_string:
	391
	392	arg_string = arg
	393	else:
	394	arg_string = arg_string + ' ' + arg
	395
[54]	396	debug_msg( 8, self.binary + ' ' + action + ' ' + filename + ' ' + arg_string )
[37]	397
[146]	398	cmd = os.popen( self.binary + ' ' + action + ' ' + filename + ' ' + arg_string )
	399	lines = cmd.readlines()
	400	cmd.close()
[37]	401
[146]	402	for line in lines:
	403
[37]	404	if line.find( 'ERROR' ) != -1:
	405
	406	error_msg = string.join( line.split( ' ' )[1:] )
[40]	407	debug_msg( 8, error_msg )
[37]	408	return 1
	409
	410	return 0
	411
[78]	412	class XMLProcessor:
	413	"""Skeleton class for XML processor's"""
	414
	415	def run( self ):
	416	"""Do main processing of XML here"""
	417
	418	pass
	419
	420	class TorqueXMLProcessor( XMLProcessor ):
	421	"""Main class for processing XML and acting with it"""
	422
	423	def __init__( self ):
	424	"""Setup initial XML connection and handlers"""
	425
	426	self.myXMLGatherer = XMLGatherer( ARCHIVE_XMLSOURCE.split( ':' )[0], ARCHIVE_XMLSOURCE.split( ':' )[1] )
	427	self.myXMLSource = self.myXMLGatherer.getFileObject()
	428	self.myXMLHandler = TorqueXMLHandler()
	429	self.myXMLError = XMLErrorHandler()
[87]	430	self.config = GangliaConfigParser( GMETAD_CONF )
[78]	431
	432	def run( self ):
	433	"""Main XML processing"""
	434
[169]	435	debug_msg( 1, 'torque_xml_thread(): started.' )
[87]	436
[78]	437	while( 1 ):
	438
	439	self.myXMLSource = self.myXMLGatherer.getFileObject()
[169]	440	debug_msg( 1, 'torque_xml_thread(): Parsing..' )
[176]	441
	442	try:
	443	xml.sax.parse( self.myXMLSource, self.myXMLHandler, self.myXMLError )
	444	except socket.error, msg:
	445	debug_msg( 0, 'ERROR: Socket error in connect to datasource!: %s' %msg )
	446
[169]	447	debug_msg( 1, 'torque_xml_thread(): Done parsing.' )
	448	debug_msg( 1, 'torque_xml_thread(): Sleeping.. (%ss)' %(str( self.config.getLowestInterval() ) ) )
[87]	449	time.sleep( self.config.getLowestInterval() )
[78]	450
[71]	451	class TorqueXMLHandler( xml.sax.handler.ContentHandler ):
[63]	452	"""Parse Torque's jobinfo XML from our plugin"""
	453
[72]	454	jobAttrs = { }
	455
[84]	456	def __init__( self ):
	457
[214]	458	self.ds = DataSQLStore( JOB_SQL_DBASE.split( '/' )[0], JOB_SQL_DBASE.split( '/' )[1] )
[98]	459	self.jobs_processed = [ ]
	460	self.jobs_to_store = [ ]
[84]	461
[183]	462	def startDocument( self ):
	463
	464	self.heartbeat = 0
	465
[63]	466	def startElement( self, name, attrs ):
	467	"""
	468	This XML will be all gmetric XML
	469	so there will be no specific start/end element
	470	just one XML statement with all info
	471	"""
	472
[73]	473	jobinfo = { }
[63]	474
[199]	475	if name == 'CLUSTER':
[63]	476
[199]	477	self.clustername = attrs.get( 'NAME', "" )
	478
	479	elif name == 'METRIC' and self.clustername in ARCHIVE_DATASOURCES:
	480
[70]	481	metricname = attrs.get( 'NAME', "" )
[63]	482
	483	if metricname == 'TOGA-HEARTBEAT':
[70]	484	self.heartbeat = attrs.get( 'VAL', "" )
[63]	485
[70]	486	elif metricname.find( 'TOGA-JOB' ) != -1:
[63]	487
[72]	488	job_id = metricname.split( 'TOGA-JOB-' )[1]
[63]	489	val = attrs.get( 'VAL', "" )
	490
[96]	491	if not job_id in self.jobs_processed:
	492	self.jobs_processed.append( job_id )
	493
[73]	494	check_change = 0
	495
	496	if self.jobAttrs.has_key( job_id ):
	497	check_change = 1
	498
[63]	499	valinfo = val.split( ' ' )
	500
	501	for myval in valinfo:
	502
[84]	503	if len( myval.split( '=' ) ) > 1:
[63]	504
[84]	505	valname = myval.split( '=' )[0]
	506	value = myval.split( '=' )[1]
[70]	507
[84]	508	if valname == 'nodes':
	509	value = value.split( ';' )
[72]	510
[84]	511	jobinfo[ valname ] = value
	512
[73]	513	if check_change:
[182]	514	if self.jobinfoChanged( self.jobAttrs, job_id, jobinfo ) and self.jobAttrs[ job_id ]['status'] in [ 'R', 'Q' ]:
[100]	515	self.jobAttrs[ job_id ]['stop_timestamp'] = ''
[82]	516	self.jobAttrs[ job_id ] = self.setJobAttrs( self.jobAttrs[ job_id ], jobinfo )
[84]	517	if not job_id in self.jobs_to_store:
	518	self.jobs_to_store.append( job_id )
	519
[102]	520	debug_msg( 6, 'jobinfo for job %s has changed' %job_id )
[73]	521	else:
	522	self.jobAttrs[ job_id ] = jobinfo
[84]	523
	524	if not job_id in self.jobs_to_store:
	525	self.jobs_to_store.append( job_id )
	526
[102]	527	debug_msg( 6, 'jobinfo for job %s has changed' %job_id )
[73]	528
[77]	529	def endDocument( self ):
[74]	530	"""When all metrics have gone, check if any jobs have finished"""
[72]	531
[182]	532	if self.heartbeat:
	533	for jobid, jobinfo in self.jobAttrs.items():
[74]	534
[182]	535	# This is an old job, not in current jobinfo list anymore
	536	# it must have finished, since we _did_ get a new heartbeat
	537	#
	538	mytime = int( jobinfo['reported'] ) + int( jobinfo['poll_interval'] )
[102]	539
[184]	540	if (mytime < self.heartbeat) and (jobid not in self.jobs_processed) and (jobinfo['status'] == 'R'):
[74]	541
[182]	542	if not jobid in self.jobs_processed:
	543	self.jobs_processed.append( jobid )
[96]	544
[182]	545	self.jobAttrs[ jobid ]['status'] = 'F'
	546	self.jobAttrs[ jobid ]['stop_timestamp'] = str( mytime )
[96]	547
[182]	548	if not jobid in self.jobs_to_store:
	549	self.jobs_to_store.append( jobid )
[74]	550
[182]	551	debug_msg( 1, 'torque_xml_thread(): Storing..' )
[87]	552
[182]	553	for jobid in self.jobs_to_store:
	554	if self.jobAttrs[ jobid ]['status'] in [ 'R', 'Q', 'F' ]:
[84]	555
[184]	556	self.ds.storeJobInfo( jobid, self.jobAttrs[ jobid ] )
	557
	558	if self.jobAttrs[ jobid ]['status'] == 'F':
	559	del self.jobAttrs[ jobid ]
	560
[182]	561	debug_msg( 1, 'torque_xml_thread(): Done storing.' )
[87]	562
[182]	563	self.jobs_processed = [ ]
	564	self.jobs_to_store = [ ]
[84]	565
[82]	566	def setJobAttrs( self, old, new ):
	567	"""
	568	Set new job attributes in old, but not lose existing fields
	569	if old attributes doesn't have those
	570	"""
	571
	572	for valname, value in new.items():
	573	old[ valname ] = value
	574
	575	return old
	576
	577
[73]	578	def jobinfoChanged( self, jobattrs, jobid, jobinfo ):
[74]	579	"""
	580	Check if jobinfo has changed from jobattrs[jobid]
	581	if it's report time is bigger than previous one
	582	and it is report time is recent (equal to heartbeat)
	583	"""
[72]	584
[87]	585	ignore_changes = [ 'reported' ]
	586
[73]	587	if jobattrs.has_key( jobid ):
	588
	589	for valname, value in jobinfo.items():
	590
[87]	591	if valname not in ignore_changes:
[73]	592
[87]	593	if jobattrs[ jobid ].has_key( valname ):
[73]	594
[87]	595	if value != jobattrs[ jobid ][ valname ]:
[73]	596
[87]	597	if jobinfo['reported'] > jobattrs[ jobid ][ 'reported' ] and jobinfo['reported'] == self.heartbeat:
	598	return 1
[73]	599
[87]	600	else:
	601	return 1
	602
[73]	603	return 0
	604
[71]	605	class GangliaXMLHandler( xml.sax.handler.ContentHandler ):
[63]	606	"""Parse Ganglia's XML"""
[3]	607
[33]	608	def __init__( self, config ):
[63]	609	"""Setup initial variables and gather info on existing rrd archive"""
	610
[33]	611	self.config = config
[35]	612	self.clusters = { }
[169]	613	debug_msg( 1, 'Checking existing toga rrd archive..' )
[44]	614	self.gatherClusters()
[169]	615	debug_msg( 1, 'Check done.' )
[33]	616
[44]	617	def gatherClusters( self ):
[63]	618	"""Find all existing clusters in archive dir"""
[44]	619
[45]	620	archive_dir = check_dir(ARCHIVE_PATH)
[44]	621
	622	hosts = [ ]
	623
	624	if os.path.exists( archive_dir ):
	625
	626	dirlist = os.listdir( archive_dir )
	627
	628	for item in dirlist:
	629
	630	clustername = item
	631
[60]	632	if not self.clusters.has_key( clustername ) and clustername in ARCHIVE_DATASOURCES:
[44]	633
	634	self.clusters[ clustername ] = RRDHandler( self.config, clustername )
	635
[6]	636	def startElement( self, name, attrs ):
[63]	637	"""Memorize appropriate data from xml start tags"""
[3]	638
[7]	639	if name == 'GANGLIA_XML':
[32]	640
	641	self.XMLSource = attrs.get( 'SOURCE', "" )
	642	self.gangliaVersion = attrs.get( 'VERSION', "" )
	643
[12]	644	debug_msg( 10, 'Found XML data: source %s version %s' %( self.XMLSource, self.gangliaVersion ) )
[6]	645
[7]	646	elif name == 'GRID':
[32]	647
	648	self.gridName = attrs.get( 'NAME', "" )
	649	self.time = attrs.get( 'LOCALTIME', "" )
	650
[12]	651	debug_msg( 10, '`-Grid found: %s' %( self.gridName ) )
[6]	652
[7]	653	elif name == 'CLUSTER':
[32]	654
	655	self.clusterName = attrs.get( 'NAME', "" )
	656	self.time = attrs.get( 'LOCALTIME', "" )
	657
[60]	658	if not self.clusters.has_key( self.clusterName ) and self.clusterName in ARCHIVE_DATASOURCES:
[32]	659
[34]	660	self.clusters[ self.clusterName ] = RRDHandler( self.config, self.clusterName )
[33]	661
[35]	662	debug_msg( 10, ' \|-Cluster found: %s' %( self.clusterName ) )
[6]	663
[60]	664	elif name == 'HOST' and self.clusterName in ARCHIVE_DATASOURCES:
[32]	665
	666	self.hostName = attrs.get( 'NAME', "" )
	667	self.hostIp = attrs.get( 'IP', "" )
	668	self.hostReported = attrs.get( 'REPORTED', "" )
	669
[12]	670	debug_msg( 10, ' \| \|-Host found: %s - ip %s reported %s' %( self.hostName, self.hostIp, self.hostReported ) )
[6]	671
[60]	672	elif name == 'METRIC' and self.clusterName in ARCHIVE_DATASOURCES:
[6]	673
[32]	674	type = attrs.get( 'TYPE', "" )
[198]	675
	676	exclude_metric = False
	677
	678	for ex_metricstr in ARCHIVE_EXCLUDE_METRICS:
[6]	679
[198]	680	orig_name = attrs.get( 'NAME', "" )
[3]	681
[198]	682	if string.lower( orig_name ) == string.lower( ex_metricstr ):
	683
	684	exclude_metric = True
	685
	686	elif re.match( ex_metricstr, orig_name ):
	687
	688	exclude_metric = True
	689
	690	if type not in UNSUPPORTED_ARCHIVE_TYPES and not exclude_metric:
	691
[32]	692	myMetric = { }
	693	myMetric['name'] = attrs.get( 'NAME', "" )
	694	myMetric['val'] = attrs.get( 'VAL', "" )
	695	myMetric['time'] = self.hostReported
[3]	696
[34]	697	self.clusters[ self.clusterName ].memMetric( self.hostName, myMetric )
[3]	698
[34]	699	debug_msg( 11, ' \| \| \|-metric: %s:%s' %( myMetric['name'], myMetric['val'] ) )
[6]	700
[34]	701	def storeMetrics( self ):
[63]	702	"""Store metrics of each cluster rrd handler"""
[9]	703
[34]	704	for clustername, rrdh in self.clusters.items():
[16]	705
[38]	706	ret = rrdh.storeMetrics()
[9]	707
[38]	708	if ret:
	709	debug_msg( 9, 'An error occured while storing metrics for cluster %s' %clustername )
	710	return 1
	711
	712	return 0
	713
[71]	714	class XMLErrorHandler( xml.sax.handler.ErrorHandler ):
	715
	716	def error( self, exception ):
	717	"""Recoverable error"""
	718
[169]	719	debug_msg( 0, 'Recoverable XML error ' + str( exception ) + ' ignored.' )
[71]	720
	721	def fatalError( self, exception ):
	722	"""Non-recoverable error"""
	723
	724	exception_str = str( exception )
	725
	726	# Ignore 'no element found' errors
	727	if exception_str.find( 'no element found' ) != -1:
[169]	728	debug_msg( 0, 'No XML data found: Socket not (re)connected or datasource not available.' )
[71]	729	return 0
	730
[170]	731	debug_msg( 0, 'FATAL ERROR: Non-recoverable XML error ' + str( exception ) )
[71]	732	sys.exit( 1 )
	733
	734	def warning( self, exception ):
	735	"""Warning"""
	736
	737	debug_msg( 0, 'Warning ' + str( exception ) )
	738
[78]	739	class XMLGatherer:
[63]	740	"""Setup a connection and file object to Ganglia's XML"""
[3]	741
[8]	742	s = None
[70]	743	fd = None
[8]	744
	745	def __init__( self, host, port ):
[63]	746	"""Store host and port for connection"""
[8]	747
[5]	748	self.host = host
	749	self.port = port
[33]	750	self.connect()
[70]	751	self.makeFileDescriptor()
[3]	752
[33]	753	def connect( self ):
[63]	754	"""Setup connection to XML source"""
[8]	755
	756	for res in socket.getaddrinfo( self.host, self.port, socket.AF_UNSPEC, socket.SOCK_STREAM ):
[32]	757
[5]	758	af, socktype, proto, canonname, sa = res
[32]	759
[5]	760	try:
[32]	761
[8]	762	self.s = socket.socket( af, socktype, proto )
[32]	763
[5]	764	except socket.error, msg:
[32]	765
[8]	766	self.s = None
[5]	767	continue
[32]	768
[5]	769	try:
[32]	770
[8]	771	self.s.connect( sa )
[32]	772
[5]	773	except socket.error, msg:
[32]	774
[70]	775	self.disconnect()
[5]	776	continue
[32]	777
[5]	778	break
[3]	779
[8]	780	if self.s is None:
[32]	781
[170]	782	debug_msg( 0, 'FATAL ERROR: Could not open socket or unable to connect to datasource!' )
[33]	783	sys.exit( 1 )
[5]	784
[33]	785	def disconnect( self ):
[63]	786	"""Close socket"""
[33]	787
	788	if self.s:
[71]	789	self.s.shutdown( 2 )
[33]	790	self.s.close()
	791	self.s = None
	792
	793	def __del__( self ):
[63]	794	"""Kill the socket before we leave"""
[33]	795
	796	self.disconnect()
	797
[70]	798	def reconnect( self ):
	799	"""Reconnect"""
[33]	800
[38]	801	if self.s:
	802	self.disconnect()
[33]	803
[70]	804	self.connect()
[5]	805
[70]	806	def makeFileDescriptor( self ):
	807	"""Make file descriptor that points to our socket connection"""
	808
	809	self.reconnect()
	810
	811	if self.s:
	812	self.fd = self.s.makefile( 'r' )
	813
	814	def getFileObject( self ):
	815	"""Connect, and return a file object"""
	816
[78]	817	self.makeFileDescriptor()
	818
[70]	819	if self.fd:
	820	return self.fd
	821
[78]	822	class GangliaXMLProcessor( XMLProcessor ):
[63]	823	"""Main class for processing XML and acting with it"""
[5]	824
[33]	825	def __init__( self ):
[63]	826	"""Setup initial XML connection and handlers"""
[33]	827
	828	self.config = GangliaConfigParser( GMETAD_CONF )
	829
[78]	830	self.myXMLGatherer = XMLGatherer( ARCHIVE_XMLSOURCE.split( ':' )[0], ARCHIVE_XMLSOURCE.split( ':' )[1] )
[70]	831	self.myXMLSource = self.myXMLGatherer.getFileObject()
[78]	832	self.myXMLHandler = GangliaXMLHandler( self.config )
	833	self.myXMLError = XMLErrorHandler()
[73]	834
[9]	835	def run( self ):
[63]	836	"""Main XML processing; start a xml and storethread"""
[8]	837
[102]	838	xml_thread = threading.Thread( None, self.processXML, 'xmlthread' )
	839	store_thread = threading.Thread( None, self.storeMetrics, 'storethread' )
[22]	840
[36]	841	while( 1 ):
	842
[102]	843	if not xml_thread.isAlive():
[36]	844	# Gather XML at the same interval as gmetad
	845
	846	# threaded call to: self.processXML()
	847	#
[169]	848	try:
	849	xml_thread = threading.Thread( None, self.processXML, 'xml_thread' )
	850	xml_thread.start()
[176]	851	except thread.error, msg:
[169]	852	debug_msg( 0, 'ERROR: Unable to start xml_thread!: '+str(msg))
	853	#return 1
[36]	854
[102]	855	if not store_thread.isAlive():
[55]	856	# Store metrics every .. sec
[36]	857
[55]	858	# threaded call to: self.storeMetrics()
	859	#
[169]	860	try:
	861	store_thread = threading.Thread( None, self.storeMetrics, 'store_thread' )
	862	store_thread.start()
[176]	863	except thread.error, msg:
[169]	864	debug_msg( 0, 'ERROR: Unable to start store_thread!: '+str(msg))
	865	#return 1
[36]	866
	867	# Just sleep a sec here, to prevent daemon from going mad. We're all threads here anyway
	868	time.sleep( 1 )
	869
[33]	870	def storeMetrics( self ):
[63]	871	"""Store metrics retained in memory to disk"""
[22]	872
[63]	873	# Store metrics somewhere between every 360 and 640 seconds
[38]	874	#
[55]	875	STORE_INTERVAL = random.randint( 360, 640 )
[22]	876
[169]	877	try:
	878	store_metric_thread = threading.Thread( None, self.storeThread, 'store_metric_thread' )
	879	store_metric_thread.start()
[176]	880	except thread.error, msg:
[169]	881	debug_msg( 0, 'ERROR: Unable to start ganglia_store_thread()!: '+str(msg) )
	882	return 1
[36]	883
[169]	884	debug_msg( 1, 'ganglia_store_thread(): started.' )
	885
	886	debug_msg( 1, 'ganglia_store_thread(): Sleeping.. (%ss)' %STORE_INTERVAL )
[36]	887	time.sleep( STORE_INTERVAL )
[169]	888	debug_msg( 1, 'ganglia_store_thread(): Done sleeping.' )
[36]	889
[102]	890	if store_metric_thread.isAlive():
[36]	891
[169]	892	debug_msg( 1, 'ganglia_store_thread(): storemetricthread() still running, waiting to finish..' )
[136]	893	store_metric_thread.join( STORE_TIMEOUT ) # Maximum time is for storing thread to finish
[169]	894	debug_msg( 1, 'ganglia_store_thread(): Done waiting.' )
[36]	895
[169]	896	debug_msg( 1, 'ganglia_store_thread(): finished.' )
[36]	897
	898	return 0
	899
[39]	900	def storeThread( self ):
[63]	901	"""Actual metric storing thread"""
[39]	902
[169]	903	debug_msg( 1, 'ganglia_store_metric_thread(): started.' )
	904	debug_msg( 1, 'ganglia_store_metric_thread(): Storing data..' )
[78]	905	ret = self.myXMLHandler.storeMetrics()
[176]	906	if ret > 0:
	907	debug_msg( 0, 'ganglia_store_metric_thread(): UNKNOWN ERROR %s while storing Metrics!' %str(ret) )
[169]	908	debug_msg( 1, 'ganglia_store_metric_thread(): Done storing.' )
	909	debug_msg( 1, 'ganglia_store_metric_thread(): finished.' )
[39]	910
[176]	911	return 0
[39]	912
[8]	913	def processXML( self ):
[63]	914	"""Process XML"""
[8]	915
[169]	916	try:
	917	parsethread = threading.Thread( None, self.parseThread, 'parsethread' )
	918	parsethread.start()
[176]	919	except thread.error, msg:
[169]	920	debug_msg( 0, 'ERROR: Unable to start ganglia_xml_thread()!: ' + str(msg) )
	921	return 1
[8]	922
[169]	923	debug_msg( 1, 'ganglia_xml_thread(): started.' )
[36]	924
[169]	925	debug_msg( 1, 'ganglia_xml_thread(): Sleeping.. (%ss)' %self.config.getLowestInterval() )
[36]	926	time.sleep( float( self.config.getLowestInterval() ) )
[169]	927	debug_msg( 1, 'ganglia_xml_thread(): Done sleeping.' )
[36]	928
	929	if parsethread.isAlive():
	930
[169]	931	debug_msg( 1, 'ganglia_xml_thread(): parsethread() still running, waiting (%ss) to finish..' %PARSE_TIMEOUT )
[47]	932	parsethread.join( PARSE_TIMEOUT ) # Maximum time for XML thread to finish
[169]	933	debug_msg( 1, 'ganglia_xml_thread(): Done waiting.' )
[36]	934
[169]	935	debug_msg( 1, 'ganglia_xml_thread(): finished.' )
[36]	936
	937	return 0
	938
[39]	939	def parseThread( self ):
[63]	940	"""Actual parsing thread"""
[39]	941
[169]	942	debug_msg( 1, 'ganglia_parse_thread(): started.' )
	943	debug_msg( 1, 'ganglia_parse_thread(): Parsing XML..' )
[78]	944	self.myXMLSource = self.myXMLGatherer.getFileObject()
[176]	945
	946	try:
	947	xml.sax.parse( self.myXMLSource, self.myXMLHandler, self.myXMLError )
	948	except socket.error, msg:
	949	debug_msg( 0, 'ERROR: Socket error in connect to datasource!: %s' %msg )
	950
[169]	951	debug_msg( 1, 'ganglia_parse_thread(): Done parsing.' )
	952	debug_msg( 1, 'ganglia_parse_thread(): finished.' )
[39]	953
[176]	954	return 0
[39]	955
[9]	956	class GangliaConfigParser:
	957
[34]	958	sources = [ ]
[9]	959
	960	def __init__( self, config ):
[63]	961	"""Parse some stuff from our gmetad's config, such as polling interval"""
[32]	962
[9]	963	self.config = config
	964	self.parseValues()
	965
[32]	966	def parseValues( self ):
[63]	967	"""Parse certain values from gmetad.conf"""
[9]	968
	969	readcfg = open( self.config, 'r' )
	970
	971	for line in readcfg.readlines():
	972
	973	if line.count( '"' ) > 1:
	974
[10]	975	if line.find( 'data_source' ) != -1 and line[0] != '#':
[9]	976
[11]	977	source = { }
	978	source['name'] = line.split( '"' )[1]
[9]	979	source_words = line.split( '"' )[2].split( ' ' )
	980
	981	for word in source_words:
	982
	983	valid_interval = 1
	984
	985	for letter in word:
[32]	986
[9]	987	if letter not in string.digits:
[32]	988
[9]	989	valid_interval = 0
	990
[10]	991	if valid_interval and len(word) > 0:
[32]	992
[9]	993	source['interval'] = word
[12]	994	debug_msg( 9, 'polling interval for %s = %s' %(source['name'], source['interval'] ) )
[33]	995
	996	# No interval found, use Ganglia's default
	997	if not source.has_key( 'interval' ):
	998	source['interval'] = 15
	999	debug_msg( 9, 'polling interval for %s defaulted to 15' %(source['name']) )
[32]	1000
[33]	1001	self.sources.append( source )
[9]	1002
	1003	def getInterval( self, source_name ):
[63]	1004	"""Return interval for source_name"""
[32]	1005
[9]	1006	for source in self.sources:
[32]	1007
[12]	1008	if source['name'] == source_name:
[32]	1009
[9]	1010	return source['interval']
[32]	1011
[9]	1012	return None
	1013
[34]	1014	def getLowestInterval( self ):
[63]	1015	"""Return the lowest interval of all clusters"""
[34]	1016
	1017	lowest_interval = 0
	1018
	1019	for source in self.sources:
	1020
	1021	if not lowest_interval or source['interval'] <= lowest_interval:
	1022
	1023	lowest_interval = source['interval']
	1024
	1025	# Return 15 when nothing is found, so that the daemon won't go insane with 0 sec delays
	1026	if lowest_interval:
	1027	return lowest_interval
	1028	else:
	1029	return 15
	1030
[9]	1031	class RRDHandler:
[63]	1032	"""Class for handling RRD activity"""
[9]	1033
[32]	1034	myMetrics = { }
[40]	1035	lastStored = { }
[47]	1036	timeserials = { }
[36]	1037	slot = None
[32]	1038
[33]	1039	def __init__( self, config, cluster ):
[63]	1040	"""Setup initial variables"""
[78]	1041
[44]	1042	self.block = 0
[9]	1043	self.cluster = cluster
[33]	1044	self.config = config
[40]	1045	self.slot = threading.Lock()
[37]	1046	self.rrdm = RRDMutator()
[42]	1047	self.gatherLastUpdates()
[9]	1048
[42]	1049	def gatherLastUpdates( self ):
[63]	1050	"""Populate the lastStored list, containing timestamps of all last updates"""
[42]	1051
	1052	cluster_dir = '%s/%s' %( check_dir(ARCHIVE_PATH), self.cluster )
	1053
	1054	hosts = [ ]
	1055
	1056	if os.path.exists( cluster_dir ):
	1057
[44]	1058	dirlist = os.listdir( cluster_dir )
[42]	1059
[44]	1060	for dir in dirlist:
[42]	1061
[44]	1062	hosts.append( dir )
[42]	1063
	1064	for host in hosts:
	1065
[47]	1066	host_dir = cluster_dir + '/' + host
	1067	dirlist = os.listdir( host_dir )
	1068
	1069	for dir in dirlist:
	1070
	1071	if not self.timeserials.has_key( host ):
	1072
	1073	self.timeserials[ host ] = [ ]
	1074
	1075	self.timeserials[ host ].append( dir )
	1076
[42]	1077	last_serial = self.getLastRrdTimeSerial( host )
	1078	if last_serial:
	1079
	1080	metric_dir = cluster_dir + '/' + host + '/' + last_serial
	1081	if os.path.exists( metric_dir ):
	1082
[44]	1083	dirlist = os.listdir( metric_dir )
[42]	1084
[44]	1085	for file in dirlist:
[42]	1086
[44]	1087	metricname = file.split( '.rrd' )[0]
[42]	1088
[44]	1089	if not self.lastStored.has_key( host ):
[42]	1090
[44]	1091	self.lastStored[ host ] = { }
[42]	1092
[44]	1093	self.lastStored[ host ][ metricname ] = self.rrdm.grabLastUpdate( metric_dir + '/' + file )
[42]	1094
[32]	1095	def getClusterName( self ):
[63]	1096	"""Return clustername"""
	1097
[32]	1098	return self.cluster
	1099
	1100	def memMetric( self, host, metric ):
[63]	1101	"""Store metric from host in memory"""
[32]	1102
[179]	1103	# <ATOMIC>
	1104	#
	1105	self.slot.acquire()
	1106
[34]	1107	if self.myMetrics.has_key( host ):
[32]	1108
[34]	1109	if self.myMetrics[ host ].has_key( metric['name'] ):
[32]	1110
[34]	1111	for mymetric in self.myMetrics[ host ][ metric['name'] ]:
[32]	1112
[34]	1113	if mymetric['time'] == metric['time']:
[32]	1114
[34]	1115	# Allready have this metric, abort
[179]	1116	self.slot.release()
[34]	1117	return 1
	1118	else:
	1119	self.myMetrics[ host ][ metric['name'] ] = [ ]
	1120	else:
[32]	1121	self.myMetrics[ host ] = { }
[34]	1122	self.myMetrics[ host ][ metric['name'] ] = [ ]
[32]	1123
[63]	1124	# Push new metric onto stack
	1125	# atomic code; only 1 thread at a time may access the stack
	1126
[32]	1127	self.myMetrics[ host ][ metric['name'] ].append( metric )
	1128
[40]	1129	self.slot.release()
[53]	1130	#
	1131	# </ATOMIC>
[40]	1132
[47]	1133	def makeUpdateList( self, host, metriclist ):
[63]	1134	"""
	1135	Make a list of update values for rrdupdate
	1136	but only those that we didn't store before
	1137	"""
[37]	1138
	1139	update_list = [ ]
[41]	1140	metric = None
[37]	1141
[47]	1142	while len( metriclist ) > 0:
[37]	1143
[53]	1144	metric = metriclist.pop( 0 )
[37]	1145
[53]	1146	if self.checkStoreMetric( host, metric ):
	1147	update_list.append( '%s:%s' %( metric['time'], metric['val'] ) )
[40]	1148
[37]	1149	return update_list
	1150
[49]	1151	def checkStoreMetric( self, host, metric ):
[63]	1152	"""Check if supplied metric if newer than last one stored"""
[40]	1153
	1154	if self.lastStored.has_key( host ):
	1155
[47]	1156	if self.lastStored[ host ].has_key( metric['name'] ):
[40]	1157
[47]	1158	if metric['time'] <= self.lastStored[ host ][ metric['name'] ]:
[40]	1159
[50]	1160	# This is old
[40]	1161	return 0
	1162
[50]	1163	return 1
	1164
[54]	1165	def memLastUpdate( self, host, metricname, metriclist ):
[63]	1166	"""
	1167	Memorize the time of the latest metric from metriclist
	1168	but only if it wasn't allready memorized
	1169	"""
[50]	1170
[54]	1171	if not self.lastStored.has_key( host ):
	1172	self.lastStored[ host ] = { }
	1173
[50]	1174	last_update_time = 0
	1175
	1176	for metric in metriclist:
	1177
[54]	1178	if metric['name'] == metricname:
[50]	1179
[54]	1180	if metric['time'] > last_update_time:
[50]	1181
[54]	1182	last_update_time = metric['time']
[40]	1183
[54]	1184	if self.lastStored[ host ].has_key( metricname ):
[52]	1185
[54]	1186	if last_update_time <= self.lastStored[ host ][ metricname ]:
[52]	1187	return 1
[40]	1188
[54]	1189	self.lastStored[ host ][ metricname ] = last_update_time
[52]	1190
[33]	1191	def storeMetrics( self ):
[63]	1192	"""
	1193	Store all metrics from memory to disk
	1194	and do it to the RRD's in appropriate timeperiod directory
	1195	"""
[33]	1196
	1197	for hostname, mymetrics in self.myMetrics.items():
	1198
	1199	for metricname, mymetric in mymetrics.items():
	1200
[53]	1201	metrics_to_store = [ ]
	1202
[63]	1203	# Pop metrics from stack for storing until none is left
	1204	# atomic code: only 1 thread at a time may access myMetrics
	1205
[53]	1206	# <ATOMIC>
[50]	1207	#
[47]	1208	self.slot.acquire()
[33]	1209
[54]	1210	while len( self.myMetrics[ hostname ][ metricname ] ) > 0:
[53]	1211
[54]	1212	if len( self.myMetrics[ hostname ][ metricname ] ) > 0:
[53]	1213
[176]	1214	try:
	1215	metrics_to_store.append( self.myMetrics[ hostname ][ metricname ].pop( 0 ) )
	1216	except IndexError, msg:
	1217
[179]	1218	# Somehow sometimes myMetrics[ hostname ][ metricname ]
	1219	# is still len 0 when the statement is executed.
	1220	# Just ignore indexerror's..
[176]	1221	pass
	1222
[53]	1223	self.slot.release()
	1224	#
	1225	# </ATOMIC>
	1226
[47]	1227	# Create a mapping table, each metric to the period where it should be stored
	1228	#
[53]	1229	metric_serial_table = self.determineSerials( hostname, metricname, metrics_to_store )
[33]	1230
[50]	1231	update_rets = [ ]
	1232
[47]	1233	for period, pmetric in metric_serial_table.items():
	1234
[146]	1235	create_ret = self.createCheck( hostname, metricname, period )
[47]	1236
	1237	update_ret = self.update( hostname, metricname, period, pmetric )
	1238
	1239	if update_ret == 0:
	1240
	1241	debug_msg( 9, 'stored metric %s for %s' %( hostname, metricname ) )
	1242	else:
	1243	debug_msg( 9, 'metric update failed' )
	1244
[146]	1245	update_rets.append( create_ret )
[50]	1246	update_rets.append( update_ret )
[47]	1247
[179]	1248	# Lets ignore errors here for now, we need to make sure last update time
	1249	# is correct!
	1250	#
	1251	#if not (1) in update_rets:
[50]	1252
[179]	1253	self.memLastUpdate( hostname, metricname, metrics_to_store )
[50]	1254
[17]	1255	def makeTimeSerial( self ):
[63]	1256	"""Generate a time serial. Seconds since epoch"""
[17]	1257
	1258	# Seconds since epoch
	1259	mytime = int( time.time() )
	1260
	1261	return mytime
	1262
[50]	1263	def makeRrdPath( self, host, metricname, timeserial ):
[63]	1264	"""Make a RRD location/path and filename"""
[17]	1265
[50]	1266	rrd_dir = '%s/%s/%s/%s' %( check_dir(ARCHIVE_PATH), self.cluster, host, timeserial )
	1267	rrd_file = '%s/%s.rrd' %( rrd_dir, metricname )
[17]	1268
	1269	return rrd_dir, rrd_file
	1270
[20]	1271	def getLastRrdTimeSerial( self, host ):
[63]	1272	"""Find the last timeserial (directory) for this host"""
[17]	1273
[19]	1274	newest_timeserial = 0
	1275
[47]	1276	for dir in self.timeserials[ host ]:
[32]	1277
[47]	1278	valid_dir = 1
[17]	1279
[47]	1280	for letter in dir:
	1281	if letter not in string.digits:
	1282	valid_dir = 0
[17]	1283
[47]	1284	if valid_dir:
	1285	timeserial = dir
	1286	if timeserial > newest_timeserial:
	1287	newest_timeserial = timeserial
[17]	1288
	1289	if newest_timeserial:
[18]	1290	return newest_timeserial
[17]	1291	else:
	1292	return 0
	1293
[47]	1294	def determinePeriod( self, host, check_serial ):
[63]	1295	"""Determine to which period (directory) this time(serial) belongs"""
[47]	1296
	1297	period_serial = 0
	1298
[56]	1299	if self.timeserials.has_key( host ):
[47]	1300
[56]	1301	for serial in self.timeserials[ host ]:
[47]	1302
[56]	1303	if check_serial >= serial and period_serial < serial:
[47]	1304
[56]	1305	period_serial = serial
	1306
[47]	1307	return period_serial
	1308
	1309	def determineSerials( self, host, metricname, metriclist ):
	1310	"""
	1311	Determine the correct serial and corresponding rrd to store
	1312	for a list of metrics
	1313	"""
	1314
	1315	metric_serial_table = { }
	1316
	1317	for metric in metriclist:
	1318
	1319	if metric['name'] == metricname:
	1320
	1321	period = self.determinePeriod( host, metric['time'] )
	1322
	1323	archive_secs = ARCHIVE_HOURS_PER_RRD * (60 * 60)
	1324
[49]	1325	if (int( metric['time'] ) - int( period ) ) > archive_secs:
[47]	1326
	1327	# This one should get it's own new period
	1328	period = metric['time']
[57]	1329
	1330	if not self.timeserials.has_key( host ):
	1331	self.timeserials[ host ] = [ ]
	1332
[50]	1333	self.timeserials[ host ].append( period )
[47]	1334
	1335	if not metric_serial_table.has_key( period ):
	1336
[49]	1337	metric_serial_table[ period ] = [ ]
[47]	1338
	1339	metric_serial_table[ period ].append( metric )
	1340
	1341	return metric_serial_table
	1342
[33]	1343	def createCheck( self, host, metricname, timeserial ):
[63]	1344	"""Check if an rrd allready exists for this metric, create if not"""
[9]	1345
[35]	1346	debug_msg( 9, 'rrdcreate: using timeserial %s for %s/%s' %( timeserial, host, metricname ) )
[47]	1347
[33]	1348	rrd_dir, rrd_file = self.makeRrdPath( host, metricname, timeserial )
[17]	1349
[9]	1350	if not os.path.exists( rrd_dir ):
[58]	1351
	1352	try:
	1353	os.makedirs( rrd_dir )
	1354
[169]	1355	except os.OSError, msg:
[58]	1356
	1357	if msg.find( 'File exists' ) != -1:
	1358
	1359	# Ignore exists errors
	1360	pass
	1361
	1362	else:
	1363
	1364	print msg
	1365	return
	1366
[14]	1367	debug_msg( 9, 'created dir %s' %( str(rrd_dir) ) )
[9]	1368
[14]	1369	if not os.path.exists( rrd_file ):
[9]	1370
[33]	1371	interval = self.config.getInterval( self.cluster )
[47]	1372	heartbeat = 8 * int( interval )
[9]	1373
[37]	1374	params = [ ]
[12]	1375
[37]	1376	params.append( '--step' )
	1377	params.append( str( interval ) )
[12]	1378
[37]	1379	params.append( '--start' )
[47]	1380	params.append( str( int( timeserial ) - 1 ) )
[12]	1381
[37]	1382	params.append( 'DS:sum:GAUGE:%d:U:U' %heartbeat )
	1383	params.append( 'RRA:AVERAGE:0.5:1:%s' %(ARCHIVE_HOURS_PER_RRD * 240) )
[13]	1384
[37]	1385	self.rrdm.create( str(rrd_file), params )
	1386
[14]	1387	debug_msg( 9, 'created rrd %s' %( str(rrd_file) ) )
	1388
[47]	1389	def update( self, host, metricname, timeserial, metriclist ):
[63]	1390	"""
	1391	Update rrd file for host with metricname
	1392	in directory timeserial with metriclist
	1393	"""
[9]	1394
[35]	1395	debug_msg( 9, 'rrdupdate: using timeserial %s for %s/%s' %( timeserial, host, metricname ) )
[9]	1396
[33]	1397	rrd_dir, rrd_file = self.makeRrdPath( host, metricname, timeserial )
[18]	1398
[47]	1399	update_list = self.makeUpdateList( host, metriclist )
[15]	1400
[41]	1401	if len( update_list ) > 0:
	1402	ret = self.rrdm.update( str(rrd_file), update_list )
[32]	1403
[41]	1404	if ret:
	1405	return 1
[27]	1406
[41]	1407	debug_msg( 9, 'updated rrd %s with %s' %( str(rrd_file), string.join( update_list ) ) )
[15]	1408
[36]	1409	return 0
	1410
[169]	1411	def daemon():
	1412	"""daemonized threading"""
[8]	1413
[169]	1414	# Fork the first child
	1415	#
	1416	pid = os.fork()
	1417
	1418	if pid > 0:
	1419
	1420	sys.exit(0) # end parent
	1421
	1422	# creates a session and sets the process group ID
	1423	#
	1424	os.setsid()
	1425
	1426	# Fork the second child
	1427	#
	1428	pid = os.fork()
	1429
	1430	if pid > 0:
	1431
	1432	sys.exit(0) # end parent
	1433
	1434	# Go to the root directory and set the umask
	1435	#
	1436	os.chdir('/')
	1437	os.umask(0)
	1438
	1439	sys.stdin.close()
	1440	sys.stdout.close()
	1441	sys.stderr.close()
	1442
	1443	os.open('/dev/null', 0)
	1444	os.dup(0)
	1445	os.dup(0)
	1446
	1447	run()
	1448
	1449	def run():
	1450	"""Threading start"""
	1451
[102]	1452	myTorqueProcessor = TorqueXMLProcessor()
	1453	myGangliaProcessor = GangliaXMLProcessor()
[8]	1454
[169]	1455	try:
[102]	1456	torque_xml_thread = threading.Thread( None, myTorqueProcessor.run, 'torque_proc_thread' )
	1457	ganglia_xml_thread = threading.Thread( None, myGangliaProcessor.run, 'ganglia_proc_thread' )
[22]	1458
[169]	1459	torque_xml_thread.start()
	1460	ganglia_xml_thread.start()
	1461
[176]	1462	except thread.error, msg:
[169]	1463	debug_msg( 0, 'FATAL ERROR: Unable to start main threads!: '+ str(msg) )
	1464	syslog.closelog()
	1465	sys.exit(1)
	1466
	1467	debug_msg( 0, 'main threading started.' )
[78]	1468
[169]	1469	def main():
	1470	"""Program startup"""
	1471
[214]	1472	if not processArgs( sys.argv[1:] ):
	1473	sys.exit( 1 )
	1474
[169]	1475	if( DAEMONIZE and USE_SYSLOG ):
	1476	syslog.openlog( 'jobarchived', syslog.LOG_NOWAIT, SYSLOG_FACILITY )
	1477
	1478	if DAEMONIZE:
	1479	daemon()
	1480	else:
	1481	run()
	1482
	1483	#
[81]	1484	# Global functions
[169]	1485	#
[81]	1486
[9]	1487	def check_dir( directory ):
[63]	1488	"""Check if directory is a proper directory. I.e.: Does _not_ end with a '/'"""
[9]	1489
	1490	if directory[-1] == '/':
	1491	directory = directory[:-1]
	1492
	1493	return directory
	1494
[12]	1495	def debug_msg( level, msg ):
[169]	1496	"""Only print msg if correct levels"""
[12]	1497
[169]	1498	if (not DAEMONIZE and DEBUG_LEVEL >= level):
	1499	sys.stderr.write( printTime() + ' - ' + msg + '\n' )
	1500
	1501	if (DAEMONIZE and USE_SYSLOG and SYSLOG_LEVEL >= level):
	1502	syslog.syslog( msg )
[12]	1503
[46]	1504	def printTime( ):
[63]	1505	"""Print current time in human readable format"""
[46]	1506
	1507	return time.strftime("%a %d %b %Y %H:%M:%S")
	1508
[63]	1509	# Ooohh, someone started me! Let's go..
[9]	1510	if __name__ == '__main__':
	1511	main()

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

Context Navigation

source: trunk/jobarchived/jobarchived.py @ 224

Download in other formats: