feat (WMS): Improved logging in executor infrastructure

Author: chrisburr

src/DIRAC/Core/Base/ExecutorMindHandler.py

@@ -124,6 +124,7 @@ def __sendTask(self, taskId, taskObj, eId, eType):
         taskStub = result["Value"]
         result = self.srv_msgCreate("ProcessTask")
         if not result["OK"]:
+            gLogger.error("Failed to create message for", "%s %r" % (taskId, result))
             return result
         msgObj = result["Value"]
         msgObj.taskId = taskId

src/DIRAC/Core/DISET/MessageClient.py

@@ -83,14 +83,12 @@ def __cbDisconnect(self, trid):
         self.__trid = False
         try:
             self.__transport.close()
-        except BaseException:
+        except Exception:
             pass
         for cb in self.__specialCallbacks["drop"]:
             try:
                 cb(self)
-            except SystemExit:
-                raise
-            except BaseException:
+            except Exception:
                 gLogger.exception("Exception while processing disconnect callbacks")
 
     def __cbRecvMsg(self, trid, msgObj):
@@ -107,7 +105,7 @@ def __cbRecvMsg(self, trid, msgObj):
                 # If no specific callback but a generic one, return the generic one
                 if msgName not in self.__callbacks:
                     return result
-            except BaseException:
+            except Exception:
                 gLogger.exception("Exception while processing callbacks", msgObj.getName())
         if msgName not in self.__callbacks:
             return S_ERROR("Unexpected message")
@@ -117,7 +115,7 @@ def __cbRecvMsg(self, trid, msgObj):
                 gLogger.error("Callback for message does not return S_OK/S_ERROR", msgName)
                 return S_ERROR("No response")
             return result
-        except BaseException:
+        except Exception:
             gLogger.exception("Exception while processing callbacks", msgName)
         return S_ERROR("No response")
 
@@ -128,6 +126,7 @@ def sendMessage(self, msgObj):
         if not self.__trid:
             result = self.connect()
             if not result["OK"]:
+                gLogger.error("Failed connect for sending", "%r" % msgObj)
                 return result
         return self.__msgBroker.sendMessage(self.__trid, msgObj)
 

src/DIRAC/Core/DISET/private/MessageBroker.py

@@ -165,12 +165,8 @@ def __listenAutoReceiveConnections(self):
 
             try:
                 events = sel.select(timeout=1)
-            except (socket.error, select.error):
-                # TODO: When can this happen?
-                time.sleep(0.001)
-                continue
-            except Exception as e:
-                gLogger.exception("Exception while selecting persistent connections", lException=e)
+            except Exception:
+                gLogger.exception("Exception while selecting persistent connections")
                 continue
 
             for key, event in events:
@@ -205,7 +201,7 @@ def __receiveMsgDataAndQueue(self, trid):
 
     def __processIncomingData(self, trid, receivedResult):
         # If keep alive, return OK
-        if "keepAlive" in receivedResult and receivedResult["keepAlive"]:
+        if receivedResult.get("keepAlive"):
             return S_OK()
         # If idle read return
         self.__trInOutLock.acquire()

src/DIRAC/Core/Utilities/ExecutorDispatcher.py

@@ -426,7 +426,7 @@ def addExecutor(self, eId, eTypes, maxTasks=1):
             self.__fillExecutors(eType)
 
     def removeExecutor(self, eId):
-        self.__log.verbose("Removing executor %s" % eId)
+        self.__log.info("Removing executor %s" % eId)
         self.__executorsLock.acquire()
         try:
             if eId not in self.__idMap:
@@ -845,6 +845,8 @@ def __msgTaskToExecutor(self, taskId, eId, eType):
             errMsg = "Send task callback did not send back an S_OK/S_ERROR structure"
             self.__log.fatal(errMsg)
             raise ValueError(errMsg)
+        if not result["OK"]:
+            self.__log.error("Failed to cbSendTask", "%r" % result)
 
 
 class UnrecoverableTaskException(Exception):