Context Navigation

source: trunk/daemon/togad.py @ 136

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

daemon/togad.py:

Typo really fixed now?

File size: 31.8 KB

Rev	Line
[3]	1	#!/usr/bin/env python
	2
[70]	3	import xml.sax
[71]	4	import xml.sax.handler
[5]	5	import socket
	6	import sys
[9]	7	import string
[12]	8	import os
	9	import os.path
[17]	10	import time
[36]	11	import threading
[38]	12	import random
[40]	13	from types import *
[60]	14	import DBClass
[3]	15
[8]	16	# Specify debugging level here;
	17	#
[38]	18	# 11 = XML: metrics
	19	# 10 = XML: host, cluster, grid, ganglia
	20	# 9 = RRD activity, gmetad config parsing
[40]	21	# 8 = RRD file activity
[84]	22	# 6 = SQL
[87]	23	# 1 = daemon threading
[8]	24	#
[102]	25	DEBUG_LEVEL = 1
[6]	26
[9]	27	# Where is the gmetad.conf located
	28	#
	29	GMETAD_CONF = '/etc/gmetad.conf'
	30
[60]	31	# Where to grab XML data from
	32	# Normally: local gmetad (port 8651)
	33	#
[63]	34	# Syntax: <hostname>:<port>
	35	#
	36	ARCHIVE_XMLSOURCE = "localhost:8651"
[60]	37
[9]	38	# List of data_source names to archive for
	39	#
[63]	40	# Syntax: [ "<clustername>", "<clustername>" ]
	41	#
[60]	42	ARCHIVE_DATASOURCES = [ "LISA Cluster" ]
[9]	43
[60]	44	# Where to store the archived rrd's
	45	#
	46	ARCHIVE_PATH = '/data/toga/rrds'
	47
[13]	48	# Amount of hours to store in one single archived .rrd
[9]	49	#
[43]	50	ARCHIVE_HOURS_PER_RRD = 12
[13]	51
[63]	52	# Toga's SQL dbase to use
[60]	53	#
[63]	54	# Syntax: <hostname>/<database>
[60]	55	#
[63]	56	TOGA_SQL_DBASE = "localhost/toga"
[60]	57
[22]	58	# Wether or not to run as a daemon in background
	59	#
	60	DAEMONIZE = 0
	61
[17]	62	######################
	63	# #
	64	# Configuration ends #
	65	# #
	66	######################
[13]	67
[60]	68	###
	69	# You'll only want to change anything below here unless you
[63]	70	# know what you are doing (i.e. your name is Ramon Bastiaans :D )
[60]	71	###
	72
[17]	73	# What XML data types not to store
[13]	74	#
[17]	75	UNSUPPORTED_ARCHIVE_TYPES = [ 'string' ]
[9]	76
[47]	77	# Maximum time (in seconds) a parsethread may run
	78	#
	79	PARSE_TIMEOUT = 60
	80
	81	# Maximum time (in seconds) a storethread may run
	82	#
	83	STORE_TIMEOUT = 360
	84
[8]	85	"""
	86	This is TOrque-GAnglia's data Daemon
	87	"""
	88
[84]	89	class DataSQLStore:
	90
	91	db_vars = None
	92	dbc = None
	93
	94	def __init__( self, hostname, database ):
	95
	96	self.db_vars = DBClass.InitVars(DataBaseName=database,
	97	User='root',
	98	Host=hostname,
	99	Password='',
	100	Dictionary='true')
	101
	102	try:
	103	self.dbc = DBClass.DB(self.db_vars)
	104	except DBClass.DBError, details:
	105	print 'Error in connection to db: %s' %details
	106	sys.exit(1)
	107
	108	def setDatabase(self, statement):
	109	ret = self.doDatabase('set', statement)
	110	return ret
	111
	112	def getDatabase(self, statement):
	113	ret = self.doDatabase('get', statement)
	114	return ret
	115
	116	def doDatabase(self, type, statement):
	117
	118	debug_msg( 6, 'doDatabase(): %s: %s' %(type, statement) )
	119	try:
	120	if type == 'set':
	121	result = self.dbc.Set( statement )
	122	self.dbc.Commit()
	123	elif type == 'get':
	124	result = self.dbc.Get( statement )
	125
	126	except DBClass.DBError, detail:
	127	operation = statement.split(' ')[0]
	128	print "%s operation on database failed while performing\n'%s'\n%s"\
	129	%(operation, statement, detail)
	130	sys.exit(1)
	131
	132	debug_msg( 6, 'doDatabase(): result: %s' %(result) )
	133	return result
	134
	135	def getNodeId( self, hostname ):
	136
[89]	137	id = self.getDatabase( "SELECT node_id FROM nodes WHERE node_hostname = '%s'" %hostname )
[84]	138
[89]	139	if len( id ) > 0:
[84]	140
[89]	141	id = id[0][0]
	142
[84]	143	return id
	144	else:
	145	return None
	146
	147	def getNodeIds( self, hostnames ):
	148
	149	ids = [ ]
	150
	151	for node in hostnames:
	152
	153	id = self.getNodeId( node )
	154
	155	if id:
	156	ids.append( id )
	157
	158	return ids
	159
	160	def getJobId( self, jobid ):
	161
	162	id = self.getDatabase( "SELECT job_id FROM jobs WHERE job_id = '%s'" %jobid )
	163
	164	if id:
[89]	165	id = id[0][0]
[84]	166
	167	return id
	168	else:
	169	return None
	170
	171	def addJob( self, job_id, jobattrs ):
	172
	173	if not self.getJobId( job_id ):
	174
	175	self.mutateJob( 'insert', job_id, jobattrs )
	176	else:
	177	self.mutateJob( 'update', job_id, jobattrs )
	178
	179	def mutateJob( self, action, job_id, jobattrs ):
	180
	181	job_values = [ 'name', 'queue', 'owner', 'requested_time', 'requested_memory', 'ppn', 'status', 'start_timestamp', 'stop_timestamp' ]
	182
	183	insert_col_str = 'job_id'
	184	insert_val_str = "'%s'" %job_id
	185	update_str = None
	186
	187	debug_msg( 6, 'mutateJob(): %s %s' %(action,job_id))
	188
[99]	189	ids = [ ]
	190
[84]	191	for valname, value in jobattrs.items():
	192
[96]	193	if valname in job_values and value != '':
[84]	194
	195	column_name = 'job_' + valname
	196
	197	if action == 'insert':
	198
	199	if not insert_col_str:
	200	insert_col_str = column_name
	201	else:
	202	insert_col_str = insert_col_str + ',' + column_name
	203
	204	if not insert_val_str:
	205	insert_val_str = value
	206	else:
	207	insert_val_str = insert_val_str + ",'%s'" %value
	208
	209	elif action == 'update':
	210
	211	if not update_str:
	212	update_str = "%s='%s'" %(column_name, value)
	213	else:
	214	update_str = update_str + ",%s='%s'" %(column_name, value)
	215
[90]	216	elif valname == 'nodes' and value:
[84]	217
[98]	218	ids = self.addNodes( value )
[86]	219	node_list = value
[84]	220
	221	if action == 'insert':
	222
	223	self.setDatabase( "INSERT INTO jobs ( %s ) VALUES ( %s )" %( insert_col_str, insert_val_str ) )
[86]	224
[99]	225	if len( ids ) > 0:
	226	self.addJobNodes( job_id, ids )
	227
[84]	228	elif action == 'update':
	229
[89]	230	self.setDatabase( "UPDATE jobs SET %s WHERE job_id=%s" %(update_str, job_id) )
[84]	231
	232	def addNodes( self, hostnames ):
	233
[98]	234	ids = [ ]
	235
[84]	236	for node in hostnames:
	237
	238	id = self.getNodeId( node )
	239
	240	if not id:
	241	self.setDatabase( "INSERT INTO nodes ( node_hostname ) VALUES ( '%s' )" %node )
[98]	242	id = self.getNodeId( node )
[84]	243
[98]	244	ids.append( id )
	245
	246	return ids
	247
[86]	248	def addJobNodes( self, jobid, nodes ):
	249
	250	for node in nodes:
	251	self.addJobNode( jobid, node )
	252
[84]	253	def addJobNode( self, jobid, nodeid ):
	254
	255	self.setDatabase( "INSERT INTO job_nodes (job_id,node_id) VALUES ( %s,%s )" %(jobid, nodeid) )
	256
	257	def storeJobInfo( self, jobid, jobattrs ):
	258
	259	self.addJob( jobid, jobattrs )
	260
[37]	261	class RRDMutator:
[63]	262	"""A class for performing RRD mutations"""
[37]	263
	264	binary = '/usr/bin/rrdtool'
	265
	266	def __init__( self, binary=None ):
[63]	267	"""Set alternate binary if supplied"""
[37]	268
	269	if binary:
	270	self.binary = binary
	271
	272	def create( self, filename, args ):
[63]	273	"""Create a new rrd with args"""
	274
[40]	275	return self.perform( 'create', '"' + filename + '"', args )
[37]	276
	277	def update( self, filename, args ):
[63]	278	"""Update a rrd with args"""
	279
[40]	280	return self.perform( 'update', '"' + filename + '"', args )
[37]	281
[42]	282	def grabLastUpdate( self, filename ):
[63]	283	"""Determine the last update time of filename rrd"""
[42]	284
	285	last_update = 0
	286
[53]	287	debug_msg( 8, self.binary + ' info "' + filename + '"' )
	288
[42]	289	for line in os.popen( self.binary + ' info "' + filename + '"' ).readlines():
	290
	291	if line.find( 'last_update') != -1:
	292
	293	last_update = line.split( ' = ' )[1]
	294
	295	if last_update:
	296	return last_update
	297	else:
	298	return 0
	299
[40]	300	def perform( self, action, filename, args ):
[63]	301	"""Perform action on rrd filename with args"""
[37]	302
	303	arg_string = None
	304
[40]	305	if type( args ) is not ListType:
	306	debug_msg( 8, 'Arguments needs to be of type List' )
	307	return 1
	308
[37]	309	for arg in args:
	310
	311	if not arg_string:
	312
	313	arg_string = arg
	314	else:
	315	arg_string = arg_string + ' ' + arg
	316
[54]	317	debug_msg( 8, self.binary + ' ' + action + ' ' + filename + ' ' + arg_string )
[37]	318
[40]	319	for line in os.popen( self.binary + ' ' + action + ' ' + filename + ' ' + arg_string ).readlines():
[37]	320
	321	if line.find( 'ERROR' ) != -1:
	322
	323	error_msg = string.join( line.split( ' ' )[1:] )
[40]	324	debug_msg( 8, error_msg )
[37]	325	return 1
	326
	327	return 0
	328
[78]	329	class XMLProcessor:
	330	"""Skeleton class for XML processor's"""
	331
	332	def daemon( self ):
	333	"""Run as daemon forever"""
	334
	335	# Fork the first child
	336	#
	337	pid = os.fork()
	338
	339	if pid > 0:
	340
	341	sys.exit(0) # end parent
	342
	343	# creates a session and sets the process group ID
	344	#
	345	os.setsid()
	346
	347	# Fork the second child
	348	#
	349	pid = os.fork()
	350
	351	if pid > 0:
	352
	353	sys.exit(0) # end parent
	354
	355	# Go to the root directory and set the umask
	356	#
	357	os.chdir('/')
	358	os.umask(0)
	359
	360	sys.stdin.close()
	361	sys.stdout.close()
[81]	362	sys.stderr.close()
[78]	363
	364	os.open('/dev/null', 0)
	365	os.dup(0)
	366	os.dup(0)
	367
	368	self.run()
	369
	370	def run( self ):
	371	"""Do main processing of XML here"""
	372
	373	pass
	374
	375	class TorqueXMLProcessor( XMLProcessor ):
	376	"""Main class for processing XML and acting with it"""
	377
	378	def __init__( self ):
	379	"""Setup initial XML connection and handlers"""
	380
	381	self.myXMLGatherer = XMLGatherer( ARCHIVE_XMLSOURCE.split( ':' )[0], ARCHIVE_XMLSOURCE.split( ':' )[1] )
	382	self.myXMLSource = self.myXMLGatherer.getFileObject()
	383	self.myXMLHandler = TorqueXMLHandler()
	384	self.myXMLError = XMLErrorHandler()
[87]	385	self.config = GangliaConfigParser( GMETAD_CONF )
[78]	386
	387	def run( self ):
	388	"""Main XML processing"""
	389
[102]	390	debug_msg( 1, printTime() + ' - torque_xml_thread(): started.' )
[87]	391
[78]	392	while( 1 ):
	393
	394	self.myXMLSource = self.myXMLGatherer.getFileObject()
[102]	395	debug_msg( 1, printTime() + ' - torque_xml_thread(): Parsing..' )
[78]	396	xml.sax.parse( self.myXMLSource, self.myXMLHandler, self.myXMLError )
[102]	397	debug_msg( 1, printTime() + ' - torque_xml_thread(): Done parsing.' )
	398	debug_msg( 1, printTime() + ' - torque_xml_thread(): Sleeping.. (%ss)' %(str( self.config.getLowestInterval() ) ) )
[87]	399	time.sleep( self.config.getLowestInterval() )
[78]	400
[71]	401	class TorqueXMLHandler( xml.sax.handler.ContentHandler ):
[63]	402	"""Parse Torque's jobinfo XML from our plugin"""
	403
[72]	404	jobAttrs = { }
	405
[84]	406	def __init__( self ):
	407
	408	self.ds = DataSQLStore( TOGA_SQL_DBASE.split( '/' )[0], TOGA_SQL_DBASE.split( '/' )[1] )
[98]	409	self.jobs_processed = [ ]
	410	self.jobs_to_store = [ ]
[84]	411
[63]	412	def startElement( self, name, attrs ):
	413	"""
	414	This XML will be all gmetric XML
	415	so there will be no specific start/end element
	416	just one XML statement with all info
	417	"""
	418
	419	heartbeat = 0
[73]	420
	421	jobinfo = { }
[63]	422
	423	if name == 'METRIC':
	424
[70]	425	metricname = attrs.get( 'NAME', "" )
[63]	426
	427	if metricname == 'TOGA-HEARTBEAT':
[70]	428	self.heartbeat = attrs.get( 'VAL', "" )
[63]	429
[70]	430	elif metricname.find( 'TOGA-JOB' ) != -1:
[63]	431
[72]	432	job_id = metricname.split( 'TOGA-JOB-' )[1]
[63]	433	val = attrs.get( 'VAL', "" )
	434
[96]	435	if not job_id in self.jobs_processed:
	436	self.jobs_processed.append( job_id )
	437
[73]	438	check_change = 0
	439
	440	if self.jobAttrs.has_key( job_id ):
	441	check_change = 1
	442
[63]	443	valinfo = val.split( ' ' )
	444
	445	for myval in valinfo:
	446
[84]	447	if len( myval.split( '=' ) ) > 1:
[63]	448
[84]	449	valname = myval.split( '=' )[0]
	450	value = myval.split( '=' )[1]
[70]	451
[84]	452	if valname == 'nodes':
	453	value = value.split( ';' )
[72]	454
[84]	455	jobinfo[ valname ] = value
	456
[73]	457	if check_change:
[100]	458	if self.jobinfoChanged( self.jobAttrs, job_id, jobinfo ) and self.jobAttrs[ job_id ]['status'] != 'E':
	459	self.jobAttrs[ job_id ]['stop_timestamp'] = ''
[82]	460	self.jobAttrs[ job_id ] = self.setJobAttrs( self.jobAttrs[ job_id ], jobinfo )
[84]	461	if not job_id in self.jobs_to_store:
	462	self.jobs_to_store.append( job_id )
	463
[102]	464	debug_msg( 6, 'jobinfo for job %s has changed' %job_id )
[73]	465	else:
	466	self.jobAttrs[ job_id ] = jobinfo
[84]	467
	468	if not job_id in self.jobs_to_store:
	469	self.jobs_to_store.append( job_id )
	470
[102]	471	debug_msg( 6, 'jobinfo for job %s has changed' %job_id )
[73]	472
[77]	473	def endDocument( self ):
[74]	474	"""When all metrics have gone, check if any jobs have finished"""
[72]	475
[74]	476	for jobid, jobinfo in self.jobAttrs.items():
	477
	478	# This is an old job, not in current jobinfo list anymore
	479	# it must have finished, since we _did_ get a new heartbeat
	480	#
[96]	481	mytime = int( jobinfo['reported'] ) + int( jobinfo['poll_interval'] )
[102]	482
[96]	483	if mytime < self.heartbeat and jobid not in self.jobs_processed and jobinfo['status'] == 'R':
[74]	484
[97]	485	if not jobid in self.jobs_processed:
[96]	486	self.jobs_processed.append( jobid )
	487
[74]	488	self.jobAttrs[ jobid ]['status'] = 'F'
[81]	489	self.jobAttrs[ jobid ]['stop_timestamp'] = str( int( jobinfo['reported'] ) + int( jobinfo['poll_interval'] ) )
[96]	490
[84]	491	if not jobid in self.jobs_to_store:
	492	self.jobs_to_store.append( jobid )
[74]	493
[102]	494	debug_msg( 1, printTime() + ' - torque_xml_thread(): Storing..' )
[87]	495
[84]	496	for jobid in self.jobs_to_store:
	497	self.ds.storeJobInfo( jobid, self.jobAttrs[ jobid ] )
	498
[102]	499	debug_msg( 1, printTime() + ' - torque_xml_thread(): Done storing.' )
[87]	500
[96]	501	self.jobs_processed = [ ]
[84]	502	self.jobs_to_store = [ ]
	503
[82]	504	def setJobAttrs( self, old, new ):
	505	"""
	506	Set new job attributes in old, but not lose existing fields
	507	if old attributes doesn't have those
	508	"""
	509
	510	for valname, value in new.items():
	511	old[ valname ] = value
	512
	513	return old
	514
	515
[73]	516	def jobinfoChanged( self, jobattrs, jobid, jobinfo ):
[74]	517	"""
	518	Check if jobinfo has changed from jobattrs[jobid]
	519	if it's report time is bigger than previous one
	520	and it is report time is recent (equal to heartbeat)
	521	"""
[72]	522
[87]	523	ignore_changes = [ 'reported' ]
	524
[73]	525	if jobattrs.has_key( jobid ):
	526
	527	for valname, value in jobinfo.items():
	528
[87]	529	if valname not in ignore_changes:
[73]	530
[87]	531	if jobattrs[ jobid ].has_key( valname ):
[73]	532
[87]	533	if value != jobattrs[ jobid ][ valname ]:
[73]	534
[87]	535	if jobinfo['reported'] > jobattrs[ jobid ][ 'reported' ] and jobinfo['reported'] == self.heartbeat:
	536	return 1
[73]	537
[87]	538	else:
	539	return 1
	540
[73]	541	return 0
	542
[71]	543	class GangliaXMLHandler( xml.sax.handler.ContentHandler ):
[63]	544	"""Parse Ganglia's XML"""
[3]	545
[33]	546	def __init__( self, config ):
[63]	547	"""Setup initial variables and gather info on existing rrd archive"""
	548
[33]	549	self.config = config
[35]	550	self.clusters = { }
[54]	551	debug_msg( 0, printTime() + ' - Checking existing toga rrd archive..' )
[44]	552	self.gatherClusters()
[54]	553	debug_msg( 0, printTime() + ' - Check done.' )
[33]	554
[44]	555	def gatherClusters( self ):
[63]	556	"""Find all existing clusters in archive dir"""
[44]	557
[45]	558	archive_dir = check_dir(ARCHIVE_PATH)
[44]	559
	560	hosts = [ ]
	561
	562	if os.path.exists( archive_dir ):
	563
	564	dirlist = os.listdir( archive_dir )
	565
	566	for item in dirlist:
	567
	568	clustername = item
	569
[60]	570	if not self.clusters.has_key( clustername ) and clustername in ARCHIVE_DATASOURCES:
[44]	571
	572	self.clusters[ clustername ] = RRDHandler( self.config, clustername )
	573
[6]	574	def startElement( self, name, attrs ):
[63]	575	"""Memorize appropriate data from xml start tags"""
[3]	576
[7]	577	if name == 'GANGLIA_XML':
[32]	578
	579	self.XMLSource = attrs.get( 'SOURCE', "" )
	580	self.gangliaVersion = attrs.get( 'VERSION', "" )
	581
[12]	582	debug_msg( 10, 'Found XML data: source %s version %s' %( self.XMLSource, self.gangliaVersion ) )
[6]	583
[7]	584	elif name == 'GRID':
[32]	585
	586	self.gridName = attrs.get( 'NAME', "" )
	587	self.time = attrs.get( 'LOCALTIME', "" )
	588
[12]	589	debug_msg( 10, '`-Grid found: %s' %( self.gridName ) )
[6]	590
[7]	591	elif name == 'CLUSTER':
[32]	592
	593	self.clusterName = attrs.get( 'NAME', "" )
	594	self.time = attrs.get( 'LOCALTIME', "" )
	595
[60]	596	if not self.clusters.has_key( self.clusterName ) and self.clusterName in ARCHIVE_DATASOURCES:
[32]	597
[34]	598	self.clusters[ self.clusterName ] = RRDHandler( self.config, self.clusterName )
[33]	599
[35]	600	debug_msg( 10, ' \|-Cluster found: %s' %( self.clusterName ) )
[6]	601
[60]	602	elif name == 'HOST' and self.clusterName in ARCHIVE_DATASOURCES:
[32]	603
	604	self.hostName = attrs.get( 'NAME', "" )
	605	self.hostIp = attrs.get( 'IP', "" )
	606	self.hostReported = attrs.get( 'REPORTED', "" )
	607
[12]	608	debug_msg( 10, ' \| \|-Host found: %s - ip %s reported %s' %( self.hostName, self.hostIp, self.hostReported ) )
[6]	609
[60]	610	elif name == 'METRIC' and self.clusterName in ARCHIVE_DATASOURCES:
[6]	611
[32]	612	type = attrs.get( 'TYPE', "" )
[6]	613
[32]	614	if type not in UNSUPPORTED_ARCHIVE_TYPES:
[3]	615
[32]	616	myMetric = { }
	617	myMetric['name'] = attrs.get( 'NAME', "" )
	618	myMetric['val'] = attrs.get( 'VAL', "" )
	619	myMetric['time'] = self.hostReported
[3]	620
[34]	621	self.clusters[ self.clusterName ].memMetric( self.hostName, myMetric )
[3]	622
[34]	623	debug_msg( 11, ' \| \| \|-metric: %s:%s' %( myMetric['name'], myMetric['val'] ) )
[6]	624
[34]	625	def storeMetrics( self ):
[63]	626	"""Store metrics of each cluster rrd handler"""
[9]	627
[34]	628	for clustername, rrdh in self.clusters.items():
[16]	629
[38]	630	ret = rrdh.storeMetrics()
[9]	631
[38]	632	if ret:
	633	debug_msg( 9, 'An error occured while storing metrics for cluster %s' %clustername )
	634	return 1
	635
	636	return 0
	637
[71]	638	class XMLErrorHandler( xml.sax.handler.ErrorHandler ):
	639
	640	def error( self, exception ):
	641	"""Recoverable error"""
	642
	643	debug_msg( 0, 'Recoverable error ' + str( exception ) )
	644
	645	def fatalError( self, exception ):
	646	"""Non-recoverable error"""
	647
	648	exception_str = str( exception )
	649
	650	# Ignore 'no element found' errors
	651	if exception_str.find( 'no element found' ) != -1:
	652	debug_msg( 1, 'No XML data found: probably socket not (re)connected.' )
	653	return 0
	654
	655	debug_msg( 0, 'Non-recoverable error ' + str( exception ) )
	656	sys.exit( 1 )
	657
	658	def warning( self, exception ):
	659	"""Warning"""
	660
	661	debug_msg( 0, 'Warning ' + str( exception ) )
	662
[78]	663	class XMLGatherer:
[63]	664	"""Setup a connection and file object to Ganglia's XML"""
[3]	665
[8]	666	s = None
[70]	667	fd = None
[8]	668
	669	def __init__( self, host, port ):
[63]	670	"""Store host and port for connection"""
[8]	671
[5]	672	self.host = host
	673	self.port = port
[33]	674	self.connect()
[70]	675	self.makeFileDescriptor()
[3]	676
[33]	677	def connect( self ):
[63]	678	"""Setup connection to XML source"""
[8]	679
	680	for res in socket.getaddrinfo( self.host, self.port, socket.AF_UNSPEC, socket.SOCK_STREAM ):
[32]	681
[5]	682	af, socktype, proto, canonname, sa = res
[32]	683
[5]	684	try:
[32]	685
[8]	686	self.s = socket.socket( af, socktype, proto )
[32]	687
[5]	688	except socket.error, msg:
[32]	689
[8]	690	self.s = None
[5]	691	continue
[32]	692
[5]	693	try:
[32]	694
[8]	695	self.s.connect( sa )
[32]	696
[5]	697	except socket.error, msg:
[32]	698
[70]	699	self.disconnect()
[5]	700	continue
[32]	701
[5]	702	break
[3]	703
[8]	704	if self.s is None:
[32]	705
[33]	706	debug_msg( 0, 'Could not open socket' )
	707	sys.exit( 1 )
[5]	708
[33]	709	def disconnect( self ):
[63]	710	"""Close socket"""
[33]	711
	712	if self.s:
[71]	713	self.s.shutdown( 2 )
[33]	714	self.s.close()
	715	self.s = None
	716
	717	def __del__( self ):
[63]	718	"""Kill the socket before we leave"""
[33]	719
	720	self.disconnect()
	721
[70]	722	def reconnect( self ):
	723	"""Reconnect"""
[33]	724
[38]	725	if self.s:
	726	self.disconnect()
[33]	727
[70]	728	self.connect()
[5]	729
[70]	730	def makeFileDescriptor( self ):
	731	"""Make file descriptor that points to our socket connection"""
	732
	733	self.reconnect()
	734
	735	if self.s:
	736	self.fd = self.s.makefile( 'r' )
	737
	738	def getFileObject( self ):
	739	"""Connect, and return a file object"""
	740
[78]	741	self.makeFileDescriptor()
	742
[70]	743	if self.fd:
	744	return self.fd
	745
[78]	746	class GangliaXMLProcessor( XMLProcessor ):
[63]	747	"""Main class for processing XML and acting with it"""
[5]	748
[33]	749	def __init__( self ):
[63]	750	"""Setup initial XML connection and handlers"""
[33]	751
	752	self.config = GangliaConfigParser( GMETAD_CONF )
	753
[78]	754	self.myXMLGatherer = XMLGatherer( ARCHIVE_XMLSOURCE.split( ':' )[0], ARCHIVE_XMLSOURCE.split( ':' )[1] )
[70]	755	self.myXMLSource = self.myXMLGatherer.getFileObject()
[78]	756	self.myXMLHandler = GangliaXMLHandler( self.config )
	757	self.myXMLError = XMLErrorHandler()
[73]	758
[9]	759	def run( self ):
[63]	760	"""Main XML processing; start a xml and storethread"""
[8]	761
[102]	762	xml_thread = threading.Thread( None, self.processXML, 'xmlthread' )
	763	store_thread = threading.Thread( None, self.storeMetrics, 'storethread' )
[22]	764
[36]	765	while( 1 ):
	766
[102]	767	if not xml_thread.isAlive():
[36]	768	# Gather XML at the same interval as gmetad
	769
	770	# threaded call to: self.processXML()
	771	#
[102]	772	xml_thread = threading.Thread( None, self.processXML, 'xml_thread' )
	773	xml_thread.start()
[36]	774
[102]	775	if not store_thread.isAlive():
[55]	776	# Store metrics every .. sec
[36]	777
[55]	778	# threaded call to: self.storeMetrics()
	779	#
[102]	780	store_thread = threading.Thread( None, self.storeMetrics, 'store_thread' )
	781	store_thread.start()
[36]	782
	783	# Just sleep a sec here, to prevent daemon from going mad. We're all threads here anyway
	784	time.sleep( 1 )
	785
[33]	786	def storeMetrics( self ):
[63]	787	"""Store metrics retained in memory to disk"""
[22]	788
[102]	789	debug_msg( 1, printTime() + ' - ganglia_store_thread(): started.' )
[39]	790
[63]	791	# Store metrics somewhere between every 360 and 640 seconds
[38]	792	#
[55]	793	STORE_INTERVAL = random.randint( 360, 640 )
[22]	794
[102]	795	store_metric_thread = threading.Thread( None, self.storeThread, 'store_metric_thread' )
	796	store_metric_thread.start()
[36]	797
[102]	798	debug_msg( 1, printTime() + ' - ganglia_store_thread(): Sleeping.. (%ss)' %STORE_INTERVAL )
[36]	799	time.sleep( STORE_INTERVAL )
[102]	800	debug_msg( 1, printTime() + ' - ganglia_store_thread(): Done sleeping.' )
[36]	801
[102]	802	if store_metric_thread.isAlive():
[36]	803
[102]	804	debug_msg( 1, printTime() + ' - ganglia_store_thread(): storemetricthread() still running, waiting to finish..' )
[136]	805	store_metric_thread.join( STORE_TIMEOUT ) # Maximum time is for storing thread to finish
[102]	806	debug_msg( 1, printTime() + ' - ganglia_store_thread(): Done waiting.' )
[36]	807
[102]	808	debug_msg( 1, printTime() + ' - ganglia_store_thread(): finished.' )
[36]	809
	810	return 0
	811
[39]	812	def storeThread( self ):
[63]	813	"""Actual metric storing thread"""
[39]	814
[102]	815	debug_msg( 1, printTime() + ' - ganglia_store_metric_thread(): started.' )
	816	debug_msg( 1, printTime() + ' - ganglia_store_metric_thread(): Storing data..' )
[78]	817	ret = self.myXMLHandler.storeMetrics()
[102]	818	debug_msg( 1, printTime() + ' - ganglia_store_metric_thread(): Done storing.' )
	819	debug_msg( 1, printTime() + ' - ganglia_store_metric_thread(): finished.' )
[39]	820
	821	return ret
	822
[8]	823	def processXML( self ):
[63]	824	"""Process XML"""
[8]	825
[87]	826	debug_msg( 1, printTime() + ' - xmlthread(): started.' )
[8]	827
[39]	828	parsethread = threading.Thread( None, self.parseThread, 'parsethread' )
[36]	829	parsethread.start()
	830
[102]	831	debug_msg( 1, printTime() + ' - ganglia_xml_thread(): Sleeping.. (%ss)' %self.config.getLowestInterval() )
[36]	832	time.sleep( float( self.config.getLowestInterval() ) )
[102]	833	debug_msg( 1, printTime() + ' - ganglia_xml_thread(): Done sleeping.' )
[36]	834
	835	if parsethread.isAlive():
	836
[102]	837	debug_msg( 1, printTime() + ' - ganglia_xml_thread(): parsethread() still running, waiting to finish..' )
[47]	838	parsethread.join( PARSE_TIMEOUT ) # Maximum time for XML thread to finish
[102]	839	debug_msg( 1, printTime() + ' - ganglia_xml_thread(): Done waiting.' )
[36]	840
[102]	841	debug_msg( 1, printTime() + ' - ganglia_xml_thread(): finished.' )
[36]	842
	843	return 0
	844
[39]	845	def parseThread( self ):
[63]	846	"""Actual parsing thread"""
[39]	847
[102]	848	debug_msg( 1, printTime() + ' - ganglia_parse_thread(): started.' )
	849	debug_msg( 1, printTime() + ' - ganglia_parse_thread(): Parsing XML..' )
[78]	850	self.myXMLSource = self.myXMLGatherer.getFileObject()
	851	ret = xml.sax.parse( self.myXMLSource, self.myXMLHandler, self.myXMLError )
[102]	852	debug_msg( 1, printTime() + ' - ganglia_parse_thread(): Done parsing.' )
	853	debug_msg( 1, printTime() + ' - ganglia_parse_thread(): finished.' )
[39]	854
	855	return ret
	856
[9]	857	class GangliaConfigParser:
	858
[34]	859	sources = [ ]
[9]	860
	861	def __init__( self, config ):
[63]	862	"""Parse some stuff from our gmetad's config, such as polling interval"""
[32]	863
[9]	864	self.config = config
	865	self.parseValues()
	866
[32]	867	def parseValues( self ):
[63]	868	"""Parse certain values from gmetad.conf"""
[9]	869
	870	readcfg = open( self.config, 'r' )
	871
	872	for line in readcfg.readlines():
	873
	874	if line.count( '"' ) > 1:
	875
[10]	876	if line.find( 'data_source' ) != -1 and line[0] != '#':
[9]	877
[11]	878	source = { }
	879	source['name'] = line.split( '"' )[1]
[9]	880	source_words = line.split( '"' )[2].split( ' ' )
	881
	882	for word in source_words:
	883
	884	valid_interval = 1
	885
	886	for letter in word:
[32]	887
[9]	888	if letter not in string.digits:
[32]	889
[9]	890	valid_interval = 0
	891
[10]	892	if valid_interval and len(word) > 0:
[32]	893
[9]	894	source['interval'] = word
[12]	895	debug_msg( 9, 'polling interval for %s = %s' %(source['name'], source['interval'] ) )
[33]	896
	897	# No interval found, use Ganglia's default
	898	if not source.has_key( 'interval' ):
	899	source['interval'] = 15
	900	debug_msg( 9, 'polling interval for %s defaulted to 15' %(source['name']) )
[32]	901
[33]	902	self.sources.append( source )
[9]	903
	904	def getInterval( self, source_name ):
[63]	905	"""Return interval for source_name"""
[32]	906
[9]	907	for source in self.sources:
[32]	908
[12]	909	if source['name'] == source_name:
[32]	910
[9]	911	return source['interval']
[32]	912
[9]	913	return None
	914
[34]	915	def getLowestInterval( self ):
[63]	916	"""Return the lowest interval of all clusters"""
[34]	917
	918	lowest_interval = 0
	919
	920	for source in self.sources:
	921
	922	if not lowest_interval or source['interval'] <= lowest_interval:
	923
	924	lowest_interval = source['interval']
	925
	926	# Return 15 when nothing is found, so that the daemon won't go insane with 0 sec delays
	927	if lowest_interval:
	928	return lowest_interval
	929	else:
	930	return 15
	931
[9]	932	class RRDHandler:
[63]	933	"""Class for handling RRD activity"""
[9]	934
[32]	935	myMetrics = { }
[40]	936	lastStored = { }
[47]	937	timeserials = { }
[36]	938	slot = None
[32]	939
[33]	940	def __init__( self, config, cluster ):
[63]	941	"""Setup initial variables"""
[78]	942
[44]	943	self.block = 0
[9]	944	self.cluster = cluster
[33]	945	self.config = config
[40]	946	self.slot = threading.Lock()
[37]	947	self.rrdm = RRDMutator()
[42]	948	self.gatherLastUpdates()
[9]	949
[42]	950	def gatherLastUpdates( self ):
[63]	951	"""Populate the lastStored list, containing timestamps of all last updates"""
[42]	952
	953	cluster_dir = '%s/%s' %( check_dir(ARCHIVE_PATH), self.cluster )
	954
	955	hosts = [ ]
	956
	957	if os.path.exists( cluster_dir ):
	958
[44]	959	dirlist = os.listdir( cluster_dir )
[42]	960
[44]	961	for dir in dirlist:
[42]	962
[44]	963	hosts.append( dir )
[42]	964
	965	for host in hosts:
	966
[47]	967	host_dir = cluster_dir + '/' + host
	968	dirlist = os.listdir( host_dir )
	969
	970	for dir in dirlist:
	971
	972	if not self.timeserials.has_key( host ):
	973
	974	self.timeserials[ host ] = [ ]
	975
	976	self.timeserials[ host ].append( dir )
	977
[42]	978	last_serial = self.getLastRrdTimeSerial( host )
	979	if last_serial:
	980
	981	metric_dir = cluster_dir + '/' + host + '/' + last_serial
	982	if os.path.exists( metric_dir ):
	983
[44]	984	dirlist = os.listdir( metric_dir )
[42]	985
[44]	986	for file in dirlist:
[42]	987
[44]	988	metricname = file.split( '.rrd' )[0]
[42]	989
[44]	990	if not self.lastStored.has_key( host ):
[42]	991
[44]	992	self.lastStored[ host ] = { }
[42]	993
[44]	994	self.lastStored[ host ][ metricname ] = self.rrdm.grabLastUpdate( metric_dir + '/' + file )
[42]	995
[32]	996	def getClusterName( self ):
[63]	997	"""Return clustername"""
	998
[32]	999	return self.cluster
	1000
	1001	def memMetric( self, host, metric ):
[63]	1002	"""Store metric from host in memory"""
[32]	1003
[34]	1004	if self.myMetrics.has_key( host ):
[32]	1005
[34]	1006	if self.myMetrics[ host ].has_key( metric['name'] ):
[32]	1007
[34]	1008	for mymetric in self.myMetrics[ host ][ metric['name'] ]:
[32]	1009
[34]	1010	if mymetric['time'] == metric['time']:
[32]	1011
[34]	1012	# Allready have this metric, abort
	1013	return 1
	1014	else:
	1015	self.myMetrics[ host ][ metric['name'] ] = [ ]
	1016	else:
[32]	1017	self.myMetrics[ host ] = { }
[34]	1018	self.myMetrics[ host ][ metric['name'] ] = [ ]
[32]	1019
[63]	1020	# Push new metric onto stack
	1021	# atomic code; only 1 thread at a time may access the stack
	1022
[53]	1023	# <ATOMIC>
[40]	1024	#
	1025	self.slot.acquire()
	1026
[32]	1027	self.myMetrics[ host ][ metric['name'] ].append( metric )
	1028
[40]	1029	self.slot.release()
[53]	1030	#
	1031	# </ATOMIC>
[40]	1032
[47]	1033	def makeUpdateList( self, host, metriclist ):
[63]	1034	"""
	1035	Make a list of update values for rrdupdate
	1036	but only those that we didn't store before
	1037	"""
[37]	1038
	1039	update_list = [ ]
[41]	1040	metric = None
[37]	1041
[47]	1042	while len( metriclist ) > 0:
[37]	1043
[53]	1044	metric = metriclist.pop( 0 )
[37]	1045
[53]	1046	if self.checkStoreMetric( host, metric ):
	1047	update_list.append( '%s:%s' %( metric['time'], metric['val'] ) )
[40]	1048
[37]	1049	return update_list
	1050
[49]	1051	def checkStoreMetric( self, host, metric ):
[63]	1052	"""Check if supplied metric if newer than last one stored"""
[40]	1053
	1054	if self.lastStored.has_key( host ):
	1055
[47]	1056	if self.lastStored[ host ].has_key( metric['name'] ):
[40]	1057
[47]	1058	if metric['time'] <= self.lastStored[ host ][ metric['name'] ]:
[40]	1059
[50]	1060	# This is old
[40]	1061	return 0
	1062
[50]	1063	return 1
	1064
[54]	1065	def memLastUpdate( self, host, metricname, metriclist ):
[63]	1066	"""
	1067	Memorize the time of the latest metric from metriclist
	1068	but only if it wasn't allready memorized
	1069	"""
[50]	1070
[54]	1071	if not self.lastStored.has_key( host ):
	1072	self.lastStored[ host ] = { }
	1073
[50]	1074	last_update_time = 0
	1075
	1076	for metric in metriclist:
	1077
[54]	1078	if metric['name'] == metricname:
[50]	1079
[54]	1080	if metric['time'] > last_update_time:
[50]	1081
[54]	1082	last_update_time = metric['time']
[40]	1083
[54]	1084	if self.lastStored[ host ].has_key( metricname ):
[52]	1085
[54]	1086	if last_update_time <= self.lastStored[ host ][ metricname ]:
[52]	1087	return 1
[40]	1088
[54]	1089	self.lastStored[ host ][ metricname ] = last_update_time
[52]	1090
[33]	1091	def storeMetrics( self ):
[63]	1092	"""
	1093	Store all metrics from memory to disk
	1094	and do it to the RRD's in appropriate timeperiod directory
	1095	"""
[33]	1096
	1097	for hostname, mymetrics in self.myMetrics.items():
	1098
	1099	for metricname, mymetric in mymetrics.items():
	1100
[53]	1101	metrics_to_store = [ ]
	1102
[63]	1103	# Pop metrics from stack for storing until none is left
	1104	# atomic code: only 1 thread at a time may access myMetrics
	1105
[53]	1106	# <ATOMIC>
[50]	1107	#
[47]	1108	self.slot.acquire()
[33]	1109
[54]	1110	while len( self.myMetrics[ hostname ][ metricname ] ) > 0:
[53]	1111
[54]	1112	if len( self.myMetrics[ hostname ][ metricname ] ) > 0:
	1113	metrics_to_store.append( self.myMetrics[ hostname ][ metricname ].pop( 0 ) )
[53]	1114
	1115	self.slot.release()
	1116	#
	1117	# </ATOMIC>
	1118
[47]	1119	# Create a mapping table, each metric to the period where it should be stored
	1120	#
[53]	1121	metric_serial_table = self.determineSerials( hostname, metricname, metrics_to_store )
[33]	1122
[50]	1123	update_rets = [ ]
	1124
[47]	1125	for period, pmetric in metric_serial_table.items():
	1126
	1127	self.createCheck( hostname, metricname, period )
	1128
	1129	update_ret = self.update( hostname, metricname, period, pmetric )
	1130
	1131	if update_ret == 0:
	1132
	1133	debug_msg( 9, 'stored metric %s for %s' %( hostname, metricname ) )
	1134	else:
	1135	debug_msg( 9, 'metric update failed' )
	1136
[50]	1137	update_rets.append( update_ret )
[47]	1138
[50]	1139	if not (1) in update_rets:
	1140
[54]	1141	self.memLastUpdate( hostname, metricname, metrics_to_store )
[50]	1142
[17]	1143	def makeTimeSerial( self ):
[63]	1144	"""Generate a time serial. Seconds since epoch"""
[17]	1145
	1146	# Seconds since epoch
	1147	mytime = int( time.time() )
	1148
	1149	return mytime
	1150
[50]	1151	def makeRrdPath( self, host, metricname, timeserial ):
[63]	1152	"""Make a RRD location/path and filename"""
[17]	1153
[50]	1154	rrd_dir = '%s/%s/%s/%s' %( check_dir(ARCHIVE_PATH), self.cluster, host, timeserial )
	1155	rrd_file = '%s/%s.rrd' %( rrd_dir, metricname )
[17]	1156
	1157	return rrd_dir, rrd_file
	1158
[20]	1159	def getLastRrdTimeSerial( self, host ):
[63]	1160	"""Find the last timeserial (directory) for this host"""
[17]	1161
[19]	1162	newest_timeserial = 0
	1163
[47]	1164	for dir in self.timeserials[ host ]:
[32]	1165
[47]	1166	valid_dir = 1
[17]	1167
[47]	1168	for letter in dir:
	1169	if letter not in string.digits:
	1170	valid_dir = 0
[17]	1171
[47]	1172	if valid_dir:
	1173	timeserial = dir
	1174	if timeserial > newest_timeserial:
	1175	newest_timeserial = timeserial
[17]	1176
	1177	if newest_timeserial:
[18]	1178	return newest_timeserial
[17]	1179	else:
	1180	return 0
	1181
[47]	1182	def determinePeriod( self, host, check_serial ):
[63]	1183	"""Determine to which period (directory) this time(serial) belongs"""
[47]	1184
	1185	period_serial = 0
	1186
[56]	1187	if self.timeserials.has_key( host ):
[47]	1188
[56]	1189	for serial in self.timeserials[ host ]:
[47]	1190
[56]	1191	if check_serial >= serial and period_serial < serial:
[47]	1192
[56]	1193	period_serial = serial
	1194
[47]	1195	return period_serial
	1196
	1197	def determineSerials( self, host, metricname, metriclist ):
	1198	"""
	1199	Determine the correct serial and corresponding rrd to store
	1200	for a list of metrics
	1201	"""
	1202
	1203	metric_serial_table = { }
	1204
	1205	for metric in metriclist:
	1206
	1207	if metric['name'] == metricname:
	1208
	1209	period = self.determinePeriod( host, metric['time'] )
	1210
	1211	archive_secs = ARCHIVE_HOURS_PER_RRD * (60 * 60)
	1212
[49]	1213	if (int( metric['time'] ) - int( period ) ) > archive_secs:
[47]	1214
	1215	# This one should get it's own new period
	1216	period = metric['time']
[57]	1217
	1218	if not self.timeserials.has_key( host ):
	1219	self.timeserials[ host ] = [ ]
	1220
[50]	1221	self.timeserials[ host ].append( period )
[47]	1222
	1223	if not metric_serial_table.has_key( period ):
	1224
[49]	1225	metric_serial_table[ period ] = [ ]
[47]	1226
	1227	metric_serial_table[ period ].append( metric )
	1228
	1229	return metric_serial_table
	1230
[33]	1231	def createCheck( self, host, metricname, timeserial ):
[63]	1232	"""Check if an rrd allready exists for this metric, create if not"""
[9]	1233
[35]	1234	debug_msg( 9, 'rrdcreate: using timeserial %s for %s/%s' %( timeserial, host, metricname ) )
[47]	1235
[33]	1236	rrd_dir, rrd_file = self.makeRrdPath( host, metricname, timeserial )
[17]	1237
[9]	1238	if not os.path.exists( rrd_dir ):
[58]	1239
	1240	try:
	1241	os.makedirs( rrd_dir )
	1242
	1243	except OSError, msg:
	1244
	1245	if msg.find( 'File exists' ) != -1:
	1246
	1247	# Ignore exists errors
	1248	pass
	1249
	1250	else:
	1251
	1252	print msg
	1253	return
	1254
[14]	1255	debug_msg( 9, 'created dir %s' %( str(rrd_dir) ) )
[9]	1256
[14]	1257	if not os.path.exists( rrd_file ):
[9]	1258
[33]	1259	interval = self.config.getInterval( self.cluster )
[47]	1260	heartbeat = 8 * int( interval )
[9]	1261
[37]	1262	params = [ ]
[12]	1263
[37]	1264	params.append( '--step' )
	1265	params.append( str( interval ) )
[12]	1266
[37]	1267	params.append( '--start' )
[47]	1268	params.append( str( int( timeserial ) - 1 ) )
[12]	1269
[37]	1270	params.append( 'DS:sum:GAUGE:%d:U:U' %heartbeat )
	1271	params.append( 'RRA:AVERAGE:0.5:1:%s' %(ARCHIVE_HOURS_PER_RRD * 240) )
[13]	1272
[37]	1273	self.rrdm.create( str(rrd_file), params )
	1274
[14]	1275	debug_msg( 9, 'created rrd %s' %( str(rrd_file) ) )
	1276
[47]	1277	def update( self, host, metricname, timeserial, metriclist ):
[63]	1278	"""
	1279	Update rrd file for host with metricname
	1280	in directory timeserial with metriclist
	1281	"""
[9]	1282
[35]	1283	debug_msg( 9, 'rrdupdate: using timeserial %s for %s/%s' %( timeserial, host, metricname ) )
[9]	1284
[33]	1285	rrd_dir, rrd_file = self.makeRrdPath( host, metricname, timeserial )
[18]	1286
[47]	1287	update_list = self.makeUpdateList( host, metriclist )
[15]	1288
[41]	1289	if len( update_list ) > 0:
	1290	ret = self.rrdm.update( str(rrd_file), update_list )
[32]	1291
[41]	1292	if ret:
	1293	return 1
[27]	1294
[41]	1295	debug_msg( 9, 'updated rrd %s with %s' %( str(rrd_file), string.join( update_list ) ) )
[15]	1296
[36]	1297	return 0
	1298
[8]	1299	def main():
[63]	1300	"""Program startup"""
[8]	1301
[102]	1302	myTorqueProcessor = TorqueXMLProcessor()
	1303	myGangliaProcessor = GangliaXMLProcessor()
[8]	1304
[63]	1305	if DAEMONIZE:
[102]	1306	torque_xml_thread = threading.Thread( None, myTorqueProcessor.daemon, 'torque_proc_thread' )
	1307	ganglia_xml_thread = threading.Thread( None, myGangliaProcessor.daemon, 'ganglia_proc_thread' )
[63]	1308	else:
[102]	1309	torque_xml_thread = threading.Thread( None, myTorqueProcessor.run, 'torque_proc_thread' )
	1310	ganglia_xml_thread = threading.Thread( None, myGangliaProcessor.run, 'ganglia_proc_thread' )
[22]	1311
[102]	1312	torque_xml_thread.start()
	1313	ganglia_xml_thread.start()
[78]	1314
[81]	1315	# Global functions
	1316
[9]	1317	def check_dir( directory ):
[63]	1318	"""Check if directory is a proper directory. I.e.: Does _not_ end with a '/'"""
[9]	1319
	1320	if directory[-1] == '/':
	1321	directory = directory[:-1]
	1322
	1323	return directory
	1324
[12]	1325	def debug_msg( level, msg ):
[63]	1326	"""Only print msg if it is not below our debug level"""
[12]	1327
	1328	if (DEBUG_LEVEL >= level):
	1329	sys.stderr.write( msg + '\n' )
	1330
[46]	1331	def printTime( ):
[63]	1332	"""Print current time in human readable format"""
[46]	1333
	1334	return time.strftime("%a %d %b %Y %H:%M:%S")
	1335
[63]	1336	# Ooohh, someone started me! Let's go..
[9]	1337	if __name__ == '__main__':
	1338	main()

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

Download in other formats: