feat (Tornado Monitoring): handlers report to Activity Monitoring

Author: chaen

src/DIRAC/Core/Tornado/Server/TornadoServer.py

@@ -211,9 +211,7 @@ def startTornado(self):
             self.__report = self.__startReportToMonitoringLoop()
             # Response time
             # Starting monitoring, IOLoop waiting time in ms, __monitoringLoopDelay is defined in seconds
-            tornado.ioloop.PeriodicCallback(
-                self.__reportToMonitoring(self.__elapsedTime), self.__monitoringLoopDelay * 1000
-            ).start()
+            tornado.ioloop.PeriodicCallback(self.__reportToMonitoring, self.__monitoringLoopDelay * 1000).start()
 
         for port, app in self.__appsSettings.items():
             sLog.debug(" - %s" % "\n - ".join([f"{k} = {ssl_options[k]}" for k in ssl_options]))
@@ -235,7 +233,7 @@ def startTornado(self):
 
         tornado.ioloop.IOLoop.current().start()
 
-    def __reportToMonitoring(self, responseTime):
+    def __reportToMonitoring(self):
         """
         Periodically reports to Monitoring
         """
@@ -250,13 +248,24 @@ def __reportToMonitoring(self, responseTime):
                 "ServiceName": "Tornado",
                 "MemoryUsage": self.__report[2],
                 "CpuPercentage": percentage,
-                "ResponseTime": responseTime,
             }
         )
         self.activityMonitoringReporter.commit()
         # Save memory usage and save realtime/CPU time for next call
         self.__report = self.__startReportToMonitoringLoop()
 
+        # For each handler,
+        for urlSpec in self.handlerManager.getHandlersDict().values():
+            # If there is more than one URL, it's
+            # most likely something that inherit from TornadoREST
+            # so don't even try to monitor...
+            if len(urlSpec["URLs"]) > 1:
+                continue
+            handler = urlSpec["URLs"][0].handler_class
+            # If there is a Monitoring reporter, call commit on it
+            if getattr(handler, "activityMonitoringReporter", None):
+                handler.activityMonitoringReporter.commit()
+
     def __startReportToMonitoringLoop(self):
         """
         Snapshot of resources to be taken at the beginning

src/DIRAC/Core/Tornado/Server/private/BaseRequestHandler.py

@@ -20,9 +20,11 @@
 import DIRAC
 
 from DIRAC import gConfig, gLogger, S_OK, S_ERROR
+from DIRAC.ConfigurationSystem.Client.Helpers.Operations import Operations
 from DIRAC.Core.Utilities import DErrno
 from DIRAC.Core.DISET.AuthManager import AuthManager
 from DIRAC.Core.Utilities.JEncode import decode, encode
+from DIRAC.Core.Utilities import Network, TimeUtilities
 from DIRAC.Core.Utilities.ReturnValues import isReturnStructure
 from DIRAC.Core.Security.X509Chain import X509Chain  # pylint: disable=import-error
 from DIRAC.Resources.IdProvider.Utilities import getProvidersForInstance
@@ -475,6 +477,12 @@ def __initialize(cls, request):
 
             cls.initializeHandler(cls._componentInfoDict)
 
+            cls.activityMonitoringReporter = None
+            if "Monitoring" in Operations().getMonitoringBackends(monitoringType="ServiceMonitoring"):
+                from DIRAC.MonitoringSystem.Client.MonitoringReporter import MonitoringReporter
+
+                cls.activityMonitoringReporter = MonitoringReporter(monitoringType="ServiceMonitoring")
+
             cls.__init_done = True
 
             return S_OK()
@@ -662,26 +670,46 @@ def on_finish(self):
         Called after the end of HTTP request.
         Log the request duration
         """
-        elapsedTime = 1000.0 * self.request.request_time()
+        elapsedTime = self.request.request_time()
+
         credentials = self.srv_getFormattedRemoteCredentials()
 
         argsString = f"OK {self._status_code}"
+        monitoringRetStatus = "Unknown"
         # Finish with DIRAC result
         if isReturnStructure(self.__result):
             if self.__result["OK"]:
                 argsString = "OK"
+                monitoringRetStatus = "OK"
             else:
                 argsString = f"ERROR: {self.__result['Message']}"
+                monitoringRetStatus = "ERROR"
                 if callStack := self.__result.pop("CallStack", None):
                     argsString += "\n" + "".join(callStack)
         # If bad HTTP status code
         if self._status_code >= 400:
             argsString = f"ERROR {self._status_code}: {self._reason}"
 
         self.log.notice(
-            "Returning response", f"{credentials} {self._fullComponentName} ({elapsedTime:.2f} ms) {argsString}"
+            "Returning response",
+            f"{credentials} {self._fullComponentName} ({1000.0 * elapsedTime:.2f} ms) {argsString}",
         )
 
+        if self.activityMonitoringReporter:
+            record = {
+                "timestamp": int(TimeUtilities.toEpochMilliSeconds()),
+                "Host": Network.getFQDN(),
+                "ServiceName": "_".join(self._fullComponentName.split("/")),
+                "Location": self.request.uri,
+                "ResponseTime": elapsedTime,
+                # Take the method name from the POST call
+                "MethodName": self.request.arguments.get("method", ["Unknown"])[0],
+                "Protocol": "https",
+                "Status": monitoringRetStatus,
+            }
+
+            self.activityMonitoringReporter.addRecord(record)
+
     def _gatherPeerCredentials(self, grants: list = None) -> dict:
         """Return a dictionary designed to work with the :py:class:`AuthManager <DIRAC.Core.DISET.AuthManager.AuthManager>`,
         already written for DISET and re-used for HTTPS.