Context Navigation

source: trunk/daemon/togad.py @ 28

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

daemon/togad.py:

Additional code for bug catching

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

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

Download in other formats: