archive-cache: synced with CE codebase

vcsserver/lib/archive_cache/backends/base.py

@@ -1,352 +1,355 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()
+    shard_cls = BaseShard
 
     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_shard(self, key) -> shard_cls:
+        index = self._hash(key) % self._shard_count
+        shard = self._shards[index]
+        return shard
 
     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) -> dict:
         """
         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 in order 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.
         """
         removal_info = {
             "removed_items": 0,
             "removed_size": 0
         }
         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 removal_info
 
         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
         removal_info['removed_items'] = removed_items
         removal_info['removed_size'] = removed_size
         log.debug('Removed %s cache archives, and reduced size by: %s',
                   removed_items, format_size(removed_size))
         return removal_info
 
     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

@@ -1,167 +1,174 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):
+    def __init__(self, index, directory, directory_folder, fs, **settings):
         self._index = index
         self._directory = directory
+        self._directory_folder = directory_folder
         self.storage_type = 'directory'
-        self.fs = fsspec.filesystem('file')
+
+        self.fs = fs
 
     @property
     def directory(self):
-        """Cache directory."""
-        return self._directory
+        """Cache directory final path."""
+        return os.path.join(self._directory, self._directory_folder)
 
     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):
     shard_name = 'shard_%03d'
+    shard_cls = FileSystemShard
 
     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
+        self._storage_path = directory  # common path for all from BaseCache
 
-        # 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._shard_count = int(self.get_conf('archive_cache.filesystem.cache_shards', pop=True))
+        if self._shard_count < 1:
+            raise ValueError('cache_shards must be 1 or more')
 
         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)
+        log.debug('Initializing %s archival cache instance under %s', self)
+        fs = fsspec.filesystem('file')
+        # check if it's ok to write, and re-create the archive cache main dir
+        # A directory is the virtual equivalent of a physical file cabinet.
+        # In other words, it's a container for organizing digital data.
+        # Unlike a folder, which can only store files, a directory can store files,
+        # subdirectories, and other directories.
+        if not fs.exists(self._directory):
+            fs.makedirs(self._directory, exist_ok=True)
+
         self._shards = tuple(
-            FileSystemShard(
+            self.shard_cls(
                 index=num,
-                directory=os.path.join(directory, self.shard_name % num),
+                directory=directory,
+                directory_folder=self.shard_name % num,
+                fs=fs,
                 **settings,
             )
-            for num in range(self._count)
+            for num in range(self._shard_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

@@ -1,158 +1,164 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):
+    def __init__(self, index, bucket, bucket_folder, fs, **settings):
         self._index = index
-        self._bucket = bucket
+        self._bucket_folder = bucket_folder
         self.storage_type = 'bucket'
+        self._bucket_main = bucket
 
-        endpoint_url = settings.pop('archive_cache.objectstore.url')
-        key = settings.pop('archive_cache.objectstore.key')
-        secret = settings.pop('archive_cache.objectstore.secret')
-
-        # TODO: Add it all over the place...
-        self._bucket_root = settings.pop('archive_cache.objectstore.bucket_root')
-
-        self.fs = fsspec.filesystem('s3', anon=False, endpoint_url=endpoint_url, key=key, secret=secret)
+        self.fs = fs
 
     @property
     def bucket(self):
-        """Cache bucket."""
-        return os.path.join(self._bucket_root, self._bucket)
+        """Cache bucket final path."""
+        return os.path.join(self._bucket_main, self._bucket_folder)
 
     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):
-        if self._bucket_root:
-            if not self.fs.exists(self._bucket_root):
-                self.fs.mkdir(self._bucket_root)
 
-        # ensure bucket exists
+        # ensure folder in 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):
-    shard_name = 'shard-bucket-%03d'
+    shard_name = 'shard-%03d'
+    shard_cls = S3Shard
 
     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._storage_path = objectstore_url  # common path for all from BaseCache
 
-        self._count = int(self.get_conf('archive_cache.objectstore.bucket_shards', pop=True))
+        self._shard_count = int(self.get_conf('archive_cache.objectstore.bucket_shards', pop=True))
+        if self._shard_count < 1:
+            raise ValueError('cache_shards must be 1 or more')
+
+        self._bucket = settings.pop('archive_cache.objectstore.bucket')
+        if not self._bucket:
+            raise ValueError('archive_cache.objectstore.bucket needs to have a value')
 
         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)
+        endpoint_url = settings.pop('archive_cache.objectstore.url')
+        key = settings.pop('archive_cache.objectstore.key')
+        secret = settings.pop('archive_cache.objectstore.secret')
+
+        log.debug('Initializing %s archival cache instance under %s', self)
+
+        fs = fsspec.filesystem('s3', anon=False, endpoint_url=endpoint_url, key=key, secret=secret)
+
+        # init main bucket
+        if not fs.exists(self._bucket):
+            fs.mkdir(self._bucket)
+
         self._shards = tuple(
-            S3Shard(
+            self.shard_cls(
                 index=num,
-                bucket=self.shard_name % num,
+                bucket=self._bucket,
+                bucket_folder=self.shard_name % num,
+                fs=fs,
                 **settings,
             )
-            for num in range(self._count)
+            for num in range(self._shard_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']