refactor: remove use of JobMonitoringClient from JobStatus

Author: fstagni

src/DIRAC/Interfaces/API/Dirac.py

@@ -46,6 +46,7 @@
 from DIRAC.WorkloadManagementSystem.Client.JobMonitoringClient import JobMonitoringClient
 from DIRAC.WorkloadManagementSystem.Client.SandboxStoreClient import SandboxStoreClient
 from DIRAC.WorkloadManagementSystem.Client.WMSClient import WMSClient
+from DIRAC.WorkloadManagementSystem.Utilities.jobAdministration import _filterJobStateTransition
 
 
 def parseArguments(args):
@@ -1450,10 +1451,13 @@ def deleteJob(self, jobID):
         # Remove any job IDs that can't change to the Killed or Deleted states
         filteredJobs = set()
         for filterState in (JobStatus.KILLED, JobStatus.DELETED):
-            filterRes = JobStatus.filterJobStateTransition(jobIDs, filterState)
-            if not filterRes["OK"]:
-                return filterRes
-            filteredJobs.update(filterRes["Value"])
+            # get a dictionary of jobID:status
+            res = JobMonitoringClient().getJobsStatus(jobIDs)
+            if not res["OK"]:
+                return res
+            js = {k: v["Status"] for k, v in res["Value"].items()}
+            # then filter
+            filteredJobs.update(_filterJobStateTransition(js, filterState))
 
         return WMSClient(useCertificates=self.useCertificates).deleteJob(list(filteredJobs))
 
@@ -1480,11 +1484,13 @@ def rescheduleJob(self, jobID):
             return ret
         jobIDs = ret["Value"]
 
-        # Remove any job IDs that can't change to the rescheduled state
-        filterRes = JobStatus.filterJobStateTransition(jobIDs, JobStatus.RESCHEDULED)
-        if not filterRes["OK"]:
-            return filterRes
-        jobIDsToReschedule = filterRes["Value"]
+        # get a dictionary of jobID:status
+        res = JobMonitoringClient().getJobsStatus(jobIDs)
+        if not res["OK"]:
+            return res
+        js = {k: v["Status"] for k, v in res["Value"].items()}
+        # then filter
+        jobIDsToReschedule = _filterJobStateTransition(js, JobStatus.RESCHEDULED)
 
         return WMSClient(useCertificates=self.useCertificates).rescheduleJob(jobIDsToReschedule)
 
@@ -1510,10 +1516,13 @@ def killJob(self, jobID):
         # Remove any job IDs that can't change to the Killed or Deleted states
         filteredJobs = set()
         for filterState in (JobStatus.KILLED, JobStatus.DELETED):
-            filterRes = JobStatus.filterJobStateTransition(jobIDs, filterState)
-            if not filterRes["OK"]:
-                return filterRes
-            filteredJobs.update(filterRes["Value"])
+            # get a dictionary of jobID:status
+            res = JobMonitoringClient().getJobsStatus(jobIDs)
+            if not res["OK"]:
+                return res
+            js = {k: v["Status"] for k, v in res["Value"].items()}
+            # then filter
+            filteredJobs.update(_filterJobStateTransition(js, filterState))
 
         return WMSClient(useCertificates=self.useCertificates).killJob(list(filteredJobs))
 

src/DIRAC/WorkloadManagementSystem/Client/JobStatus.py

@@ -2,9 +2,7 @@
 This module contains constants and lists for the possible job states.
 """
 
-from DIRAC import S_OK
 from DIRAC.Core.Utilities.StateMachine import State, StateMachine
-from DIRAC.WorkloadManagementSystem.Client.JobMonitoringClient import JobMonitoringClient
 
 #:
 SUBMITTING = "Submitting"
@@ -95,26 +93,3 @@ def __init__(self, state):
             RECEIVED: State(1, [SCOUTING, CHECKING, STAGING, WAITING, FAILED, DELETED, KILLED], defState=RECEIVED),
             SUBMITTING: State(0, [RECEIVED, CHECKING, DELETED, KILLED], defState=SUBMITTING),  # initial state
         }
-
-
-def filterJobStateTransition(jobIDs, candidateState):
-    """Given a list of jobIDs, return a list that are allowed to transition
-    to the given candidate state.
-    """
-    allowedJobs = []
-
-    if not isinstance(jobIDs, list):
-        jobIDs = [jobIDs]
-
-    res = JobMonitoringClient().getJobsStatus(jobIDs)
-    if not res["OK"]:
-        return res
-
-    for jobID in jobIDs:
-        if jobID in res["Value"]:
-            curState = res["Value"][jobID]["Status"]
-            stateRes = JobsStateMachine(curState).getNextState(candidateState)
-            if stateRes["OK"]:
-                if stateRes["Value"] == candidateState:
-                    allowedJobs.append(jobID)
-    return S_OK(allowedJobs)

src/DIRAC/WorkloadManagementSystem/Utilities/jobAdministration.py

@@ -84,17 +84,11 @@ def kill_delete_jobs(right, validJobList, nonauthJobList=[], force=False):
         jobStates = result["Value"]
 
         # Get the jobs allowed to transition to the Killed state
-        filterRes = _filterJobStateTransition(jobStates, JobStatus.KILLED)
-        if not filterRes["OK"]:
-            return filterRes
-        killJobList.extend(filterRes["Value"])
+        killJobList.extend(_filterJobStateTransition(jobStates, JobStatus.KILLED))
 
         if right == RIGHT_DELETE:
             # Get the jobs allowed to transition to the Deleted state
-            filterRes = _filterJobStateTransition(jobStates, JobStatus.DELETED)
-            if not filterRes["OK"]:
-                return filterRes
-            deleteJobList.extend(filterRes["Value"])
+            deleteJobList.extend(_filterJobStateTransition(jobStates, JobStatus.DELETED))
 
         for jobID in killJobList:
             result = _killJob(jobID, force=force)
@@ -141,4 +135,4 @@ def _filterJobStateTransition(jobStates, candidateState):
         if stateRes["OK"]:
             if stateRes["Value"] == candidateState:
                 allowedJobs.append(js[0])
-    return S_OK(allowedJobs)
+    return allowedJobs