Context Navigation

source: trunk/daemon/jobarchived.py @ 198

Last change on this file since 198 was 198, checked in by bastiaans, 19 years ago

daemon/jobarchived.py:

Now with support for ARCHIVE_EXCLUDE_METRICS ! List of metric names (string or regexp) not to archive

File size: 34.8 KB

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

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

Download in other formats: