librbd: permit last mirror peer to be removed from mirror snapshot

This applies only in the case where there are no additional mirror
snapshots in order to preserve the current state.

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

diff --git a/src/cls/rbd/cls_rbd.cc b/src/cls/rbd/cls_rbd.cc
index 6e7cfbd2ee091211fcf8c385107773256054e173..b6ae25d7c91b0313214d211bfe16ac1d242b5185 100644 (file)
--- a/src/cls/rbd/cls_rbd.cc
+++ b/src/cls/rbd/cls_rbd.cc
@@ -5696,8 +5696,21 @@ int image_snapshot_unlink_peer(cls_method_context_t hctx,
   }
 
   if (mirror_ns->mirror_peer_uuids.size() == 1) {
-    // return a special error when trying to unlink the last peer
-    return -ERESTART;
+    // if this is the last peer to unlink and we have at least one additional
+    // newer mirror snapshot, return a special error to inform the caller it
+    // should remove the snapshot instead.
+    auto search_lambda = [snap_id](const cls_rbd_snap& snap_meta) {
+      if (snap_meta.id > snap_id &&
+          boost::get<cls::rbd::MirrorSnapshotNamespace>(
+                   &snap_meta.snapshot_namespace) != nullptr) {
+        return -EEXIST;
+      }
+      return 0;
+    };
+    r = image::snapshot::iterate(hctx, search_lambda);
+    if (r == -EEXIST) {
+      return -ERESTART;
+    }
   }
 
   mirror_ns->mirror_peer_uuids.erase(mirror_peer_uuid);

diff --git a/src/librbd/mirror/snapshot/UnlinkPeerRequest.cc b/src/librbd/mirror/snapshot/UnlinkPeerRequest.cc
index 40fc0af55261519f6bc3a71a036186fae465afc1..67f51dc4bd654e57447e08f8913d66889cf2740c 100644 (file)
--- a/src/librbd/mirror/snapshot/UnlinkPeerRequest.cc
+++ b/src/librbd/mirror/snapshot/UnlinkPeerRequest.cc
@@ -61,42 +61,52 @@ void UnlinkPeerRequest<I>::unlink_peer() {
   CephContext *cct = m_image_ctx->cct;
 
   m_image_ctx->image_lock.lock_shared();
+  int r = -ENOENT;
+  cls::rbd::MirrorSnapshotNamespace* mirror_ns = nullptr;
+  bool newer_mirror_snapshots = false;
+  for (auto snap_it = m_image_ctx->snap_info.find(m_snap_id);
+       snap_it != m_image_ctx->snap_info.end(); ++snap_it) {
+    if (snap_it->first == m_snap_id) {
+      r = 0;
+      mirror_ns = boost::get<cls::rbd::MirrorSnapshotNamespace>(
+        &snap_it->second.snap_namespace);
+    } else if (boost::get<cls::rbd::MirrorSnapshotNamespace>(
+                 &snap_it->second.snap_namespace) != nullptr) {
+      newer_mirror_snapshots = true;
+      break;
+    }
+  }
 
-  auto snap_info = m_image_ctx->get_snap_info(m_snap_id);
-  if (!snap_info) {
+  if (r == -ENOENT) {
     m_image_ctx->image_lock.unlock_shared();
-    finish(-ENOENT);
+    finish(r);
     return;
   }
 
-  auto info = boost::get<cls::rbd::MirrorSnapshotNamespace>(
-    &snap_info->snap_namespace);
-  if (info == nullptr) {
-    lderr(cct) << "not mirror primary snapshot (snap_id=" << m_snap_id << ")"
-               << dendl;
+  if (mirror_ns == nullptr) {
+    lderr(cct) << "not mirror snapshot (snap_id=" << m_snap_id << ")" << dendl;
     m_image_ctx->image_lock.unlock_shared();
     finish(-EINVAL);
     return;
   }
 
-  if (info->mirror_peer_uuids.count(m_mirror_peer_uuid) == 0 ||
-      info->mirror_peer_uuids.size() == 1U) {
+  // if there is or will be no more peers in the mirror snapshot and we have
+  // a more recent mirror snapshot, remove the older one
+  if ((mirror_ns->mirror_peer_uuids.count(m_mirror_peer_uuid) == 0) ||
+      (mirror_ns->mirror_peer_uuids.size() == 1U && newer_mirror_snapshots)) {
     m_image_ctx->image_lock.unlock_shared();
     remove_snapshot();
     return;
   }
-
   m_image_ctx->image_lock.unlock_shared();
 
   ldout(cct, 20) << dendl;
-
   librados::ObjectWriteOperation op;
   librbd::cls_client::mirror_image_snapshot_unlink_peer(&op, m_snap_id,
                                                         m_mirror_peer_uuid);
   auto aio_comp = create_rados_callback<
     UnlinkPeerRequest<I>, &UnlinkPeerRequest<I>::handle_unlink_peer>(this);
-  int r = m_image_ctx->md_ctx.aio_operate(m_image_ctx->header_oid, aio_comp,
-                                          &op);
+  r = m_image_ctx->md_ctx.aio_operate(m_image_ctx->header_oid, aio_comp, &op);
   ceph_assert(r == 0);
   aio_comp->release();
 }

diff --git a/src/librbd/mirror/snapshot/UnlinkPeerRequest.h b/src/librbd/mirror/snapshot/UnlinkPeerRequest.h
index c42f89200d0cb9c4cca52bb9d6c13ae5cef8c779..184f7ccd9569f82e7d556b768d116b9d7dfe163d 100644 (file)
--- a/src/librbd/mirror/snapshot/UnlinkPeerRequest.h
+++ b/src/librbd/mirror/snapshot/UnlinkPeerRequest.h
@@ -48,11 +48,14 @@ private:
    *    |                     *  or last)     |
    *    |                     *               |
    *    |\---------------> UNLINK_PEER --> NOTIFY_UPDATE
-   *    |   (peer not last)
-   *    |
+   *    |   (not last peer or
+   *    |    no newer mirror
+   *    |    snap exists)
    *    |
    *    |\---------------> REMOVE_SNAPSHOT
-   *    |   (peer last)       |
+   *    |   (last peer and    |
+   *    |    newer mirror     |
+   *    |    snap exists)     |
    *    |                     |
    *    |(peer not found)     |
    *    v                     |

diff --git a/src/test/cls_rbd/test_cls_rbd.cc b/src/test/cls_rbd/test_cls_rbd.cc
index f6a59ffd8c810f76668b7680223934b73cc76cca..ef753b749809be5587d14f4ca5c15ce7a97877db 100644 (file)
--- a/src/test/cls_rbd/test_cls_rbd.cc
+++ b/src/test/cls_rbd/test_cls_rbd.cc
@@ -2243,7 +2243,7 @@ TEST_F(TestClsRbd, mirror_snapshot) {
     cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"peer1", "peer2"}, "",
     CEPH_NOSNAP};
   cls::rbd::MirrorSnapshotNamespace non_primary = {
-    cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "uuid", 123};
+    cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {"peer1"}, "uuid", 123};
   librados::ObjectWriteOperation op;
   ::librbd::cls_client::snapshot_add(&op, 1, "primary", primary);
   ::librbd::cls_client::snapshot_add(&op, 2, "non_primary", non_primary);
@@ -2284,6 +2284,8 @@ TEST_F(TestClsRbd, mirror_snapshot) {
     &snap.snapshot_namespace);
   ASSERT_NE(nullptr, nsn);
   ASSERT_EQ(non_primary, *nsn);
+  ASSERT_EQ(1U, nsn->mirror_peer_uuids.size());
+  ASSERT_EQ(1U, nsn->mirror_peer_uuids.count("peer1"));
   ASSERT_FALSE(nsn->complete);
   ASSERT_EQ(nsn->last_copied_object_number, 0);
 
@@ -2296,6 +2298,8 @@ TEST_F(TestClsRbd, mirror_snapshot) {
   ASSERT_TRUE(nsn->complete);
   ASSERT_EQ(nsn->last_copied_object_number, 10);
 
+  ASSERT_EQ(0, mirror_image_snapshot_unlink_peer(&ioctx, oid, 2, "peer1"));
+
   ASSERT_EQ(0, snapshot_remove(&ioctx, oid, 1));
   ASSERT_EQ(0, snapshot_remove(&ioctx, oid, 2));
 }

diff --git a/src/test/librbd/mirror/snapshot/test_mock_UnlinkPeerRequest.cc b/src/test/librbd/mirror/snapshot/test_mock_UnlinkPeerRequest.cc
index 5de92de30d9bc4a02c1b18bb21161cad5df6b1db..eba9b5cbf3feaca304b40657430ce96f8edb34e6 100644 (file)
--- a/src/test/librbd/mirror/snapshot/test_mock_UnlinkPeerRequest.cc
+++ b/src/test/librbd/mirror/snapshot/test_mock_UnlinkPeerRequest.cc
@@ -165,6 +165,7 @@ TEST_F(TestMockMirrorSnapshotUnlinkPeerRequest, RemoveSnapshot) {
     cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"peer_uuid"},
     "", CEPH_NOSNAP};
   auto snap_id = snap_create(mock_image_ctx, ns, "mirror_snap");
+  snap_create(mock_image_ctx, ns, "mirror_snap2");
 
   expect_get_snap_info(mock_image_ctx, snap_id);
 
@@ -334,6 +335,7 @@ TEST_F(TestMockMirrorSnapshotUnlinkPeerRequest, RemoveSnapshotError) {
     cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"peer_uuid"},
     "", CEPH_NOSNAP};
   auto snap_id = snap_create(mock_image_ctx, ns, "mirror_snap");
+  snap_create(mock_image_ctx, ns, "mirror_snap2");
 
   expect_get_snap_info(mock_image_ctx, snap_id);
 

diff --git a/src/test/librbd/test_mirroring.cc b/src/test/librbd/test_mirroring.cc
index e0145c889d0d796d8ecafe308b9fed23830217f8..cb69dcfbc1d317809b27381ac34bbdfc075106ff 100644 (file)
--- a/src/test/librbd/test_mirroring.cc
+++ b/src/test/librbd/test_mirroring.cc
@@ -1242,6 +1242,8 @@ TEST_F(TestMirroring, SnapshotUnlinkPeer)
   ASSERT_EQ(0, image.mirror_image_enable2(RBD_MIRROR_IMAGE_MODE_SNAPSHOT));
   uint64_t snap_id;
   ASSERT_EQ(0, image.mirror_image_create_snapshot(&snap_id));
+  uint64_t snap_id2;
+  ASSERT_EQ(0, image.mirror_image_create_snapshot(&snap_id2));
   librbd::snap_mirror_namespace_t mirror_snap;
   ASSERT_EQ(0, image.snap_get_mirror_namespace(snap_id, &mirror_snap,
                                                sizeof(mirror_snap)));