fix: fixes to StatesAccountingAgent for new fields

Author: fstagni

src/DIRAC/AccountingSystem/Client/Types/BaseAccountingType.py

@@ -97,7 +97,9 @@ def setValuesFromDict(self, dataDict):
         if errKeys:
             return S_ERROR(f"Key(s) {', '.join(errKeys)} are not valid")
         for key in dataDict:
-            self.setValueByKey(key, dataDict[key])
+            res = self.setValueByKey(key, dataDict[key])
+            if not res["OK"]:
+                return res
         return S_OK()
 
     def getValue(self, key):

src/DIRAC/WorkloadManagementSystem/Agent/StatesAccountingAgent.py

@@ -32,17 +32,22 @@ class StatesAccountingAgent(AgentModule):
     __summaryKeyFieldsMapping = [
         "Status",
         "Site",
-        "User",
-        "UserGroup",
+        "Owner",
+        "OwnerGroup",
         "JobGroup",
         "VO",
         "JobType",
         "ApplicationStatus",
         "MinorStatus",
     ]
-    __summaryDefinedFields = [("ApplicationStatus", "unset"), ("MinorStatus", "unset")]
-    __summaryValueFieldsMapping = ["Jobs", "Reschedules"]
-    __renameFieldsMapping = {"JobType": "JobSplitType"}
+    __summaryValueFieldsMapping = ["JobCount", "RescheduleSum"]
+    __renameFieldsMapping = {
+        "Owner": "User",
+        "OwnerGroup": "UserGroup",
+        "JobType": "JobSplitType",
+        "JobCount": "Jobs",
+        "RescheduleSum": "Reschedules",
+    }
 
     # PilotsHistory fields
     __pilotsMapping = ["GridSite", "ComputingElement", "GridType", "Status", "VO", "NumOfPilots"]
@@ -98,7 +103,7 @@ def execute(self):
                     f"{result['Message']}: won't commit PilotsHistory at this cycle",
                 )
 
-            values = result["Value"][1]
+            values = result["Value"]
             for record in values:
                 rD = {}
                 for iP, _ in enumerate(self.__pilotsMapping):
@@ -114,25 +119,20 @@ def execute(self):
 
         # WMSHistory to Monitoring or Accounting
         self.log.info(f"Committing WMSHistory to {'and '.join(self.jobMonitoringOption)} backend")
-        result = JobDB()._query("SELECT * FROM JobsHistorySummary")
+        result = JobDB()._query(
+            f"SELECT {','.join(self.__summaryKeyFieldsMapping + self.__summaryValueFieldsMapping)} FROM JobsHistorySummary ORDER BY {','.join(self.__summaryKeyFieldsMapping)}"
+        )
         if not result["OK"]:
             self.log.error("Can't get the JobDB summary", f"{result['Message']}: won't commit WMSHistory at this cycle")
             return S_ERROR()
 
-        values = result["Value"][1]
-
+        values = result["Value"]
         now = datetime.datetime.utcnow()
         self.log.info("Start sending WMSHistory records")
         for record in values:
             rD = {}
-            for fV in self.__summaryDefinedFields:
-                rD[fV[0]] = fV[1]
-            for iP, _ in enumerate(self.__summaryKeyFieldsMapping):
-                fieldName = self.__summaryKeyFieldsMapping[iP]
+            for iP, fieldName in enumerate(self.__summaryKeyFieldsMapping + self.__summaryValueFieldsMapping):
                 rD[self.__renameFieldsMapping.get(fieldName, fieldName)] = record[iP]
-            record = record[len(self.__summaryKeyFieldsMapping) :]
-            for iP, _ in enumerate(self.__summaryValueFieldsMapping):
-                rD[self.__summaryValueFieldsMapping[iP]] = int(record[iP])
 
             for backend in self.datastores:
                 if backend.lower() == "monitoring":
@@ -143,6 +143,8 @@ def execute(self):
                     self.datastores["Monitoring"].addRecord(rD)
 
                 elif backend.lower() == "accounting":
+                    rD.pop("VO")  # Remove VO field for Accounting
+                    rD.pop("ApplicationStatus")  # Remove ApplicationStatus for Accounting
                     acWMS = WMSHistory()
                     acWMS.setStartTime(now)
                     acWMS.setEndTime(now)