From 19fcec84d8d7d21e796c7624e521b60d28ee21ed Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Sun, 7 Apr 2024 20:45:59 +0200 Subject: Adding upstream version 16.2.11+ds. Signed-off-by: Daniel Baumann --- .../mgr/volumes/fs/operations/clone_index.py | 98 ++++++++++++++++++++++ 1 file changed, 98 insertions(+) create mode 100644 src/pybind/mgr/volumes/fs/operations/clone_index.py (limited to 'src/pybind/mgr/volumes/fs/operations/clone_index.py') diff --git a/src/pybind/mgr/volumes/fs/operations/clone_index.py b/src/pybind/mgr/volumes/fs/operations/clone_index.py new file mode 100644 index 000000000..a2b31f858 --- /dev/null +++ b/src/pybind/mgr/volumes/fs/operations/clone_index.py @@ -0,0 +1,98 @@ +import os +import uuid +import stat +import errno +import logging +from contextlib import contextmanager + +import cephfs + +from .index import Index +from ..exception import IndexException, VolumeException +from ..fs_util import list_one_entry_at_a_time + +log = logging.getLogger(__name__) + +class CloneIndex(Index): + SUB_GROUP_NAME = "clone" + PATH_MAX = 4096 + + @property + def path(self): + return os.path.join(super(CloneIndex, self).path, CloneIndex.SUB_GROUP_NAME.encode('utf-8')) + + def _track(self, sink_path): + tracking_id = str(uuid.uuid4()) + source_path = os.path.join(self.path, tracking_id.encode('utf-8')) + log.info("tracking-id {0} for path {1}".format(tracking_id, sink_path)) + + self.fs.symlink(sink_path, source_path) + return tracking_id + + def track(self, sink_path): + try: + return self._track(sink_path) + except (VolumeException, cephfs.Error) as e: + if isinstance(e, cephfs.Error): + e = IndexException(-e.args[0], e.args[1]) + elif isinstance(e, VolumeException): + e = IndexException(e.errno, e.error_str) + raise e + + def untrack(self, tracking_id): + log.info("untracking {0}".format(tracking_id)) + source_path = os.path.join(self.path, tracking_id.encode('utf-8')) + try: + self.fs.unlink(source_path) + except cephfs.Error as e: + raise IndexException(-e.args[0], e.args[1]) + + def get_oldest_clone_entry(self, exclude=[]): + min_ctime_entry = None + exclude_tracking_ids = [v[0] for v in exclude] + log.debug("excluded tracking ids: {0}".format(exclude_tracking_ids)) + for entry in list_one_entry_at_a_time(self.fs, self.path): + dname = entry.d_name + dpath = os.path.join(self.path, dname) + st = self.fs.lstat(dpath) + if dname not in exclude_tracking_ids and stat.S_ISLNK(st.st_mode): + if min_ctime_entry is None or st.st_ctime < min_ctime_entry[1].st_ctime: + min_ctime_entry = (dname, st) + if min_ctime_entry: + try: + linklen = min_ctime_entry[1].st_size + sink_path = self.fs.readlink(os.path.join(self.path, min_ctime_entry[0]), CloneIndex.PATH_MAX) + return (min_ctime_entry[0], sink_path[:linklen]) + except cephfs.Error as e: + raise IndexException(-e.args[0], e.args[1]) + return None + + def find_clone_entry_index(self, sink_path): + try: + for entry in list_one_entry_at_a_time(self.fs, self.path): + dname = entry.d_name + dpath = os.path.join(self.path, dname) + st = self.fs.lstat(dpath) + if stat.S_ISLNK(st.st_mode): + target_path = self.fs.readlink(dpath, CloneIndex.PATH_MAX) + if sink_path == target_path[:st.st_size]: + return dname + return None + except cephfs.Error as e: + raise IndexException(-e.args[0], e.args[1]) + +def create_clone_index(fs, vol_spec): + clone_index = CloneIndex(fs, vol_spec) + try: + fs.mkdirs(clone_index.path, 0o700) + except cephfs.Error as e: + raise IndexException(-e.args[0], e.args[1]) + +@contextmanager +def open_clone_index(fs, vol_spec): + clone_index = CloneIndex(fs, vol_spec) + try: + fs.stat(clone_index.path) + except cephfs.Error as e: + raise IndexException(-e.args[0], e.args[1]) + yield clone_index -- cgit v1.2.3