Refactor Measure block metrics to be homeserver-scoped (v2) (#18601)

Refactor `Measure` block metrics to be homeserver-scoped (add
`server_name` label to block metrics).

Part of #18592


### Testing strategy

#### See behavior of previous `metrics` listener

 1. Add the `metrics` listener in your `homeserver.yaml`
    ```yaml
    listeners:
      - port: 9323
        type: metrics
        bind_addresses: ['127.0.0.1']
    ```
1. Start the homeserver: `poetry run synapse_homeserver --config-path
homeserver.yaml`
 1. Fetch `http://localhost:9323/metrics`
1. Observe response includes the block metrics
(`synapse_util_metrics_block_count`,
`synapse_util_metrics_block_in_flight`, etc)


#### See behavior of the `http` `metrics` resource

1. Add the `metrics` resource to a new or existing `http` listeners in
your `homeserver.yaml`
    ```yaml
    listeners:
      - port: 9322
        type: http
        bind_addresses: ['127.0.0.1']
        resources:
          - names: [metrics]
            compress: false
    ```
1. Start the homeserver: `poetry run synapse_homeserver --config-path
homeserver.yaml`
1. Fetch `http://localhost:9322/_synapse/metrics` (it's just a `GET`
request so you can even do in the browser)
1. Observe response includes the block metrics
(`synapse_util_metrics_block_count`,
`synapse_util_metrics_block_in_flight`, etc)

Author: MadLittleMods

changelog.d/18601.misc

@@ -0,0 +1 @@
+Refactor `Measure` block metrics to be homeserver-scoped.

synapse/federation/send_queue.py

@@ -156,7 +156,9 @@ def _clear_queue(self) -> None:
 
     def _clear_queue_before_pos(self, position_to_delete: int) -> None:
         """Clear all the queues from before a given position"""
-        with Measure(self.clock, "send_queue._clear"):
+        with Measure(
+            self.clock, name="send_queue._clear", server_name=self.server_name
+        ):
             # Delete things out of presence maps
             keys = self.presence_destinations.keys()
             i = self.presence_destinations.bisect_left(position_to_delete)

synapse/federation/sender/__init__.py

@@ -657,7 +657,11 @@ async def handle_room_events(events: List[EventBase]) -> None:
                     logger.debug(
                         "Handling %i events in room %s", len(events), events[0].room_id
                     )
-                    with Measure(self.clock, "handle_room_events"):
+                    with Measure(
+                        self.clock,
+                        name="handle_room_events",
+                        server_name=self.server_name,
+                    ):
                         for event in events:
                             await handle_event(event)
 

synapse/federation/sender/transaction_manager.py

@@ -58,7 +58,7 @@ class TransactionManager:
     """
 
     def __init__(self, hs: "synapse.server.HomeServer"):
-        self._server_name = hs.hostname
+        self.server_name = hs.hostname  # nb must be called this for @measure_func
         self.clock = hs.get_clock()  # nb must be called this for @measure_func
         self._store = hs.get_datastores().main
         self._transaction_actions = TransactionActions(self._store)
@@ -116,7 +116,7 @@ async def send_new_transaction(
             transaction = Transaction(
                 origin_server_ts=int(self.clock.time_msec()),
                 transaction_id=txn_id,
-                origin=self._server_name,
+                origin=self.server_name,
                 destination=destination,
                 pdus=[p.get_pdu_json() for p in pdus],
                 edus=[edu.get_dict() for edu in edus],

synapse/handlers/appservice.py

@@ -73,6 +73,7 @@
 
 class ApplicationServicesHandler:
     def __init__(self, hs: "HomeServer"):
+        self.server_name = hs.hostname
         self.store = hs.get_datastores().main
         self.is_mine_id = hs.is_mine_id
         self.appservice_api = hs.get_application_service_api()
@@ -120,7 +121,9 @@ def notify_interested_services(self, max_token: RoomStreamToken) -> None:
 
     @wrap_as_background_process("notify_interested_services")
     async def _notify_interested_services(self, max_token: RoomStreamToken) -> None:
-        with Measure(self.clock, "notify_interested_services"):
+        with Measure(
+            self.clock, name="notify_interested_services", server_name=self.server_name
+        ):
             self.is_processing = True
             try:
                 upper_bound = -1
@@ -329,7 +332,11 @@ async def _notify_interested_services_ephemeral(
         users: Collection[Union[str, UserID]],
     ) -> None:
         logger.debug("Checking interested services for %s", stream_key)
-        with Measure(self.clock, "notify_interested_services_ephemeral"):
+        with Measure(
+            self.clock,
+            name="notify_interested_services_ephemeral",
+            server_name=self.server_name,
+        ):
             for service in services:
                 if stream_key == StreamKeyType.TYPING:
                     # Note that we don't persist the token (via set_appservice_stream_type_pos)

synapse/handlers/delayed_events.py

@@ -54,6 +54,7 @@
 
 class DelayedEventsHandler:
     def __init__(self, hs: "HomeServer"):
+        self.server_name = hs.hostname
         self._store = hs.get_datastores().main
         self._storage_controllers = hs.get_storage_controllers()
         self._config = hs.config
@@ -159,7 +160,9 @@ async def _unsafe_process_new_event(self) -> None:
 
         # Loop round handling deltas until we're up to date
         while True:
-            with Measure(self._clock, "delayed_events_delta"):
+            with Measure(
+                self._clock, name="delayed_events_delta", server_name=self.server_name
+            ):
                 room_max_stream_ordering = self._store.get_room_max_stream_ordering()
                 if self._event_pos == room_max_stream_ordering:
                     return

synapse/handlers/device.py

@@ -526,6 +526,8 @@ class DeviceHandler(DeviceWorkerHandler):
     def __init__(self, hs: "HomeServer"):
         super().__init__(hs)
 
+        self.server_name = hs.hostname  # nb must be called this for @measure_func
+        self.clock = hs.get_clock()  # nb must be called this for @measure_func
         self.federation_sender = hs.get_federation_sender()
         self._account_data_handler = hs.get_account_data_handler()
         self._storage_controllers = hs.get_storage_controllers()
@@ -1215,7 +1217,8 @@ class DeviceListUpdater(DeviceListWorkerUpdater):
     def __init__(self, hs: "HomeServer", device_handler: DeviceHandler):
         self.store = hs.get_datastores().main
         self.federation = hs.get_federation_client()
-        self.clock = hs.get_clock()
+        self.server_name = hs.hostname  # nb must be called this for @measure_func
+        self.clock = hs.get_clock()  # nb must be called this for @measure_func
         self.device_handler = device_handler
         self._notifier = hs.get_notifier()
 

synapse/handlers/message.py

@@ -476,16 +476,16 @@ async def _expire_event(self, event_id: str) -> None:
 class EventCreationHandler:
     def __init__(self, hs: "HomeServer"):
         self.hs = hs
+        self.validator = EventValidator()
+        self.event_builder_factory = hs.get_event_builder_factory()
+        self.server_name = hs.hostname  # nb must be called this for @measure_func
+        self.clock = hs.get_clock()  # nb must be called this for @measure_func
         self.auth_blocking = hs.get_auth_blocking()
         self._event_auth_handler = hs.get_event_auth_handler()
         self.store = hs.get_datastores().main
         self._storage_controllers = hs.get_storage_controllers()
         self.state = hs.get_state_handler()
-        self.clock = hs.get_clock()
-        self.validator = EventValidator()
         self.profile_handler = hs.get_profile_handler()
-        self.event_builder_factory = hs.get_event_builder_factory()
-        self.server_name = hs.hostname
         self.notifier = hs.get_notifier()
         self.config = hs.config
         self.require_membership_for_aliases = (

synapse/handlers/presence.py

@@ -747,6 +747,7 @@ async def bump_presence_active_time(
 class PresenceHandler(BasePresenceHandler):
     def __init__(self, hs: "HomeServer"):
         super().__init__(hs)
+        self.server_name = hs.hostname
         self.wheel_timer: WheelTimer[str] = WheelTimer()
         self.notifier = hs.get_notifier()
 
@@ -941,7 +942,9 @@ async def _update_states(
 
         now = self.clock.time_msec()
 
-        with Measure(self.clock, "presence_update_states"):
+        with Measure(
+            self.clock, name="presence_update_states", server_name=self.server_name
+        ):
             # NOTE: We purposefully don't await between now and when we've
             # calculated what we want to do with the new states, to avoid races.
 
@@ -1497,7 +1500,9 @@ async def _process_presence() -> None:
     async def _unsafe_process(self) -> None:
         # Loop round handling deltas until we're up to date
         while True:
-            with Measure(self.clock, "presence_delta"):
+            with Measure(
+                self.clock, name="presence_delta", server_name=self.server_name
+            ):
                 room_max_stream_ordering = self.store.get_room_max_stream_ordering()
                 if self._event_pos == room_max_stream_ordering:
                     return
@@ -1759,6 +1764,7 @@ def __init__(self, hs: "HomeServer"):
         # Same with get_presence_router:
         #
         #   AuthHandler -> Notifier -> PresenceEventSource -> ModuleApi -> AuthHandler
+        self.server_name = hs.hostname
         self.get_presence_handler = hs.get_presence_handler
         self.get_presence_router = hs.get_presence_router
         self.clock = hs.get_clock()
@@ -1792,7 +1798,9 @@ async def get_new_events(
         user_id = user.to_string()
         stream_change_cache = self.store.presence_stream_cache
 
-        with Measure(self.clock, "presence.get_new_events"):
+        with Measure(
+            self.clock, name="presence.get_new_events", server_name=self.server_name
+        ):
             if from_key is not None:
                 from_key = int(from_key)
 

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()
@@ -710,7 +711,9 @@ async def ephemeral_by_room(
 
         sync_config = sync_result_builder.sync_config
 
-        with Measure(self.clock, "ephemeral_by_room"):
+        with Measure(
+            self.clock, name="ephemeral_by_room", server_name=self.server_name
+        ):
             typing_key = since_token.typing_key if since_token else 0
 
             room_ids = sync_result_builder.joined_room_ids
@@ -783,7 +786,9 @@ async def _load_filtered_recents(
                 and current token to send down to clients.
             newly_joined_room
         """
-        with Measure(self.clock, "load_filtered_recents"):
+        with Measure(
+            self.clock, name="load_filtered_recents", server_name=self.server_name
+        ):
             timeline_limit = sync_config.filter_collection.timeline_limit()
             block_all_timeline = (
                 sync_config.filter_collection.blocks_all_room_timeline()
@@ -1174,7 +1179,9 @@ async def compute_state_delta(
         # updates even if they occurred logically before the previous event.
         # TODO(mjark) Check for new redactions in the state events.
 
-        with Measure(self.clock, "compute_state_delta"):
+        with Measure(
+            self.clock, name="compute_state_delta", server_name=self.server_name
+        ):
             # The memberships needed for events in the timeline.
             # Only calculated when `lazy_load_members` is on.
             members_to_fetch: Optional[Set[str]] = None
@@ -1791,7 +1798,9 @@ async def unread_notifs_for_room_id(
             # the DB.
             return RoomNotifCounts.empty()
 
-        with Measure(self.clock, "unread_notifs_for_room_id"):
+        with Measure(
+            self.clock, name="unread_notifs_for_room_id", server_name=self.server_name
+        ):
             return await self.store.get_unread_event_push_actions_by_room_for_user(
                 room_id,
                 sync_config.user.to_string(),