Context Navigation

source: trunk/daemon/togad.py @ 21

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

daemon/togad.py:

Minor adjustments for running daemon'ized

Will now let daemon run a day for testing

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

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

Download in other formats: