Changeset 792 for branches

Timestamp:

04/05/13 14:35:28 (10 years ago)

Author:

ramonb

Message:

• prevent that 1 (or more) failed SQL insert/updates hangs the entire job store thread
• check success/failure of SQL actions and perform commit/rollback accordingly

File:

• branches/0.4/jobarchived/jobarchived.py (modified) (10 diffs)

branches/0.4/jobarchived/jobarchived.py

@@ -340 +340 @@
 
     def Commit(self):
-        self.SQL.commit()
+
+        return self.SQL.commit()
+
+    def Rollback( self ):
+
+        return self.SQL.rollback()
 
 class DataSQLStore:
@@ -364 +369 @@
 
     def setDatabase(self, statement):
+
         ret = self.doDatabase('set', statement)
         return ret
         
     def getDatabase(self, statement):
+
         ret = self.doDatabase('get', statement)
         return ret
+
+    def doCommit( self ):
+
+        return self.dbc.Commit()
+
+    def doRollback( self ):
+
+        return self.dbc.Rollback()
 
     def doDatabase(self, type, statement):
@@ -377 +392 @@
             if type == 'set':
                 result = self.dbc.Set( statement )
-                self.dbc.Commit()
             elif type == 'get':
                 result = self.dbc.Get( statement )
@@ -383 +397 @@
         except DBError, detail:
             operation = statement.split(' ')[0]
-            debug_msg( 0, 'FATAL ERROR: ' +operation+ ' on database failed while doing ['+statement+'] full msg: '+str(detail) )
-            sys.exit(1)
+            debug_msg( 0, 'ERROR: ' +operation+ ' on database failed while doing ['+statement+'] full msg: '+str(detail) )
+            return False
 
         debug_msg( 10, 'doDatabase(): result: %s' %(result) )
@@ -392 +406 @@
 
         id = self.getDatabase( "SELECT job_id,node_id FROM job_nodes WHERE job_id = '%s' AND node_id = '%s'" %(job_id, node_id) )
+        if not id:
+            return False
+
         if len( id ) > 0:
 
             if len( id[0] ) > 0 and id[0] != '':
             
-                return 1
-
-        return 0
+                return True
+
+        return False
 
     def getNodeId( self, hostname ):
@@ -440 +457 @@
         if not self.getJobId( job_id ):
 
-            self.mutateJob( 'insert', job_id, jobattrs ) 
+            return self.mutateJob( 'insert', job_id, jobattrs ) 
         else:
-            self.mutateJob( 'update', job_id, jobattrs )
+            return self.mutateJob( 'update', job_id, jobattrs )
 
     def mutateJob( self, action, job_id, jobattrs ):
@@ -507 +524 @@
         if action == 'insert':
 
-            self.setDatabase( "INSERT INTO jobs ( %s ) VALUES ( %s )" %( insert_col_str, insert_val_str ) )
+            db_ret = self.setDatabase( "INSERT INTO jobs ( %s ) VALUES ( %s )" %( insert_col_str, insert_val_str ) )
 
         elif action == 'update':
 
-            self.setDatabase( "UPDATE jobs SET %s WHERE job_id='%s'" %(update_str, job_id) )
+            db_ret = self.setDatabase( "UPDATE jobs SET %s WHERE job_id='%s'" %(update_str, job_id) )
 
         if len( ids ) > 0:
             self.addJobNodes( job_id, ids )
+
+        return db_ret
 
     def addNodes( self, hostnames, domain ):
@@ -547 +566 @@
     def storeJobInfo( self, jobid, jobattrs ):
 
-        self.addJob( jobid, jobattrs )
+        return self.addJob( jobid, jobattrs )
 
     def checkStaleJobs( self ):
@@ -947 +966 @@
         debug_msg( 1, 'job_xml_thread(): Found %s updated jobs.' %len(self.jobs_to_store) )
 
+        failed_store = [ ]
+        succes_store = [ ]
+
         if len( self.jobs_to_store ) > 0:
 
             debug_msg( 1, 'job_xml_thread(): Storing jobs to database..' )
 
-            while len( self.jobs_to_store ) > 0:
+            for n in range( 0, len(self.jobs_to_store ) ):
+
+                if len( self.jobs_to_store ) == 0:
+                    break
 
                 jobid = self.jobs_to_store.pop( 0 )
 
-                self.ds.storeJobInfo( jobid, self.jobAttrs[ jobid ] )
+                db_ok = self.ds.storeJobInfo( jobid, self.jobAttrs[ jobid ] )
+
+                if not db_ok:
+
+                    self.ds.doRollback()
+                    failed_store.append( jobid )
+                    continue
+
+                self.ds.doCommit()
+                succes_store.append( jobid )
 
                 if not jobid in jobs_finished:
@@ -969 +1003 @@
                     del self.jobAttrs[ jobid ]
 
-            debug_msg( 1, 'job_xml_thread(): Done storing.' )
+            result_str = 'succesfully stored: %s jobs' %str(len(succes_store))
+
+            if len( failed_store ) > 0:
+                result_str = result_str + ' - failed to store: %s jobs - deferred to next interval' %str(len(failed_store))
+
+            debug_msg( 1, 'job_xml_thread(): Done storing. %s' %result_str )
 
         else: