feat: use summary table and triggers (PilotAgentsDB.PilotAgents, summaries for StatesAccounting)

F

Author: fstagni

src/DIRAC/WorkloadManagementSystem/Agent/StatesAccountingAgent.py

@@ -43,7 +43,7 @@ class StatesAccountingAgent(AgentModule):
     __renameFieldsMapping = {"JobType": "JobSplitType"}
 
     # PilotsHistory fields
-    __pilotsMapping = ["GridSite", "GridType", "Status", "NumOfPilots"]
+    __pilotsMapping = ["GridSite", "GridType", "Status", "VO", "NumOfPilots"]
 
     def initialize(self):
         """Standard initialization"""
@@ -85,7 +85,8 @@ def execute(self):
         # PilotsHistory to Monitoring
         if "Monitoring" in self.pilotMonitoringOption:
             self.log.info("Committing PilotsHistory to Monitoring")
-            result = PilotAgentsDB().getSummarySnapshot()
+            sql = "SELECT * FROM PilotsHistorySummary ORDER BY GridSite, DestinationSite, Status, VO;"
+            result = PilotAgentsDB()._query(sql)
             now = datetime.datetime.utcnow()
             if not result["OK"]:
                 self.log.error(

src/DIRAC/WorkloadManagementSystem/DB/PilotAgentsDB.py

@@ -943,15 +943,12 @@ def getPilotSummaryWeb(self, selectDict, sortList, startItem, maxItems):
     def getPilotMonitorSelectors(self):
         """Get distinct values for the Pilot Monitor page selectors"""
 
-        paramNames = ["VO", "GridType", "Status", "DestinationSite", "GridSite"]
-
         resultDict = {}
-        for param in paramNames:
-            result = self.getDistinctAttributeValues("PilotAgents", param)
-            if result["OK"]:
-                resultDict[param] = result["Value"]
-            else:
+        for param in ["VO", "Status", "DestinationSite", "GridSite"]:
+            result = self.getDistinctAttributeValues("PilotsHistorySummary", param)
+            if not result["OK"]:
                 resultDict = []
+            resultDict[param] = result["Value"]    
 
         return S_OK(resultDict)
 
@@ -1048,18 +1045,6 @@ def getPilotMonitorWeb(self, selectDict, sortList, startItem, maxItems):
 
         return S_OK(resultDict)
 
-    def getSummarySnapshot(self, requestedFields=False):
-        """Get the summary snapshot for a given combination"""
-        if not requestedFields:
-            requestedFields = ["GridSite", "GridType", "Status"]
-        valueFields = ["COUNT(PilotID)"]
-        defString = ", ".join(requestedFields)
-        valueString = ", ".join(valueFields)
-        result = self._query(f"SELECT {defString}, {valueString} FROM PilotAgents GROUP BY {defString}")
-        if not result["OK"]:
-            return result
-        return S_OK(((requestedFields + valueFields), result["Value"]))
-
 
 class PivotedPilotSummaryTable:
     """

src/DIRAC/WorkloadManagementSystem/DB/PilotAgentsDB.sql

@@ -1,5 +1,3 @@
--- $Header: /tmp/libdirac/tmp.stZoy15380/dirac/DIRAC3/DIRAC/WorkloadManagementSystem/DB/PilotAgentsDB.sql,v 1.20 2009/08/26 09:39:53 rgracian Exp $
-
 -- ------------------------------------------------------------------------------
 --
 --  Schema definition for the PilotAgentsDB database - containing the Pilots status
@@ -48,7 +46,6 @@ CREATE TABLE `PilotAgents` (
   KEY `Statuskey` (`GridSite`,`DestinationSite`,`Status`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
 
-
 DROP TABLE IF EXISTS `JobToPilotMapping`;
 CREATE TABLE `JobToPilotMapping` (
   `PilotID` INT(11) UNSIGNED NOT NULL,
@@ -65,3 +62,77 @@ CREATE TABLE `PilotOutput` (
   `StdError` MEDIUMTEXT,
   PRIMARY KEY (`PilotID`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+
+-- ------------------------------------------------------------------------------
+-- summary table and triggers
+-- ------------------------------------------------------------------------------
+
+-- summary for PilotsHistory
+
+DROP TABLE IF EXISTS `PilotsHistorySummary`;
+CREATE TABLE `PilotsHistorySummary` (
+  `GridSite` VARCHAR(128),
+  `DestinationSite` VARCHAR(128),
+  `Status` VARCHAR(32),
+  `VO` VARCHAR(128),
+  `PilotCount` INT,
+  PRIMARY KEY (`GridSite`,`DestinationSite`,`Status`, `VO`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+
+-- now the triggers
+
+DELIMITER //
+
+CREATE TRIGGER trg_PilotAgents_insert
+AFTER INSERT ON PilotAgents
+FOR EACH ROW
+BEGIN
+  INSERT INTO PilotsHistorySummary (GridSite, DestinationSite, Status, VO, PilotCount)
+  VALUES (NEW.GridSite, NEW.DestinationSite, NEW.Status, NEW.VO, 1)
+  ON DUPLICATE KEY UPDATE PilotCount = PilotCount + 1;
+END;
+//
+
+CREATE TRIGGER trg_PilotAgents_delete
+AFTER DELETE ON PilotAgents
+FOR EACH ROW
+BEGIN
+  UPDATE PilotsHistorySummary
+  SET PilotCount = PilotCount - 1
+  WHERE GridSite = OLD.GridSite
+    AND DestinationSite = OLD.DestinationSite
+    AND Status = OLD.Status
+    AND VO = OLD.VO;
+
+  -- Optional cleanup (remove zero rows)
+  DELETE FROM PilotsHistorySummary WHERE PilotCount = 0;
+END;
+//
+
+CREATE TRIGGER trg_PilotAgents_update_status
+AFTER UPDATE ON PilotAgents
+FOR EACH ROW
+BEGIN
+  IF OLD.Status != NEW.Status THEN
+
+    -- Decrease count from old status
+    UPDATE PilotsHistorySummary
+    SET PilotCount = PilotCount - 1
+    WHERE GridSite = OLD.GridSite
+      AND DestinationSite = OLD.DestinationSite
+      AND Status = OLD.Status
+      AND VO = OLD.VO;
+
+    -- Delete row if count drops to zero
+    DELETE FROM PilotsHistorySummary WHERE PilotCount = 0;
+
+    -- Increase count for new status
+    INSERT INTO PilotsHistorySummary (GridSite, DestinationSite, Status, VO, PilotCount)
+    VALUES (NEW.GridSite, NEW.DestinationSite, NEW.Status, NEW.VO, 1)
+    ON DUPLICATE KEY UPDATE PilotCount = PilotCount + 1;
+
+  END IF;
+END;
+//

tests/Integration/WorkloadManagementSystem/Test_PilotAgentsDB.py

@@ -186,3 +186,96 @@ def test_PivotedPilotSummaryTable():
     assert "Total" in columns
 
     cleanUpPilots(pilotRef)
+
+
+# Parse date strings into datetime objects
+def process_data(data):
+    converted_data = []
+
+    for row in data:
+        # date fields
+        date_indices = [10, 11]  # Positions of date fields
+        for i in date_indices:
+            if not row[i]:
+                row[i] = None
+            else:
+                try:
+                    row[i] = datetime.strptime(row[i], "%Y-%m-%d %H:%M:%S")
+                except ValueError:
+                    # Handle invalid dates
+                    row[i] = None
+        # Convert other fields to appropriate types
+        int_indices = [0, 1]  # Positions of integer fields
+        for i in int_indices:
+            if not row[i]:
+                row[i] = 0
+            else:
+                try:
+                    row[i] = int(row[i])
+                except ValueError:
+                    # Handle invalid integers
+                    row[i] = 0
+        float_indices = [9]  # Positions of float fields
+        for i in float_indices:
+            if not row[i]:
+                row[i] = 0
+            else:
+                try:
+                    row[i] = float(row[i])
+                except ValueError:
+                    # Handle invalid float
+                    row[i] = 0
+        converted_data.append(tuple(row))
+    return converted_data
+
+
+def test_summarySnapshot():
+    # first delete all pilots
+    sql = "DELETE FROM PilotAgents"
+    res = paDB._update(sql)
+    assert res["OK"], res["Message"]
+    sql = "DELETE FROM PilotsHistorySummary"
+    res = paDB._update(sql)
+    assert res["OK"], res["Message"]
+
+    # insert some predefined pilots to test the summary snapshot
+    with open("pilots.csv", newline="", encoding="utf-8") as csvfile:
+        csvreader = csv.reader(csvfile)
+        data = list(csvreader)
+        processed_data = process_data(data)
+        placeholders = ",".join(["%s"] * len(processed_data[0]))
+        sql = f"INSERT INTO PilotAgents (InitialJobID, CurrentJobID, PilotJobReference, PilotStamp, DestinationSite, Queue, GridSite, VO, GridType, BenchMark, SubmissionTime, LastUpdateTime, Status, StatusReason, AccountingSent) VALUES ({placeholders})"
+        res = paDB._updatemany(sql, processed_data)
+        assert res["OK"], res["Message"]
+    sql = "SELECT * FROM PilotsHistorySummary ORDER BY GridSite, DestinationSite, Status, VO;"
+    result = PilotAgentsDB()._query(sql)
+    assert result["OK"], result["Message"]
+    values = result["Value"][1]
+    assert len(values) == 5, "Expected 5 record in the summary"
+    # Check it corresponds to the basic "GROUP BY" query
+    sql = "SELECT GridSite, DestinationSite, Status, VO, COUNT(*) FROM PilotAgents GROUP BY GridSite, DestinationSite, Status, VO ORDER BY GridSite, DestinationSite, Status, VO;"
+    result_grouped = PilotAgentsDB()._query(sql)
+    assert result_grouped["OK"], result_grouped["Message"]
+    sql = "SELECT * FROM PilotsHistorySummary ORDER BY GridSite, DestinationSite, Status, VO;"
+    result_summary = PilotAgentsDB()._query(sql)
+    assert result_summary["OK"], result_summary["Message"]
+    assert result_grouped["Value"] == result_summary["Value"], "Summary and grouped query results differ"
+
+    # deleting now
+    with open("pilots.csv", newline="", encoding="utf-8") as csvfile:
+        csvreader = csv.reader(csvfile)
+        data = list(csvreader)
+        processed_data = process_data(data)
+        pilotStamps = [row[3] for row in processed_data]
+        pilotStampsStr = ",".join("'" + p + "'" for p in pilotStamps)
+        sql = f"DELETE FROM PilotAgents WHERE PilotStamp IN (%s)" % pilotStampsStr
+        res = paDB._update(sql)
+        assert res["OK"], res["Message"]
+    # Check it corresponds to the basic "GROUP BY" query
+    sql = "SELECT GridSite, DestinationSite, Status, VO, COUNT(*) FROM PilotAgents GROUP BY GridSite, DestinationSite, Status, VO ORDER BY GridSite, DestinationSite, Status, VO;"
+    result_grouped = PilotAgentsDB()._query(sql)
+    assert result_grouped["OK"], result_grouped["Message"]
+    sql = "select * FROM PilotsHistorySummary ORDER BY GridSite, DestinationSite, Status, VO;"
+    result_summary = PilotAgentsDB()._query(sql)
+    assert result_summary["OK"], result_summary["Message"]
+    assert result_grouped["Value"] == result_summary["Value"], "Summary and grouped query results differ"