Merge 35a74f4d80 into 8482cb11bc

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

qa/tasks/cephfs/test_mirroring.py

@@ -264,7 +264,7 @@ class TestMirroring(CephFSTestCase):
             vafter = res[TestMirroring.PERF_COUNTER_KEY_NAME_CEPHFS_MIRROR_FS][0]
             self.assertGreater(vafter["counters"]["mirroring_peers"], vbefore["counters"]["mirroring_peers"])
 
-    def peer_remove(self, fs_name, fs_id, peer_spec):
+    def peer_remove(self, fs_name, fs_id, peer_spec, verify_dircount=True):
         res = self.mirror_daemon_command(f'counter dump for fs: {fs_name}', 'counter', 'dump')
         vbefore = res[TestMirroring.PERF_COUNTER_KEY_NAME_CEPHFS_MIRROR_FS][0]
 
@@ -272,9 +272,10 @@ class TestMirroring(CephFSTestCase):
         self.run_ceph_cmd("fs", "snapshot", "mirror", "peer_remove", fs_name, peer_uuid)
         time.sleep(10)
         # verify via asok
-        res = self.mirror_daemon_command(f'mirror status for fs: {fs_name}',
-                                         'fs', 'mirror', 'status', f'{fs_name}@{fs_id}')
-        self.assertTrue(res['peers'] == {} and res['snap_dirs']['dir_count'] == 0)
+        if verify_dircount:
+            res = self.mirror_daemon_command(f'mirror status for fs: {fs_name}',
+                                             'fs', 'mirror', 'status', f'{fs_name}@{fs_id}')
+            self.assertTrue(res['peers'] == {} and res['snap_dirs']['dir_count'] == 0)
 
         res = self.mirror_daemon_command(f'counter dump for fs: {fs_name}', 'counter', 'dump')
         vafter = res[TestMirroring.PERF_COUNTER_KEY_NAME_CEPHFS_MIRROR_FS][0]
@@ -812,6 +813,56 @@ class TestMirroring(CephFSTestCase):
                 return peer_uuid
         return None
 
+    def sync_stat_omap_key(self, fs_name, peer_uuid, dir_path):
+        dir_rel = dir_path.lstrip('/')
+        return f'sync_stat/{fs_name}/{peer_uuid}/{dir_rel}'
+
+    def list_sync_stat_omap_keys(self, fs_name, peer_uuid=None):
+        p = self.mount_a.client_remote.run(
+            args=['rados', '-p', self.fs.metadata_pool_name,
+                  'listomapvals', 'cephfs_mirror'],
+            stdout=StringIO(), stderr=StringIO(), timeout=30,
+            check_status=True, label='list sync stat omap keys')
+        p.wait()
+        prefix = f'sync_stat/{fs_name}/'
+        if peer_uuid:
+            prefix = f'{prefix}{peer_uuid}/'
+        keys = []
+        for line in p.stdout.getvalue().splitlines():
+            stripped = line.strip()
+            if stripped.startswith(prefix):
+                keys.append(stripped)
+        return keys
+
+    @retry_assert(timeout=60, interval=2)
+    def wait_sync_stat_omap_key(self, fs_name, peer_uuid, dir_path):
+        expected = self.sync_stat_omap_key(fs_name, peer_uuid, dir_path)
+        keys = self.list_sync_stat_omap_keys(fs_name, peer_uuid)
+        self.assertIn(expected, keys, msg=f'expected omap key {expected}, got {keys}')
+
+    @retry_assert(timeout=60, interval=2)
+    def assert_sync_stat_omap_keys_removed(self, fs_name, peer_uuid=None):
+        keys = self.list_sync_stat_omap_keys(fs_name, peer_uuid)
+        self.assertEqual(keys, [], msg=f'stale sync stat omap keys: {keys}')
+
+    def setup_sync_stat_omap(self, dir_name='sync_stat_omap_dir'):
+        self.setup_mount_b(mds_perm='rw')
+        self.enable_mirroring(self.primary_fs_name, self.primary_fs_id)
+        peer_spec = "client.mirror_remote@ceph"
+        self.peer_add(self.primary_fs_name, self.primary_fs_id, peer_spec,
+                      self.secondary_fs_name)
+        self.mount_a.run_shell(['mkdir', dir_name])
+        self.mount_a.create_n_files(f'{dir_name}/file', 10, sync=True)
+        self.add_directory(self.primary_fs_name, self.primary_fs_id, f'/{dir_name}')
+        snap_name = 'snap0'
+        self.mount_a.run_shell(['mkdir', f'{dir_name}/.snap/{snap_name}'])
+        self.check_peer_status_idle(self.primary_fs_name, self.primary_fs_id,
+                                    peer_spec, f'/{dir_name}', snap_name, 1)
+        peer_uuid = self.get_peer_uuid(peer_spec)
+        self.wait_sync_stat_omap_key(
+            self.primary_fs_name, peer_uuid, f'/{dir_name}')
+        return peer_spec, peer_uuid, f'/{dir_name}'
+
     def get_daemon_admin_socket(self):
         """overloaded by teuthology override (fs/mirror/clients/mirror.yaml)"""
         return "/var/run/ceph/cephfs-mirror.asok"
@@ -3064,6 +3115,43 @@ class TestMirroring(CephFSTestCase):
         self.assertEqual(res, {'metrics': {}})
         self.disable_mirroring(self.primary_fs_name, self.primary_fs_id)
 
+    def test_cephfs_mirror_sync_stat_omap_removed_on_peer_remove(self):
+        """peer_remove purges persisted sync-stat omap entries for the peer."""
+        peer_spec, peer_uuid, _dir_path = self.setup_sync_stat_omap(
+            dir_name='sync_stat_omap_peer_remove')
+        self.peer_remove(self.primary_fs_name, self.primary_fs_id, peer_spec, False)
+        self.assert_sync_stat_omap_keys_removed(
+            self.primary_fs_name, peer_uuid)
+        self.disable_mirroring(self.primary_fs_name, self.primary_fs_id)
+
+    def test_cephfs_mirror_sync_stat_omap_removed_on_disable(self):
+        """mirror disable purges persisted sync-stat omap entries."""
+        _peer_spec, peer_uuid, _dir_path = self.setup_sync_stat_omap(
+            dir_name='sync_stat_omap_disable')
+        keys_before = self.list_sync_stat_omap_keys(
+            self.primary_fs_name, peer_uuid)
+        self.assertTrue(keys_before)
+        self.disable_mirroring(self.primary_fs_name, self.primary_fs_id)
+        self.assert_sync_stat_omap_keys_removed(self.primary_fs_name)
+
+    def test_cephfs_mirror_sync_stat_omap_preserved_on_restart(self):
+        """Daemon restart preserves persisted sync-stat omap entries."""
+        peer_spec, peer_uuid, dir_path = self.setup_sync_stat_omap(
+            dir_name='sync_stat_omap_restart')
+        keys_before = self.list_sync_stat_omap_keys(
+            self.primary_fs_name, peer_uuid)
+        self.assertTrue(keys_before)
+        self.restart_mirror_daemon()
+        self.wait_sync_stat_omap_key(
+            self.primary_fs_name, peer_uuid, dir_path)
+        keys_after = self.list_sync_stat_omap_keys(
+            self.primary_fs_name, peer_uuid)
+        self.assertEqual(set(keys_before), set(keys_after))
+        self.remove_directory(self.primary_fs_name, self.primary_fs_id,
+                              dir_path)
+        self.peer_remove(self.primary_fs_name, self.primary_fs_id, peer_spec)
+        self.disable_mirroring(self.primary_fs_name, self.primary_fs_id)
+
     def test_mgr_snapshot_mirror_status_errors(self):
         """Mgr status returns expected errors for invalid inputs."""
         self.setup_mount_b(mds_perm='rw')

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;