Merge branch 'develop' into madlittlemods/18592-refactor-histogram

Conflicts:
	scripts-dev/mypy_synapse_plugin.py
	synapse/http/request_metrics.py
	synapse/replication/tcp/external_cache.py
	synapse/storage/database.py

Author: MadLittleMods

changelog.d/18656.misc

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

scripts-dev/mypy_synapse_plugin.py

@@ -60,6 +60,7 @@ def get_function_signature_hook(
         self, fullname: str
     ) -> Optional[Callable[[FunctionSigContext], FunctionLike]]:
         if fullname in (
+            "prometheus_client.metrics.Counter",
             "prometheus_client.metrics.Histogram",
             # TODO: Add other prometheus_client metrics that need checking as we
             # refactor, see https://github.com/element-hq/synapse/issues/18592

synapse/appservice/api.py

@@ -48,6 +48,7 @@
 from synapse.events.utils import SerializeEventConfig, serialize_event
 from synapse.http.client import SimpleHttpClient, is_unknown_endpoint
 from synapse.logging import opentracing
+from synapse.metrics import SERVER_NAME_LABEL
 from synapse.types import DeviceListUpdates, JsonDict, JsonMapping, ThirdPartyInstanceID
 from synapse.util.caches.response_cache import ResponseCache
 
@@ -59,29 +60,31 @@
 sent_transactions_counter = Counter(
     "synapse_appservice_api_sent_transactions",
     "Number of /transactions/ requests sent",
-    ["service"],
+    labelnames=["service", SERVER_NAME_LABEL],
 )
 
 failed_transactions_counter = Counter(
     "synapse_appservice_api_failed_transactions",
     "Number of /transactions/ requests that failed to send",
-    ["service"],
+    labelnames=["service", SERVER_NAME_LABEL],
 )
 
 sent_events_counter = Counter(
-    "synapse_appservice_api_sent_events", "Number of events sent to the AS", ["service"]
+    "synapse_appservice_api_sent_events",
+    "Number of events sent to the AS",
+    labelnames=["service", SERVER_NAME_LABEL],
 )
 
 sent_ephemeral_counter = Counter(
     "synapse_appservice_api_sent_ephemeral",
     "Number of ephemeral events sent to the AS",
-    ["service"],
+    labelnames=["service", SERVER_NAME_LABEL],
 )
 
 sent_todevice_counter = Counter(
     "synapse_appservice_api_sent_todevice",
     "Number of todevice messages sent to the AS",
-    ["service"],
+    labelnames=["service", SERVER_NAME_LABEL],
 )
 
 HOUR_IN_MS = 60 * 60 * 1000
@@ -382,6 +385,7 @@ async def push_bulk(
                     "left": list(device_list_summary.left),
                 }
 
+        labels = {"service": service.id, SERVER_NAME_LABEL: self.server_name}
         try:
             args = None
             if self.config.use_appservice_legacy_authorization:
@@ -399,10 +403,10 @@ async def push_bulk(
                     service.url,
                     [event.get("event_id") for event in events],
                 )
-            sent_transactions_counter.labels(service.id).inc()
-            sent_events_counter.labels(service.id).inc(len(serialized_events))
-            sent_ephemeral_counter.labels(service.id).inc(len(ephemeral))
-            sent_todevice_counter.labels(service.id).inc(len(to_device_messages))
+            sent_transactions_counter.labels(**labels).inc()
+            sent_events_counter.labels(**labels).inc(len(serialized_events))
+            sent_ephemeral_counter.labels(**labels).inc(len(ephemeral))
+            sent_todevice_counter.labels(**labels).inc(len(to_device_messages))
             return True
         except CodeMessageException as e:
             logger.warning(
@@ -421,7 +425,7 @@ async def push_bulk(
                 ex.args,
                 exc_info=logger.isEnabledFor(logging.DEBUG),
             )
-        failed_transactions_counter.labels(service.id).inc()
+        failed_transactions_counter.labels(**labels).inc()
         return False
 
     async def claim_client_keys(

synapse/federation/federation_client.py

@@ -74,6 +74,7 @@
 from synapse.http.client import is_unknown_endpoint
 from synapse.http.types import QueryParams
 from synapse.logging.opentracing import SynapseTags, log_kv, set_tag, tag_args, trace
+from synapse.metrics import SERVER_NAME_LABEL
 from synapse.types import JsonDict, StrCollection, UserID, get_domain_from_id
 from synapse.types.handlers.policy_server import RECOMMENDATION_OK, RECOMMENDATION_SPAM
 from synapse.util.async_helpers import concurrently_execute
@@ -85,7 +86,9 @@
 
 logger = logging.getLogger(__name__)
 
-sent_queries_counter = Counter("synapse_federation_client_sent_queries", "", ["type"])
+sent_queries_counter = Counter(
+    "synapse_federation_client_sent_queries", "", labelnames=["type", SERVER_NAME_LABEL]
+)
 
 
 PDU_RETRY_TIME_MS = 1 * 60 * 1000
@@ -209,7 +212,10 @@ async def make_query(
         Returns:
             The JSON object from the response
         """
-        sent_queries_counter.labels(query_type).inc()
+        sent_queries_counter.labels(
+            type=query_type,
+            **{SERVER_NAME_LABEL: self.server_name},
+        ).inc()
 
         return await self.transport_layer.make_query(
             destination,
@@ -231,7 +237,10 @@ async def query_client_keys(
         Returns:
             The JSON object from the response
         """
-        sent_queries_counter.labels("client_device_keys").inc()
+        sent_queries_counter.labels(
+            type="client_device_keys",
+            **{SERVER_NAME_LABEL: self.server_name},
+        ).inc()
         return await self.transport_layer.query_client_keys(
             destination, content, timeout
         )
@@ -242,7 +251,10 @@ async def query_user_devices(
         """Query the device keys for a list of user ids hosted on a remote
         server.
         """
-        sent_queries_counter.labels("user_devices").inc()
+        sent_queries_counter.labels(
+            type="user_devices",
+            **{SERVER_NAME_LABEL: self.server_name},
+        ).inc()
         return await self.transport_layer.query_user_devices(
             destination, user_id, timeout
         )
@@ -264,7 +276,10 @@ async def claim_client_keys(
         Returns:
             The JSON object from the response
         """
-        sent_queries_counter.labels("client_one_time_keys").inc()
+        sent_queries_counter.labels(
+            type="client_one_time_keys",
+            **{SERVER_NAME_LABEL: self.server_name},
+        ).inc()
 
         # Convert the query with counts into a stable and unstable query and check
         # if attempting to claim more than 1 OTK.

synapse/federation/federation_server.py

@@ -105,12 +105,18 @@
 
 logger = logging.getLogger(__name__)
 
-received_pdus_counter = Counter("synapse_federation_server_received_pdus", "")
+received_pdus_counter = Counter(
+    "synapse_federation_server_received_pdus", "", labelnames=[SERVER_NAME_LABEL]
+)
 
-received_edus_counter = Counter("synapse_federation_server_received_edus", "")
+received_edus_counter = Counter(
+    "synapse_federation_server_received_edus", "", labelnames=[SERVER_NAME_LABEL]
+)
 
 received_queries_counter = Counter(
-    "synapse_federation_server_received_queries", "", ["type"]
+    "synapse_federation_server_received_queries",
+    "",
+    labelnames=["type", SERVER_NAME_LABEL],
 )
 
 pdu_process_time = Histogram(
@@ -436,7 +442,9 @@ async def _handle_pdus_in_txn(
             report back to the sending server.
         """
 
-        received_pdus_counter.inc(len(transaction.pdus))
+        received_pdus_counter.labels(**{SERVER_NAME_LABEL: self.server_name}).inc(
+            len(transaction.pdus)
+        )
 
         origin_host, _ = parse_server_name(origin)
 
@@ -555,7 +563,7 @@ async def _handle_edus_in_txn(self, origin: str, transaction: Transaction) -> No
         """Process the EDUs in a received transaction."""
 
         async def _process_edu(edu_dict: JsonDict) -> None:
-            received_edus_counter.inc()
+            received_edus_counter.labels(**{SERVER_NAME_LABEL: self.server_name}).inc()
 
             edu = Edu(
                 origin=origin,
@@ -670,7 +678,10 @@ async def on_pdu_request(
     async def on_query_request(
         self, query_type: str, args: Dict[str, str]
     ) -> Tuple[int, Dict[str, Any]]:
-        received_queries_counter.labels(query_type).inc()
+        received_queries_counter.labels(
+            type=query_type,
+            **{SERVER_NAME_LABEL: self.server_name},
+        ).inc()
         resp = await self.registry.on_query(query_type, args)
         return 200, resp
 

synapse/federation/sender/__init__.py

@@ -190,11 +190,13 @@
 sent_pdus_destination_dist_count = Counter(
     "synapse_federation_client_sent_pdu_destinations_count",
     "Number of PDUs queued for sending to one or more destinations",
+    labelnames=[SERVER_NAME_LABEL],
 )
 
 sent_pdus_destination_dist_total = Counter(
     "synapse_federation_client_sent_pdu_destinations",
     "Total number of PDUs queued for sending across all destinations",
+    labelnames=[SERVER_NAME_LABEL],
 )
 
 # Time (in s) to wait before trying to wake up destinations that have
@@ -710,13 +712,19 @@ async def handle_room_events(events: List[EventBase]) -> None:
                         "federation_sender"
                     ).set(ts)
 
-                    events_processed_counter.inc(len(event_entries))
+                    events_processed_counter.labels(
+                        **{SERVER_NAME_LABEL: self.server_name}
+                    ).inc(len(event_entries))
 
-                    event_processing_loop_room_count.labels("federation_sender").inc(
-                        len(events_by_room)
-                    )
+                    event_processing_loop_room_count.labels(
+                        name="federation_sender",
+                        **{SERVER_NAME_LABEL: self.server_name},
+                    ).inc(len(events_by_room))
 
-                event_processing_loop_counter.labels("federation_sender").inc()
+                event_processing_loop_counter.labels(
+                    name="federation_sender",
+                    **{SERVER_NAME_LABEL: self.server_name},
+                ).inc()
 
                 synapse.metrics.event_processing_positions.labels(
                     "federation_sender"
@@ -737,8 +745,12 @@ async def _send_pdu(self, pdu: EventBase, destinations: Iterable[str]) -> None:
         if not destinations:
             return
 
-        sent_pdus_destination_dist_total.inc(len(destinations))
-        sent_pdus_destination_dist_count.inc()
+        sent_pdus_destination_dist_total.labels(
+            **{SERVER_NAME_LABEL: self.server_name}
+        ).inc(len(destinations))
+        sent_pdus_destination_dist_count.labels(
+            **{SERVER_NAME_LABEL: self.server_name}
+        ).inc()
 
         assert pdu.internal_metadata.stream_ordering
 

synapse/federation/sender/per_destination_queue.py

@@ -40,7 +40,7 @@
 from synapse.handlers.presence import format_user_presence_state
 from synapse.logging import issue9533_logger
 from synapse.logging.opentracing import SynapseTags, set_tag
-from synapse.metrics import sent_transactions_counter
+from synapse.metrics import SERVER_NAME_LABEL, sent_transactions_counter
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.types import JsonDict, ReadReceipt
 from synapse.util.retryutils import NotRetryingDestination, get_retry_limiter
@@ -56,13 +56,15 @@
 
 
 sent_edus_counter = Counter(
-    "synapse_federation_client_sent_edus", "Total number of EDUs successfully sent"
+    "synapse_federation_client_sent_edus",
+    "Total number of EDUs successfully sent",
+    labelnames=[SERVER_NAME_LABEL],
 )
 
 sent_edus_by_type = Counter(
     "synapse_federation_client_sent_edus_by_type",
     "Number of sent EDUs successfully sent, by event type",
-    ["type"],
+    labelnames=["type", SERVER_NAME_LABEL],
 )
 
 
@@ -368,10 +370,17 @@ async def _transaction_transmission_loop(self) -> None:
                         self._destination, pending_pdus, pending_edus
                     )
 
-                    sent_transactions_counter.inc()
-                    sent_edus_counter.inc(len(pending_edus))
+                    sent_transactions_counter.labels(
+                        **{SERVER_NAME_LABEL: self.server_name}
+                    ).inc()
+                    sent_edus_counter.labels(
+                        **{SERVER_NAME_LABEL: self.server_name}
+                    ).inc(len(pending_edus))
                     for edu in pending_edus:
-                        sent_edus_by_type.labels(edu.edu_type).inc()
+                        sent_edus_by_type.labels(
+                            type=edu.edu_type,
+                            **{SERVER_NAME_LABEL: self.server_name},
+                        ).inc()
 
         except NotRetryingDestination as e:
             logger.debug(
@@ -596,7 +605,9 @@ async def _catch_up_transmission_loop(self) -> None:
                     self._destination, room_catchup_pdus, []
                 )
 
-                sent_transactions_counter.inc()
+                sent_transactions_counter.labels(
+                    **{SERVER_NAME_LABEL: self.server_name}
+                ).inc()
 
                 # We pulled this from the DB, so it'll be non-null
                 assert pdu.internal_metadata.stream_ordering

synapse/handlers/appservice.py

@@ -69,7 +69,9 @@
 
 logger = logging.getLogger(__name__)
 
-events_processed_counter = Counter("synapse_handlers_appservice_events_processed", "")
+events_processed_counter = Counter(
+    "synapse_handlers_appservice_events_processed", "", labelnames=[SERVER_NAME_LABEL]
+)
 
 
 class ApplicationServicesHandler:
@@ -209,13 +211,19 @@ async def handle_room_events(events: Iterable[EventBase]) -> None:
                         "appservice_sender"
                     ).set(upper_bound)
 
-                    events_processed_counter.inc(len(events))
+                    events_processed_counter.labels(
+                        **{SERVER_NAME_LABEL: self.server_name}
+                    ).inc(len(events))
 
-                    event_processing_loop_room_count.labels("appservice_sender").inc(
-                        len(events_by_room)
-                    )
+                    event_processing_loop_room_count.labels(
+                        name="appservice_sender",
+                        **{SERVER_NAME_LABEL: self.server_name},
+                    ).inc(len(events_by_room))
 
-                    event_processing_loop_counter.labels("appservice_sender").inc()
+                    event_processing_loop_counter.labels(
+                        name="appservice_sender",
+                        **{SERVER_NAME_LABEL: self.server_name},
+                    ).inc()
 
                     if events:
                         now = self.clock.time_msec()

synapse/handlers/auth.py

@@ -70,6 +70,7 @@
 from synapse.http.server import finish_request, respond_with_html
 from synapse.http.site import SynapseRequest
 from synapse.logging.context import defer_to_thread
+from synapse.metrics import SERVER_NAME_LABEL
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.storage.databases.main.registration import (
     LoginTokenExpired,
@@ -95,7 +96,7 @@
 invalid_login_token_counter = Counter(
     "synapse_user_login_invalid_login_tokens",
     "Counts the number of rejected m.login.token on /login",
-    ["reason"],
+    labelnames=["reason", SERVER_NAME_LABEL],
 )
 
 
@@ -1478,11 +1479,20 @@ async def consume_login_token(self, login_token: str) -> LoginTokenLookupResult:
         try:
             return await self.store.consume_login_token(login_token)
         except LoginTokenExpired:
-            invalid_login_token_counter.labels("expired").inc()
+            invalid_login_token_counter.labels(
+                reason="expired",
+                **{SERVER_NAME_LABEL: self.server_name},
+            ).inc()
         except LoginTokenReused:
-            invalid_login_token_counter.labels("reused").inc()
+            invalid_login_token_counter.labels(
+                reason="reused",
+                **{SERVER_NAME_LABEL: self.server_name},
+            ).inc()
         except NotFoundError:
-            invalid_login_token_counter.labels("not found").inc()
+            invalid_login_token_counter.labels(
+                reason="not found",
+                **{SERVER_NAME_LABEL: self.server_name},
+            ).inc()
 
         raise AuthError(403, "Invalid login token", errcode=Codes.FORBIDDEN)