Context Navigation

source: trunk/daemon/togad.py @ 32

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

daemon/togad.py:

Begin of RRD metric handling rewrite to in-memory rrd's

File size: 11.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	#
[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
[32]	65	def __init__( self ):
	66	self.metrics = [ ]
	67	self.clusters = { }
[6]	68
	69	def startElement( self, name, attrs ):
[8]	70	"Store appropriate data from xml start tags"
[3]	71
[7]	72	if name == 'GANGLIA_XML':
[32]	73
	74	self.XMLSource = attrs.get( 'SOURCE', "" )
	75	self.gangliaVersion = attrs.get( 'VERSION', "" )
	76
[12]	77	debug_msg( 10, 'Found XML data: source %s version %s' %( self.XMLSource, self.gangliaVersion ) )
[6]	78
[7]	79	elif name == 'GRID':
[32]	80
	81	self.gridName = attrs.get( 'NAME', "" )
	82	self.time = attrs.get( 'LOCALTIME', "" )
	83
[12]	84	debug_msg( 10, '`-Grid found: %s' %( self.gridName ) )
[6]	85
[7]	86	elif name == 'CLUSTER':
[32]	87
	88	self.clusterName = attrs.get( 'NAME', "" )
	89	self.time = attrs.get( 'LOCALTIME', "" )
	90
	91	self.clusters[ self.clusterName ] = RRDHandler( self.clusterName )
	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
[32]	114	self.clusters[ self.clusterName ].memMetric( self.hostname, myMetric )
[3]	115
[32]	116	debug_msg( 11, ' \| \| \|-metric: %s:%s' %( myMetric['name'], myMetric['val'] ) )
[6]	117
[18]	118	def storeMetrics( self, hostname, timeserial ):
[9]	119
	120	for metric in self.metrics:
[17]	121	if metric['type'] not in UNSUPPORTED_ARCHIVE_TYPES:
[16]	122
[18]	123	self.rrd.createCheck( hostname, metric, timeserial )
	124	self.rrd.update( hostname, metric, timeserial )
[16]	125	debug_msg( 9, 'stored metric %s for %s: %s' %( hostname, metric['name'], metric['val'] ) )
[19]	126	#sys.exit(1)
[9]	127
[5]	128	class GangliaXMLGatherer:
[8]	129	"Setup a connection and file object to Ganglia's XML"
[3]	130
[8]	131	s = None
	132
	133	def __init__( self, host, port ):
	134	"Store host and port for connection"
	135
[5]	136	self.host = host
	137	self.port = port
[3]	138
[8]	139	def __del__( self ):
	140	"Kill the socket before we leave"
	141
	142	self.s.close()
	143
	144	def getFileObject( self ):
	145	"Connect, and return a file object"
	146
	147	for res in socket.getaddrinfo( self.host, self.port, socket.AF_UNSPEC, socket.SOCK_STREAM ):
[32]	148
[5]	149	af, socktype, proto, canonname, sa = res
[32]	150
[5]	151	try:
[32]	152
[8]	153	self.s = socket.socket( af, socktype, proto )
[32]	154
[5]	155	except socket.error, msg:
[32]	156
[8]	157	self.s = None
[5]	158	continue
[32]	159
[5]	160	try:
[32]	161
[8]	162	self.s.connect( sa )
[32]	163
[5]	164	except socket.error, msg:
[32]	165
[8]	166	self.s.close()
	167	self.s = None
[5]	168	continue
[32]	169
[5]	170	break
[3]	171
[8]	172	if self.s is None:
[32]	173
[8]	174	print 'Could not open socket'
[5]	175	sys.exit(1)
	176
[8]	177	return self.s.makefile( 'r' )
[5]	178
[8]	179	class GangliaXMLProcessor:
[5]	180
[9]	181	def daemon( self ):
[8]	182	"Run as daemon forever"
[5]	183
[8]	184	self.DAEMON = 1
[5]	185
[8]	186	# Fork the first child
	187	#
	188	pid = os.fork()
[32]	189
[8]	190	if pid > 0:
[7]	191
[32]	192	sys.exit(0) # end parent
	193
[8]	194	# creates a session and sets the process group ID
	195	#
	196	os.setsid()
[7]	197
[8]	198	# Fork the second child
	199	#
	200	pid = os.fork()
[32]	201
[8]	202	if pid > 0:
[5]	203
[32]	204	sys.exit(0) # end parent
	205
[8]	206	# Go to the root directory and set the umask
	207	#
	208	os.chdir('/')
	209	os.umask(0)
	210
	211	sys.stdin.close()
	212	sys.stdout.close()
[21]	213	sys.stderr.close()
[8]	214
	215	os.open('/dev/null', 0)
	216	os.dup(0)
	217	os.dup(0)
	218
	219	self.run()
	220
[22]	221	def printTime( self ):
	222
	223	return time.strftime("%a, %d %b %Y %H:%M:%S")
	224
[9]	225	def run( self ):
[8]	226	"Main thread"
	227
[32]	228	while( 1 ):
[8]	229
[22]	230	debug_msg( 7, self.printTime() + ' - mainthread() - xmlthread() started' )
	231	pid = os.fork()
	232
	233	if pid == 0:
	234	# Child - XML Thread
	235	#
	236	# Process XML and exit
	237
	238	debug_msg( 7, self.printTime() + ' - xmlthread() - Start XML processing..' )
	239	self.processXML()
	240	debug_msg( 7, self.printTime() + ' - xmlthread() - Done processing; exiting.' )
[28]	241	sys.exit( 0 )
[22]	242
	243	elif pid > 0:
	244	# Parent - Daemon Thread
	245
	246	debug_msg( 7, self.printTime() + ' - mainthread() - Sleep '+ str(GRAB_INTERVAL) +'s: zzzzz..' )
	247	time.sleep( GRAB_INTERVAL )
	248	debug_msg( 7, self.printTime() + ' - mainthread() - Awoken: waiting for XML thread..' )
	249
	250	r = os.wait()
[28]	251	ret = r[1]
	252	if ret != 0:
	253	debug_msg( 7, self.printTime() + ' - mainthread() - Done waiting: ERROR! xmlthread() exited with status %d' %(ret) )
[29]	254	if DEBUG_LEVEL>=7: sys.exit( 1 )
[28]	255	else:
[22]	256
[28]	257	debug_msg( 7, self.printTime() + ' - mainthread() - Done waiting: xmlthread() finished succesfully' )
[22]	258
[8]	259	def processXML( self ):
	260	"Process XML"
	261
	262	myXMLGatherer = GangliaXMLGatherer( 'localhost', 8651 )
	263
	264	myParser = make_parser()
	265	myHandler = GangliaXMLHandler()
	266	myParser.setContentHandler( myHandler )
	267
	268	myParser.parse( myXMLGatherer.getFileObject() )
	269
[9]	270	class GangliaConfigParser:
	271
	272	sources = [ ]
	273
	274	def __init__( self, config ):
[32]	275
[9]	276	self.config = config
	277	self.parseValues()
	278
[32]	279	def parseValues( self ):
[9]	280	"Parse certain values from gmetad.conf"
	281
	282	readcfg = open( self.config, 'r' )
	283
	284	for line in readcfg.readlines():
	285
	286	if line.count( '"' ) > 1:
	287
[10]	288	if line.find( 'data_source' ) != -1 and line[0] != '#':
[9]	289
[11]	290	source = { }
	291	source['name'] = line.split( '"' )[1]
[9]	292	source_words = line.split( '"' )[2].split( ' ' )
	293
	294	for word in source_words:
	295
	296	valid_interval = 1
	297
	298	for letter in word:
[32]	299
[9]	300	if letter not in string.digits:
[32]	301
[9]	302	valid_interval = 0
	303
[10]	304	if valid_interval and len(word) > 0:
[32]	305
[9]	306	source['interval'] = word
[12]	307	debug_msg( 9, 'polling interval for %s = %s' %(source['name'], source['interval'] ) )
[9]	308
	309	# No interval found, use Ganglia's default
	310	if not source.has_key( 'interval' ):
[32]	311
[9]	312	source['interval'] = 15
[12]	313	debug_msg( 9, 'polling interval for %s defaulted to 15' %(source['name']) )
[9]	314
	315	self.sources.append( source )
	316
	317	def getInterval( self, source_name ):
[32]	318
[9]	319	for source in self.sources:
[32]	320
[12]	321	if source['name'] == source_name:
[32]	322
[9]	323	return source['interval']
[32]	324
[9]	325	return None
	326
	327	class RRDHandler:
	328
[32]	329	myMetrics = { }
	330
[9]	331	def __init__( self, cluster ):
	332	self.cluster = cluster
	333	self.gmetad_conf = GangliaConfigParser( GMETAD_CONF )
	334
[32]	335	def getClusterName( self ):
	336	return self.cluster
	337
	338	def memMetric( self, host, metric ):
	339
	340	for m in self.myMetrics[ host ]:
	341
	342	if m['time'] == metric['time']:
	343
	344	# Allready have this metric, abort
	345	return 1
	346
	347	if not self.myMetrics.has_key( host ):
	348
	349	self.myMetrics[ host ] = { }
	350
	351	if not self.myMetrics[ host ].has_key( metric['name'] ):
	352
	353	self.myMetrics[ host ][ metric['name'] ] = [ ]
	354
	355	self.myMetrics[ host ][ metric['name'] ].append( metric )
	356
	357	def makeUpdateString( self, host, metric ):
	358
	359	update_string = ''
	360
	361	for m in self.myMetrics[ host ][ metric['name'] ]:
	362
	363	update_string = update_string + ' %s:%s' %( metric['time'], metric['val'] )
	364
	365	return update_string
	366
[17]	367	def makeTimeSerial( self ):
[32]	368	"Generate a time serial. Seconds since epoch"
[17]	369
	370	# YYYYMMDDhhmmss: 20050321143411
	371	#mytime = time.strftime( "%Y%m%d%H%M%S" )
	372
	373	# Seconds since epoch
	374	mytime = int( time.time() )
	375
	376	return mytime
	377
[20]	378	def makeRrdPath( self, host, metric=None, timeserial=None ):
[32]	379	"""
	380	Make a RRD location/path and filename
	381	If a metric or timeserial are supplied the complete locations
	382	will be made, else just the host directory
	383	"""
[17]	384
[20]	385	if not timeserial:
	386	rrd_dir = '%s/%s/%s' %( check_dir(ARCHIVE_PATH), self.cluster, host )
	387	else:
	388	rrd_dir = '%s/%s/%s/%s' %( check_dir(ARCHIVE_PATH), self.cluster, host, timeserial )
	389	if metric:
	390	rrd_file = '%s/%s.rrd' %( rrd_dir, metric['name'] )
	391	else:
	392	rrd_file = None
[17]	393
	394	return rrd_dir, rrd_file
	395
[20]	396	def getLastRrdTimeSerial( self, host ):
[32]	397	"""
	398	Find the last timeserial (directory) for this host
	399	This is determined once every host
	400	"""
[17]	401
[20]	402	rrd_dir, rrd_file = self.makeRrdPath( host )
[17]	403
[19]	404	newest_timeserial = 0
	405
[17]	406	if os.path.exists( rrd_dir ):
[32]	407
[17]	408	for root, dirs, files in os.walk( rrd_dir ):
	409
[20]	410	for dir in dirs:
[17]	411
[20]	412	valid_dir = 1
[17]	413
[20]	414	for letter in dir:
	415	if letter not in string.digits:
	416	valid_dir = 0
[17]	417
[20]	418	if valid_dir:
	419	timeserial = dir
[17]	420	if timeserial > newest_timeserial:
	421	newest_timeserial = timeserial
	422
	423	if newest_timeserial:
[18]	424	return newest_timeserial
[17]	425	else:
	426	return 0
	427
[20]	428	def checkNewRrdPeriod( self, host, current_timeserial ):
[32]	429	"""
	430	Check if current timeserial belongs to recent time period
	431	or should become a new period (and file).
[17]	432
[32]	433	Returns the serial of the correct time period
	434	"""
	435
[20]	436	last_timeserial = int( self.getLastRrdTimeSerial( host ) )
[30]	437	debug_msg( 9, 'last timeserial of %s is %s' %( host, last_timeserial ) )
[17]	438
	439	if not last_timeserial:
[18]	440	serial = current_timeserial
	441	else:
[17]	442
[18]	443	archive_secs = ARCHIVE_HOURS_PER_RRD * (60 * 60)
[17]	444
[18]	445	if (current_timeserial - last_timeserial) >= archive_secs:
	446	serial = current_timeserial
	447	else:
	448	serial = last_timeserial
[17]	449
[18]	450	return serial
	451
[20]	452	def createCheck( self, host, metric, timeserial ):
[9]	453	"Check if an .rrd allready exists for this metric, create if not"
	454
[30]	455	debug_msg( 9, 'rrdcreate: using timeserial %s for %s/%s' %( timeserial, host, metric['name'] ) )
[9]	456
[17]	457	rrd_dir, rrd_file = self.makeRrdPath( host, metric, timeserial )
	458
[9]	459	if not os.path.exists( rrd_dir ):
	460	os.makedirs( rrd_dir )
[14]	461	debug_msg( 9, 'created dir %s' %( str(rrd_dir) ) )
[9]	462
[14]	463	if not os.path.exists( rrd_file ):
[9]	464
[14]	465	interval = self.gmetad_conf.getInterval( self.cluster )
	466	heartbeat = 8 * int(interval)
[9]	467
[14]	468	param_step1 = '--step'
	469	param_step2 = str( interval )
[12]	470
[14]	471	param_start1 = '--start'
	472	param_start2 = str( int( metric['time'] ) - 1 )
[12]	473
[14]	474	param_ds = 'DS:sum:GAUGE:%d:U:U' %heartbeat
	475	param_rra = 'RRA:AVERAGE:0.5:1:%s' %(ARCHIVE_HOURS_PER_RRD * 240)
[12]	476
[14]	477	rrdtool.create( str(rrd_file), param_step1, param_step2, param_start1, param_start2, param_ds, param_rra )
[13]	478
[14]	479	debug_msg( 9, 'created rrd %s' %( str(rrd_file) ) )
	480
[20]	481	def update( self, host, metric, timeserial ):
[9]	482
[30]	483	debug_msg( 9, 'rrdupdate: using timeserial %s for %s/%s' %( timeserial, host, metric['name'] ) )
[9]	484
[18]	485	rrd_dir, rrd_file = self.makeRrdPath( host, metric, timeserial )
	486
[15]	487	timestamp = metric['time']
	488	val = metric['val']
[9]	489
[15]	490	update_string = '%s:%s' %(timestamp, val)
	491
[27]	492	try:
[32]	493
[27]	494	rrdtool.update( str(rrd_file), str(update_string) )
[32]	495
[28]	496	except rrdtool.error, detail:
[32]	497
[27]	498	debug_msg( 0, 'EXCEPTION! While trying to update rrd:' )
	499	debug_msg( 0, '\trrd %s with %s' %( str(rrd_file), update_string ) )
[28]	500	debug_msg( 0, str(detail) )
[32]	501
[28]	502	sys.exit( 1 )
[27]	503
[15]	504	debug_msg( 9, 'updated rrd %s with %s' %( str(rrd_file), update_string ) )
	505
[8]	506	def main():
	507	"Program startup"
	508
	509	myProcessor = GangliaXMLProcessor()
	510
[22]	511	if DAEMONIZE:
	512	myProcessor.daemon()
	513	else:
	514	myProcessor.run()
	515
[9]	516	def check_dir( directory ):
	517	"Check if directory is a proper directory. I.e.: Does _not_ end with a '/'"
	518
	519	if directory[-1] == '/':
	520	directory = directory[:-1]
	521
	522	return directory
	523
[12]	524	def debug_msg( level, msg ):
	525
	526	if (DEBUG_LEVEL >= level):
	527	sys.stderr.write( msg + '\n' )
	528
[5]	529	# Let's go
[9]	530	if __name__ == '__main__':
	531	main()

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

Download in other formats: