Context Navigation

source: trunk/daemon/togad.py @ 20

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

daemon/togad.py:

Directory listing each host for each metric took _ages_ to do a update run

Rewrote timestamping style to: <cluster>/<host>/<time>/<metric>

This requires only 1 dirlisting per host, which speeds up the app considerably:

A update run now takes 7 seconds, in stead of 2 minutes

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

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

Download in other formats: