rbd-mirror: basic integration with sync throttling

snapshot-based mirroring did not have any throttling to prevent
too many concurrent syncs from running. Since each sync might need
to iterate over every object of an image, that could potentially
put an extreme burden on the remote cluster.

A future PR will add a more intelligent throttle based on the actual
number of objects needed to be scanned.

Signed-off-by: Jason Dillaman <dillaman@redhat.com>

diff --git a/src/test/rbd_mirror/image_replayer/snapshot/test_mock_Replayer.cc b/src/test/rbd_mirror/image_replayer/snapshot/test_mock_Replayer.cc
index 391b4000a6c7034351b964b2e95b99d67a2bdd7c..b2e95bad097e146e0d99d4d3d39afa0774638c38 100644 (file)
--- a/src/test/rbd_mirror/image_replayer/snapshot/test_mock_Replayer.cc
+++ b/src/test/rbd_mirror/image_replayer/snapshot/test_mock_Replayer.cc
@@ -221,6 +221,10 @@ namespace mirror {
 
 template <>
 struct InstanceWatcher<librbd::MockTestImageCtx> {
+  MOCK_METHOD1(cancel_sync_request, void(const std::string&));
+  MOCK_METHOD2(notify_sync_request, void(const std::string&,
+                                         Context*));
+  MOCK_METHOD1(notify_sync_complete, void(const std::string&));
 };
 
 template <>
@@ -497,6 +501,19 @@ public:
       }));
   }
 
+  void expect_notify_sync_request(MockInstanceWatcher& mock_instance_watcher,
+                                  const std::string& image_id, int r) {
+    EXPECT_CALL(mock_instance_watcher, notify_sync_request(image_id, _))
+      .WillOnce(WithArg<1>(Invoke([this, r](Context* ctx) {
+        m_threads->work_queue->queue(ctx, r);
+      })));
+  }
+
+  void expect_notify_sync_complete(MockInstanceWatcher& mock_instance_watcher,
+                                   const std::string& image_id) {
+    EXPECT_CALL(mock_instance_watcher, notify_sync_complete(image_id));
+  }
+
   void expect_image_copy(MockImageCopyRequest& mock_image_copy_request,
                          uint64_t src_snap_id_start, uint64_t src_snap_id_end,
                          uint64_t dst_snap_id_start,
@@ -728,6 +745,7 @@ TEST_F(TestMockImageReplayerSnapshotReplayer, SyncSnapshot) {
   expect_create_non_primary_request(mock_create_non_primary_request,
                                     false, "remote mirror uuid", 1,
                                     {{1, CEPH_NOSNAP}}, 11, 0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id, 0);
   MockImageCopyRequest mock_image_copy_request;
   expect_image_copy(mock_image_copy_request, 0, 1, 0, {},
                     {{1, CEPH_NOSNAP}}, 0);
@@ -736,6 +754,7 @@ TEST_F(TestMockImageReplayerSnapshotReplayer, SyncSnapshot) {
   expect_mirror_image_snapshot_set_copy_progress(
     mock_local_image_ctx, 11, true, 0, 0);
   expect_notify_update(mock_local_image_ctx);
+  expect_notify_sync_complete(mock_instance_watcher, mock_local_image_ctx.id);
 
   // sync snap4
   expect_load_image_meta(mock_image_meta, false, 0);
@@ -754,6 +773,7 @@ TEST_F(TestMockImageReplayerSnapshotReplayer, SyncSnapshot) {
   expect_create_non_primary_request(mock_create_non_primary_request,
                                     false, "remote mirror uuid", 4,
                                     {{1, 11}, {4, CEPH_NOSNAP}}, 14, 0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id, 0);
   expect_image_copy(mock_image_copy_request, 1, 4, 11, {},
                     {{1, 11}, {4, CEPH_NOSNAP}}, 0);
   expect_apply_image_state(mock_apply_state_request, 0);
@@ -763,6 +783,7 @@ TEST_F(TestMockImageReplayerSnapshotReplayer, SyncSnapshot) {
   MockUnlinkPeerRequest mock_unlink_peer_request;
   expect_unlink_peer(mock_unlink_peer_request, 1, "remote mirror peer uuid",
                      0);
+  expect_notify_sync_complete(mock_instance_watcher, mock_local_image_ctx.id);
 
   // idle
   expect_load_image_meta(mock_image_meta, false, 0);
@@ -844,6 +865,7 @@ TEST_F(TestMockImageReplayerSnapshotReplayer, InterruptedSync) {
   expect_is_refresh_required(mock_remote_image_ctx, false);
   MockGetImageStateRequest mock_get_image_state_request;
   expect_get_image_state(mock_get_image_state_request, 11, 0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id, 0);
   MockImageCopyRequest mock_image_copy_request;
   expect_image_copy(mock_image_copy_request, 0, 1, 0,
                     librbd::deep_copy::ObjectNumber{123U},
@@ -929,6 +951,7 @@ TEST_F(TestMockImageReplayerSnapshotReplayer, RemoteImageDemoted) {
   expect_create_non_primary_request(mock_create_non_primary_request,
                                     true, "remote mirror uuid", 1,
                                     {{1, CEPH_NOSNAP}}, 11, 0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id, 0);
   MockImageCopyRequest mock_image_copy_request;
   expect_image_copy(mock_image_copy_request, 0, 1, 0, {},
                     {{1, CEPH_NOSNAP}}, 0);
@@ -1556,6 +1579,75 @@ TEST_F(TestMockImageReplayerSnapshotReplayer, CreateNonPrimarySnapshotError) {
                                         mock_remote_image_ctx));
 }
 
+TEST_F(TestMockImageReplayerSnapshotReplayer, RequestSyncError) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // inject snapshot
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"}, "",
+       0U, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+
+  // sync snap1
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  MockSnapshotCopyRequest mock_snapshot_copy_request;
+  expect_snapshot_copy(mock_snapshot_copy_request, 0, 1, 0, {{1, CEPH_NOSNAP}},
+                       0);
+  MockGetImageStateRequest mock_get_image_state_request;
+  expect_get_image_state(mock_get_image_state_request, 1, 0);
+  MockCreateNonPrimaryRequest mock_create_non_primary_request;
+  expect_create_non_primary_request(mock_create_non_primary_request,
+                                    false, "remote mirror uuid", 1,
+                                    {{1, CEPH_NOSNAP}}, 11, 0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id,
+                             -ECANCELED);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete and expect replay complete
+  ASSERT_EQ(0, wait_for_notification(1));
+  ASSERT_FALSE(mock_replayer.is_replaying());
+  ASSERT_EQ(-ECANCELED, mock_replayer.get_error_code());
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+
+}
+
 TEST_F(TestMockImageReplayerSnapshotReplayer, CopyImageError) {
   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
   librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
@@ -1608,6 +1700,7 @@ TEST_F(TestMockImageReplayerSnapshotReplayer, CopyImageError) {
   expect_create_non_primary_request(mock_create_non_primary_request,
                                     false, "remote mirror uuid", 1,
                                     {{1, CEPH_NOSNAP}}, 11, 0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id, 0);
   MockImageCopyRequest mock_image_copy_request;
   expect_image_copy(mock_image_copy_request, 0, 1, 0, {},
                     {{1, CEPH_NOSNAP}}, -EINVAL);
@@ -1677,6 +1770,7 @@ TEST_F(TestMockImageReplayerSnapshotReplayer, UpdateNonPrimarySnapshotError) {
   expect_create_non_primary_request(mock_create_non_primary_request,
                                     false, "remote mirror uuid", 1,
                                     {{1, CEPH_NOSNAP}}, 11, 0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id, 0);
   MockImageCopyRequest mock_image_copy_request;
   expect_image_copy(mock_image_copy_request, 0, 1, 0, {},
                     {{1, CEPH_NOSNAP}}, 0);
@@ -1759,6 +1853,7 @@ TEST_F(TestMockImageReplayerSnapshotReplayer, UnlinkPeerError) {
   expect_create_non_primary_request(mock_create_non_primary_request,
                                     false, "remote mirror uuid", 2,
                                     {{2, CEPH_NOSNAP}}, 12, 0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id, 0);
   MockImageCopyRequest mock_image_copy_request;
   expect_image_copy(mock_image_copy_request, 1, 2, 11, {},
                     {{2, CEPH_NOSNAP}}, 0);
@@ -1770,6 +1865,7 @@ TEST_F(TestMockImageReplayerSnapshotReplayer, UnlinkPeerError) {
   MockUnlinkPeerRequest mock_unlink_peer_request;
   expect_unlink_peer(mock_unlink_peer_request, 1, "remote mirror peer uuid",
                      -EINVAL);
+  expect_notify_sync_complete(mock_instance_watcher, mock_local_image_ctx.id);
 
   // wake-up replayer
   update_watch_ctx->handle_notify();

diff --git a/src/tools/rbd_mirror/image_replayer/snapshot/Replayer.cc b/src/tools/rbd_mirror/image_replayer/snapshot/Replayer.cc
index 55083c610197db2816a2e11250ff84014937c836..4cc2d1b78f9df5e55296adccd7fcdc14b62116de 100644 (file)
--- a/src/tools/rbd_mirror/image_replayer/snapshot/Replayer.cc
+++ b/src/tools/rbd_mirror/image_replayer/snapshot/Replayer.cc
@@ -175,6 +175,10 @@ void Replayer<I>::shut_down(Context* on_finish) {
   std::swap(m_state, state);
 
   if (state == STATE_REPLAYING) {
+    // if a sync request was pending, request a cancelation
+    m_instance_watcher->cancel_sync_request(
+      m_state_builder->local_image_ctx->id);
+
     // TODO interrupt snapshot copy and image copy state machines even if remote
     // cluster is unreachable
     dout(10) << "shut down pending on completion of snapshot replay" << dendl;
@@ -711,7 +715,7 @@ void Replayer<I>::handle_get_local_image_state(int r) {
     return;
   }
 
-  copy_image();
+  request_sync();
 }
 
 template <typename I>
@@ -740,6 +744,37 @@ void Replayer<I>::handle_create_non_primary_snapshot(int r) {
 
   dout(15) << "local_snap_id_end=" << m_local_snap_id_end << dendl;
 
+  request_sync();
+}
+
+template <typename I>
+void Replayer<I>::request_sync() {
+  dout(10) << dendl;
+
+  std::unique_lock locker{m_lock};
+  auto ctx = create_async_context_callback(
+    m_threads->work_queue, create_context_callback<
+      Replayer<I>, &Replayer<I>::handle_request_sync>(this));
+  m_instance_watcher->notify_sync_request(m_state_builder->local_image_ctx->id,
+                                          ctx);
+}
+
+template <typename I>
+void Replayer<I>::handle_request_sync(int r) {
+  dout(10) << "r=" << r << dendl;
+
+  if (r == -ECANCELED) {
+    dout(5) << "image-sync canceled" << dendl;
+    handle_replay_complete(r, "image-sync canceled");
+    return;
+  } else if (r < 0) {
+    derr << "failed to request image-sync: " << cpp_strerror(r) << dendl;
+    handle_replay_complete(r, "failed to request image-sync");
+    return;
+  }
+
+  m_sync_in_progress = true;
+
   copy_image();
 }
 
@@ -943,6 +978,10 @@ void Replayer<I>::finish_sync() {
   {
     std::unique_lock locker{m_lock};
     notify_status_updated();
+
+    m_sync_in_progress = false;
+    m_instance_watcher->notify_sync_complete(
+      m_state_builder->local_image_ctx->id);
   }
 
   if (is_replay_interrupted()) {
@@ -1140,6 +1179,12 @@ void Replayer<I>::handle_replay_complete(std::unique_lock<ceph::mutex>* locker,
     m_error_description = description;
   }
 
+  if (m_sync_in_progress) {
+    m_sync_in_progress = false;
+    m_instance_watcher->notify_sync_complete(
+      m_state_builder->local_image_ctx->id);
+  }
+
   if (m_state != STATE_REPLAYING && m_state != STATE_IDLE) {
     return;
   }

diff --git a/src/tools/rbd_mirror/image_replayer/snapshot/Replayer.h b/src/tools/rbd_mirror/image_replayer/snapshot/Replayer.h
index f0b1fccd56be4c4901d0f0c97c8772cf941efcb5..e7536f5a6fa26ea96694f7eaa23597221917d6db 100644 (file)
--- a/src/tools/rbd_mirror/image_replayer/snapshot/Replayer.h
+++ b/src/tools/rbd_mirror/image_replayer/snapshot/Replayer.h
@@ -126,6 +126,9 @@ private:
    *    |                     |/----------------------/ |
    *    |                     |                         |
    *    |                     v                         |
+   *    |                 REQUEST_SYNC                  |
+   *    |                     |                         |
+   *    |                     v                         |
    *    |                 COPY_IMAGE                    |
    *    |                     |                         |
    *    |                     v                         |
@@ -221,6 +224,7 @@ private:
 
   bool m_remote_image_updated = false;
   bool m_updating_sync_point = false;
+  bool m_sync_in_progress = false;
 
   void load_local_image_meta();
   void handle_load_local_image_meta(int r);
@@ -246,6 +250,9 @@ private:
   void create_non_primary_snapshot();
   void handle_create_non_primary_snapshot(int r);
 
+  void request_sync();
+  void handle_request_sync(int r);
+
   void copy_image();
   void handle_copy_image(int r);
   void handle_copy_image_progress(uint64_t object_number,