Add metrics to track how the rate limiter is affecting requests (sleep/reject)

changelog.d/13534.misc

@@ -0,0 +1 @@
+Add metrics to track how the rate limiter is affecting requests (sleep/reject).

synapse/util/ratelimitutils.py

@@ -18,6 +18,8 @@
 import typing
 from typing import Any, DefaultDict, Iterator, List, Set
 
+from prometheus_client.core import Counter
+
 from twisted.internet import defer
 
 from synapse.api.errors import LimitExceededError
@@ -36,6 +38,11 @@
 logger = logging.getLogger(__name__)
 
 
+# Track how much the ratelimiter is affecting requests
+rate_limit_sleep_counter = Counter("synapse_rate_limit_sleep", "")
+rate_limit_reject_counter = Counter("synapse_rate_limit_reject", "")
+
+
 class FederationRateLimiter:
  def __init__(self, clock: Clock, config: FederationRatelimitSettings):
  def new_limiter() -> "_PerHostRatelimiter":
@@ -60,7 +67,7 @@ def ratelimit(self, host: str) -> "_GeneratorContextManager[defer.Deferred[None]
  Returns:
  context manager which returns a deferred.
  """
- return self.ratelimiters[host].ratelimit()
+ return self.ratelimiters[host].ratelimit(host)
 
 
 class _PerHostRatelimiter:
@@ -95,12 +102,14 @@ def __init__(self, clock: Clock, config: FederationRatelimitSettings):
  self.request_times: List[int] = []
 
  @contextlib.contextmanager
- def ratelimit(self) -> "Iterator[defer.Deferred[None]]":
+ def ratelimit(self, host: str) -> "Iterator[defer.Deferred[None]]":
  # `contextlib.contextmanager` takes a generator and turns it into a
  # context manager. The generator should only yield once with a value
  # to be returned by manager.
  # Exceptions will be reraised at the yield.
 
+ self.host = host
+
  request_id = object()
  ret = self._on_enter(request_id)
  try:
@@ -120,6 +129,8 @@ def _on_enter(self, request_id: object) -> "defer.Deferred[None]":
  # sleeping or in the ready queue).
  queue_size = len(self.ready_request_queue) + len(self.sleeping_requests)
  if queue_size > self.reject_limit:
+ logger.debug("Ratelimiter(%s): rejecting request", self.host)
+ rate_limit_reject_counter.inc()
  raise LimitExceededError(
  retry_after_ms=int(self.window_size / self.sleep_limit)
  )
@@ -131,7 +142,8 @@ def queue_request() -> "defer.Deferred[None]":
  queue_defer: defer.Deferred[None] = defer.Deferred()
  self.ready_request_queue[request_id] = queue_defer
  logger.info(
- "Ratelimiter: queueing request (queue now %i items)",
+ "Ratelimiter(%s): queueing request (queue now %i items)",
+ self.host,
  len(self.ready_request_queue),
  )
 
@@ -140,19 +152,28 @@ def queue_request() -> "defer.Deferred[None]":
  return defer.succeed(None)
 
  logger.debug(
- "Ratelimit [%s]: len(self.request_times)=%d",
+ "Ratelimit(%s) [%s]: len(self.request_times)=%d",
+ self.host,
  id(request_id),
  len(self.request_times),
  )
 
  if len(self.request_times) > self.sleep_limit:
- logger.debug("Ratelimiter: sleeping request for %f sec", self.sleep_sec)
+ logger.debug(
+ "Ratelimiter(%s) [%s]: sleeping request for %f sec",
+ self.host,
+ id(request_id),
+ self.sleep_sec,
+ )
+ rate_limit_sleep_counter.inc()
  ret_defer = run_in_background(self.clock.sleep, self.sleep_sec)
 
  self.sleeping_requests.add(request_id)
 
  def on_wait_finished(_: Any) -> "defer.Deferred[None]":
- logger.debug("Ratelimit [%s]: Finished sleeping", id(request_id))
+ logger.debug(
+ "Ratelimit(%s) [%s]: Finished sleeping", self.host, id(request_id)
+ )
  self.sleeping_requests.discard(request_id)
  queue_defer = queue_request()
  return queue_defer
@@ -162,7 +183,9 @@ def on_wait_finished(_: Any) -> "defer.Deferred[None]":
  ret_defer = queue_request()
 
  def on_start(r: object) -> object:
- logger.debug("Ratelimit [%s]: Processing req", id(request_id))
+ logger.debug(
+ "Ratelimit(%s) [%s]: Processing req", self.host, id(request_id)
+ )
  self.current_processing.add(request_id)
  return r
 
@@ -187,7 +210,7 @@ def on_both(r: object) -> object:
  return make_deferred_yieldable(ret_defer)
 
  def _on_exit(self, request_id: object) -> None:
- logger.debug("Ratelimit [%s]: Processed req", id(request_id))
+ logger.debug("Ratelimit(%s) [%s]: Processed req", self.host, id(request_id))
  self.current_processing.discard(request_id)
  try:
  # start processing the next item on the queue.