Context Navigation

source: trunk/jobarchived/jobarchived.py @ 369

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

jobarchived/jobarchived.py:

autocreate rrd archive dirs

Property svn:keywords set to Id

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

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

Download in other formats: