Context Navigation

jobarchived.py @ 199

Last change on this file since 199 was 199, checked in by bastiaans, 18 years ago

daemon/jobarchived.py:

Bugfix: no longer store jobs in database that are not in ARCHIVE_DATSOURCES

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

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

Context Navigation

source: trunk/daemon/jobarchived.py @ 199

Download in other formats: