From 8bbeba48f483681b62b1a50d473a33aaaab745a7 Mon Sep 17 00:00:00 2001 From: Dale Wahl Date: Wed, 11 Nov 2026 10:10:37 +0100 Subject: [PATCH] memcache update: the simple bits --- common/config_manager.py | 181 ++++++++++++++++----------- tests/test_config_memcache.py | 227 ++++++++++++++++++++++++++++++++++ webtool/__init__.py | 6 +- 3 files changed, 342 insertions(+), 72 deletions(-) create mode 100644 tests/test_config_memcache.py diff --git a/common/config_manager.py b/common/config_manager.py index a1f71f5e5..7c0c676eb 100644 --- a/common/config_manager.py +++ b/common/config_manager.py @@ -1,5 +1,6 @@ import itertools import threading +import hashlib import pickle import time import json @@ -16,6 +17,9 @@ import configparser import os +# Errors that mean memcache cannot be used right now +MEMCACHE_ERRORS = (SystemError, ValueError, MemcacheError, ConnectionError, OSError) + class CacheMiss: """ Helper class to distinguish memcache misses from true `None` values @@ -40,6 +44,13 @@ class ConfigManager(BaseConfigReader): # Prevents creating a new TCP connection per request in threaded/gunicorn contexts. _memcache_tls = threading.local() + # Cached values expire after this many seconds. (Stops a value that could not be + # removed from memcache from being read forever by other 4CAT processes.) + memcache_expire = 600 + # Once memcache could not be reached, it is not used again until 4CAT restarts + _memcache_down = False + _memcache_down_lock = threading.Lock() + def __init__(self, db=None): # ensure core settings (including database config) are loaded self.load_core_settings() @@ -184,7 +195,13 @@ def get_memcache(self): The config reader can optionally use Memcache to keep fetched values in memory. + + :return MemcacheClient|None: The client, or `None` if memcache is not + configured or could not be reached """ + if ConfigManager._memcache_down: + return None + # Reuse per-thread client if already initialised. existing = getattr(self._memcache_tls, "client", None) if existing: @@ -193,21 +210,61 @@ def get_memcache(self): server = self.get("MEMCACHE_SERVER") if server: try: - memcache = MemcacheClient(server, serde=serde.pickle_serde, key_prefix=b"4cat-config") + memcache = MemcacheClient(server, serde=serde.pickle_serde, key_prefix=b"4cat-config", + connect_timeout=1, timeout=1) # do one test fetch to test if connection is valid memcache.set("4cat-init-dummy", time.time()) - memcache.init_thread_id = threading.get_ident() self._memcache_tls.client = memcache return memcache - except (SystemError, ValueError, MemcacheError, ConnectionError, OSError): - # we have no access to the logger here so we simply pass - # later we can detect elsewhere that a memcache address is - # configured but no connection is there - then we can log - # config reader still works without memcache - pass + except MEMCACHE_ERRORS as e: + self._memcache_failed(e) return None + def _memcache_command(self, command, *args, failed=None, **kwargs): + """ + Run a command on memcache, if it can be used + + Memcache only makes reading settings faster, so an error there must not + stop 4CAT: the caller reads from the database instead. A failed command + is tried once more with a new connection via pymemcache. If that fails too, + memcache is not used again until 4CAT restarts. In that process at least. + + :param str command: Name of the client method to run + :param failed: What to return if memcache cannot be used + :return: What the command returns, or `failed` + """ + error = None + for _ in range(2): + memcache = self.get_memcache() + if not memcache: + # memcache not available (either not configured or down) + return failed + + try: + return getattr(memcache, command)(*args, **kwargs) + except MEMCACHE_ERRORS as e: + error = e + + self._memcache_failed(error) + return failed + + def _memcache_failed(self, error): + """ + Stop using memcache after it could not be reached + + :param Exception error: The error memcache ran into + """ + with ConfigManager._memcache_down_lock: + was_up = not ConfigManager._memcache_down + ConfigManager._memcache_down = True + + if was_up and self.logger: + self.logger.warning(f"Memcache cannot be reached ({error!r}). This 4CAT process reads settings from " + f"the database instead until 4CAT is restarted. Restart 4CAT once memcache works " + f"again; until then, settings changed here can take up to " + f"{self.memcache_expire // 60} minutes to apply elsewhere in 4CAT.") + def close_memcache(self): """Close and dispose this thread's memcache client. @@ -288,7 +345,7 @@ def get_all_setting_names(self, with_core=True): return settings - def get_all(self, is_json=False, user=None, tags=None, with_core=True, memcache=None): + def get_all(self, is_json=False, user=None, tags=None, with_core=True): """ Get all known settings @@ -304,16 +361,15 @@ def get_all(self, is_json=False, user=None, tags=None, with_core=True, memcache= with the given tag, and returns that if one exists. First matching tag wins. :param bool with_core: Also include core (i.e. config.ini) settings - :param MemcacheClient memcache: Memcache client. If `None`, a thread-local client will be used. :return dict: Setting value, as a dictionary with setting names as keys and setting values as values. """ for setting in self.get_all_setting_names(with_core=with_core): - yield setting, self.get(setting, None, is_json, user, tags, memcache) + yield setting, self.get(setting, None, is_json, user, tags) - def get(self, attribute_name, default=None, is_json=False, user=None, tags=None, memcache=None): + def get(self, attribute_name, default=None, is_json=False, user=None, tags=None): """ Get a setting's value from the database @@ -329,7 +385,6 @@ def get(self, attribute_name, default=None, is_json=False, user=None, tags=None, provided, the method checks if a special value for the setting exists with the given tag, and returns that if one exists. First matching tag wins. - :param MemcacheClient memcache: Memcache client. If `None`, a thread-local client will be used. :return: Setting value, or the provided fallback, or `None`. """ @@ -353,7 +408,7 @@ def get(self, attribute_name, default=None, is_json=False, user=None, tags=None, # interacts badly with get_all() if tags: tags = tags.copy() - tags = self.get_active_tags(user, tags, memcache) + tags = self.get_active_tags(user, tags) # global settings are only ever stored with the global (empty) tag via # set(), so there can be no valid tag-specific overrides for them. @@ -367,21 +422,10 @@ def get(self, attribute_name, default=None, is_json=False, user=None, tags=None, # end to fall back to the global value if no specific one exists. tags.append("") - # Obtain thread-local memcache client if not explicitly given. - if not memcache: - memcache = self.get_memcache() - # first check if we have all the values in memcache, in which case we # do not need a database query - if memcache: - if threading.get_ident() != memcache.init_thread_id: - raise RuntimeError("Thread-unsafe use of memcache! Please make sure you are using a configuration " - "wrapper to read with a thread-local memcache connection.") - - cached_values = {tag: memcache.get(self._get_memcache_id(attribute_name, tag), default=CacheMiss) for tag in tags} - - else: - cached_values = {t: CacheMiss for t in tags} + cached_values = {tag: self._memcache_command("get", self._get_memcache_id(attribute_name, tag), + default=CacheMiss, failed=CacheMiss) for tag in tags} # for the tags we could not get from memcache, run a database query # (and save to cache if possible) @@ -392,9 +436,9 @@ def get(self, attribute_name, default=None, is_json=False, user=None, tags=None, replacements = (attribute_name, tuple(missing_tags)) queried_settings = {setting["tag"]: setting["value"] for setting in self.db.fetchall(query, replacements)} - if memcache: - for tag, value in queried_settings.items(): - memcache.set(self._get_memcache_id(attribute_name, tag), value) + for tag, value in queried_settings.items(): + self._memcache_command("set", self._get_memcache_id(attribute_name, tag), value, + expire=self.memcache_expire) cached_values.update(queried_settings) @@ -406,13 +450,13 @@ def get(self, attribute_name, default=None, is_json=False, user=None, tags=None, # replace the magic value with a CacheMiss in the dict that will be # parsed unconfigured_magic = "__unconfigured__" - if memcache: - for tag in [t for t in cached_values if cached_values[t] is CacheMiss]: - # should this be more magic? - memcache.set(self._get_memcache_id(attribute_name, tag), unconfigured_magic) + for tag in [t for t in cached_values if cached_values[t] is CacheMiss]: + # should this be more magic? + self._memcache_command("set", self._get_memcache_id(attribute_name, tag), unconfigured_magic, + expire=self.memcache_expire) - for tag in [t for t in cached_values if cached_values[t] == unconfigured_magic]: - cached_values[tag] = CacheMiss + for tag in [t for t in cached_values if cached_values[t] == unconfigured_magic]: + cached_values[tag] = CacheMiss # now we may still have some CacheMisses in the values dict, if there # was no setting in the database with that tag. So, find the first @@ -437,7 +481,7 @@ def get(self, attribute_name, default=None, is_json=False, user=None, tags=None, return value - def get_active_tags(self, user=None, tags=None, memcache=None): + def get_active_tags(self, user=None, tags=None): """ Get active tags for given user/tag list @@ -450,7 +494,6 @@ def get_active_tags(self, user=None, tags=None, memcache=None): provided, the method checks if a special value for the setting exists with the given tag, and returns that if one exists. First matching tag wins. - :param MemcacheClient memcache: Memcache client. If `None`, a thread-local client will be used. :return list: List of tags """ # be flexible about the input types here @@ -466,19 +509,13 @@ def get_active_tags(self, user=None, tags=None, memcache=None): # that user's tags (including the 'special' user: tag) and add them # to the list if user: - user_tags = CacheMiss - - if not memcache: - memcache = self.get_memcache() - - if memcache: - memcache_id = f"_usertags-{user}" - user_tags = memcache.get(memcache_id, default=CacheMiss) + memcache_id = self._memcache_key(f"_usertags-{user}") + user_tags = self._memcache_command("get", memcache_id, default=CacheMiss, failed=CacheMiss) if user_tags is CacheMiss: user_tags = self.db.fetchone("SELECT tags FROM users WHERE name = %s", (user,)) - if user_tags and memcache: - memcache.set(memcache_id, user_tags) + if user_tags: + self._memcache_command("set", memcache_id, user_tags, expire=self.memcache_expire) if user_tags: try: @@ -491,7 +528,7 @@ def get_active_tags(self, user=None, tags=None, memcache=None): return tags - def set(self, attribute_name, value, is_json=False, tag="", overwrite_existing=True, memcache=None): + def set(self, attribute_name, value, is_json=False, tag="", overwrite_existing=True): """ Insert OR set value for a setting @@ -504,7 +541,6 @@ def set(self, attribute_name, value, is_json=False, tag="", overwrite_existing=T be serialised into a JSON string :param bool overwrite_existing: True will overwrite existing setting, False will do nothing if setting exists :param str tag: Tag to write setting for - :param MemcacheClient memcache: Memcache client. If `None`, a thread-local client will be used. :return int: number of updated rows """ @@ -532,13 +568,8 @@ def set(self, attribute_name, value, is_json=False, tag="", overwrite_existing=T updated_rows = self.db.cursor.rowcount self.db.log.debug(f"Updated setting for {attribute_name}: {value} (tag: {tag})") - if not memcache: - memcache = self.get_memcache() - - if memcache: - # invalidate any cached value for this setting - memcache_id = self._get_memcache_id(attribute_name, tag) - memcache.delete(memcache_id) + # invalidate any cached value for this setting + self._memcache_command("delete", self._get_memcache_id(attribute_name, tag), noreply=False) return updated_rows @@ -552,9 +583,7 @@ def delete_for_tag(self, attribute_name, tag): """ self.db.delete("settings", where={"name": attribute_name, "tag": tag}) updated_rows = self.db.cursor.rowcount - client = self.get_memcache() - if client: - client.delete(self._get_memcache_id(attribute_name, tag)) + self._memcache_command("delete", self._get_memcache_id(attribute_name, tag), noreply=False) return updated_rows def clear_cache(self): @@ -563,10 +592,7 @@ def clear_cache(self): Called when the backend restarts - helps start with a blank slate. """ - client = self.get_memcache() - if not client: - return - client.flush_all() + self._memcache_command("flush_all", noreply=False) def uncache_user_tags(self, users): """ @@ -578,11 +604,8 @@ def uncache_user_tags(self, users): :param list users: List of users, as usernames or User objects """ - client = self.get_memcache() - if client: - for user in users: - user = self._normalise_user(user) - client.delete(f"_usertags-{user}") + keys = [self._memcache_key(f"_usertags-{self._normalise_user(user)}") for user in users] + self._memcache_command("delete_many", keys, noreply=False) def _normalise_user(self, user): """ @@ -634,7 +657,25 @@ def _get_memcache_id(self, attribute_name, tags=None): if tag_bit: memcache_id += f"-{'-'.join(tag_bit)}" - return memcache_id.encode("ascii") + return self._memcache_key(memcache_id) + + @staticmethod + def _memcache_key(key): + """ + Make a key that memcache accepts + + Memcache keys must be short and cannot contain spaces or non-ASCII + characters (pymemcache). Keys contain user names, which can contain + anything, so we check and hash if necessary. + + :param str key: + :return str: + """ + # check key short and only plain English letters, digits and punctuation (no spaces) + if key.isascii() and key.isprintable() and " " not in key and len(key) <= 200: + return key + # or hash the key + return "hash-" + hashlib.sha256(key.encode()).hexdigest() def __getattr__(self, attr): """ diff --git a/tests/test_config_memcache.py b/tests/test_config_memcache.py new file mode 100644 index 000000000..27733c5c4 --- /dev/null +++ b/tests/test_config_memcache.py @@ -0,0 +1,227 @@ +""" +Tests for how the config manager uses memcache + +Memcache only makes reading settings faster. When it cannot be used, settings +are read from the database instead, and nothing may fail because of it. +""" +import json + +import pytest + +from unittest.mock import MagicMock, patch +from pymemcache.exceptions import MemcacheUnexpectedCloseError + +from common.config_manager import ConfigManager + + +class FakeMemcacheServer: + """ + Values held by a memcache server that can be restarted or stopped + """ + def __init__(self): + self.values = {} + self.expire = {} + self.up = True + self.started = 0 + self.connects = 0 + + def restart(self): + # a restart closes all connections and empties memcache + self.values = {} + self.started += 1 + + +class FakeMemcacheClient: + """ + Stands in for pymemcache's client + + Like pymemcache, it closes its connection after an error, and opens a new + one on the next command. + """ + def __init__(self, server): + self.server = server + self.sock = None + + def _run(self, command): + if self.sock is None: + self.server.connects += 1 + if not self.server.up: + raise ConnectionRefusedError("Connection refused") + self.sock = self.server.started + + if not self.server.up or self.sock != self.server.started: + self.sock = None + raise MemcacheUnexpectedCloseError() + + return command() + + def get(self, key, default=None): + return self._run(lambda: self.server.values.get(key, default)) + + def set(self, key, value, expire=0, noreply=None): + def store(): + self.server.values[key] = value + self.server.expire[key] = expire + return True + return self._run(store) + + def delete(self, key, noreply=None): + return self._run(lambda: self.server.values.pop(key, None) is not None) + + def delete_many(self, keys, noreply=None): + def remove(): + for key in keys: + self.server.values.pop(key, None) + return True + return self._run(remove) + + def flush_all(self, noreply=None): + return self._run(lambda: self.server.values.clear()) + + +class FakeDatabase: + """ + Stands in for the settings table + """ + def __init__(self): + self.settings = {} + self.cursor = MagicMock(rowcount=1) + self.log = MagicMock() + + def fetchall(self, query, replacements): + name, tags = replacements + return [{"tag": tag, "value": value} for (setting, tag), value in self.settings.items() + if setting == name and tag in tags] + + def execute(self, query, replacements): + name, value, tag = replacements + self.settings[(name, tag)] = value + + +@pytest.fixture +def server(): + return FakeMemcacheServer() + + +@pytest.fixture +def config(server): + """ + A config manager with a fake database and a fake memcache server + """ + instance = object.__new__(ConfigManager) + instance.core_settings = {"MEMCACHE_SERVER": "memcached:11211"} + instance.config_definition = {} + instance.db = FakeDatabase() + instance.logger = MagicMock() + + ConfigManager._memcache_down = False + ConfigManager._memcache_tls.__dict__.clear() + + with patch("common.config_manager.MemcacheClient", lambda *args, **kwargs: FakeMemcacheClient(server)): + yield instance + + ConfigManager._memcache_down = False + ConfigManager._memcache_tls.__dict__.clear() + + +def test_values_are_cached_with_an_expiry_time(config, server): + config.set("4cat.name", "Test 4CAT") + + assert config.get("4cat.name") == "Test 4CAT" + key = config._get_memcache_id("4cat.name", "") + assert server.values[key] == json.dumps("Test 4CAT") + assert server.expire[key] == ConfigManager.memcache_expire + + +def test_restart_is_not_noticed(config, server): + config.set("4cat.name", "Test 4CAT") + config.get("4cat.name") + + server.restart() + + assert config.get("4cat.name") == "Test 4CAT" + assert not ConfigManager._memcache_down + config.logger.warning.assert_not_called() + # cached again after the restart + assert config._get_memcache_id("4cat.name", "") in server.values + + +def test_changed_setting_is_removed_after_restart(config, server): + config.set("4cat.name", "Old name") + config.get("4cat.name") + + server.restart() + config.set("4cat.name", "New name") + + assert not ConfigManager._memcache_down + assert config.get("4cat.name") == "New name" + + +def test_outage_reads_from_database_until_restart(config, server): + config.set("4cat.name", "Test 4CAT") + config.get("4cat.name") + + server.up = False + assert config.get("4cat.name") == "Test 4CAT" + assert ConfigManager._memcache_down + assert config.logger.warning.call_count == 1 + + # memcache is not tried again, not even once it works again + connects = server.connects + server.up = True + assert config.get("4cat.name") == "Test 4CAT" + assert server.connects == connects + assert config.logger.warning.call_count == 1 + + +def test_unreachable_from_the_start(config, server): + server.up = False + + assert config.get("4cat.name", default="fallback") == "fallback" + assert ConfigManager._memcache_down + assert config.logger.warning.call_count == 1 + + +def test_change_while_down_leaves_old_value_cached(config, server): + """ + A setting changed while this process cannot reach memcache stays cached + with its old value, for processes that can still reach it, until the + value expires + """ + config.set("4cat.name", "Old name") + config.get("4cat.name") + key = config._get_memcache_id("4cat.name", "") + + server.up = False + config.set("4cat.name", "New name") + server.up = True + + assert config.get("4cat.name") == "New name" + assert server.values[key] == json.dumps("Old name") + assert server.expire[key] == ConfigManager.memcache_expire + + +def test_cache_clear_does_not_fail_while_down(config, server): + server.up = False + config.clear_cache() + config.uncache_user_tags(["someone@example.com"]) + + +def test_without_memcache_nothing_is_cached(config, server): + config.core_settings = {"MEMCACHE_SERVER": None} + config.set("4cat.name", "Test 4CAT") + + assert config.get("4cat.name") == "Test 4CAT" + assert server.connects == 0 + + +@pytest.mark.parametrize("name", ["jürgen@example.com", "first last@example.com", "x" * 300 + "@example.com"]) +def test_unusual_user_names_make_valid_keys(name): + key = ConfigManager._memcache_key(f"_usertags-{name}") + + assert key.startswith("hash-") + assert key.isascii() and key.isprintable() and " " not in key and len(key) <= 200 + + +def test_ordinary_keys_are_kept(): + assert ConfigManager._memcache_key("4cat.name-user:someone@example.com") == "4cat.name-user:someone@example.com" diff --git a/webtool/__init__.py b/webtool/__init__.py index f7cd24b8c..d0799197c 100644 --- a/webtool/__init__.py +++ b/webtool/__init__.py @@ -139,9 +139,11 @@ def time_this(func): app.login_manager.login_view = "user.show_login" # initialize rate limiter - memcache can serve as the storage backend, if not -# available, direct memory storage will be used +# available, direct memory storage will be used, also when memcache stops +# working later if config.get_memcache(): - app.limiter = Limiter(app=app, key_func=get_remote_address, storage_uri=f"memcached://{config.get('MEMCACHE_SERVER')}") + app.limiter = Limiter(app=app, key_func=get_remote_address, storage_uri=f"memcached://{config.get('MEMCACHE_SERVER')}", + in_memory_fallback_enabled=True, swallow_errors=True) else: app.limiter = Limiter(app=app, key_func=get_remote_address)