Context Navigation

source: trunk/jobarchived/jobarchived.py @ 296

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

jobarchived/jobarchived.py:

changed msg to be more descriptive

Property svn:keywords set to Id

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

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

Download in other formats: