From: Karthik U S Date: Wed, 10 Jun 2026 05:48:25 +0000 (+0530) Subject: mgr/mirroring,tools/cephfs_mirror: Handle checkpoint state transition X-Git-Tag: testing/wip-khiremat-testing-20260701.115536^2~2 X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=7a80895db88144f7b3bc9dbcd6035b5f5b3a712f;p=ceph-ci.git mgr/mirroring,tools/cephfs_mirror: Handle checkpoint state transition When a new checkpoint is being added or when the daemon gets restarted, it will check whether the newly created checkpoint or any other old checkpoints have already been mirrored onto the remote peer. If so, it will transition to the correct state by checking for the highest snap id present on the remote, and setting all the checkpoints which have snap id lesser than or equal to that of the remote to COMPLETE. This is done by sending an acquire notification from the mirroring module to the mirror daemon, which is handled in the add_directory path, by adding the directory to be checked for the state transition in the tick thread. This path gets triggered: a) when the daemon gets restarted b) when the peer mapping changes c) by sending the acquire notification from checkpoint add/now CLIs. d) when mirroring module restarts Fixes: https://tracker.ceph.com/issues/73454 Signed-off-by: Karthik U S --- diff --git a/src/pybind/mgr/mirroring/fs/dir_map/policy.py b/src/pybind/mgr/mirroring/fs/dir_map/policy.py index da61bec940b..c9c111cacef 100644 --- a/src/pybind/mgr/mirroring/fs/dir_map/policy.py +++ b/src/pybind/mgr/mirroring/fs/dir_map/policy.py @@ -102,7 +102,8 @@ class Policy: if dir_state: return {'instance_id': dir_state.instance_id, 'mapped_time': dir_state.mapped_time, - 'purging': dir_state.purging} + 'purging': dir_state.purging, + 'state': dir_state.state} return None def get_tracked_instance_id(self, dir_path): diff --git a/src/pybind/mgr/mirroring/fs/snapshot_mirror.py b/src/pybind/mgr/mirroring/fs/snapshot_mirror.py index 82075612264..e7b0ccba79a 100644 --- a/src/pybind/mgr/mirroring/fs/snapshot_mirror.py +++ b/src/pybind/mgr/mirroring/fs/snapshot_mirror.py @@ -30,7 +30,7 @@ from .dir_map.create import create_mirror_object 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 .dir_map.state_transition import ActionType, State from .checkpoint import ( Checkpoint, checkpoint_from_snap, @@ -193,6 +193,17 @@ class FSPolicy: finally: self.op_tracker.finish_async_op() + def handle_checkpoint_acquire_ack(self, dir_path, r): + """Ack for checkpoint refresh acquire; does not advance the policy state machine.""" + log.debug(f'handle_checkpoint_acquire_ack: {dir_path} r={r}') + with self.lock: + try: + if self.stopping.is_set(): + log.debug('handle_checkpoint_acquire_ack: policy shutting down') + return + finally: + self.op_tracker.finish_async_op() + def process_updates(self): def acquire_message(dir_path): return json.dumps({'dir_path': dir_path, @@ -1024,6 +1035,40 @@ class FSSnapshotMirror: def _client_snapdir(self): return self.mgr.get_foreign_ceph_option('client', 'client_snapdir') + def _send_acquire_notification(self, fs_name, dir_path): + """Send acquire notification directly to daemon for a directory.""" + try: + with self.lock: + fspolicy = self.pool_policy.get(fs_name, None) + if not fspolicy: + log.warning(f'filesystem {fs_name} is not mirrored') + return + + with fspolicy.lock: + lookup_info = fspolicy.policy.lookup(dir_path) + if not lookup_info: + log.warning(f'directory {dir_path} not found in policy map') + return + + if lookup_info.get('state') != State.ASSOCIATED: + log.debug(f'directory {dir_path} is not associated yet ' + f'(state={lookup_info.get("state")}), skipping acquire notification') + return + + instance_id = lookup_info['instance_id'] + if not instance_id: + log.warning(f'directory {dir_path} not mapped to any instance yet') + return + + acquire_msg = json.dumps({'dir_path': dir_path, 'mode': 'acquire'}) + + log.debug(f'sending acquire notification for {dir_path} to instance {instance_id}') + fspolicy.op_tracker.start_async_op() + fspolicy.notifier.notify(dir_path, (instance_id, acquire_msg), + fspolicy.handle_checkpoint_acquire_ack) + except Exception as e: + log.error(f'failed to send acquire notification for {dir_path}: {e}') + def checkpoint_add(self, fs_name, dir_path, snap_name): """Add a checkpoint for a snapshot via snapshot metadata on the primary filesystem.""" try: @@ -1034,6 +1079,9 @@ class FSSnapshotMirror: info = self.checkpoint.snap_info(fsh, dir_path, snap_name) self.checkpoint.write_metadata(fsh, dir_path, snap_name, info) + # Send acquire notification to trigger checkpoint state initialization + self._send_acquire_notification(fs_name, dir_path) + result = { 'status': 'success', 'message': f'checkpoint added for snapshot {snap_name}', @@ -1121,6 +1169,9 @@ class FSSnapshotMirror: snap_name, info = self.checkpoint.get_latest_snap(fsh, dir_path) self.checkpoint.write_metadata(fsh, dir_path, snap_name, info) + # Send acquire notification to trigger checkpoint state initialization + self._send_acquire_notification(fs_name, dir_path) + result = { 'status': 'success', 'message': 'checkpoint created on latest snapshot', diff --git a/src/tools/cephfs_mirror/PeerReplayer.cc b/src/tools/cephfs_mirror/PeerReplayer.cc index 202f9b1f61f..8cb8dbdd183 100644 --- a/src/tools/cephfs_mirror/PeerReplayer.cc +++ b/src/tools/cephfs_mirror/PeerReplayer.cc @@ -47,6 +47,7 @@ enum { l_cephfs_mirror_peer_replayer_last_synced_end, l_cephfs_mirror_peer_replayer_last_synced_duration, l_cephfs_mirror_peer_replayer_last_synced_bytes, + l_cephfs_mirror_peer_replayer_add_directory, l_cephfs_mirror_peer_replayer_last, }; @@ -289,6 +290,8 @@ PeerReplayer::PeerReplayer(CephContext *cct, FSMirror *fs_mirror, "last_synced_duration", "Last Synced Duration", "lsdn", prio); plb.add_u64_counter(l_cephfs_mirror_peer_replayer_last_synced_bytes, "last_synced_bytes", "Last Synced Bytes", "lsbt", prio); + plb.add_u64_counter(l_cephfs_mirror_peer_replayer_add_directory, + "add_directory", "Add Directory Calls", "adir", prio); m_perf_counters = plb.create_perf_counters(); m_cct->get_perfcounters_collection()->add(m_perf_counters); } @@ -724,10 +727,15 @@ void PeerReplayer::shutdown() { void PeerReplayer::add_directory(string_view dir_root) { dout(20) << ": dir_root=" << dir_root << dendl; + if (m_perf_counters) { + m_perf_counters->inc(l_cephfs_mirror_peer_replayer_add_directory); + } auto _dir_root = std::string(dir_root); { std::scoped_lock locker(m_lock); + m_checkpoint_init_pending.insert(_dir_root); + if (std::find(m_directories.begin(), m_directories.end(), _dir_root) != m_directories.end()) { dout(10) << ": dir_root=" << _dir_root << " already in replay list" << dendl; @@ -766,6 +774,7 @@ void PeerReplayer::remove_directory(string_view dir_root, bool purging) { if (it1 == m_registered.end()) { if (purging) { remove_persisted_dir_sync_stat(_dir_root); + m_checkpoint_init_pending.erase(_dir_root); } remove_directory_perf_counters(_dir_root); m_snap_sync_stats.erase(_dir_root); @@ -1421,6 +1430,82 @@ void PeerReplayer::checkpoint_sync_complete(const std::string &dir_root, << " snap_name=" << snap_name << dendl; } +void PeerReplayer::initialize_checkpoints(const std::string &dir_root) { + dout(10) << ": dir_root=" << dir_root << dendl; + + // Build local snapshot map + std::map local_snap_map; + int r = build_snap_map(dir_root, &local_snap_map, false); + if (r < 0) { + derr << ": failed to build local snap map for dir_root=" << dir_root + << ": " << cpp_strerror(r) << dendl; + return; + } + if (local_snap_map.empty()) { + dout(10) << ": no local snapshots for dir_root=" << dir_root << dendl; + return; + } + + // Build remote snapshot map + std::map remote_snap_map; + r = build_snap_map(dir_root, &remote_snap_map, true); + if (r < 0) { + derr << ": failed to build remote snap map for dir_root=" << dir_root + << ": " << cpp_strerror(r) << dendl; + return; + } + if (remote_snap_map.empty()) { + dout(10) << ": no remote snapshots for dir_root=" << dir_root << dendl; + return; + } + + // Get the highest snap_id from remote + uint64_t remote_highest_snap_id = remote_snap_map.rbegin()->first; + + dout(10) << ": remote_highest_snap_id=" << remote_highest_snap_id << dendl; + + // Iterate through local snapshots and mark checkpoints as COMPLETE + // if their snap_id is <= remote_highest_snap_id + for (const auto &[snap_id, snap_name] : local_snap_map) { + // Read snapshot metadata + auto snap_path = snapshot_path(m_cct, dir_root, snap_name); + std::map snap_metadata; + r = read_snap_metadata(m_local_mount, snap_path, &snap_metadata); + if (r < 0) { + derr << ": failed to read snap metadata for snap_id=" << snap_id + << " snap_name=" << snap_name << ": " << cpp_strerror(r) << dendl; + continue; + } + + if (!has_checkpoint(snap_metadata)) { + continue; + } + + CheckpointInfo info = read_checkpoint_metadata(snap_id, snap_name, snap_metadata); + // Update checkpoints that are in CREATED or FAILED status and have snap_id <= remote_highest_snap_id + if ((info.status == CheckpointStatus::CREATED || info.status == CheckpointStatus::FAILED) + && snap_id <= remote_highest_snap_id) { + dout(10) << ": marking checkpoint as COMPLETE for snap_id=" << snap_id + << " snap_name=" << snap_name << " (previous status: " + << checkpoint_status_to_string(info.status) << ")" << dendl; + + info.status = CheckpointStatus::COMPLETE; + info.updated_at = ceph_clock_now(); + info.error_msg.clear(); // Clear any previous error message + + r = write_checkpoint_metadata(m_cct, m_local_mount, dir_root, snap_name, snap_metadata, info); + if (r < 0) { + derr << ": failed to update checkpoint status for snap_id=" << snap_id + << " snap_name=" << snap_name << " dir_root=" << dir_root + << ": " << cpp_strerror(r) << dendl; + } else { + dout(10) << ": successfully marked checkpoint as COMPLETE for snap_id=" << snap_id + << " snap_name=" << snap_name << dendl; + } + } + } +} + void PeerReplayer::checkpoint_sync_failed(const std::string &dir_root, uint64_t snap_id, const std::string &snap_name, @@ -3232,6 +3317,10 @@ void PeerReplayer::run_tick() { refresh_directory_current_sync_perf_counters(kv.first); } + std::vector checkpoint_dirs( + m_checkpoint_init_pending.begin(), m_checkpoint_init_pending.end()); + m_checkpoint_init_pending.clear(); + // persist sync stats to omap for registered directories std::vector dirs; dirs.reserve(m_registered.size()); @@ -3242,6 +3331,9 @@ void PeerReplayer::run_tick() { for (const auto &dir_root : dirs) { persist_dir_sync_stat(dir_root); } + for (const auto &dir_root : checkpoint_dirs) { + initialize_checkpoints(dir_root); + } locker.lock(); } } diff --git a/src/tools/cephfs_mirror/PeerReplayer.h b/src/tools/cephfs_mirror/PeerReplayer.h index 4cf99201bb4..0a12fe87d25 100644 --- a/src/tools/cephfs_mirror/PeerReplayer.h +++ b/src/tools/cephfs_mirror/PeerReplayer.h @@ -692,6 +692,7 @@ private: std::vector m_directories; std::map m_snap_sync_stats; std::set m_purging_directories; + std::set m_checkpoint_init_pending; MountRef m_local_mount; ServiceDaemon *m_service_daemon; PeerReplayerAdminSocketHook *m_asok_hook = nullptr; @@ -767,6 +768,7 @@ private: int build_snap_map(const std::string &dir_root, std::map *snap_map, bool is_remote=false); + void initialize_checkpoints(const std::string &dir_root); void checkpoint_sync_complete(const std::string &dir_root, uint64_t synced_snap_id, const std::string &snap_name); void checkpoint_sync_failed(const std::string &dir_root, uint64_t snap_id,