Context Navigation

source: trunk/daemon/togad.py @ 35

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

daemon/togad.py:

More bugfixes

Need to find a way to share data between Child and Parent or find a different way to handle it

File size: 13.8 KB

Rev	Line
[3]	1	#!/usr/bin/env python
	2
[5]	3	from xml.sax import make_parser
	4	from xml.sax.handler import ContentHandler
	5	import socket
	6	import sys
[9]	7	import rrdtool
	8	import string
[12]	9	import os
	10	import os.path
[17]	11	import time
	12	import re
[3]	13
[8]	14	# Specify debugging level here;
	15	#
[30]	16	# <=11 = metric XML
	17	# <=10 = host,cluster,grid,ganglia XML
	18	# <=9 = RRD activity,gmetad config parsing
	19	# <=8 = host processing
[31]	20	# <=7 = daemon threading - NOTE: Daemon will 'halt on all errors' from this level
[8]	21	#
[34]	22	DEBUG_LEVEL = 10
[6]	23
[9]	24	# Where is the gmetad.conf located
	25	#
	26	GMETAD_CONF = '/etc/gmetad.conf'
	27
	28	# Where to store the archived rrd's
	29	#
	30	ARCHIVE_PATH = '/data/toga/rrds'
	31
	32	# List of data_source names to archive for
	33	#
	34	ARCHIVE_SOURCES = [ "LISA Cluster" ]
	35
[13]	36	# Amount of hours to store in one single archived .rrd
[9]	37	#
[22]	38	ARCHIVE_HOURS_PER_RRD = 12
[13]	39
[22]	40	# Wether or not to run as a daemon in background
	41	#
	42	DAEMONIZE = 0
	43
[17]	44	######################
	45	# #
	46	# Configuration ends #
	47	# #
	48	######################
[13]	49
[17]	50	# What XML data types not to store
[13]	51	#
[17]	52	UNSUPPORTED_ARCHIVE_TYPES = [ 'string' ]
[9]	53
[8]	54	"""
	55	This is TOrque-GAnglia's data Daemon
	56	"""
	57
[6]	58	class GangliaXMLHandler( ContentHandler ):
[8]	59	"Parse Ganglia's XML"
[3]	60
[33]	61	def __init__( self, config ):
	62	self.config = config
[35]	63	self.clusters = { }
[33]	64
[6]	65	def startElement( self, name, attrs ):
[8]	66	"Store appropriate data from xml start tags"
[3]	67
[7]	68	if name == 'GANGLIA_XML':
[32]	69
	70	self.XMLSource = attrs.get( 'SOURCE', "" )
	71	self.gangliaVersion = attrs.get( 'VERSION', "" )
	72
[12]	73	debug_msg( 10, 'Found XML data: source %s version %s' %( self.XMLSource, self.gangliaVersion ) )
[6]	74
[7]	75	elif name == 'GRID':
[32]	76
	77	self.gridName = attrs.get( 'NAME', "" )
	78	self.time = attrs.get( 'LOCALTIME', "" )
	79
[12]	80	debug_msg( 10, '`-Grid found: %s' %( self.gridName ) )
[6]	81
[7]	82	elif name == 'CLUSTER':
[32]	83
	84	self.clusterName = attrs.get( 'NAME', "" )
	85	self.time = attrs.get( 'LOCALTIME', "" )
	86
[35]	87	if not self.clusters.has_key( self.clusterName ) and self.clusterName in ARCHIVE_SOURCES:
[32]	88
[34]	89	self.clusters[ self.clusterName ] = RRDHandler( self.config, self.clusterName )
[33]	90
[35]	91	debug_msg( 10, ' \|-Cluster found: %s' %( self.clusterName ) )
[6]	92
[14]	93	elif name == 'HOST' and self.clusterName in ARCHIVE_SOURCES:
[32]	94
	95	self.hostName = attrs.get( 'NAME', "" )
	96	self.hostIp = attrs.get( 'IP', "" )
	97	self.hostReported = attrs.get( 'REPORTED', "" )
	98
[12]	99	debug_msg( 10, ' \| \|-Host found: %s - ip %s reported %s' %( self.hostName, self.hostIp, self.hostReported ) )
[6]	100
[14]	101	elif name == 'METRIC' and self.clusterName in ARCHIVE_SOURCES:
[6]	102
[32]	103	type = attrs.get( 'TYPE', "" )
[6]	104
[32]	105	if type not in UNSUPPORTED_ARCHIVE_TYPES:
[3]	106
[32]	107	myMetric = { }
	108	myMetric['name'] = attrs.get( 'NAME', "" )
	109	myMetric['val'] = attrs.get( 'VAL', "" )
	110	myMetric['time'] = self.hostReported
[3]	111
[34]	112	self.clusters[ self.clusterName ].memMetric( self.hostName, myMetric )
[3]	113
[34]	114	debug_msg( 11, ' \| \| \|-metric: %s:%s' %( myMetric['name'], myMetric['val'] ) )
[6]	115
[34]	116	def storeMetrics( self ):
[9]	117
[34]	118	for clustername, rrdh in self.clusters.items():
[16]	119
[34]	120	print 'storing metrics of cluster ' + clustername
	121	rrdh.storeMetrics()
[9]	122
[5]	123	class GangliaXMLGatherer:
[8]	124	"Setup a connection and file object to Ganglia's XML"
[3]	125
[8]	126	s = None
	127
	128	def __init__( self, host, port ):
	129	"Store host and port for connection"
	130
[5]	131	self.host = host
	132	self.port = port
[33]	133	self.connect()
[3]	134
[33]	135	def connect( self ):
	136	"Setup connection to XML source"
[8]	137
	138	for res in socket.getaddrinfo( self.host, self.port, socket.AF_UNSPEC, socket.SOCK_STREAM ):
[32]	139
[5]	140	af, socktype, proto, canonname, sa = res
[32]	141
[5]	142	try:
[32]	143
[8]	144	self.s = socket.socket( af, socktype, proto )
[32]	145
[5]	146	except socket.error, msg:
[32]	147
[8]	148	self.s = None
[5]	149	continue
[32]	150
[5]	151	try:
[32]	152
[8]	153	self.s.connect( sa )
[32]	154
[5]	155	except socket.error, msg:
[32]	156
[8]	157	self.s.close()
	158	self.s = None
[5]	159	continue
[32]	160
[5]	161	break
[3]	162
[8]	163	if self.s is None:
[32]	164
[33]	165	debug_msg( 0, 'Could not open socket' )
	166	sys.exit( 1 )
[5]	167
[33]	168	def disconnect( self ):
	169	"Close socket"
	170
	171	if self.s:
	172	self.s.close()
	173	self.s = None
	174
	175	def __del__( self ):
	176	"Kill the socket before we leave"
	177
	178	self.disconnect()
	179
	180	def getFileObject( self ):
	181	"Connect, and return a file object"
	182
	183	if not self.s:
	184	self.connect()
	185
[8]	186	return self.s.makefile( 'r' )
[5]	187
[8]	188	class GangliaXMLProcessor:
[5]	189
[33]	190	def __init__( self ):
	191	"Setup initial XML connection and handlers"
	192
	193	self.config = GangliaConfigParser( GMETAD_CONF )
	194
	195	self.myXMLGatherer = GangliaXMLGatherer( 'localhost', 8651 )
	196	self.myParser = make_parser()
	197	self.myHandler = GangliaXMLHandler( self.config )
	198	self.myParser.setContentHandler( self.myHandler )
	199
[9]	200	def daemon( self ):
[8]	201	"Run as daemon forever"
[5]	202
[8]	203	self.DAEMON = 1
[5]	204
[8]	205	# Fork the first child
	206	#
	207	pid = os.fork()
[32]	208
[8]	209	if pid > 0:
[7]	210
[32]	211	sys.exit(0) # end parent
	212
[8]	213	# creates a session and sets the process group ID
	214	#
	215	os.setsid()
[7]	216
[8]	217	# Fork the second child
	218	#
	219	pid = os.fork()
[32]	220
[8]	221	if pid > 0:
[5]	222
[32]	223	sys.exit(0) # end parent
	224
[8]	225	# Go to the root directory and set the umask
	226	#
	227	os.chdir('/')
	228	os.umask(0)
	229
	230	sys.stdin.close()
	231	sys.stdout.close()
[21]	232	sys.stderr.close()
[8]	233
	234	os.open('/dev/null', 0)
	235	os.dup(0)
	236	os.dup(0)
	237
	238	self.run()
	239
[22]	240	def printTime( self ):
[33]	241	"Print current time in human readable format"
[22]	242
[33]	243	return time.strftime("%a %d %b %Y %H:%M:%S")
[22]	244
[33]	245	def grabXML( self ):
	246
	247	debug_msg( 7, self.printTime() + ' - mainthread() - xmlthread() started' )
	248	pid = os.fork()
	249
	250	if pid == 0:
	251	# Child - XML Thread
	252	#
	253	# Process XML and exit
	254
	255	debug_msg( 7, self.printTime() + ' - xmlthread() - Start XML processing..' )
	256	self.processXML()
	257	debug_msg( 7, self.printTime() + ' - xmlthread() - Done processing; exiting.' )
	258	sys.exit( 0 )
	259
	260	elif pid > 0:
	261	# Parent - Time/sleep Thread
	262	#
	263	# Make sure XML is processed on time and at regular intervals
	264
[34]	265	debug_msg( 7, self.printTime() + ' - mainthread() - Sleep '+ str( self.config.getLowestInterval() ) +'s: zzzzz..' )
	266	time.sleep( float( self.config.getLowestInterval() ) )
[33]	267	debug_msg( 7, self.printTime() + ' - mainthread() - Awoken: waiting for XML thread..' )
	268
	269	r = os.wait()
	270	ret = r[1]
	271	if ret != 0:
	272	debug_msg( 7, self.printTime() + ' - mainthread() - Done waiting: ERROR! xmlthread() exited with status %d' %(ret) )
	273	if DEBUG_LEVEL>=7: sys.exit( 1 )
	274	else:
	275
	276	debug_msg( 7, self.printTime() + ' - mainthread() - Done waiting: xmlthread() finished succesfully' )
	277
[9]	278	def run( self ):
[8]	279	"Main thread"
	280
[33]	281	# Daemonized not working yet
	282	if DAEMONIZE:
[22]	283	pid = os.fork()
	284
[33]	285	# Handle XML grabbing in Child
[22]	286	if pid == 0:
	287
[33]	288	while( 1 ):
	289	self.grabXML()
[22]	290
[33]	291	# Do scheduled RRD storing in Parent
	292	#elif pid > ):
[22]	293
[33]	294	else:
[35]	295	#self.grabXML()
	296	self.processXML()
[33]	297	self.storeMetrics()
[22]	298
[33]	299	def storeMetrics( self ):
	300	"Store metrics retained in memory to disk"
[22]	301
[33]	302	self.myHandler.storeMetrics()
[22]	303
[8]	304	def processXML( self ):
	305	"Process XML"
	306
[33]	307	self.myParser.parse( self.myXMLGatherer.getFileObject() )
[8]	308
[9]	309	class GangliaConfigParser:
	310
[34]	311	sources = [ ]
[9]	312
	313	def __init__( self, config ):
[32]	314
[9]	315	self.config = config
	316	self.parseValues()
	317
[32]	318	def parseValues( self ):
[9]	319	"Parse certain values from gmetad.conf"
	320
	321	readcfg = open( self.config, 'r' )
	322
	323	for line in readcfg.readlines():
	324
	325	if line.count( '"' ) > 1:
	326
[10]	327	if line.find( 'data_source' ) != -1 and line[0] != '#':
[9]	328
[11]	329	source = { }
	330	source['name'] = line.split( '"' )[1]
[9]	331	source_words = line.split( '"' )[2].split( ' ' )
	332
	333	for word in source_words:
	334
	335	valid_interval = 1
	336
	337	for letter in word:
[32]	338
[9]	339	if letter not in string.digits:
[32]	340
[9]	341	valid_interval = 0
	342
[10]	343	if valid_interval and len(word) > 0:
[32]	344
[9]	345	source['interval'] = word
[12]	346	debug_msg( 9, 'polling interval for %s = %s' %(source['name'], source['interval'] ) )
[33]	347
	348	# No interval found, use Ganglia's default
	349	if not source.has_key( 'interval' ):
	350	source['interval'] = 15
	351	debug_msg( 9, 'polling interval for %s defaulted to 15' %(source['name']) )
[32]	352
[33]	353	self.sources.append( source )
[9]	354
	355	def getInterval( self, source_name ):
[32]	356
[9]	357	for source in self.sources:
[32]	358
[12]	359	if source['name'] == source_name:
[32]	360
[9]	361	return source['interval']
[32]	362
[9]	363	return None
	364
[34]	365	def getLowestInterval( self ):
	366
	367	lowest_interval = 0
	368
	369	for source in self.sources:
	370
	371	if not lowest_interval or source['interval'] <= lowest_interval:
	372
	373	lowest_interval = source['interval']
	374
	375	# Return 15 when nothing is found, so that the daemon won't go insane with 0 sec delays
	376	if lowest_interval:
	377	return lowest_interval
	378	else:
	379	return 15
	380
[9]	381	class RRDHandler:
	382
[32]	383	myMetrics = { }
	384
[33]	385	def __init__( self, config, cluster ):
[9]	386	self.cluster = cluster
[33]	387	self.config = config
[9]	388
[32]	389	def getClusterName( self ):
	390	return self.cluster
	391
	392	def memMetric( self, host, metric ):
	393
[34]	394	if self.myMetrics.has_key( host ):
[32]	395
[34]	396	if self.myMetrics[ host ].has_key( metric['name'] ):
[32]	397
[34]	398	for mymetric in self.myMetrics[ host ][ metric['name'] ]:
[32]	399
[34]	400	if mymetric['time'] == metric['time']:
[32]	401
[34]	402	# Allready have this metric, abort
	403	return 1
	404	else:
	405	self.myMetrics[ host ][ metric['name'] ] = [ ]
	406	else:
[32]	407	self.myMetrics[ host ] = { }
[34]	408	self.myMetrics[ host ][ metric['name'] ] = [ ]
[32]	409
	410
	411	self.myMetrics[ host ][ metric['name'] ].append( metric )
	412
[35]	413	def makeUpdateString( self, host, metricname ):
[32]	414
	415	update_string = ''
	416
[35]	417	for m in self.myMetrics[ host ][ metricname ]:
[32]	418
[35]	419	update_string = update_string + ' %s:%s' %( m['time'], m['val'] )
[32]	420
	421	return update_string
	422
[33]	423	def storeMetrics( self ):
	424
	425	for hostname, mymetrics in self.myMetrics.items():
	426
	427	for metricname, mymetric in mymetrics.items():
	428
[35]	429	mytime = self.makeTimeSerial()
	430	self.createCheck( hostname, metricname, mytime )
	431	update_okay = self.update( hostname, metricname, mytime )
[33]	432
	433	if not update_okay:
	434
	435	del self.myMetrics[ hostname ][ metricname ]
	436	debug_msg( 9, 'stored metric %s for %s' %( hostname, metricname ) )
	437	else:
	438	debug_msg( 9, 'metric update failed' )
	439
	440	sys.exit(1)
	441
[17]	442	def makeTimeSerial( self ):
[32]	443	"Generate a time serial. Seconds since epoch"
[17]	444
	445	# Seconds since epoch
	446	mytime = int( time.time() )
	447
	448	return mytime
	449
[33]	450	def makeRrdPath( self, host, metricname=None, timeserial=None ):
[32]	451	"""
	452	Make a RRD location/path and filename
	453	If a metric or timeserial are supplied the complete locations
	454	will be made, else just the host directory
	455	"""
[17]	456
[20]	457	if not timeserial:
	458	rrd_dir = '%s/%s/%s' %( check_dir(ARCHIVE_PATH), self.cluster, host )
	459	else:
	460	rrd_dir = '%s/%s/%s/%s' %( check_dir(ARCHIVE_PATH), self.cluster, host, timeserial )
[35]	461	if metricname:
[33]	462	rrd_file = '%s/%s.rrd' %( rrd_dir, metricname )
[20]	463	else:
	464	rrd_file = None
[17]	465
	466	return rrd_dir, rrd_file
	467
[20]	468	def getLastRrdTimeSerial( self, host ):
[32]	469	"""
	470	Find the last timeserial (directory) for this host
	471	This is determined once every host
	472	"""
[17]	473
[20]	474	rrd_dir, rrd_file = self.makeRrdPath( host )
[17]	475
[19]	476	newest_timeserial = 0
	477
[17]	478	if os.path.exists( rrd_dir ):
[32]	479
[17]	480	for root, dirs, files in os.walk( rrd_dir ):
	481
[20]	482	for dir in dirs:
[17]	483
[20]	484	valid_dir = 1
[17]	485
[20]	486	for letter in dir:
	487	if letter not in string.digits:
	488	valid_dir = 0
[17]	489
[20]	490	if valid_dir:
	491	timeserial = dir
[17]	492	if timeserial > newest_timeserial:
	493	newest_timeserial = timeserial
	494
	495	if newest_timeserial:
[18]	496	return newest_timeserial
[17]	497	else:
	498	return 0
	499
[20]	500	def checkNewRrdPeriod( self, host, current_timeserial ):
[32]	501	"""
	502	Check if current timeserial belongs to recent time period
	503	or should become a new period (and file).
[17]	504
[32]	505	Returns the serial of the correct time period
	506	"""
	507
[20]	508	last_timeserial = int( self.getLastRrdTimeSerial( host ) )
[30]	509	debug_msg( 9, 'last timeserial of %s is %s' %( host, last_timeserial ) )
[17]	510
	511	if not last_timeserial:
[18]	512	serial = current_timeserial
	513	else:
[17]	514
[18]	515	archive_secs = ARCHIVE_HOURS_PER_RRD * (60 * 60)
[17]	516
[18]	517	if (current_timeserial - last_timeserial) >= archive_secs:
	518	serial = current_timeserial
	519	else:
	520	serial = last_timeserial
[17]	521
[18]	522	return serial
	523
[33]	524	def getFirstTime( self, host, metricname ):
	525	"Get the first time of a metric we know of"
	526
	527	first_time = 0
	528
	529	for metric in self.myMetrics[ host ][ metricname ]:
	530
	531	if not first_time or metric['time'] <= first_time:
	532
	533	first_time = metric['time']
	534
[35]	535	return first_time
	536
[33]	537	def createCheck( self, host, metricname, timeserial ):
[9]	538	"Check if an .rrd allready exists for this metric, create if not"
	539
[35]	540	debug_msg( 9, 'rrdcreate: using timeserial %s for %s/%s' %( timeserial, host, metricname ) )
[9]	541
[33]	542	rrd_dir, rrd_file = self.makeRrdPath( host, metricname, timeserial )
[17]	543
[9]	544	if not os.path.exists( rrd_dir ):
	545	os.makedirs( rrd_dir )
[14]	546	debug_msg( 9, 'created dir %s' %( str(rrd_dir) ) )
[9]	547
[14]	548	if not os.path.exists( rrd_file ):
[9]	549
[33]	550	interval = self.config.getInterval( self.cluster )
[14]	551	heartbeat = 8 * int(interval)
[9]	552
[14]	553	param_step1 = '--step'
	554	param_step2 = str( interval )
[12]	555
[14]	556	param_start1 = '--start'
[33]	557	param_start2 = str( int( self.getFirstTime( host, metricname ) ) - 1 )
[12]	558
[14]	559	param_ds = 'DS:sum:GAUGE:%d:U:U' %heartbeat
	560	param_rra = 'RRA:AVERAGE:0.5:1:%s' %(ARCHIVE_HOURS_PER_RRD * 240)
[12]	561
[14]	562	rrdtool.create( str(rrd_file), param_step1, param_step2, param_start1, param_start2, param_ds, param_rra )
[13]	563
[14]	564	debug_msg( 9, 'created rrd %s' %( str(rrd_file) ) )
	565
[33]	566	def update( self, host, metricname, timeserial ):
[9]	567
[35]	568	debug_msg( 9, 'rrdupdate: using timeserial %s for %s/%s' %( timeserial, host, metricname ) )
[9]	569
[33]	570	rrd_dir, rrd_file = self.makeRrdPath( host, metricname, timeserial )
[18]	571
[33]	572	#timestamp = metric['time']
	573	#val = metric['val']
[9]	574
[33]	575	#update_string = '%s:%s' %(timestamp, val)
	576	update_string = self.makeUpdateString( host, metricname )
[15]	577
[27]	578	try:
[32]	579
[27]	580	rrdtool.update( str(rrd_file), str(update_string) )
[32]	581
[28]	582	except rrdtool.error, detail:
[32]	583
[27]	584	debug_msg( 0, 'EXCEPTION! While trying to update rrd:' )
	585	debug_msg( 0, '\trrd %s with %s' %( str(rrd_file), update_string ) )
[28]	586	debug_msg( 0, str(detail) )
[32]	587
[33]	588	return 1
[27]	589
[15]	590	debug_msg( 9, 'updated rrd %s with %s' %( str(rrd_file), update_string ) )
	591
[8]	592	def main():
	593	"Program startup"
	594
	595	myProcessor = GangliaXMLProcessor()
	596
[22]	597	if DAEMONIZE:
	598	myProcessor.daemon()
	599	else:
	600	myProcessor.run()
	601
[9]	602	def check_dir( directory ):
	603	"Check if directory is a proper directory. I.e.: Does _not_ end with a '/'"
	604
	605	if directory[-1] == '/':
	606	directory = directory[:-1]
	607
	608	return directory
	609
[12]	610	def debug_msg( level, msg ):
	611
	612	if (DEBUG_LEVEL >= level):
	613	sys.stderr.write( msg + '\n' )
	614
[5]	615	# Let's go
[9]	616	if __name__ == '__main__':
	617	main()

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

Download in other formats: