From d37b725ad68bd130f3727802f9ee4568a1df0d35 Mon Sep 17 00:00:00 2001 From: Patrick Cloke Date: Fri, 20 Oct 2023 15:49:46 -0400 Subject: [PATCH 1/6] Support reactor timing on more reactors. --- mypy.ini | 4 +- synapse/metrics/_reactor_metrics.py | 118 +++++++++++++++++++++------- 2 files changed, 93 insertions(+), 29 deletions(-) diff --git a/mypy.ini b/mypy.ini index fdfe9432fcc7..1a2b9ea410b0 100644 --- a/mypy.ini +++ b/mypy.ini @@ -37,8 +37,8 @@ files = build_rust.py [mypy-synapse.metrics._reactor_metrics] -# This module imports select.epoll. That exists on Linux, but doesn't on macOS. -# See https://github.com/matrix-org/synapse/pull/11771. +# This module pokes at the internals of OS-specific classes, to appease mypy +# on different systems we add additional ignores. warn_unused_ignores = False [mypy-synapse.util.caches.treecache] diff --git a/synapse/metrics/_reactor_metrics.py b/synapse/metrics/_reactor_metrics.py index a2c6e6842d0c..8f6abf5a67a2 100644 --- a/synapse/metrics/_reactor_metrics.py +++ b/synapse/metrics/_reactor_metrics.py @@ -12,17 +12,46 @@ # See the License for the specific language governing permissions and # limitations under the License. -import select +import logging import time -from typing import Any, Iterable, List, Tuple +from selectors import SelectSelector, _PollLikeSelector # type: ignore[attr-defined] +from typing import Any, Callable, Iterable from prometheus_client import Histogram, Metric from prometheus_client.core import REGISTRY, GaugeMetricFamily from twisted.internet import reactor +from twisted.internet.asyncioreactor import AsyncioSelectorReactor +from twisted.internet.selectreactor import SelectReactor from synapse.metrics._types import Collector +try: + from selectors import KqueueSelector +except ImportError: + + class KqueueSelector: # type: ignore[no-redef] + pass + + +try: + from twisted.internet.epollreactor import EPollReactor +except ImportError: + + class EPollReactor: # type: ignore[no-redef] + pass + + +try: + from twisted.internet.pollreactor import PollReactor +except ImportError: + + class PollReactor: # type: ignore[no-redef] + pass + + +logger = logging.getLogger(__name__) + # # Twisted reactor metrics # @@ -34,52 +63,87 @@ ) -class EpollWrapper: - """a wrapper for an epoll object which records the time between polls""" +class CallWrapper: + """a wrapper for a callable which records the time between calls""" - def __init__(self, poller: "select.epoll"): # type: ignore[name-defined] + def __init__(self, wrapped: Callable[..., Any]): self.last_polled = time.time() - self._poller = poller + self._wrapped = wrapped - def poll(self, *args, **kwargs) -> List[Tuple[int, int]]: # type: ignore[no-untyped-def] - # record the time since poll() was last called. This gives a good proxy for + def __call__(self, *args, **kwargs) -> Any: # type: ignore[no-untyped-def] + # record the time since this was last called. This gives a good proxy for # how long it takes to run everything in the reactor - ie, how long anything # waiting for the next tick will have to wait. tick_time.observe(time.time() - self.last_polled) - ret = self._poller.poll(*args, **kwargs) + ret = self._wrapped(*args, **kwargs) self.last_polled = time.time() return ret + +class ObjWrapper: + """a wrapper for an callable which records the time between calls""" + + def __init__(self, wrapped: Any, method: str): + self._wrapped = wrapped + self._method = method + self._wrapped_method = CallWrapper(getattr(wrapped, method)) + def __getattr__(self, item: str) -> Any: - return getattr(self._poller, item) + if item == self._method: + return self._wrapped_method + + return getattr(self._wrapped, item) class ReactorLastSeenMetric(Collector): - def __init__(self, epoll_wrapper: EpollWrapper): - self._epoll_wrapper = epoll_wrapper + def __init__(self, call_wrapper: CallWrapper): + self._call_wrapper = call_wrapper def collect(self) -> Iterable[Metric]: cm = GaugeMetricFamily( "python_twisted_reactor_last_seen", "Seconds since the Twisted reactor was last seen", ) - cm.add_metric([], time.time() - self._epoll_wrapper.last_polled) + cm.add_metric([], time.time() - self._call_wrapper.last_polled) yield cm -try: - # if the reactor has a `_poller` attribute, which is an `epoll` object - # (ie, it's an EPollReactor), we wrap the `epoll` with a thing that will - # measure the time between ticks - from select import epoll # type: ignore[attr-defined] - - poller = reactor._poller # type: ignore[attr-defined] -except (AttributeError, ImportError): - pass -else: - if isinstance(poller, epoll): - poller = EpollWrapper(poller) - reactor._poller = poller # type: ignore[attr-defined] - REGISTRY.register(ReactorLastSeenMetric(poller)) +# Twisted has already select a reasonable reactor for us, so assumptions can be +# made about the shape. +wrapper = None +if isinstance(reactor, (PollReactor, EPollReactor)): + wrapper = reactor._poller.poll = CallWrapper(reactor._poller.poll) + +elif isinstance(reactor, SelectReactor): + from twisted.internet import selectreactor + + wrapper = selectreactor._select = CallWrapper(selectreactor._select) + +elif isinstance(reactor, AsyncioSelectorReactor): + # For asyncio we need to go deeper. + asyncio_loop = reactor._asyncioEventloop # A sub-class of BaseEventLoop, + + # If an unexpected asyncio loop implementation is used, these might fail. + try: + # A sub-class of BaseSelector. + selector = asyncio_loop._selector # type: ignore[attr-defined] + + if isinstance(selector, SelectSelector): + wrapper = selector._select = CallWrapper(selector._select) # type: ignore[attr-defined] + + elif isinstance(selector, _PollLikeSelector): + selector._selector = ObjWrapper(selector._selector, "poll") # type: ignore[attr-defined] + wrapper = selector._selector._wrapped_method # type: ignore[attr-defined] + + elif isinstance(selector, KqueueSelector): + selector._selector = ObjWrapper(selector._selector, "control") # type: ignore[attr-defined] + wrapper = selector._selector._wrapped_method # type: ignore[attr-defined] + + # XXX Will not work on (Windows-only) ProactorEventLoop. + except AttributeError: + logger.warn("Unexpected asyncio loop: %r", asyncio_loop) + +if wrapper: + REGISTRY.register(ReactorLastSeenMetric(wrapper)) From cbf796433966cc278283cf18001d21893a6a8d4a Mon Sep 17 00:00:00 2001 From: Patrick Cloke Date: Fri, 20 Oct 2023 15:53:09 -0400 Subject: [PATCH 2/6] Add a comment. --- synapse/metrics/_reactor_metrics.py | 1 + 1 file changed, 1 insertion(+) diff --git a/synapse/metrics/_reactor_metrics.py b/synapse/metrics/_reactor_metrics.py index 8f6abf5a67a2..00946d85eb13 100644 --- a/synapse/metrics/_reactor_metrics.py +++ b/synapse/metrics/_reactor_metrics.py @@ -133,6 +133,7 @@ def collect(self) -> Iterable[Metric]: if isinstance(selector, SelectSelector): wrapper = selector._select = CallWrapper(selector._select) # type: ignore[attr-defined] + # poll, epoll, and /dev/poll. elif isinstance(selector, _PollLikeSelector): selector._selector = ObjWrapper(selector._selector, "poll") # type: ignore[attr-defined] wrapper = selector._selector._wrapped_method # type: ignore[attr-defined] From fdb3ce0035b87242efe99e347fc560c2d075e3b3 Mon Sep 17 00:00:00 2001 From: Patrick Cloke Date: Fri, 20 Oct 2023 15:53:28 -0400 Subject: [PATCH 3/6] Newsfragment --- changelog.d/16532.misc | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog.d/16532.misc diff --git a/changelog.d/16532.misc b/changelog.d/16532.misc new file mode 100644 index 000000000000..437e00210b96 --- /dev/null +++ b/changelog.d/16532.misc @@ -0,0 +1 @@ +Support reactor tick timings on more types of event loops. From 6318c33a1c213acb1220b6822b9cd9fc58ccdfee Mon Sep 17 00:00:00 2001 From: Patrick Cloke Date: Fri, 20 Oct 2023 15:58:48 -0400 Subject: [PATCH 4/6] Minor tweaks. --- synapse/metrics/_reactor_metrics.py | 28 ++++++++++++++++------------ 1 file changed, 16 insertions(+), 12 deletions(-) diff --git a/synapse/metrics/_reactor_metrics.py b/synapse/metrics/_reactor_metrics.py index 00946d85eb13..0ff4113381f6 100644 --- a/synapse/metrics/_reactor_metrics.py +++ b/synapse/metrics/_reactor_metrics.py @@ -20,9 +20,8 @@ from prometheus_client import Histogram, Metric from prometheus_client.core import REGISTRY, GaugeMetricFamily -from twisted.internet import reactor +from twisted.internet import reactor, selectreactor from twisted.internet.asyncioreactor import AsyncioSelectorReactor -from twisted.internet.selectreactor import SelectReactor from synapse.metrics._types import Collector @@ -64,7 +63,7 @@ class PollReactor: # type: ignore[no-redef] class CallWrapper: - """a wrapper for a callable which records the time between calls""" + """A wrapper for a callable which records the time between calls""" def __init__(self, wrapped: Callable[..., Any]): self.last_polled = time.time() @@ -83,15 +82,21 @@ def __call__(self, *args, **kwargs) -> Any: # type: ignore[no-untyped-def] class ObjWrapper: - """a wrapper for an callable which records the time between calls""" + """A wrapper for an object which wraps a specified method in CallWrapper. - def __init__(self, wrapped: Any, method: str): + Other methods/attributes are passed to the original object. + + This is necessary when the wrapped object does not allow the attribute to be + overwritten. + """ + + def __init__(self, wrapped: Any, method_name: str): self._wrapped = wrapped - self._method = method - self._wrapped_method = CallWrapper(getattr(wrapped, method)) + self._method_name = method_name + self._wrapped_method = CallWrapper(getattr(wrapped, method_name)) def __getattr__(self, item: str) -> Any: - if item == self._method: + if item == self._method_name: return self._wrapped_method return getattr(self._wrapped, item) @@ -116,13 +121,12 @@ def collect(self) -> Iterable[Metric]: if isinstance(reactor, (PollReactor, EPollReactor)): wrapper = reactor._poller.poll = CallWrapper(reactor._poller.poll) -elif isinstance(reactor, SelectReactor): - from twisted.internet import selectreactor - +elif isinstance(reactor, selectreactor.SelectReactor): + # Twisted uses a module-level _select function. wrapper = selectreactor._select = CallWrapper(selectreactor._select) elif isinstance(reactor, AsyncioSelectorReactor): - # For asyncio we need to go deeper. + # For asyncio look at the underlying asyncio event loop. asyncio_loop = reactor._asyncioEventloop # A sub-class of BaseEventLoop, # If an unexpected asyncio loop implementation is used, these might fail. From 20350c18c626c32fb91b4be29ad180cd30efdc2c Mon Sep 17 00:00:00 2001 From: Patrick Cloke Date: Mon, 23 Oct 2023 08:00:41 -0400 Subject: [PATCH 5/6] Fix-up patching EPollReactor. --- synapse/metrics/_reactor_metrics.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/synapse/metrics/_reactor_metrics.py b/synapse/metrics/_reactor_metrics.py index 0ff4113381f6..0fa485daaffb 100644 --- a/synapse/metrics/_reactor_metrics.py +++ b/synapse/metrics/_reactor_metrics.py @@ -119,7 +119,8 @@ def collect(self) -> Iterable[Metric]: # made about the shape. wrapper = None if isinstance(reactor, (PollReactor, EPollReactor)): - wrapper = reactor._poller.poll = CallWrapper(reactor._poller.poll) + reactor._poller = ObjWrapper(reactor._poller, "poll") # type: ignore[attr-defined] + wrapper = reactor._poller._wrapped_method # type: ignore[attr-defined] elif isinstance(reactor, selectreactor.SelectReactor): # Twisted uses a module-level _select function. From 2eac501f88e1843f3c2fb68c080c21f51756421d Mon Sep 17 00:00:00 2001 From: Patrick Cloke Date: Thu, 2 Nov 2023 08:47:40 -0400 Subject: [PATCH 6/6] Add a bigger try-except. --- synapse/metrics/_reactor_metrics.py | 34 +++++++++++++++++------------ 1 file changed, 20 insertions(+), 14 deletions(-) diff --git a/synapse/metrics/_reactor_metrics.py b/synapse/metrics/_reactor_metrics.py index 0fa485daaffb..dd486dd3e288 100644 --- a/synapse/metrics/_reactor_metrics.py +++ b/synapse/metrics/_reactor_metrics.py @@ -118,20 +118,19 @@ def collect(self) -> Iterable[Metric]: # Twisted has already select a reasonable reactor for us, so assumptions can be # made about the shape. wrapper = None -if isinstance(reactor, (PollReactor, EPollReactor)): - reactor._poller = ObjWrapper(reactor._poller, "poll") # type: ignore[attr-defined] - wrapper = reactor._poller._wrapped_method # type: ignore[attr-defined] +try: + if isinstance(reactor, (PollReactor, EPollReactor)): + reactor._poller = ObjWrapper(reactor._poller, "poll") # type: ignore[attr-defined] + wrapper = reactor._poller._wrapped_method # type: ignore[attr-defined] -elif isinstance(reactor, selectreactor.SelectReactor): - # Twisted uses a module-level _select function. - wrapper = selectreactor._select = CallWrapper(selectreactor._select) + elif isinstance(reactor, selectreactor.SelectReactor): + # Twisted uses a module-level _select function. + wrapper = selectreactor._select = CallWrapper(selectreactor._select) -elif isinstance(reactor, AsyncioSelectorReactor): - # For asyncio look at the underlying asyncio event loop. - asyncio_loop = reactor._asyncioEventloop # A sub-class of BaseEventLoop, + elif isinstance(reactor, AsyncioSelectorReactor): + # For asyncio look at the underlying asyncio event loop. + asyncio_loop = reactor._asyncioEventloop # A sub-class of BaseEventLoop, - # If an unexpected asyncio loop implementation is used, these might fail. - try: # A sub-class of BaseSelector. selector = asyncio_loop._selector # type: ignore[attr-defined] @@ -147,9 +146,16 @@ def collect(self) -> Iterable[Metric]: selector._selector = ObjWrapper(selector._selector, "control") # type: ignore[attr-defined] wrapper = selector._selector._wrapped_method # type: ignore[attr-defined] - # XXX Will not work on (Windows-only) ProactorEventLoop. - except AttributeError: - logger.warn("Unexpected asyncio loop: %r", asyncio_loop) + else: + # E.g. this does not support the (Windows-only) ProactorEventLoop. + logger.warning( + "Skipping configuring ReactorLastSeenMetric: unexpected asyncio loop selector: %r via %r", + selector, + asyncio_loop, + ) +except Exception as e: + logger.warning("Configuring ReactorLastSeenMetric failed: %r", e) + if wrapper: REGISTRY.register(ReactorLastSeenMetric(wrapper))