feat: JobDB snapshot: use a summary table for jobs in final state

Author: fstagni

src/DIRAC/WorkloadManagementSystem/Agent/StatesAccountingAgent.py

@@ -49,6 +49,8 @@ def initialize(self):
         """Standard initialization"""
         # This agent will always loop every 15 minutes
         self.am_setOption("PollingTime", 900)
+        # This agent will restart every 24 hours
+        self.am_setOption("MaxCycles", 96)
 
         # Check whether to send to Monitoring or Accounting or both
         self.jobMonitoringOption = Operations().getMonitoringBackends(monitoringType="WMSHistory")
@@ -65,6 +67,11 @@ def initialize(self):
         if "Monitoring" in self.pilotMonitoringOption:
             self.pilotReporter = MonitoringReporter(monitoringType="PilotsHistory", failoverQueueName=messageQueue)
 
+        res = JobDB().fillJobsHistorySummary()
+        if not res["OK"]:
+            self.log.error("Could not fill the JobDB summary", res["Message"])
+            return S_ERROR()
+
         self.__jobDBFields = []
         for field in self.__summaryKeyFieldsMapping:
             if field == "User":

src/DIRAC/WorkloadManagementSystem/DB/JobDB.py

@@ -20,7 +20,7 @@
 from DIRAC.Core.Utilities.ClassAd.ClassAdLight import ClassAd
 from DIRAC.Core.Utilities.Decorators import deprecated
 from DIRAC.Core.Utilities.DErrno import EWMSJMAN, EWMSSUBM, cmpError
-from DIRAC.Core.Utilities.ReturnValues import S_ERROR, S_OK, convertToReturnValue, returnValueOrRaise, SErrorException
+from DIRAC.Core.Utilities.ReturnValues import S_ERROR, S_OK, SErrorException, convertToReturnValue, returnValueOrRaise
 from DIRAC.FrameworkSystem.Client.Logger import contextLogger
 from DIRAC.ResourceStatusSystem.Client.SiteStatus import SiteStatus
 from DIRAC.WorkloadManagementSystem.Client import JobMinorStatus, JobStatus
@@ -1510,14 +1510,53 @@ def setJobCommandStatus(self, jobID, command, status):
         return self._update(f"UPDATE JobCommands SET Status={status} WHERE JobID={jobID} AND Command={command}")
 
     #####################################################################################
+    def fillJobsHistorySummary(self):
+        """Fill the JobsHistorySummary table with the summary of the jobs in a final state"""
+        
+        defString = "Status, Site, Owner, OwnerGroup, JobGroup, JobType, ApplicationStatus, MinorStatus"
+        valuesString = "COUNT(JobID), SUM(RescheduleCounter)"
+        final_states = "', '".join(JobStatus.JOB_FINAL_STATES + JobStatus.JOB_REALLY_FINAL_STATES)
+        final_states = f"'{final_states}'"
+
+        query = (
+            f"INSERT INTO JobsHistorySummary SELECT {defString}, {valuesString} "
+            f"FROM Jobs WHERE Status IN ({final_states}) AND LastUpdateTime < UTC_DATE() "
+            f"GROUP BY {defString}"
+        )
+        result = self._update(query)
+        if not result["OK"]:
+            return result
+        return S_OK(result["Value"])
+    
     def getSummarySnapshot(self, requestedFields=False):
         """Get the summary snapshot for a given combination"""
         if not requestedFields:
             requestedFields = ["Status", "MinorStatus", "Site", "Owner", "OwnerGroup", "JobGroup"]
         valueFields = ["COUNT(JobID)", "SUM(RescheduleCounter)"]
         defString = ", ".join(requestedFields)
         valueString = ", ".join(valueFields)
-        result = self._query(f"SELECT {defString}, {valueString} FROM Jobs GROUP BY {defString}")
+        final_states = "', '".join(JobStatus.JOB_FINAL_STATES + JobStatus.JOB_REALLY_FINAL_STATES)
+        final_states = f"'{final_states}'"
+
+        query = f"SELECT {defString}, {valueString} FROM ("
+        # All jobs that are NOT in a final state
+        query += (
+            f"SELECT {defString}, {valueString}, COUNT(JobID), SUM(RescheduleCounter) "
+            f"FROM Jobs WHERE STATUS NOT IN ({final_states}) "
+            f"GROUP BY {defString}, {valueString} "
+        )
+        query += "UNION ALL "
+        # Recent jobs only (today) that are in a final state
+        query += (
+            f"SELECT {defString}, {valueString}, COUNT(JobID), SUM(RescheduleCounter) "
+            f"FROM Jobs WHERE Status IN ({final_states}) AND LastUpdateTime >= UTC_DATE() "
+            f"GROUP BY {defString}, {valueString} "
+        )
+        query += "UNION ALL "   
+        # Cached history (of jobs in a final state)
+        query += f"SELECT * FROM JobsHistorySummary) AS combined GROUP BY {defString}, {valueString}"
+
+        result = self._query(query)
         if not result["OK"]:
             return result
         return S_OK(((requestedFields + valueFields), result["Value"]))

src/DIRAC/WorkloadManagementSystem/DB/JobDB.sql

@@ -132,3 +132,19 @@ CREATE TABLE `JobCommands` (
   PRIMARY KEY (`JobID`,`Arguments`,`ReceptionTime`),
   FOREIGN KEY (`JobID`) REFERENCES `Jobs`(`JobID`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+-- ------------------------------------------------------------------------------
+DROP TABLE IF EXISTS `JobsHistorySummary`;
+CREATE TABLE `JobsHistorySummary` (
+  `Status` VARCHAR(32),
+  `Site` VARCHAR(100),
+  `Owner` VARCHAR(32),
+  `OwnerGroup` VARCHAR(128),
+  `JobGroup` VARCHAR(32),
+  `JobType` VARCHAR(32),
+  `ApplicationStatus` VARCHAR(255),
+  `MinorStatus` VARCHAR(128),
+  `JobCount` INT,
+  `RescheduleSum` INT,
+  PRIMARY KEY (Status, Site, Owner, OwnerGroup, JobGroup, JobType, ApplicationStatus, MinorStatus)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;