Context Navigation

source: trunk/daemon/togad.py @ 16

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

daemon/togad.py:

Storing and updating now working for ALL hosts and metrics

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

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

Download in other formats: