feat(DISET): add throttle duration tracking and diagnostic logging

Track when throttling starts and log state transitions with
queue/thread diagnostics. Periodic warnings are emitted every 30s
while throttling persists, making it easier to diagnose stuck
services from logs without needing to attach a debugger.

Adds Service.throttleDiagnostics() to expose queue size, max queue,
active threads, and max threads for logging purposes.

Author: ryuwd

src/DIRAC/Core/DISET/ServiceReactor.py

@@ -35,6 +35,9 @@
 #: This sleep is repeated for as long as Service.wantsThrottle is truthy
 THROTTLE_SERVICE_SLEEP_SECONDS = 0.25
 
+#: Interval between periodic throttle warning messages (seconds)
+THROTTLE_LOG_INTERVAL_SECONDS = 30
+
 
 class ServiceReactor:
     __transportExtraKeywords = {
@@ -200,6 +203,8 @@ def __acceptIncomingConnection(self, svcName=False):
                                   services at the same time
         """
         sel = self.__getListeningSelector(svcName)
+        throttleStartedAt = None
+        lastThrottleLog = 0
         while self.__alive:
             clientTransport = None
             try:
@@ -226,10 +231,30 @@ def __acceptIncomingConnection(self, svcName=False):
             # Handle throttling: reject all connections while overloaded
             # to prevent queue growth when threads are stuck
             if self.__services[svcName].wantsThrottle:
-                gLogger.warn("Rejecting client due to throttling", str(clientTransport.getRemoteAddress()))
+                now = time.time()
+                if throttleStartedAt is None:
+                    throttleStartedAt = now
+                    diag = self.__services[svcName].throttleDiagnostics()
+                    gLogger.warn(
+                        f"Service {svcName} entering throttle mode",
+                        f"queue={diag['queue']}/{diag['maxQueue']}, " f"threads={diag['threads']}/{diag['maxThreads']}",
+                    )
+                    lastThrottleLog = now
+                elif now - lastThrottleLog >= THROTTLE_LOG_INTERVAL_SECONDS:
+                    duration = now - throttleStartedAt
+                    diag = self.__services[svcName].throttleDiagnostics()
+                    gLogger.warn(
+                        f"Service {svcName} still throttling after {duration:.0f}s",
+                        f"queue={diag['queue']}/{diag['maxQueue']}, " f"threads={diag['threads']}/{diag['maxThreads']}",
+                    )
+                    lastThrottleLog = now
                 clientTransport.close()
                 time.sleep(THROTTLE_SERVICE_SLEEP_SECONDS)
                 continue
+            if throttleStartedAt is not None:
+                duration = time.time() - throttleStartedAt
+                gLogger.info(f"Service {svcName} throttle cleared after {duration:.1f}s")
+                throttleStartedAt = None
             # Handle connection
             self.__stats.connectionStablished()
             self.__services[svcName].handleConnection(clientTransport)

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

@@ -286,6 +286,15 @@ def wantsThrottle(self):
         nQueued = self._threadPool._work_queue.qsize()
         return nQueued > self._cfg.getMaxWaitingPetitions()
 
+    def throttleDiagnostics(self):
+        """Return a dict of diagnostics useful for throttle logging"""
+        return {
+            "queue": self._threadPool._work_queue.qsize(),
+            "maxQueue": self._cfg.getMaxWaitingPetitions(),
+            "threads": len(self._threadPool._threads),
+            "maxThreads": self._cfg.getMaxThreads(),
+        }
+
     # Threaded process function
     def _processInThread(self, clientTransport):
         """