Context Navigation

source: trunk/daemon/togad.py @ 31

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

daemon/togad.py:

Bug was caused by incorrect timestamping

Now host reporting time is used

New bug: when a ganglia monitor has a big interval
it will report the same value for a second xmlthread run.

One cannot update an rrd twice with the same timestamp
Will have to check timestamp of metrics each run

This implicates that for every metric for every host the
timestamp of the last xml run must be kept in memory

Might as well rewrite rrd code with in-memory metric storage
And do scheduled/queued rrd updates, in stead of every xml run: big performance increase

File size: 11.1 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	#
[30]	22	DEBUG_LEVEL = 8
[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
[21]	40	# Interval at which to grab&store XML
	41	#
	42	GRAB_INTERVAL = 15
	43
[22]	44	# Wether or not to run as a daemon in background
	45	#
	46	DAEMONIZE = 0
	47
[17]	48	######################
	49	# #
	50	# Configuration ends #
	51	# #
	52	######################
[13]	53
[17]	54	# What XML data types not to store
[13]	55	#
[17]	56	UNSUPPORTED_ARCHIVE_TYPES = [ 'string' ]
[9]	57
[8]	58	"""
	59	This is TOrque-GAnglia's data Daemon
	60	"""
	61
[6]	62	class GangliaXMLHandler( ContentHandler ):
[8]	63	"Parse Ganglia's XML"
[3]	64
[6]	65	metrics = [ ]
	66
	67	def startElement( self, name, attrs ):
[8]	68	"Store appropriate data from xml start tags"
[3]	69
[7]	70	if name == 'GANGLIA_XML':
	71	self.XMLSource = attrs.get('SOURCE',"")
	72	self.gangliaVersion = attrs.get('VERSION',"")
[12]	73	debug_msg( 10, 'Found XML data: source %s version %s' %( self.XMLSource, self.gangliaVersion ) )
[6]	74
[7]	75	elif name == 'GRID':
	76	self.gridName = attrs.get('NAME',"")
[12]	77	self.time = attrs.get('LOCALTIME',"")
	78	debug_msg( 10, '`-Grid found: %s' %( self.gridName ) )
[6]	79
[7]	80	elif name == 'CLUSTER':
	81	self.clusterName = attrs.get('NAME',"")
[12]	82	self.time = attrs.get('LOCALTIME',"")
[9]	83	self.rrd = RRDHandler( self.clusterName )
[12]	84	debug_msg( 10, ' \|-Cluster found: %s' %( self.clusterName ) )
[6]	85
[14]	86	elif name == 'HOST' and self.clusterName in ARCHIVE_SOURCES:
[7]	87	self.hostName = attrs.get('NAME',"")
	88	self.hostIp = attrs.get('IP',"")
	89	self.hostReported = attrs.get('REPORTED',"")
[9]	90	# Reset the metrics list for each host
	91	self.metrics = [ ]
[12]	92	debug_msg( 10, ' \| \|-Host found: %s - ip %s reported %s' %( self.hostName, self.hostIp, self.hostReported ) )
[6]	93
[14]	94	elif name == 'METRIC' and self.clusterName in ARCHIVE_SOURCES:
[6]	95	myMetric = { }
[7]	96	myMetric['name'] = attrs.get('NAME',"")
	97	myMetric['val'] = attrs.get('VAL',"")
[31]	98	myMetric['time'] = self.hostReported
[16]	99	myMetric['type'] = attrs.get('TYPE',"")
[6]	100
	101	self.metrics.append( myMetric )
[12]	102	debug_msg( 11, ' \| \| \|-metric: %s:%s' %( myMetric['name'], myMetric['val'] ) )
[6]	103
[5]	104	return
[3]	105
[9]	106	def endElement( self, name ):
[12]	107	#if name == 'GANGLIA_XML':
[3]	108
[12]	109	#if name == 'GRID':
[3]	110
[12]	111	#if name == 'CLUSTER':
[6]	112
[14]	113	if name == 'HOST' and self.clusterName in ARCHIVE_SOURCES:
[6]	114
[18]	115	# Determine time here, so all use same time in this run
	116	mytime = self.rrd.makeTimeSerial()
[20]	117	correct_serial = self.rrd.checkNewRrdPeriod( self.hostName, mytime )
[18]	118
[31]	119	#debug_msg( 8, 'time %s: Storing metrics for %s' %(mytime, self.hostName) )
	120	print 'Storing metrics for %s:' %(self.hostName),
[20]	121	self.storeMetrics( self.hostName, correct_serial )
[18]	122
[12]	123	#if name == 'METRIC':
[6]	124
[18]	125	def storeMetrics( self, hostname, timeserial ):
[9]	126
	127	for metric in self.metrics:
[17]	128	if metric['type'] not in UNSUPPORTED_ARCHIVE_TYPES:
[16]	129
[18]	130	self.rrd.createCheck( hostname, metric, timeserial )
[31]	131	print ' [%s.%s]' %(metric['name'], metric['time']),
[18]	132	self.rrd.update( hostname, metric, timeserial )
[16]	133	debug_msg( 9, 'stored metric %s for %s: %s' %( hostname, metric['name'], metric['val'] ) )
[19]	134	#sys.exit(1)
[31]	135	print
[9]	136
	137
[5]	138	class GangliaXMLGatherer:
[8]	139	"Setup a connection and file object to Ganglia's XML"
[3]	140
[8]	141	s = None
	142
	143	def __init__( self, host, port ):
	144	"Store host and port for connection"
	145
[5]	146	self.host = host
	147	self.port = port
[3]	148
[8]	149	def __del__( self ):
	150	"Kill the socket before we leave"
	151
	152	self.s.close()
	153
	154	def getFileObject( self ):
	155	"Connect, and return a file object"
	156
	157	for res in socket.getaddrinfo( self.host, self.port, socket.AF_UNSPEC, socket.SOCK_STREAM ):
[5]	158	af, socktype, proto, canonname, sa = res
	159	try:
[8]	160	self.s = socket.socket( af, socktype, proto )
[5]	161	except socket.error, msg:
[8]	162	self.s = None
[5]	163	continue
	164	try:
[8]	165	self.s.connect( sa )
[5]	166	except socket.error, msg:
[8]	167	self.s.close()
	168	self.s = None
[5]	169	continue
	170	break
[3]	171
[8]	172	if self.s is None:
	173	print 'Could not open socket'
[5]	174	sys.exit(1)
	175
[8]	176	return self.s.makefile( 'r' )
[5]	177
[8]	178	class GangliaXMLProcessor:
[5]	179
[9]	180	def daemon( self ):
[8]	181	"Run as daemon forever"
[5]	182
[8]	183	self.DAEMON = 1
[5]	184
[8]	185	# Fork the first child
	186	#
	187	pid = os.fork()
	188	if pid > 0:
	189	sys.exit(0) # end parrent
[7]	190
[8]	191	# creates a session and sets the process group ID
	192	#
	193	os.setsid()
[7]	194
[8]	195	# Fork the second child
	196	#
	197	pid = os.fork()
	198	if pid > 0:
	199	sys.exit(0) # end parrent
[5]	200
[8]	201	# Go to the root directory and set the umask
	202	#
	203	os.chdir('/')
	204	os.umask(0)
	205
	206	sys.stdin.close()
	207	sys.stdout.close()
[21]	208	sys.stderr.close()
[8]	209
	210	os.open('/dev/null', 0)
	211	os.dup(0)
	212	os.dup(0)
	213
	214	self.run()
	215
[22]	216	def printTime( self ):
	217
	218	return time.strftime("%a, %d %b %Y %H:%M:%S")
	219
[9]	220	def run( self ):
[8]	221	"Main thread"
	222
	223	while ( 1 ):
	224
[22]	225	debug_msg( 7, self.printTime() + ' - mainthread() - xmlthread() started' )
	226	pid = os.fork()
	227
	228	if pid == 0:
	229	# Child - XML Thread
	230	#
	231	# Process XML and exit
	232
	233	debug_msg( 7, self.printTime() + ' - xmlthread() - Start XML processing..' )
	234	self.processXML()
	235	debug_msg( 7, self.printTime() + ' - xmlthread() - Done processing; exiting.' )
[28]	236	sys.exit( 0 )
[22]	237
	238	elif pid > 0:
	239	# Parent - Daemon Thread
	240
	241	debug_msg( 7, self.printTime() + ' - mainthread() - Sleep '+ str(GRAB_INTERVAL) +'s: zzzzz..' )
	242	time.sleep( GRAB_INTERVAL )
	243	debug_msg( 7, self.printTime() + ' - mainthread() - Awoken: waiting for XML thread..' )
	244
	245	r = os.wait()
[28]	246	ret = r[1]
	247	if ret != 0:
	248	debug_msg( 7, self.printTime() + ' - mainthread() - Done waiting: ERROR! xmlthread() exited with status %d' %(ret) )
[29]	249	if DEBUG_LEVEL>=7: sys.exit( 1 )
[28]	250	else:
[22]	251
[28]	252	debug_msg( 7, self.printTime() + ' - mainthread() - Done waiting: xmlthread() finished succesfully' )
[22]	253
[8]	254	def processXML( self ):
	255	"Process XML"
	256
	257	myXMLGatherer = GangliaXMLGatherer( 'localhost', 8651 )
	258
	259	myParser = make_parser()
	260	myHandler = GangliaXMLHandler()
	261	myParser.setContentHandler( myHandler )
	262
	263	myParser.parse( myXMLGatherer.getFileObject() )
	264
[9]	265	class GangliaConfigParser:
	266
	267	sources = [ ]
	268
	269	def __init__( self, config ):
	270	self.config = config
	271	self.parseValues()
	272
	273	def parseValues(self):
	274	"Parse certain values from gmetad.conf"
	275
	276	readcfg = open( self.config, 'r' )
	277
	278	for line in readcfg.readlines():
	279
	280	if line.count( '"' ) > 1:
	281
[10]	282	if line.find( 'data_source' ) != -1 and line[0] != '#':
[9]	283
[11]	284	source = { }
	285	source['name'] = line.split( '"' )[1]
[9]	286	source_words = line.split( '"' )[2].split( ' ' )
	287
	288	for word in source_words:
	289
	290	valid_interval = 1
	291
	292	for letter in word:
	293	if letter not in string.digits:
	294	valid_interval = 0
	295
[10]	296	if valid_interval and len(word) > 0:
[9]	297	source['interval'] = word
[12]	298	debug_msg( 9, 'polling interval for %s = %s' %(source['name'], source['interval'] ) )
[9]	299
	300	# No interval found, use Ganglia's default
	301	if not source.has_key( 'interval' ):
	302	source['interval'] = 15
[12]	303	debug_msg( 9, 'polling interval for %s defaulted to 15' %(source['name']) )
[9]	304
	305	self.sources.append( source )
	306
	307	def getInterval( self, source_name ):
	308	for source in self.sources:
[12]	309	if source['name'] == source_name:
[9]	310	return source['interval']
	311	return None
	312
	313	class RRDHandler:
	314
	315	def __init__( self, cluster ):
	316	self.cluster = cluster
	317	self.gmetad_conf = GangliaConfigParser( GMETAD_CONF )
	318
[17]	319	def makeTimeSerial( self ):
	320
	321	# YYYYMMDDhhmmss: 20050321143411
	322	#mytime = time.strftime( "%Y%m%d%H%M%S" )
	323
	324	# Seconds since epoch
	325	mytime = int( time.time() )
	326
	327	return mytime
	328
[20]	329	def makeRrdPath( self, host, metric=None, timeserial=None ):
[17]	330
[20]	331	if not timeserial:
	332	rrd_dir = '%s/%s/%s' %( check_dir(ARCHIVE_PATH), self.cluster, host )
	333	else:
	334	rrd_dir = '%s/%s/%s/%s' %( check_dir(ARCHIVE_PATH), self.cluster, host, timeserial )
	335	if metric:
	336	rrd_file = '%s/%s.rrd' %( rrd_dir, metric['name'] )
	337	else:
	338	rrd_file = None
[17]	339
	340	return rrd_dir, rrd_file
	341
[20]	342	def getLastRrdTimeSerial( self, host ):
[17]	343
[20]	344	rrd_dir, rrd_file = self.makeRrdPath( host )
[17]	345
[19]	346	newest_timeserial = 0
	347
[17]	348	if os.path.exists( rrd_dir ):
	349	for root, dirs, files in os.walk( rrd_dir ):
	350
[20]	351	for dir in dirs:
[17]	352
[20]	353	valid_dir = 1
[17]	354
[20]	355	for letter in dir:
	356	if letter not in string.digits:
	357	valid_dir = 0
[17]	358
[20]	359	if valid_dir:
	360	timeserial = dir
[17]	361	if timeserial > newest_timeserial:
	362	newest_timeserial = timeserial
	363
	364	if newest_timeserial:
[18]	365	return newest_timeserial
[17]	366	else:
	367	return 0
	368
[20]	369	def checkNewRrdPeriod( self, host, current_timeserial ):
[17]	370
[20]	371	last_timeserial = int( self.getLastRrdTimeSerial( host ) )
[30]	372	debug_msg( 9, 'last timeserial of %s is %s' %( host, last_timeserial ) )
[17]	373
	374	if not last_timeserial:
[18]	375	serial = current_timeserial
	376	else:
[17]	377
[18]	378	archive_secs = ARCHIVE_HOURS_PER_RRD * (60 * 60)
[17]	379
[18]	380	if (current_timeserial - last_timeserial) >= archive_secs:
	381	serial = current_timeserial
	382	else:
	383	serial = last_timeserial
[17]	384
[18]	385	return serial
	386
[20]	387	def createCheck( self, host, metric, timeserial ):
[9]	388	"Check if an .rrd allready exists for this metric, create if not"
	389
[30]	390	debug_msg( 9, 'rrdcreate: using timeserial %s for %s/%s' %( timeserial, host, metric['name'] ) )
[9]	391
[17]	392	rrd_dir, rrd_file = self.makeRrdPath( host, metric, timeserial )
	393
[9]	394	if not os.path.exists( rrd_dir ):
	395	os.makedirs( rrd_dir )
[14]	396	debug_msg( 9, 'created dir %s' %( str(rrd_dir) ) )
[9]	397
[14]	398	if not os.path.exists( rrd_file ):
[9]	399
[14]	400	interval = self.gmetad_conf.getInterval( self.cluster )
	401	heartbeat = 8 * int(interval)
[9]	402
[14]	403	param_step1 = '--step'
	404	param_step2 = str( interval )
[12]	405
[14]	406	param_start1 = '--start'
	407	param_start2 = str( int( metric['time'] ) - 1 )
[12]	408
[14]	409	param_ds = 'DS:sum:GAUGE:%d:U:U' %heartbeat
	410	param_rra = 'RRA:AVERAGE:0.5:1:%s' %(ARCHIVE_HOURS_PER_RRD * 240)
[12]	411
[14]	412	rrdtool.create( str(rrd_file), param_step1, param_step2, param_start1, param_start2, param_ds, param_rra )
[13]	413
[14]	414	debug_msg( 9, 'created rrd %s' %( str(rrd_file) ) )
	415
[20]	416	def update( self, host, metric, timeserial ):
[9]	417
[30]	418	debug_msg( 9, 'rrdupdate: using timeserial %s for %s/%s' %( timeserial, host, metric['name'] ) )
[9]	419
[18]	420	rrd_dir, rrd_file = self.makeRrdPath( host, metric, timeserial )
	421
[15]	422	timestamp = metric['time']
	423	val = metric['val']
[9]	424
[15]	425	update_string = '%s:%s' %(timestamp, val)
	426
[27]	427	try:
	428	rrdtool.update( str(rrd_file), str(update_string) )
[28]	429	except rrdtool.error, detail:
[27]	430	debug_msg( 0, 'EXCEPTION! While trying to update rrd:' )
	431	debug_msg( 0, '\trrd %s with %s' %( str(rrd_file), update_string ) )
[28]	432	debug_msg( 0, str(detail) )
	433	sys.exit( 1 )
[27]	434
[15]	435	debug_msg( 9, 'updated rrd %s with %s' %( str(rrd_file), update_string ) )
	436
[8]	437	def main():
	438	"Program startup"
	439
	440	myProcessor = GangliaXMLProcessor()
	441
[22]	442	if DAEMONIZE:
	443	myProcessor.daemon()
	444	else:
	445	myProcessor.run()
	446
[9]	447	def check_dir( directory ):
	448	"Check if directory is a proper directory. I.e.: Does _not_ end with a '/'"
	449
	450	if directory[-1] == '/':
	451	directory = directory[:-1]
	452
	453	return directory
	454
[12]	455	def debug_msg( level, msg ):
	456
	457	if (DEBUG_LEVEL >= level):
	458	sys.stderr.write( msg + '\n' )
	459
[5]	460	# Let's go
[9]	461	if __name__ == '__main__':
	462	main()

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

Download in other formats: