Context Navigation

source: trunk/daemon/togad.py @ 30

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

daemon/togad.py:

Changed debugging levels/messages

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

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

Download in other formats: