fix: pre-select the fill_time instead of using UTC_DATE

Author: fstagni

src/DIRAC/WorkloadManagementSystem/Agent/StatesAccountingAgent.py

@@ -68,11 +68,15 @@ def initialize(self):
         if "Monitoring" in self.pilotMonitoringOption:
             self.pilotReporter = MonitoringReporter(monitoringType="PilotsHistory", failoverQueueName=messageQueue)
 
-        threadJobDB = threading.Thread(target=JobDB().fillJobsHistorySummary)
+        # self.fill_time: now -1h
+        now = datetime.datetime.utcnow() - datetime.timedelta(hours=1)
+        self.fill_time = now.strftime("%Y-%m-%d %H:%M:%S")
+
+        threadJobDB = threading.Thread(target=lambda: JobDB().fillJobsHistorySummary(self.fill_time))
         threadJobDB.daemon = True
         threadJobDB.start()
 
-        threadPilotDB = threading.Thread(target=PilotAgentsDB().fillPilotsHistorySummary)
+        threadPilotDB = threading.Thread(target=lambda: PilotAgentsDB().fillPilotsHistorySummary(self.fill_time))
         threadPilotDB.daemon = True
         threadPilotDB.start()
 
@@ -96,7 +100,7 @@ def execute(self):
         # PilotsHistory to Monitoring
         if "Monitoring" in self.pilotMonitoringOption:
             self.log.info("Committing PilotsHistory to Monitoring")
-            result = PilotAgentsDB().getSummarySnapshot()
+            result = PilotAgentsDB().getSummarySnapshot(fill_time=self.fill_time)
             now = datetime.datetime.utcnow()
             if not result["OK"]:
                 self.log.error(
@@ -119,7 +123,7 @@ 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)
+        result = JobDB().getSummarySnapshot(fill_time=self.fill_time, requestedFields=self.__jobDBFields)
         now = datetime.datetime.utcnow()
         if not result["OK"]:
             self.log.error("Can't get the JobDB summary", f"{result['Message']}: won't commit WMSHistory at this cycle")

src/DIRAC/WorkloadManagementSystem/DB/JobDB.py

@@ -1510,7 +1510,7 @@ def setJobCommandStatus(self, jobID, command, status):
         return self._update(f"UPDATE JobCommands SET Status={status} WHERE JobID={jobID} AND Command={command}")
 
     #####################################################################################
-    def fillJobsHistorySummary(self):
+    def fillJobsHistorySummary(self, fill_time: str):
         """Fill the JobsHistorySummary table with the summary of the jobs in a final state"""
 
         # Create the staging table
@@ -1525,7 +1525,7 @@ def fillJobsHistorySummary(self):
         final_states = f"'{final_states}'"
         query = (
             f"INSERT INTO JobsHistorySummary_staging SELECT {defString}, {valuesString} "
-            f"FROM Jobs WHERE Status IN ({final_states}) AND LastUpdateTime < UTC_DATE() "
+            f"FROM Jobs WHERE Status IN ({final_states}) AND LastUpdateTime < '{fill_time}' "
             f"GROUP BY {defString}"
         )
         if not (result := self._update(query))["OK"]:
@@ -1539,7 +1539,7 @@ def fillJobsHistorySummary(self):
         )
         return self._update(sql)
 
-    def getSummarySnapshot(self, requestedFields=False):
+    def getSummarySnapshot(self, fill_time: str, requestedFields=False):
         """Get the summary snapshot for a given combination"""
         if not requestedFields:
             requestedFields = ["Status", "MinorStatus", "Site", "Owner", "OwnerGroup", "JobGroup"]
@@ -1555,10 +1555,10 @@ def getSummarySnapshot(self, requestedFields=False):
             f"GROUP BY {defString} "
         )
         query += "UNION ALL "
-        # Recent jobs only (today) that are in a final state
+        # Recent jobs only (after fill_time) that are in a final state
         query += (
             f"SELECT {defString}, COUNT(JobID) AS JobCount, SUM(RescheduleCounter) AS RescheduleSum "
-            f"FROM Jobs WHERE Status IN ({final_states}) AND LastUpdateTime >= UTC_DATE() "
+            f"FROM Jobs WHERE Status IN ({final_states}) AND LastUpdateTime >= '{fill_time}' "
             f"GROUP BY {defString} "
         )
         query += "UNION ALL "

src/DIRAC/WorkloadManagementSystem/DB/PilotAgentsDB.py

@@ -1047,7 +1047,7 @@ def getPilotMonitorWeb(self, selectDict, sortList, startItem, maxItems):
 
         return S_OK(resultDict)
 
-    def fillPilotsHistorySummary(self):
+    def fillPilotsHistorySummary(self, fill_time: str):
         """Fill the PilotsHistorySummary table with the summary of the Pilots in a final state"""
 
         # Create the staging table
@@ -1063,7 +1063,7 @@ def fillPilotsHistorySummary(self):
 
         query = (
             f"INSERT INTO PilotsHistorySummary_staging SELECT {defString}, {valuesString} "
-            f"FROM PilotAgents WHERE Status IN ({final_states}) AND LastUpdateTime < UTC_DATE() "
+            f"FROM PilotAgents WHERE Status IN ({final_states}) AND LastUpdateTime < '{fill_time}' "
             f"GROUP BY {defString}"
         )
         if not (result := self._update(query))["OK"]:
@@ -1077,7 +1077,7 @@ def fillPilotsHistorySummary(self):
         )
         return self._update(sql)
 
-    def getSummarySnapshot(self, requestedFields=False):
+    def getSummarySnapshot(self, fill_time: str, requestedFields=False):
         """Get the summary snapshot for a given combination"""
         if not requestedFields:
             requestedFields = ["GridSite", "GridType", "Status"]
@@ -1097,7 +1097,7 @@ def getSummarySnapshot(self, requestedFields=False):
         # Recent Pilots only (today) that are in a final state
         query += (
             f"SELECT {defString}, {valueString} "
-            f"FROM PilotAgents WHERE Status IN ({final_states}) AND LastUpdateTime >= UTC_DATE() "
+            f"FROM PilotAgents WHERE Status IN ({final_states}) AND LastUpdateTime >= '{fill_time}' "
             f"GROUP BY {defString} "
         )
         query += "UNION ALL "

tests/Integration/WorkloadManagementSystem/Test_JobDB.py

@@ -520,9 +520,11 @@ def test_summarySnapshot():
         res = jobDB._updatemany(sql, processed_data)
         assert res["OK"], res["Message"]
     # Act
-    res = jobDB.fillJobsHistorySummary()
+    now = datetime.utcnow() - timedelta(hours=1)
+    fill_time = now.strftime("%Y-%m-%d %H:%M:%S")
+    res = jobDB.fillJobsHistorySummary(fill_time)
     assert res["OK"], res["Message"]
-    res = jobDB.getSummarySnapshot()
+    res = jobDB.getSummarySnapshot(fill_time)
     assert res["OK"], res["Message"]
     requestedFields = ["Status", "MinorStatus", "Site", "Owner", "OwnerGroup", "JobGroup"]
     defString = ", ".join(requestedFields)

tests/Integration/WorkloadManagementSystem/Test_PilotAgentsDB.py

@@ -6,7 +6,7 @@
 # pylint: disable=wrong-import-position
 
 import csv
-from datetime import datetime
+from datetime import datetime, timedelta
 from unittest.mock import patch
 
 import DIRAC
@@ -242,9 +242,11 @@ def test_summarySnapshot():
         res = paDB._updatemany(sql, processed_data)
         assert res["OK"], res["Message"]
     # Act
-    res = paDB.fillPilotsHistorySummary()
+    now = datetime.utcnow() - timedelta(hours=1)
+    fill_time = now.strftime("%Y-%m-%d %H:%M:%S")
+    res = paDB.fillPilotsHistorySummary(fill_time)
     assert res["OK"], res["Message"]
-    res = paDB.getSummarySnapshot()
+    res = paDB.getSummarySnapshot(fill_time)
     assert res["OK"], res["Message"]
     requestedFields = ["GridSite", "GridType", "Status"]
     defString = ", ".join(requestedFields)