Context Navigation

source: trunk/daemon/togad.py @ 27

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

daemon/togad.py:

Added try/except, to catch a bug that occures from time to time

File size: 10.5 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	#
[19]	21	DEBUG_LEVEL = 7
[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.' )
	231	sys.exit(0)
	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()
	241
	242	debug_msg( 7, self.printTime() + ' - mainthread() - Done waiting.' )
	243
[8]	244	def processXML( self ):
	245	"Process XML"
	246
	247	myXMLGatherer = GangliaXMLGatherer( 'localhost', 8651 )
	248
	249	myParser = make_parser()
	250	myHandler = GangliaXMLHandler()
	251	myParser.setContentHandler( myHandler )
	252
	253	myParser.parse( myXMLGatherer.getFileObject() )
	254
[9]	255	class GangliaConfigParser:
	256
	257	sources = [ ]
	258
	259	def __init__( self, config ):
	260	self.config = config
	261	self.parseValues()
	262
	263	def parseValues(self):
	264	"Parse certain values from gmetad.conf"
	265
	266	readcfg = open( self.config, 'r' )
	267
	268	for line in readcfg.readlines():
	269
	270	if line.count( '"' ) > 1:
	271
[10]	272	if line.find( 'data_source' ) != -1 and line[0] != '#':
[9]	273
[11]	274	source = { }
	275	source['name'] = line.split( '"' )[1]
[9]	276	source_words = line.split( '"' )[2].split( ' ' )
	277
	278	for word in source_words:
	279
	280	valid_interval = 1
	281
	282	for letter in word:
	283	if letter not in string.digits:
	284	valid_interval = 0
	285
[10]	286	if valid_interval and len(word) > 0:
[9]	287	source['interval'] = word
[12]	288	debug_msg( 9, 'polling interval for %s = %s' %(source['name'], source['interval'] ) )
[9]	289
	290	# No interval found, use Ganglia's default
	291	if not source.has_key( 'interval' ):
	292	source['interval'] = 15
[12]	293	debug_msg( 9, 'polling interval for %s defaulted to 15' %(source['name']) )
[9]	294
	295	self.sources.append( source )
	296
	297	def getInterval( self, source_name ):
	298	for source in self.sources:
[12]	299	if source['name'] == source_name:
[9]	300	return source['interval']
	301	return None
	302
	303	class RRDHandler:
	304
	305	def __init__( self, cluster ):
	306	self.cluster = cluster
	307	self.gmetad_conf = GangliaConfigParser( GMETAD_CONF )
	308
[17]	309	def makeTimeSerial( self ):
	310
	311	# YYYYMMDDhhmmss: 20050321143411
	312	#mytime = time.strftime( "%Y%m%d%H%M%S" )
	313
	314	# Seconds since epoch
	315	mytime = int( time.time() )
	316
	317	return mytime
	318
[20]	319	def makeRrdPath( self, host, metric=None, timeserial=None ):
[17]	320
[20]	321	if not timeserial:
	322	rrd_dir = '%s/%s/%s' %( check_dir(ARCHIVE_PATH), self.cluster, host )
	323	else:
	324	rrd_dir = '%s/%s/%s/%s' %( check_dir(ARCHIVE_PATH), self.cluster, host, timeserial )
	325	if metric:
	326	rrd_file = '%s/%s.rrd' %( rrd_dir, metric['name'] )
	327	else:
	328	rrd_file = None
[17]	329
	330	return rrd_dir, rrd_file
	331
[20]	332	def getLastRrdTimeSerial( self, host ):
[17]	333
[20]	334	rrd_dir, rrd_file = self.makeRrdPath( host )
[17]	335
[19]	336	newest_timeserial = 0
	337
[17]	338	if os.path.exists( rrd_dir ):
	339	for root, dirs, files in os.walk( rrd_dir ):
	340
[20]	341	for dir in dirs:
[17]	342
[20]	343	valid_dir = 1
[17]	344
[20]	345	for letter in dir:
	346	if letter not in string.digits:
	347	valid_dir = 0
[17]	348
[20]	349	if valid_dir:
	350	timeserial = dir
[17]	351	if timeserial > newest_timeserial:
	352	newest_timeserial = timeserial
	353
	354	if newest_timeserial:
[18]	355	return newest_timeserial
[17]	356	else:
	357	return 0
	358
[20]	359	def checkNewRrdPeriod( self, host, current_timeserial ):
[17]	360
[20]	361	last_timeserial = int( self.getLastRrdTimeSerial( host ) )
	362	debug_msg( 8, 'last timeserial of %s is %s' %( host, last_timeserial ) )
[17]	363
	364	if not last_timeserial:
[18]	365	serial = current_timeserial
	366	else:
[17]	367
[18]	368	archive_secs = ARCHIVE_HOURS_PER_RRD * (60 * 60)
[17]	369
[18]	370	if (current_timeserial - last_timeserial) >= archive_secs:
	371	serial = current_timeserial
	372	else:
	373	serial = last_timeserial
[17]	374
[18]	375	return serial
	376
[20]	377	def createCheck( self, host, metric, timeserial ):
[9]	378	"Check if an .rrd allready exists for this metric, create if not"
	379
[18]	380	debug_msg( 8, 'rrdcreate: using timeserial %s for %s/%s' %( timeserial, host, metric['name'] ) )
[9]	381
[17]	382	rrd_dir, rrd_file = self.makeRrdPath( host, metric, timeserial )
	383
[9]	384	if not os.path.exists( rrd_dir ):
	385	os.makedirs( rrd_dir )
[14]	386	debug_msg( 9, 'created dir %s' %( str(rrd_dir) ) )
[9]	387
[14]	388	if not os.path.exists( rrd_file ):
[9]	389
[14]	390	interval = self.gmetad_conf.getInterval( self.cluster )
	391	heartbeat = 8 * int(interval)
[9]	392
[14]	393	param_step1 = '--step'
	394	param_step2 = str( interval )
[12]	395
[14]	396	param_start1 = '--start'
	397	param_start2 = str( int( metric['time'] ) - 1 )
[12]	398
[14]	399	param_ds = 'DS:sum:GAUGE:%d:U:U' %heartbeat
	400	param_rra = 'RRA:AVERAGE:0.5:1:%s' %(ARCHIVE_HOURS_PER_RRD * 240)
[12]	401
[14]	402	rrdtool.create( str(rrd_file), param_step1, param_step2, param_start1, param_start2, param_ds, param_rra )
[13]	403
[14]	404	debug_msg( 9, 'created rrd %s' %( str(rrd_file) ) )
	405
[20]	406	def update( self, host, metric, timeserial ):
[9]	407
[18]	408	debug_msg( 8, 'rrdupdate: using timeserial %s for %s/%s' %( timeserial, host, metric['name'] ) )
[9]	409
[18]	410	rrd_dir, rrd_file = self.makeRrdPath( host, metric, timeserial )
	411
[15]	412	timestamp = metric['time']
	413	val = metric['val']
[9]	414
[15]	415	update_string = '%s:%s' %(timestamp, val)
	416
[27]	417	try:
	418	rrdtool.update( str(rrd_file), str(update_string) )
	419	except error, detail:
	420	debug_msg( 0, 'EXCEPTION! While trying to update rrd:' )
	421	debug_msg( 0, '\trrd %s with %s' %( str(rrd_file), update_string ) )
	422	debug_msg( 0, detail )
	423
[15]	424	debug_msg( 9, 'updated rrd %s with %s' %( str(rrd_file), update_string ) )
	425
[8]	426	def main():
	427	"Program startup"
	428
	429	myProcessor = GangliaXMLProcessor()
	430
[22]	431	if DAEMONIZE:
	432	myProcessor.daemon()
	433	else:
	434	myProcessor.run()
	435
[9]	436	def check_dir( directory ):
	437	"Check if directory is a proper directory. I.e.: Does _not_ end with a '/'"
	438
	439	if directory[-1] == '/':
	440	directory = directory[:-1]
	441
	442	return directory
	443
[12]	444	def debug_msg( level, msg ):
	445
	446	if (DEBUG_LEVEL >= level):
	447	sys.stderr.write( msg + '\n' )
	448
[5]	449	# Let's go
[9]	450	if __name__ == '__main__':
	451	main()

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

Download in other formats: