Refactor cache metrics to be homeserver-scoped

changelog.d/18604.misc

@@ -0,0 +1 @@
+Refactor cache metrics to be homeserver-scoped.

synapse/api/auth/base.py

@@ -172,7 +172,7 @@ async def validate_appservice_can_control_user_id(
         """
 
         # It's ok if the app service is trying to use the sender from their registration
-        if app_service.sender == user_id:
+        if app_service.sender.to_string() == user_id:
             pass
         # Check to make sure the app service is allowed to control the user
         elif not app_service.is_interested_in_user(user_id):

synapse/api/auth/msc3861_delegated.py

@@ -176,6 +176,7 @@ def __init__(self, hs: "HomeServer"):
         assert self._config.client_id, "No client_id provided"
         assert auth_method is not None, "Invalid client_auth_method provided"
 
+        self.server_name = hs.hostname
         self._clock = hs.get_clock()
         self._http_client = hs.get_proxied_http_client()
         self._hostname = hs.hostname
@@ -206,8 +207,9 @@ def __init__(self, hs: "HomeServer"):
         #   In this case, the device still exists and it's not the end of the world for
         #   the old access token to continue working for a short time.
         self._introspection_cache: ResponseCache[str] = ResponseCache(
-            self._clock,
-            "token_introspection",
+            clock=self._clock,
+            name="token_introspection",
+            server_name=self.server_name,
             timeout_ms=120_000,
             # don't log because the keys are access tokens
             enable_logging=False,

synapse/appservice/__init__.py

@@ -78,7 +78,7 @@ def __init__(
         self,
         token: str,
         id: str,
-        sender: str,
+        sender: UserID,
         url: Optional[str] = None,
         namespaces: Optional[JsonDict] = None,
         hs_token: Optional[str] = None,
@@ -96,6 +96,8 @@ def __init__(
         self.hs_token = hs_token
         # The full Matrix ID for this application service's sender.
         self.sender = sender
+        # The application service user should be part of the server's domain.
+        self.server_name = sender.domain
         self.namespaces = self._check_namespaces(namespaces)
         self.id = id
         self.ip_range_whitelist = ip_range_whitelist
@@ -223,7 +225,7 @@ def is_interested_in_user(
         """
         return (
             # User is the appservice's configured sender_localpart user
-            user_id == self.sender
+            user_id == self.sender.to_string()
             # User is in the appservice's user namespace
             or self.is_user_in_namespace(user_id)
         )
@@ -347,7 +349,7 @@ def is_room_id_in_namespace(self, room_id: str) -> bool:
     def is_exclusive_user(self, user_id: str) -> bool:
         return (
             self._is_exclusive(ApplicationService.NS_USERS, user_id)
-            or user_id == self.sender
+            or user_id == self.sender.to_string()
         )
 
     def is_interested_in_protocol(self, protocol: str) -> bool:

synapse/appservice/api.py

@@ -126,11 +126,15 @@ class ApplicationServiceApi(SimpleHttpClient):
 
     def __init__(self, hs: "HomeServer"):
         super().__init__(hs)
+        self.server_name = hs.hostname
         self.clock = hs.get_clock()
         self.config = hs.config.appservice
 
         self.protocol_meta_cache: ResponseCache[Tuple[str, str]] = ResponseCache(
-            hs.get_clock(), "as_protocol_meta", timeout_ms=HOUR_IN_MS
+            clock=hs.get_clock(),
+            name="as_protocol_meta",
+            server_name=self.server_name,
+            timeout_ms=HOUR_IN_MS,
         )
 
     def _get_headers(self, service: "ApplicationService") -> Dict[bytes, List[bytes]]:

synapse/appservice/scheduler.py

@@ -319,7 +319,7 @@ async def _compute_msc3202_otk_counts_and_fallback_keys(
         users: Set[str] = set()
 
         # The sender is always included
-        users.add(service.sender)
+        users.add(service.sender.to_string())
 
         # All AS users that would receive the PDUs or EDUs sent to these rooms
         # are classed as 'interesting'.

synapse/config/appservice.py

@@ -122,8 +122,7 @@ def _load_appservice(
     localpart = as_info["sender_localpart"]
     if urlparse.quote(localpart) != localpart:
         raise ValueError("sender_localpart needs characters which are not URL encoded.")
-    user = UserID(localpart, hostname)
-    user_id = user.to_string()
+    user_id = UserID(localpart, hostname)
 
     # Rate limiting for users of this AS is on by default (excludes sender)
     rate_limited = as_info.get("rate_limited")

synapse/federation/federation_client.py

@@ -137,13 +137,14 @@ def __init__(self, hs: "HomeServer"):
         self.state = hs.get_state_handler()
         self.transport_layer = hs.get_federation_transport_client()
 
-        self.hostname = hs.hostname
+        self.server_name = hs.hostname
         self.signing_key = hs.signing_key
 
         # Cache mapping `event_id` to a tuple of the event itself and the `pull_origin`
         # (which server we pulled the event from)
         self._get_pdu_cache: ExpiringCache[str, Tuple[EventBase, str]] = ExpiringCache(
             cache_name="get_pdu_cache",
+            server_name=self.server_name,
             clock=self._clock,
             max_len=1000,
             expiry_ms=120 * 1000,
@@ -162,6 +163,7 @@ def __init__(self, hs: "HomeServer"):
             Tuple[JsonDict, Sequence[JsonDict], Sequence[JsonDict], Sequence[str]],
         ] = ExpiringCache(
             cache_name="get_room_hierarchy_cache",
+            server_name=self.server_name,
             clock=self._clock,
             max_len=1000,
             expiry_ms=5 * 60 * 1000,
@@ -1068,7 +1070,7 @@ async def send_request(destination: str) -> Tuple[str, EventBase, RoomVersion]:
             # there's some we never care about
             ev = builder.create_local_event_from_event_dict(
                 self._clock,
-                self.hostname,
+                self.server_name,
                 self.signing_key,
                 room_version=room_version,
                 event_dict=pdu_dict,

synapse/federation/federation_server.py

@@ -160,7 +160,10 @@ def __init__(self, hs: "HomeServer"):
 
         # We cache results for transaction with the same ID
         self._transaction_resp_cache: ResponseCache[Tuple[str, str]] = ResponseCache(
-            hs.get_clock(), "fed_txn_handler", timeout_ms=30000
+            clock=hs.get_clock(),
+            name="fed_txn_handler",
+            server_name=self.server_name,
+            timeout_ms=30000,
         )
 
         self.transaction_actions = TransactionActions(self.store)
@@ -170,10 +173,18 @@ def __init__(self, hs: "HomeServer"):
         # We cache responses to state queries, as they take a while and often
         # come in waves.
         self._state_resp_cache: ResponseCache[Tuple[str, Optional[str]]] = (
-            ResponseCache(hs.get_clock(), "state_resp", timeout_ms=30000)
+            ResponseCache(
+                clock=hs.get_clock(),
+                name="state_resp",
+                server_name=self.server_name,
+                timeout_ms=30000,
+            )
         )
         self._state_ids_resp_cache: ResponseCache[Tuple[str, str]] = ResponseCache(
-            hs.get_clock(), "state_ids_resp", timeout_ms=30000
+            clock=hs.get_clock(),
+            name="state_ids_resp",
+            server_name=self.server_name,
+            timeout_ms=30000,
         )
 
         self._federation_metrics_domains = (

synapse/handlers/appservice.py

@@ -839,7 +839,7 @@ async def _is_unknown_user(self, user_id: str) -> bool:
 
         # user not found; could be the AS though, so check.
         services = self.store.get_app_services()
-        service_list = [s for s in services if s.sender == user_id]
+        service_list = [s for s in services if s.sender.to_string() == user_id]
         return len(service_list) == 0
 
     async def _check_user_exists(self, user_id: str) -> bool:

synapse/handlers/device.py

@@ -1213,6 +1213,7 @@ class DeviceListUpdater(DeviceListWorkerUpdater):
     "Handles incoming device list updates from federation and updates the DB"
 
     def __init__(self, hs: "HomeServer", device_handler: DeviceHandler):
+        self.server_name = hs.hostname
         self.store = hs.get_datastores().main
         self.federation = hs.get_federation_client()
         self.clock = hs.get_clock()
@@ -1232,6 +1233,7 @@ def __init__(self, hs: "HomeServer", device_handler: DeviceHandler):
         # resyncs.
         self._seen_updates: ExpiringCache[str, Set[str]] = ExpiringCache(
             cache_name="device_update_edu",
+            server_name=self.server_name,
             clock=self.clock,
             max_len=10000,
             expiry_ms=30 * 60 * 1000,

synapse/handlers/directory.py

@@ -406,7 +406,7 @@ def can_modify_alias(self, alias: RoomAlias, user_id: Optional[str] = None) -> b
         ]
 
         for service in interested_services:
-            if user_id == service.sender:
+            if user_id == service.sender.to_string():
                 # this user IS the app service so they can do whatever they like
                 return True
             elif service.is_exclusive_alias(alias.to_string()):

synapse/handlers/initial_sync.py

@@ -60,6 +60,7 @@
 
 class InitialSyncHandler:
     def __init__(self, hs: "HomeServer"):
+        self.server_name = hs.hostname
         self.store = hs.get_datastores().main
         self.auth = hs.get_auth()
         self.state_handler = hs.get_state_handler()
@@ -77,7 +78,11 @@ def __init__(self, hs: "HomeServer"):
                 bool,
                 bool,
             ]
-        ] = ResponseCache(hs.get_clock(), "initial_sync_cache")
+        ] = ResponseCache(
+            clock=hs.get_clock(),
+            name="initial_sync_cache",
+            server_name=self.server_name,
+        )
         self._event_serializer = hs.get_event_client_serializer()
         self._storage_controllers = hs.get_storage_controllers()
         self._state_storage_controller = self._storage_controllers.state

synapse/handlers/message.py

@@ -558,8 +558,9 @@ def __init__(self, hs: "HomeServer"):
         self._external_cache_joined_hosts_updates: Optional[ExpiringCache] = None
         if self._external_cache.is_enabled():
             self._external_cache_joined_hosts_updates = ExpiringCache(
-                "_external_cache_joined_hosts_updates",
-                self.clock,
+                cache_name="_external_cache_joined_hosts_updates",
+                server_name=self.server_name,
+                clock=self.clock,
                 expiry_ms=30 * 60 * 1000,
             )
 

synapse/handlers/profile.py

@@ -55,6 +55,7 @@ class ProfileHandler:
     """
 
     def __init__(self, hs: "HomeServer"):
+        self.server_name = hs.hostname
         self.store = hs.get_datastores().main
         self.clock = hs.get_clock()
         self.hs = hs

synapse/handlers/room.py

@@ -118,6 +118,7 @@ class EventContext:
 
 class RoomCreationHandler:
     def __init__(self, hs: "HomeServer"):
+        self.server_name = hs.hostname
         self.store = hs.get_datastores().main
         self._storage_controllers = hs.get_storage_controllers()
         self.auth = hs.get_auth()
@@ -174,7 +175,10 @@ def __init__(self, hs: "HomeServer"):
         # succession, only process the first attempt and return its result to
         # subsequent requests
         self._upgrade_response_cache: ResponseCache[Tuple[str, str]] = ResponseCache(
-            hs.get_clock(), "room_upgrade", timeout_ms=FIVE_MINUTES_IN_MS
+            clock=hs.get_clock(),
+            name="room_upgrade",
+            server_name=self.server_name,
+            timeout_ms=FIVE_MINUTES_IN_MS,
         )
         self._server_notices_mxid = hs.config.servernotices.server_notices_mxid
 

synapse/handlers/room_list.py

@@ -61,16 +61,26 @@
 
 class RoomListHandler:
     def __init__(self, hs: "HomeServer"):
+        self.server_name = hs.hostname
         self.store = hs.get_datastores().main
         self._storage_controllers = hs.get_storage_controllers()
         self.hs = hs
         self.enable_room_list_search = hs.config.roomdirectory.enable_room_list_search
         self.response_cache: ResponseCache[
             Tuple[Optional[int], Optional[str], Optional[ThirdPartyInstanceID]]
-        ] = ResponseCache(hs.get_clock(), "room_list")
+        ] = ResponseCache(
+            clock=hs.get_clock(),
+            name="room_list",
+            server_name=self.server_name,
+        )
         self.remote_response_cache: ResponseCache[
             Tuple[str, Optional[int], Optional[str], bool, Optional[str]]
-        ] = ResponseCache(hs.get_clock(), "remote_room_list", timeout_ms=30 * 1000)
+        ] = ResponseCache(
+            clock=hs.get_clock(),
+            name="remote_room_list",
+            server_name=self.server_name,
+            timeout_ms=30 * 1000,
+        )
 
     async def get_local_public_room_list(
         self,

synapse/handlers/room_summary.py

@@ -96,6 +96,7 @@ class RoomSummaryHandler:
     _PAGINATION_SESSION_VALIDITY_PERIOD_MS = 5 * 60 * 1000
 
     def __init__(self, hs: "HomeServer"):
+        self.server_name = hs.hostname
         self._event_auth_handler = hs.get_event_auth_handler()
         self._store = hs.get_datastores().main
         self._storage_controllers = hs.get_storage_controllers()
@@ -121,8 +122,9 @@ def __init__(self, hs: "HomeServer"):
                 Optional[Tuple[str, ...]],
             ]
         ] = ResponseCache(
-            hs.get_clock(),
-            "get_room_hierarchy",
+            clock=hs.get_clock(),
+            name="get_room_hierarchy",
+            server_name=self.server_name,
         )
         self._msc3266_enabled = hs.config.experimental.msc3266_enabled
 

synapse/handlers/sync.py

@@ -329,6 +329,7 @@ class E2eeSyncResult:
 
 class SyncHandler:
     def __init__(self, hs: "HomeServer"):
+        self.server_name = hs.hostname
         self.hs_config = hs.config
         self.store = hs.get_datastores().main
         self.notifier = hs.get_notifier()
@@ -352,17 +353,19 @@ def __init__(self, hs: "HomeServer"):
         #    cached result any more, and we could flush the entry from the cache to save
         #    memory.
         self.response_cache: ResponseCache[SyncRequestKey] = ResponseCache(
-            hs.get_clock(),
-            "sync",
+            clock=hs.get_clock(),
+            name="sync",
+            server_name=self.server_name,
             timeout_ms=hs.config.caches.sync_response_cache_duration,
         )
 
         # ExpiringCache((User, Device)) -> LruCache(user_id => event_id)
         self.lazy_loaded_members_cache: ExpiringCache[
             Tuple[str, Optional[str]], LruCache[str, str]
         ] = ExpiringCache(
-            "lazy_loaded_members_cache",
-            self.clock,
+            cache_name="lazy_loaded_members_cache",
+            server_name=self.server_name,
+            clock=self.clock,
             max_len=0,
             expiry_ms=LAZY_LOADED_MEMBERS_CACHE_MAX_AGE,
         )
@@ -1129,7 +1132,7 @@ def get_lazy_loaded_members_cache(
         )
         if cache is None:
             logger.debug("creating LruCache for %r", cache_key)
-            cache = LruCache(LAZY_LOADED_MEMBERS_CACHE_MAX_SIZE)
+            cache = LruCache(max_size=LAZY_LOADED_MEMBERS_CACHE_MAX_SIZE)
             self.lazy_loaded_members_cache[cache_key] = cache
         else:
             logger.debug("found LruCache for %r", cache_key)

synapse/handlers/typing.py

@@ -263,6 +263,7 @@ def __init__(self, hs: "HomeServer"):
 
         assert hs.get_instance_name() in hs.config.worker.writers.typing
 
+        self.server_name = hs.hostname
         self.auth = hs.get_auth()
         self.notifier = hs.get_notifier()
         self.event_auth_handler = hs.get_event_auth_handler()
@@ -280,7 +281,9 @@ def __init__(self, hs: "HomeServer"):
 
         # caches which room_ids changed at which serials
         self._typing_stream_change_cache = StreamChangeCache(
-            "TypingStreamChangeCache", self._latest_room_serial
+            name="TypingStreamChangeCache",
+            server_name=self.server_name,
+            current_stream_pos=self._latest_room_serial,
         )
 
     def _handle_timeout_for_member(self, now: int, member: RoomMember) -> None: