fix(resources): condor command line call compatible with v24

Author: antonio falabella

src/DIRAC/Resources/Computing/BatchSystems/Condor.py

@@ -6,6 +6,7 @@
 from __future__ import print_function
 from __future__ import absolute_import
 from __future__ import division
+import json
 import re
 import tempfile
 import subprocess
@@ -87,63 +88,34 @@
 """
 
 
-def parseCondorStatus(lines, jobID):
+def getCondorStatus(jobMetadata):
     """parse the condor_q or condor_history output for the job status
 
-    :param lines: list of lines from the output of the condor commands, each line is a tuple of jobID, statusID, and holdReasonCode
-    :type lines: python:list
-    :param str jobID: jobID of condor job, e.g.: 123.53
+    :param jobMetadata: dict with job metadata
+    :type lines: dict[str, str | int]
     :returns: Status as known by DIRAC, and a reason if the job is being held
     """
-    jobID = str(jobID)
-
-    holdReason = ""
-    status = None
-    for line in lines:
-        l = line.strip().split()
-
-        # Make sure the job ID exists
-        if len(l) < 1 or l[0] != jobID:
-            continue
-
-        # Make sure the status is present and is an integer
-        try:
-            status = int(l[1])
-        except (ValueError, IndexError):
-            break
-
-        # Stop here if the status is not held (5): result should be found in STATES_MAP
-        if status != 5:
-            break
-
-        # A job can be held for various reasons,
-        # we need to further investigate with the holdReasonCode & holdReasonSubCode
-        # Details in:
-        # https://htcondor.readthedocs.io/en/latest/classad-attributes/job-classad-attributes.html#HoldReasonCode
-
-        # By default, a held (5) job is defined as Aborted in STATES_MAP, but there might be some exceptions
-        status = 3
-        try:
-            holdReasonCode = l[2]
-            holdReasonSubcode = l[3]
-            holdReason = " ".join(l[4:])
-        except IndexError:
-            # This should not happen in theory
-            # Just set the status to unknown such as
-            status = None
-            holdReasonCode = "undefined"
-            holdReasonSubcode = "undefined"
-            break
-
-        # If holdReasonCode is 3 (The PERIODIC_HOLD expression evaluated to True. Or, ON_EXIT_HOLD was true)
-        # And subcode is HOLD_REASON_SUBCODE, then it means the job failed by itself, it needs to be marked as Failed
-        if holdReasonCode == "3" and holdReasonSubcode == HOLD_REASON_SUBCODE:
-            status = 5
-        # If holdReasonCode is 16 (Input files are being spooled), the job should be marked as Waiting
-        elif holdReasonCode == "16":
-            status = 1
-
-    return (STATES_MAP.get(status, "Unknown"), holdReason)
+    if jobMetadata["JobStatus"] != 5:
+        # If the job is not held, we can return the status directly
+        return (STATES_MAP.get(jobMetadata["JobStatus"], "Unknown"), "")
+
+    # A job can be held for various reasons,
+    # we need to further investigate with the holdReasonCode & holdReasonSubCode
+    # Details in:
+    # https://htcondor.readthedocs.io/en/latest/classad-attributes/job-classad-attributes.html#HoldReasonCode
+
+    # By default, a held (5) job is defined as Aborted in STATES_MAP, but there might be some exceptions
+    status = 3
+
+    # If holdReasonCode is 3 (The PERIODIC_HOLD expression evaluated to True. Or, ON_EXIT_HOLD was true)
+    # And subcode is HOLD_REASON_SUBCODE, then it means the job failed by itself, it needs to be marked as Failed
+    if jobMetadata["HoldReasonCode"] == 3 and jobMetadata["HoldReasonSubCode"] == HOLD_REASON_SUBCODE:
+        status = 5
+    # If holdReasonCode is 16 (Input files are being spooled), the job should be marked as Waiting
+    elif jobMetadata["HoldReasonCode"] == 16:
+        status = 1
+
+    return (STATES_MAP.get(status, "Unknown"), jobMetadata["HoldReason"])
 
 
 class Condor(object):
@@ -283,7 +255,6 @@ def killJob(self, **kwargs):
 
     def getJobStatus(self, **kwargs):
         """Get status of the jobs in the given list"""
-
         resultDict = {}
 
         MANDATORY_PARAMETERS = ["JobIDList"]
@@ -299,15 +270,12 @@ def getJobStatus(self, **kwargs):
             resultDict["Message"] = "Empty job list"
             return resultDict
 
-        user = kwargs.get("User")
-        if not user:
-            user = os.environ.get("USER")
-        if not user:
-            resultDict["Status"] = -1
-            resultDict["Message"] = "No user name"
-            return resultDict
+        # Prepare the command to get the status of the jobs
+        cmdJobs = " ".join(str(jobID) for jobID in jobIDList)
+        attributes = "ClusterId,ProcId,JobStatus,HoldReasonCode,HoldReasonSubCode,HoldReason"
 
-        cmd = "condor_q -submitter %s -af:j JobStatus HoldReasonCode HoldReasonSubCode HoldReason" % user
+        # Get the status of the jobs currently active
+        cmd = "condor_q %s -attributes %s -json" % (cmdJobs, attributes)
         sp = subprocess.Popen(
             shlex.split(cmd),
             stdout=subprocess.PIPE,
@@ -317,16 +285,15 @@ def getJobStatus(self, **kwargs):
         output, error = sp.communicate()
         status = sp.returncode
 
-        if status != 0:
+        if status != 0 or not output:
             resultDict["Status"] = status
             resultDict["Message"] = error
             return resultDict
 
-        qList = output.strip().split("\n")
+        jobMetadata = json.loads(output)
 
-        condorHistCall = (
-            "condor_history -af:j JobStatus HoldReasonCode HoldReasonSubCode HoldReason -submitter %s" % user
-        )
+        # Get the status of the jobs in the history
+        condorHistCall = "condor_history %s -attributes %s -json" % (cmdJobs, attributes)
         sp = subprocess.Popen(
             shlex.split(condorHistCall),
             stdout=subprocess.PIPE,
@@ -335,15 +302,26 @@ def getJobStatus(self, **kwargs):
         )
         output, _ = sp.communicate()
         status = sp.returncode
-        if status == 0:
-            for line in output.split("\n"):
-                qList.append(line)
+
+        if status != 0 or not output:
+            resultDict["Status"] = status
+            resultDict["Message"] = error
+            return resultDict
+
+        jobMetadata += json.loads(output)
 
         statusDict = {}
-        if len(qList):
-            for job in jobIDList:
-                job = str(job)
-                statusDict[job], _ = parseCondorStatus(qList, job)
+        # Build a set of job IDs found in jobMetadata
+        foundJobIDs = set()
+        for jobDict in jobMetadata:
+            jobID = "%s.%s" % (jobDict["ClusterId"], jobDict["ProcId"])
+            statusDict[jobID], _ = getCondorStatus(jobDict)
+            foundJobIDs.add(jobID)
+
+        # For job IDs not found, set status to "Unknown"
+        for jobID in jobIDList:
+            if str(jobID) not in foundJobIDs:
+                statusDict[str(jobID)] = "Unknown"
 
         # Final output
         status = 0
@@ -355,19 +333,30 @@ def getCEStatus(self, **kwargs):
         """Get the overall status of the CE"""
         resultDict = {}
 
-        user = kwargs.get("User")
-        if not user:
-            user = os.environ.get("USER")
-        if not user:
+        cmd = "condor_q -totals -json"
+        sp = subprocess.Popen(
+            shlex.split(cmd),
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            universal_newlines=True,
+        )
+        output, error = sp.communicate()
+        status = sp.returncode
+
+        if status != 0 or not output:
             resultDict["Status"] = -1
-            resultDict["Message"] = "No user name"
+            resultDict["Message"] = error
             return resultDict
 
-        waitingJobs = 0
-        runningJobs = 0
+        jresult = json.loads(output)
+        resultDict["Status"] = 0
+        resultDict["Waiting"] = jresult[0]["Idle"]
+        resultDict["Running"] = jresult[0]["Running"]
 
+        # We also need to check the hold jobs, some of them are actually waiting (e.g. for input files)
+        cmd = 'condor_q -json -constraint "JobStatus == 5" -attributes HoldReasonCode'
         sp = subprocess.Popen(
-            shlex.split("condor_q -submitter %s" % user),
+            shlex.split(cmd),
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             universal_newlines=True,
@@ -376,33 +365,17 @@ def getCEStatus(self, **kwargs):
         status = sp.returncode
 
         if status != 0:
-            if "no record" in output:
-                resultDict["Status"] = 0
-                resultDict["Waiting"] = waitingJobs
-                resultDict["Running"] = runningJobs
-                return resultDict
-            resultDict["Status"] = status
+            resultDict["Status"] = -1
             resultDict["Message"] = error
             return resultDict
 
-        if "no record" in output:
-            resultDict["Status"] = 0
-            resultDict["Waiting"] = waitingJobs
-            resultDict["Running"] = runningJobs
+        # If there are no held jobs, we can return the result
+        if not output:
             return resultDict
 
-        if output:
-            lines = output.split("\n")
-            for line in lines:
-                if not line.strip():
-                    continue
-                if " I " in line:
-                    waitingJobs += 1
-                elif " R " in line:
-                    runningJobs += 1
+        jresult = json.loads(output)
+        for job_metadata in jresult:
+            if job_metadata["HoldReasonCode"] == 16:
+                resultDict["Waiting"] += 1
 
-        # Final output
-        resultDict["Status"] = 0
-        resultDict["Waiting"] = waitingJobs
-        resultDict["Running"] = runningJobs
         return resultDict

src/DIRAC/Resources/Computing/HTCondorCEComputingElement.py

@@ -50,6 +50,7 @@
 
 import datetime
 import errno
+import json
 import os
 import subprocess
 import tempfile
@@ -63,7 +64,7 @@
 from DIRAC.Core.Utilities.List import breakListIntoChunks
 from DIRAC.Core.Utilities.Subprocess import systemCall
 from DIRAC.FrameworkSystem.private.authorization.utils.Tokens import writeToTokenFile
-from DIRAC.Resources.Computing.BatchSystems.Condor import HOLD_REASON_SUBCODE, parseCondorStatus, subTemplate
+from DIRAC.Resources.Computing.BatchSystems.Condor import HOLD_REASON_SUBCODE, getCondorStatus, subTemplate
 from DIRAC.Resources.Computing.ComputingElement import ComputingElement
 from DIRAC.WorkloadManagementSystem.Client import PilotStatus
 
@@ -400,45 +401,59 @@ def getJobStatus(self, jobIDList):
         if isinstance(jobIDList, str):
             jobIDList = [jobIDList]
 
+        self.tokenFile = None
         resultDict = {}
         condorIDs = {}
         # Get all condorIDs so we can just call condor_q and condor_history once
         for jobReference in jobIDList:
             jobReference = jobReference.split(":::")[0]
-            condorIDs[jobReference] = self._jobReferenceToCondorID(jobReference)
+            condorIDs[self._jobReferenceToCondorID(jobReference)] = jobReference
 
-        self.tokenFile = None
+        attributes = "ClusterId,ProcId,JobStatus,HoldReasonCode,HoldReasonSubCode,HoldReason"
 
         qList = []
-        for _condorIDs in breakListIntoChunks(condorIDs.values(), 100):
-            # This will return a list of 1245.75 3 undefined undefined undefined
+        for _condorIDs in breakListIntoChunks(condorIDs.keys(), 100):
             cmd = ["condor_q"]
             cmd.extend(self.remoteScheddOptions.strip().split(" "))
             cmd.extend(_condorIDs)
-            cmd.extend(["-af:j", "JobStatus", "HoldReasonCode", "HoldReasonSubCode", "HoldReason"])
+            cmd.extend(["-attributes", attributes])
+            cmd.extend(["-json"])
             result = self._executeCondorCommand(cmd, keepTokenFile=True)
             if not result["OK"]:
                 return result
 
-            qList.extend(result["Value"].split("\n"))
+            if result["Value"]:
+                qList.extend(json.loads(result["Value"]))
 
             condorHistCall = ["condor_history"]
             condorHistCall.extend(self.remoteScheddOptions.strip().split(" "))
             condorHistCall.extend(_condorIDs)
-            condorHistCall.extend(["-af:j", "JobStatus", "HoldReasonCode", "HoldReasonSubCode", "HoldReason"])
+            condorHistCall.extend(["-attributes", attributes])
+            condorHistCall.extend(["-json"])
             result = self._executeCondorCommand(cmd, keepTokenFile=True)
             if not result["OK"]:
                 return result
 
-            qList.extend(result["Value"].split("\n"))
+            if result["Value"]:
+                qList.extend(json.loads(result["Value"]))
 
-        for job, jobID in condorIDs.items():
-            jobStatus, reason = parseCondorStatus(qList, jobID)
+        foundJobIDs = set()
+        for jobMetadata in qList:
+            jobStatus, reason = getCondorStatus(jobMetadata)
+            condorId = f"{jobMetadata['ClusterId']}.{jobMetadata['ProcId']}"
+            jobReference = condorIDs.get(condorId)
 
             if jobStatus == PilotStatus.ABORTED:
-                self.log.verbose("Job", f"{jobID} held: {reason}")
+                self.log.verbose("Job", f"{jobReference} held: {reason}")
+
+            resultDict[jobReference] = jobStatus
+            foundJobIDs.add(jobReference)
 
-            resultDict[job] = jobStatus
+        # Check if we have any jobs that were not found in the condor_q or condor_history
+        for jobReference in condorIDs.values():
+            if jobReference not in foundJobIDs:
+                self.log.verbose("Job", f"{jobReference} not found in condor_q or condor_history")
+                resultDict[jobReference] = PilotStatus.UNKNOWN
 
         self.tokenFile = None