From 158884820ee2ab7982ae4a75f571730ba5c3b439 Mon Sep 17 00:00:00 2001
From: Venky Shankar <vshankar@redhat.com>
Date: Mon, 8 Mar 2021 04:48:56 -0500
Subject: [PATCH] cephfs-mirror: restart failed mirror filesystem instances

Signed-off-by: Venky Shankar <vshankar@redhat.com>
---
 src/common/options.cc                      |  6 +++++
 src/tools/cephfs_mirror/FSMirror.h         |  9 ++++++-
 src/tools/cephfs_mirror/InstanceWatcher.cc |  2 ++
 src/tools/cephfs_mirror/InstanceWatcher.h  |  6 +++++
 src/tools/cephfs_mirror/Mirror.cc          | 31 +++++++++++++++++-----
 src/tools/cephfs_mirror/Mirror.h           |  2 ++
 src/tools/cephfs_mirror/MirrorWatcher.cc   |  2 ++
 src/tools/cephfs_mirror/MirrorWatcher.h    |  6 +++++
 8 files changed, 56 insertions(+), 8 deletions(-)

diff --git a/src/common/options.cc b/src/common/options.cc
index 8673abc4cf05d..783caf2f112cc 100644
--- a/src/common/options.cc
+++ b/src/common/options.cc
@@ -9069,6 +9069,12 @@ std::vector<Option> get_cephfs_mirror_options() {
     .set_min(1)
     .set_description("failed directory retry interval for synchronization")
     .set_long_description("interval in seconds to retry synchronization for failed directories."),
+
+    Option("cephfs_mirror_restart_mirror_on_failure_interval", Option::TYPE_SECS, Option::LEVEL_ADVANCED)
+    .set_default(20)
+    .set_min(0)
+    .set_description("interval to restart failed mirror instances")
+    .set_long_description("Interval in seconds to restart failed mirror instances. Setting to zero (0) disables restarting failed mirror instances."),
     });
 }
 
diff --git a/src/tools/cephfs_mirror/FSMirror.h b/src/tools/cephfs_mirror/FSMirror.h
index 53af92e5a69b8..e598cd12d5dc5 100644
--- a/src/tools/cephfs_mirror/FSMirror.h
+++ b/src/tools/cephfs_mirror/FSMirror.h
@@ -40,11 +40,18 @@ public:
     return m_stopping;
   }
 
-  bool is_failed() {
+  bool is_init_failed() {
     std::scoped_lock locker(m_lock);
     return m_init_failed;
   }
 
+  bool is_failed() {
+    std::scoped_lock locker(m_lock);
+    return m_init_failed ||
+           m_instance_watcher->is_failed() ||
+           m_mirror_watcher->is_failed();
+  }
+
   bool is_blocklisted() {
     std::scoped_lock locker(m_lock);
     return is_blocklisted(locker);
diff --git a/src/tools/cephfs_mirror/InstanceWatcher.cc b/src/tools/cephfs_mirror/InstanceWatcher.cc
index 5775e572082cb..9c357da318548 100644
--- a/src/tools/cephfs_mirror/InstanceWatcher.cc
+++ b/src/tools/cephfs_mirror/InstanceWatcher.cc
@@ -116,8 +116,10 @@ void InstanceWatcher::handle_rewatch_complete(int r) {
     m_blocklisted = true;
   } else if (r == -ENOENT) {
     derr << ": mirroring object deleted" << dendl;
+    m_failed = true;
   } else if (r < 0) {
     derr << ": rewatch error: " << cpp_strerror(r) << dendl;
+    m_failed = true;
   }
 }
 
diff --git a/src/tools/cephfs_mirror/InstanceWatcher.h b/src/tools/cephfs_mirror/InstanceWatcher.h
index 8d12ce9344468..06edf5da97a27 100644
--- a/src/tools/cephfs_mirror/InstanceWatcher.h
+++ b/src/tools/cephfs_mirror/InstanceWatcher.h
@@ -49,6 +49,11 @@ public:
     return m_blocklisted;
   }
 
+  bool is_failed() {
+    std::scoped_lock locker(m_lock);
+    return m_failed;
+  }
+
 private:
   librados::IoCtx &m_ioctx;
   Listener &m_listener;
@@ -59,6 +64,7 @@ private:
   Context *m_on_shutdown_finish = nullptr;
 
   bool m_blocklisted = false;
+  bool m_failed = false;
 
   void create_instance();
   void handle_create_instance(int r);
diff --git a/src/tools/cephfs_mirror/Mirror.cc b/src/tools/cephfs_mirror/Mirror.cc
index 6d28263adec27..9389c77f1cf82 100644
--- a/src/tools/cephfs_mirror/Mirror.cc
+++ b/src/tools/cephfs_mirror/Mirror.cc
@@ -196,6 +196,7 @@ Mirror::Mirror(CephContext *cct, const std::vector<const char*> &args,
     m_msgr(msgr),
     m_listener(this),
     m_last_blocklist_check(ceph_clock_now()),
+    m_last_failure_check(ceph_clock_now()),
     m_local(new librados::Rados()) {
   auto thread_pool = &(cct->lookup_or_create_singleton_object<ThreadPoolSingleton>(
                          "cephfs::mirror::thread_pool", false, cct));
@@ -349,7 +350,6 @@ void Mirror::enable_mirroring(const Filesystem &filesystem, uint64_t local_pool_
 
   auto &mirror_action = m_mirror_actions.at(filesystem);
   if (is_restart) {
-    ceph_assert(mirror_action.action_in_progress);
     mirror_action.fs_mirror.reset();
   } else {
     ceph_assert(!mirror_action.action_in_progress);
@@ -384,7 +384,7 @@ void Mirror::handle_disable_mirroring(const Filesystem &filesystem, int r) {
   std::scoped_lock locker(m_lock);
   auto &mirror_action = m_mirror_actions.at(filesystem);
 
-  if (!mirror_action.fs_mirror->is_failed()) {
+  if (!mirror_action.fs_mirror->is_init_failed()) {
     ceph_assert(mirror_action.action_in_progress);
     mirror_action.action_in_progress = false;
     m_cond.notify_all();
@@ -406,7 +406,7 @@ void Mirror::disable_mirroring(const Filesystem &filesystem, Context *on_finish)
   ceph_assert(mirror_action.fs_mirror);
   ceph_assert(!mirror_action.action_in_progress);
 
-  if (mirror_action.fs_mirror->is_failed()) {
+  if (mirror_action.fs_mirror->is_init_failed()) {
     dout(10) << ": init failed for filesystem=" << filesystem << dendl;
     m_work_queue->queue(on_finish, -EINVAL);
     return;
@@ -478,16 +478,28 @@ void Mirror::peer_removed(const Filesystem &filesystem, const Peer &peer) {
 void Mirror::update_fs_mirrors() {
   dout(20) << dendl;
 
+  auto now = ceph_clock_now();
   double blocklist_interval = g_ceph_context->_conf.get_val<std::chrono::seconds>
     ("cephfs_mirror_restart_mirror_on_blocklist_interval").count();
-  auto now = ceph_clock_now();
   bool check_blocklist = blocklist_interval > 0 && ((now - m_last_blocklist_check) >= blocklist_interval);
 
+  double failed_interval = g_ceph_context->_conf.get_val<std::chrono::seconds>
+    ("cephfs_mirror_restart_mirror_on_failure_interval").count();
+  bool check_failure = failed_interval > 0 && ((now - m_last_failure_check) >= failed_interval);
+
   {
     std::scoped_lock locker(m_lock);
     for (auto &[filesystem, mirror_action] : m_mirror_actions) {
-      if (check_blocklist && !mirror_action.action_in_progress
-          && mirror_action.fs_mirror && mirror_action.fs_mirror->is_blocklisted()) {
+      if (check_failure && !mirror_action.action_in_progress &&
+          mirror_action.fs_mirror && mirror_action.fs_mirror->is_failed()) {
+        // about to restart failed mirror instance -- nothing
+        // should interfere
+        dout(5) << ": filesystem=" << filesystem << " failed mirroring -- restarting" << dendl;
+        auto peers = mirror_action.fs_mirror->get_peers();
+        mirror_action.action_ctxs.push_front(
+          new C_RestartMirroring(this, filesystem, mirror_action.pool_id, peers));
+      } else if (check_blocklist && !mirror_action.action_in_progress &&
+                 mirror_action.fs_mirror && mirror_action.fs_mirror->is_blocklisted()) {
         // about to restart blocklisted mirror instance -- nothing
         // should interfere
         dout(5) << ": filesystem=" << filesystem << " is blocklisted -- restarting" << dendl;
@@ -505,6 +517,9 @@ void Mirror::update_fs_mirrors() {
     if (check_blocklist) {
       m_last_blocklist_check = now;
     }
+    if (check_failure) {
+      m_last_failure_check = now;
+    }
   }
 
   schedule_mirror_update_task();
@@ -553,12 +568,14 @@ void Mirror::run() {
                 {return !mirror_action.action_in_progress;});
     if (mirror_action.fs_mirror &&
         !mirror_action.fs_mirror->is_stopping() &&
-        !mirror_action.fs_mirror->is_failed()) {
+        !mirror_action.fs_mirror->is_init_failed()) {
       C_SaferCond cond;
       mirror_action.fs_mirror->shutdown(new C_AsyncCallback<ContextWQ>(m_work_queue, &cond));
       int r = cond.wait();
       dout(10) << ": shutdown filesystem=" << filesystem << ", r=" << r << dendl;
     }
+
+    mirror_action.fs_mirror.reset();
   }
 }
 
diff --git a/src/tools/cephfs_mirror/Mirror.h b/src/tools/cephfs_mirror/Mirror.h
index 3d6dde7301055..2e50a2a8bdbdc 100644
--- a/src/tools/cephfs_mirror/Mirror.h
+++ b/src/tools/cephfs_mirror/Mirror.h
@@ -101,6 +101,8 @@ private:
   std::map<Filesystem, MirrorAction> m_mirror_actions;
 
   utime_t m_last_blocklist_check;
+  utime_t m_last_failure_check;
+
   RadosRef m_local;
   std::unique_ptr<ServiceDaemon> m_service_daemon;
 
diff --git a/src/tools/cephfs_mirror/MirrorWatcher.cc b/src/tools/cephfs_mirror/MirrorWatcher.cc
index f3e4e377b2c3a..26b88d077e598 100644
--- a/src/tools/cephfs_mirror/MirrorWatcher.cc
+++ b/src/tools/cephfs_mirror/MirrorWatcher.cc
@@ -95,8 +95,10 @@ void MirrorWatcher::handle_rewatch_complete(int r) {
     m_blocklisted = true;
   } else if (r == -ENOENT) {
     derr << ": mirroring object deleted" << dendl;
+    m_failed = true;
   } else if (r < 0) {
     derr << ": rewatch error: " << cpp_strerror(r) << dendl;
+    m_failed = true;
   }
 }
 
diff --git a/src/tools/cephfs_mirror/MirrorWatcher.h b/src/tools/cephfs_mirror/MirrorWatcher.h
index 1d6cbef013e1d..c4d4f452294d8 100644
--- a/src/tools/cephfs_mirror/MirrorWatcher.h
+++ b/src/tools/cephfs_mirror/MirrorWatcher.h
@@ -47,6 +47,11 @@ public:
     return m_blocklisted;
   }
 
+  bool is_failed() {
+    std::scoped_lock locker(m_lock);
+    return m_failed;
+  }
+
 private:
   librados::IoCtx &m_ioctx;
   FSMirror *m_fs_mirror;
@@ -59,6 +64,7 @@ private:
   Context *m_on_shutdown_finish = nullptr;
 
   bool m_blocklisted = false;
+  bool m_failed = false;
 
   void register_watcher();
   void handle_register_watcher(int r);
-- 
2.39.5