Context Navigation

source: trunk/daemon/togad.py @ 190

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

daemon/togad.py:

Fix: don't save node info from jobs in Q state
Fix: save node info on update

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

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

Download in other formats: