Use homeserver specific Prometheus CollectorRegistry for metrics instead of global REGISTRY

changelog.d/18443.misc

@@ -0,0 +1 @@
+Refactor metrics to collect in homeserver-specific registry.

docs/metrics-howto.md

@@ -53,15 +53,17 @@
         type: http
         x_forwarded: true
         bind_addresses: ['::1', '127.0.0.1']
-
         resources:
           - names: [client, federation]
             compress: false
 
       # beginning of the new metrics listener
       - port: 9000
-        type: metrics
+        type: http
         bind_addresses: ['::1', '127.0.0.1']
+        resources:
+          - names: [metrics]
+            compress: false
     ```
 
 1.  Restart Synapse.

schema/synapse-config.schema.yaml

@@ -451,8 +451,7 @@ properties:
           type: string
           description: >-
             The type of listener. Normally `http`, but other valid options are
-            [`manhole`](../../manhole.md) and
-            [`metrics`](../../metrics-howto.md).
+            [`manhole`](../../manhole.md).
           enum:
             - http
             - manhole

synapse/api/auth/msc3861_delegated.py

@@ -207,6 +207,7 @@ def __init__(self, hs: "HomeServer"):
         #   the old access token to continue working for a short time.
         self._introspection_cache: ResponseCache[str] = ResponseCache(
             self._clock,
+            hs.get_cache_manager(),
             "token_introspection",
             timeout_ms=120_000,
             # don't log because the keys are access tokens

synapse/app/_base.py

@@ -74,9 +74,13 @@
 from synapse.http.site import SynapseSite
 from synapse.logging.context import PreserveLoggingContext
 from synapse.logging.opentracing import init_tracer
-from synapse.metrics import install_gc_manager, register_threadpool
-from synapse.metrics.background_process_metrics import wrap_as_background_process
-from synapse.metrics.jemalloc import setup_jemalloc_stats
+from synapse.metrics import CPUMetrics, install_gc_manager, register_threadpool
+from synapse.metrics._gc import GCCounts, PyPyGCStats, running_on_pypy
+from synapse.metrics._reactor_metrics import setup_reactor_metrics
+from synapse.metrics.background_process_metrics import (
+    BackgroundProcessCollector,
+    wrap_as_background_process,
+)
 from synapse.module_api.callbacks.spamchecker_callbacks import load_legacy_spam_checkers
 from synapse.module_api.callbacks.third_party_event_rules_callbacks import (
     load_legacy_third_party_event_rules,
@@ -178,7 +182,6 @@ def start_reactor(
 
     def run() -> None:
         logger.info("Running")
-        setup_jemalloc_stats()
         change_resource_limit(soft_file_limit)
         if gc_thresholds:
             gc.set_threshold(*gc_thresholds)
@@ -283,20 +286,6 @@ async def wrapper() -> None:
     reactor.callWhenRunning(lambda: defer.ensureDeferred(wrapper()))
 
 
-def listen_metrics(bind_addresses: StrCollection, port: int) -> None:
-    """
-    Start Prometheus metrics server.
-    """
-    from prometheus_client import start_http_server as start_http_server_prometheus
-
-    from synapse.metrics import RegistryProxy
-
-    for host in bind_addresses:
-        logger.info("Starting metrics listener on %s:%d", host, port)
-        _set_prometheus_client_use_created_metrics(False)
-        start_http_server_prometheus(port, addr=host, registry=RegistryProxy)
-
-
 def _set_prometheus_client_use_created_metrics(new_value: bool) -> None:
     """
     Sets whether prometheus_client should expose `_created`-suffixed metrics for
@@ -607,6 +596,7 @@ def run_sighup(*args: Any, **kwargs: Any) -> None:
     )
 
     setup_sentry(hs)
+    setup_global_metrics(hs)
     setup_sdnotify(hs)
 
     # If background tasks are running on the main process or this is the worker in
@@ -694,6 +684,20 @@ def setup_sentry(hs: "HomeServer") -> None:
     global_scope.set_tag("worker_name", name)
 
 
+def setup_global_metrics(hs: "HomeServer") -> None:
+    """Set up global metrics for this homeserver.
+
+    This is called after the homeserver has been set up, but before any
+    listeners are started.
+    """
+    CPUMetrics(registry=hs.metrics_collector_registry)
+    if running_on_pypy:
+        PyPyGCStats(registry=hs.metrics_collector_registry)
+        GCCounts(registry=hs.metrics_collector_registry)
+    BackgroundProcessCollector(registry=hs.metrics_collector_registry)
+    setup_reactor_metrics(registry=hs.metrics_collector_registry)
+
+
 def setup_sdnotify(hs: "HomeServer") -> None:
     """Adds process state hooks to tell systemd what we are up to."""
 

synapse/app/generic_worker.py

@@ -49,7 +49,7 @@
 from synapse.federation.transport.server import TransportLayerServer
 from synapse.http.server import JsonResource, OptionsResource
 from synapse.logging.context import LoggingContext
-from synapse.metrics import METRICS_PREFIX, MetricsResource, RegistryProxy
+from synapse.metrics import METRICS_PREFIX, MetricsResource
 from synapse.replication.http import REPLICATION_PREFIX, ReplicationRestResource
 from synapse.rest import ClientRestResource, admin
 from synapse.rest.health import HealthResource
@@ -186,7 +186,9 @@ def _listen_http(self, listener_config: ListenerConfig) -> None:
         for res in listener_config.http_options.resources:
             for name in res.names:
                 if name == "metrics":
-                    resources[METRICS_PREFIX] = MetricsResource(RegistryProxy)
+                    resources[METRICS_PREFIX] = MetricsResource(
+                        self.metrics_collector_registry
+                    )
                 elif name == "client":
                     resource: Resource = ClientRestResource(self)
 
@@ -283,23 +285,6 @@ def start_listening(self) -> None:
                     raise ConfigError(
                         "Can not using a unix socket for manhole at this time."
                     )
-
-            elif listener.type == "metrics":
-                if not self.config.metrics.enable_metrics:
-                    logger.warning(
-                        "Metrics listener configured, but enable_metrics is not True!"
-                    )
-                else:
-                    if isinstance(listener, TCPListenerConfig):
-                        _base.listen_metrics(
-                            listener.bind_addresses,
-                            listener.port,
-                        )
-                    else:
-                        raise ConfigError(
-                            "Can not use a unix socket for metrics at this time."
-                        )
-
             else:
                 logger.warning("Unsupported listener type: %s", listener.type)
 

synapse/app/homeserver.py

@@ -60,7 +60,7 @@
     StaticResource,
 )
 from synapse.logging.context import LoggingContext
-from synapse.metrics import METRICS_PREFIX, MetricsResource, RegistryProxy
+from synapse.metrics import METRICS_PREFIX, MetricsResource
 from synapse.replication.http import REPLICATION_PREFIX, ReplicationRestResource
 from synapse.rest import ClientRestResource, admin
 from synapse.rest.health import HealthResource
@@ -252,7 +252,9 @@ def _configure_named_resource(
             resources[SERVER_KEY_PREFIX] = KeyResource(self)
 
         if name == "metrics" and self.config.metrics.enable_metrics:
-            metrics_resource: Resource = MetricsResource(RegistryProxy)
+            metrics_resource: Resource = MetricsResource(
+                self.metrics_collector_registry
+            )
             if compress:
                 metrics_resource = gz_wrap(metrics_resource)
             resources[METRICS_PREFIX] = metrics_resource
@@ -286,22 +288,6 @@ def start_listening(self) -> None:
                     raise ConfigError(
                         "Can not use a unix socket for manhole at this time."
                     )
-            elif listener.type == "metrics":
-                if not self.config.metrics.enable_metrics:
-                    logger.warning(
-                        "Metrics listener configured, but enable_metrics is not True!"
-                    )
-                else:
-                    if isinstance(listener, TCPListenerConfig):
-                        _base.listen_metrics(
-                            listener.bind_addresses,
-                            listener.port,
-                        )
-                    else:
-                        raise ConfigError(
-                            "Can not use a unix socket for metrics at this time."
-                        )
-
             else:
                 # this shouldn't happen, as the listener type should have been checked
                 # during parsing

synapse/appservice/api.py

@@ -56,33 +56,6 @@
 
 logger = logging.getLogger(__name__)
 
-sent_transactions_counter = Counter(
-    "synapse_appservice_api_sent_transactions",
-    "Number of /transactions/ requests sent",
-    ["service"],
-)
-
-failed_transactions_counter = Counter(
-    "synapse_appservice_api_failed_transactions",
-    "Number of /transactions/ requests that failed to send",
-    ["service"],
-)
-
-sent_events_counter = Counter(
-    "synapse_appservice_api_sent_events", "Number of events sent to the AS", ["service"]
-)
-
-sent_ephemeral_counter = Counter(
-    "synapse_appservice_api_sent_ephemeral",
-    "Number of ephemeral events sent to the AS",
-    ["service"],
-)
-
-sent_todevice_counter = Counter(
-    "synapse_appservice_api_sent_todevice",
-    "Number of todevice messages sent to the AS",
-    ["service"],
-)
 
 HOUR_IN_MS = 60 * 60 * 1000
 
@@ -130,7 +103,45 @@ def __init__(self, hs: "HomeServer"):
         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
+            hs.get_clock(),
+            hs.get_cache_manager(),
+            "as_protocol_meta",
+            timeout_ms=HOUR_IN_MS,
+        )
+
+        self.sent_transactions_counter = Counter(
+            "synapse_appservice_api_sent_transactions",
+            "Number of /transactions/ requests sent",
+            ["service"],
+            registry=hs.metrics_collector_registry,
+        )
+
+        self.failed_transactions_counter = Counter(
+            "synapse_appservice_api_failed_transactions",
+            "Number of /transactions/ requests that failed to send",
+            ["service"],
+            registry=hs.metrics_collector_registry,
+        )
+
+        self.sent_events_counter = Counter(
+            "synapse_appservice_api_sent_events",
+            "Number of events sent to the AS",
+            ["service"],
+            registry=hs.metrics_collector_registry,
+        )
+
+        self.sent_ephemeral_counter = Counter(
+            "synapse_appservice_api_sent_ephemeral",
+            "Number of ephemeral events sent to the AS",
+            ["service"],
+            registry=hs.metrics_collector_registry,
+        )
+
+        self.sent_todevice_counter = Counter(
+            "synapse_appservice_api_sent_todevice",
+            "Number of todevice messages sent to the AS",
+            ["service"],
+            registry=hs.metrics_collector_registry,
         )
 
     def _get_headers(self, service: "ApplicationService") -> Dict[bytes, List[bytes]]:
@@ -395,10 +406,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))
+            self.sent_transactions_counter.labels(service.id).inc()
+            self.sent_events_counter.labels(service.id).inc(len(serialized_events))
+            self.sent_ephemeral_counter.labels(service.id).inc(len(ephemeral))
+            self.sent_todevice_counter.labels(service.id).inc(len(to_device_messages))
             return True
         except CodeMessageException as e:
             logger.warning(
@@ -417,7 +428,7 @@ async def push_bulk(
                 ex.args,
                 exc_info=logger.isEnabledFor(logging.DEBUG),
             )
-        failed_transactions_counter.labels(service.id).inc()
+        self.failed_transactions_counter.labels(service.id).inc()
         return False
 
     async def claim_client_keys(

synapse/federation/federation_client.py

@@ -85,8 +85,6 @@
 
 logger = logging.getLogger(__name__)
 
-sent_queries_counter = Counter("synapse_federation_client_sent_queries", "", ["type"])
-
 
 PDU_RETRY_TIME_MS = 1 * 60 * 1000
 
@@ -145,6 +143,7 @@ def __init__(self, hs: "HomeServer"):
         self._get_pdu_cache: ExpiringCache[str, Tuple[EventBase, str]] = ExpiringCache(
             cache_name="get_pdu_cache",
             clock=self._clock,
+            cache_manager=hs.get_cache_manager(),
             max_len=1000,
             expiry_ms=120 * 1000,
             reset_expiry_on_get=False,
@@ -163,11 +162,19 @@ def __init__(self, hs: "HomeServer"):
         ] = ExpiringCache(
             cache_name="get_room_hierarchy_cache",
             clock=self._clock,
+            cache_manager=hs.get_cache_manager(),
             max_len=1000,
             expiry_ms=5 * 60 * 1000,
             reset_expiry_on_get=False,
         )
 
+        self.sent_queries_counter = Counter(
+            "synapse_federation_client_sent_queries",
+            "",
+            ["type"],
+            registry=hs.metrics_collector_registry,
+        )
+
     def _clear_tried_cache(self) -> None:
         """Clear pdu_destination_tried cache"""
         now = self._clock.time_msec()
@@ -207,7 +214,7 @@ async def make_query(
         Returns:
             The JSON object from the response
         """
-        sent_queries_counter.labels(query_type).inc()
+        self.sent_queries_counter.labels(query_type).inc()
 
         return await self.transport_layer.make_query(
             destination,
@@ -229,7 +236,7 @@ async def query_client_keys(
         Returns:
             The JSON object from the response
         """
-        sent_queries_counter.labels("client_device_keys").inc()
+        self.sent_queries_counter.labels("client_device_keys").inc()
         return await self.transport_layer.query_client_keys(
             destination, content, timeout
         )
@@ -240,7 +247,7 @@ 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()
+        self.sent_queries_counter.labels("user_devices").inc()
         return await self.transport_layer.query_user_devices(
             destination, user_id, timeout
         )
@@ -262,7 +269,7 @@ async def claim_client_keys(
         Returns:
             The JSON object from the response
         """
-        sent_queries_counter.labels("client_one_time_keys").inc()
+        self.sent_queries_counter.labels("client_one_time_keys").inc()
 
         # Convert the query with counts into a stable and unstable query and check
         # if attempting to claim more than 1 OTK.