Index: trunk/daemon/togad.py
===================================================================
--- trunk/daemon/togad.py	(revision 62)
+++ trunk/daemon/togad.py	(revision 63)
@@ -9,7 +9,5 @@
 import os.path
 import time
-import re
 import threading
-import mutex
 import random
 from types import *
@@ -30,17 +28,15 @@
 GMETAD_CONF = '/etc/gmetad.conf'
 
-# Wether or not to maintain a archive of all ganglia node data
-# Note: This will require a significant amount of hd space
-#	depending on your cluster size
-#
-ARCHIVE = 0
-
 # Where to grab XML data from
 # Normally: local gmetad (port 8651)
 #
-ARCHIVE_SOURCE = "localhost:8651"
+# Syntax: <hostname>:<port>
+#
+ARCHIVE_XMLSOURCE = "localhost:8651"
 
 # List of data_source names to archive for
 #
+# Syntax: [ "<clustername>", "<clustername>" ]
+#
 ARCHIVE_DATASOURCES = [ "LISA Cluster" ]
 
@@ -53,19 +49,9 @@
 ARCHIVE_HOURS_PER_RRD = 12
 
-# Wether or not to run a seperate Toga jobinfo server
-#
-TOGA_SERVER = 1
-
-# On what interfaces to listen
-#
-TOGA_SERVER_IP = [ '127.0.0.1' ]
-
-# On what port to listen
-#
-TOGA_SERVER_PORT = 9048
-
-# Toga's SQL dbase name to use
-#
-TOGA_SERVER_SQL_DBASE = "toga"
+# Toga's SQL dbase to use
+#
+# Syntax: <hostname>/<database>
+#
+TOGA_SQL_DBASE = "localhost/toga"
 
 # Wether or not to run as a daemon in background
@@ -81,5 +67,5 @@
 ###
 # You'll only want to change anything below here unless you 
-# know what you are doing (i.e. your name is Ramon Bastiaans)
+# know what you are doing (i.e. your name is Ramon Bastiaans :D )
 ###
 
@@ -100,76 +86,11 @@
 """
 
-class TogaServer:
-
-	sockets = [ ]
-
-
-
-	def __init__( self ):
-
-		s = None
-		for host in TOGA_SERVER_IP:
-
-			for res in socket.getaddrinfo( host, TOGA_SERVER_PORT, socket.AF_UNSPEC, socket.SOCK_STREAM, 0, socket.AI_PASSIVE ):
-
-				af, socktype, proto, canonname, sa = res
-
-				try:
-
-					s = socket.socket( af, socktype, proto )
-
-				except socket.error, msg:
-
-					s = None
-					continue
-
-				try:
-					s.bind( sa )
-					s.listen( 1 )
-
-				except socket.error, msg:
-
-					s.close()
-					s = None
-					continue
-					break
-
-				if not self.s:
-
-					debug_msg( 6, 'Could not open socket' )
-					return None
-
-				else:
-
-					self.sockets.append( s )
-
-	def run( self ):
-
-		for s in self.sockets:
-
-			while( 1 ):
-
-				conn, addr = s.accept()
-				pid = os.fork()
-
-				if pid == 0:
-
-					debug_msg( 6, 'New connection to %s' %addr[0] )
-
-					leesme = conn.makefile( 'r' )
-					for line in leesme.readlines():
-						print line
-					
-					conn.close()
-					conn.shutdown( 2 )
-
-					sys.exit( 0 )
-
 class RRDMutator:
-	"A class for handling .rrd mutations"
+	"""A class for performing RRD mutations"""
 
 	binary = '/usr/bin/rrdtool'
 
 	def __init__( self, binary=None ):
+		"""Set alternate binary if supplied"""
 
 		if binary:
@@ -177,10 +98,15 @@
 
 	def create( self, filename, args ):
+		"""Create a new rrd with args"""
+
 		return self.perform( 'create', '"' + filename + '"', args )
 
 	def update( self, filename, args ):
+		"""Update a rrd with args"""
+
 		return self.perform( 'update', '"' + filename + '"', args )
 
 	def grabLastUpdate( self, filename ):
+		"""Determine the last update time of filename rrd"""
 
 		last_update = 0
@@ -200,4 +126,5 @@
 
 	def perform( self, action, filename, args ):
+		"""Perform action on rrd filename with args"""
 
 		arg_string = None
@@ -227,8 +154,47 @@
 		return 0
 
+class TorqueXMLHandler( ContentHandler ):
+	"""Parse Torque's jobinfo XML from our plugin"""
+
+	def __init__( self ):
+
+		pass
+
+	def startElement( self, name, attrs ):
+		"""
+		This XML will be all gmetric XML
+		so there will be no specific start/end element
+		just one XML statement with all info
+		"""
+		
+		heartbeat = 0
+
+		if name == 'METRIC':
+
+			metricname = attrss.get( 'NAME', "" )
+
+			if metricname == 'TOGA-HEARTBEAT':
+
+				if not self.heartbeat:
+					self.heartbeat = attrs.get( 'VAL', "" )
+
+			if metricname.find( 'TOGA-JOB' ) != -1:
+
+				job_id = name.split( 'TOGA-JOB-' )[1]
+				val = attrs.get( 'VAL', "" )
+
+				valinfo = val.split( ' ' )
+
+				for myval in valinfo:
+
+					name = valinfo.split( '=' )[0]
+					value = valinfo.split( '=' )[1]
+
 class GangliaXMLHandler( ContentHandler ):
-	"Parse Ganglia's XML"
+	"""Parse Ganglia's XML"""
 
 	def __init__( self, config ):
+		"""Setup initial variables and gather info on existing rrd archive"""
+
 		self.config = config
 		self.clusters = { }
@@ -238,4 +204,5 @@
 
 	def gatherClusters( self ):
+		"""Find all existing clusters in archive dir"""
 
 		archive_dir = check_dir(ARCHIVE_PATH)
@@ -256,5 +223,5 @@
 
 	def startElement( self, name, attrs ):
-		"Store appropriate data from xml start tags"
+		"""Memorize appropriate data from xml start tags"""
 
 		if name == 'GANGLIA_XML':
@@ -307,4 +274,5 @@
 
 	def storeMetrics( self ):
+		"""Store metrics of each cluster rrd handler"""
 
 		for clustername, rrdh in self.clusters.items():
@@ -319,10 +287,10 @@
 
 class GangliaXMLGatherer:
-	"Setup a connection and file object to Ganglia's XML"
+	"""Setup a connection and file object to Ganglia's XML"""
 
 	s = None
 
 	def __init__( self, host, port ):
-		"Store host and port for connection"
+		"""Store host and port for connection"""
 
 		self.host = host
@@ -331,5 +299,5 @@
 
 	def connect( self ):
-		"Setup connection to XML source"
+		"""Setup connection to XML source"""
 
 		for res in socket.getaddrinfo( self.host, self.port, socket.AF_UNSPEC, socket.SOCK_STREAM ):
@@ -364,5 +332,5 @@
 
 	def disconnect( self ):
-		"Close socket"
+		"""Close socket"""
 
 		if self.s:
@@ -371,10 +339,10 @@
 
 	def __del__( self ):
-		"Kill the socket before we leave"
+		"""Kill the socket before we leave"""
 
 		self.disconnect()
 
 	def getFileObject( self ):
-		"Connect, and return a file object"
+		"""Connect, and return a file object"""
 
 		if self.s:
@@ -388,11 +356,12 @@
 
 class GangliaXMLProcessor:
+	"""Main class for processing XML and acting with it"""
 
 	def __init__( self ):
-		"Setup initial XML connection and handlers"
+		"""Setup initial XML connection and handlers"""
 
 		self.config = GangliaConfigParser( GMETAD_CONF )
 
-		self.myXMLGatherer = GangliaXMLGatherer( ARCHIVE_SOURCE.split( ':' )[0], ARCHIVE_SOURCE.split( ':' )[1] ) 
+		self.myXMLGatherer = GangliaXMLGatherer( ARCHIVE_XMLSOURCE.split( ':' )[0], ARCHIVE_XMLSOURCE.split( ':' )[1] ) 
 		self.myParser = make_parser()   
 		self.myHandler = GangliaXMLHandler( self.config )
@@ -400,7 +369,5 @@
 
 	def daemon( self ):
-		"Run as daemon forever"
-
-		self.DAEMON = 1
+		"""Run as daemon forever"""
 
 		# Fork the first child
@@ -440,10 +407,10 @@
 
 	def printTime( self ):
-		"Print current time in human readable format"
+		"""Print current time in human readable format for logging"""
 
 		return time.strftime("%a %d %b %Y %H:%M:%S")
 
 	def run( self ):
-		"Main thread"
+		"""Main XML processing; start a xml and storethread"""
 
 		xmlthread = threading.Thread( None, self.processXML, 'xmlthread' )
@@ -472,9 +439,9 @@
 
 	def storeMetrics( self ):
-		"Store metrics retained in memory to disk"
+		"""Store metrics retained in memory to disk"""
 
 		debug_msg( 7, self.printTime() + ' - storethread(): started.' )
 
-		# Store metrics somewhere between every 60 and 180 seconds
+		# Store metrics somewhere between every 360 and 640 seconds
 		#
 		STORE_INTERVAL = random.randint( 360, 640 )
@@ -498,4 +465,5 @@
 
 	def storeThread( self ):
+		"""Actual metric storing thread"""
 
 		debug_msg( 7, self.printTime() + ' - storemetricthread(): started.' )
@@ -508,5 +476,5 @@
 
 	def processXML( self ):
-		"Process XML"
+		"""Process XML"""
 
 		debug_msg( 7, self.printTime() + ' - xmlthread(): started.' )
@@ -530,4 +498,5 @@
 
 	def parseThread( self ):
+		"""Actual parsing thread"""
 
 		debug_msg( 7, self.printTime() + ' - parsethread(): started.' )
@@ -544,4 +513,5 @@
 
 	def __init__( self, config ):
+		"""Parse some stuff from our gmetad's config, such as polling interval"""
 
 		self.config = config
@@ -549,5 +519,5 @@
 
 	def parseValues( self ):
-		"Parse certain values from gmetad.conf"
+		"""Parse certain values from gmetad.conf"""
 
 		readcfg = open( self.config, 'r' )
@@ -586,4 +556,5 @@
 
 	def getInterval( self, source_name ):
+		"""Return interval for source_name"""
 
 		for source in self.sources:
@@ -596,4 +567,5 @@
 
 	def getLowestInterval( self ):
+		"""Return the lowest interval of all clusters"""
 
 		lowest_interval = 0
@@ -612,4 +584,5 @@
 
 class RRDHandler:
+	"""Class for handling RRD activity"""
 
 	myMetrics = { }
@@ -619,4 +592,5 @@
 
 	def __init__( self, config, cluster ):
+		"""Setup initial variables"""
 		self.block = 0
 		self.cluster = cluster
@@ -627,5 +601,5 @@
 
 	def gatherLastUpdates( self ):
-		"Populate the lastStored list, containing timestamps of all last updates"
+		"""Populate the lastStored list, containing timestamps of all last updates"""
 
 		cluster_dir = '%s/%s' %( check_dir(ARCHIVE_PATH), self.cluster )
@@ -673,7 +647,10 @@
 
 	def getClusterName( self ):
+		"""Return clustername"""
+
 		return self.cluster
 
 	def memMetric( self, host, metric ):
+		"""Store metric from host in memory"""
 
 		if self.myMetrics.has_key( host ):
@@ -693,4 +670,7 @@
 			self.myMetrics[ host ][ metric['name'] ] = [ ]
 
+		# Push new metric onto stack
+		# atomic code; only 1 thread at a time may access the stack
+
 		# <ATOMIC>
 		#
@@ -704,4 +684,8 @@
 
 	def makeUpdateList( self, host, metriclist ):
+		"""
+		Make a list of update values for rrdupdate
+		but only those that we didn't store before
+		"""
 
 		update_list = [ ]
@@ -718,4 +702,5 @@
 
 	def checkStoreMetric( self, host, metric ):
+		"""Check if supplied metric if newer than last one stored"""
 
 		if self.lastStored.has_key( host ):
@@ -731,4 +716,8 @@
 
 	def memLastUpdate( self, host, metricname, metriclist ):
+		"""
+		Memorize the time of the latest metric from metriclist
+		but only if it wasn't allready memorized
+		"""
 
 		if not self.lastStored.has_key( host ):
@@ -753,4 +742,8 @@
 
 	def storeMetrics( self ):
+		"""
+		Store all metrics from memory to disk
+		and do it to the RRD's in appropriate timeperiod directory
+		"""
 
 		for hostname, mymetrics in self.myMetrics.items():	
@@ -759,4 +752,7 @@
 
 				metrics_to_store = [ ]
+
+				# Pop metrics from stack for storing until none is left
+				# atomic code: only 1 thread at a time may access myMetrics
 
 				# <ATOMIC>
@@ -798,5 +794,5 @@
 
 	def makeTimeSerial( self ):
-		"Generate a time serial. Seconds since epoch"
+		"""Generate a time serial. Seconds since epoch"""
 
 		# Seconds since epoch
@@ -806,9 +802,5 @@
 
 	def makeRrdPath( self, host, metricname, timeserial ):
-		"""
-		Make a RRD location/path and filename
-		If a metric or timeserial are supplied the complete locations
-		will be made, else just the host directory
-		"""
+		"""Make a RRD location/path and filename"""
 
 		rrd_dir = '%s/%s/%s/%s' %( check_dir(ARCHIVE_PATH), self.cluster, host, timeserial )
@@ -818,8 +810,5 @@
 
 	def getLastRrdTimeSerial( self, host ):
-		"""
-		Find the last timeserial (directory) for this host
-		This is determined once every host
-		"""
+		"""Find the last timeserial (directory) for this host"""
 
 		newest_timeserial = 0
@@ -844,4 +833,5 @@
 
 	def determinePeriod( self, host, check_serial ):
+		"""Determine to which period (directory) this time(serial) belongs"""
 
 		period_serial = 0
@@ -892,5 +882,5 @@
 
 	def createCheck( self, host, metricname, timeserial ):
-		"Check if an .rrd allready exists for this metric, create if not"
+		"""Check if an rrd allready exists for this metric, create if not"""
 
 		debug_msg( 9, 'rrdcreate: using timeserial %s for %s/%s' %( timeserial, host, metricname ) )
@@ -938,4 +928,8 @@
 
 	def update( self, host, metricname, timeserial, metriclist ):
+		"""
+		Update rrd file for host with metricname
+		in directory timeserial with metriclist
+		"""
 
 		debug_msg( 9, 'rrdupdate: using timeserial %s for %s/%s' %( timeserial, host, metricname ) )
@@ -956,26 +950,15 @@
 
 def main():
-	"Program startup"
-
-	if TOGA_SERVER:
-	
-		myServer = TogaServer()
-
-		if DAEMONIZE:
-			myServer.daemon()
-		else:
-			myServer.run()
-
-	if ARCHIVE:
-
-		myProcessor = GangliaXMLProcessor()
-
-		if DAEMONIZE:
-			myProcessor.daemon()
-		else:
-			myProcessor.run()
+	"""Program startup"""
+
+	myProcessor = GangliaXMLProcessor()
+
+	if DAEMONIZE:
+		myProcessor.daemon()
+	else:
+		myProcessor.run()
 
 def check_dir( directory ):
-	"Check if directory is a proper directory. I.e.: Does _not_ end with a '/'"
+	"""Check if directory is a proper directory. I.e.: Does _not_ end with a '/'"""
 
 	if directory[-1] == '/':
@@ -985,4 +968,5 @@
 
 def debug_msg( level, msg ):
+	"""Only print msg if it is not below our debug level"""
 
 	if (DEBUG_LEVEL >= level):
@@ -990,9 +974,9 @@
 
 def printTime( ):
-	"Print current time in human readable format"
+	"""Print current time in human readable format"""
 
 	return time.strftime("%a %d %b %Y %H:%M:%S")
 
-# Let's go
+# Ooohh, someone started me! Let's go..
 if __name__ == '__main__':
 	main()
