Merge pull request #5724 from DIRACGridBot/cherry-pick-2-cdd7b018f-integration

[sweep:integration] Fixes for setting the Jobs Status

Author: fstagni

src/DIRAC/Core/Utilities/Time.py

@@ -160,8 +160,7 @@ def toString(myDate=None):
       and day carries the sign.
     To keep internal consistency we are using:
       [hour]:[min]:[sec]:[microsec]
-      where min, sec, microsec are alwys positive intergers and hour carries the
-      sign.
+      where min, sec, microsec are always positive integers and hour carries the sign.
     """
     if isinstance(myDate, _dateTimeType):
         return str(myDate)

src/DIRAC/FrameworkSystem/Service/SystemAdministratorHandler.py

@@ -309,7 +309,7 @@ def export_updateSoftware(self, version):
                 stderr=subprocess.PIPE,
                 universal_newlines=True,
                 check=False,
-                timeout=300,
+                timeout=600,
             )
             if r.returncode != 0:
                 stderr = [x for x in r.stderr.split("\n") if not x.startswith("Extracting : ")]

src/DIRAC/Interfaces/API/Dirac.py

@@ -56,6 +56,7 @@
 from DIRAC.DataManagementSystem.Client.DataManager import DataManager
 from DIRAC.Resources.Storage.StorageElement import StorageElement
 from DIRAC.Resources.Catalog.FileCatalog import FileCatalog
+from DIRAC.WorkloadManagementSystem.Client import JobStatus
 from DIRAC.WorkloadManagementSystem.Client.WMSClient import WMSClient
 from DIRAC.WorkloadManagementSystem.Client.SandboxStoreClient import SandboxStoreClient
 from DIRAC.WorkloadManagementSystem.Client.JobMonitoringClient import JobMonitoringClient
@@ -204,7 +205,11 @@ def retrieveRepositorySandboxes(self, requestedStates=None, destinationDirectory
             gLogger.warn("No repository is initialised")
             return S_OK()
         if requestedStates is None:
-            requestedStates = ["Done", "Failed", "Completed"]  # because users dont care about completed
+            requestedStates = [
+                JobStatus.DONE,
+                JobStatus.FAILED,
+                JobStatus.COMPLETED,
+            ]  # because users dont care about completed
         jobs = self.jobRepo.readRepository()["Value"]
         for jobID in sorted(jobs):
             jobDict = jobs[jobID]

src/DIRAC/WorkloadManagementSystem/Agent/JobAgent.py

@@ -341,7 +341,7 @@ def execute(self):
                         par_name=thisp, par_value=gConfig.getValue("/LocalSite/%s" % thisp, "Unknown"), sendFlag=False
                     )
 
-            jobReport.setJobStatus(status=JobStatus.MATCHED, minorStatus="Job Received by Agent", sendFlag=False)
+            jobReport.setJobStatus(minorStatus="Job Received by Agent", sendFlag=False)
             result_setupProxy = self._setupProxy(ownerDN, jobGroup)
             if not result_setupProxy["OK"]:
                 return self._rescheduleFailedJob(jobID, result_setupProxy["Message"], self.stopOnApplicationFailure)
@@ -526,7 +526,7 @@ def _checkInstallSoftware(self, jobID, jobParams, resourceParams, jobReport):
             self.log.verbose(msg)
             return S_OK(msg)
 
-        jobReport.setJobStatus(status=JobStatus.MATCHED, minorStatus="Installing Software", sendFlag=False)
+        jobReport.setJobStatus(minorStatus="Installing Software", sendFlag=False)
         softwareDist = jobParams["SoftwareDistModule"]
         self.log.verbose("Found VO Software Distribution module", ": %s" % (softwareDist))
         argumentsDict = {"Job": jobParams, "CE": resourceParams}
@@ -579,7 +579,7 @@ def _submitJob(
             return result
 
         wrapperFile = result["Value"]
-        jobReport.setJobStatus(status=JobStatus.MATCHED, minorStatus="Submitting To CE")
+        jobReport.setJobStatus(minorStatus="Submitting To CE")
 
         self.log.info("Submitting JobWrapper", "%s to %sCE" % (os.path.basename(wrapperFile), self.ceName))
 

src/DIRAC/WorkloadManagementSystem/Client/JobReport.py

@@ -36,7 +36,7 @@ def setJob(self, jobID):
         self.jobID = jobID
 
     def setJobStatus(self, status="", minorStatus="", applicationStatus="", sendFlag=True):
-        """Send job status information to the JobState service for jobID"""
+        """Accumulate and possibly send job status information to the JobState service"""
 
         timeStamp = Time.toString()
         # add job status record

src/DIRAC/WorkloadManagementSystem/Client/JobStatus.py

@@ -85,7 +85,7 @@ def __init__(self, state):
             FAILED: State(12, [RESCHEDULED, DELETED], defState=FAILED),
             DONE: State(11, [DELETED], defState=DONE),
             COMPLETED: State(10, [DONE, FAILED], defState=COMPLETED),
-            COMPLETING: State(9, [DONE, FAILED, COMPLETED], defState=COMPLETING),
+            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),
             RESCHEDULED: State(6, [WAITING, RECEIVED, DELETED], defState=RESCHEDULED),

src/DIRAC/WorkloadManagementSystem/Client/Matcher.py

@@ -9,7 +9,6 @@
 __RCSID__ = "$Id"
 
 import time
-import re
 
 from DIRAC import gLogger, convertToPy3VersionNumber
 
@@ -18,9 +17,9 @@
 from DIRAC.Core.Security import Properties
 from DIRAC.ConfigurationSystem.Client.Helpers import Registry
 from DIRAC.ConfigurationSystem.Client.Helpers.Operations import Operations
+from DIRAC.WorkloadManagementSystem.Client import JobStatus
 from DIRAC.WorkloadManagementSystem.Client.Limiter import Limiter
 from DIRAC.WorkloadManagementSystem.Client import PilotStatus
-
 from DIRAC.WorkloadManagementSystem.DB.TaskQueueDB import TaskQueueDB, singleValueDefFields, multiValueMatchFields
 from DIRAC.WorkloadManagementSystem.DB.PilotAgentsDB import PilotAgentsDB
 from DIRAC.WorkloadManagementSystem.DB.JobDB import JobDB
@@ -266,7 +265,7 @@ def _reportStatus(self, resourceDict, jobID):
         else:
             self.log.verbose("Set job attributes for jobID", jobID)
 
-        result = self.jlDB.addLoggingRecord(jobID, status="Matched", minorStatus="Assigned", source="Matcher")
+        result = self.jlDB.addLoggingRecord(jobID, status=JobStatus.MATCHED, minorStatus="Assigned", source="Matcher")
         if not result["OK"]:
             self.log.error(
                 "Problem reporting job status", "addLoggingRecord, jobID = %s: %s" % (jobID, result["Message"])

src/DIRAC/WorkloadManagementSystem/DB/JobDB.py

@@ -2015,7 +2015,7 @@ def setJobCommand(self, jobID, command, arguments=None):
         return self._update(req)
 
     #####################################################################################
-    def getJobCommand(self, jobID, status="Received"):
+    def getJobCommand(self, jobID, status=JobStatus.RECEIVED):
         """Get a command to be passed to the job together with the
         next heart beat
         """

src/DIRAC/WorkloadManagementSystem/JobWrapper/JobWrapper.py

@@ -194,7 +194,7 @@ def __init__(self, jobID=None, jobReport=None):
     #############################################################################
     def initialize(self, arguments):
         """Initializes parameters and environment for job."""
-        self.__report(status=JobStatus.RUNNING, minorStatus=JobMinorStatus.JOB_INITIALIZATION)
+        self.__report(status=JobStatus.RUNNING, minorStatus=JobMinorStatus.JOB_INITIALIZATION, sendFlag=True)
         self.log.info("Starting Job Wrapper Initialization for Job", self.jobID)
         self.jobArgs = arguments["Job"]
         self.log.verbose(self.jobArgs)
@@ -371,8 +371,8 @@ def execute(self):
                 self.log.verbose("%s = %s" % (nameEnv, valEnv))
 
         if os.path.exists(executable):
-            # it's in fact not yet running: it will be in few lines
-            self.__report(status=JobStatus.RUNNING, minorStatus=JobMinorStatus.APPLICATION, sendFlag=True)
+            # the actual executable is not yet running: it will be in few lines
+            self.__report(minorStatus=JobMinorStatus.APPLICATION, sendFlag=True)
             spObject = Subprocess(timeout=False, bufferLimit=int(self.bufferLimit))
             command = executable
             if jobArguments:
@@ -482,9 +482,7 @@ def execute(self):
                 self.__report(status=JobStatus.COMPLETING, minorStatus=JobMinorStatus.APP_ERRORS, sendFlag=True)
                 if status in (DErrno.EWMSRESC, DErrno.EWMSRESC & 255):  # the status will be truncated to 0xDE (222)
                     self.log.verbose("job will be rescheduled")
-                    self.__report(
-                        status=JobStatus.COMPLETING, minorStatus=JobMinorStatus.GOING_RESCHEDULE, sendFlag=True
-                    )
+                    self.__report(minorStatus=JobMinorStatus.GOING_RESCHEDULE, sendFlag=True)
                     return S_ERROR(DErrno.EWMSRESC, "Job will be rescheduled")
 
         else:
@@ -564,7 +562,9 @@ def __getCPUHMS(self, cpuTime):
     #############################################################################
     def resolveInputData(self):
         """Input data is resolved here using a VO specific plugin module."""
-        self.__report(status=JobStatus.RUNNING, minorStatus=JobMinorStatus.INPUT_DATA_RESOLUTION, sendFlag=True)
+        self.__report(
+            minorStatus=JobMinorStatus.INPUT_DATA_RESOLUTION, sendFlag=True
+        )  # if we are here, the status should be "Running"
 
         # What is this input data? - and exit if there's no input
         inputData = self.jobArgs["InputData"]
@@ -899,7 +899,7 @@ def __resolveOutputSandboxFiles(self, outputSandbox):
     def __transferOutputDataFiles(self, outputData, outputSE, outputPath):
         """Performs the upload and registration in the File Catalog(s)"""
         self.log.verbose("Uploading output data files")
-        self.__report(status=JobStatus.COMPLETING, minorStatus=JobMinorStatus.UPLOADING_OUTPUT_DATA)
+        self.__report(minorStatus=JobMinorStatus.UPLOADING_OUTPUT_DATA)  # the major status should be "Completing"
         self.log.info("Output data files %s to be uploaded to %s SE" % (", ".join(outputData), outputSE))
         missing = []
         uploaded = []
@@ -1091,7 +1091,7 @@ def transferInputSandbox(self, inputSandbox):
         sandboxFiles = []
         registeredISB = []
         lfns = []
-        self.__report(status=JobStatus.RUNNING, minorStatus=JobMinorStatus.DOWNLOADING_INPUT_SANDBOX)
+        self.__report(minorStatus=JobMinorStatus.DOWNLOADING_INPUT_SANDBOX)  # Should be in "Running" status
         if not isinstance(inputSandbox, (list, tuple)):
             inputSandbox = [inputSandbox]
         for isb in inputSandbox:
@@ -1118,21 +1118,19 @@ def transferInputSandbox(self, inputSandbox):
                     self.log.info("Downloading Input SandBox %s" % isb)
                     result = SandboxStoreClient().downloadSandbox(isb)
                     if not result["OK"]:
-                        self.__report(
-                            status=JobStatus.RUNNING, minorStatus=JobMinorStatus.FAILED_DOWNLOADING_INPUT_SANDBOX
-                        )
+                        self.__report(minorStatus=JobMinorStatus.FAILED_DOWNLOADING_INPUT_SANDBOX)
                         return S_ERROR("Cannot download Input sandbox %s: %s" % (isb, result["Message"]))
                     else:
                         self.inputSandboxSize += result["Value"]
 
         if lfns:
             self.log.info("Downloading Input SandBox LFNs, number of files to get", len(lfns))
-            self.__report(status=JobStatus.RUNNING, minorStatus=JobMinorStatus.DOWNLOADING_INPUT_SANDBOX_LFN)
+            self.__report(minorStatus=JobMinorStatus.DOWNLOADING_INPUT_SANDBOX_LFN)
             lfns = [fname.replace("LFN:", "").replace("lfn:", "") for fname in lfns]
             download = self.dm.getFile(lfns)
             if not download["OK"]:
                 self.log.warn(download)
-                self.__report(status=JobStatus.RUNNING, minorStatus=JobMinorStatus.FAILED_DOWNLOADING_INPUT_SANDBOX_LFN)
+                self.__report(minorStatus=JobMinorStatus.FAILED_DOWNLOADING_INPUT_SANDBOX_LFN)
                 return S_ERROR(download["Message"])
             failed = download["Value"]["Failed"]
             if failed:
@@ -1303,16 +1301,6 @@ def sendFailoverRequest(self):
         request.JobID = self.jobID
         request.SourceComponent = "Job_%s" % self.jobID
 
-        # JobReport part first
-        result = self.jobReport.generateForwardDISET()
-        if result["OK"]:
-            if isinstance(result["Value"], Operation):
-                self.log.info("Adding a job state update DISET operation to the request")
-                request.addOperation(result["Value"])
-        else:
-            self.log.warn("JobReportFailure", "Could not generate a forwardDISET operation: %s" % result["Message"])
-            self.log.warn("JobReportFailure", "The job won't fail, but the jobLogging info might be incomplete")
-
         # Failover transfer requests
         for storedOperation in self.failoverTransfer.request:
             request.addOperation(storedOperation)
@@ -1325,6 +1313,16 @@ def sendFailoverRequest(self):
             for storedOperation in requestStored:
                 request.addOperation(storedOperation)
 
+        # JobReport part
+        result = self.jobReport.generateForwardDISET()
+        if result["OK"]:
+            if isinstance(result["Value"], Operation):
+                self.log.info("Adding a job state update DISET operation to the request")
+                request.addOperation(result["Value"])
+        else:
+            self.log.warn("JobReportFailure", "Could not generate a forwardDISET operation: %s" % result["Message"])
+            self.log.warn("JobReportFailure", "The job won't fail, but the jobLogging info might be incomplete")
+
         if len(request):
             # The request is ready, send it now
             isValid = RequestValidator().validate(request)

src/DIRAC/WorkloadManagementSystem/Service/JobStateUpdateHandler.py

@@ -57,14 +57,12 @@ def initializeHandler(cls, svcInfoDict):
     @classmethod
     def export_updateJobFromStager(cls, jobID, status):
         """Simple call back method to be used by the stager."""
-        if status == JobStatus.DONE:
+        if status == "Done":
             jobStatus = JobStatus.CHECKING
             minorStatus = "JobScheduling"
-        elif status == JobStatus.FAILED:
-            jobStatus = JobStatus.FAILED
-            minorStatus = "Staging input files failed"
         else:
-            return S_ERROR("updateJobFromStager: %s status not known." % status)
+            jobStatus = None
+            minorStatus = "Staging input files failed"
 
         infoStr = None
         trials = 10