|
|
# Copyright (C) 2016-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 errno
|
|
|
import os
|
|
|
import hashlib
|
|
|
import functools
|
|
|
import time
|
|
|
import logging
|
|
|
|
|
|
from .. import config_keys
|
|
|
from ..exceptions import FileOverSizeException
|
|
|
from ..backends.base import BaseFileStoreBackend, fsspec, BaseShard, ShardFileReader
|
|
|
|
|
|
from ....lib.ext_json import json
|
|
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
|
|
|
class LegacyFileSystemShard(BaseShard):
|
|
|
# legacy ver
|
|
|
METADATA_VER = 'v2'
|
|
|
BACKEND_TYPE = config_keys.backend_legacy_filesystem
|
|
|
storage_type: str = 'dir_struct'
|
|
|
|
|
|
# legacy suffix
|
|
|
metadata_suffix: str = '.meta'
|
|
|
|
|
|
@classmethod
|
|
|
def _sub_store_from_filename(cls, filename):
|
|
|
return filename[:2]
|
|
|
|
|
|
@classmethod
|
|
|
def apply_counter(cls, counter, filename):
|
|
|
name_counted = '%d-%s' % (counter, filename)
|
|
|
return name_counted
|
|
|
|
|
|
@classmethod
|
|
|
def safe_make_dirs(cls, dir_path):
|
|
|
if not os.path.exists(dir_path):
|
|
|
try:
|
|
|
os.makedirs(dir_path)
|
|
|
except OSError as e:
|
|
|
if e.errno != errno.EEXIST:
|
|
|
raise
|
|
|
return
|
|
|
|
|
|
@classmethod
|
|
|
def resolve_name(cls, name, directory):
|
|
|
"""
|
|
|
Resolves a unique name and the correct path. If a filename
|
|
|
for that path already exists then a numeric prefix with values > 0 will be
|
|
|
added, for example test.jpg -> 1-test.jpg etc. initially file would have 0 prefix.
|
|
|
|
|
|
:param name: base name of file
|
|
|
:param directory: absolute directory path
|
|
|
"""
|
|
|
|
|
|
counter = 0
|
|
|
while True:
|
|
|
name_counted = cls.apply_counter(counter, name)
|
|
|
|
|
|
# sub_store prefix to optimize disk usage, e.g some_path/ab/final_file
|
|
|
sub_store: str = cls._sub_store_from_filename(name_counted)
|
|
|
sub_store_path: str = os.path.join(directory, sub_store)
|
|
|
cls.safe_make_dirs(sub_store_path)
|
|
|
|
|
|
path = os.path.join(sub_store_path, name_counted)
|
|
|
if not os.path.exists(path):
|
|
|
return name_counted, path
|
|
|
counter += 1
|
|
|
|
|
|
def __init__(self, index, directory, directory_folder, fs, **settings):
|
|
|
self._index: int = index
|
|
|
self._directory: str = directory
|
|
|
self._directory_folder: str = directory_folder
|
|
|
self.fs = fs
|
|
|
|
|
|
@property
|
|
|
def dir_struct(self) -> str:
|
|
|
"""Cache directory final path."""
|
|
|
return os.path.join(self._directory, '0-')
|
|
|
|
|
|
def _write_file(self, full_path, iterator, max_filesize, mode='wb'):
|
|
|
|
|
|
# ensure dir exists
|
|
|
destination, _ = os.path.split(full_path)
|
|
|
if not self.fs.exists(destination):
|
|
|
self.fs.makedirs(destination)
|
|
|
|
|
|
writer = self.fs.open(full_path, mode)
|
|
|
|
|
|
digest = hashlib.sha256()
|
|
|
oversize_cleanup = False
|
|
|
with writer:
|
|
|
size = 0
|
|
|
for chunk in iterator:
|
|
|
size += len(chunk)
|
|
|
digest.update(chunk)
|
|
|
writer.write(chunk)
|
|
|
|
|
|
if max_filesize and size > max_filesize:
|
|
|
# free up the copied file, and raise exc
|
|
|
oversize_cleanup = True
|
|
|
break
|
|
|
|
|
|
writer.flush()
|
|
|
# Get the file descriptor
|
|
|
fd = writer.fileno()
|
|
|
|
|
|
# Sync the file descriptor to disk, helps with NFS cases...
|
|
|
os.fsync(fd)
|
|
|
|
|
|
if oversize_cleanup:
|
|
|
self.fs.rm(full_path)
|
|
|
raise FileOverSizeException(f'given file is over size limit ({max_filesize}): {full_path}')
|
|
|
|
|
|
sha256 = digest.hexdigest()
|
|
|
log.debug('written new artifact under %s, sha256: %s', full_path, sha256)
|
|
|
return size, sha256
|
|
|
|
|
|
def _store(self, key: str, uid_key, value_reader, max_filesize: int | None = None, metadata: dict | None = None, **kwargs):
|
|
|
|
|
|
filename = key
|
|
|
uid_filename = uid_key
|
|
|
|
|
|
# NOTE:, also apply N- Counter...
|
|
|
uid_filename, full_path = self.resolve_name(uid_filename, self._directory)
|
|
|
|
|
|
# STORE METADATA
|
|
|
# TODO: make it compatible, and backward proof
|
|
|
_metadata = {
|
|
|
"version": self.METADATA_VER,
|
|
|
|
|
|
"filename": filename,
|
|
|
"filename_uid_path": full_path,
|
|
|
"filename_uid": uid_filename,
|
|
|
"sha256": "", # NOTE: filled in by reader iteration
|
|
|
|
|
|
"store_time": time.time(),
|
|
|
|
|
|
"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, max_filesize)
|
|
|
_metadata['size'] = size
|
|
|
_metadata['sha256'] = sha256
|
|
|
|
|
|
# after storing the artifacts, we write the metadata present
|
|
|
_metadata_file, metadata_file_path = self.get_metadata_filename(uid_filename)
|
|
|
|
|
|
with self.fs.open(metadata_file_path, 'wb') as f:
|
|
|
f.write(json.dumps(_metadata))
|
|
|
|
|
|
return uid_filename, _metadata
|
|
|
|
|
|
def store_path(self, uid_filename):
|
|
|
"""
|
|
|
Returns absolute file path of the uid_filename
|
|
|
"""
|
|
|
prefix_dir = ''
|
|
|
if '/' in uid_filename:
|
|
|
prefix_dir, filename = uid_filename.split('/')
|
|
|
sub_store = self._sub_store_from_filename(filename)
|
|
|
else:
|
|
|
sub_store = self._sub_store_from_filename(uid_filename)
|
|
|
|
|
|
return os.path.join(self._directory, prefix_dir, sub_store, uid_filename)
|
|
|
|
|
|
def metadata_convert(self, uid_filename, metadata):
|
|
|
# NOTE: backward compat mode here... this is for file created PRE 5.2 system
|
|
|
if 'meta_ver' in metadata:
|
|
|
full_path = self.store_path(uid_filename)
|
|
|
metadata = {
|
|
|
"_converted": True,
|
|
|
"_org": metadata,
|
|
|
"version": self.METADATA_VER,
|
|
|
"store_type": self.BACKEND_TYPE,
|
|
|
|
|
|
"filename": metadata['filename'],
|
|
|
"filename_uid_path": full_path,
|
|
|
"filename_uid": uid_filename,
|
|
|
"sha256": metadata['sha256'],
|
|
|
|
|
|
"store_time": metadata['time'],
|
|
|
|
|
|
"size": metadata['size']
|
|
|
}
|
|
|
return metadata
|
|
|
|
|
|
def _fetch(self, key, presigned_url_expires: int = 0):
|
|
|
if key not in self:
|
|
|
log.exception(f'requested key={key} not found in {self}')
|
|
|
raise KeyError(key)
|
|
|
|
|
|
metadata = self.get_metadata(key)
|
|
|
|
|
|
file_path = metadata['filename_uid_path']
|
|
|
if presigned_url_expires and presigned_url_expires > 0:
|
|
|
metadata['url'] = self.fs.url(file_path, expires=presigned_url_expires)
|
|
|
|
|
|
return ShardFileReader(self.fs.open(file_path, 'rb')), metadata
|
|
|
|
|
|
def delete(self, key):
|
|
|
return self._delete(key)
|
|
|
|
|
|
def _delete(self, key):
|
|
|
if key not in self:
|
|
|
log.exception(f'requested key={key} not found in {self}')
|
|
|
raise KeyError(key)
|
|
|
|
|
|
metadata = self.get_metadata(key)
|
|
|
metadata_file, metadata_file_path = self.get_metadata_filename(key)
|
|
|
artifact_file_path = metadata['filename_uid_path']
|
|
|
self.fs.rm(artifact_file_path)
|
|
|
self.fs.rm(metadata_file_path)
|
|
|
|
|
|
def get_metadata_filename(self, uid_filename) -> tuple[str, str]:
|
|
|
|
|
|
metadata_file: str = f'{uid_filename}{self.metadata_suffix}'
|
|
|
uid_path_in_store = self.store_path(uid_filename)
|
|
|
|
|
|
metadata_file_path = f'{uid_path_in_store}{self.metadata_suffix}'
|
|
|
return metadata_file, metadata_file_path
|
|
|
|
|
|
|
|
|
class LegacyFileSystemBackend(BaseFileStoreBackend):
|
|
|
_shard_cls = LegacyFileSystemShard
|
|
|
|
|
|
def __init__(self, settings):
|
|
|
super().__init__(settings)
|
|
|
|
|
|
store_dir = self.get_conf(config_keys.legacy_filesystem_storage_path)
|
|
|
directory = os.path.expanduser(store_dir)
|
|
|
|
|
|
self._directory = directory
|
|
|
self._storage_path = directory # common path for all from BaseCache
|
|
|
|
|
|
log.debug('Initializing %s file_store instance', self)
|
|
|
fs = fsspec.filesystem('file')
|
|
|
|
|
|
if not fs.exists(self._directory):
|
|
|
fs.makedirs(self._directory, exist_ok=True)
|
|
|
|
|
|
# legacy system uses single shard
|
|
|
self._shards = tuple(
|
|
|
[
|
|
|
self._shard_cls(
|
|
|
index=0,
|
|
|
directory=directory,
|
|
|
directory_folder='',
|
|
|
fs=fs,
|
|
|
**settings,
|
|
|
)
|
|
|
]
|
|
|
)
|
|
|
|
|
|
@classmethod
|
|
|
def get_shard_index(cls, filename: str, num_shards) -> int:
|
|
|
# legacy filesystem doesn't use shards, and always uses single shard
|
|
|
return 0
|
|
|
|
