tools/cephfs_mirror: Remove sync-stat omap metrics on peer remove and mirror disable

Remove persisted sync-stat omap entries when a peer is removed or
mirroring is disabled for a filesystem, since those peer UUIDs are not
reused. Preserve omap on internal blocklist/failure restarts so sync
can resume where it left off.

Also update the relevant documentation

Fixes: https://tracker.ceph.com/issues/78464
Signed-off-by: Kotresh HR <khiremat@redhat.com>

doc/dev/cephfs-mirroring.rst

@@ -358,11 +358,14 @@ Each omap value includes metadata fields written by ``cephfs-mirror``:
   persist cadence and Manager caching (``snapshot_mirror_metrics_cache_enabled``
   and ``snapshot_mirror_metrics_cache_ttl``).
 
-Omap entries are removed when a directory is removed from mirroring. All metric
-fields are written to omap; on daemon restart only ``last_synced_snap`` metadata
-is loaded back. Per-session counters (``snaps_synced``, ``snaps_deleted``,
-``snaps_renamed``) are persisted but not loaded and therefore start at zero each
-session.
+Omap entries for a directory are removed when that directory is removed from
+mirroring. Entries for a peer (all of its directories) are removed when the peer
+is removed, and all entries for a file system are removed when mirroring is
+disabled for that file system. Internal blocklist/failure restarts of a mirror
+instance preserve omap so sync can resume. All metric fields are written to
+omap; on daemon restart only ``last_synced_snap`` metadata is loaded back.
+Per-session counters (``snaps_synced``, ``snaps_deleted``, ``snaps_renamed``)
+are persisted but not loaded and therefore start at zero each session.
 
 See :ref:`Directory snapshot sync metrics<cephfs_mirroring_mgr_snapshot_status>`
 and :ref:`Snapshot sync metric fields<cephfs_mirroring_sync_metric_fields>` in

src/tools/cephfs_mirror/FSMirror.cc

@@ -217,12 +217,13 @@ void FSMirror::init(Context *on_finish) {
   init_instance_watcher(on_finish);
 }
 
-void FSMirror::shutdown(Context *on_finish) {
+void FSMirror::shutdown(Context *on_finish, bool purge_persisted_sync_stats) {
   dout(20) << dendl;
 
   {
     std::scoped_lock locker(m_lock);
     m_stopping = true;
+    m_purge_persisted_sync_stats_on_shutdown = purge_persisted_sync_stats;
     if (m_on_init_finish != nullptr) {
       dout(10) << ": delaying shutdown -- init in progress" << dendl;
       m_on_shutdown_finish = new LambdaContext([this, on_finish](int r) {
@@ -245,12 +246,22 @@ void FSMirror::shutdown(Context *on_finish) {
 void FSMirror::shutdown_peer_replayers() {
   dout(20) << dendl;
 
+  bool purge_persisted_sync_stats = false;
+  {
+    std::scoped_lock locker(m_lock);
+    purge_persisted_sync_stats = m_purge_persisted_sync_stats_on_shutdown;
+  }
+
   for (auto &[peer, peer_replayer] : m_peer_replayers) {
     dout(5) << ": shutting down replayer for peer=" << peer << dendl;
     shutdown_replayer(peer_replayer.get());
   }
   m_peer_replayers.clear();
 
+  if (purge_persisted_sync_stats) {
+    remove_persisted_sync_stats_by_prefix(PeerReplayer::sync_stat_omap_prefix(m_filesystem));
+  }
+
   shutdown_mirror_watcher();
 }
 
@@ -411,6 +422,53 @@ void FSMirror::handle_release_directory(string_view dir_path, bool purging) {
   }
 }
 
+void FSMirror::remove_persisted_sync_stats_by_prefix(std::string_view prefix) {
+  std::string prefix_str(prefix);
+  dout(5) << ": removing persisted sync stats with prefix=" << prefix_str << dendl;
+
+  // PeerReplayer submits persist/remove via a shallow copy of m_ioctx.
+  // Join only stops producer threads; flush outstanding AIOs so a late
+  // omap_set cannot recreate keys after the purge below.
+  int r = m_ioctx.aio_flush();
+  if (r < 0) {
+    derr << ": failed to flush outstanding sync-stat AIOs: "
+         << cpp_strerror(r) << dendl;
+  }
+
+  constexpr uint64_t max_return = 256;
+  std::string start_after;
+  bool more = true;
+
+  while (more) {
+    std::map<std::string, bufferlist> vals;
+    r = m_ioctx.omap_get_vals2(CEPHFS_MIRROR_OBJECT, start_after, prefix_str,
+                               max_return, &vals, &more);
+    if (r == -ENOENT) {
+      return;
+    }
+    if (r < 0) {
+      derr << ": failed to list sync stat omap keys: " << cpp_strerror(r) << dendl;
+      return;
+    }
+    if (vals.empty()) {
+      return;
+    }
+
+    std::set<std::string> keys;
+    for (const auto &[key, _] : vals) {
+      keys.insert(key);
+    }
+
+    r = m_ioctx.omap_rm_keys(CEPHFS_MIRROR_OBJECT, keys);
+    if (r < 0) {
+      derr << ": failed to remove sync stat omap keys: " << cpp_strerror(r) << dendl;
+      return;
+    }
+
+    start_after = vals.rbegin()->first;
+  }
+}
+
 void FSMirror::add_peer(const Peer &peer) {
   dout(10) << ": peer=" << peer << dendl;
 
@@ -452,9 +510,10 @@ void FSMirror::remove_peer(const Peer &peer) {
   }
 
   if (replayer) {
-    dout(5) << ": shutting down replayers for peer=" << peer << dendl;
+    dout(5) << ": shutting down replayer for peer=" << peer << dendl;
     shutdown_replayer(replayer.get());
   }
+  remove_persisted_sync_stats_by_prefix(PeerReplayer::sync_stat_omap_prefix(m_filesystem, peer));
   if (m_perf_counters) {
     m_perf_counters->dec(l_cephfs_mirror_fs_mirror_peers);
   }

src/tools/cephfs_mirror/FSMirror.h

@@ -30,7 +30,7 @@ public:
   ~FSMirror();
 
   void init(Context *on_finish);
-  void shutdown(Context *on_finish);
+  void shutdown(Context *on_finish, bool purge_persisted_sync_stats=false);
 
   void add_peer(const Peer &peer);
   void remove_peer(const Peer &peer);
@@ -160,6 +160,7 @@ private:
 
   int m_retval = 0;
   bool m_stopping = false;
+  bool m_purge_persisted_sync_stats_on_shutdown = false;
   bool m_init_failed = false;
   Context *m_on_init_finish = nullptr;
   Context *m_on_shutdown_finish = nullptr;
@@ -190,6 +191,8 @@ private:
 
   void handle_acquire_directory(std::string_view dir_path);
   void handle_release_directory(std::string_view dir_path, bool purging);
+
+  void remove_persisted_sync_stats_by_prefix(std::string_view prefix);
 };
 
 } // namespace mirror

src/tools/cephfs_mirror/Mirror.cc

@@ -112,7 +112,7 @@ struct Mirror::C_DisableMirroring : Context {
   void disable_mirroring() {
     Context *ctx = new C_CallbackAdapter<C_DisableMirroring,
                                          &C_DisableMirroring::handle_disable_mirroring>(this);
-    mirror->disable_mirroring(filesystem, ctx);
+    mirror->disable_mirroring(filesystem, ctx, true);
   }
 
   void handle_disable_mirroring(int r) {
@@ -471,7 +471,8 @@ void Mirror::handle_disable_mirroring(const Filesystem &filesystem, int r) {
   }
 }
 
-void Mirror::disable_mirroring(const Filesystem &filesystem, Context *on_finish) {
+void Mirror::disable_mirroring(const Filesystem &filesystem, Context *on_finish,
+                               bool purge_persisted_sync_stats) {
   ceph_assert(ceph_mutex_is_locked(m_lock));
 
   auto &mirror_action = m_mirror_actions.at(filesystem);
@@ -485,7 +486,9 @@ void Mirror::disable_mirroring(const Filesystem &filesystem, Context *on_finish)
   }
 
   mirror_action.action_in_progress = true;
-  mirror_action.fs_mirror->shutdown(new C_AsyncCallback<ContextWQ>(m_work_queue, on_finish));
+  mirror_action.fs_mirror->shutdown(
+    new C_AsyncCallback<ContextWQ>(m_work_queue, on_finish),
+    purge_persisted_sync_stats);
 }
 
 void Mirror::mirroring_disabled(const Filesystem &filesystem) {

src/tools/cephfs_mirror/Mirror.h

@@ -121,7 +121,8 @@ private:
   void handle_enable_mirroring(const Filesystem &filesystem, const Peers &peers, int r);
 
   // mirror disable callback
-  void disable_mirroring(const Filesystem &filesystem, Context *on_finish);
+  void disable_mirroring(const Filesystem &filesystem, Context *on_finish,
+                         bool purge_persisted_sync_stats=false);
   void handle_disable_mirroring(const Filesystem &filesystem, int r);
 
   // peer update callback

src/tools/cephfs_mirror/PeerReplayer.cc

@@ -2,6 +2,7 @@
 // vim: ts=8 sw=2 sts=2 expandtab
 
 #include <stack>
+#include <vector>
 #include <fcntl.h>
 #include <algorithm>
 #include <sys/time.h>
@@ -785,14 +786,22 @@ void PeerReplayer::remove_directory(string_view dir_root, bool purging) {
   }
 }
 
+std::string PeerReplayer::sync_stat_omap_prefix(const Filesystem &filesystem) {
+  return CEPHFS_MIRROR_SYNC_STAT_OMAP_PREFIX + "/" + filesystem.fs_name + "/";
+}
+
+std::string PeerReplayer::sync_stat_omap_prefix(const Filesystem &filesystem,
+                                                const Peer &peer) {
+  return sync_stat_omap_prefix(filesystem) + peer.uuid + "/";
+}
+
 std::string PeerReplayer::peer_sync_stat_omap_key(std::string_view dir_root) const {
   // dir_root is usually absolute (e.g. "/d0"); avoid ".../uuid//d0" from an extra slash.
   std::string d(dir_root);
   while (!d.empty() && d.front() == '/') {
     d.erase(0, 1);
   }
-  return PEER_SYNC_STAT_KEY_PREFIX + "/" + m_filesystem.fs_name + "/" + m_peer.uuid
-         + "/" + d;
+  return sync_stat_omap_prefix(m_filesystem, m_peer) + d;
 }
 
 void PeerReplayer::apply_persisted_dir_sync_stat(SnapSyncStat &sync_stat,

src/tools/cephfs_mirror/PeerReplayer.h

@@ -51,6 +51,10 @@ public:
   // remove a directory from queue
   void remove_directory(std::string_view dir_root, bool purging = false);
 
+  static std::string sync_stat_omap_prefix(const Filesystem &filesystem);
+  static std::string sync_stat_omap_prefix(const Filesystem &filesystem,
+                                           const Peer &peer);
+
   // admin socket helpers
   void peer_status(Formatter *f);
 
@@ -62,7 +66,6 @@ private:
 
   inline static const std::string SERVICE_DAEMON_FAILED_DIR_COUNT_KEY = "failure_count";
   inline static const std::string SERVICE_DAEMON_RECOVERED_DIR_COUNT_KEY = "recovery_count";
-  inline static const std::string PEER_SYNC_STAT_KEY_PREFIX = "sync_stat";
 
   using Snapshot = std::pair<std::string, uint64_t>;
 

src/tools/cephfs_mirror/Types.h

@@ -19,6 +19,7 @@ namespace cephfs {
 namespace mirror {
 
 static const std::string CEPHFS_MIRROR_OBJECT("cephfs_mirror");
+static const std::string CEPHFS_MIRROR_SYNC_STAT_OMAP_PREFIX("sync_stat");
 
 typedef std::variant<bool, uint64_t, std::string> AttributeValue;
 typedef std::map<std::string, AttributeValue> Attributes;