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..8df0e40215 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/metrics/__init__.py b/synapse/metrics/__init__.py index cf9aaa7f2d..dcfea587d7 100644 --- a/synapse/metrics/__init__.py +++ b/synapse/metrics/__init__.py @@ -61,6 +61,7 @@ from twisted.web.server import Request # This module is imported for its side effects; flake8 needn't warn that it's unused. import synapse.metrics._reactor_metrics # noqa: F401 from synapse.metrics._gc import MIN_TIME_BETWEEN_GCS, install_gc_manager +from synapse.metrics._thread_cpu import thread_cpu_metrics from synapse.metrics._types import Collector from synapse.synapse_rust import get_rustc_version from synapse.types import StrSequence @@ -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..7983b1fae9 --- /dev/null +++ b/synapse/metrics/_thread_cpu.py @@ -0,0 +1,178 @@ +# +# 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.""" + + # When the thread started, in ticks since boot. Tells apart a new thread + # that reuses the ID of one that exited. + start_time: int + # The user and system CPU the thread had used, in ticks. + utime: int + stime: int + + +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, each + registered threadpool, the tokio runtime, and `other` for everything else. + + 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 + + self._threadpools: "WeakKeyDictionary[ThreadPool, str]" = WeakKeyDictionary() + + # Guards the fields below, as the metrics listener may scrape from any + # thread. + self._lock = threading.Lock() + # 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_`.""" + self._threadpools[threadpool] = name + + def _threadpool_thread_types(self) -> dict[int, str]: + """Map the native IDs of registered threadpool threads to their thread + type.""" + types: dict[int, str] = {} + 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, threadpool_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. + threadpool_types: The output of `_threadpool_thread_types`. + + Returns: + `main`, `threadpool_`, `tokio` or `other`. + """ + if tid == os.getpid(): + return "main" + + threadpool_type = threadpool_types.get(tid) + if threadpool_type is not None: + return threadpool_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.""" + try: + tids = [int(tid) for tid in os.listdir("/proc/self/task")] + except OSError: + return + + threadpool_types = self._threadpool_thread_types() + + with self._lock: + 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. + continue + + # The thread name may contain spaces or brackets, so split after + # its closing bracket. See `man 5 proc` for the field numbers. + 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 + else: + prev_utime = prev_stime = 0 + + thread_type = self._thread_type(tid, threadpool_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)