From: Karthik U S Date: Sat, 16 May 2026 00:49:59 +0000 (+0530) Subject: mgr/mirroring: Add mirroring checkpoint CLIs X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=959bc2eb07cacf70a4da09170e1c9bbed4b9423f;p=ceph.git mgr/mirroring: Add mirroring checkpoint CLIs Add mgr checkpoint add/remove/ls/now commands that read and write checkpoint metadata on the primary filesystem via do_snap_md_op. Sample CLIs: ceph fs snapshot mirror checkpoint add ceph fs snapshot mirror checkpoint remove ceph fs snapshot mirror checkpoint ls ceph fs snapshot mirror checkpoint now Fixes: https://tracker.ceph.com/issues/73454 Signed-off-by: Karthik U S --- diff --git a/src/pybind/mgr/mirroring/fs/checkpoint.py b/src/pybind/mgr/mirroring/fs/checkpoint.py new file mode 100644 index 00000000000..87bef8851a1 --- /dev/null +++ b/src/pybind/mgr/mirroring/fs/checkpoint.py @@ -0,0 +1,147 @@ +import errno +import os +import time +from datetime import datetime + +import cephfs + +from .exception import MirrorException + +# Snapshot metadata keys for mirroring checkpoints (must match Checkpoint.h). +CHECKPOINT_STATUS_KEY = 'cephfs.mirror.checkpoint.status' +CHECKPOINT_CREATED_AT_KEY = 'cephfs.mirror.checkpoint.created_at' +CHECKPOINT_UPDATED_AT_KEY = 'cephfs.mirror.checkpoint.updated_at' +CHECKPOINT_ERROR_MSG_KEY = 'cephfs.mirror.checkpoint.error_msg' +CHECKPOINT_METADATA_KEYS = ( + CHECKPOINT_STATUS_KEY, + CHECKPOINT_CREATED_AT_KEY, + CHECKPOINT_UPDATED_AT_KEY, + CHECKPOINT_ERROR_MSG_KEY, +) +CHECKPOINT_STATUS_CREATED = 0 +CHECKPOINT_STATUS_COMPLETE = 1 +CHECKPOINT_STATUS_FAILED = 2 +CHECKPOINT_STATUS_NAMES = { + CHECKPOINT_STATUS_CREATED: 'created', + CHECKPOINT_STATUS_COMPLETE: 'complete', + CHECKPOINT_STATUS_FAILED: 'failed', +} + + +def get_checkpoint_epoch(): + """Get current time as UNIX epoch timestamp (9 decimal places, matches C++ %.9f).""" + return f'{time.time():.9f}' + + +def format_epoch_timestamp(epoch_str): + """Format UNIX epoch timestamp for display in ISO format with timezone.""" + try: + epoch = float(epoch_str) + dt = datetime.fromtimestamp(epoch).astimezone() + return dt.strftime('%Y-%m-%dT%H:%M:%S.%f%z') + except (ValueError, OSError): + return epoch_str + + +def snap_path(dir_path, snap_dir, snap_name): + return os.path.join(dir_path, snap_dir, snap_name) + + +def checkpoint_status_to_string(status_val): + try: + return CHECKPOINT_STATUS_NAMES[int(status_val)] + except (ValueError, KeyError): + return 'unknown' + + +def checkpoint_from_snap(snap_id, snap_name, metadata): + created_at = metadata.get(CHECKPOINT_CREATED_AT_KEY, '') + updated_at = metadata.get(CHECKPOINT_UPDATED_AT_KEY, '') + + cp = { + 'snap_id': snap_id, + 'snap_name': snap_name, + 'status': checkpoint_status_to_string( + metadata.get(CHECKPOINT_STATUS_KEY, CHECKPOINT_STATUS_CREATED)), + 'created_at': format_epoch_timestamp(created_at) if created_at else '', + 'updated_at': format_epoch_timestamp(updated_at) if updated_at else '', + } + if CHECKPOINT_ERROR_MSG_KEY in metadata: + cp['error_msg'] = metadata[CHECKPOINT_ERROR_MSG_KEY] + return cp + + +def is_checkpointed(metadata): + return CHECKPOINT_STATUS_KEY in metadata + + +class Checkpoint: + def __init__(self, get_snapdir): + self.get_snapdir = get_snapdir + + def list_directory_snapshots(self, fsh, dir_path): + snap_dir = self.get_snapdir() + snap_names = [] + try: + with fsh.opendir(os.path.join(dir_path, snap_dir)) as d_handle: + ent = fsh.readdir(d_handle) + while ent: + name = ent.d_name.decode('utf-8') + if name not in ('.', '..') and not name.startswith('_'): + snap_names.append(name) + ent = fsh.readdir(d_handle) + except cephfs.Error as e: + if e.errno == errno.ENOENT: + return [] + raise MirrorException(-e.errno, f'failed to list snapshots: {e}') + return snap_names + + def snap_info(self, fsh, dir_path, snap_name): + path = snap_path(dir_path, self.get_snapdir(), snap_name) + try: + return fsh.snap_info(path) + except cephfs.Error as e: + raise MirrorException(-e.errno, + f"snapshot '{snap_name}' not found under {dir_path}") + + def get_latest_snap(self, fsh, dir_path): + latest_id = 0 + latest_name = None + latest_info = None + + for snap_name in self.list_directory_snapshots(fsh, dir_path): + info = self.snap_info(fsh, dir_path, snap_name) + if info['id'] > latest_id: + latest_id = info['id'] + latest_name = snap_name + latest_info = info + + if latest_name is None: + raise MirrorException(-errno.ENOENT, f'no snapshots found under {dir_path}') + return latest_name, latest_info + + def write_metadata(self, fsh, dir_path, snap_name, snap_info): + if is_checkpointed(snap_info.get('metadata', {})): + raise MirrorException(-errno.EEXIST, 'checkpoint already exists for snapshot') + + path = snap_path(dir_path, self.get_snapdir(), snap_name) + now = get_checkpoint_epoch() + to_write = { + CHECKPOINT_STATUS_KEY: str(CHECKPOINT_STATUS_CREATED), + CHECKPOINT_CREATED_AT_KEY: now, + CHECKPOINT_UPDATED_AT_KEY: now, + } + + op = cephfs.CEPH_SNAP_MD_OP_CREATE | cephfs.CEPH_SNAP_MD_OP_EXCL + for key, val in to_write.items(): + fsh.do_snap_md_op(path, key, val, op) + + def remove_metadata(self, fsh, dir_path, snap_name, snap_info): + metadata = snap_info.get('metadata', {}) + if not is_checkpointed(metadata): + raise MirrorException(-errno.ENOENT, 'checkpoint not found for snapshot') + + path = snap_path(dir_path, self.get_snapdir(), snap_name) + for key in CHECKPOINT_METADATA_KEYS: + if key in metadata: + fsh.do_snap_md_op(path, key, '', cephfs.CEPH_SNAP_MD_OP_REMOVE) diff --git a/src/pybind/mgr/mirroring/fs/snapshot_mirror.py b/src/pybind/mgr/mirroring/fs/snapshot_mirror.py index 0d88613e2d6..82075612264 100644 --- a/src/pybind/mgr/mirroring/fs/snapshot_mirror.py +++ b/src/pybind/mgr/mirroring/fs/snapshot_mirror.py @@ -31,6 +31,11 @@ from .dir_map.load import load_dir_map, load_instances from .dir_map.update import UpdateDirMapRequest, UpdateInstanceRequest from .dir_map.policy import Policy from .dir_map.state_transition import ActionType +from .checkpoint import ( + Checkpoint, + checkpoint_from_snap, + is_checkpointed, +) log = logging.getLogger(__name__) @@ -304,6 +309,7 @@ class FSSnapshotMirror: self.lock = threading.Lock() self.refresh_pool_policy() self.local_fs = CephfsClient(mgr) + self.checkpoint = Checkpoint(self._client_snapdir) def notify(self, notify_type: NotifyType): log.debug(f'got notify type {notify_type}') @@ -992,3 +998,142 @@ class FSSnapshotMirror: return 0, json.dumps(daemons), '' except MirrorException as me: return me.args[0], '', me.args[1] + + def _validate_checkpoint_dir(self, fs_name, dir_path): + """Validate filesystem and mirrored directory; return normalized dir path.""" + if not self.filesystem_exist(fs_name): + raise MirrorException(-errno.ENOENT, f'filesystem {fs_name} does not exist') + + fspolicy = self.pool_policy.get(fs_name, None) + if not fspolicy: + raise MirrorException(-errno.EINVAL, f'filesystem {fs_name} is not mirrored') + + dir_path = norm_path(dir_path) + if not dir_path: + raise MirrorException(-errno.EINVAL, 'directory path is required') + + lookup_info = fspolicy.policy.lookup(dir_path) + if not lookup_info: + raise MirrorException(-errno.ENOENT, f'directory {dir_path} is not tracked') + + if lookup_info['purging']: + raise MirrorException(-errno.EINVAL, f'directory {dir_path} is under removal') + + return dir_path + + def _client_snapdir(self): + return self.mgr.get_foreign_ceph_option('client', 'client_snapdir') + + def checkpoint_add(self, fs_name, dir_path, snap_name): + """Add a checkpoint for a snapshot via snapshot metadata on the primary filesystem.""" + try: + with self.lock: + dir_path = self._validate_checkpoint_dir(fs_name, dir_path) + + with open_filesystem(self.local_fs, fs_name) as fsh: + info = self.checkpoint.snap_info(fsh, dir_path, snap_name) + self.checkpoint.write_metadata(fsh, dir_path, snap_name, info) + + result = { + 'status': 'success', + 'message': f'checkpoint added for snapshot {snap_name}', + 'dir_root': dir_path, + 'snap_id': info['id'], + 'snap_name': snap_name, + 'checkpoint_status': 'created', + } + return 0, json.dumps(result), '' + except MirrorException as me: + return me.args[0], '', me.args[1] + except cephfs.Error as e: + return -e.errno, '', f'failed to add checkpoint: {e}' + except Exception as e: + log.error(f'failed to add checkpoint: {e}') + return -errno.EINVAL, '', f'failed to add checkpoint: {str(e)}' + + def checkpoint_remove(self, fs_name, dir_path, snap_name): + """Remove a checkpoint from a snapshot via snapshot metadata on the primary filesystem.""" + try: + with self.lock: + dir_path = self._validate_checkpoint_dir(fs_name, dir_path) + + with open_filesystem(self.local_fs, fs_name) as fsh: + info = self.checkpoint.snap_info(fsh, dir_path, snap_name) + self.checkpoint.remove_metadata(fsh, dir_path, snap_name, info) + + result = { + 'status': 'success', + 'message': f'checkpoint removed for snapshot {snap_name}', + 'dir_root': dir_path, + 'snap_name': snap_name, + } + return 0, json.dumps(result), '' + except MirrorException as me: + return me.args[0], '', me.args[1] + except cephfs.Error as e: + return -e.errno, '', f'failed to remove checkpoint: {e}' + except Exception as e: + log.error(f'failed to remove checkpoint: {e}') + return -errno.EINVAL, '', f'failed to remove checkpoint: {str(e)}' + + def checkpoint_ls(self, fs_name, dir_path, format='json'): + """List all checkpoints for a directory from snapshot metadata on the primary filesystem.""" + try: + with self.lock: + dir_path = self._validate_checkpoint_dir(fs_name, dir_path) + + checkpoints = [] + with open_filesystem(self.local_fs, fs_name) as fsh: + for snap_name in self.checkpoint.list_directory_snapshots(fsh, dir_path): + try: + info = self.checkpoint.snap_info(fsh, dir_path, snap_name) + except MirrorException as me: + log.warning( + f'failed to get snapshot info for {snap_name!r} ' + f'under {dir_path}: {me.args[1]}') + continue + md = info.get('metadata', {}) + if is_checkpointed(md): + checkpoints.append( + checkpoint_from_snap(info['id'], snap_name, md)) + + checkpoints.sort(key=lambda cp: cp['snap_id']) + result = {'dir_root': dir_path, 'checkpoints': checkpoints} + + if format == 'json-pretty': + return 0, json.dumps(result, indent=2), '' + return 0, json.dumps(result), '' + except MirrorException as me: + return me.args[0], '', me.args[1] + except cephfs.Error as e: + return -e.errno, '', f'failed to list checkpoints: {e}' + except Exception as e: + log.error(f'failed to list checkpoints: {e}') + return -errno.EINVAL, '', f'failed to list checkpoints: {str(e)}' + + def checkpoint_now(self, fs_name, dir_path): + """Create a checkpoint on the latest snapshot via snapshot metadata.""" + try: + with self.lock: + dir_path = self._validate_checkpoint_dir(fs_name, dir_path) + + with open_filesystem(self.local_fs, fs_name) as fsh: + snap_name, info = self.checkpoint.get_latest_snap(fsh, dir_path) + self.checkpoint.write_metadata(fsh, dir_path, snap_name, info) + + result = { + 'status': 'success', + 'message': 'checkpoint created on latest snapshot', + 'dir_root': dir_path, + 'snap_id': info['id'], + 'snap_name': snap_name, + 'checkpoint_status': 'created', + } + return 0, json.dumps(result), '' + except MirrorException as me: + return me.args[0], '', me.args[1] + except cephfs.Error as e: + return -e.errno, '', f'failed to create checkpoint: {e}' + except Exception as e: + log.error(f'failed to create checkpoint: {e}') + return -errno.EINVAL, '', f'failed to create checkpoint: {str(e)}' diff --git a/src/pybind/mgr/mirroring/module.py b/src/pybind/mgr/mirroring/module.py index 3fbe145864e..829d713436b 100644 --- a/src/pybind/mgr/mirroring/module.py +++ b/src/pybind/mgr/mirroring/module.py @@ -141,3 +141,34 @@ class Module(MgrModule): """Get snapshot mirror metrics for a filesystem (optional mirrored directory and peer)""" return self.fs_snapshot_mirror.metrics_status(fs_name, mirrored_dir_path, peer_uuid) + + @MirroringCLICommand.Write('fs snapshot mirror checkpoint add') + def snapshot_mirror_checkpoint_add(self, + fs_name: str, + path: str, + snap_name: str): + """Add a checkpoint for a snapshot""" + return self.fs_snapshot_mirror.checkpoint_add(fs_name, path, snap_name) + + @MirroringCLICommand.Write('fs snapshot mirror checkpoint remove') + def snapshot_mirror_checkpoint_remove(self, + fs_name: str, + path: str, + snap_name: str): + """Remove a checkpoint from a snapshot""" + return self.fs_snapshot_mirror.checkpoint_remove(fs_name, path, snap_name) + + @MirroringCLICommand.Read('fs snapshot mirror checkpoint ls') + def snapshot_mirror_checkpoint_ls(self, + fs_name: str, + path: str, + format: str = 'json'): + """List all checkpoints for a directory""" + return self.fs_snapshot_mirror.checkpoint_ls(fs_name, path, format) + + @MirroringCLICommand.Write('fs snapshot mirror checkpoint now') + def snapshot_mirror_checkpoint_now(self, + fs_name: str, + path: str): + """Create a checkpoint on the latest snapshot""" + return self.fs_snapshot_mirror.checkpoint_now(fs_name, path)