Refactor Gauge metrics to be homeserver-scoped

changelog.d/18725.misc

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

contrib/grafana/synapse.json

@@ -4396,7 +4396,7 @@
               "exemplar": false,
               "expr": "(time() - max without (job, index, host) (avg_over_time(synapse_federation_last_received_pdu_time[10m]))) / 60",
               "instant": false,
-              "legendFormat": "{{server_name}} ",
+              "legendFormat": "{{origin_server_name}} ",
               "range": true,
               "refId": "A"
             }
@@ -4518,7 +4518,7 @@
               "exemplar": false,
               "expr": "(time() - max without (job, index, host) (avg_over_time(synapse_federation_last_sent_pdu_time[10m]))) / 60",
               "instant": false,
-              "legendFormat": "{{server_name}}",
+              "legendFormat": "{{destination_server_name}}",
               "range": true,
               "refId": "A"
             }

docs/upgrade.md

@@ -117,6 +117,25 @@ each upgrade are complete before moving on to the next upgrade, to avoid
 stacking them up. You can monitor the currently running background updates with
 [the Admin API](usage/administration/admin_api/background_updates.html#status).
 
+# Upgrading to v1.136.0
+
+## Metric labels have changed on `synapse_federation_last_received_pdu_time` and `synapse_federation_last_sent_pdu_time`
+
+Previously, the `synapse_federation_last_received_pdu_time` and
+`synapse_federation_last_sent_pdu_time` metrics both used the `server_name` label to
+differentiate between different servers that we send and receive events from.
+
+Since we're now using the `server_name` label to differentiate between different Synapse
+homeserver instances running in the same process, these metrics have been changed as follows:
+
+ - `synapse_federation_last_received_pdu_time` now uses the `origin_server_name` label
+ - `synapse_federation_last_sent_pdu_time` now uses the `destination_server_name` label
+
+The Grafana dashboard JSON in `contrib/grafana/synapse.json` has been updated to reflect
+this change but you will need to manually update your own existing Grafana dashboards
+using these metrics.
+
+
 # Upgrading to v1.135.0
 
 ## `on_user_registration` module API callback may now run on any worker

scripts-dev/mypy_synapse_plugin.py

@@ -28,7 +28,7 @@
 import mypy.types
 from mypy.erasetype import remove_instance_last_known_values
 from mypy.errorcodes import ErrorCode
-from mypy.nodes import ARG_NAMED_OPT, ListExpr, NameExpr, TempNode, Var
+from mypy.nodes import ARG_NAMED_OPT, ListExpr, NameExpr, TempNode, TupleExpr, Var
 from mypy.plugin import (
     FunctionLike,
     FunctionSigContext,
@@ -61,6 +61,7 @@ def get_function_signature_hook(
     ) -> Optional[Callable[[FunctionSigContext], FunctionLike]]:
         if fullname in (
             "prometheus_client.metrics.Counter",
+            "prometheus_client.metrics.Gauge",
             # TODO: Add other prometheus_client metrics that need checking as we
             # refactor, see https://github.com/element-hq/synapse/issues/18592
         ):
@@ -98,8 +99,8 @@ def check_prometheus_metric_instantiation(ctx: FunctionSigContext) -> CallableTy
     ensures metrics are correctly separated by homeserver.
 
     There are also some metrics that apply at the process level, such as CPU usage,
-    Python garbage collection, Twisted reactor tick time which shouldn't have the
-    `SERVER_NAME_LABEL`. In those cases, use use a type ignore comment to disable the
+    Python garbage collection, and Twisted reactor tick time, which shouldn't have the
+    `SERVER_NAME_LABEL`. In those cases, use a type ignore comment to disable the
     check, e.g. `# type: ignore[missing-server-name-label]`.
     """
     # The true signature, this isn't being modified so this is what will be returned.
@@ -136,7 +137,7 @@ def check_prometheus_metric_instantiation(ctx: FunctionSigContext) -> CallableTy
     # ]
     # ```
     labelnames_arg_expression = ctx.args[2][0] if len(ctx.args[2]) > 0 else None
-    if isinstance(labelnames_arg_expression, ListExpr):
+    if isinstance(labelnames_arg_expression, (ListExpr, TupleExpr)):
         # Check if the `labelnames` argument includes the `server_name` label (`SERVER_NAME_LABEL`).
         for labelname_expression in labelnames_arg_expression.items:
             if (

synapse/app/_base.py

@@ -525,8 +525,12 @@ async def start(hs: "HomeServer") -> None:
     )
 
     # Register the threadpools with our metrics.
-    register_threadpool("default", reactor.getThreadPool())
-    register_threadpool("gai_resolver", resolver_threadpool)
+    register_threadpool(
+        name="default", server_name=server_name, threadpool=reactor.getThreadPool()
+    )
+    register_threadpool(
+        name="gai_resolver", server_name=server_name, threadpool=resolver_threadpool
+    )
 
     # Set up the SIGHUP machinery.
     if hasattr(signal, "SIGHUP"):

synapse/app/phone_stats_home.py

@@ -28,6 +28,7 @@
 
 from twisted.internet import defer
 
+from synapse.metrics import SERVER_NAME_LABEL
 from synapse.metrics.background_process_metrics import (
     run_as_background_process,
 )
@@ -57,16 +58,25 @@
 _stats_process: List[Tuple[int, "resource.struct_rusage"]] = []
 
 # Gauges to expose monthly active user control metrics
-current_mau_gauge = Gauge("synapse_admin_mau_current", "Current MAU")
+current_mau_gauge = Gauge(
+    "synapse_admin_mau_current",
+    "Current MAU",
+    labelnames=[SERVER_NAME_LABEL],
+)
 current_mau_by_service_gauge = Gauge(
     "synapse_admin_mau_current_mau_by_service",
     "Current MAU by service",
-    ["app_service"],
+    labelnames=["app_service", SERVER_NAME_LABEL],
+)
+max_mau_gauge = Gauge(
+    "synapse_admin_mau_max",
+    "MAU Limit",
+    labelnames=[SERVER_NAME_LABEL],
 )
-max_mau_gauge = Gauge("synapse_admin_mau_max", "MAU Limit")
 registered_reserved_users_mau_gauge = Gauge(
     "synapse_admin_mau_registered_reserved_users",
     "Registered users with reserved threepids",
+    labelnames=[SERVER_NAME_LABEL],
 )
 
 
@@ -237,13 +247,21 @@ async def _generate_monthly_active_users() -> None:
                     await store.get_monthly_active_count_by_service()
                 )
                 reserved_users = await store.get_registered_reserved_users()
-            current_mau_gauge.set(float(current_mau_count))
+            current_mau_gauge.labels(**{SERVER_NAME_LABEL: server_name}).set(
+                float(current_mau_count)
+            )
 
             for app_service, count in current_mau_count_by_service.items():
-                current_mau_by_service_gauge.labels(app_service).set(float(count))
-
-            registered_reserved_users_mau_gauge.set(float(len(reserved_users)))
-            max_mau_gauge.set(float(hs.config.server.max_mau_value))
+                current_mau_by_service_gauge.labels(
+                    app_service=app_service, **{SERVER_NAME_LABEL: server_name}
+                ).set(float(count))
+
+            registered_reserved_users_mau_gauge.labels(
+                **{SERVER_NAME_LABEL: server_name}
+            ).set(float(len(reserved_users)))
+            max_mau_gauge.labels(**{SERVER_NAME_LABEL: server_name}).set(
+                float(hs.config.server.max_mau_value)
+            )
 
         return run_as_background_process(
             "generate_monthly_active_users",

synapse/federation/federation_server.py

@@ -127,7 +127,7 @@
 last_pdu_ts_metric = Gauge(
     "synapse_federation_last_received_pdu_time",
     "The timestamp of the last PDU which was successfully received from the given domain",
-    labelnames=("server_name",),
+    labelnames=("origin_server_name", SERVER_NAME_LABEL),
 )
 
 
@@ -554,7 +554,9 @@ async def process_pdu(pdu: EventBase) -> JsonDict:
         )
 
         if newest_pdu_ts and origin in self._federation_metrics_domains:
-            last_pdu_ts_metric.labels(server_name=origin).set(newest_pdu_ts / 1000)
+            last_pdu_ts_metric.labels(
+                origin_server_name=origin, **{SERVER_NAME_LABEL: self.server_name}
+            ).set(newest_pdu_ts / 1000)
 
         return pdu_results
 

synapse/federation/sender/__init__.py

@@ -705,10 +705,12 @@ async def handle_room_events(events: List[EventBase]) -> None:
                     assert ts is not None
 
                     synapse.metrics.event_processing_lag.labels(
-                        "federation_sender"
+                        name="federation_sender",
+                        **{SERVER_NAME_LABEL: self.server_name},
                     ).set(now - ts)
                     synapse.metrics.event_processing_last_ts.labels(
-                        "federation_sender"
+                        name="federation_sender",
+                        **{SERVER_NAME_LABEL: self.server_name},
                     ).set(ts)
 
                     events_processed_counter.labels(
@@ -726,7 +728,7 @@ async def handle_room_events(events: List[EventBase]) -> None:
                 ).inc()
 
                 synapse.metrics.event_processing_positions.labels(
-                    "federation_sender"
+                    name="federation_sender", **{SERVER_NAME_LABEL: self.server_name}
                 ).set(next_token)
 
         finally:

synapse/federation/sender/transaction_manager.py

@@ -34,6 +34,7 @@
     tags,
     whitelisted_homeserver,
 )
+from synapse.metrics import SERVER_NAME_LABEL
 from synapse.types import JsonDict
 from synapse.util import json_decoder
 from synapse.util.metrics import measure_func
@@ -47,7 +48,7 @@
 last_pdu_ts_metric = Gauge(
     "synapse_federation_last_sent_pdu_time",
     "The timestamp of the last PDU which was successfully sent to the given domain",
-    labelnames=("server_name",),
+    labelnames=("destination_server_name", SERVER_NAME_LABEL),
 )
 
 
@@ -191,6 +192,7 @@ def json_data_cb() -> JsonDict:
 
             if pdus and destination in self._federation_metrics_domains:
                 last_pdu = pdus[-1]
-                last_pdu_ts_metric.labels(server_name=destination).set(
-                    last_pdu.origin_server_ts / 1000
-                )
+                last_pdu_ts_metric.labels(
+                    destination_server_name=destination,
+                    **{SERVER_NAME_LABEL: self.server_name},
+                ).set(last_pdu.origin_server_ts / 1000)

synapse/handlers/appservice.py

@@ -207,7 +207,8 @@ async def handle_room_events(events: Iterable[EventBase]) -> None:
                     await self.store.set_appservice_last_pos(upper_bound)
 
                     synapse.metrics.event_processing_positions.labels(
-                        "appservice_sender"
+                        name="appservice_sender",
+                        **{SERVER_NAME_LABEL: self.server_name},
                     ).set(upper_bound)
 
                     events_processed_counter.labels(
@@ -230,10 +231,12 @@ async def handle_room_events(events: Iterable[EventBase]) -> None:
                         assert ts is not None
 
                         synapse.metrics.event_processing_lag.labels(
-                            "appservice_sender"
+                            name="appservice_sender",
+                            **{SERVER_NAME_LABEL: self.server_name},
                         ).set(now - ts)
                         synapse.metrics.event_processing_last_ts.labels(
-                            "appservice_sender"
+                            name="appservice_sender",
+                            **{SERVER_NAME_LABEL: self.server_name},
                         ).set(ts)
             finally:
                 self.is_processing = False

synapse/handlers/delayed_events.py

@@ -22,7 +22,7 @@
 from synapse.api.ratelimiting import Ratelimiter
 from synapse.config.workers import MAIN_PROCESS_INSTANCE_NAME
 from synapse.logging.opentracing import set_tag
-from synapse.metrics import event_processing_positions
+from synapse.metrics import SERVER_NAME_LABEL, event_processing_positions
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.replication.http.delayed_events import (
     ReplicationAddedDelayedEventRestServlet,
@@ -191,7 +191,9 @@ async def _unsafe_process_new_event(self) -> None:
                 self._event_pos = max_pos
 
                 # Expose current event processing position to prometheus
-                event_processing_positions.labels("delayed_events").set(max_pos)
+                event_processing_positions.labels(
+                    name="delayed_events", **{SERVER_NAME_LABEL: self.server_name}
+                ).set(max_pos)
 
                 await self._store.update_delayed_events_stream_pos(max_pos)
 

synapse/handlers/presence.py

@@ -1568,9 +1568,9 @@ async def _unsafe_process(self) -> None:
                 self._event_pos = max_pos
 
                 # Expose current event processing position to prometheus
-                synapse.metrics.event_processing_positions.labels("presence").set(
-                    max_pos
-                )
+                synapse.metrics.event_processing_positions.labels(
+                    name="presence", **{SERVER_NAME_LABEL: self.server_name}
+                ).set(max_pos)
 
     async def _handle_state_delta(self, room_id: str, deltas: List[StateDelta]) -> None:
         """Process current state deltas for the room to find new joins that need

synapse/handlers/room_member.py

@@ -49,7 +49,7 @@
 from synapse.handlers.state_deltas import MatchChange, StateDeltasHandler
 from synapse.handlers.worker_lock import NEW_EVENT_DURING_PURGE_LOCK_NAME
 from synapse.logging import opentracing
-from synapse.metrics import event_processing_positions
+from synapse.metrics import SERVER_NAME_LABEL, event_processing_positions
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.replication.http.push import ReplicationCopyPusherRestServlet
 from synapse.storage.databases.main.state_deltas import StateDelta
@@ -2255,7 +2255,9 @@ async def _unsafe_process(self) -> None:
             self.pos = max_pos
 
             # Expose current event processing position to prometheus
-            event_processing_positions.labels("room_forgetter").set(max_pos)
+            event_processing_positions.labels(
+                name="room_forgetter", **{SERVER_NAME_LABEL: self.server_name}
+            ).set(max_pos)
 
             await self._store.update_room_forgetter_stream_pos(max_pos)
 

synapse/handlers/stats.py

@@ -32,7 +32,7 @@
 )
 
 from synapse.api.constants import EventContentFields, EventTypes, Membership
-from synapse.metrics import event_processing_positions
+from synapse.metrics import SERVER_NAME_LABEL, event_processing_positions
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.storage.databases.main.state_deltas import StateDelta
 from synapse.types import JsonDict
@@ -147,7 +147,9 @@ async def _unsafe_process(self) -> None:
 
             logger.debug("Handled room stats to %s -> %s", self.pos, max_pos)
 
-            event_processing_positions.labels("stats").set(max_pos)
+            event_processing_positions.labels(
+                name="stats", **{SERVER_NAME_LABEL: self.server_name}
+            ).set(max_pos)
 
             self.pos = max_pos
 

synapse/handlers/user_directory.py

@@ -35,6 +35,7 @@
 )
 from synapse.api.errors import Codes, SynapseError
 from synapse.handlers.state_deltas import MatchChange, StateDeltasHandler
+from synapse.metrics import SERVER_NAME_LABEL
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.storage.databases.main.state_deltas import StateDelta
 from synapse.storage.databases.main.user_directory import SearchResult
@@ -262,9 +263,9 @@ async def _unsafe_process(self) -> None:
                 self.pos = max_pos
 
                 # Expose current event processing position to prometheus
-                synapse.metrics.event_processing_positions.labels("user_dir").set(
-                    max_pos
-                )
+                synapse.metrics.event_processing_positions.labels(
+                    name="user_dir", **{SERVER_NAME_LABEL: self.server_name}
+                ).set(max_pos)
 
                 await self.store.update_user_directory_stream_pos(max_pos)