Merge pull request #5999 from rupozzi/pilothistory

[integration] PilotsHistory Monitoring

Author: fstagni

docs/source/AdministratorGuide/Systems/MonitoringSystem/index.rst

@@ -10,9 +10,10 @@ Monitoring System
 Overview
 =========
 
-The Monitoring system is used to monitor various components of DIRAC. Currently, we have five monitoring types:
+The Monitoring system is used to monitor various components of DIRAC. Currently, we have several monitoring types:
 
   - WMSHistory: for monitoring the DIRAC WMS
+  - PilotsHistory: for monitoring of DIRAC pilots
   - Component Monitoring: for monitoring DIRAC components such as services, agents, etc.
   - RMS Monitoring: for monitoring the DIRAC RequestManagement System (mostly the Request Executing Agent).
   - PilotSubmission Monitoring: for monitoring the DIRAC pilot submission statistics from SiteDirector agents
@@ -122,6 +123,11 @@ You can configure the MQ in the local dirac.cfg file where the agent is running:
   Note: the JSON file already contains the index patterns needed for the visualizations. You may need to adapt the index patterns to your existing ones.
 
 
+Enable PilotsHistory monitoring
+===============================
+In order to enable PilotsHistory monitoring you need to set the flag ``monitoringEnabled = True`` in Operations/Defaults.
+
+
 Enable Component monitoring
 ===========================
 

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

@@ -0,0 +1,36 @@
+""" Definition for PilotsHistory Monitoring type.
+    Filled by the agent "WorkloadManagement/StatesAccountingAgent"
+"""
+
+from DIRAC.MonitoringSystem.Client.Types.BaseType import BaseType
+
+
+class PilotsHistory(BaseType):
+    """
+    .. class:: PilotsHistoryMonitorType
+    """
+
+    def __init__(self):
+        """
+        :param self: self reference
+        """
+
+        super().__init__()
+
+        self.keyFields = ["TaskQueueID", "GridSite", "GridType", "Status"]
+
+        self.monitoringFields = ["NumOfPilots"]
+
+        self.index = "pilotshistory_index"
+
+        self.addMapping(
+            {
+                "TaskQueueID": {"type": "keyword"},
+                "GridSite": {"type": "keyword"},
+                "GridType": {"type": "keyword"},
+                "Status": {"type": "keyword"},
+                "NumOfPilots": {"type": "long"},
+            }
+        )
+
+        self.checkType()

src/DIRAC/WorkloadManagementSystem/Agent/StatesAccountingAgent.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 = ["NumOfPilots"]
+
     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.log.info("Committing to %s backend" % "and ".join(self.backends))
+        self.monitoringEnabled = Operations().getValue("MonitoringEnabled", False)
 
+        messageQueue = self.am_getOption("MessageQueue", "dirac.wmshistory")
+        pilotMessageQueue = self.am_getOption("MessageQueue", "dirac.monitoring")
         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", failoverQueueName=pilotMessageQueue)
 
         self.__jobDBFields = []
         for field in self.__summaryKeyFieldsMapping:
@@ -66,27 +75,62 @@ def initialize(self):
 
     def execute(self):
         """Main execution method"""
-        # Get the WMS Snapshot!
+
+        # PilotsHistory to Monitoring
+        if self.monitoringEnabled:
+            self.log.info("Committing PilotsHistory to Monitoring")
+            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 PilotsHistory at this cycle" % result["Message"],
+                )
+                return S_ERROR()
+
+            values = result["Value"][1]
+            for record in values:
+                record = record[1:]
+                rD = {}
+                for iP, _ in enumerate(self.__pilotKeyFields):
+                    rD[self.__pilotKeyFields[iP]] = record[iP]
+                record = record[len(self.__pilotKeyFields) :]
+                for iP, _ in enumerate(self.__pilotValueFields):
+                    rD[self.__pilotValueFields[iP]] = int(record[iP])
+                rD["timestamp"] = int(Time.toEpoch(now))
+                self.log.debug("Adding following PilotsHistory record to Reporter: \n", rD)
+                self.pilotReporter.addRecord(rD)
+
+            self.log.info("Committing to Monitoring...")
+            result = self.pilotReporter.commit()
+            if not result["OK"]:
+                self.log.error("Could not commit to Monitoring", result["Message"])
+                return result
+            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 = {}
             for fV in self.__summaryDefinedFields:
                 rD[fV[0]] = fV[1]
-            for iP in range(len(self.__summaryKeyFieldsMapping)):
+            for iP, _ in enumerate(self.__summaryKeyFieldsMapping):
                 fieldName = self.__summaryKeyFieldsMapping[iP]
                 rD[self.__renameFieldsMapping.get(fieldName, fieldName)] = record[iP]
             record = record[len(self.__summaryKeyFieldsMapping) :]
-            for iP in range(len(self.__summaryValueFieldsMapping)):
+            for iP, _ in enumerate(self.__summaryValueFieldsMapping):
                 rD[self.__summaryValueFieldsMapping[iP]] = int(record[iP])
 
             for backend in self.datastores:
@@ -101,16 +145,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/DB/PilotAgentsDB.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:
     """

tests/Integration/Monitoring/Test_MonitoringReporter.py

@@ -63,7 +63,7 @@
 wmsMonitoringReporter = MonitoringReporter(monitoringType="WMSHistory")
 componentMonitoringReporter = MonitoringReporter(monitoringType="ComponentMonitoring")
 pilotMonitoringReporter = MonitoringReporter(monitoringType="PilotSubmissionMonitoring")
-
+pilotsHistoryReporter = MonitoringReporter(monitoringType="PilotsHistory")
 data = [
     {
         "Status": "Waiting",
@@ -852,6 +852,25 @@
     },
 ]
 
+pilotsHistoryData = [
+    {
+        "TaskQueueID": "1240451",
+        "GridSite": "LCG.CNAF.it",
+        "GridType": "",
+        "Status": "failed",
+        "Pilots": "7",
+        "timestamp": 1649161714,
+    },
+    {
+        "TaskQueueID": "12401",
+        "GridSite": "LCG.CNAF.it",
+        "GridType": "",
+        "Status": "failed",
+        "Pilots": "7",
+        "timestamp": 1649161714,
+    },
+]
+
 
 def test_addWMSRecords():
     for record in data:
@@ -875,3 +894,11 @@ def test_addPilotSubmissionRecords():
     result = pilotMonitoringReporter.commit()
     assert result["OK"]
     assert result["Value"] == len(pilotMonitoringData)
+
+
+def test_addPilotHistoryRecords():
+    for record in pilotsHistoryData:
+        pilotsHistoryReporter.addRecord(record)
+    result = pilotsHistoryReporter.commit()
+    assert result["OK"]
+    assert result["Value"] == len(pilotsHistoryData)