diff --git a/src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py b/src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py index 5b33f31feaa..60889323714 100755 --- a/src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py +++ b/src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py @@ -561,34 +561,39 @@ def _kickStuckJobs(self): """Reschedule jobs stuck in initialization status Rescheduled, Matched.""" message = "" - - checkTime = DiracTime.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 {JobStatus.MATCHED} status") - result = rescheduleJobs(jobIDs) + for status, stuckTime in ((JobStatus.MATCHED, self.matchedTime), (JobStatus.RESCHEDULED, self.rescheduledTime)): + checkTime = DiracTime.utcnow() - stuckTime * second + result = self.jobDB.selectJobs({"Status": status}, older=checkTime) if not result["OK"]: - message = f"Failed to reschedule jobs stuck in {JobStatus.MATCHED} status" - message += "\n" + result["Message"] - - checkTime = DiracTime.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 + 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 {JobStatus.RESCHEDULED} status") - result = rescheduleJobs(jobIDs) - if not result["OK"]: - message = f"Failed to reschedule jobs stuck in {JobStatus.RESCHEDULED} status" - message += "\n" + result["Message"] + 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 + result = rescheduleJobs( + [jobID], + source="StalledJobAgent", + jobDB=self.jobDB, + taskQueueDB=self.taskQueueDB, + jobLoggingDB=self.logDB, + ) + if not result["OK"]: + failedJobIDs.append(jobID) + if failedJobIDs: + message += f"Failed to reschedule jobs stuck in {status} status: {failedJobIDs}\n" if message: self.log.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 0f49936cdb2..b2481844df9 100644 --- a/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_StalledJobAgent.py +++ b/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_StalledJobAgent.py @@ -72,3 +72,25 @@ 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 + reschedule = mocker.patch( + "DIRAC.WorkloadManagementSystem.Agent.StalledJobAgent.rescheduleJobs", return_value={"OK": True} + ) + + assert sja._kickStuckJobs()["OK"] + reschedule.assert_called_once() + assert reschedule.call_args.args == ([1],) + assert reschedule.call_args.kwargs["source"] == "StalledJobAgent" diff --git a/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py b/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py index b5e8cb005ad..97056e6972e 100755 --- a/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py +++ b/src/DIRAC/WorkloadManagementSystem/DB/JobDB.py @@ -405,16 +405,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: - cmd = "UPDATE Jobs SET LastUpdateTime=UTC_TIMESTAMP() WHERE JobID=%s" - args = [str(jobID)] - if myDate: - cmd += " AND LastUpdateTime < %s" - args.append(myDate) - return self._update(cmd, args=args) + if update and myDate: + cmd = "UPDATE Jobs SET LastUpdateTime=UTC_TIMESTAMP() WHERE JobID=%s AND LastUpdateTime < %s" + return self._update(cmd, args=[str(jobID), myDate]) else: return res @@ -474,7 +470,9 @@ def setJobAttributes(self, jobID, attrNames, attrValues, update=False, myDate=No statusDone = False 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")) @@ -507,12 +505,14 @@ def setJobAttributes(self, jobID, attrNames, attrValues, update=False, myDate=No return self._update(cmd, args=args) - 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 @@ -543,6 +543,8 @@ def setJobsMajorStatus(self, jIDList, candidateStatus, force=False): newStatuses[jID] = nextState cmd = "INSERT INTO Jobs (JobID, Status) VALUES (%s, %s) ON DUPLICATE KEY UPDATE Status=VALUES(Status)" + if update: + cmd += ", LastUpdateTime=UTC_TIMESTAMP()" return self._updatemany(cmd, data=newStatuses.items()) def setJobStatus(self, jobID, status="", minorStatus="", applicationStatus="", force=False):