feat: Added function to send pilot submission to ES

Author: rupozzi

src/DIRAC/MonitoringSystem/Client/MonitoringReporter.py

@@ -15,11 +15,6 @@
 a MQ service is available, if the MQ is not working a failover will be performed.
 
 """
-from __future__ import absolute_import
-from __future__ import division
-from __future__ import print_function
-
-__RCSID__ = "$Id$"
 
 import threading
 import json

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

@@ -14,7 +14,7 @@ def __init__(self):
 
         self.monitoringFields = ["NumTotal", "NumSucceeded"]
 
-        self.index = "pilot-stats_index"
+        self.index = "pilotstats_index"
 
         self.addMapping(
             {

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

@@ -3,9 +3,6 @@
 
     Filled by the agent "WorkloadManagement/StatesAccountingAgent"
 """
-from __future__ import absolute_import
-from __future__ import division
-from __future__ import print_function
 
 from DIRAC.MonitoringSystem.Client.Types.BaseType import BaseType
 

src/DIRAC/MonitoringSystem/DB/MonitoringDB.py

@@ -32,11 +32,6 @@
 
 
 """
-from __future__ import absolute_import
-from __future__ import division
-from __future__ import print_function
-
-__RCSID__ = "$Id$"
 
 import time
 import calendar

src/DIRAC/WorkloadManagementSystem/Agent/SiteDirector.py

@@ -15,16 +15,17 @@
 from concurrent.futures import ThreadPoolExecutor
 
 import DIRAC
-from DIRAC import S_OK, gConfig
+from DIRAC import S_OK, S_ERROR, gConfig
 from DIRAC.ConfigurationSystem.Client.Helpers import CSGlobals, Registry
 from DIRAC.ConfigurationSystem.Client.Helpers.Operations import Operations
 from DIRAC.ConfigurationSystem.Client.Helpers.Resources import getCESiteMapping
 from DIRAC.Core.Base.AgentModule import AgentModule
-from DIRAC.Core.Utilities.Time import dateTime, second
+from DIRAC.Core.Utilities.Time import dateTime, second, toEpoch
 from DIRAC.Core.Utilities.ObjectLoader import ObjectLoader
 from DIRAC.FrameworkSystem.Client.ProxyManagerClient import gProxyManager
 from DIRAC.AccountingSystem.Client.Types.Pilot import Pilot as PilotAccounting
 from DIRAC.AccountingSystem.Client.Types.PilotSubmission import PilotSubmission as PilotSubmissionAccounting
+from DIRAC.MonitoringSystem.Client.MonitoringReporter import MonitoringReporter
 from DIRAC.MonitoringSystem.Client.Types.PilotMonitoring import PilotMonitoring as PilotSubmissionMonitoring
 from DIRAC.AccountingSystem.Client.DataStoreClient import gDataStoreClient
 from DIRAC.WorkloadManagementSystem.Client import PilotStatus
@@ -722,7 +723,15 @@ def _submitPilotsToQueue(self, pilotsToSubmit, ce, queue):
                     0,
                     "Failed",
                 )
-
+            if self.sendSubmissionMonitoring:
+                self.sendPilotSubmissionMonitoring(
+                    self.queueDict[queue]["Site"],
+                    self.queueDict[queue]["CEName"],
+                    self.queueDict[queue]["QueueName"],
+                    pilotsToSubmit,
+                    0,
+                    "Failed",
+                )
             self.failedQueues[queue] += 1
             return submitResult
 
@@ -745,6 +754,15 @@ def _submitPilotsToQueue(self, pilotsToSubmit, ce, queue):
                 len(pilotList),
                 "Succeeded",
             )
+        if self.sendSubmissionMonitoring:
+            self.sendPilotSubmissionMonitoring(
+                self.queueDict[queue]["Site"],
+                self.queueDict[queue]["CEName"],
+                self.queueDict[queue]["QueueName"],
+                len(pilotList),
+                len(pilotList),
+                "Succeeded",
+            )
 
         return S_OK((pilotList, stampDict))
 
@@ -1365,3 +1383,40 @@ def sendPilotSubmissionAccounting(self, siteName, ceName, queueName, numTotal, n
         if not result["OK"]:
             self.log.error("Error in Commit:" + result["Message"])
             return result
+
+    def sendPilotSubmissionMonitoring(self, siteName, ceName, queueName, numTotal, numSucceeded, status):
+        """Send pilot submission accounting record
+
+        :param str siteName:     Site name
+        :param str ceName:       CE name
+        :param str queueName:    queue Name
+        :param int numTotal:     Total number of submission
+        :param int numSucceeded: Total number of submission succeeded
+        :param str status:       'Succeeded' or 'Failed'
+
+        :returns: S_OK / S_ERROR
+        """
+
+        pilotMonitoringReporter = MonitoringReporter(monitoringType="PilotMonitoring")
+        pilotMonitoringData = [
+            {
+                "HostName": "",
+                "SiteDirector": "",
+                "Site": siteName,
+                "CE": ceName,
+                "Queue": queueName,
+                "Status": status,
+                "NumTotal": numTotal,
+                "NumSucceded": numSucceeded,
+                "timestamp": int(toEpoch(dateTime())),
+            }
+        ]
+        pilotMonitoringReporter.addRecord(pilotMonitoringData)
+        result = pilotMonitoringReporter.commit()
+
+        self.log.verbose("Committing pilot submission to monitoring")
+        if not result["OK"]:
+            self.log.error("Couldn't commit pilot submission to monitoring", result["Message"])
+            return S_ERROR()
+        self.log.verbose("Done committing to monitoring")
+        return S_OK()