Index: trunk/ippToPsps/jython/ipptopspsdb.py =================================================================== --- trunk/ippToPsps/jython/ipptopspsdb.py (revision 32460) +++ trunk/ippToPsps/jython/ipptopspsdb.py (revision 32461) @@ -20,39 +20,5 @@ super(IppToPspsDb, self).__init__(logger, doc, "ipptopspsdatabase") - ''' - Creates a new batch - ''' - def createNewBatch(self, batchType, stageID, survey, dvoDb, datastoreProduct): - - sql = "INSERT INTO batch ( \ - batch_type, \ - stage_id, \ - survey, \ - dvo_db, \ - datastore_product \ - ) VALUES ( \ - '" + batchType + "', \ - " + str(stageID) + ", \ - '" + survey + "', \ - '" + dvoDb + "', \ - '" + datastoreProduct + "' \ - )" - - self.execute(sql) - - sql = "SELECT MAX(batch_id) FROM batch" - - batchID = -1; - - try: - rs = self.executeQuery(sql) - rs.first() - batchID = rs.getInt(1) - except: - self.logger.exception("Unable to get batch ID") - - self.logger.debug("Created new batch in ippToPsps database with batchID = %d" % batchID) - - return batchID; + self.MAX_FAILS = 3 ''' @@ -132,5 +98,5 @@ Returns a list of processed batch IDs that have not yet been loaded to the ODM ''' - def getBatchIDsUnloadedToODM(self, epoch, dvoGpc1Label): + def getUnloadedBatchIDs(self, epoch, dvoGpc1Label, batchType): sql = "SELECT DISTINCT batch_id \ @@ -139,4 +105,5 @@ AND dvo_db = '" + dvoGpc1Label + "' \ AND loaded_to_datastore = 1 \ + AND batch_type = '" + batchType + "' \ AND loaded_to_ODM = 0" @@ -157,7 +124,7 @@ ''' - Returns a list of processed batch IDs that have not failed to load and are not yet merged, for this epoch and dvo label - ''' - def getUnfinishedBatchIDs(self, epoch, dvoGpc1Label): + Returns a list of processed batch IDs that are merge_worthy, but not yet merged, for this epoch and dvo label + ''' + def getUnmergedBatchIDs(self, epoch, dvoGpc1Label, batchType): sql = "SELECT DISTINCT batch_id \ @@ -165,20 +132,19 @@ WHERE timestamp > '" + epoch + "' \ AND dvo_db = '" + dvoGpc1Label + "' \ - AND loaded_to_datastore = 1 \ - AND merged != 1 \ - AND loaded_to_ODM != -1" - - - ids = [] - try: - rs = self.executeQuery(sql) - while (rs.next()): - ids.append(rs.getInt(1)) - except: - self.logger.exception("Can't query for unfinished batch ids in ipptopsps Db") - - rs.close() - - self.logger.debug("Found %d unfinished items" % len(ids)) + AND loaded_to_ODM = 1 \ + AND batch_type = '" + batchType + "' \ + AND merged != 1" + + ids = [] + try: + rs = self.executeQuery(sql) + while (rs.next()): + ids.append(rs.getInt(1)) + except: + self.logger.exception("Can't query for unmerged batch ids in ipptopsps Db") + + rs.close() + + self.logger.debug("Found %d un-merged items" % len(ids)) return ids @@ -300,5 +266,5 @@ ''' def getStages(self): - return ['processed', 'loaded_to_datastore', 'loaded_to_ODM', 'merge_worthy', 'merged', 'deleted_datastore', 'deleted_dxlayer', 'deleted_local'] + return ['processed', 'loaded_to_datastore', 'loaded_to_ODM', 'merge_worthy', 'deleted_datastore', 'merged', 'deleted_dxlayer', 'deleted_local'] ''' @@ -328,5 +294,5 @@ def getTimeOfLastBatchPublished(self, batchType, epoch, dvoGpc1Label): - minutes = None + seconds = None sql = "SELECT TIMESTAMPDIFF(SECOND, timestamp, now()) \ @@ -427,4 +393,27 @@ ''' + Is someone processing this item right now? Checks whether this was attempted within the last 2 HOUR + ''' + def processingNow(self, batchType, stage_id, epoch, dvoGpc1Label): + + sql = "SELECT COUNT(*) \ + FROM batch \ + WHERE stage_id = " + str(stage_id) + " \ + AND batch_type = '" + batchType + "' \ + AND dvo_db = '" + dvoGpc1Label + "' \ + AND timestamp BETWEEN now() - INTERVAL 2 HOUR AND now()" + + try: + rs = self.executeQuery(sql) + rs.first() + if rs.getInt(1) > 0: + self.logger.errorPair(str(stage_id) + " is already being processed", "skipping") + return True + else: + return False + except: + self.logger.exception("Unable to check whether this batch is being processed") + + ''' Have we already processed and published this batch? ''' @@ -454,5 +443,5 @@ Has this stage_id consistently failed to process? ''' - def consistentlyFailed(self, batchType, stage_id, epoch, dvoGpc1Label, count): + def consistentlyFailed(self, batchType, stage_id, epoch, dvoGpc1Label): sql = "SELECT COUNT(*) \ @@ -467,6 +456,6 @@ rs = self.executeQuery(sql) rs.first() - if rs.getInt(1) >= count: - self.logger.errorPair(str(stage_id) + " has failed %d times" % count, "skipping") + if rs.getInt(1) >= self.MAX_FAILS: + self.logger.errorPair(str(stage_id) + " has failed %d times" % self.MAX_FAILS, "skipping") return True else: @@ -475,4 +464,77 @@ self.logger.exception("Unable to check whether this batch has consistently failed") + ''' + Returns a list of stage_ids that have failed the max number of times + ''' + def getConsistentlyFailedIDs(self, batchType, epoch, dvoGpc1Label): + + sql = "SELECT DISTINCT stage_id, COUNT(stage_id) AS numoccur \ + FROM batch \ + WHERE batch_type = '" + batchType + "' \ + AND timestamp > '" + epoch + "' \ + AND dvo_db = '" + dvoGpc1Label + "' \ + AND processed = -1 \ + GROUP BY stage_id HAVING numoccur >= " + str(self.MAX_FAILS) + + ids = [] + try: + rs = self.executeQuery(sql) + while (rs.next()): + ids.append(rs.getInt(1)) + except: + self.logger.exception("Can't query for consistantly failed ids in ipptopsps Db") + + rs.close() + + self.logger.debug("Found %d consistantly failed stage_ids for batch type='%s'" % (len(ids), batchType)) + + return ids + + ''' + Locks the batch table. This will wait if the lock is held by someone else + ''' + def lockBatchTable(self): + self.lockTable("batch") + + ''' + Creates a new batch. This is done with a locked table in order that no two clients can start work on the same item + ''' + def createNewBatch(self, batchType, stageID, survey, epoch, dvoDb, datastoreProduct, force): + + batchID = -1; + + if force or \ + (not self.alreadyProcessed(batchType, stageID, epoch, dvoDb) \ + and not self.processingNow(batchType, stageID, epoch, dvoDb) \ + and not self.consistentlyFailed(batchType, stageID, epoch, dvoDb)): + + sql = "INSERT INTO batch ( \ + batch_type, \ + stage_id, \ + survey, \ + dvo_db, \ + datastore_product \ + ) VALUES ( \ + '" + batchType + "', \ + " + str(stageID) + ", \ + '" + survey + "', \ + '" + dvoDb + "', \ + '" + datastoreProduct + "' \ + )" + + self.execute(sql) + + sql = "SELECT MAX(batch_id) FROM batch" + + try: + rs = self.executeQuery(sql) + rs.first() + batchID = rs.getInt(1) + self.logger.debug("Created new batch in ippToPsps database with batchID = %d" % batchID) + except: + self.logger.exception("Unable to get batch ID") + batchID = -1 + + return batchID;