diff --git a/changelog.d/20405.feature b/changelog.d/20405.feature new file mode 100644 index 0000000000..9097fefdc0 --- /dev/null +++ b/changelog.d/20405.feature @@ -0,0 +1 @@ +Add a `synapse_thread_cpu_seconds_total` metric that reports CPU time per thread type, e.g. the main thread, each threadpool and the Rust tokio runtime. diff --git a/scripts-dev/mypy_synapse_plugin.py b/scripts-dev/mypy_synapse_plugin.py index ac1fee7f67..f97b74fc33 100644 --- a/scripts-dev/mypy_synapse_plugin.py +++ b/scripts-dev/mypy_synapse_plugin.py @@ -147,6 +147,7 @@ prometheus_metric_fullname_to_label_arg_map: Mapping[str, ArgLocation | None] = "synapse.metrics._gc.PyPyGCStats": None, "synapse.metrics._reactor_metrics.ReactorLastSeenMetric": None, "synapse.metrics.CPUMetrics": None, + "synapse.metrics.thread_cpu.ThreadCPUMetrics": None, "synapse.metrics.jemalloc.JemallocCollector": None, "synapse.util.metrics.DynamicCollectorRegistry": None, "synapse.metrics.background_process_metrics._Collector": None, diff --git a/synapse/app/_base.py b/synapse/app/_base.py index 1e40ae5baf..430fe53ec8 100644 --- a/synapse/app/_base.py +++ b/synapse/app/_base.py @@ -345,6 +345,13 @@ def listen_metrics( port, addr=host, registry=RegistryProxy ) servers.append((server, thread)) + + # Note that we do not register these threads with `ThreadCPUMetrics` as they + # don't actually do much work themselves and simply spawn another new thread + # per-request (where the real metrics scrape work happens). It's not worth it to + # track these with `ThreadCPUMetrics` as we assume they won't use much CPU and + # so are OK with them being tracked in the `other` bucket. And we can't easily + # register those sub-threads where the metric scrapes actually happen. return servers diff --git a/synapse/logging/opentracing.py b/synapse/logging/opentracing.py index 6e4e029163..2fe713a6b5 100644 --- a/synapse/logging/opentracing.py +++ b/synapse/logging/opentracing.py @@ -175,6 +175,7 @@ import enum import inspect import logging import re +import threading from functools import wraps from typing import ( TYPE_CHECKING, @@ -201,6 +202,7 @@ from twisted.web.http import Request from twisted.web.http_headers import Headers from synapse.config import ConfigError +from synapse.metrics.thread_cpu import thread_cpu_metrics from synapse.util.json import json_decoder, json_encoder if TYPE_CHECKING: @@ -492,8 +494,26 @@ def init_tracer(hs: "HomeServer") -> None: tracer = config.create_tracer(RustReporter(), config.sampler) opentracing.set_global_tracer(tracer) else: + # Register the jaeger tracing thread with the CPU metrics system. + # + # jaeger-client's reporter sends spans from a thread it starts without a + # name, so we have to manually register its CPU time with a type of its + # own. To do this we find all threads that started during the + # initialisation and register them. + # + # This is best effort, but works in practice for the jaeger reporter + # thread. Technically, other Python threads could be started during this + # period (though unlikely to happen during Synapse startup), or newer + # versions of the library could defer starting of threads. At worse this + # means that some threads CPU time may be incorrectly attributed to + # `jaeger-reporter` or `other`. + threads_before = set(threading.enumerate()) config.initialize_tracer() + for thread in threading.enumerate(): + if thread not in threads_before: + thread_cpu_metrics.register_thread("jaeger-reporter", thread) + # Whitelisting diff --git a/synapse/metrics/__init__.py b/synapse/metrics/__init__.py index cf9aaa7f2d..8e04dbf423 100644 --- a/synapse/metrics/__init__.py +++ b/synapse/metrics/__init__.py @@ -62,6 +62,7 @@ from twisted.web.server import Request import synapse.metrics._reactor_metrics # noqa: F401 from synapse.metrics._gc import MIN_TIME_BETWEEN_GCS, install_gc_manager from synapse.metrics._types import Collector +from synapse.metrics.thread_cpu import thread_cpu_metrics from synapse.synapse_rust import get_rustc_version from synapse.types import StrSequence from synapse.util import SYNAPSE_VERSION @@ -765,6 +766,8 @@ def register_threadpool(*, name: str, server_name: str, threadpool: ThreadPool) name=name, **{SERVER_NAME_LABEL: server_name} ).set_function(lambda: len(threadpool.working)) + thread_cpu_metrics.register_threadpool(name, threadpool) + class MetricsResource(Resource): """ diff --git a/synapse/metrics/thread_cpu.py b/synapse/metrics/thread_cpu.py new file mode 100644 index 0000000000..fbc5ac91d2 --- /dev/null +++ b/synapse/metrics/thread_cpu.py @@ -0,0 +1,226 @@ +# +# This file is licensed under the Affero General Public License (AGPL) version 3. +# +# Copyright (C) 2026 Element Creations Ltd +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# See the GNU Affero General Public License for more details: +# . +# + +"""Prometheus metrics for CPU time per thread type.""" + +import logging +import os +import threading +from collections import Counter +from typing import Iterable +from weakref import WeakKeyDictionary + +import attr +from prometheus_client.core import REGISTRY, CounterMetricFamily, Metric + +from twisted.python.threadpool import ThreadPool + +from synapse.metrics._types import Collector + +logger = logging.getLogger(__name__) + + +@attr.s(slots=True, frozen=True, auto_attribs=True) +class _ThreadSample: + """A thread's CPU usage as of one scrape.""" + + start_time: int + """When the thread started, in ticks since boot. Used to distinguish a new + thread that reuses the ID of one that exited. + """ + + utime: int + """User CPU time used by the thread, in ticks.""" + + stime: int + """System CPU time used by the thread, in ticks.""" + + +class ThreadCPUMetrics(Collector): + """Reports the CPU time used by each type of thread in the process. + + Threads are grouped by what they run: the main (reactor) thread, a + registered type, the `tokio` runtime, and `other` for everything else. + + Note: "other" can include threads like prometheus HTTP server threads. + + Linux only, as it reads `/proc/self/task`. The kernel only reports CPU time + for live threads, so the collector sums the CPU time each thread used + between scrapes. The CPU time a thread uses between the last scrape and + its exit is not counted. + """ + + def __init__(self) -> None: + # The kernel reports thread CPU time in clock ticks. This converts + # ticks to seconds. It is 100 on almost every Linux system. + self._ticks_per_sec = 100 + try: + self._ticks_per_sec = os.sysconf("SC_CLK_TCK") + except (ValueError, TypeError, AttributeError): + pass + + # Guards the fields below, as the metrics listener may scrape from any + # thread. + self._lock = threading.Lock() + + self._threadpools: "WeakKeyDictionary[ThreadPool, str]" = WeakKeyDictionary() + self._threads: "WeakKeyDictionary[threading.Thread, str]" = WeakKeyDictionary() + + # Thread ID to the thread's sample from the previous scrape. + self._last_seen: dict[int, _ThreadSample] = {} + # (thread type, mode) to the total ticks used by threads of that type. + self._totals: Counter[tuple[str, str]] = Counter() + + def register_threadpool(self, name: str, threadpool: ThreadPool) -> None: + """Report the threads of `threadpool` as the type `threadpool_`. + + A threadpool keeps the name it was first registered under. + """ + with self._lock: + self._threadpools.setdefault(threadpool, name) + + def register_thread(self, thread_type: str, thread: threading.Thread) -> None: + """Report `thread` as the type `thread_type`.""" + with self._lock: + self._threads[thread] = thread_type + + def _registered_thread_types(self) -> dict[int, str]: + """Map the native IDs of registered threads and threadpool threads to + their thread type. + + Requires the caller to hold `self._lock`. + """ + + assert self._lock.locked() + + types: dict[int, str] = {} + for thread, thread_type in list(self._threads.items()): + if thread.native_id is not None and thread.is_alive(): + types[thread.native_id] = thread_type + + for threadpool, name in list(self._threadpools.items()): + # Twisted never removes exited threads from `threads`. The kernel + # may have given their IDs to new threads. + for thread in list(threadpool.threads): + if thread.native_id is not None and thread.is_alive(): + types[thread.native_id] = f"threadpool_{name}" + + return types + + def _thread_type(self, tid: int, registered_types: dict[int, str]) -> str: + """Work out the thread type to report a thread's CPU time under. + + Args: + tid: The thread's native ID. + registered_types: The output of `_registered_thread_types`. + + Returns: + `main`, a registered type, `tokio` or `other`. + """ + if tid == os.getpid(): + return "main" + + registered_type = registered_types.get(tid) + if registered_type is not None: + return registered_type + + try: + with open(f"/proc/self/task/{tid}/comm") as f: + comm = f.read().strip() + except OSError: + return "other" + + # Tokio names its worker and blocking threads `tokio-rt-worker`, or + # `tokio-runtime-worker` in older versions. + if comm.startswith("tokio-"): + return "tokio" + return "other" + + def collect(self) -> Iterable[Metric]: + """Add the CPU time each live thread used since the previous scrape to + its type's total, and report the totals. + + A thread that exits loses the CPU it used since the last scrape. + """ + try: + tids = [int(tid) for tid in os.listdir("/proc/self/task")] + except OSError: + return + + with self._lock: + registered_types = self._registered_thread_types() + + last_seen: dict[int, _ThreadSample] = {} + for tid in tids: + try: + with open(f"/proc/self/task/{tid}/stat") as f: + line = f.read() + except OSError: + # The thread exited since we listed it. We lose this bit of + # CPU tracing, but it's hard to recover it after the thread + # has exited. + # + # We can see in the metrics how much thread CPU has been + # lost, by comparing total CPU vs. the sum of CPU used by + # all threads. + continue + + # The thread name may contain spaces or brackets, so split after + # its closing bracket. See `man 5 proc_pid_stat` for the field + # numbers (note they're off by three due to zero-based indexing + # and we have split off the first two fields). + # + # Example of the start of such a line: + # + # 171453 (cat) R 30798 171453 ... + fields = line.rsplit(")", 1)[1].split() + utime = int(fields[11]) + stime = int(fields[12]) + start_time = int(fields[19]) + + prev = self._last_seen.get(tid) + if prev is not None and prev.start_time == start_time: + prev_utime, prev_stime = prev.utime, prev.stime + + # It's possible (though unlikely) that a thread has exited + # and got reused in the same tick, so ensure that times have + # strictly increased. + if utime < prev_utime or stime < prev_stime: + prev_utime = prev_stime = 0 + else: + prev_utime = prev_stime = 0 + + thread_type = self._thread_type(tid, registered_types) + self._totals[(thread_type, "user")] += utime - prev_utime + self._totals[(thread_type, "system")] += stime - prev_stime + + last_seen[tid] = _ThreadSample(start_time, utime, stime) + + self._last_seen = last_seen + + # This is a process-level metric, so it does not have the `SERVER_NAME_LABEL`. + cpu = CounterMetricFamily( # type: ignore[missing-server-name-label] + "synapse_thread_cpu_seconds", + "CPU time used by each type of thread in the process", + labels=["thread_type", "mode"], + ) + for (thread_type, mode), ticks in sorted(self._totals.items()): + cpu.add_metric([thread_type, mode], ticks / self._ticks_per_sec) + + yield cpu + + +# This is a process-level metric, so it does not have the `SERVER_NAME_LABEL`. +thread_cpu_metrics = ThreadCPUMetrics() # type: ignore[missing-server-name-label] +REGISTRY.register(thread_cpu_metrics) diff --git a/synapse/module_api/__init__.py b/synapse/module_api/__init__.py index 58ccddc792..24b80c2dbd 100644 --- a/synapse/module_api/__init__.py +++ b/synapse/module_api/__init__.py @@ -82,6 +82,7 @@ from synapse.logging.context import ( from synapse.metrics.background_process_metrics import ( run_as_background_process as _run_as_background_process, ) +from synapse.metrics.thread_cpu import thread_cpu_metrics from synapse.module_api.callbacks.account_validity_callbacks import ( IS_USER_EXPIRED_CALLBACK, ON_LEGACY_ADMIN_REQUEST, @@ -1764,6 +1765,12 @@ class ModuleApi: Returns: The return value of the function once ran in a thread. """ + # Report the CPU time of the module's threads under the pool's name. + thread_pool_name = "module-other" + if threadpool.name: + thread_pool_name = "module-" + threadpool.name + thread_cpu_metrics.register_threadpool(thread_pool_name, threadpool) + return await defer_to_threadpool( self._hs.get_reactor(), threadpool, f, *args, **kwargs ) diff --git a/tests/metrics/test_thread_cpu.py b/tests/metrics/test_thread_cpu.py new file mode 100644 index 0000000000..e88fe3b126 --- /dev/null +++ b/tests/metrics/test_thread_cpu.py @@ -0,0 +1,131 @@ +# +# This file is licensed under the Affero General Public License (AGPL) version 3. +# +# Copyright (C) 2026 Element Creations Ltd +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# See the GNU Affero General Public License for more details: +# . +# + +import os +import threading +import time + +from synapse.metrics.thread_cpu import ThreadCPUMetrics +from synapse.util.duration import Duration + +from tests import unittest + +# How much CPU time to burn on each thread, in seconds. +BURN = Duration(milliseconds=200) + +# How far a thread's reported CPU may fall short of its burn. The kernel reports +# CPU time in whole ticks (rounded down), so we allow some tolerance (here ~5 +# ticks). +TOLERANCE = Duration(milliseconds=50) + + +def _burn_cpu() -> None: + """Spin on the calling thread until it has used `BURN_SECONDS` of CPU.""" + start = time.thread_time() + while time.thread_time() - start < BURN.as_secs(): + pass + + +def _collect(metrics: ThreadCPUMetrics) -> dict[str, float]: + """Scrape `metrics` and map each thread type to its CPU seconds. + + User and system time are summed. The kernel splits a thread's CPU time + between the two by sampling, so a burn in user code can show up partly as + system time. + """ + totals: dict[str, float] = {} + for metric in metrics.collect(): + for sample in metric.samples: + if sample.name == "synapse_thread_cpu_seconds_total": + thread_type = sample.labels["thread_type"] + totals[thread_type] = totals.get(thread_type, 0.0) + sample.value + return totals + + +class ThreadCPUMetricsTestCase(unittest.TestCase): + def test_cpu_adds_up(self) -> None: + """CPU used between two scrapes is reported under the right thread + type, and the types add up to the CPU the whole process used. + """ + if not os.path.isdir("/proc/self/task"): + self.skipTest( + "Skipping as `ThreadCPUMetrics` tests require /proc/self/task (Linux only)" + ) + + # A fresh collector, so threads registered elsewhere don't count. It + # reports process-level metrics, so it has no `SERVER_NAME_LABEL`. + metrics = ThreadCPUMetrics() # type: ignore[missing-server-name-label] + + # First record the base CPU metrics. + before = _collect(metrics) + process_before = sum(os.times()[:2]) + + # First we setup and burn some CPU on a separate thread. + + # The collector only sees live threads, so the thread waits for + # `release` until after the second scrape. + burnt = threading.Event() + release = threading.Event() + + def run_on_thread() -> None: + _burn_cpu() + burnt.set() + release.wait(timeout=Duration(seconds=10).as_secs()) + + # Since `/proc//stat` puts the thread name in parenthesis and uses + # spaces as field separators, choose a thread name with those same characters + # to ensure we parse it correctly. + # + # Note that the thread name only gets passed to the kernel from Python + # 3.14 onwards. + thread = threading.Thread(target=run_on_thread, name="a) b) (c") + + # Cleanup functions are run in reverse order, so first we set the + # release event (to wake the thread up), and then we wait for it exit. + self.addCleanup(thread.join) + self.addCleanup(release.set) + + metrics.register_thread("test-thread", thread) + + thread.start() + + # Wait for the separate thread to burn CPU before continuing. + self.assertTrue(burnt.wait(timeout=Duration(seconds=10).as_secs())) + + # Next burn some CPU on the main thread. + _burn_cpu() + + # Now collect the CPU metrics after burning on both threads. + after = _collect(metrics) + process_delta = sum(os.times()[:2]) - process_before + + # Check that each thread burned at least the expected amount of CPU. + + def delta(thread_type: str) -> Duration: + secs = after[thread_type] - before.get(thread_type, 0.0) + return Duration(seconds=secs) + + self.assertGreaterEqual(delta("main"), BURN - TOLERANCE) + self.assertGreaterEqual(delta("test-thread"), BURN - TOLERANCE) + + # Check that the sum of the per-thread CPU deltas is (more-or-less) the + # same as the whole process CPU delta. + + # The sum over threads can lag the process total by up to a tick per + # thread. Allow a couple more ticks for the scrapes themselves. + tick = 1 / os.sysconf("SC_CLK_TCK") + slack = (len(os.listdir("/proc/self/task")) + 2) * tick + + threads_delta = sum(after.values()) - sum(before.values()) + self.assertLessEqual(abs(threads_delta - process_delta), slack)