Context Navigation

source: trunk/daemon/togad.py @ 34

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

daemon/togad.py:

Numerous bugfixes Data is still not written, storeMetrics doesn't run

File size: 13.7 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	clusters = { }
	62	config = None
[6]	63
[33]	64	def __init__( self, config ):
	65	self.config = config
	66
[6]	67	def startElement( self, name, attrs ):
[8]	68	"Store appropriate data from xml start tags"
[3]	69
[7]	70	if name == 'GANGLIA_XML':
[32]	71
	72	self.XMLSource = attrs.get( 'SOURCE', "" )
	73	self.gangliaVersion = attrs.get( 'VERSION', "" )
	74
[12]	75	debug_msg( 10, 'Found XML data: source %s version %s' %( self.XMLSource, self.gangliaVersion ) )
[6]	76
[7]	77	elif name == 'GRID':
[32]	78
	79	self.gridName = attrs.get( 'NAME', "" )
	80	self.time = attrs.get( 'LOCALTIME', "" )
	81
[12]	82	debug_msg( 10, '`-Grid found: %s' %( self.gridName ) )
[6]	83
[7]	84	elif name == 'CLUSTER':
[32]	85
	86	self.clusterName = attrs.get( 'NAME', "" )
	87	self.time = attrs.get( 'LOCALTIME', "" )
	88
[33]	89	if not self.clusters.has_key( self.clusterName ):
[32]	90
[34]	91	self.clusters[ self.clusterName ] = RRDHandler( self.config, self.clusterName )
[33]	92
[12]	93	debug_msg( 10, ' \|-Cluster found: %s' %( self.clusterName ) )
[6]	94
[14]	95	elif name == 'HOST' and self.clusterName in ARCHIVE_SOURCES:
[32]	96
	97	self.hostName = attrs.get( 'NAME', "" )
	98	self.hostIp = attrs.get( 'IP', "" )
	99	self.hostReported = attrs.get( 'REPORTED', "" )
	100
[12]	101	debug_msg( 10, ' \| \|-Host found: %s - ip %s reported %s' %( self.hostName, self.hostIp, self.hostReported ) )
[6]	102
[14]	103	elif name == 'METRIC' and self.clusterName in ARCHIVE_SOURCES:
[6]	104
[32]	105	type = attrs.get( 'TYPE', "" )
[6]	106
[32]	107	if type not in UNSUPPORTED_ARCHIVE_TYPES:
[3]	108
[32]	109	myMetric = { }
	110	myMetric['name'] = attrs.get( 'NAME', "" )
	111	myMetric['val'] = attrs.get( 'VAL', "" )
	112	myMetric['time'] = self.hostReported
[3]	113
[34]	114	self.clusters[ self.clusterName ].memMetric( self.hostName, myMetric )
[3]	115
[34]	116	debug_msg( 11, ' \| \| \|-metric: %s:%s' %( myMetric['name'], myMetric['val'] ) )
[6]	117
[34]	118	def storeMetrics( self ):
[9]	119
[34]	120	for clustername, rrdh in self.clusters.items():
[16]	121
[34]	122	print 'storing metrics of cluster ' + clustername
	123	rrdh.storeMetrics()
[9]	124
[5]	125	class GangliaXMLGatherer:
[8]	126	"Setup a connection and file object to Ganglia's XML"
[3]	127
[8]	128	s = None
	129
	130	def __init__( self, host, port ):
	131	"Store host and port for connection"
	132
[5]	133	self.host = host
	134	self.port = port
[33]	135	self.connect()
[3]	136
[33]	137	def connect( self ):
	138	"Setup connection to XML source"
[8]	139
	140	for res in socket.getaddrinfo( self.host, self.port, socket.AF_UNSPEC, socket.SOCK_STREAM ):
[32]	141
[5]	142	af, socktype, proto, canonname, sa = res
[32]	143
[5]	144	try:
[32]	145
[8]	146	self.s = socket.socket( af, socktype, proto )
[32]	147
[5]	148	except socket.error, msg:
[32]	149
[8]	150	self.s = None
[5]	151	continue
[32]	152
[5]	153	try:
[32]	154
[8]	155	self.s.connect( sa )
[32]	156
[5]	157	except socket.error, msg:
[32]	158
[8]	159	self.s.close()
	160	self.s = None
[5]	161	continue
[32]	162
[5]	163	break
[3]	164
[8]	165	if self.s is None:
[32]	166
[33]	167	debug_msg( 0, 'Could not open socket' )
	168	sys.exit( 1 )
[5]	169
[33]	170	def disconnect( self ):
	171	"Close socket"
	172
	173	if self.s:
	174	self.s.close()
	175	self.s = None
	176
	177	def __del__( self ):
	178	"Kill the socket before we leave"
	179
	180	self.disconnect()
	181
	182	def getFileObject( self ):
	183	"Connect, and return a file object"
	184
	185	if not self.s:
	186	self.connect()
	187
[8]	188	return self.s.makefile( 'r' )
[5]	189
[8]	190	class GangliaXMLProcessor:
[5]	191
[33]	192	def __init__( self ):
	193	"Setup initial XML connection and handlers"
	194
	195	self.config = GangliaConfigParser( GMETAD_CONF )
	196
	197	self.myXMLGatherer = GangliaXMLGatherer( 'localhost', 8651 )
	198	self.myParser = make_parser()
	199	self.myHandler = GangliaXMLHandler( self.config )
	200	self.myParser.setContentHandler( self.myHandler )
	201
[9]	202	def daemon( self ):
[8]	203	"Run as daemon forever"
[5]	204
[8]	205	self.DAEMON = 1
[5]	206
[8]	207	# Fork the first child
	208	#
	209	pid = os.fork()
[32]	210
[8]	211	if pid > 0:
[7]	212
[32]	213	sys.exit(0) # end parent
	214
[8]	215	# creates a session and sets the process group ID
	216	#
	217	os.setsid()
[7]	218
[8]	219	# Fork the second child
	220	#
	221	pid = os.fork()
[32]	222
[8]	223	if pid > 0:
[5]	224
[32]	225	sys.exit(0) # end parent
	226
[8]	227	# Go to the root directory and set the umask
	228	#
	229	os.chdir('/')
	230	os.umask(0)
	231
	232	sys.stdin.close()
	233	sys.stdout.close()
[21]	234	sys.stderr.close()
[8]	235
	236	os.open('/dev/null', 0)
	237	os.dup(0)
	238	os.dup(0)
	239
	240	self.run()
	241
[22]	242	def printTime( self ):
[33]	243	"Print current time in human readable format"
[22]	244
[33]	245	return time.strftime("%a %d %b %Y %H:%M:%S")
[22]	246
[33]	247	def grabXML( self ):
	248
	249	debug_msg( 7, self.printTime() + ' - mainthread() - xmlthread() started' )
	250	pid = os.fork()
	251
	252	if pid == 0:
	253	# Child - XML Thread
	254	#
	255	# Process XML and exit
	256
	257	debug_msg( 7, self.printTime() + ' - xmlthread() - Start XML processing..' )
	258	self.processXML()
	259	debug_msg( 7, self.printTime() + ' - xmlthread() - Done processing; exiting.' )
	260	sys.exit( 0 )
	261
	262	elif pid > 0:
	263	# Parent - Time/sleep Thread
	264	#
	265	# Make sure XML is processed on time and at regular intervals
	266
[34]	267	debug_msg( 7, self.printTime() + ' - mainthread() - Sleep '+ str( self.config.getLowestInterval() ) +'s: zzzzz..' )
	268	time.sleep( float( self.config.getLowestInterval() ) )
[33]	269	debug_msg( 7, self.printTime() + ' - mainthread() - Awoken: waiting for XML thread..' )
	270
	271	r = os.wait()
	272	ret = r[1]
	273	if ret != 0:
	274	debug_msg( 7, self.printTime() + ' - mainthread() - Done waiting: ERROR! xmlthread() exited with status %d' %(ret) )
	275	if DEBUG_LEVEL>=7: sys.exit( 1 )
	276	else:
	277
	278	debug_msg( 7, self.printTime() + ' - mainthread() - Done waiting: xmlthread() finished succesfully' )
	279
[9]	280	def run( self ):
[8]	281	"Main thread"
	282
[33]	283	# Daemonized not working yet
	284	if DAEMONIZE:
[22]	285	pid = os.fork()
	286
[33]	287	# Handle XML grabbing in Child
[22]	288	if pid == 0:
	289
[33]	290	while( 1 ):
	291	self.grabXML()
[22]	292
[33]	293	# Do scheduled RRD storing in Parent
	294	#elif pid > ):
[22]	295
[33]	296	else:
	297	self.grabXML()
	298	self.storeMetrics()
[22]	299
[33]	300	def storeMetrics( self ):
	301	"Store metrics retained in memory to disk"
[22]	302
[33]	303	self.myHandler.storeMetrics()
[22]	304
[8]	305	def processXML( self ):
	306	"Process XML"
	307
[33]	308	self.myParser.parse( self.myXMLGatherer.getFileObject() )
[8]	309
[9]	310	class GangliaConfigParser:
	311
[34]	312	sources = [ ]
[9]	313
	314	def __init__( self, config ):
[32]	315
[9]	316	self.config = config
	317	self.parseValues()
	318
[32]	319	def parseValues( self ):
[9]	320	"Parse certain values from gmetad.conf"
	321
	322	readcfg = open( self.config, 'r' )
	323
	324	for line in readcfg.readlines():
	325
	326	if line.count( '"' ) > 1:
	327
[10]	328	if line.find( 'data_source' ) != -1 and line[0] != '#':
[9]	329
[11]	330	source = { }
	331	source['name'] = line.split( '"' )[1]
[9]	332	source_words = line.split( '"' )[2].split( ' ' )
	333
	334	for word in source_words:
	335
	336	valid_interval = 1
	337
	338	for letter in word:
[32]	339
[9]	340	if letter not in string.digits:
[32]	341
[9]	342	valid_interval = 0
	343
[10]	344	if valid_interval and len(word) > 0:
[32]	345
[9]	346	source['interval'] = word
[12]	347	debug_msg( 9, 'polling interval for %s = %s' %(source['name'], source['interval'] ) )
[33]	348
	349	# No interval found, use Ganglia's default
	350	if not source.has_key( 'interval' ):
	351	source['interval'] = 15
	352	debug_msg( 9, 'polling interval for %s defaulted to 15' %(source['name']) )
[32]	353
[33]	354	self.sources.append( source )
[9]	355
	356	def getInterval( self, source_name ):
[32]	357
[9]	358	for source in self.sources:
[32]	359
[12]	360	if source['name'] == source_name:
[32]	361
[9]	362	return source['interval']
[32]	363
[9]	364	return None
	365
[34]	366	def getLowestInterval( self ):
	367
	368	lowest_interval = 0
	369
	370	for source in self.sources:
	371
	372	if not lowest_interval or source['interval'] <= lowest_interval:
	373
	374	lowest_interval = source['interval']
	375
	376	# Return 15 when nothing is found, so that the daemon won't go insane with 0 sec delays
	377	if lowest_interval:
	378	return lowest_interval
	379	else:
	380	return 15
	381
[9]	382	class RRDHandler:
	383
[32]	384	myMetrics = { }
	385
[33]	386	def __init__( self, config, cluster ):
[9]	387	self.cluster = cluster
[33]	388	self.config = config
[9]	389
[32]	390	def getClusterName( self ):
	391	return self.cluster
	392
	393	def memMetric( self, host, metric ):
	394
[34]	395	if self.myMetrics.has_key( host ):
[32]	396
[34]	397	if self.myMetrics[ host ].has_key( metric['name'] ):
[32]	398
[34]	399	for mymetric in self.myMetrics[ host ][ metric['name'] ]:
[32]	400
[34]	401	if mymetric['time'] == metric['time']:
[32]	402
[34]	403	# Allready have this metric, abort
	404	return 1
	405	else:
	406	self.myMetrics[ host ][ metric['name'] ] = [ ]
	407	else:
[32]	408	self.myMetrics[ host ] = { }
[34]	409	self.myMetrics[ host ][ metric['name'] ] = [ ]
[32]	410
	411
	412	self.myMetrics[ host ][ metric['name'] ].append( metric )
	413
	414	def makeUpdateString( self, host, metric ):
	415
	416	update_string = ''
	417
	418	for m in self.myMetrics[ host ][ metric['name'] ]:
	419
	420	update_string = update_string + ' %s:%s' %( metric['time'], metric['val'] )
	421
	422	return update_string
	423
[33]	424	def storeMetrics( self ):
	425
	426	for hostname, mymetrics in self.myMetrics.items():
	427
	428	for metricname, mymetric in mymetrics.items():
	429
	430	self.rrd.createCheck( hostname, metricname, timeserial )
	431	update_okay = self.rrd.update( hostname, metricname, timeserial )
	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 )
	461	if metric:
[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
	535	def createCheck( self, host, metricname, timeserial ):
[9]	536	"Check if an .rrd allready exists for this metric, create if not"
	537
[30]	538	debug_msg( 9, 'rrdcreate: using timeserial %s for %s/%s' %( timeserial, host, metric['name'] ) )
[9]	539
[33]	540	rrd_dir, rrd_file = self.makeRrdPath( host, metricname, timeserial )
[17]	541
[9]	542	if not os.path.exists( rrd_dir ):
	543	os.makedirs( rrd_dir )
[14]	544	debug_msg( 9, 'created dir %s' %( str(rrd_dir) ) )
[9]	545
[14]	546	if not os.path.exists( rrd_file ):
[9]	547
[33]	548	interval = self.config.getInterval( self.cluster )
[14]	549	heartbeat = 8 * int(interval)
[9]	550
[14]	551	param_step1 = '--step'
	552	param_step2 = str( interval )
[12]	553
[14]	554	param_start1 = '--start'
[33]	555	param_start2 = str( int( self.getFirstTime( host, metricname ) ) - 1 )
[12]	556
[14]	557	param_ds = 'DS:sum:GAUGE:%d:U:U' %heartbeat
	558	param_rra = 'RRA:AVERAGE:0.5:1:%s' %(ARCHIVE_HOURS_PER_RRD * 240)
[12]	559
[14]	560	rrdtool.create( str(rrd_file), param_step1, param_step2, param_start1, param_start2, param_ds, param_rra )
[13]	561
[14]	562	debug_msg( 9, 'created rrd %s' %( str(rrd_file) ) )
	563
[33]	564	def update( self, host, metricname, timeserial ):
[9]	565
[30]	566	debug_msg( 9, 'rrdupdate: using timeserial %s for %s/%s' %( timeserial, host, metric['name'] ) )
[9]	567
[33]	568	rrd_dir, rrd_file = self.makeRrdPath( host, metricname, timeserial )
[18]	569
[33]	570	#timestamp = metric['time']
	571	#val = metric['val']
[9]	572
[33]	573	#update_string = '%s:%s' %(timestamp, val)
	574	update_string = self.makeUpdateString( host, metricname )
[15]	575
[27]	576	try:
[32]	577
[27]	578	rrdtool.update( str(rrd_file), str(update_string) )
[32]	579
[28]	580	except rrdtool.error, detail:
[32]	581
[27]	582	debug_msg( 0, 'EXCEPTION! While trying to update rrd:' )
	583	debug_msg( 0, '\trrd %s with %s' %( str(rrd_file), update_string ) )
[28]	584	debug_msg( 0, str(detail) )
[32]	585
[33]	586	return 1
[27]	587
[15]	588	debug_msg( 9, 'updated rrd %s with %s' %( str(rrd_file), update_string ) )
	589
[8]	590	def main():
	591	"Program startup"
	592
	593	myProcessor = GangliaXMLProcessor()
	594
[22]	595	if DAEMONIZE:
	596	myProcessor.daemon()
	597	else:
	598	myProcessor.run()
	599
[9]	600	def check_dir( directory ):
	601	"Check if directory is a proper directory. I.e.: Does _not_ end with a '/'"
	602
	603	if directory[-1] == '/':
	604	directory = directory[:-1]
	605
	606	return directory
	607
[12]	608	def debug_msg( level, msg ):
	609
	610	if (DEBUG_LEVEL >= level):
	611	sys.stderr.write( msg + '\n' )
	612
[5]	613	# Let's go
[9]	614	if __name__ == '__main__':
	615	main()

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

Download in other formats: