Context Navigation

source: trunk/jobarchived/jobarchived.py @ 293

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

jobarchived/jobarchived.py:

Fixed: XMLGatherer'ing
Added XMLGather locking
Removed conflicting XMLGatherer instances

Property svn:keywords set to Id

File size: 37.9 KB

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

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

Download in other formats: