fix(caches): synced rc_cache module with rhodecode-ce

vcsserver/lib/rc_cache/__init__.py

@@ -1,114 +1,114 b''
 # RhodeCode VCSServer provides access to different vcs backends via network.
 # Copyright (C) 2014-2023 RhodeCode GmbH
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation; either version 3 of the License, or
 # (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU General Public License
 # along with this program; if not, write to the Free Software Foundation,
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
 
 import logging
 import threading
 
 from dogpile.cache import register_backend
 
 from . import region_meta
 from .utils import (
     backend_key_generator,
     clear_cache_namespace,
     get_default_cache_settings,
     get_or_create_region,
     make_region,
     str2bool,
 )
 
 module_name = 'vcsserver'
 
 register_backend(
     "dogpile.cache.rc.memory_lru", f"{module_name}.lib.rc_cache.backends",
     "LRUMemoryBackend")
 
 register_backend(
     "dogpile.cache.rc.file_namespace", f"{module_name}.lib.rc_cache.backends",
     "FileNamespaceBackend")
 
 register_backend(
     "dogpile.cache.rc.redis", f"{module_name}.lib.rc_cache.backends",
     "RedisPickleBackend")
 
 register_backend(
     "dogpile.cache.rc.redis_msgpack", f"{module_name}.lib.rc_cache.backends",
     "RedisMsgPackBackend")
 
 
 log = logging.getLogger(__name__)
 
 
 CACHE_OBJ_CACHE_VER = 'v2'
 
 CLEAR_DELETE = 'delete'
 CLEAR_INVALIDATE = 'invalidate'
 
 
-def async_creation_runner(cache, somekey, creator, mutex):
+def async_creation_runner(cache, cache_key, creator, mutex):
 
     def runner():
         try:
             value = creator()
-            cache.set(somekey, value)
+            cache.set(cache_key, value)
         finally:
             mutex.release()
 
     thread = threading.Thread(target=runner)
     thread.start()
 
 
 def configure_dogpile_cache(settings):
     cache_dir = settings.get('cache_dir')
     if cache_dir:
         region_meta.dogpile_config_defaults['cache_dir'] = cache_dir
 
     rc_cache_data = get_default_cache_settings(settings, prefixes=['rc_cache.'])
 
     # inspect available namespaces
     avail_regions = set()
     for key in rc_cache_data.keys():
         namespace_name = key.split('.', 1)[0]
         if namespace_name in avail_regions:
             continue
 
         avail_regions.add(namespace_name)
         log.debug('dogpile: found following cache regions: %s', namespace_name)
 
         new_region = make_region(
             name=namespace_name,
             function_key_generator=None,
             async_creation_runner=None
         )
 
         new_region.configure_from_config(settings, f'rc_cache.{namespace_name}.')
         new_region.function_key_generator = backend_key_generator(new_region.actual_backend)
 
         async_creator = str2bool(settings.pop(f'rc_cache.{namespace_name}.async_creator', 'false'))
         if async_creator:
             log.debug('configuring region %s with async creator', new_region)
             new_region.async_creation_runner = async_creation_runner
 
         if log.isEnabledFor(logging.DEBUG):
             region_args = dict(backend=new_region.actual_backend,
                                region_invalidator=new_region.region_invalidator.__class__)
             log.debug('dogpile: registering a new region key=`%s` args=%s', namespace_name, region_args)
 
         region_meta.dogpile_cache_regions[namespace_name] = new_region
 
 
 def includeme(config):
     configure_dogpile_cache(config.registry.settings)

vcsserver/lib/rc_cache/backends.py

@@ -1,267 +1,303 b''
 # RhodeCode VCSServer provides access to different vcs backends via network.
 # Copyright (C) 2014-2023 RhodeCode GmbH
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation; either version 3 of the License, or
 # (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU General Public License
 # along with this program; if not, write to the Free Software Foundation,
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
 
 #import errno
 import fcntl
 import functools
 import logging
 import os
 import pickle
 #import time
 
 #import gevent
 import msgpack
 import redis
 
 flock_org = fcntl.flock
 from typing import Union
 
 from dogpile.cache.api import Deserializer, Serializer
 from dogpile.cache.backends import file as file_backend
 from dogpile.cache.backends import memory as memory_backend
 from dogpile.cache.backends import redis as redis_backend
 from dogpile.cache.backends.file import FileLock
 from dogpile.cache.util import memoized_property
 
 from vcsserver.lib.memory_lru_dict import LRUDict, LRUDictDebug
 from vcsserver.str_utils import safe_bytes, safe_str
 from vcsserver.type_utils import str2bool
 
 _default_max_size = 1024
 
 log = logging.getLogger(__name__)
 
 
 class LRUMemoryBackend(memory_backend.MemoryBackend):
     key_prefix = 'lru_mem_backend'
     pickle_values = False
 
     def __init__(self, arguments):
         self.max_size = arguments.pop('max_size', _default_max_size)
 
         LRUDictClass = LRUDict
         if arguments.pop('log_key_count', None):
             LRUDictClass = LRUDictDebug
 
         arguments['cache_dict'] = LRUDictClass(self.max_size)
         super().__init__(arguments)
 
     def __repr__(self):
         return f'{self.__class__}(maxsize=`{self.max_size}`)'
 
     def __str__(self):
         return self.__repr__()
 
     def delete(self, key):
         try:
             del self._cache[key]
         except KeyError:
             # we don't care if key isn't there at deletion
             pass
 
+    def list_keys(self, prefix):
+        return list(self._cache.keys())
+
     def delete_multi(self, keys):
         for key in keys:
             self.delete(key)
 
+    def delete_multi_by_prefix(self, prefix):
+        cache_keys = self.list_keys(prefix=prefix)
+        num_affected_keys = len(cache_keys)
+        if num_affected_keys:
+            self.delete_multi(cache_keys)
+        return num_affected_keys
+
 
 class PickleSerializer:
     serializer: None | Serializer = staticmethod(  # type: ignore
         functools.partial(pickle.dumps, protocol=pickle.HIGHEST_PROTOCOL)
     )
     deserializer: None | Deserializer = staticmethod(  # type: ignore
         functools.partial(pickle.loads)
     )
 
 
 class MsgPackSerializer:
     serializer: None | Serializer = staticmethod(  # type: ignore
         msgpack.packb
     )
     deserializer: None | Deserializer = staticmethod(  # type: ignore
         functools.partial(msgpack.unpackb, use_list=False)
     )
 
 
 class CustomLockFactory(FileLock):
 
     pass
 
 
 class FileNamespaceBackend(PickleSerializer, file_backend.DBMBackend):
     key_prefix = 'file_backend'
 
     def __init__(self, arguments):
         arguments['lock_factory'] = CustomLockFactory
         db_file = arguments.get('filename')
 
         log.debug('initialing cache-backend=%s db in %s', self.__class__.__name__, db_file)
         db_file_dir = os.path.dirname(db_file)
         if not os.path.isdir(db_file_dir):
             os.makedirs(db_file_dir)
 
         try:
             super().__init__(arguments)
         except Exception:
             log.exception('Failed to initialize db at: %s', db_file)
             raise
 
     def __repr__(self):
         return f'{self.__class__}(file=`{self.filename}`)'
 
     def __str__(self):
         return self.__repr__()
 
     def _get_keys_pattern(self, prefix: bytes = b''):
         return b'%b:%b' % (safe_bytes(self.key_prefix), safe_bytes(prefix))
 
     def list_keys(self, prefix: bytes = b''):
         prefix = self._get_keys_pattern(prefix)
 
         def cond(dbm_key: bytes):
             if not prefix:
                 return True
 
             if dbm_key.startswith(prefix):
                 return True
             return False
 
         with self._dbm_file(True) as dbm:
             try:
                 return list(filter(cond, dbm.keys()))
             except Exception:
                 log.error('Failed to fetch DBM keys from DB: %s', self.get_store())
                 raise
 
+    def delete_multi_by_prefix(self, prefix):
+        cache_keys = self.list_keys(prefix=prefix)
+        num_affected_keys = len(cache_keys)
+        if num_affected_keys:
+            self.delete_multi(cache_keys)
+        return num_affected_keys
+
     def get_store(self):
         return self.filename
 
 
 class BaseRedisBackend(redis_backend.RedisBackend):
     key_prefix = ''
 
     def __init__(self, arguments):
         self.db_conn = arguments.get('host', '') or arguments.get('url', '') or 'redis-host'
         super().__init__(arguments)
 
         self._lock_timeout = self.lock_timeout
         self._lock_auto_renewal = str2bool(arguments.pop("lock_auto_renewal", True))
 
         if self._lock_auto_renewal and not self._lock_timeout:
             # set default timeout for auto_renewal
             self._lock_timeout = 30
 
     def __repr__(self):
         return f'{self.__class__}(conn=`{self.db_conn}`)'
 
     def __str__(self):
         return self.__repr__()
 
     def _create_client(self):
         args = {}
 
         if self.url is not None:
             args.update(url=self.url)
 
         else:
             args.update(
                 host=self.host, password=self.password,
                 port=self.port, db=self.db
             )
 
         connection_pool = redis.ConnectionPool(**args)
         self.writer_client = redis.StrictRedis(
             connection_pool=connection_pool
         )
         self.reader_client = self.writer_client
 
     def _get_keys_pattern(self, prefix: bytes = b''):
         return b'%b:%b*' % (safe_bytes(self.key_prefix), safe_bytes(prefix))
 
     def list_keys(self, prefix: bytes = b''):
         prefix = self._get_keys_pattern(prefix)
         return self.reader_client.keys(prefix)
 
+    def delete_multi_by_prefix(self, prefix, use_lua=False):
+        if use_lua:
+            # high efficient LUA script to delete ALL keys by prefix...
+            lua = """local keys = redis.call('keys', ARGV[1])
+                     for i=1,#keys,5000 do
+                     redis.call('del', unpack(keys, i, math.min(i+(5000-1), #keys)))
+                     end
+                     return #keys"""
+            num_affected_keys = self.writer_client.eval(
+                lua,
+                0,
+                f"{prefix}*")
+        else:
+            cache_keys = self.list_keys(prefix=prefix)
+            num_affected_keys = len(cache_keys)
+            if num_affected_keys:
+                self.delete_multi(cache_keys)
+        return num_affected_keys
+
     def get_store(self):
         return self.reader_client.connection_pool
 
     def get_mutex(self, key):
         if self.distributed_lock:
             lock_key = f'_lock_{safe_str(key)}'
             return get_mutex_lock(
                 self.writer_client, lock_key,
                 self._lock_timeout,
                 auto_renewal=self._lock_auto_renewal
             )
         else:
             return None
 
 
 class RedisPickleBackend(PickleSerializer, BaseRedisBackend):
     key_prefix = 'redis_pickle_backend'
     pass
 
 
 class RedisMsgPackBackend(MsgPackSerializer, BaseRedisBackend):
     key_prefix = 'redis_msgpack_backend'
     pass
 
 
 def get_mutex_lock(client, lock_key, lock_timeout, auto_renewal=False):
     from vcsserver.lib._vendor import redis_lock
 
     class _RedisLockWrapper:
         """LockWrapper for redis_lock"""
 
         @classmethod
         def get_lock(cls):
             return redis_lock.Lock(
                 redis_client=client,
                 name=lock_key,
                 expire=lock_timeout,
                 auto_renewal=auto_renewal,
                 strict=True,
             )
 
         def __repr__(self):
             return f"{self.__class__.__name__}:{lock_key}"
 
         def __str__(self):
             return f"{self.__class__.__name__}:{lock_key}"
 
         def __init__(self):
             self.lock = self.get_lock()
             self.lock_key = lock_key
 
         def acquire(self, wait=True):
             log.debug('Trying to acquire Redis lock for key %s', self.lock_key)
             try:
                 acquired = self.lock.acquire(wait)
                 log.debug('Got lock for key %s, %s', self.lock_key, acquired)
                 return acquired
             except redis_lock.AlreadyAcquired:
                 return False
             except redis_lock.AlreadyStarted:
                 # refresh thread exists, but it also means we acquired the lock
                 return True
 
         def release(self):
             try:
                 self.lock.release()
             except redis_lock.NotAcquired:
                 pass
 
     return _RedisLockWrapper()

vcsserver/lib/rc_cache/utils.py

@@ -1,248 +1,245 b''
 # RhodeCode VCSServer provides access to different vcs backends via network.
 # Copyright (C) 2014-2023 RhodeCode GmbH
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation; either version 3 of the License, or
 # (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU General Public License
 # along with this program; if not, write to the Free Software Foundation,
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
 
 import functools
 import logging
 import os
 import threading
 import time
 
 import decorator
 from dogpile.cache import CacheRegion
 
 
 from vcsserver.utils import sha1
 from vcsserver.str_utils import safe_bytes
 from vcsserver.type_utils import str2bool # noqa :required by imports from .utils
 
 from . import region_meta
 
 log = logging.getLogger(__name__)
 
 
 class RhodeCodeCacheRegion(CacheRegion):
 
     def __repr__(self):
-        return f'{self.__class__}(name={self.name})'
+        return f'`{self.__class__.__name__}(name={self.name}, backend={self.backend.__class__})`'
 
     def conditional_cache_on_arguments(
             self, namespace=None,
             expiration_time=None,
             should_cache_fn=None,
             to_str=str,
             function_key_generator=None,
             condition=True):
         """
         Custom conditional decorator, that will not touch any dogpile internals if
         condition isn't meet. This works a bit different from should_cache_fn
         And it's faster in cases we don't ever want to compute cached values
         """
         expiration_time_is_callable = callable(expiration_time)
         if not namespace:
             namespace = getattr(self, '_default_namespace', None)
 
         if function_key_generator is None:
             function_key_generator = self.function_key_generator
 
         def get_or_create_for_user_func(func_key_generator, user_func, *arg, **kw):
 
             if not condition:
                 log.debug('Calling un-cached method:%s', user_func.__name__)
                 start = time.time()
                 result = user_func(*arg, **kw)
                 total = time.time() - start
                 log.debug('un-cached method:%s took %.4fs', user_func.__name__, total)
                 return result
 
             key = func_key_generator(*arg, **kw)
 
             timeout = expiration_time() if expiration_time_is_callable \
                 else expiration_time
 
             log.debug('Calling cached method:`%s`', user_func.__name__)
             return self.get_or_create(key, user_func, timeout, should_cache_fn, (arg, kw))
 
         def cache_decorator(user_func):
             if to_str is str:
                 # backwards compatible
                 key_generator = function_key_generator(namespace, user_func)
             else:
                 key_generator = function_key_generator(namespace, user_func, to_str=to_str)
 
             def refresh(*arg, **kw):
                 """
                 Like invalidate, but regenerates the value instead
                 """
                 key = key_generator(*arg, **kw)
                 value = user_func(*arg, **kw)
                 self.set(key, value)
                 return value
 
             def invalidate(*arg, **kw):
                 key = key_generator(*arg, **kw)
                 self.delete(key)
 
             def set_(value, *arg, **kw):
                 key = key_generator(*arg, **kw)
                 self.set(key, value)
 
             def get(*arg, **kw):
                 key = key_generator(*arg, **kw)
                 return self.get(key)
 
             user_func.set = set_
             user_func.invalidate = invalidate
             user_func.get = get
             user_func.refresh = refresh
             user_func.key_generator = key_generator
             user_func.original = user_func
 
             # Use `decorate` to preserve the signature of :param:`user_func`.
             return decorator.decorate(user_func, functools.partial(
                 get_or_create_for_user_func, key_generator))
 
         return cache_decorator
 
 
 def make_region(*arg, **kw):
     return RhodeCodeCacheRegion(*arg, **kw)
 
 
 def get_default_cache_settings(settings, prefixes=None):
     prefixes = prefixes or []
     cache_settings = {}
     for key in settings.keys():
         for prefix in prefixes:
             if key.startswith(prefix):
                 name = key.split(prefix)[1].strip()
                 val = settings[key]
                 if isinstance(val, str):
                     val = val.strip()
                 cache_settings[name] = val
     return cache_settings
 
 
 def compute_key_from_params(*args):
     """
     Helper to compute key from given params to be used in cache manager
     """
     return sha1(safe_bytes("_".join(map(str, args))))
 
 
 def custom_key_generator(backend, namespace, fn):
     func_name = fn.__name__
 
     def generate_key(*args):
         backend_pref = getattr(backend, 'key_prefix', None) or 'backend_prefix'
         namespace_pref = namespace or 'default_namespace'
         arg_key = compute_key_from_params(*args)
         final_key = f"{backend_pref}:{namespace_pref}:{func_name}_{arg_key}"
 
         return final_key
 
     return generate_key
 
 
 def backend_key_generator(backend):
     """
     Special wrapper that also sends over the backend to the key generator
     """
     def wrapper(namespace, fn):
         return custom_key_generator(backend, namespace, fn)
     return wrapper
 
 
 def get_or_create_region(region_name, region_namespace: str = None, use_async_runner=False):
     from .backends import FileNamespaceBackend
     from . import async_creation_runner
 
     region_obj = region_meta.dogpile_cache_regions.get(region_name)
     if not region_obj:
         reg_keys = list(region_meta.dogpile_cache_regions.keys())
         raise OSError(f'Region `{region_name}` not in configured: {reg_keys}.')
 
     region_uid_name = f'{region_name}:{region_namespace}'
 
     # Special case for ONLY the FileNamespaceBackend backend. We register one-file-per-region
     if isinstance(region_obj.actual_backend, FileNamespaceBackend):
         if not region_namespace:
             raise ValueError(f'{FileNamespaceBackend} used requires to specify region_namespace param')
 
         region_exist = region_meta.dogpile_cache_regions.get(region_namespace)
         if region_exist:
             log.debug('Using already configured region: %s', region_namespace)
             return region_exist
 
         expiration_time = region_obj.expiration_time
 
         cache_dir = region_meta.dogpile_config_defaults['cache_dir']
         namespace_cache_dir = cache_dir
 
         # we default the namespace_cache_dir to our default cache dir.
         # however, if this backend is configured with filename= param, we prioritize that
         # so all caches within that particular region, even those namespaced end up in the same path
         if region_obj.actual_backend.filename:
             namespace_cache_dir = os.path.dirname(region_obj.actual_backend.filename)
 
         if not os.path.isdir(namespace_cache_dir):
             os.makedirs(namespace_cache_dir)
         new_region = make_region(
             name=region_uid_name,
             function_key_generator=backend_key_generator(region_obj.actual_backend)
         )
 
         namespace_filename = os.path.join(
             namespace_cache_dir, f"{region_name}_{region_namespace}.cache_db")
         # special type that allows 1db per namespace
         new_region.configure(
             backend='dogpile.cache.rc.file_namespace',
             expiration_time=expiration_time,
             arguments={"filename": namespace_filename}
         )
 
         # create and save in region caches
         log.debug('configuring new region: %s', region_uid_name)
         region_obj = region_meta.dogpile_cache_regions[region_namespace] = new_region
 
     region_obj._default_namespace = region_namespace
     if use_async_runner:
         region_obj.async_creation_runner = async_creation_runner
     return region_obj
 
 
-def clear_cache_namespace(cache_region: str | RhodeCodeCacheRegion, cache_namespace_uid: str, method: str):
+def clear_cache_namespace(cache_region: str | RhodeCodeCacheRegion, cache_namespace_uid: str, method: str) -> int:
     from . import CLEAR_DELETE, CLEAR_INVALIDATE
 
     if not isinstance(cache_region, RhodeCodeCacheRegion):
         cache_region = get_or_create_region(cache_region, cache_namespace_uid)
-    log.debug('clearing cache region: %s with method=%s', cache_region, method)
+    log.debug('clearing cache region: %s [prefix:%s] with method=%s',
+              cache_region, cache_namespace_uid, method)
 
-    num_affected_keys = None
+    num_affected_keys = 0
 
     if method == CLEAR_INVALIDATE:
         # NOTE: The CacheRegion.invalidate() method’s default mode of
         # operation is to set a timestamp local to this CacheRegion in this Python process only.
         # It does not impact other Python processes or regions as the timestamp is only stored locally in memory.
         cache_region.invalidate(hard=True)
 
     if method == CLEAR_DELETE:
-        cache_keys = cache_region.backend.list_keys(prefix=cache_namespace_uid)
-        num_affected_keys = len(cache_keys)
-        if num_affected_keys:
-            cache_region.delete_multi(cache_keys)
-
+        num_affected_keys = cache_region.backend.delete_multi_by_prefix(prefix=cache_namespace_uid)
     return num_affected_keys