Merge pull request #6886 from chaen/v8.0_FEAT_serviceMonitoring

[8.0 ] Implement actual service monitoring

Author: fstagni

src/DIRAC/Core/DISET/private/Service.py

@@ -555,17 +555,28 @@ def _executeAction(self, trid, proposalTuple, handlerObj):
             response = handlerObj._rh_executeAction(proposalTuple)
             if not response["OK"]:
                 return response
+            retVal = response["Value"][0]
             if self.activityMonitoring:
+                _actionType, actionName = proposalTuple[1]
+                retStatus = "Unknown"
+                if isReturnStructure(retVal):
+                    if retVal["OK"]:
+                        retStatus = "OK"
+                    else:
+                        retStatus = "ERROR"
                 self.activityMonitoringReporter.addRecord(
                     {
                         "timestamp": int(TimeUtilities.toEpochMilliSeconds()),
                         "Host": Network.getFQDN(),
                         "ServiceName": "_".join(self._name.split("/")),
                         "Location": self._cfg.getURL(),
                         "ResponseTime": response["Value"][1],
+                        "MethodName": actionName,
+                        "Protocol": "dips",
+                        "Status": retStatus,
                     }
                 )
-            return response["Value"][0]
+            return retVal
         except Exception as e:
             gLogger.exception("Exception while executing handler action")
             return S_ERROR(f"Server error while executing action: {str(e)}")

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

@@ -176,6 +176,12 @@ def startTornado(self):
         Starts the tornado server when ready.
         This method never returns.
         """
+
+        # If we are running with python3, Tornado will use asyncio,
+        # and we have to convince it to let us run in a different thread
+        # This statement must be placed before setting PeriodicCallback
+        asyncio.set_event_loop_policy(tornado.platform.asyncio.AnyThreadEventLoopPolicy())
+
         # If there is no services loaded:
         if not self.__calculateAppSettings():
             raise Exception("There is no services loaded, please check your configuration")
@@ -205,13 +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()
-
-        # If we are running with python3, Tornado will use asyncio,
-        # and we have to convince it to let us run in a different thread
-        asyncio.set_event_loop_policy(tornado.platform.asyncio.AnyThreadEventLoopPolicy())
+            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]))
@@ -233,7 +233,7 @@ def startTornado(self):
 
         tornado.ioloop.IOLoop.current().start()
 
-    def __reportToMonitoring(self, responseTime):
+    def __reportToMonitoring(self):
         """
         Periodically reports to Monitoring
         """
@@ -248,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.