Context Navigation

source: trunk/daemon/jobarchived.py @ 198

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

daemon/jobarchived.py:

Now with support for ARCHIVE_EXCLUDE_METRICS ! List of metric names (string or regexp) not to archive

File size: 34.8 KB

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

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

Download in other formats: