Index: trunk/jobarchived/jobarchived.py
===================================================================
--- trunk/jobarchived/jobarchived.py	(revision 289)
+++ trunk/jobarchived/jobarchived.py	(revision 292)
@@ -34,6 +34,6 @@
 def processArgs( args ):
 
-        SHORT_L = 'c:'
-        LONG_L = 'config='
+        SHORT_L	= 'c:'
+        LONG_L	= 'config='
 
         config_filename = None
@@ -106,17 +106,17 @@
 	global DEBUG_LEVEL, USE_SYSLOG, SYSLOG_LEVEL, SYSLOG_FACILITY, GMETAD_CONF, ARCHIVE_XMLSOURCE, ARCHIVE_DATASOURCES, ARCHIVE_PATH, ARCHIVE_HOURS_PER_RRD, ARCHIVE_EXCLUDE_METRICS, JOB_SQL_DBASE, DAEMONIZE, RRDTOOL
 
-	ARCHIVE_PATH = cfg.get( 'DEFAULT', 'ARCHIVE_PATH' )
-
-	ARCHIVE_HOURS_PER_RRD = cfg.getint( 'DEFAULT', 'ARCHIVE_HOURS_PER_RRD' )
-
-	DEBUG_LEVEL = cfg.getint( 'DEFAULT', 'DEBUG_LEVEL' )
-
-	USE_SYSLOG = cfg.getboolean( 'DEFAULT', 'USE_SYSLOG' )
-
-	SYSLOG_LEVEL = cfg.getint( 'DEFAULT', 'SYSLOG_LEVEL' )
+	ARCHIVE_PATH		= cfg.get( 'DEFAULT', 'ARCHIVE_PATH' )
+
+	ARCHIVE_HOURS_PER_RRD	= cfg.getint( 'DEFAULT', 'ARCHIVE_HOURS_PER_RRD' )
+
+	DEBUG_LEVEL		= cfg.getint( 'DEFAULT', 'DEBUG_LEVEL' )
+
+	USE_SYSLOG		= cfg.getboolean( 'DEFAULT', 'USE_SYSLOG' )
+
+	SYSLOG_LEVEL		= cfg.getint( 'DEFAULT', 'SYSLOG_LEVEL' )
 
 	try:
 
-		SYSLOG_FACILITY = eval( 'syslog.LOG_' + cfg.get( 'DEFAULT', 'SYSLOG_FACILITY' ) )
+		SYSLOG_FACILITY	= eval( 'syslog.LOG_' + cfg.get( 'DEFAULT', 'SYSLOG_FACILITY' ) )
 
 	except AttributeError, detail:
@@ -125,17 +125,17 @@
 		sys.exit( 1 )
 
-	GMETAD_CONF = cfg.get( 'DEFAULT', 'GMETAD_CONF' )
-
-	ARCHIVE_XMLSOURCE = cfg.get( 'DEFAULT', 'ARCHIVE_XMLSOURCE' )
-
-        ARCHIVE_DATASOURCES = getlist( cfg.get( 'DEFAULT', 'ARCHIVE_DATASOURCES' ) )
-
-	ARCHIVE_EXCLUDE_METRICS = getlist( cfg.get( 'DEFAULT', 'ARCHIVE_EXCLUDE_METRICS' ) )
-
-	JOB_SQL_DBASE = cfg.get( 'DEFAULT', 'JOB_SQL_DBASE' )
-
-	DAEMONIZE = cfg.getboolean( 'DEFAULT', 'DAEMONIZE' )
-
-	RRDTOOL = cfg.get( 'DEFAULT', 'RRDTOOL' )
+	GMETAD_CONF		= cfg.get( 'DEFAULT', 'GMETAD_CONF' )
+
+	ARCHIVE_XMLSOURCE	= cfg.get( 'DEFAULT', 'ARCHIVE_XMLSOURCE' )
+
+        ARCHIVE_DATASOURCES	= getlist( cfg.get( 'DEFAULT', 'ARCHIVE_DATASOURCES' ) )
+
+	ARCHIVE_EXCLUDE_METRICS	= getlist( cfg.get( 'DEFAULT', 'ARCHIVE_EXCLUDE_METRICS' ) )
+
+	JOB_SQL_DBASE		= cfg.get( 'DEFAULT', 'JOB_SQL_DBASE' )
+
+	DAEMONIZE		= cfg.getboolean( 'DEFAULT', 'DAEMONIZE' )
+
+	RRDTOOL			= cfg.get( 'DEFAULT', 'RRDTOOL' )
 
 	return True
@@ -264,9 +264,9 @@
 	def mutateJob( self, action, job_id, jobattrs ):
 
-		job_values = [ 'name', 'queue', 'owner', 'requested_time', 'requested_memory', 'ppn', 'status', 'start_timestamp', 'stop_timestamp' ]
-
-		insert_col_str = 'job_id'
-		insert_val_str = "'%s'" %job_id
-		update_str = None
+		job_values	= [ 'name', 'queue', 'owner', 'requested_time', 'requested_memory', 'ppn', 'status', 'start_timestamp', 'stop_timestamp' ]
+
+		insert_col_str	= 'job_id'
+		insert_val_str	= "'%s'" %job_id
+		update_str	= None
 
 		debug_msg( 6, 'mutateJob(): %s %s' %(action,job_id))
@@ -340,6 +340,6 @@
 		for node in hostnames:
 
-			node = '%s.%s' %( node, domain )
-			id = self.getNodeId( node )
+			node	= '%s.%s' %( node, domain )
+			id	= self.getNodeId( node )
 	
 			if not id:
@@ -395,9 +395,15 @@
 		debug_msg( 8, self.binary + ' info "' + filename + '"' )
 
-		for line in os.popen( self.binary + ' info "' + filename + '"' ).readlines():
+		my_pipe		= os.popen( self.binary + ' info "' + filename + '"' )
+
+		for line in my_pipe.readlines():
 
 			if line.find( 'last_update') != -1:
 
 				last_update = line.split( ' = ' )[1]
+
+		if my_pipe:
+
+			my_pipe.close()
 
 		if last_update:
@@ -425,6 +431,7 @@
 		debug_msg( 8, self.binary + ' ' + action + ' ' + filename + ' ' + arg_string  )
 
-		cmd = os.popen( self.binary + ' ' + action + ' ' + filename + ' ' + arg_string )
-		lines = cmd.readlines()
+		cmd	= os.popen( self.binary + ' ' + action + ' ' + filename + ' ' + arg_string )
+		lines	= cmd.readlines()
+
 		cmd.close()
 
@@ -489,11 +496,11 @@
 	def __init__( self ):
 
-		self.ds = DataSQLStore( JOB_SQL_DBASE.split( '/' )[0], JOB_SQL_DBASE.split( '/' )[1] )
-		self.jobs_processed = [ ]
-		self.jobs_to_store = [ ]
+		self.ds			= DataSQLStore( JOB_SQL_DBASE.split( '/' )[0], JOB_SQL_DBASE.split( '/' )[1] )
+		self.jobs_processed	= [ ]
+		self.jobs_to_store	= [ ]
 
 	def startDocument( self ):
 
-		self.heartbeat = 0
+		self.heartbeat	= 0
 
 	def startElement( self, name, attrs ):
@@ -519,8 +526,9 @@
 			elif metricname.find( 'MONARCH-JOB' ) != -1:
 
-				job_id = metricname.split( 'MONARCH-JOB-' )[1].split( '-' )[0]
-				val = attrs.get( 'VAL', "" )
+				job_id	= metricname.split( 'MONARCH-JOB-' )[1].split( '-' )[0]
+				val	= attrs.get( 'VAL', "" )
 
 				if not job_id in self.jobs_processed:
+
 					self.jobs_processed.append( job_id )
 
@@ -528,4 +536,5 @@
 
 				if self.jobAttrs.has_key( job_id ):
+
 					check_change = 1
 
@@ -536,6 +545,6 @@
 					if len( myval.split( '=' ) ) > 1:
 
-						valname = myval.split( '=' )[0]
-						value = myval.split( '=' )[1]
+						valname	= myval.split( '=' )[0]
+						value	= myval.split( '=' )[1]
 
 						if valname == 'nodes':
@@ -546,6 +555,6 @@
 				if check_change:
 					if self.jobinfoChanged( self.jobAttrs, job_id, jobinfo ) and self.jobAttrs[ job_id ]['status'] in [ 'R', 'Q' ]:
-						self.jobAttrs[ job_id ]['stop_timestamp'] = ''
-						self.jobAttrs[ job_id ] = self.setJobAttrs( self.jobAttrs[ job_id ], jobinfo )
+						self.jobAttrs[ job_id ]['stop_timestamp']	= ''
+						self.jobAttrs[ job_id ]				= self.setJobAttrs( self.jobAttrs[ job_id ], jobinfo )
 						if not job_id in self.jobs_to_store:
 							self.jobs_to_store.append( job_id )
@@ -594,6 +603,6 @@
 			debug_msg( 1, 'torque_xml_thread(): Done storing.' )
 
-			self.jobs_processed = [ ]
-			self.jobs_to_store = [ ]
+			self.jobs_processed	= [ ]
+			self.jobs_to_store	= [ ]
 
 	def setJobAttrs( self, old, new ):
@@ -642,8 +651,8 @@
 		"""Setup initial variables and gather info on existing rrd archive"""
 
-		self.config = config
-		self.clusters = { }
+		self.config	= config
+		self.clusters	= { }
 		debug_msg( 1, 'Checking existing toga rrd archive..' )
-		self.gatherClusters()
+		#self.gatherClusters()
 		debug_msg( 1, 'Check done.' )
 
@@ -651,11 +660,11 @@
 		"""Find all existing clusters in archive dir"""
 
-		archive_dir = check_dir(ARCHIVE_PATH)
-
-		hosts = [ ]
+		archive_dir	= check_dir(ARCHIVE_PATH)
+
+		hosts		= [ ]
 
 		if os.path.exists( archive_dir ):
 
-			dirlist = os.listdir( archive_dir )
+			dirlist	= os.listdir( archive_dir )
 
 			for item in dirlist:
@@ -672,6 +681,6 @@
 		if name == 'GANGLIA_XML':
 
-			self.XMLSource = attrs.get( 'SOURCE', "" )
-			self.gangliaVersion = attrs.get( 'VERSION', "" )
+			self.XMLSource		= attrs.get( 'SOURCE', "" )
+			self.gangliaVersion	= attrs.get( 'VERSION', "" )
 
 			debug_msg( 10, 'Found XML data: source %s version %s' %( self.XMLSource, self.gangliaVersion ) )
@@ -679,6 +688,6 @@
 		elif name == 'GRID':
 
-			self.gridName = attrs.get( 'NAME', "" )
-			self.time = attrs.get( 'LOCALTIME', "" )
+			self.gridName	= attrs.get( 'NAME', "" )
+			self.time	= attrs.get( 'LOCALTIME', "" )
 
 			debug_msg( 10, '`-Grid found: %s' %( self.gridName ) )
@@ -686,6 +695,6 @@
 		elif name == 'CLUSTER':
 
-			self.clusterName = attrs.get( 'NAME', "" )
-			self.time = attrs.get( 'LOCALTIME', "" )
+			self.clusterName	= attrs.get( 'NAME', "" )
+			self.time		= attrs.get( 'LOCALTIME', "" )
 
 			if not self.clusters.has_key( self.clusterName ) and self.clusterName in ARCHIVE_DATASOURCES:
@@ -697,7 +706,7 @@
 		elif name == 'HOST' and self.clusterName in ARCHIVE_DATASOURCES:     
 
-			self.hostName = attrs.get( 'NAME', "" )
-			self.hostIp = attrs.get( 'IP', "" )
-			self.hostReported = attrs.get( 'REPORTED', "" )
+			self.hostName		= attrs.get( 'NAME', "" )
+			self.hostIp		= attrs.get( 'IP', "" )
+			self.hostReported	= attrs.get( 'REPORTED', "" )
 
 			debug_msg( 10, ' | |-Host found: %s - ip %s reported %s' %( self.hostName, self.hostIp, self.hostReported ) )
@@ -723,8 +732,8 @@
 			if type not in UNSUPPORTED_ARCHIVE_TYPES and not exclude_metric:
 
-				myMetric = { }
-				myMetric['name'] = attrs.get( 'NAME', "" )
-				myMetric['val'] = attrs.get( 'VAL', "" )
-				myMetric['time'] = self.hostReported
+				myMetric		= { }
+				myMetric['name']	= attrs.get( 'NAME', "" )
+				myMetric['val']		= attrs.get( 'VAL', "" )
+				myMetric['time']	= self.hostReported
 
 				self.clusters[ self.clusterName ].memMetric( self.hostName, myMetric )
@@ -1070,7 +1079,7 @@
 				if line.find( 'data_source' ) != -1 and line[0] != '#':
 
-					source = { }
-					source['name'] = line.split( '"' )[1]
-					source_words = line.split( '"' )[2].split( ' ' )
+					source		= { }
+					source['name']	= line.split( '"' )[1]
+					source_words	= line.split( '"' )[2].split( ' ' )
 
 					for word in source_words:
@@ -1135,9 +1144,10 @@
 		"""Setup initial variables"""
 
-		self.block = 0
-		self.cluster = cluster
-		self.config = config
-		self.slot = threading.Lock()
-		self.rrdm = RRDMutator( RRDTOOL )
+		self.block	= 0
+		self.cluster	= cluster
+		self.config	= config
+		self.slot	= threading.Lock()
+		self.rrdm	= RRDMutator( RRDTOOL )
+
 		self.gatherLastUpdates()
 
@@ -1159,6 +1169,6 @@
 		for host in hosts:
 
-			host_dir = cluster_dir + '/' + host
-			dirlist = os.listdir( host_dir )
+			host_dir	= cluster_dir + '/' + host
+			dirlist		= os.listdir( host_dir )
 
 			for dir in dirlist:
@@ -1171,7 +1181,9 @@
 
 			last_serial = self.getLastRrdTimeSerial( host )
+
 			if last_serial:
 
 				metric_dir = cluster_dir + '/' + host + '/' + last_serial
+
 				if os.path.exists( metric_dir ):
 
@@ -1214,6 +1226,6 @@
 				self.myMetrics[ host ][ metric['name'] ] = [ ]
 		else:
-			self.myMetrics[ host ] = { }
-			self.myMetrics[ host ][ metric['name'] ] = [ ]
+			self.myMetrics[ host ]				= { }
+			self.myMetrics[ host ][ metric['name'] ]	= [ ]
 
 		# Push new metric onto stack
@@ -1232,6 +1244,6 @@
 		"""
 
-		update_list = [ ]
-		metric = None
+		update_list	= [ ]
+		metric		= None
 
 		while len( metriclist ) > 0:
@@ -1240,4 +1252,5 @@
 
 			if self.checkStoreMetric( host, metric ):
+
 				update_list.append( '%s:%s' %( metric['time'], metric['val'] ) )
 
@@ -1359,6 +1372,6 @@
 		"""Make a RRD location/path and filename"""
 
-		rrd_dir = '%s/%s/%s/%s' %( check_dir(ARCHIVE_PATH), self.cluster, host, timeserial )
-		rrd_file = '%s/%s.rrd' %( rrd_dir, metricname )
+		rrd_dir		= '%s/%s/%s/%s'	%( check_dir(ARCHIVE_PATH), self.cluster, host, timeserial )
+		rrd_file	= '%s/%s.rrd'	%( rrd_dir, metricname )
 
 		return rrd_dir, rrd_file
@@ -1414,7 +1427,7 @@
 			if metric['name'] == metricname:
 
-				period = self.determinePeriod( host, metric['time'] )	
-
-				archive_secs = ARCHIVE_HOURS_PER_RRD * (60 * 60)
+				period		= self.determinePeriod( host, metric['time'] )	
+
+				archive_secs	= ARCHIVE_HOURS_PER_RRD * (60 * 60)
 
 				if (int( metric['time'] ) - int( period ) ) > archive_secs:
@@ -1464,8 +1477,8 @@
 		if not os.path.exists( rrd_file ):
 
-			interval = self.config.getInterval( self.cluster )
-			heartbeat = 8 * int( interval )
-
-			params = [ ]
+			interval	= self.config.getInterval( self.cluster )
+			heartbeat	= 8 * int( interval )
+
+			params		= [ ]
 
 			params.append( '--step' )
@@ -1490,7 +1503,7 @@
 		debug_msg( 9, 'rrdupdate: using timeserial %s for %s/%s' %( timeserial, host, metricname ) )
 
-		rrd_dir, rrd_file = self.makeRrdPath( host, metricname, timeserial )
-
-		update_list = self.makeUpdateList( host, metriclist )
+		rrd_dir, rrd_file	= self.makeRrdPath( host, metricname, timeserial )
+
+		update_list		= self.makeUpdateList( host, metriclist )
 
 		if len( update_list ) > 0:
@@ -1551,6 +1564,6 @@
 
 	try:
-		torque_xml_thread = threading.Thread( None, myTorqueProcessor.run, 'torque_proc_thread' )
-		ganglia_xml_thread = threading.Thread( None, myGangliaProcessor.run, 'ganglia_proc_thread' )
+		torque_xml_thread	= threading.Thread( None, myTorqueProcessor.run, 'torque_proc_thread' )
+		ganglia_xml_thread	= threading.Thread( None, myGangliaProcessor.run, 'ganglia_proc_thread' )
 
 		torque_xml_thread.start()
