diff --git a/synapse/logging/opentracing.py b/synapse/logging/opentracing.py index 6e4e029163..19728b1e8a 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, @@ -492,8 +493,17 @@ def init_tracer(hs: "HomeServer") -> None: tracer = config.create_tracer(RustReporter(), config.sampler) opentracing.set_global_tracer(tracer) else: + threads_before = set(threading.enumerate()) config.initialize_tracer() + # jaeger-client's reporter sends spans from a thread it starts without + # a name. Give its CPU time a type of its own. + from synapse.metrics._thread_cpu import thread_cpu_metrics + + 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/_thread_cpu.py b/synapse/metrics/_thread_cpu.py index 7983b1fae9..0c896a3fd4 100644 --- a/synapse/metrics/_thread_cpu.py +++ b/synapse/metrics/_thread_cpu.py @@ -65,6 +65,7 @@ class ThreadCPUMetrics(Collector): pass self._threadpools: "WeakKeyDictionary[ThreadPool, str]" = WeakKeyDictionary() + self._threads: "WeakKeyDictionary[threading.Thread, str]" = WeakKeyDictionary() # Guards the fields below, as the metrics listener may scrape from any # thread. @@ -78,10 +79,18 @@ class ThreadCPUMetrics(Collector): """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.""" + def register_thread(self, thread_type: str, thread: threading.Thread) -> None: + """Report `thread` as the type `thread_type`.""" + 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.""" 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. @@ -91,22 +100,22 @@ class ThreadCPUMetrics(Collector): return types - def _thread_type(self, tid: int, threadpool_types: dict[int, str]) -> str: + 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. - threadpool_types: The output of `_threadpool_thread_types`. + registered_types: The output of `_registered_thread_types`. Returns: - `main`, `threadpool_`, `tokio` or `other`. + `main`, a registered type, `tokio` or `other`. """ if tid == os.getpid(): return "main" - threadpool_type = threadpool_types.get(tid) - if threadpool_type is not None: - return threadpool_type + 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: @@ -128,7 +137,7 @@ class ThreadCPUMetrics(Collector): except OSError: return - threadpool_types = self._threadpool_thread_types() + registered_types = self._registered_thread_types() with self._lock: last_seen: dict[int, _ThreadSample] = {} @@ -153,7 +162,7 @@ class ThreadCPUMetrics(Collector): else: prev_utime = prev_stime = 0 - thread_type = self._thread_type(tid, threadpool_types) + thread_type = self._thread_type(tid, registered_types) self._totals[(thread_type, "user")] += utime - prev_utime self._totals[(thread_type, "system")] += stime - prev_stime