Merge commit 'dc80a0762' into dinsic

* commit 'dc80a0762':
  1.16.0rc1
  Back out MSC2625 implementation (#7761)
  Additional configuration options for auto-join rooms (#7763)
  Add some metrics for inbound and outbound federation processing times (#7755)
  Explain the purpose of the "tests" conditional dependency requirement (#7751)
  Add another yield point to state res v2 (#7746)
  Move flake8 to end. Don't exit script on failure (#7738)
  Make tox actions work on Debian 10 (#7703)
  Yield during large v2 state res. (#7735)
  add org.matrix.login.jwt so that m.login.jwt can be deprecated (#7675)
  Set Content-Length for Metrics requests (#7730)
  Sync ignored table names in synapse_port_db to current database schema (#7717)
  Allow local media to be marked as safe from being quarantined. (#7718)
  Convert directory handler to async/await (#7727)
  Speed up state res v2 across large state differences. (#7725)

1 files changed, 21 insertions, 16 deletions

diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py

index afe0a8238b..e704cf2f44 100644
--- a/synapse/federation/federation_server.py
+++ b/synapse/federation/federation_server.py
@@ -18,7 +18,7 @@ import logging
 from typing import Any, Callable, Dict, List, Match, Optional, Tuple, Union
 
 from canonicaljson import json
-from prometheus_client import Counter
+from prometheus_client import Counter, Histogram
 
 from twisted.internet import defer
 from twisted.internet.abstract import isIPAddress
@@ -70,6 +70,10 @@ received_queries_counter = Counter(
     "synapse_federation_server_received_queries", "", ["type"]
 )
 
+pdu_process_time = Histogram(
+    "synapse_federation_server_pdu_process_time", "Time taken to process an event",
+)
+
 
 class FederationServer(FederationBase):
     def __init__(self, hs):
@@ -271,21 +275,22 @@ class FederationServer(FederationBase):
 
             for pdu in pdus_by_room[room_id]:
                 event_id = pdu.event_id
-                with nested_logging_context(event_id):
-                    try:
-                        await self._handle_received_pdu(origin, pdu)
-                        pdu_results[event_id] = {}
-                    except FederationError as e:
-                        logger.warning("Error handling PDU %s: %s", event_id, e)
-                        pdu_results[event_id] = {"error": str(e)}
-                    except Exception as e:
-                        f = failure.Failure()
-                        pdu_results[event_id] = {"error": str(e)}
-                        logger.error(
-                            "Failed to handle PDU %s",
-                            event_id,
-                            exc_info=(f.type, f.value, f.getTracebackObject()),
-                        )
+                with pdu_process_time.time():
+                    with nested_logging_context(event_id):
+                        try:
+                            await self._handle_received_pdu(origin, pdu)
+                            pdu_results[event_id] = {}
+                        except FederationError as e:
+                            logger.warning("Error handling PDU %s: %s", event_id, e)
+                            pdu_results[event_id] = {"error": str(e)}
+                        except Exception as e:
+                            f = failure.Failure()
+                            pdu_results[event_id] = {"error": str(e)}
+                            logger.error(
+                                "Failed to handle PDU %s",
+                                event_id,
+                                exc_info=(f.type, f.value, f.getTracebackObject()),
+                            )
 
         await concurrently_execute(
             process_pdus_for_room, pdus_by_room.keys(), TRANSACTION_CONCURRENCY_LIMIT