sweep: #5744 WMS job status fixes

Author: fstagni

src/DIRAC/WorkloadManagementSystem/Agent/JobAgent.py

@@ -36,6 +36,7 @@
 from DIRAC.WorkloadManagementSystem.Client.MatcherClient import MatcherClient
 from DIRAC.WorkloadManagementSystem.Client.PilotManagerClient import PilotManagerClient
 from DIRAC.WorkloadManagementSystem.Client.JobManagerClient import JobManagerClient
+from DIRAC.WorkloadManagementSystem.Client.JobStateUpdateClient import JobStateUpdateClient
 from DIRAC.WorkloadManagementSystem.Client.JobReport import JobReport
 from DIRAC.WorkloadManagementSystem.Client import JobStatus
 from DIRAC.WorkloadManagementSystem.Utilities.Utils import createJobWrapper
@@ -315,7 +316,7 @@ def execute(self):
             self.log.debug("After %sCE submitJob()" % (self.ceName))
         except Exception as subExcept:  # pylint: disable=broad-except
             self.log.exception("Exception in submission", "", lException=subExcept, lExcInfo=True)
-            result = self._rescheduleFailedJob(jobID, "Job processing failed with exception")
+            result = self._rescheduleFailedJob(jobID, "Job processing failed with exception", direct=True)
             return self._finish(result["Message"], self.stopOnApplicationFailure)
 
         return S_OK("Job Agent cycle complete")
@@ -751,20 +752,23 @@ def _finish(self, message, stop=True):
         return S_OK(message)
 
     #############################################################################
-    def _rescheduleFailedJob(self, jobID, message):
+    def _rescheduleFailedJob(self, jobID, message, direct=False):
         """
         Set Job Status to "Rescheduled" and issue a reschedule command to the Job Manager
         """
 
         self.log.warn("Failure ==> rescheduling", "(during %s)" % (message))
 
-        jobReport = JobReport(int(jobID), "JobAgent@%s" % self.siteName)
-
-        # Setting a job parameter does not help since the job will be rescheduled,
-        # instead set the status with the cause and then another status showing the
-        # reschedule operation.
-
-        jobReport.setJobStatus(status=JobStatus.RESCHEDULED, applicationStatus=message, sendFlag=True)
+        if direct:
+            JobStateUpdateClient().setJobStatus(
+                int(jobID), status=JobStatus.RESCHEDULED, applicationStatus=message, source="JobAgent@%s", force=True
+            )
+        else:
+            jobReport = JobReport(int(jobID), "JobAgent@%s" % self.siteName)
+            # Setting a job parameter does not help since the job will be rescheduled,
+            # instead set the status with the cause and then another status showing the
+            # reschedule operation.
+            jobReport.setJobStatus(status=JobStatus.RESCHEDULED, applicationStatus=message, sendFlag=True)
 
         self.log.info("Job will be rescheduled")
         result = JobManagerClient().rescheduleJob(jobID)

src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py

@@ -239,7 +239,7 @@ def _failStalledJobs(self, jobID):
                     return result
                 site = result["Value"]
                 if site in self.stalledJobsToRescheduleSites:
-                    return self._updateJobStatus(jobID, JobStatus.RESCHEDULED, minorStatus=setFailed)
+                    return self._updateJobStatus(jobID, JobStatus.RESCHEDULED, minorStatus=setFailed, force=True)
 
             return self._updateJobStatus(jobID, JobStatus.FAILED, minorStatus=setFailed)
 
@@ -314,7 +314,7 @@ def _getLatestUpdateTime(self, job):
             return S_OK(latestUpdate)
 
     #############################################################################
-    def _updateJobStatus(self, job, status, minorStatus=None):
+    def _updateJobStatus(self, job, status, minorStatus=None, force=False):
         """This method updates the job status in the JobDB"""
 
         if not self.am_getOption("Enable", True):
@@ -323,7 +323,7 @@ def _updateJobStatus(self, job, status, minorStatus=None):
         toRet = S_OK()
 
         self.log.debug("self.jobDB.setJobAttribute(%s,'Status','%s',update=True)" % (job, status))
-        result = self.jobDB.setJobAttribute(job, "Status", status, update=True)
+        result = self.jobDB.setJobAttribute(job, "Status", status, update=True, force=force)
         if not result["OK"]:
             self.log.error("Failed setting Status", "%s for job %d: %s" % (status, job, result["Message"]))
             toRet = result

src/DIRAC/WorkloadManagementSystem/Client/JobStatus.py

@@ -80,9 +80,9 @@ def __init__(self, state):
             COMPLETED: State(10, [DONE, FAILED], defState=COMPLETED),
             COMPLETING: State(9, [DONE, FAILED, COMPLETED, STALLED, KILLED], defState=COMPLETING),
             STALLED: State(8, [RUNNING, FAILED, KILLED], defState=STALLED),
-            RUNNING: State(7, [STALLED, DONE, FAILED, COMPLETING, KILLED, RECEIVED], defState=RUNNING),
+            RUNNING: State(7, [STALLED, DONE, FAILED, RESCHEDULED, COMPLETING, KILLED, RECEIVED], defState=RUNNING),
             RESCHEDULED: State(6, [WAITING, RECEIVED, DELETED], defState=RESCHEDULED),
-            MATCHED: State(5, [RUNNING, FAILED, KILLED], defState=MATCHED),
+            MATCHED: State(5, [RUNNING, FAILED, RESCHEDULED, KILLED], defState=MATCHED),
             WAITING: State(4, [MATCHED, RESCHEDULED, DELETED], defState=WAITING),
             STAGING: State(3, [WAITING, FAILED, KILLED], defState=STAGING),
             CHECKING: State(2, [STAGING, WAITING, RESCHEDULED, FAILED, DELETED], defState=CHECKING),

src/DIRAC/WorkloadManagementSystem/JobWrapper/JobWrapper.py

@@ -1518,8 +1518,7 @@ def rescheduleFailedJob(jobID, minorStatus, jobReport=None):
 
         gLogger.info("Job will be rescheduled after exception during execution of the JobWrapper")
 
-        jobManager = JobManagerClient()
-        result = jobManager.rescheduleJob(int(jobID))
+        result = JobManagerClient().rescheduleJob(int(jobID))
         if not result["OK"]:
             gLogger.warn(result["Message"])
             if "Maximum number of reschedulings is reached" in result["Message"]: