Repository: dmwm/wmcore
Issue: JobAccountant terminated but the heart still beats

spotted https://cmsweb.cern.ch/reqmgr2/fetch?rid=vlimant_ACDC0_task_TOP-PhaseIITDRFall17wmLHEGS-00017__v1_T_171121_151106_5101 this by chance, in running since 5 days and 70 jobs created, nothing has ran as far as wmstats/gwmsmon says.
I could not spot anything specifically  wrong with the WQE https://cmsweb.cern.ch/couchdb/workqueue/_design/WorkQueue/_rewrite/element/d7affcd8abd8c053781fbd3688802b2a 

Changed files:
src/python/WMCore/WorkerThreads/BaseWorkerThread.py, src/python/WMComponent/JobAccountant/AccountantWorker.py

diff --git a/src/python/WMComponent/JobAccountant/AccountantWorker.py b/src/python/WMComponent/JobAccountant/AccountantWorker.py
index eb4785e95..338bf9ce5 100644
--- a/src/python/WMComponent/JobAccountant/AccountantWorker.py
+++ b/src/python/WMComponent/JobAccountant/AccountantWorker.py
@@ -106,9 +106,9 @@ class AccountantWorker(WMConnectionBase):
                                                     database=config.ACDC.database)
 
         jobDBurl = sanitizeURL(config.JobStateMachine.couchurl)['url']
-        jobDBName = config.JobStateMachine.couchDBName
-        jobCouchdb = CouchServer(jobDBurl)
-        self.fwjrCouchDB = jobCouchdb.connectDatabase("%s/fwjrs" % jobDBName)
+        self.jobDBName = config.JobStateMachine.couchDBName
+        self.jobCouchdb = CouchServer(jobDBurl)
+        self.fwjrCouchDB = None
         self.localWMStats = WMStatsWriter(config.TaskArchiver.localWMStatsURL, appName="WMStatsAgent")
 
         # Hold data for later commital
@@ -590,6 +590,9 @@ class AccountantWorker(WMConnectionBase):
 
         Associate a logArchive output to its parent job
         """
+        if self.fwjrCouchDB is None:
+            self.fwjrCouchDB = self.jobCouchdb.connectDatabase("%s/fwjrs" % self.jobDBName)
+
         inputFileList = fwkJobReport.getAllInputFiles()
         requestName = task.split('/')[1]
         keys = []
diff --git a/src/python/WMCore/WorkerThreads/BaseWorkerThread.py b/src/python/WMCore/WorkerThreads/BaseWorkerThread.py
index c0508ca0d..098941119 100644
--- a/src/python/WMCore/WorkerThreads/BaseWorkerThread.py
+++ b/src/python/WMCore/WorkerThreads/BaseWorkerThread.py
@@ -95,6 +95,29 @@ class BaseWorkerThread(object):
         """
         logging.error("Calling algorithm on BaseWorkerThread: Override me!")
 
+    def setUpHeartbeat(self, myThread):
+        # heartbeat needed to be called in self.initInThread
+        # to get the right name but before the self.setup
+        if hasattr(self.component.config, "Agent"):
+            self.useHeartbeat = getattr(self.component.config.Agent, "useHeartbeat", True)
+            self.workerName = myThread.getName()
+
+        if self.useHeartbeat:
+            self.heartbeatAPI.registerWorker(self.workerName)
+        return
+
+    def setUpLogDB(self, myThread):
+        # setup logDB
+        if hasattr(self.component.config, "General") and \\
+                hasattr(self.component.config.General, "central_logdb_url") and \\
+                hasattr(self.component.config, "Agent"):
+            from WMCore.Services.LogDB.LogDB import LogDB
+            myThread.logdbClient = LogDB(self.component.config.General.central_logdb_url,
+                                         self.component.config.Agent.hostName, logger=logging)
+        else:
+            myThread.logdbClient = None
+        return
+
     def initInThread(self, parameters):
         """
         Called when the thread is actually running in its own thread. Performs
@@ -121,6 +144,9 @@ class BaseWorkerThread(object):
         logging.info("Initialising default transaction")
         myThread.transaction = Transaction(myThread.dbi)
 
+        self.setUpHeartbeat(myThread)
+        self.setUpLogDB(myThread)
+
         # Call worker setup
         self.setup(parameters)
         myThread.transaction.commit()
@@ -130,6 +156,7 @@ class BaseWorkerThread(object):
         Thread entry point; handles synchronisation with run and terminate
         conditions
         """
+        errMsg = ""
         try:
             msg = "Initialising worker thread %s" % str(self)
             logging.info(msg)
@@ -140,26 +167,8 @@ class BaseWorkerThread(object):
             msg = "Worker thread %s started" % str(self)
             logging.info(msg)
 
-            # heartbeat needed to be called after self.initInThread
-            # to get the right name
             myThread = threading.currentThread()
 
-            if hasattr(self.component.config, "Agent"):
-                self.useHeartbeat = getattr(self.component.config.Agent, "useHeartbeat", True)
-                self.workerName = myThread.getName()
-
-            if hasattr(self.component.config, "General") and \\
-                    hasattr(self.component.config.General, "central_logdb_url") and \\
-                    hasattr(self.component.config, "Agent"):
-                from WMCore.Services.LogDB.LogDB import LogDB
-                myThread.logdbClient = LogDB(self.component.config.General.central_logdb_url,
-                                             self.component.config.Agent.hostName, logger=logging)
-            else:
-                myThread.logdbClient = None
-
-            if self.useHeartbeat:
-                self.heartbeatAPI.registerWorker(self.workerName)
-
             # Run event loop while termination is not flagged
             while not self.notifyTerminate.isSet():
                 # Check manager hasn't paused threads
@@ -190,19 +199,17 @@ class BaseWorkerThread(object):
                         except Exception as ex:
                             if myThread.transaction.transaction is not None:
                                 myThread.transaction.rollback()
-                            msg = "Error in worker algorithm (1):\\nBacktrace:\\n "
-                            msg += (" %s %s" % (str(self), str(ex)))
+                            errMsg = "Error in worker algorithm (1):\\nBacktrace:\\n "
+                            errMsg += (" %s %s" % (str(self), str(ex)))
                             stackTrace = traceback.format_tb(sys.exc_info()[2], None)
                             for stackFrame in stackTrace:
-                                msg += stackFrame
-                            logging.error(msg)
+                                errMsg += stackFrame
+                            logging.error(errMsg)
                             # force entire component to terminate
                             try:
                                 self.component.prepareToStop()
                             except Exception as ex:
                                 logging.error("Failed to halt component after worker crash: %s", str(ex))
-                            if self.useHeartbeat:
-                                self.heartbeatAPI.updateWorkerError(self.workerName, msg)
                             raise ex
                         # Put the thread to sleep
                         self.sleepThread()
@@ -217,6 +224,13 @@ class BaseWorkerThread(object):
             for stackFrame in stackTrace:
                 msg += stackFrame
             logging.error(msg)
+            # send heartbeat message that thread is terminated
+            if self.useHeartbeat:
+                msg = errMsg or msg
+                try:
+                    self.heartbeatAPI.updateWorkerError(self.workerName, errMsg)
+                except Exception as ex:
+                    logging.error("Heartbeat error update failed %s" % str(ex))
 
         # Indicate to manager that thread is done
         self.terminateCallback(threading.currentThread().name)