Merge pull request #8204 from fstagni/90_fixes14

[9.0] fix: getJobParameters expects a list

Author: chrisburr

src/DIRAC/MonitoringSystem/Service/WebAppHandler.py

@@ -188,10 +188,9 @@ def export_getJobPageSummaryWeb(self, selectDict, sortList, startItem, maxItems,
                 )
                 summaryJobList = validJobs
 
-            res = self.jobDB.getJobsAttributes(summaryJobList)
-            if not res["OK"]:
-                return res
-            return S_OK(strToIntDict(res["Value"]))
+            result = self.jobDB.getJobsAttributes(summaryJobList)
+            if not result["OK"]:
+                return result
 
             summaryDict = result["Value"]
             # If no jobs can be selected after the properties check
@@ -352,14 +351,9 @@ def export_getDistinctAttributeValues(cls, attribute, selectDict):
 
     @classmethod
     def export_getTransformationFilesSummaryWeb(cls, selectDict, sortList, startItem, maxItems):
-        selectColumns = (["TransformationID", "Status", "UsedSE", "TargetSE"],)
-        timeStamp = ("LastUpdate",)
-        statusColumn = ("Status",)
         fromDate = selectDict.get("FromDate", None)
         if fromDate:
             del selectDict["FromDate"]
-        # if not fromDate:
-        #  fromDate = last_update
         toDate = selectDict.get("ToDate", None)
         if toDate:
             del selectDict["ToDate"]
@@ -369,26 +363,21 @@ def export_getTransformationFilesSummaryWeb(cls, selectDict, sortList, startItem
         else:
             orderAttribute = None
         # Get the columns that match the selection
-        fcn = None
-        fcnName = "getTransformationFiles"
-        if hasattr(cls.transformationDB, fcnName) and callable(getattr(cls.transformationDB, fcnName)):
-            fcn = getattr(cls.transformationDB, fcnName)
-        if not fcn:
-            return S_ERROR(f"Unable to invoke gTransformationDB.{fcnName}, it isn't a member function")
-        res = fcn(condDict=selectDict, older=toDate, newer=fromDate, timeStamp=timeStamp, orderAttribute=orderAttribute)
+        res = cls.transformationDB.getTransformationFiles(
+            condDict=selectDict, older=toDate, newer=fromDate, timeStamp="LastUpdate", orderAttribute=orderAttribute
+        )
         if not res["OK"]:
             return res
-
-        # The full list of columns in contained here
         allRows = res["Value"]
+
         # Prepare the standard structure now within the resultDict dictionary
         resultDict = {}
-        # Create the total records entry
         resultDict["TotalRecords"] = len(allRows)
 
-        # Get the rows which are within the selected window
-        if resultDict["TotalRecords"] == 0:
+        if not allRows:
             return S_OK(resultDict)
+
+        # Get the rows which are within the selected window
         ini = startItem
         last = ini + maxItems
         if ini >= resultDict["TotalRecords"]:
@@ -403,25 +392,25 @@ def export_getTransformationFilesSummaryWeb(cls, selectDict, sortList, startItem
 
         # Create the ParameterNames entry
         resultDict["ParameterNames"] = list(selectedRows[0].keys())
-        # Find which element in the tuple contains the requested status
-        if statusColumn not in resultDict["ParameterNames"]:
-            return S_ERROR("Provided status column not present")
 
         # Generate the status dictionary
         statusDict = {}
         for row in selectedRows:
-            status = row[statusColumn]
+            status = row["Status"]
             statusDict[status] = statusDict.setdefault(status, 0) + 1
         resultDict["Extras"] = statusDict
 
         # Obtain the distinct values of the selection parameters
         res = cls.transformationDB.getTableDistinctAttributeValues(
-            "TransformationFiles", selectColumns, selectDict, older=toDate, newer=fromDate
+            "TransformationFiles",
+            ["TransformationID", "Status", "UsedSE", "TargetSE"],
+            selectDict,
+            older=toDate,
+            newer=fromDate,
         )
-        distinctSelections = zip(selectColumns, [])
-        if res["OK"]:
-            distinctSelections = res["Value"]
-        resultDict["Selections"] = distinctSelections
+        if not res["OK"]:
+            return res
+        resultDict["Selections"] = res["Value"]
 
         return S_OK(resultDict)
 

src/DIRAC/WorkloadManagementSystem/Agent/StalledJobAgent.py

@@ -254,7 +254,7 @@ def _failStalledJobs(self, jobID):
 
     def _getJobPilotStatus(self, jobID):
         """Get the job pilot status."""
-        result = getJobParameters(jobID, "Pilot_Reference")
+        result = getJobParameters([jobID], "Pilot_Reference")
         if not result["OK"]:
             return result
         pilotReference = result["Value"].get("Pilot_Reference")
@@ -389,11 +389,11 @@ def _sendAccounting(self, jobID):
             if lastHeartBeatTime is not None and lastHeartBeatTime > endTime:
                 endTime = lastHeartBeatTime
 
-            result = getJobParameters(jobID, "CPUNormalizationFactor")
-            if not result["OK"] or not result["Value"]:
+            result = getJobParameters([jobID], "CPUNormalizationFactor")
+            if not result["OK"] or not result["Value"] or not result["Value"].get("CPUNormalizationFactor"):
                 self.log.error(
                     "Error getting Job Parameter CPUNormalizationFactor, setting 0",
-                    result.get("Message", "No such value"),
+                    result.get("Message"),
                 )
                 cpuNormalization = 0.0
             else:

src/DIRAC/WorkloadManagementSystem/DB/JobParametersDB.py

@@ -69,7 +69,7 @@ def _createIndex(self, indexName: str) -> None:
                 raise RuntimeError(result["Message"])
             self.log.always("Index created:", indexName)
 
-    def getJobParameters(self, jobIDs: int | list[int], vo: str, paramList=None) -> dict:
+    def getJobParameters(self, jobIDs: int | list[int], vo: str, paramList: list[str] | None = None) -> dict:
         """Get Job Parameters defined for jobID.
           Returns a dictionary with the Job Parameters.
           If paramList is empty - all the parameters are returned.

src/DIRAC/WorkloadManagementSystem/Service/JobMonitoringHandler.py

@@ -247,7 +247,7 @@ def export_getJobParameter(self, jobID, parName):
         :param str/int jobID: one single Job ID
         :param str parName: one single parameter name
         """
-        res = self.elasticJobParametersDB.getJobParameters(int(jobID), self.vo, [parName])
+        res = getJobParameters([int(jobID)], parName, self.vo or "")
         if not res["OK"]:
             return res
         return S_OK(res["Value"].get(int(jobID), {}))
@@ -266,7 +266,7 @@ def export_getJobOptParameters(cls, jobID):
     def export_getJobParameters(self, jobIDs, parName=None):
         """
         :param str/int/list jobIDs: one single job ID or a list of them
-        :param str parName: one single parameter name, a list or None (meaning all of them)
+        :param str parName: one single parameter name, or None (meaning all of them)
         """
         if not isinstance(jobIDs, list):
             jobIDs = [jobIDs]

tests/Integration/FutureClient/WorkloadManagement/Test_JobMonitoring.py

@@ -108,12 +108,6 @@ def test_getJobSite():
     pytest.skip()
 
 
-def test_getJobStats():
-    # JobMonitoringClient().getJobStats(attribute: str, selectDict: dict)
-    method = JobMonitoringClient().getJobStats
-    pytest.skip()
-
-
 def test_getJobSummary():
     # JobMonitoringClient().getJobSummary(jobID: int)
     method = JobMonitoringClient().getJobSummary