feat: use summary table and triggers (JobDB.Jobs, summaries for StatesAccounting)

Author: fstagni

src/DIRAC/WorkloadManagementSystem/Agent/StatesAccountingAgent.py

@@ -34,6 +34,7 @@ class StatesAccountingAgent(AgentModule):
         "User",
         "UserGroup",
         "JobGroup",
+        "VO",
         "JobType",
         "ApplicationStatus",
         "MinorStatus",
@@ -110,14 +111,14 @@ def execute(self):
 
         # WMSHistory to Monitoring or Accounting
         self.log.info(f"Committing WMSHistory to {'and '.join(self.jobMonitoringOption)} backend")
-        result = JobDB().getSummarySnapshot(self.__jobDBFields)
-        now = datetime.datetime.utcnow()
+        result = JobDB()._query("SELECT * FROM JobsHistorySummary")
         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]
 
+        now = datetime.datetime.utcnow()
         self.log.info("Start sending WMSHistory records")
         for record in values:
             rD = {}

src/DIRAC/WorkloadManagementSystem/DB/JobDB.py

@@ -26,10 +26,10 @@
 from DIRAC.Core.Utilities.ReturnValues import (
     S_ERROR,
     S_OK,
+    DReturnType,
+    SErrorException,
     convertToReturnValue,
     returnValueOrRaise,
-    SErrorException,
-    DReturnType,
 )
 from DIRAC.FrameworkSystem.Client.Logger import contextLogger
 from DIRAC.ResourceStatusSystem.Client.SiteStatus import SiteStatus
@@ -107,7 +107,7 @@ def __getAttributeNames(self):
     def getDistinctJobAttributes(self, attribute, condDict=None, older=None, newer=None, timeStamp="LastUpdateTime"):
         """Get distinct values of the job attribute under specified conditions"""
         return self.getDistinctAttributeValues(
-            "Jobs", attribute, condDict=condDict, older=older, newer=newer, timeStamp=timeStamp
+            "JobsHistorySummary", attribute, condDict=condDict, older=older, newer=newer, timeStamp=timeStamp
         )
 
     #############################################################################
@@ -1538,19 +1538,6 @@ def setJobCommandStatus(self, jobID, command, status):
 
         return self._update(f"UPDATE JobCommands SET Status={status} WHERE JobID={jobID} AND Command={command}")
 
-    #####################################################################################
-    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}")
-        if not result["OK"]:
-            return result
-        return S_OK(((requestedFields + valueFields), result["Value"]))
-
     def removeInfoFromHeartBeatLogging(self, status, delTime, maxLines):
         """Remove HeartBeatLoggingInfo from DB.
 

src/DIRAC/WorkloadManagementSystem/DB/JobDB.sql

@@ -40,7 +40,7 @@ CREATE TABLE `Jobs` (
   `JobName` VARCHAR(128) NOT NULL DEFAULT 'Unknown',
   `Owner` VARCHAR(64) NOT NULL DEFAULT 'Unknown',
   `OwnerGroup` VARCHAR(128) NOT NULL DEFAULT 'Unknown',
-  `VO` VARCHAR(32) NOT NULL DEFAULT 'Unknown',
+  `VO` VARCHAR(128) NOT NULL DEFAULT 'Unknown',
   `SubmissionTime` DATETIME DEFAULT NULL,
   `RescheduleTime` DATETIME DEFAULT NULL,
   `LastUpdateTime` DATETIME DEFAULT NULL,
@@ -126,9 +126,110 @@ CREATE TABLE `JobCommands` (
   `JobID` INT(11) UNSIGNED NOT NULL,
   `Command` VARCHAR(100) NOT NULL,
   `Arguments` VARCHAR(100) NOT NULL,
-  `Status` VARCHAR(64) NOT NULL DEFAULT 'Received',
+  `Status` VARCHAR(32) NOT NULL DEFAULT 'Received',
   `ReceptionTime` DATETIME NOT NULL,
   `ExecutionTime` DATETIME DEFAULT NULL,
   PRIMARY KEY (`JobID`,`Arguments`,`ReceptionTime`),
   FOREIGN KEY (`JobID`) REFERENCES `Jobs`(`JobID`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+
+-- ------------------------------------------------------------------------------
+-- summary table and triggers
+-- ------------------------------------------------------------------------------
+
+-- summary for JobsHistory
+
+DROP TABLE IF EXISTS `JobsHistorySummary`;
+CREATE TABLE `JobsHistorySummary` (
+  `ID` INT AUTO_INCREMENT PRIMARY KEY,
+  `Status` VARCHAR(32),
+  `Site` VARCHAR(100),
+  `Owner` VARCHAR(32),
+  `OwnerGroup` VARCHAR(128),
+  `VO` VARCHAR(128),
+  `JobGroup` VARCHAR(32),
+  `JobType` VARCHAR(32),
+  `ApplicationStatus` VARCHAR(255),
+  `MinorStatus` VARCHAR(128),
+  `JobCount` INT,
+  RescheduleSum INT,
+  UNIQUE KEY uq_summary (
+    `Status`, 
+    `Site`, 
+    `Owner`, 
+    `OwnerGroup`(32), 
+    `VO`,
+    `JobGroup`,
+    `JobType`,
+    `ApplicationStatus`(128),
+    `MinorStatus`
+  )
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+
+-- now the triggers
+
+DELIMITER //
+
+CREATE TRIGGER trg_Jobs_insert
+AFTER INSERT ON Jobs
+FOR EACH ROW
+BEGIN
+  INSERT INTO JobsHistorySummary (Status, Site, Owner, OwnerGroup, VO, JobGroup, JobType, ApplicationStatus, MinorStatus, JobCount, RescheduleSum)
+  VALUES (NEW.Status, NEW.Site, NEW.Owner, NEW.OwnerGroup, NEW.VO, NEW.JobGroup, NEW.JobType, NEW.ApplicationStatus, NEW.MinorStatus, 1, NEW.RescheduleCounter)
+  ON DUPLICATE KEY UPDATE JobCount = JobCount + 1, RescheduleSum = RescheduleSum + NEW.RescheduleCounter;
+END;
+//
+
+CREATE TRIGGER trg_Jobs_delete
+AFTER DELETE ON Jobs
+FOR EACH ROW
+BEGIN
+  UPDATE JobsHistorySummary
+  SET JobCount = JobCount - 1, RescheduleSum = RescheduleSum - OLD.RescheduleCounter
+  WHERE Status = OLD.Status
+    AND Site = OLD.Site
+    AND Owner = OLD.Owner
+    AND OwnerGroup = OLD.OwnerGroup
+    AND VO = OLD.VO
+    AND JobGroup = OLD.JobGroup
+    AND JobType = OLD.JobType
+    AND ApplicationStatus = OLD.ApplicationStatus
+    AND MinorStatus = OLD.MinorStatus;
+
+  -- Optional cleanup (remove zero rows)
+  DELETE FROM JobsHistorySummary WHERE JobCount = 0;
+END;
+//
+
+CREATE TRIGGER trg_Jobs_update_status
+AFTER UPDATE ON Jobs
+FOR EACH ROW
+BEGIN
+  IF OLD.Status != NEW.Status THEN
+
+    -- Decrease count from old status
+    UPDATE JobsHistorySummary
+    SET JobCount = JobCount - 1, RescheduleSum = RescheduleSum - OLD.RescheduleCounter
+    WHERE Status = OLD.Status
+      AND Site = OLD.Site
+      AND Owner = OLD.Owner
+      AND OwnerGroup = OLD.OwnerGroup
+      AND VO = OLD.VO
+      AND JobGroup = OLD.JobGroup
+      AND JobType = OLD.JobType
+      AND ApplicationStatus = OLD.ApplicationStatus
+      AND MinorStatus = OLD.MinorStatus;
+
+    -- Delete row if count drops to zero
+    DELETE FROM JobsHistorySummary WHERE JobCount = 0;
+
+    -- Increase count for new status
+    INSERT INTO JobsHistorySummary (Status, Site, Owner, OwnerGroup, JobGroup, VO, JobType, ApplicationStatus, MinorStatus, JobCount, RescheduleSum)
+    VALUES (NEW.Status, NEW.Site, NEW.Owner, NEW.OwnerGroup, NEW.JobGroup, NEW.VO, NEW.JobType, NEW.ApplicationStatus, NEW.MinorStatus, 1, NEW.RescheduleCounter)
+    ON DUPLICATE KEY UPDATE JobCount = JobCount + 1, RescheduleSum = RescheduleSum + NEW.RescheduleCounter;
+
+  END IF;
+END;
+//