From 8a4b3738f3dcf49f70c08204ad457559dcef4112 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff <1389908+richvdh@users.noreply.github.com> Date: Thu, 4 Mar 2021 16:40:18 +0000 Subject: Replace `last_*_pdu_age` metrics with timestamps (#9540) Following the advice at https://prometheus.io/docs/practices/instrumentation/#timestamps-not-time-since, it's preferable to export unix timestamps, not ages. There doesn't seem to be any particular naming convention for timestamp metrics. --- synapse/federation/federation_server.py | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) (limited to 'synapse/federation/federation_server.py') diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index 2f832b47f6..362895bf42 100644 --- a/synapse/federation/federation_server.py +++ b/synapse/federation/federation_server.py @@ -90,10 +90,9 @@ pdu_process_time = Histogram( "Time taken to process an event", ) - -last_pdu_age_metric = Gauge( - "synapse_federation_last_received_pdu_age", - "The age (in seconds) of the last PDU successfully received from the given domain", +last_pdu_ts_metric = Gauge( + "synapse_federation_last_received_pdu_time", + "The timestamp of the last PDU which was successfully received from the given domain", labelnames=("server_name",), ) @@ -369,8 +368,7 @@ class FederationServer(FederationBase): ) if newest_pdu_ts and origin in self._federation_metrics_domains: - newest_pdu_age = self._clock.time_msec() - newest_pdu_ts - last_pdu_age_metric.labels(server_name=origin).set(newest_pdu_age / 1000) + last_pdu_ts_metric.labels(server_name=origin).set(newest_pdu_ts / 1000) return pdu_results -- cgit 1.5.1 From d6196efafcc312472464c882ab630bc3fbf7bd37 Mon Sep 17 00:00:00 2001 From: Jonathan de Jong Date: Mon, 8 Mar 2021 20:00:07 +0100 Subject: Add ResponseCache tests. (#9458) --- changelog.d/9458.misc | 1 + synapse/appservice/api.py | 2 +- synapse/federation/federation_server.py | 13 ++-- synapse/handlers/initial_sync.py | 2 +- synapse/handlers/room.py | 2 +- synapse/handlers/room_list.py | 4 +- synapse/handlers/sync.py | 2 +- synapse/replication/http/_base.py | 9 ++- synapse/util/caches/response_cache.py | 10 +-- tests/util/caches/test_responsecache.py | 131 ++++++++++++++++++++++++++++++++ 10 files changed, 156 insertions(+), 20 deletions(-) create mode 100644 changelog.d/9458.misc create mode 100644 tests/util/caches/test_responsecache.py (limited to 'synapse/federation/federation_server.py') diff --git a/changelog.d/9458.misc b/changelog.d/9458.misc new file mode 100644 index 0000000000..8ceeed1352 --- /dev/null +++ b/changelog.d/9458.misc @@ -0,0 +1 @@ +Add tests to ResponseCache. \ No newline at end of file diff --git a/synapse/appservice/api.py b/synapse/appservice/api.py index 93c2aabcca..9d3bbe3b8b 100644 --- a/synapse/appservice/api.py +++ b/synapse/appservice/api.py @@ -90,7 +90,7 @@ class ApplicationServiceApi(SimpleHttpClient): self.clock = hs.get_clock() self.protocol_meta_cache = ResponseCache( - hs, "as_protocol_meta", timeout_ms=HOUR_IN_MS + hs.get_clock(), "as_protocol_meta", timeout_ms=HOUR_IN_MS ) # type: ResponseCache[Tuple[str, str]] async def query_user(self, service, user_id): diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index 362895bf42..7657697bfa 100644 --- a/synapse/federation/federation_server.py +++ b/synapse/federation/federation_server.py @@ -22,6 +22,7 @@ from typing import ( Awaitable, Callable, Dict, + Iterable, List, Optional, Tuple, @@ -98,7 +99,7 @@ last_pdu_ts_metric = Gauge( class FederationServer(FederationBase): - def __init__(self, hs): + def __init__(self, hs: "HomeServer"): super().__init__(hs) self.auth = hs.get_auth() @@ -118,7 +119,7 @@ class FederationServer(FederationBase): # We cache results for transaction with the same ID self._transaction_resp_cache = ResponseCache( - hs, "fed_txn_handler", timeout_ms=30000 + hs.get_clock(), "fed_txn_handler", timeout_ms=30000 ) # type: ResponseCache[Tuple[str, str]] self.transaction_actions = TransactionActions(self.store) @@ -128,10 +129,10 @@ class FederationServer(FederationBase): # We cache responses to state queries, as they take a while and often # come in waves. self._state_resp_cache = ResponseCache( - hs, "state_resp", timeout_ms=30000 + hs.get_clock(), "state_resp", timeout_ms=30000 ) # type: ResponseCache[Tuple[str, str]] self._state_ids_resp_cache = ResponseCache( - hs, "state_ids_resp", timeout_ms=30000 + hs.get_clock(), "state_ids_resp", timeout_ms=30000 ) # type: ResponseCache[Tuple[str, str]] self._federation_metrics_domains = ( @@ -453,7 +454,9 @@ class FederationServer(FederationBase): self, room_id: str, event_id: str ) -> Dict[str, list]: if event_id: - pdus = await self.handler.get_state_for_pdu(room_id, event_id) + pdus = await self.handler.get_state_for_pdu( + room_id, event_id + ) # type: Iterable[EventBase] else: pdus = (await self.state.get_current_state(room_id)).values() diff --git a/synapse/handlers/initial_sync.py b/synapse/handlers/initial_sync.py index 71a5076672..13f8152283 100644 --- a/synapse/handlers/initial_sync.py +++ b/synapse/handlers/initial_sync.py @@ -48,7 +48,7 @@ class InitialSyncHandler(BaseHandler): self.clock = hs.get_clock() self.validator = EventValidator() self.snapshot_cache = ResponseCache( - hs, "initial_sync_cache" + hs.get_clock(), "initial_sync_cache" ) # type: ResponseCache[Tuple[str, Optional[StreamToken], Optional[StreamToken], str, Optional[int], bool, bool]] self._event_serializer = hs.get_event_client_serializer() self.storage = hs.get_storage() diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py index a488df10d6..4b3d0d72e3 100644 --- a/synapse/handlers/room.py +++ b/synapse/handlers/room.py @@ -121,7 +121,7 @@ class RoomCreationHandler(BaseHandler): # succession, only process the first attempt and return its result to # subsequent requests self._upgrade_response_cache = ResponseCache( - hs, "room_upgrade", timeout_ms=FIVE_MINUTES_IN_MS + hs.get_clock(), "room_upgrade", timeout_ms=FIVE_MINUTES_IN_MS ) # type: ResponseCache[Tuple[str, str]] self._server_notices_mxid = hs.config.server_notices_mxid diff --git a/synapse/handlers/room_list.py b/synapse/handlers/room_list.py index 14f14db449..8bfc46c654 100644 --- a/synapse/handlers/room_list.py +++ b/synapse/handlers/room_list.py @@ -44,10 +44,10 @@ class RoomListHandler(BaseHandler): super().__init__(hs) self.enable_room_list_search = hs.config.enable_room_list_search self.response_cache = ResponseCache( - hs, "room_list" + hs.get_clock(), "room_list" ) # type: ResponseCache[Tuple[Optional[int], Optional[str], ThirdPartyInstanceID]] self.remote_response_cache = ResponseCache( - hs, "remote_room_list", timeout_ms=30 * 1000 + hs.get_clock(), "remote_room_list", timeout_ms=30 * 1000 ) # type: ResponseCache[Tuple[str, Optional[int], Optional[str], bool, Optional[str]]] async def get_local_public_room_list( diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index 4e8ed7b33f..f50257cd57 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -244,7 +244,7 @@ class SyncHandler: self.event_sources = hs.get_event_sources() self.clock = hs.get_clock() self.response_cache = ResponseCache( - hs, "sync" + hs.get_clock(), "sync" ) # type: ResponseCache[Tuple[Any, ...]] self.state = hs.get_state_handler() self.auth = hs.get_auth() diff --git a/synapse/replication/http/_base.py b/synapse/replication/http/_base.py index 8a3f113e76..b7aa0c280f 100644 --- a/synapse/replication/http/_base.py +++ b/synapse/replication/http/_base.py @@ -18,7 +18,7 @@ import logging import re import urllib from inspect import signature -from typing import Dict, List, Tuple +from typing import TYPE_CHECKING, Dict, List, Tuple from prometheus_client import Counter, Gauge @@ -28,6 +28,9 @@ from synapse.logging.opentracing import inject_active_span_byte_dict, trace from synapse.util.caches.response_cache import ResponseCache from synapse.util.stringutils import random_string +if TYPE_CHECKING: + from synapse.server import HomeServer + logger = logging.getLogger(__name__) _pending_outgoing_requests = Gauge( @@ -88,10 +91,10 @@ class ReplicationEndpoint(metaclass=abc.ABCMeta): CACHE = True RETRY_ON_TIMEOUT = True - def __init__(self, hs): + def __init__(self, hs: "HomeServer"): if self.CACHE: self.response_cache = ResponseCache( - hs, "repl." + self.NAME, timeout_ms=30 * 60 * 1000 + hs.get_clock(), "repl." + self.NAME, timeout_ms=30 * 60 * 1000 ) # type: ResponseCache[str] # We reserve `instance_name` as a parameter to sending requests, so we diff --git a/synapse/util/caches/response_cache.py b/synapse/util/caches/response_cache.py index 32228f42ee..46ea8e0964 100644 --- a/synapse/util/caches/response_cache.py +++ b/synapse/util/caches/response_cache.py @@ -13,17 +13,15 @@ # See the License for the specific language governing permissions and # limitations under the License. import logging -from typing import TYPE_CHECKING, Any, Callable, Dict, Generic, Optional, TypeVar +from typing import Any, Callable, Dict, Generic, Optional, TypeVar from twisted.internet import defer from synapse.logging.context import make_deferred_yieldable, run_in_background +from synapse.util import Clock from synapse.util.async_helpers import ObservableDeferred from synapse.util.caches import register_cache -if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer - logger = logging.getLogger(__name__) T = TypeVar("T") @@ -37,11 +35,11 @@ class ResponseCache(Generic[T]): used rather than trying to compute a new response. """ - def __init__(self, hs: "HomeServer", name: str, timeout_ms: float = 0): + def __init__(self, clock: Clock, name: str, timeout_ms: float = 0): # Requests that haven't finished yet. self.pending_result_cache = {} # type: Dict[T, ObservableDeferred] - self.clock = hs.get_clock() + self.clock = clock self.timeout_sec = timeout_ms / 1000.0 self._name = name diff --git a/tests/util/caches/test_responsecache.py b/tests/util/caches/test_responsecache.py new file mode 100644 index 0000000000..f9a187b8de --- /dev/null +++ b/tests/util/caches/test_responsecache.py @@ -0,0 +1,131 @@ +# Copyright 2021 The Matrix.org Foundation C.I.C. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from synapse.util.caches.response_cache import ResponseCache + +from tests.server import get_clock +from tests.unittest import TestCase + + +class DeferredCacheTestCase(TestCase): + """ + A TestCase class for ResponseCache. + + The test-case function naming has some logic to it in it's parts, here's some notes about it: + wait: Denotes tests that have an element of "waiting" before its wrapped result becomes available + (Generally these just use .delayed_return instead of .instant_return in it's wrapped call.) + expire: Denotes tests that test expiry after assured existence. + (These have cache with a short timeout_ms=, shorter than will be tested through advancing the clock) + """ + + def setUp(self): + self.reactor, self.clock = get_clock() + + def with_cache(self, name: str, ms: int = 0) -> ResponseCache: + return ResponseCache(self.clock, name, timeout_ms=ms) + + @staticmethod + async def instant_return(o: str) -> str: + return o + + async def delayed_return(self, o: str) -> str: + await self.clock.sleep(1) + return o + + def test_cache_hit(self): + cache = self.with_cache("keeping_cache", ms=9001) + + expected_result = "howdy" + + wrap_d = cache.wrap(0, self.instant_return, expected_result) + + self.assertEqual( + expected_result, + self.successResultOf(wrap_d), + "initial wrap result should be the same", + ) + self.assertEqual( + expected_result, + self.successResultOf(cache.get(0)), + "cache should have the result", + ) + + def test_cache_miss(self): + cache = self.with_cache("trashing_cache", ms=0) + + expected_result = "howdy" + + wrap_d = cache.wrap(0, self.instant_return, expected_result) + + self.assertEqual( + expected_result, + self.successResultOf(wrap_d), + "initial wrap result should be the same", + ) + self.assertIsNone(cache.get(0), "cache should not have the result now") + + def test_cache_expire(self): + cache = self.with_cache("short_cache", ms=1000) + + expected_result = "howdy" + + wrap_d = cache.wrap(0, self.instant_return, expected_result) + + self.assertEqual(expected_result, self.successResultOf(wrap_d)) + self.assertEqual( + expected_result, + self.successResultOf(cache.get(0)), + "cache should still have the result", + ) + + # cache eviction timer is handled + self.reactor.pump((2,)) + + self.assertIsNone(cache.get(0), "cache should not have the result now") + + def test_cache_wait_hit(self): + cache = self.with_cache("neutral_cache") + + expected_result = "howdy" + + wrap_d = cache.wrap(0, self.delayed_return, expected_result) + self.assertNoResult(wrap_d) + + # function wakes up, returns result + self.reactor.pump((2,)) + + self.assertEqual(expected_result, self.successResultOf(wrap_d)) + + def test_cache_wait_expire(self): + cache = self.with_cache("medium_cache", ms=3000) + + expected_result = "howdy" + + wrap_d = cache.wrap(0, self.delayed_return, expected_result) + self.assertNoResult(wrap_d) + + # stop at 1 second to callback cache eviction callLater at that time, then another to set time at 2 + self.reactor.pump((1, 1)) + + self.assertEqual(expected_result, self.successResultOf(wrap_d)) + self.assertEqual( + expected_result, + self.successResultOf(cache.get(0)), + "cache should still have the result", + ) + + # (1 + 1 + 2) > 3.0, cache eviction timer is handled + self.reactor.pump((2,)) + + self.assertIsNone(cache.get(0), "cache should not have the result now") -- cgit 1.5.1 From 7fdc6cefb3017f3c4bbe1d824e1de249ac29d219 Mon Sep 17 00:00:00 2001 From: Patrick Cloke Date: Tue, 9 Mar 2021 07:41:32 -0500 Subject: Fix additional type hints. (#9543) Type hint fixes due to Twisted 21.2.0 adding type hints. --- changelog.d/9543.misc | 1 + synapse/config/logger.py | 5 ++++- synapse/federation/federation_server.py | 2 +- synapse/handlers/pagination.py | 2 +- synapse/http/federation/well_known_resolver.py | 3 ++- synapse/logging/context.py | 6 ++++-- tests/replication/_base.py | 27 ++++++++++++++++---------- tests/server.py | 2 +- tests/test_utils/logging_setup.py | 2 +- 9 files changed, 32 insertions(+), 18 deletions(-) create mode 100644 changelog.d/9543.misc (limited to 'synapse/federation/federation_server.py') diff --git a/changelog.d/9543.misc b/changelog.d/9543.misc new file mode 100644 index 0000000000..14c7b78dd9 --- /dev/null +++ b/changelog.d/9543.misc @@ -0,0 +1 @@ +Fix incorrect type hints. diff --git a/synapse/config/logger.py b/synapse/config/logger.py index e56cf846f5..999aecce5c 100644 --- a/synapse/config/logger.py +++ b/synapse/config/logger.py @@ -21,8 +21,10 @@ import threading from string import Template import yaml +from zope.interface import implementer from twisted.logger import ( + ILogObserver, LogBeginner, STDLibLogObserver, eventAsText, @@ -227,7 +229,8 @@ def _setup_stdlib_logging(config, log_config_path, logBeginner: LogBeginner) -> threadlocal = threading.local() - def _log(event): + @implementer(ILogObserver) + def _log(event: dict) -> None: if "log_text" in event: if event["log_text"].startswith("DNSDatagramProtocol starting on "): return diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index 7657697bfa..ffc735ba25 100644 --- a/synapse/federation/federation_server.py +++ b/synapse/federation/federation_server.py @@ -361,7 +361,7 @@ class FederationServer(FederationBase): logger.error( "Failed to handle PDU %s", event_id, - exc_info=(f.type, f.value, f.getTracebackObject()), + exc_info=(f.type, f.value, f.getTracebackObject()), # type: ignore ) await concurrently_execute( diff --git a/synapse/handlers/pagination.py b/synapse/handlers/pagination.py index 059064a4eb..66dc886c81 100644 --- a/synapse/handlers/pagination.py +++ b/synapse/handlers/pagination.py @@ -285,7 +285,7 @@ class PaginationHandler: except Exception: f = Failure() logger.error( - "[purge] failed", exc_info=(f.type, f.value, f.getTracebackObject()) + "[purge] failed", exc_info=(f.type, f.value, f.getTracebackObject()) # type: ignore ) self._purges_by_id[purge_id].status = PurgeStatus.STATUS_FAILED finally: diff --git a/synapse/http/federation/well_known_resolver.py b/synapse/http/federation/well_known_resolver.py index 4def7d7633..ecd63e6596 100644 --- a/synapse/http/federation/well_known_resolver.py +++ b/synapse/http/federation/well_known_resolver.py @@ -322,7 +322,8 @@ def _cache_period_from_headers( def _parse_cache_control(headers: Headers) -> Dict[bytes, Optional[bytes]]: cache_controls = {} - for hdr in headers.getRawHeaders(b"cache-control", []): + cache_control_headers = headers.getRawHeaders(b"cache-control") or [] + for hdr in cache_control_headers: for directive in hdr.split(b","): splits = [x.strip() for x in directive.split(b"=", 1)] k = splits[0].lower() diff --git a/synapse/logging/context.py b/synapse/logging/context.py index 78e27bfb00..1a7ea4fa96 100644 --- a/synapse/logging/context.py +++ b/synapse/logging/context.py @@ -669,7 +669,7 @@ def preserve_fn(f): return g -def run_in_background(f, *args, **kwargs): +def run_in_background(f, *args, **kwargs) -> defer.Deferred: """Calls a function, ensuring that the current context is restored after return from the function, and that the sentinel context is set once the deferred returned by the function completes. @@ -697,8 +697,10 @@ def run_in_background(f, *args, **kwargs): if isinstance(res, types.CoroutineType): res = defer.ensureDeferred(res) + # At this point we should have a Deferred, if not then f was a synchronous + # function, wrap it in a Deferred for consistency. if not isinstance(res, defer.Deferred): - return res + return defer.succeed(res) if res.called and not res.paused: # The function should have maintained the logcontext, so we can diff --git a/tests/replication/_base.py b/tests/replication/_base.py index f6a6aed35e..20940c8107 100644 --- a/tests/replication/_base.py +++ b/tests/replication/_base.py @@ -22,6 +22,7 @@ from twisted.internet.protocol import Protocol from twisted.internet.task import LoopingCall from twisted.web.http import HTTPChannel from twisted.web.resource import Resource +from twisted.web.server import Request, Site from synapse.app.generic_worker import ( GenericWorkerReplicationHandler, @@ -32,7 +33,10 @@ from synapse.http.site import SynapseRequest, SynapseSite from synapse.replication.http import ReplicationRestResource from synapse.replication.tcp.handler import ReplicationCommandHandler from synapse.replication.tcp.protocol import ClientReplicationStreamProtocol -from synapse.replication.tcp.resource import ReplicationStreamProtocolFactory +from synapse.replication.tcp.resource import ( + ReplicationStreamProtocolFactory, + ServerReplicationStreamProtocol, +) from synapse.server import HomeServer from synapse.util import Clock @@ -59,7 +63,9 @@ class BaseStreamTestCase(unittest.HomeserverTestCase): # build a replication server server_factory = ReplicationStreamProtocolFactory(hs) self.streamer = hs.get_replication_streamer() - self.server = server_factory.buildProtocol(None) + self.server = server_factory.buildProtocol( + None + ) # type: ServerReplicationStreamProtocol # Make a new HomeServer object for the worker self.reactor.lookups["testserv"] = "1.2.3.4" @@ -155,9 +161,7 @@ class BaseStreamTestCase(unittest.HomeserverTestCase): request_factory = OneShotRequestFactory() # Set up the server side protocol - channel = _PushHTTPChannel(self.reactor) - channel.requestFactory = request_factory - channel.site = self.site + channel = _PushHTTPChannel(self.reactor, request_factory, self.site) # Connect client to server and vice versa. client_to_server_transport = FakeTransport( @@ -188,8 +192,9 @@ class BaseStreamTestCase(unittest.HomeserverTestCase): fetching updates for given stream. """ + path = request.path # type: bytes # type: ignore self.assertRegex( - request.path, + path, br"^/_synapse/replication/get_repl_stream_updates/%s/[^/]+$" % (stream_name.encode("ascii"),), ) @@ -390,9 +395,7 @@ class BaseMultiWorkerStreamTestCase(unittest.HomeserverTestCase): request_factory = OneShotRequestFactory() # Set up the server side protocol - channel = _PushHTTPChannel(self.reactor) - channel.requestFactory = request_factory - channel.site = self._hs_to_site[hs] + channel = _PushHTTPChannel(self.reactor, request_factory, self._hs_to_site[hs]) # Connect client to server and vice versa. client_to_server_transport = FakeTransport( @@ -475,9 +478,13 @@ class _PushHTTPChannel(HTTPChannel): makes it very hard to test. """ - def __init__(self, reactor: IReactorTime): + def __init__( + self, reactor: IReactorTime, request_factory: Callable[..., Request], site: Site + ): super().__init__() self.reactor = reactor + self.requestFactory = request_factory + self.site = site self._pull_to_push_producer = None # type: Optional[_PullToPushProducer] diff --git a/tests/server.py b/tests/server.py index 939a0008ca..863f6da738 100644 --- a/tests/server.py +++ b/tests/server.py @@ -188,7 +188,7 @@ class FakeSite: def make_request( reactor, - site: Site, + site: Union[Site, FakeSite], method, path, content=b"", diff --git a/tests/test_utils/logging_setup.py b/tests/test_utils/logging_setup.py index 52ae5c5713..74568b34f8 100644 --- a/tests/test_utils/logging_setup.py +++ b/tests/test_utils/logging_setup.py @@ -28,7 +28,7 @@ class ToTwistedHandler(logging.Handler): def emit(self, record): log_entry = self.format(record) log_level = record.levelname.lower().replace("warning", "warn") - self.tx_log.emit( + self.tx_log.emit( # type: ignore twisted.logger.LogLevel.levelWithName(log_level), "{entry}", entry=log_entry ) -- cgit 1.5.1