diff options
| -rw-r--r-- | synapse/federation/transaction_queue.py | 16 |
1 files changed, 14 insertions, 2 deletions
diff --git a/synapse/federation/transaction_queue.py b/synapse/federation/transaction_queue.py index 9dc7849b17..b9d3f89324 100644 --- a/synapse/federation/transaction_queue.py +++ b/synapse/federation/transaction_queue.py @@ -25,12 +25,15 @@ from synapse.util.logcontext import PreserveLoggingContext from synapse.util.retryutils import ( get_retry_limiter, NotRetryingDestination, ) +import synapse.metrics import logging logger = logging.getLogger(__name__) +metrics = synapse.metrics.get_metrics_for(__name__) + class TransactionQueue(object): """This class makes sure we only have one transaction in flight at @@ -56,9 +59,9 @@ class TransactionQueue(object): # Is a mapping from destination -> list of # tuple(pending pdus, deferred, order) - self.pending_pdus_by_dest = {} + self.pending_pdus_by_dest = pdus = {} # destination -> list of tuple(edu, deferred) - self.pending_edus_by_dest = {} + self.pending_edus_by_dest = edus = {} # destination -> list of tuple(failure, deferred) self.pending_failures_by_dest = {} @@ -66,6 +69,15 @@ class TransactionQueue(object): # HACK to get unique tx id self._next_txn_id = int(self._clock.time_msec()) + metrics.register_callback("pending_pdus", + lambda: {(dest,): len(pdus[dest]) for dest in pdus.keys()}, + keys=["dest"], + ) + metrics.register_callback("pending_edus", + lambda: {(dest,): len(edus[dest]) for dest in edus.keys()}, + keys=["dest"], + ) + def can_send_to(self, destination): """Can we send messages to the given server? |