diff --git a/src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py b/src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py index ca6766fcc42..b849eecf08d 100755 --- a/src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py +++ b/src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py @@ -515,34 +515,34 @@ def _kickStuckJobs(self): """Reschedule jobs stuck in initialization status Rescheduled, Matched""" message = "" + for status, stuckTime in ((JobStatus.MATCHED, self.matchedTime), (JobStatus.RESCHEDULED, self.rescheduledTime)): + checkTime = datetime.datetime.utcnow() - stuckTime * second + result = self.jobDB.selectJobs({"Status": status}, older=checkTime) + if not result["OK"]: + self.log.error("Failed to select jobs", result["Message"]) + return result - checkTime = datetime.datetime.utcnow() - self.matchedTime * second - result = self.jobDB.selectJobs({"Status": JobStatus.MATCHED}, older=checkTime) - if not result["OK"]: - self.log.error("Failed to select jobs", result["Message"]) - return result - - jobIDs = result["Value"] - if jobIDs: - self.log.info(f"Rescheduling {len(jobIDs)} jobs stuck in Matched status") - result = self.jobDB.rescheduleJobs(jobIDs) - if "FailedJobs" in result: - message = f"Failed to reschedule {len(result['FailedJobs'])} jobs stuck in Matched status" - - checkTime = datetime.datetime.utcnow() - self.rescheduledTime * second - result = self.jobDB.selectJobs({"Status": JobStatus.RESCHEDULED}, older=checkTime) - if not result["OK"]: - self.log.error("Failed to select jobs", result["Message"]) - return result - - jobIDs = result["Value"] - if jobIDs: - self.log.info(f"Rescheduling {len(jobIDs)} jobs stuck in Rescheduled status") - result = self.jobDB.rescheduleJobs(jobIDs) - if "FailedJobs" in result: + jobIDs = result["Value"] + if not jobIDs: + continue + self.log.info(f"Rescheduling {len(jobIDs)} jobs stuck in {status} status") + failedJobIDs = [] + for jobID in jobIDs: + # The job may have moved on since it was selected (e.g. a pilot picked it up), + # so check again just before rescheduling it + result = self.jobDB.selectJobs({"JobID": jobID, "Status": status}, older=checkTime) + if not result["OK"]: + failedJobIDs.append(jobID) + continue + if not result["Value"]: + self.log.info("Job is no longer stuck, not rescheduling", f"{jobID} ({status})") + continue + if not self.jobDB.rescheduleJob(jobID)["OK"]: + failedJobIDs.append(jobID) + if failedJobIDs: if message: message += "\n" - message += f"Failed to reschedule {len(result['FailedJobs'])} jobs stuck in Rescheduled status" + message += f"Failed to reschedule {len(failedJobIDs)} jobs stuck in {status} status" if message: return S_ERROR(message) diff --git a/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_StalledJobAgent.py b/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_StalledJobAgent.py index dbf4ce2fbf7..382cc5baeb1 100644 --- a/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_StalledJobAgent.py +++ b/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_StalledJobAgent.py @@ -49,3 +49,21 @@ def test__sjaFunctions(sja): assert sja._kickStuckJobs()["OK"] assert sja._failStalledJobs(0)["OK"] assert not sja._markStalledJobs(0)["OK"] + + +def test__kickStuckJobs_recheck(sja, mocker): + """Jobs that move on between the selection and the rescheduling are left alone""" + stillStuck = {1} + + def selectJobs(condDict, older=None): + if "JobID" in condDict: + return {"OK": True, "Value": [condDict["JobID"]] if condDict["JobID"] in stillStuck else []} + if condDict["Status"] == "Matched": + return {"OK": True, "Value": [1, 2]} + return {"OK": True, "Value": []} + + sja.jobDB.selectJobs.side_effect = selectJobs + sja.jobDB.rescheduleJob.return_value = {"OK": True} + + assert sja._kickStuckJobs()["OK"] + sja.jobDB.rescheduleJob.assert_called_once_with(1) diff --git a/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py b/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py index 39354ec45f4..fe5719f681a 100755 --- a/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py +++ b/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py @@ -467,13 +467,12 @@ def setJobAttribute(self, jobID, attrName, attrValue, update=False, myDate=None, if attrName == "Status": # Treat this update separately - res = self.setJobsMajorStatus([jobID], attrValue, force=force) + res = self.setJobsMajorStatus([jobID], attrValue, force=force, update=update and not myDate) if not res["OK"]: return res - if update: + if update and myDate: cmd = f"UPDATE Jobs SET LastUpdateTime=UTC_TIMESTAMP() WHERE JobID={jobID}" - if myDate: - cmd += f" AND LastUpdateTime < {myDate}" + cmd += f" AND LastUpdateTime < {myDate}" return self._update(cmd) else: return res @@ -542,7 +541,9 @@ def setJobAttributes(self, jobID, attrNames, attrValues, update=False, myDate=No if "Status" in attrNames: # Treat this update separately - res = self.setJobsMajorStatus(jIDList, attrValues[attrNames.index("Status")], force=force) + res = self.setJobsMajorStatus( + jIDList, attrValues[attrNames.index("Status")], force=force, update=update and not myDate + ) if not res["OK"]: return res attrValues.pop(attrNames.index("Status")) @@ -567,12 +568,14 @@ def setJobAttributes(self, jobID, attrNames, attrValues, update=False, myDate=No return self._update(cmd) - def setJobsMajorStatus(self, jIDList, candidateStatus, force=False): + def setJobsMajorStatus(self, jIDList, candidateStatus, force=False, update=False): """ Sets jobs major status, considering the JobStateMachine result :param list jIDList: list of one or more job IDs :param str candidateStatus: candidate major Status + :param bool update: also refresh the LastUpdateTime stamp in the same statement, so that + the new status is never visible together with the old time stamp """ # get the current statuses of the jobs @@ -614,6 +617,8 @@ def setJobsMajorStatus(self, jIDList, candidateStatus, force=False): cmd += ",".join(ns) cmd += " ON DUPLICATE KEY UPDATE Status=VALUES(Status)" + if update: + cmd += ", LastUpdateTime=UTC_TIMESTAMP()" return self._update(cmd)