feat(disk-cache): rewrite diskcache backend to be k8s and NFS safe.

We no longer use diskcache lib, but instead a homebrew solution that uses redis lock and simple filesystem storage

vcsserver/lib/rc_cache/archive_cache/__init__.py

@@ -0,0 +1,31 b''
+# RhodeCode VCSServer provides access to different vcs backends via network.
+# Copyright (C) 2014-2024 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
+
+from .fanout_cache import get_archival_cache_store
+from .fanout_cache import get_archival_config
+
+from .utils import archive_iterator
+from .utils import ArchiveCacheLock
+
+
+def includeme(config):
+    # NOTE: for vcsserver, we lazy init this and config is sent from RhodeCode
+    return
+
+    # init our cache at start
+    settings = config.get_settings()
+    get_archival_cache_store(settings)

vcsserver/lib/rc_cache/archive_cache/fanout_cache.py

@@ -0,0 +1,258 b''
+# RhodeCode VCSServer provides access to different vcs backends via network.
+# Copyright (C) 2014-2024 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 codecs
+import contextlib
+import functools
+import os
+import logging
+import time
+import typing
+import zlib
+
+from vcsserver.lib.rc_json import json
+from .lock import GenerationLock
+
+log = logging.getLogger(__name__)
+
+cache_meta = None
+
+UNKNOWN = -241
+NO_VAL = -917
+
+MODE_BINARY = 'BINARY'
+
+
+class FileSystemCache:
+
+    def __init__(self, index, directory, **settings):
+        self._index = index
+        self._directory = directory
+
+    def _write_file(self, full_path, iterator, mode, encoding=None):
+        full_dir, _ = os.path.split(full_path)
+
+        for count in range(1, 11):
+            with contextlib.suppress(OSError):
+                os.makedirs(full_dir)
+
+            try:
+                # Another cache may have deleted the directory before
+                # the file could be opened.
+                writer = open(full_path, mode, encoding=encoding)
+            except OSError:
+                if count == 10:
+                    # Give up after 10 tries to open the file.
+                    raise
+                continue
+
+            with writer:
+                size = 0
+                for chunk in iterator:
+                    size += len(chunk)
+                    writer.write(chunk)
+                return size
+
+    def _get_keyfile(self, key):
+        return os.path.join(self._directory, f'{key}.key')
+
+    def store(self, key, value_reader, metadata):
+        filename, full_path = self.random_filename()
+        key_file = self._get_keyfile(key)
+
+        # STORE METADATA
+        _metadata = {
+            "version": "v1",
+            "timestamp": time.time(),
+            "filename": filename,
+            "full_path": full_path,
+            "key_file": key_file,
+        }
+        if metadata:
+            _metadata.update(metadata)
+
+        reader = functools.partial(value_reader.read, 2**22)
+
+        iterator = iter(reader, b'')
+        size = self._write_file(full_path, iterator, 'xb')
+
+        # after archive is finished, we create a key to save the presence of the binary file
+        with open(key_file, 'wb') as f:
+            f.write(json.dumps(_metadata))
+
+        return key, size, MODE_BINARY, filename, _metadata
+
+    def fetch(self, key) -> tuple[typing.BinaryIO, dict]:
+        if key not in self:
+            raise KeyError(key)
+
+        key_file = self._get_keyfile(key)
+        with open(key_file, 'rb') as f:
+            metadata = json.loads(f.read())
+
+        filename = metadata['filename']
+
+        return open(os.path.join(self._directory, filename), 'rb'), metadata
+
+    def random_filename(self):
+        """Return filename and full-path tuple for file storage.
+
+        Filename will be a randomly generated 28 character hexadecimal string
+        with ".archive_cache" suffixed. Two levels of sub-directories will be used to
+        reduce the size of directories. On older filesystems, lookups in
+        directories with many files may be slow.
+        """
+
+        hex_name = codecs.encode(os.urandom(16), 'hex').decode('utf-8')
+        sub_dir = os.path.join(hex_name[:2], hex_name[2:4])
+        name = hex_name[4:] + '.archive_cache'
+        filename = os.path.join(sub_dir, name)
+        full_path = os.path.join(self._directory, filename)
+        return filename, full_path
+
+    def hash(self, key):
+        """Compute portable hash for `key`.
+
+        :param key: key to hash
+        :return: hash value
+
+        """
+        mask = 0xFFFFFFFF
+        return zlib.adler32(key.encode('utf-8')) & mask  # noqa
+
+    def __contains__(self, key):
+        """Return `True` if `key` matching item is found in cache.
+
+        :param key: key matching item
+        :return: True if key matching item
+
+        """
+        key_file = self._get_keyfile(key)
+        return os.path.exists(key_file)
+
+
+class FanoutCache:
+    """Cache that shards keys and values."""
+
+    def __init__(
+        self, directory=None, **settings
+    ):
+        """Initialize cache instance.
+
+        :param str directory: cache directory
+        :param settings: settings dict
+
+        """
+        if directory is None:
+            raise ValueError('directory cannot be None')
+
+        directory = str(directory)
+        directory = os.path.expanduser(directory)
+        directory = os.path.expandvars(directory)
+        self._directory = directory
+
+        self._count = settings.pop('cache_shards')
+        self._locking_url = settings.pop('locking_url')
+
+        self._shards = tuple(
+            FileSystemCache(
+                index=num,
+                directory=os.path.join(directory, 'shard_%03d' % num),
+                **settings,
+            )
+            for num in range(self._count)
+        )
+        self._hash = self._shards[0].hash
+
+    def get_lock(self, lock_key):
+        return GenerationLock(lock_key, self._locking_url)
+
+    def _get_shard(self, key) -> FileSystemCache:
+        index = self._hash(key) % self._count
+        shard = self._shards[index]
+        return shard
+
+    def store(self, key, value_reader, metadata=None):
+        shard = self._get_shard(key)
+        return shard.store(key, value_reader, metadata)
+
+    def fetch(self, key):
+        """Return file handle corresponding to `key` from cache.
+        """
+        shard = self._get_shard(key)
+        return shard.fetch(key)
+
+    def has_key(self, key):
+        """Return `True` if `key` matching item is found in cache.
+
+        :param key: key for item
+        :return: True if key is found
+
+        """
+        shard = self._get_shard(key)
+        return key in shard
+
+    def __contains__(self, item):
+        return self.has_key(item)
+
+
+def get_archival_config(config):
+
+    final_config = {
+
+    }
+
+    for k, v in config.items():
+        if k.startswith('archive_cache'):
+            final_config[k] = v
+
+    return final_config
+
+
+def get_archival_cache_store(config):
+
+    global cache_meta
+    if cache_meta is not None:
+        return cache_meta
+
+    config = get_archival_config(config)
+    backend = config['archive_cache.backend.type']
+    if backend != 'filesystem':
+        raise ValueError('archive_cache.backend.type only supports "filesystem"')
+
+    archive_cache_locking_url = config['archive_cache.locking.url']
+    archive_cache_dir = config['archive_cache.filesystem.store_dir']
+    archive_cache_size_gb = config['archive_cache.filesystem.cache_size_gb']
+    archive_cache_shards = config['archive_cache.filesystem.cache_shards']
+    archive_cache_eviction_policy = config['archive_cache.filesystem.eviction_policy']
+
+    log.debug('Initializing archival cache instance under %s', archive_cache_dir)
+
+    # check if it's ok to write, and re-create the archive cache
+    if not os.path.isdir(archive_cache_dir):
+        os.makedirs(archive_cache_dir, exist_ok=True)
+
+    d_cache = FanoutCache(
+        archive_cache_dir,
+        locking_url=archive_cache_locking_url,
+        cache_shards=archive_cache_shards,
+        cache_size_limit=archive_cache_size_gb * 1024 * 1024 * 1024,
+        cache_eviction_policy=archive_cache_eviction_policy
+    )
+    cache_meta = d_cache
+    return cache_meta
+

vcsserver/lib/rc_cache/archive_cache/lock.py

@@ -0,0 +1,59 b''
+# RhodeCode VCSServer provides access to different vcs backends via network.
+# Copyright (C) 2014-2024 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 redis
+from vcsserver.lib._vendor import redis_lock
+
+from .utils import ArchiveCacheLock
+
+
+class GenerationLock:
+    """
+    Locking mechanism that detects if a lock is acquired
+
+    with GenerationLock(lock_key):
+        compute_archive()
+    """
+    lock_timeout = 7200
+
+    def __init__(self, lock_key, url):
+        self.lock_key = lock_key
+        self._create_client(url)
+        self.lock = self.get_lock()
+
+    def _create_client(self, url):
+        connection_pool = redis.ConnectionPool.from_url(url)
+        self.writer_client = redis.StrictRedis(
+            connection_pool=connection_pool
+        )
+        self.reader_client = self.writer_client
+
+    def get_lock(self):
+        return redis_lock.Lock(
+            redis_client=self.writer_client,
+            name=self.lock_key,
+            expire=self.lock_timeout,
+            strict=True
+        )
+
+    def __enter__(self):
+        acquired = self.lock.acquire(blocking=False)
+        if not acquired:
+            raise ArchiveCacheLock('Failed to create a lock')
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.lock.release()

vcsserver/lib/rc_cache/archive_cache/utils.py

@@ -0,0 +1,29 b''
+# RhodeCode VCSServer provides access to different vcs backends via network.
+# Copyright (C) 2014-2024 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
+
+
+class ArchiveCacheLock(Exception):
+    pass
+
+
+def archive_iterator(_reader, block_size: int = 4096 * 512):
+    # 4096 * 64 = 64KB
+    while 1:
+        data = _reader.read(block_size)
+        if not data:
+            break
+        yield data

requirements.txt

@@ -1,74 +1,73 b''
 # deps, generated via pipdeptree --exclude setuptools,wheel,pipdeptree,pip -f | tr '[:upper:]' '[:lower:]'
 
 async-timeout==4.0.3
 atomicwrites==1.4.1
 celery==5.3.6
   billiard==4.2.0
   click==8.1.3
   click-didyoumean==0.3.0
     click==8.1.3
   click-plugins==1.1.1
     click==8.1.3
   click-repl==0.2.0
     click==8.1.3
     prompt-toolkit==3.0.38
       wcwidth==0.2.6
     six==1.16.0
   kombu==5.3.5
     amqp==5.2.0
       vine==5.1.0
     vine==5.1.0
   python-dateutil==2.8.2
     six==1.16.0
   tzdata==2024.1
   vine==5.1.0
 contextlib2==21.6.0
-diskcache==5.6.3
 dogpile.cache==1.3.3
   decorator==5.1.1
   stevedore==5.1.0
     pbr==5.11.1
 dulwich==0.21.6
   urllib3==1.26.14
 gunicorn==21.2.0
   packaging==24.0
 hg-evolve==11.0.2
 importlib-metadata==6.0.0
   zipp==3.15.0
 mercurial==6.3.3
 more-itertools==9.1.0
 msgpack==1.0.8
 orjson==3.10.3
 psutil==5.9.8
 py==1.11.0
 pygit2==1.13.3
   cffi==1.16.0
     pycparser==2.21
 pygments==2.15.1
 pyparsing==3.1.1
 pyramid==2.0.2
   hupper==1.12
   plaster==1.1.2
   plaster-pastedeploy==1.0.1
     pastedeploy==3.1.0
     plaster==1.1.2
   translationstring==1.4
   venusian==3.0.0
   webob==1.8.7
   zope.deprecation==5.0.0
   zope.interface==6.3.0
 redis==5.0.4
   async-timeout==4.0.3
 repoze.lru==0.7
 scandir==1.10.0
 setproctitle==1.3.3
 subvertpy==0.11.0
 waitress==3.0.0
 wcwidth==0.2.6
 
 
 ## test related requirements
 #-r requirements_test.txt
 
 ## uncomment to add the debug libraries
 #-r requirements_debug.txt

vcsserver/base.py

@@ -1,193 +1,187 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 os
 import sys
 import tempfile
 import logging
 import urllib.parse
 
 from vcsserver.lib.rc_cache.archive_cache import get_archival_cache_store
 
 from vcsserver import exceptions
 from vcsserver.exceptions import NoContentException
 from vcsserver.hgcompat import archival
 from vcsserver.str_utils import safe_bytes
 from vcsserver.lib.exc_tracking import format_exc
 log = logging.getLogger(__name__)
 
 
 class RepoFactory:
     """
     Utility to create instances of repository
 
     It provides internal caching of the `repo` object based on
     the :term:`call context`.
     """
     repo_type = None
 
     def __init__(self):
         pass
 
     def _create_config(self, path, config):
         config = {}
         return config
 
     def _create_repo(self, wire, create):
         raise NotImplementedError()
 
     def repo(self, wire, create=False):
         raise NotImplementedError()
 
 
 def obfuscate_qs(query_string):
     if query_string is None:
         return None
 
     parsed = []
     for k, v in urllib.parse.parse_qsl(query_string, keep_blank_values=True):
         if k in ['auth_token', 'api_key']:
             v = "*****"
         parsed.append((k, v))
 
     return '&'.join('{}{}'.format(
         k, f'={v}' if v else '') for k, v in parsed)
 
 
 def raise_from_original(new_type, org_exc: Exception):
     """
     Raise a new exception type with original args and traceback.
     """
     exc_info = sys.exc_info()
     exc_type, exc_value, exc_traceback = exc_info
     new_exc = new_type(*exc_value.args)
 
     # store the original traceback into the new exc
     new_exc._org_exc_tb = format_exc(exc_info)
 
     try:
         raise new_exc.with_traceback(exc_traceback)
     finally:
         del exc_traceback
 
 
 class ArchiveNode:
     def __init__(self, path, mode, is_link, raw_bytes):
         self.path = path
         self.mode = mode
         self.is_link = is_link
         self.raw_bytes = raw_bytes
 
 
 def store_archive_in_cache(node_walker, archive_key, kind, mtime, archive_at_path, archive_dir_name,
                            commit_id, write_metadata=True, extra_metadata=None, cache_config=None):
     """
     Function that would store generate archive and send it to a dedicated backend store
     In here we use diskcache
 
     :param node_walker: a generator returning nodes to add to archive
     :param archive_key: key used to store the path
     :param kind: archive kind
     :param mtime: time of creation
     :param archive_at_path: default '/' the path at archive was started.
         If this is not '/' it means it's a partial archive
     :param archive_dir_name: inside dir name when creating an archive
     :param commit_id: commit sha of revision archive was created at
     :param write_metadata:
     :param extra_metadata:
     :param cache_config:
 
     walker should be a file walker, for example,
         def node_walker():
             for file_info in files:
                 yield ArchiveNode(fn, mode, is_link, ctx[fn].data)
     """
     extra_metadata = extra_metadata or {}
 
     d_cache = get_archival_cache_store(config=cache_config)
 
     if archive_key in d_cache:
-        with d_cache as d_cache_reader:
-            reader, tag = d_cache_reader.get(archive_key, read=True, tag=True, retry=True)
+        reader, metadata = d_cache.fetch(archive_key)
             return reader.name
 
     archive_tmp_path = safe_bytes(tempfile.mkstemp()[1])
     log.debug('Creating new temp archive in %s', archive_tmp_path)
 
     if kind == "tgz":
         archiver = archival.tarit(archive_tmp_path, mtime, b"gz")
     elif kind == "tbz2":
         archiver = archival.tarit(archive_tmp_path, mtime, b"bz2")
     elif kind == 'zip':
         archiver = archival.zipit(archive_tmp_path, mtime)
     else:
         raise exceptions.ArchiveException()(
             f'Remote does not support: "{kind}" archive type.')
 
     for f in node_walker(commit_id, archive_at_path):
         f_path = os.path.join(safe_bytes(archive_dir_name), safe_bytes(f.path).lstrip(b'/'))
+
         try:
             archiver.addfile(f_path, f.mode, f.is_link, f.raw_bytes())
         except NoContentException:
             # NOTE(marcink): this is a special case for SVN so we can create "empty"
             # directories which are not supported by archiver
             archiver.addfile(os.path.join(f_path, b'.dir'), f.mode, f.is_link, b'')
 
-    if write_metadata:
         metadata = dict([
             ('commit_id', commit_id),
             ('mtime', mtime),
         ])
         metadata.update(extra_metadata)
-
+    if write_metadata:
         meta = [safe_bytes(f"{f_name}:{value}") for f_name, value in metadata.items()]
         f_path = os.path.join(safe_bytes(archive_dir_name), b'.archival.txt')
         archiver.addfile(f_path, 0o644, False, b'\n'.join(meta))
 
     archiver.done()
 
-    # ensure set & get are atomic
-    with d_cache.transact():
-
         with open(archive_tmp_path, 'rb') as archive_file:
-            add_result = d_cache.set(archive_key, archive_file, read=True, tag='db-name', retry=True)
+        add_result = d_cache.store(archive_key, archive_file, metadata=metadata)
             if not add_result:
                 log.error('Failed to store cache for key=%s', archive_key)
 
         os.remove(archive_tmp_path)
 
-        reader, tag = d_cache.get(archive_key, read=True, tag=True, retry=True)
-        if not reader:
-            raise AssertionError(f'empty reader on key={archive_key} added={add_result}')
+    reader, metadata = d_cache.fetch(archive_key)
 
         return reader.name
 
 
 class BinaryEnvelope:
     def __init__(self, val):
         self.val = val
 
 
 class BytesEnvelope(bytes):
     def __new__(cls, content):
         if isinstance(content, bytes):
             return super().__new__(cls, content)
         else:
             raise TypeError('BytesEnvelope content= param must be bytes. Use BinaryEnvelope to wrap other types')
 
 
 class BinaryBytesEnvelope(BytesEnvelope):
     pass