feat: Added method for retrieving snapshot of PilotsHistory

Author: rupozzi

src/DIRAC/MonitoringSystem/Client/Types/PilotsHistory.py

@@ -1,5 +1,5 @@
 """ Definition for PilotsHistory Monitoring type.
-    Filled by the agent "WorkloadManagement/PilotsHistoryAgent"
+    Filled by the agent "WorkloadManagement/StatesAccountingAgent"
 """
 
 from DIRAC.MonitoringSystem.Client.Types.BaseType import BaseType

src/DIRAC/WorkloadManagementSystem/Agent/PilotsHistoryAgent.py

@@ -1,4 +1,4 @@
-"""  StatesAccountingAgent sends periodically numbers of jobs in various states for various
+"""  StatesAccountingAgent sends periodically numbers of jobs and pilots in various states for various
      sites to the Monitoring system to create historical plots.
 
 .. literalinclude:: ../ConfigTemplate.cfg
@@ -8,20 +8,24 @@
   :caption: StatesAccountingAgent options
 """
 from DIRAC import S_OK, S_ERROR
+from DIRAC.ConfigurationSystem.Client.Helpers.Operations import Operations
 from DIRAC.Core.Base.AgentModule import AgentModule
 from DIRAC.Core.Utilities import Time
 from DIRAC.AccountingSystem.Client.Types.WMSHistory import WMSHistory
 from DIRAC.AccountingSystem.Client.DataStoreClient import DataStoreClient
 from DIRAC.MonitoringSystem.Client.MonitoringReporter import MonitoringReporter
 from DIRAC.WorkloadManagementSystem.DB.JobDB import JobDB
+from DIRAC.WorkloadManagementSystem.DB.PilotAgentsDB import PilotAgentsDB
 
 
 class StatesAccountingAgent(AgentModule):
     """Agent that every 15 minutes will report
     to the AccountingDB (MySQL) or the Monitoring DB (ElasticSearch), or both,
     a snapshot of the JobDB.
+    Also sends a snapshot of PilotAgentsDB to Monitoring.
     """
 
+    # WMSHistory fields
     __summaryKeyFieldsMapping = [
         "Status",
         "Site",
@@ -36,24 +40,29 @@ class StatesAccountingAgent(AgentModule):
     __summaryValueFieldsMapping = ["Jobs", "Reschedules"]
     __renameFieldsMapping = {"JobType": "JobSplitType"}
 
+    # PilotsHistory fields
+    __pilotKeyFields = ["TaskQueueID", "GridSite", "GridType", "Status"]
+    __pilotValueFields = ["Pilots"]
+
     def initialize(self):
         """Standard initialization"""
         # This agent will always loop every 15 minutes
         self.am_setOption("PollingTime", 900)
 
         self.backends = self.am_getOption("Backends", "Accounting").replace(" ", "").split(",")
-        messageQueue = self.am_getOption("MessageQueue", "dirac.wmshistory")
+        self.monitoringEnabled = Operations().getValue("monitoringEnabled", "False")
 
-        self.log.info("Committing to %s backend" % "and ".join(self.backends))
+        messageQueue = self.am_getOption("MessageQueue", "dirac.wmshistory")
 
         self.datastores = {}  # For storing the clients to Accounting and Monitoring
 
         if "Accounting" in self.backends:
             self.datastores["Accounting"] = DataStoreClient(retryGraceTime=900)
-        if "Monitoring" in self.backends:
+        if "Monitoring" in self.backends or self.monitoringEnabled:
             self.datastores["Monitoring"] = MonitoringReporter(
                 monitoringType="WMSHistory", failoverQueueName=messageQueue
             )
+            self.pilotReporter = MonitoringReporter(monitoringType="PilotsHistory")
 
         self.__jobDBFields = []
         for field in self.__summaryKeyFieldsMapping:
@@ -66,17 +75,50 @@ def initialize(self):
 
     def execute(self):
         """Main execution method"""
-        # Get the WMS Snapshot!
+
+        # PilotsHistory to Monitoring
+        if self.monitoringEnabled:
+            result = PilotAgentsDB.getSummarySnapshot(self.__pilotKeyFields)
+            now = Time.dateTime()
+            if not result["OK"]:
+                self.log.error(
+                    "Can't get the PilotAgentsDB summary", "%s: won't commit at this cycle" % result["Message"]
+                )
+                return S_ERROR()
+
+            values = result["Value"][1]
+            for record in values:
+                record = record[1:]
+                rD = {}
+                for iP in range(len(self.__pilotKeyFields)):
+                    rD[self.__pilotKeyFields[iP]] = record[iP]
+                record = record[len(self.__pilotKeyFields) :]
+                for iP in range(len(self.__pilotValueFields)):
+                    rD[self.__pilotValueFields[iP]] = int(record[iP])
+                rD["timestamp"] = int(Time.toEpoch(now))
+                self.log.verbose("Adding following PilotsHistory record to Reporter: \n", rD)
+                self.pilotReporter.addRecord(rD)
+
+            self.log.info("Committing PilotsHistory to Monitoring")
+            result = self.pilotReporter.commit()
+            if not result["OK"]:
+                self.log.error("Could not commit PilotsHistory to Monitoring")
+                return S_ERROR()
+            self.log.verbose("Done committing PilotsHistory to Monitoring")
+
+        # WMSHistory to Monitoring or Accounting
+        self.log.info("Committing WMSHistory to %s backend" % "and ".join(self.backends))
         result = JobDB().getSummarySnapshot(self.__jobDBFields)
         now = Time.dateTime()
         if not result["OK"]:
-            self.log.error("Can't get the JobDB summary", "%s: won't commit at this cycle" % result["Message"])
+            self.log.error(
+                "Can't get the JobDB summary", "%s: won't commit WMSHistory at this cycle" % result["Message"]
+            )
             return S_ERROR()
 
-        # Now we try to commit
         values = result["Value"][1]
 
-        self.log.info("Start sending records")
+        self.log.info("Start sending WMSHistory records")
         for record in values:
             record = record[1:]
             rD = {}
@@ -101,16 +143,16 @@ def execute(self):
                     acWMS.setValuesFromDict(rD)
                     retVal = acWMS.checkValues()
                     if not retVal["OK"]:
-                        self.log.error("Invalid accounting record ", "%s -> %s" % (retVal["Message"], rD))
+                        self.log.error("Invalid WMSHistory accounting record ", "%s -> %s" % (retVal["Message"], rD))
                     else:
                         self.datastores["Accounting"].addRegister(acWMS)
 
         for backend, datastore in self.datastores.items():
-            self.log.info("Committing to %s backend" % backend)
+            self.log.info("Committing WMSHistory records to %s backend" % backend)
             result = datastore.commit()
             if not result["OK"]:
-                self.log.error("Couldn't commit WMS history to %s" % backend, result["Message"])
+                self.log.error("Couldn't commit WMSHistory to %s" % backend, result["Message"])
                 return S_ERROR()
-            self.log.verbose("Done committing to %s backend" % backend)
+            self.log.verbose("Done committing WMSHistory to %s backend" % backend)
 
         return S_OK()

src/DIRAC/WorkloadManagementSystem/Agent/StatesAccountingAgent.py

@@ -1173,6 +1173,17 @@ def getPilotMonitorWeb(self, selectDict, sortList, startItem, maxItems):
 
         return S_OK(resultDict)
 
+    def getSummarySnapshot(self, requestedFields=False):
+        """Get the summary snapshot for a given combination"""
+        requestedFields = ["TaskQueueID", "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:
     """