Index: trunk/jobarchived/jobarchived.conf-dist
===================================================================
--- trunk/jobarchived/jobarchived.conf-dist	(revision 293)
+++ trunk/jobarchived/jobarchived.conf-dist	(revision 295)
@@ -77,4 +77,11 @@
 JOB_SQL_DBASE			: localhost/jobarchive
 
+# Timeout for jobs in archive
+#
+# Assume job has already finished while jobarchived was not running
+# after this amount of hours: the it will be finished anyway in the database
+#
+JOB_TIMEOUT			: 168
+
 # Location of rrdtool binary
 #
Index: trunk/jobarchived/jobarchived.py
===================================================================
--- trunk/jobarchived/jobarchived.py	(revision 293)
+++ trunk/jobarchived/jobarchived.py	(revision 295)
@@ -104,5 +104,5 @@
 	cfg.read( filename )
 
-	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
+	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, JOB_TIMEOUT
 
 	ARCHIVE_PATH		= cfg.get( 'DEFAULT', 'ARCHIVE_PATH' )
@@ -134,4 +134,6 @@
 
 	JOB_SQL_DBASE		= cfg.get( 'DEFAULT', 'JOB_SQL_DBASE' )
+
+	JOB_TIMEOUT		= cfg.getint( 'DEFAULT', 'JOB_TIMEOUT' )
 
 	DAEMONIZE		= cfg.getboolean( 'DEFAULT', 'DAEMONIZE' )
@@ -367,4 +369,64 @@
 		self.addJob( jobid, jobattrs )
 
+	def checkStaleJobs( self ):
+
+		q = "SELECT * from jobs WHERE job_status != 'F'"
+
+		r = self.getDatabase( q )
+
+		if len( r ) == 0:
+
+			return None
+
+		cleanjobs	= [ ]
+		timeoutjobs	= [ ]
+
+		jobtimeout_sec	= JOB_TIMEOUT * (60 * 60)
+		cur_time	= time.time()
+
+		for row in r:
+
+			job_id			= row[0]
+			job_requested_time	= row[4]
+			job_status		= row[7]
+			job_start_timestamp	= row[8]
+
+			if job_status == 'Q' or not job_start_timestamp:
+
+				cleanjobs.append( job_id )
+
+			else:
+
+				start_timestamp = int( job_start_timestamp )
+
+				if ( cur_time - start_timestamp ) > jobtimeout_sec:
+
+					if job_requested_time:
+
+						rtime_epoch	= reqtime2epoch( job_requested_time )
+					else:
+						rtime_epoch	= None
+					
+					timeoutjobs.append( (job_id, job_start_timestamp, rtime_epoch) )
+
+		debug_msg( 1, 'Found ' + str( len( cleanjobs ) ) + ' stale jobs in database: deleting entries' )
+
+		for j in cleanjobs:
+
+			q = "DELETE FROM jobs WHERE job_id = '" + str( j ) + "'"
+			self.setDatabase( q )
+
+		debug_msg( 1, 'Found ' + str( len( timeoutjobs ) ) + ' timed out jobs in database: closing entries' )
+
+		for j in timeoutjobs:
+
+			( i, s, r )		= j
+
+			if r:
+				new_end_timestamp	= int( s ) + r
+
+			q = "UPDATE jobs SET job_stop_timestamp = '" + str( new_end_timestamp ) + "' WHERE job_id = '" + str(i) + "'"
+			self.setDatabase( q )
+
 class RRDMutator:
 	"""A class for performing RRD mutations"""
@@ -457,5 +519,5 @@
 	"""Main class for processing XML and acting with it"""
 
-	def __init__( self, XMLSource ):
+	def __init__( self, XMLSource, DataStore ):
 		"""Setup initial XML connection and handlers"""
 
@@ -463,5 +525,5 @@
 		#self.myXMLSource	= self.myXMLGatherer.getFileObject()
 		self.myXMLSource	= XMLSource
-		self.myXMLHandler	= TorqueXMLHandler()
+		self.myXMLHandler	= TorqueXMLHandler( DataStore )
 		self.myXMLError		= XMLErrorHandler()
 
@@ -494,7 +556,8 @@
 	jobAttrs = { }
 
-	def __init__( self ):
-
-		self.ds			= DataSQLStore( JOB_SQL_DBASE.split( '/' )[0], JOB_SQL_DBASE.split( '/' )[1] )
+	def __init__( self, datastore ):
+
+		#self.ds			= DataSQLStore( JOB_SQL_DBASE.split( '/' )[0], JOB_SQL_DBASE.split( '/' )[1] )
+		self.ds			= datastore
 		self.jobs_processed	= [ ]
 		self.jobs_to_store	= [ ]
@@ -594,5 +657,5 @@
 
 			for jobid in self.jobs_to_store:
-				if self.jobAttrs[ jobid ]['status'] in [ 'R', 'Q', 'F' ]:
+				if self.jobAttrs[ jobid ]['status'] in [ 'R', 'F' ]:
 
 					self.ds.storeJobInfo( jobid, self.jobAttrs[ jobid ] )
@@ -648,9 +711,13 @@
 	"""Parse Ganglia's XML"""
 
-	def __init__( self, config ):
+	def __init__( self, config, datastore ):
 		"""Setup initial variables and gather info on existing rrd archive"""
 
 		self.config	= config
 		self.clusters	= { }
+		self.ds		= datastore
+		debug_msg( 1, 'Checking database..' )
+		self.ds.checkStaleJobs()
+		debug_msg( 1, 'Check done.' )
 		debug_msg( 1, 'Checking existing toga rrd archive..' )
 		self.gatherClusters()
@@ -930,5 +997,5 @@
 	"""Main class for processing XML and acting with it"""
 
-	def __init__( self, XMLSource ):
+	def __init__( self, XMLSource, DataStore ):
 		"""Setup initial XML connection and handlers"""
 
@@ -938,5 +1005,6 @@
 		#self.myXMLSource	= self.myXMLGatherer.getFileObject()
 		self.myXMLSource	= XMLSource
-		self.myXMLHandler	= GangliaXMLHandler( self.config )
+		self.ds			= DataStore
+		self.myXMLHandler	= GangliaXMLHandler( self.config, self.ds )
 		self.myXMLError		= XMLErrorHandler()
 
@@ -1567,7 +1635,8 @@
 
 	myXMLSource		= XMLGatherer( ARCHIVE_XMLSOURCE.split( ':' )[0], ARCHIVE_XMLSOURCE.split( ':' )[1] )
-
-	myTorqueProcessor	= TorqueXMLProcessor( myXMLSource )
-	myGangliaProcessor	= GangliaXMLProcessor( myXMLSource )
+	myDataStore		= DataSQLStore( JOB_SQL_DBASE.split( '/' )[0], JOB_SQL_DBASE.split( '/' )[1] )
+
+	myTorqueProcessor	= TorqueXMLProcessor( myXMLSource, myDataStore )
+	myGangliaProcessor	= GangliaXMLProcessor( myXMLSource, myDataStore )
 
 	try:
@@ -1611,4 +1680,14 @@
 	return directory
 
+def reqtime2epoch( rtime ):
+
+	(hours, minutes, seconds )	= rtime.split( ':' )
+
+	etime	= int(seconds)
+	etime	= etime + ( int(minutes) * 60 )
+	etime	= etime + ( int(hours) * 60 * 60 )
+
+	return etime
+
 def debug_msg( level, msg ):
 	"""Only print msg if correct levels"""
