feat(archive-cache): synced with CE codebase

vcsserver/lib/archive_cache/__init__.py

@@ -0,0 +1,79 b''
+# Copyright (C) 2015-2024 RhodeCode GmbH
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License, version 3
+# (only), as published by the Free Software Foundation.
+#
+# 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 Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+# This program is dual-licensed. If you wish to learn more about the
+# RhodeCode Enterprise Edition, including its added features, Support services,
+# and proprietary license terms, please see https://rhodecode.com/licenses/
+
+import logging
+
+from .backends.fanout_cache import FileSystemFanoutCache
+from .backends.objectstore_cache import ObjectStoreCache
+
+from .utils import archive_iterator  # noqa
+from .lock import ArchiveCacheGenerationLock # noqa
+
+log = logging.getLogger(__name__)
+
+
+cache_meta = None
+
+
+def includeme(config):
+    return  # vcsserver gets its config from rhodecode on a remote call
+    # init our cache at start
+    settings = config.get_settings()
+    get_archival_cache_store(settings)
+
+
+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, always_init=False):
+
+    global cache_meta
+    if cache_meta is not None and not always_init:
+        return cache_meta
+
+    config = get_archival_config(config)
+    backend = config['archive_cache.backend.type']
+
+    archive_cache_locking_url = config['archive_cache.locking.url']
+
+    match backend:
+        case 'filesystem':
+            d_cache = FileSystemFanoutCache(
+                locking_url=archive_cache_locking_url,
+                **config
+            )
+        case 'objectstore':
+            d_cache = ObjectStoreCache(
+                locking_url=archive_cache_locking_url,
+                **config
+            )
+        case _:
+            raise ValueError(f'archive_cache.backend.type only supports "filesystem" or "objectstore" got {backend} ')
+
+    cache_meta = d_cache
+    return cache_meta

vcsserver/lib/archive_cache/backends/__init__.py

@@ -0,0 +1,17 b''
+# Copyright (C) 2015-2024 RhodeCode GmbH
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License, version 3
+# (only), as published by the Free Software Foundation.
+#
+# 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 Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+# This program is dual-licensed. If you wish to learn more about the
+# RhodeCode Enterprise Edition, including its added features, Support services,
+# and proprietary license terms, please see https://rhodecode.com/licenses/

vcsserver/lib/archive_cache/backends/base.py

@@ -0,0 +1,348 b''
+# Copyright (C) 2015-2024 RhodeCode GmbH
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License, version 3
+# (only), as published by the Free Software Foundation.
+#
+# 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 Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+# This program is dual-licensed. If you wish to learn more about the
+# RhodeCode Enterprise Edition, including its added features, Support services,
+# and proprietary license terms, please see https://rhodecode.com/licenses/
+
+import os
+import functools
+import logging
+import typing
+import time
+import zlib
+
+from ...ext_json import json
+from ..utils import StatsDB, NOT_GIVEN, ShardFileReader, EVICTION_POLICY, format_size
+from ..lock import GenerationLock
+
+log = logging.getLogger(__name__)
+
+
+class BaseShard:
+    storage_type: str = ''
+    fs = None
+
+    @classmethod
+    def hash(cls, 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 _write_file(self, full_path, read_iterator, mode):
+        raise NotImplementedError
+
+    def _get_keyfile(self, key):
+        raise NotImplementedError
+
+    def random_filename(self):
+        raise NotImplementedError
+
+    def _store(self, key, value_reader, metadata, mode):
+        (filename,  # hash-name
+         full_path  # full-path/hash-name
+         ) = self.random_filename()
+
+        key_file, key_file_path = self._get_keyfile(key)
+
+        # STORE METADATA
+        _metadata = {
+            "version": "v1",
+
+            "key_file": key_file,  # this is the .key.json file storing meta
+            "key_file_path": key_file_path,  # full path to key_file
+            "archive_key": key,  # original name we stored archive under, e.g my-archive.zip
+            "archive_filename": filename,  # the actual filename we stored that file under
+            "archive_full_path": full_path,
+
+            "store_time": time.time(),
+            "access_count": 0,
+            "access_time": 0,
+
+            "size": 0
+        }
+        if metadata:
+            _metadata.update(metadata)
+
+        read_iterator = iter(functools.partial(value_reader.read, 2**22), b'')
+        size, sha256 = self._write_file(full_path, read_iterator, mode)
+        _metadata['size'] = size
+        _metadata['sha256'] = sha256
+
+        # after archive is finished, we create a key to save the presence of the binary file
+        with self.fs.open(key_file_path, 'wb') as f:
+            f.write(json.dumps(_metadata))
+
+        return key, filename, size, _metadata
+
+    def _fetch(self,  key, retry, retry_attempts, retry_backoff):
+        if retry is NOT_GIVEN:
+            retry = False
+        if retry_attempts is NOT_GIVEN:
+            retry_attempts = 0
+
+        if retry and retry_attempts > 0:
+            for attempt in range(1, retry_attempts + 1):
+                if key in self:
+                    break
+                # we didn't find the key, wait retry_backoff N seconds, and re-check
+                time.sleep(retry_backoff)
+
+        if key not in self:
+            log.exception(f'requested key={key} not found in {self} retry={retry}, attempts={retry_attempts}')
+            raise KeyError(key)
+
+        key_file, key_file_path = self._get_keyfile(key)
+        with self.fs.open(key_file_path, 'rb') as f:
+            metadata = json.loads(f.read())
+
+        archive_path = metadata['archive_full_path']
+
+        try:
+            return ShardFileReader(self.fs.open(archive_path, 'rb')), metadata
+        finally:
+            # update usage stats, count and accessed
+            metadata["access_count"] = metadata.get("access_count", 0) + 1
+            metadata["access_time"] = time.time()
+            log.debug('Updated %s with access snapshot, access_count=%s access_time=%s',
+                      key_file, metadata['access_count'], metadata['access_time'])
+            with self.fs.open(key_file_path, 'wb') as f:
+                f.write(json.dumps(metadata))
+
+    def _remove(self, key):
+        if key not in self:
+            log.exception(f'requested key={key} not found in {self}')
+            raise KeyError(key)
+
+        key_file, key_file_path = self._get_keyfile(key)
+        with self.fs.open(key_file_path, 'rb') as f:
+            metadata = json.loads(f.read())
+
+        archive_path = metadata['archive_full_path']
+        self.fs.rm(archive_path)
+        self.fs.rm(key_file_path)
+        return 1
+
+    @property
+    def storage_medium(self):
+        return getattr(self, self.storage_type)
+
+    @property
+    def key_suffix(self):
+        return 'key.json'
+
+    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, key_file_path = self._get_keyfile(key)
+        return self.fs.exists(key_file_path)
+
+
+class BaseCache:
+    _locking_url: str = ''
+    _storage_path: str = ''
+    _config = {}
+    retry = False
+    retry_attempts = 0
+    retry_backoff = 1
+    _shards = tuple()
+
+    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
+
+        """
+        return self.has_key(key)
+
+    def __repr__(self):
+        return f'<{self.__class__.__name__}(storage={self._storage_path})>'
+
+    @classmethod
+    def gb_to_bytes(cls, gb):
+        return gb * (1024 ** 3)
+
+    @property
+    def storage_path(self):
+        return self._storage_path
+
+    @classmethod
+    def get_stats_db(cls):
+        return StatsDB()
+
+    def get_conf(self, key, pop=False):
+        if key not in self._config:
+            raise ValueError(f"No configuration key '{key}', please make sure it exists in archive_cache config")
+        val = self._config[key]
+        if pop:
+            del self._config[key]
+        return val
+
+    def _get_shard(self, key):
+        raise NotImplementedError
+
+    def _get_size(self, shard, archive_path):
+        raise NotImplementedError
+
+    def store(self, key, value_reader, metadata=None):
+        shard = self._get_shard(key)
+        return shard.store(key, value_reader, metadata)
+
+    def fetch(self, key, retry=NOT_GIVEN, retry_attempts=NOT_GIVEN) -> tuple[typing.BinaryIO, dict]:
+        """
+        Return file handle corresponding to `key` from specific shard cache.
+        """
+        if retry is NOT_GIVEN:
+            retry = self.retry
+        if retry_attempts is NOT_GIVEN:
+            retry_attempts = self.retry_attempts
+        retry_backoff = self.retry_backoff
+
+        shard = self._get_shard(key)
+        return shard.fetch(key, retry=retry, retry_attempts=retry_attempts, retry_backoff=retry_backoff)
+
+    def remove(self, key):
+        shard = self._get_shard(key)
+        return shard.remove(key)
+
+    def has_key(self, archive_key):
+        """Return `True` if `key` matching item is found in cache.
+
+        :param archive_key: key for item, this is a unique archive name we want to store data under. e.g my-archive-svn.zip
+        :return: True if key is found
+
+        """
+        shard = self._get_shard(archive_key)
+        return archive_key in shard
+
+    def iter_keys(self):
+        for shard in self._shards:
+            if shard.fs.exists(shard.storage_medium):
+                for path, _dirs, _files in shard.fs.walk(shard.storage_medium):
+                    for key_file_path in _files:
+                        if key_file_path.endswith(shard.key_suffix):
+                            yield shard, key_file_path
+
+    def get_lock(self, lock_key):
+        return GenerationLock(lock_key, self._locking_url)
+
+    def evict(self, policy=None, size_limit=None) -> int:
+        """
+        Remove old items based on the conditions
+
+
+        explanation of this algo:
+        iterate over each shard, then for each shard iterate over the .key files
+        read the key files metadata stored. This gives us a full list of keys, cached_archived, their size and
+        access data, time creation, and access counts.
+
+        Store that into a memory DB so we can run different sorting strategies easily.
+        Summing the size is a sum sql query.
+
+        Then we run a sorting strategy based on eviction policy.
+        We iterate over sorted keys, and remove each checking if we hit the overall limit.
+        """
+
+        policy = policy or self._eviction_policy
+        size_limit = size_limit or self._cache_size_limit
+
+        select_policy = EVICTION_POLICY[policy]['evict']
+
+        log.debug('Running eviction policy \'%s\', and checking for size limit: %s',
+                  policy, format_size(size_limit))
+
+        if select_policy is None:
+            return 0
+
+        db = self.get_stats_db()
+
+        data = []
+        cnt = 1
+
+        for shard, key_file in self.iter_keys():
+            with shard.fs.open(os.path.join(shard.storage_medium, key_file), 'rb') as f:
+                metadata = json.loads(f.read())
+
+            key_file_path = os.path.join(shard.storage_medium, key_file)
+
+            archive_key = metadata['archive_key']
+            archive_path = metadata['archive_full_path']
+
+            size = metadata.get('size')
+            if not size:
+                # in case we don't have size re-calc it...
+                size = self._get_size(shard, archive_path)
+
+            data.append([
+                cnt,
+                key_file,
+                key_file_path,
+                archive_key,
+                archive_path,
+                metadata.get('store_time', 0),
+                metadata.get('access_time', 0),
+                metadata.get('access_count', 0),
+                size,
+            ])
+            cnt += 1
+
+        # Insert bulk data using executemany
+        db.bulk_insert(data)
+
+        total_size = db.get_total_size()
+        log.debug('Analyzed %s keys, occupying: %s, running eviction to match %s',
+                  len(data), format_size(total_size), format_size(size_limit))
+
+        removed_items = 0
+        removed_size = 0
+        for key_file, archive_key, size in db.get_sorted_keys(select_policy):
+            # simulate removal impact BEFORE removal
+            total_size -= size
+
+            if total_size <= size_limit:
+                # we obtained what we wanted...
+                break
+
+            self.remove(archive_key)
+            removed_items += 1
+            removed_size += size
+
+        log.debug('Removed %s cache archives, and reduced size by: %s',
+                  removed_items, format_size(removed_size))
+        return removed_items
+
+    def get_statistics(self):
+        total_files = 0
+        total_size = 0
+        meta = {}
+
+        for shard, key_file in self.iter_keys():
+            json_key = f"{shard.storage_medium}/{key_file}"
+            with shard.fs.open(json_key, 'rb') as f:
+                total_files += 1
+                metadata = json.loads(f.read())
+                total_size += metadata['size']
+
+        return total_files, total_size, meta
+

vcsserver/lib/archive_cache/backends/fanout_cache.py

@@ -0,0 +1,166 b''
+# Copyright (C) 2015-2024 RhodeCode GmbH
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License, version 3
+# (only), as published by the Free Software Foundation.
+#
+# 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 Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+# This program is dual-licensed. If you wish to learn more about the
+# RhodeCode Enterprise Edition, including its added features, Support services,
+# and proprietary license terms, please see https://rhodecode.com/licenses/
+
+import codecs
+import hashlib
+import logging
+import os
+
+import fsspec
+
+from .base import BaseCache, BaseShard
+from ..utils import ShardFileReader, NOT_GIVEN
+from ...type_utils import str2bool
+
+log = logging.getLogger(__name__)
+
+
+class FileSystemShard(BaseShard):
+
+    def __init__(self, index, directory, **settings):
+        self._index = index
+        self._directory = directory
+        self.storage_type = 'directory'
+        self.fs = fsspec.filesystem('file')
+
+    @property
+    def directory(self):
+        """Cache directory."""
+        return self._directory
+
+    def _get_keyfile(self, archive_key) -> tuple[str, str]:
+        key_file = f'{archive_key}.{self.key_suffix}'
+        return key_file, os.path.join(self.directory, key_file)
+
+    def _get_writer(self, path, mode):
+        for count in range(1, 11):
+            try:
+                # Another cache may have deleted the directory before
+                # the file could be opened.
+                return self.fs.open(path, mode)
+            except OSError:
+                if count == 10:
+                    # Give up after 10 tries to open the file.
+                    raise
+                continue
+
+    def _write_file(self, full_path, iterator, mode):
+        # ensure dir exists
+        destination, _ = os.path.split(full_path)
+        if not self.fs.exists(destination):
+            self.fs.makedirs(destination)
+
+        writer = self._get_writer(full_path, mode)
+
+        digest = hashlib.sha256()
+        with writer:
+            size = 0
+            for chunk in iterator:
+                size += len(chunk)
+                digest.update(chunk)
+                writer.write(chunk)
+            writer.flush()
+            # Get the file descriptor
+            fd = writer.fileno()
+
+            # Sync the file descriptor to disk, helps with NFS cases...
+            os.fsync(fd)
+            sha256 = digest.hexdigest()
+            log.debug('written new archive cache under %s, sha256: %s', full_path, sha256)
+            return size, sha256
+
+    def store(self, key, value_reader, metadata: dict | None = None):
+        return self._store(key, value_reader, metadata, mode='xb')
+
+    def fetch(self, key, retry=NOT_GIVEN, retry_attempts=NOT_GIVEN, retry_backoff=1) -> tuple[ShardFileReader, dict]:
+        return self._fetch(key, retry, retry_attempts, retry_backoff)
+
+    def remove(self, key):
+        return self._remove(key)
+
+    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')
+
+        archive_name = hex_name[4:] + '.archive_cache'
+        filename = f"{hex_name[:2]}/{hex_name[2:4]}/{archive_name}"
+
+        full_path = os.path.join(self.directory, filename)
+        return archive_name, full_path
+
+    def __repr__(self):
+        return f'{self.__class__.__name__}(index={self._index}, dir={self.directory})'
+
+
+class FileSystemFanoutCache(BaseCache):
+
+    def __init__(self, locking_url, **settings):
+        """
+        Initialize file system cache instance.
+
+        :param str locking_url: redis url for a lock
+        :param settings: settings dict
+
+        """
+        self._locking_url = locking_url
+        self._config = settings
+        cache_dir = self.get_conf('archive_cache.filesystem.store_dir')
+        directory = str(cache_dir)
+        directory = os.path.expanduser(directory)
+        directory = os.path.expandvars(directory)
+        self._directory = directory
+        self._storage_path = directory
+
+        # check if it's ok to write, and re-create the archive cache
+        if not os.path.isdir(self._directory):
+            os.makedirs(self._directory, exist_ok=True)
+
+        self._count = int(self.get_conf('archive_cache.filesystem.cache_shards', pop=True))
+
+        self._eviction_policy = self.get_conf('archive_cache.filesystem.eviction_policy', pop=True)
+        self._cache_size_limit = self.gb_to_bytes(int(self.get_conf('archive_cache.filesystem.cache_size_gb')))
+
+        self.retry = str2bool(self.get_conf('archive_cache.filesystem.retry', pop=True))
+        self.retry_attempts = int(self.get_conf('archive_cache.filesystem.retry_attempts', pop=True))
+        self.retry_backoff = int(self.get_conf('archive_cache.filesystem.retry_backoff', pop=True))
+
+        log.debug('Initializing archival cache instance under %s', self._directory)
+        self._shards = tuple(
+            FileSystemShard(
+                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_shard(self, key) -> FileSystemShard:
+        index = self._hash(key) % self._count
+        shard = self._shards[index]
+        return shard
+
+    def _get_size(self, shard, archive_path):
+        return os.stat(archive_path).st_size

vcsserver/lib/archive_cache/backends/objectstore_cache.py

@@ -0,0 +1,150 b''
+# Copyright (C) 2015-2024 RhodeCode GmbH
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License, version 3
+# (only), as published by the Free Software Foundation.
+#
+# 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 Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+# This program is dual-licensed. If you wish to learn more about the
+# RhodeCode Enterprise Edition, including its added features, Support services,
+# and proprietary license terms, please see https://rhodecode.com/licenses/
+
+import codecs
+import hashlib
+import logging
+import os
+
+import fsspec
+
+from .base import BaseCache, BaseShard
+from ..utils import ShardFileReader, NOT_GIVEN
+from ...type_utils import str2bool
+
+log = logging.getLogger(__name__)
+
+
+class S3Shard(BaseShard):
+
+    def __init__(self, index, bucket, **settings):
+        self._index = index
+        self._bucket = bucket
+        self.storage_type = 'bucket'
+
+        endpoint_url = settings.pop('archive_cache.objectstore.url')
+        key = settings.pop('archive_cache.objectstore.key')
+        secret = settings.pop('archive_cache.objectstore.secret')
+
+        self.fs = fsspec.filesystem('s3', anon=False, endpoint_url=endpoint_url, key=key, secret=secret)
+
+    @property
+    def bucket(self):
+        """Cache bucket."""
+        return self._bucket
+
+    def _get_keyfile(self, archive_key) -> tuple[str, str]:
+        key_file = f'{archive_key}-{self.key_suffix}'
+        return key_file, os.path.join(self.bucket, key_file)
+
+    def _get_writer(self, path, mode):
+        return self.fs.open(path, 'wb')
+
+    def _write_file(self, full_path, iterator, mode):
+        # ensure bucket exists
+        destination = self.bucket
+        if not self.fs.exists(destination):
+            self.fs.mkdir(destination, s3_additional_kwargs={})
+
+        writer = self._get_writer(full_path, mode)
+
+        digest = hashlib.sha256()
+        with writer:
+            size = 0
+            for chunk in iterator:
+                size += len(chunk)
+                digest.update(chunk)
+                writer.write(chunk)
+
+            sha256 = digest.hexdigest()
+            log.debug('written new archive cache under %s, sha256: %s', full_path, sha256)
+            return size, sha256
+
+    def store(self, key, value_reader, metadata: dict | None = None):
+        return self._store(key, value_reader, metadata, mode='wb')
+
+    def fetch(self, key, retry=NOT_GIVEN, retry_attempts=NOT_GIVEN, retry_backoff=1) -> tuple[ShardFileReader, dict]:
+        return self._fetch(key, retry, retry_attempts, retry_backoff)
+
+    def remove(self, key):
+        return self._remove(key)
+
+    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')
+
+        archive_name = hex_name[4:] + '.archive_cache'
+        filename = f"{hex_name[:2]}-{hex_name[2:4]}-{archive_name}"
+
+        full_path = os.path.join(self.bucket, filename)
+        return archive_name, full_path
+
+    def __repr__(self):
+        return f'{self.__class__.__name__}(index={self._index}, bucket={self.bucket})'
+
+
+class ObjectStoreCache(BaseCache):
+
+    def __init__(self, locking_url, **settings):
+        """
+        Initialize objectstore cache instance.
+
+        :param str locking_url: redis url for a lock
+        :param settings: settings dict
+
+        """
+        self._locking_url = locking_url
+        self._config = settings
+
+        objectstore_url = self.get_conf('archive_cache.objectstore.url')
+        self._storage_path = objectstore_url
+
+        self._count = int(self.get_conf('archive_cache.objectstore.bucket_shards', pop=True))
+
+        self._eviction_policy = self.get_conf('archive_cache.objectstore.eviction_policy', pop=True)
+        self._cache_size_limit = self.gb_to_bytes(int(self.get_conf('archive_cache.objectstore.cache_size_gb')))
+
+        self.retry = str2bool(self.get_conf('archive_cache.objectstore.retry', pop=True))
+        self.retry_attempts = int(self.get_conf('archive_cache.objectstore.retry_attempts', pop=True))
+        self.retry_backoff = int(self.get_conf('archive_cache.objectstore.retry_backoff', pop=True))
+
+        log.debug('Initializing archival cache instance under %s', objectstore_url)
+        self._shards = tuple(
+            S3Shard(
+                index=num,
+                bucket='rhodecode-archivecache-%03d' % num,
+                **settings,
+            )
+            for num in range(self._count)
+        )
+        self._hash = self._shards[0].hash
+
+    def _get_shard(self, key) -> S3Shard:
+        index = self._hash(key) % self._count
+        shard = self._shards[index]
+        return shard
+
+    def _get_size(self, shard, archive_path):
+        return shard.fs.info(archive_path)['size']

vcsserver/lib/archive_cache/lock.py

@@ -0,0 +1,62 b''
+# Copyright (C) 2015-2024 RhodeCode GmbH
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License, version 3
+# (only), as published by the Free Software Foundation.
+#
+# 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 Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+# This program is dual-licensed. If you wish to learn more about the
+# RhodeCode Enterprise Edition, including its added features, Support services,
+# and proprietary license terms, please see https://rhodecode.com/licenses/
+
+import redis
+from .._vendor import redis_lock
+
+
+class ArchiveCacheGenerationLock(Exception):
+    pass
+
+
+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 ArchiveCacheGenerationLock('Failed to create a lock')
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.lock.release()

vcsserver/lib/archive_cache/utils.py

@@ -0,0 +1,134 b''
+# Copyright (C) 2015-2024 RhodeCode GmbH
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License, version 3
+# (only), as published by the Free Software Foundation.
+#
+# 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 Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+# This program is dual-licensed. If you wish to learn more about the
+# RhodeCode Enterprise Edition, including its added features, Support services,
+# and proprietary license terms, please see https://rhodecode.com/licenses/
+
+import sqlite3
+import s3fs.core
+
+NOT_GIVEN = -917
+
+
+EVICTION_POLICY = {
+    'none': {
+        'evict': None,
+    },
+    'least-recently-stored': {
+        'evict': 'SELECT {fields} FROM archive_cache ORDER BY store_time',
+    },
+    'least-recently-used': {
+        'evict': 'SELECT {fields} FROM archive_cache ORDER BY access_time',
+    },
+    'least-frequently-used': {
+        'evict': 'SELECT {fields} FROM archive_cache ORDER BY access_count',
+    },
+}
+
+
+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
+
+
+def format_size(size):
+    # Convert size in bytes to a human-readable format (e.g., KB, MB, GB)
+    for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
+        if size < 1024:
+            return f"{size:.2f} {unit}"
+        size /= 1024
+
+
+class StatsDB:
+
+    def __init__(self):
+        self.connection = sqlite3.connect(':memory:')
+        self._init_db()
+
+    def _init_db(self):
+        qry = '''
+            CREATE TABLE IF NOT EXISTS archive_cache (
+             rowid INTEGER PRIMARY KEY,
+             key_file TEXT,
+             key_file_path TEXT,
+             archive_key TEXT,
+             archive_path TEXT,
+             store_time REAL,
+             access_time REAL,
+             access_count INTEGER DEFAULT 0,
+             size INTEGER DEFAULT 0
+             )
+        '''
+
+        self.sql(qry)
+        self.connection.commit()
+
+    @property
+    def sql(self):
+        return self.connection.execute
+
+    def bulk_insert(self, rows):
+        qry = '''
+            INSERT INTO archive_cache (
+             rowid,
+             key_file,
+             key_file_path,
+             archive_key,
+             archive_path,
+             store_time,
+             access_time,
+             access_count,
+             size
+            )
+            VALUES (
+            ?, ?, ?, ?, ?, ?, ?, ?, ?
+            )
+        '''
+        cursor = self.connection.cursor()
+        cursor.executemany(qry, rows)
+        self.connection.commit()
+
+    def get_total_size(self):
+        qry = 'SELECT COALESCE(SUM(size), 0) FROM archive_cache'
+        ((total_size,),) = self.sql(qry).fetchall()
+        return total_size
+
+    def get_sorted_keys(self, select_policy):
+        select_policy_qry = select_policy.format(fields='key_file, archive_key, size')
+        return self.sql(select_policy_qry).fetchall()
+
+
+class ShardFileReader:
+
+    def __init__(self, file_like_reader):
+        self._file_like_reader = file_like_reader
+
+    def __getattr__(self, item):
+        if isinstance(self._file_like_reader, s3fs.core.S3File):
+            match item:
+                case 'name':
+                    # S3 FileWrapper doesn't support name attribute, and we use it
+                    return self._file_like_reader.full_name
+                case _:
+                    return getattr(self._file_like_reader, item)
+        else:
+            return getattr(self._file_like_reader, item)
+
+    def __repr__(self):
+        return f'<{self.__class__.__name__}={self._file_like_reader}>'

requirements.txt

@@ -1,73 +1,102 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
 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
+fsspec==2024.6.0
 gunicorn==21.2.0
   packaging==24.0
 hg-evolve==11.1.3
 importlib-metadata==6.0.0
   zipp==3.15.0
 mercurial==6.7.4
 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
+s3fs==2024.6.0
+  aiobotocore==2.13.0
+    aiohttp==3.9.5
+      aiosignal==1.3.1
+        frozenlist==1.4.1
+      attrs==22.2.0
+      frozenlist==1.4.1
+      multidict==6.0.5
+      yarl==1.9.4
+        idna==3.4
+        multidict==6.0.5
+    aioitertools==0.11.0
+    botocore==1.34.106
+      jmespath==1.0.1
+      python-dateutil==2.8.2
+        six==1.16.0
+      urllib3==1.26.14
+    wrapt==1.16.0
+  aiohttp==3.9.5
+    aiosignal==1.3.1
+      frozenlist==1.4.1
+    attrs==22.2.0
+    frozenlist==1.4.1
+    multidict==6.0.5
+    yarl==1.9.4
+      idna==3.4
+      multidict==6.0.5
+  fsspec==2024.6.0
 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,187 +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.lib.archive_cache import get_archival_cache_store
 
 from vcsserver import exceptions
 from vcsserver.exceptions import NoContentException
 from vcsserver.hgcompat import archival
 from vcsserver.lib.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:
         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'')
 
     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()
 
     with open(archive_tmp_path, 'rb') as archive_file:
         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, 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

vcsserver/http_main.py

@@ -1,774 +1,774 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 io
 import os
 import platform
 import sys
 import locale
 import logging
 import uuid
 import time
 import wsgiref.util
 import tempfile
 import psutil
 
 from itertools import chain
 
 import msgpack
 import configparser
 
 from pyramid.config import Configurator
 from pyramid.wsgi import wsgiapp
 from pyramid.response import Response
 
 from vcsserver.base import BytesEnvelope, BinaryEnvelope
 from vcsserver.lib.ext_json import json
 from vcsserver.config.settings_maker import SettingsMaker
 from vcsserver.lib.str_utils import safe_int
 from vcsserver.lib.statsd_client import StatsdClient
 from vcsserver.tweens.request_wrapper import get_headers_call_context
 
 import vcsserver
 from vcsserver import remote_wsgi, scm_app, settings, hgpatches
 from vcsserver.git_lfs.app import GIT_LFS_CONTENT_TYPE, GIT_LFS_PROTO_PAT
 from vcsserver.echo_stub import remote_wsgi as remote_wsgi_stub
 from vcsserver.echo_stub.echo_app import EchoApp
 from vcsserver.exceptions import HTTPRepoLocked, HTTPRepoBranchProtected
 from vcsserver.lib.exc_tracking import store_exception, format_exc
 from vcsserver.server import VcsServer
 
 strict_vcs = True
 
 git_import_err = None
 try:
     from vcsserver.remote.git_remote import GitFactory, GitRemote
 except ImportError as e:
     GitFactory = None
     GitRemote = None
     git_import_err = e
     if strict_vcs:
         raise
 
 
 hg_import_err = None
 try:
     from vcsserver.remote.hg_remote import MercurialFactory, HgRemote
 except ImportError as e:
     MercurialFactory = None
     HgRemote = None
     hg_import_err = e
     if strict_vcs:
         raise
 
 
 svn_import_err = None
 try:
     from vcsserver.remote.svn_remote import SubversionFactory, SvnRemote
 except ImportError as e:
     SubversionFactory = None
     SvnRemote = None
     svn_import_err = e
     if strict_vcs:
         raise
 
 log = logging.getLogger(__name__)
 
 # due to Mercurial/glibc2.27 problems we need to detect if locale settings are
 # causing problems and "fix" it in case they do and fallback to LC_ALL = C
 
 try:
     locale.setlocale(locale.LC_ALL, '')
 except locale.Error as e:
     log.error(
         'LOCALE ERROR: failed to set LC_ALL, fallback to LC_ALL=C, org error: %s', e)
     os.environ['LC_ALL'] = 'C'
 
 
 def _is_request_chunked(environ):
     stream = environ.get('HTTP_TRANSFER_ENCODING', '') == 'chunked'
     return stream
 
 
 def log_max_fd():
     try:
         maxfd = psutil.Process().rlimit(psutil.RLIMIT_NOFILE)[1]
         log.info('Max file descriptors value: %s', maxfd)
     except Exception:
         pass
 
 
 class VCS:
     def __init__(self, locale_conf=None, cache_config=None):
         self.locale = locale_conf
         self.cache_config = cache_config
         self._configure_locale()
 
         log_max_fd()
 
         if GitFactory and GitRemote:
             git_factory = GitFactory()
             self._git_remote = GitRemote(git_factory)
         else:
             log.error("Git client import failed: %s", git_import_err)
 
         if MercurialFactory and HgRemote:
             hg_factory = MercurialFactory()
             self._hg_remote = HgRemote(hg_factory)
         else:
             log.error("Mercurial client import failed: %s", hg_import_err)
 
         if SubversionFactory and SvnRemote:
             svn_factory = SubversionFactory()
 
             # hg factory is used for svn url validation
             hg_factory = MercurialFactory()
             self._svn_remote = SvnRemote(svn_factory, hg_factory=hg_factory)
         else:
             log.error("Subversion client import failed: %s", svn_import_err)
 
         self._vcsserver = VcsServer()
 
     def _configure_locale(self):
         if self.locale:
             log.info('Settings locale: `LC_ALL` to %s', self.locale)
         else:
             log.info('Configuring locale subsystem based on environment variables')
         try:
             # If self.locale is the empty string, then the locale
             # module will use the environment variables. See the
             # documentation of the package `locale`.
             locale.setlocale(locale.LC_ALL, self.locale)
 
             language_code, encoding = locale.getlocale()
             log.info(
                 'Locale set to language code "%s" with encoding "%s".',
                 language_code, encoding)
         except locale.Error:
             log.exception('Cannot set locale, not configuring the locale system')
 
 
 class WsgiProxy:
     def __init__(self, wsgi):
         self.wsgi = wsgi
 
     def __call__(self, environ, start_response):
         input_data = environ['wsgi.input'].read()
         input_data = msgpack.unpackb(input_data)
 
         error = None
         try:
             data, status, headers = self.wsgi.handle(
                 input_data['environment'], input_data['input_data'],
                 *input_data['args'], **input_data['kwargs'])
         except Exception as e:
             data, status, headers = [], None, None
             error = {
                 'message': str(e),
                 '_vcs_kind': getattr(e, '_vcs_kind', None)
             }
 
         start_response(200, {})
         return self._iterator(error, status, headers, data)
 
     def _iterator(self, error, status, headers, data):
         initial_data = [
             error,
             status,
             headers,
         ]
 
         for d in chain(initial_data, data):
             yield msgpack.packb(d)
 
 
 def not_found(request):
     return {'status': '404 NOT FOUND'}
 
 
 class VCSViewPredicate:
     def __init__(self, val, config):
         self.remotes = val
 
     def text(self):
         return f'vcs view method = {list(self.remotes.keys())}'
 
     phash = text
 
     def __call__(self, context, request):
         """
         View predicate that returns true if given backend is supported by
         defined remotes.
         """
         backend = request.matchdict.get('backend')
         return backend in self.remotes
 
 
 class HTTPApplication:
     ALLOWED_EXCEPTIONS = ('KeyError', 'URLError')
 
     remote_wsgi = remote_wsgi
     _use_echo_app = False
 
     def __init__(self, settings=None, global_config=None):
 
         self.config = Configurator(settings=settings)
         # Init our statsd at very start
         self.config.registry.statsd = StatsdClient.statsd
         self.config.registry.vcs_call_context = {}
 
         self.global_config = global_config
         self.config.include('vcsserver.lib.rc_cache')
-        self.config.include('vcsserver.lib.rc_cache.archive_cache')
+        self.config.include('vcsserver.lib.archive_cache')
 
         settings_locale = settings.get('locale', '') or 'en_US.UTF-8'
         vcs = VCS(locale_conf=settings_locale, cache_config=settings)
         self._remotes = {
             'hg': vcs._hg_remote,
             'git': vcs._git_remote,
             'svn': vcs._svn_remote,
             'server': vcs._vcsserver,
         }
         if settings.get('dev.use_echo_app', 'false').lower() == 'true':
             self._use_echo_app = True
             log.warning("Using EchoApp for VCS operations.")
             self.remote_wsgi = remote_wsgi_stub
 
         self._configure_settings(global_config, settings)
 
         self._configure()
 
     def _configure_settings(self, global_config, app_settings):
         """
         Configure the settings module.
         """
         settings_merged = global_config.copy()
         settings_merged.update(app_settings)
 
         binary_dir = app_settings['core.binary_dir']
 
         settings.BINARY_DIR = binary_dir
 
         # Store the settings to make them available to other modules.
         vcsserver.PYRAMID_SETTINGS = settings_merged
         vcsserver.CONFIG = settings_merged
 
     def _configure(self):
         self.config.add_renderer(name='msgpack', factory=self._msgpack_renderer_factory)
 
         self.config.add_route('service', '/_service')
         self.config.add_route('status', '/status')
         self.config.add_route('hg_proxy', '/proxy/hg')
         self.config.add_route('git_proxy', '/proxy/git')
 
         # rpc methods
         self.config.add_route('vcs', '/{backend}')
 
         # streaming rpc remote methods
         self.config.add_route('vcs_stream', '/{backend}/stream')
 
         # vcs operations clone/push as streaming
         self.config.add_route('stream_git', '/stream/git/*repo_name')
         self.config.add_route('stream_hg', '/stream/hg/*repo_name')
 
         self.config.add_view(self.status_view, route_name='status', renderer='json')
         self.config.add_view(self.service_view, route_name='service', renderer='msgpack')
 
         self.config.add_view(self.hg_proxy(), route_name='hg_proxy')
         self.config.add_view(self.git_proxy(), route_name='git_proxy')
         self.config.add_view(self.vcs_view, route_name='vcs', renderer='msgpack',
                              vcs_view=self._remotes)
         self.config.add_view(self.vcs_stream_view, route_name='vcs_stream',
                              vcs_view=self._remotes)
 
         self.config.add_view(self.hg_stream(), route_name='stream_hg')
         self.config.add_view(self.git_stream(), route_name='stream_git')
 
         self.config.add_view_predicate('vcs_view', VCSViewPredicate)
 
         self.config.add_notfound_view(not_found, renderer='json')
 
         self.config.add_view(self.handle_vcs_exception, context=Exception)
 
         self.config.add_tween(
             'vcsserver.tweens.request_wrapper.RequestWrapperTween',
         )
         self.config.add_request_method(
             'vcsserver.lib.request_counter.get_request_counter',
             'request_count')
 
     def wsgi_app(self):
         return self.config.make_wsgi_app()
 
     def _vcs_view_params(self, request):
         remote = self._remotes[request.matchdict['backend']]
         payload = msgpack.unpackb(request.body, use_list=True)
 
         method = payload.get('method')
         params = payload['params']
         wire = params.get('wire')
         args = params.get('args')
         kwargs = params.get('kwargs')
         context_uid = None
 
         request.registry.vcs_call_context = {
             'method': method,
             'repo_name': payload.get('_repo_name'),
         }
 
         if wire:
             try:
                 wire['context'] = context_uid = uuid.UUID(wire['context'])
             except KeyError:
                 pass
             args.insert(0, wire)
         repo_state_uid = wire.get('repo_state_uid') if wire else None
 
         # NOTE(marcink): trading complexity for slight performance
         if log.isEnabledFor(logging.DEBUG):
             # also we SKIP printing out any of those methods args since they maybe excessive
             just_args_methods = {
                 'commitctx': ('content', 'removed', 'updated'),
                 'commit': ('content', 'removed', 'updated')
             }
             if method in just_args_methods:
                 skip_args = just_args_methods[method]
                 call_args = ''
                 call_kwargs = {}
                 for k in kwargs:
                     if k in skip_args:
                         # replace our skip key with dummy
                         call_kwargs[k] = f'RemovedParam({k})'
                     else:
                         call_kwargs[k] = kwargs[k]
             else:
                 call_args = args[1:]
                 call_kwargs = kwargs
 
             log.debug('Method requested:`%s` with args:%s kwargs:%s context_uid: %s, repo_state_uid:%s',
                       method, call_args, call_kwargs, context_uid, repo_state_uid)
 
         statsd = request.registry.statsd
         if statsd:
             statsd.incr(
                 'vcsserver_method_total', tags=[
                     f"method:{method}",
                 ])
         return payload, remote, method, args, kwargs
 
     def vcs_view(self, request):
 
         payload, remote, method, args, kwargs = self._vcs_view_params(request)
         payload_id = payload.get('id')
 
         try:
             resp = getattr(remote, method)(*args, **kwargs)
         except Exception as e:
             exc_info = list(sys.exc_info())
             exc_type, exc_value, exc_traceback = exc_info
 
             org_exc = getattr(e, '_org_exc', None)
             org_exc_name = None
             org_exc_tb = ''
             if org_exc:
                 org_exc_name = org_exc.__class__.__name__
                 org_exc_tb = getattr(e, '_org_exc_tb', '')
                 # replace our "faked" exception with our org
                 exc_info[0] = org_exc.__class__
                 exc_info[1] = org_exc
 
             should_store_exc = True
             if org_exc:
                 def get_exc_fqn(_exc_obj):
                     module_name = getattr(org_exc.__class__, '__module__', 'UNKNOWN')
                     return module_name + '.' + org_exc_name
 
                 exc_fqn = get_exc_fqn(org_exc)
 
                 if exc_fqn in ['mercurial.error.RepoLookupError',
                                'vcsserver.exceptions.RefNotFoundException']:
                     should_store_exc = False
 
             if should_store_exc:
                 store_exception(id(exc_info), exc_info, request_path=request.path)
 
             tb_info = format_exc(exc_info)
 
             type_ = e.__class__.__name__
             if type_ not in self.ALLOWED_EXCEPTIONS:
                 type_ = None
 
             resp = {
                 'id': payload_id,
                 'error': {
                     'message': str(e),
                     'traceback': tb_info,
                     'org_exc': org_exc_name,
                     'org_exc_tb': org_exc_tb,
                     'type': type_
                 }
             }
 
             try:
                 resp['error']['_vcs_kind'] = getattr(e, '_vcs_kind', None)
             except AttributeError:
                 pass
         else:
             resp = {
                 'id': payload_id,
                 'result': resp
             }
         log.debug('Serving data for method %s', method)
         return resp
 
     def vcs_stream_view(self, request):
         payload, remote, method, args, kwargs = self._vcs_view_params(request)
         # this method has a stream: marker we remove it here
         method = method.split('stream:')[-1]
         chunk_size = safe_int(payload.get('chunk_size')) or 4096
 
         resp = getattr(remote, method)(*args, **kwargs)
 
         def get_chunked_data(method_resp):
             stream = io.BytesIO(method_resp)
             while 1:
                 chunk = stream.read(chunk_size)
                 if not chunk:
                     break
                 yield chunk
 
         response = Response(app_iter=get_chunked_data(resp))
         response.content_type = 'application/octet-stream'
 
         return response
 
     def status_view(self, request):
         import vcsserver
         _platform_id = platform.uname()[1] or 'instance'
 
         return {
             "status": "OK",
             "vcsserver_version": vcsserver.get_version(),
             "platform": _platform_id,
             "pid": os.getpid(),
         }
 
     def service_view(self, request):
         import vcsserver
 
         payload = msgpack.unpackb(request.body, use_list=True)
         server_config, app_config = {}, {}
 
         try:
             path = self.global_config['__file__']
             config = configparser.RawConfigParser()
 
             config.read(path)
 
             if config.has_section('server:main'):
                 server_config = dict(config.items('server:main'))
             if config.has_section('app:main'):
                 app_config = dict(config.items('app:main'))
 
         except Exception:
             log.exception('Failed to read .ini file for display')
 
         environ = list(os.environ.items())
 
         resp = {
             'id': payload.get('id'),
             'result': dict(
                 version=vcsserver.get_version(),
                 config=server_config,
                 app_config=app_config,
                 environ=environ,
                 payload=payload,
             )
         }
         return resp
 
     def _msgpack_renderer_factory(self, info):
 
         def _render(value, system):
             bin_type = False
             res = value.get('result')
             if isinstance(res, BytesEnvelope):
                 log.debug('Result is wrapped in BytesEnvelope type')
                 bin_type = True
             elif isinstance(res, BinaryEnvelope):
                 log.debug('Result is wrapped in BinaryEnvelope type')
                 value['result'] = res.val
                 bin_type = True
 
             request = system.get('request')
             if request is not None:
                 response = request.response
                 ct = response.content_type
                 if ct == response.default_content_type:
                     response.content_type = 'application/x-msgpack'
                     if bin_type:
                         response.content_type = 'application/x-msgpack-bin'
 
             return msgpack.packb(value, use_bin_type=bin_type)
         return _render
 
     def set_env_from_config(self, environ, config):
         dict_conf = {}
         try:
             for elem in config:
                 if elem[0] == 'rhodecode':
                     dict_conf = json.loads(elem[2])
                     break
         except Exception:
             log.exception('Failed to fetch SCM CONFIG')
             return
 
         username = dict_conf.get('username')
         if username:
             environ['REMOTE_USER'] = username
             # mercurial specific, some extension api rely on this
             environ['HGUSER'] = username
 
         ip = dict_conf.get('ip')
         if ip:
             environ['REMOTE_HOST'] = ip
 
         if _is_request_chunked(environ):
             # set the compatibility flag for webob
             environ['wsgi.input_terminated'] = True
 
     def hg_proxy(self):
         @wsgiapp
         def _hg_proxy(environ, start_response):
             app = WsgiProxy(self.remote_wsgi.HgRemoteWsgi())
             return app(environ, start_response)
         return _hg_proxy
 
     def git_proxy(self):
         @wsgiapp
         def _git_proxy(environ, start_response):
             app = WsgiProxy(self.remote_wsgi.GitRemoteWsgi())
             return app(environ, start_response)
         return _git_proxy
 
     def hg_stream(self):
         if self._use_echo_app:
             @wsgiapp
             def _hg_stream(environ, start_response):
                 app = EchoApp('fake_path', 'fake_name', None)
                 return app(environ, start_response)
             return _hg_stream
         else:
             @wsgiapp
             def _hg_stream(environ, start_response):
                 log.debug('http-app: handling hg stream')
                 call_context = get_headers_call_context(environ)
 
                 repo_path = call_context['repo_path']
                 repo_name = call_context['repo_name']
                 config = call_context['repo_config']
 
                 app = scm_app.create_hg_wsgi_app(
                     repo_path, repo_name, config)
 
                 # Consistent path information for hgweb
                 environ['PATH_INFO'] = call_context['path_info']
                 environ['REPO_NAME'] = repo_name
                 self.set_env_from_config(environ, config)
 
                 log.debug('http-app: starting app handler '
                           'with %s and process request', app)
                 return app(environ, ResponseFilter(start_response))
             return _hg_stream
 
     def git_stream(self):
         if self._use_echo_app:
             @wsgiapp
             def _git_stream(environ, start_response):
                 app = EchoApp('fake_path', 'fake_name', None)
                 return app(environ, start_response)
             return _git_stream
         else:
             @wsgiapp
             def _git_stream(environ, start_response):
                 log.debug('http-app: handling git stream')
 
                 call_context = get_headers_call_context(environ)
 
                 repo_path = call_context['repo_path']
                 repo_name = call_context['repo_name']
                 config = call_context['repo_config']
 
                 environ['PATH_INFO'] = call_context['path_info']
                 self.set_env_from_config(environ, config)
 
                 content_type = environ.get('CONTENT_TYPE', '')
 
                 path = environ['PATH_INFO']
                 is_lfs_request = GIT_LFS_CONTENT_TYPE in content_type
                 log.debug(
                     'LFS: Detecting if request `%s` is LFS server path based '
                     'on content type:`%s`, is_lfs:%s',
                     path, content_type, is_lfs_request)
 
                 if not is_lfs_request:
                     # fallback detection by path
                     if GIT_LFS_PROTO_PAT.match(path):
                         is_lfs_request = True
                     log.debug(
                         'LFS: fallback detection by path of: `%s`, is_lfs:%s',
                         path, is_lfs_request)
 
                 if is_lfs_request:
                     app = scm_app.create_git_lfs_wsgi_app(
                         repo_path, repo_name, config)
                 else:
                     app = scm_app.create_git_wsgi_app(
                         repo_path, repo_name, config)
 
                 log.debug('http-app: starting app handler '
                           'with %s and process request', app)
 
                 return app(environ, start_response)
 
             return _git_stream
 
     def handle_vcs_exception(self, exception, request):
         _vcs_kind = getattr(exception, '_vcs_kind', '')
 
         if _vcs_kind == 'repo_locked':
             headers_call_context = get_headers_call_context(request.environ)
             status_code = safe_int(headers_call_context['locked_status_code'])
 
             return HTTPRepoLocked(
                 title=str(exception), status_code=status_code, headers=[('X-Rc-Locked', '1')])
 
         elif _vcs_kind == 'repo_branch_protected':
             # Get custom repo-branch-protected status code if present.
             return HTTPRepoBranchProtected(
                 title=str(exception), headers=[('X-Rc-Branch-Protection', '1')])
 
         exc_info = request.exc_info
         store_exception(id(exc_info), exc_info)
 
         traceback_info = 'unavailable'
         if request.exc_info:
             traceback_info = format_exc(request.exc_info)
 
         log.error(
             'error occurred handling this request for path: %s, \n%s',
             request.path, traceback_info)
 
         statsd = request.registry.statsd
         if statsd:
             exc_type = f"{exception.__class__.__module__}.{exception.__class__.__name__}"
             statsd.incr('vcsserver_exception_total',
                         tags=[f"type:{exc_type}"])
         raise exception
 
 
 class ResponseFilter:
 
     def __init__(self, start_response):
         self._start_response = start_response
 
     def __call__(self, status, response_headers, exc_info=None):
         headers = tuple(
             (h, v) for h, v in response_headers
             if not wsgiref.util.is_hop_by_hop(h))
         return self._start_response(status, headers, exc_info)
 
 
 def sanitize_settings_and_apply_defaults(global_config, settings):
     _global_settings_maker = SettingsMaker(global_config)
     settings_maker = SettingsMaker(settings)
 
     settings_maker.make_setting('logging.autoconfigure', False, parser='bool')
 
     logging_conf = os.path.join(os.path.dirname(global_config.get('__file__')), 'logging.ini')
     settings_maker.enable_logging(logging_conf)
 
     # Default includes, possible to change as a user
     pyramid_includes = settings_maker.make_setting('pyramid.includes', [], parser='list:newline')
     log.debug("Using the following pyramid.includes: %s", pyramid_includes)
 
     settings_maker.make_setting('__file__', global_config.get('__file__'))
 
     settings_maker.make_setting('pyramid.default_locale_name', 'en')
     settings_maker.make_setting('locale', 'en_US.UTF-8')
 
     settings_maker.make_setting(
         'core.binary_dir', '/usr/local/bin/rhodecode_bin/vcs_bin',
         default_when_empty=True, parser='string:noquote')
 
     temp_store = tempfile.gettempdir()
     default_cache_dir = os.path.join(temp_store, 'rc_cache')
     # save default, cache dir, and use it for all backends later.
     default_cache_dir = settings_maker.make_setting(
         'cache_dir',
         default=default_cache_dir, default_when_empty=True,
         parser='dir:ensured')
 
     # exception store cache
     settings_maker.make_setting(
         'exception_tracker.store_path',
         default=os.path.join(default_cache_dir, 'exc_store'), default_when_empty=True,
         parser='dir:ensured'
     )
 
     # repo_object cache defaults
     settings_maker.make_setting(
         'rc_cache.repo_object.backend',
         default='dogpile.cache.rc.file_namespace',
         parser='string')
     settings_maker.make_setting(
         'rc_cache.repo_object.expiration_time',
         default=30 * 24 * 60 * 60,  # 30days
         parser='int')
     settings_maker.make_setting(
         'rc_cache.repo_object.arguments.filename',
         default=os.path.join(default_cache_dir, 'vcsserver_cache_repo_object.db'),
         parser='string')
 
     # statsd
     settings_maker.make_setting('statsd.enabled', False, parser='bool')
     settings_maker.make_setting('statsd.statsd_host', 'statsd-exporter', parser='string')
     settings_maker.make_setting('statsd.statsd_port', 9125, parser='int')
     settings_maker.make_setting('statsd.statsd_prefix', '')
     settings_maker.make_setting('statsd.statsd_ipv6', False, parser='bool')
 
     settings_maker.env_expand()
 
 
 def main(global_config, **settings):
     start_time = time.time()
     log.info('Pyramid app config starting')
 
     if MercurialFactory:
         hgpatches.patch_largefiles_capabilities()
         hgpatches.patch_subrepo_type_mapping()
 
     # Fill in and sanitize the defaults & do ENV expansion
     sanitize_settings_and_apply_defaults(global_config, settings)
 
     # init and bootstrap StatsdClient
     StatsdClient.setup(settings)
 
     pyramid_app = HTTPApplication(settings=settings, global_config=global_config).wsgi_app()
     total_time = time.time() - start_time
     log.info('Pyramid app created and configured in %.2fs', total_time)
     return pyramid_app