--- /dev/null
+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)
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__)
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}')
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)}'
"""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)
projects / ceph.git / commitdiff