cls/rbd: method to unlink peer from mirror snapshot

Signed-off-by: Mykola Golub <mgolub@suse.com>

diff --git a/src/cls/rbd/cls_rbd.cc b/src/cls/rbd/cls_rbd.cc
index 1f97e275e2b54d52e3a1fa373c79efcd1a12b261..54d0d90a358b5e8cd38bf946203fd192f7f2dc9d 100644 (file)
--- a/src/cls/rbd/cls_rbd.cc
+++ b/src/cls/rbd/cls_rbd.cc
@@ -5662,6 +5662,48 @@ int mirror_image_map_list(cls_method_context_t hctx,
   return 0;
 }
 
+int image_snapshot_unlink_peer(cls_method_context_t hctx,
+                               uint64_t snap_id,
+                               std::string mirror_peer_uuid) {
+  cls_rbd_snap snap;
+  std::string snap_key;
+  key_from_snap_id(snap_id, &snap_key);
+  int r = read_key(hctx, snap_key, &snap);
+  if (r < 0) {
+    if (r != -ENOENT) {
+      CLS_ERR("Could not read snapshot meta off disk: %s",
+              cpp_strerror(r).c_str());
+    }
+    return r;
+  }
+
+  auto primary = boost::get<cls::rbd::MirrorPrimarySnapshotNamespace>(
+    &snap.snapshot_namespace);
+  if (primary == nullptr) {
+    CLS_LOG(5, "mirror_image_snapshot_unlink_peer " \
+            "not mirroring snapshot snap_id=%" PRIu64, snap_id);
+    return -EINVAL;
+  }
+
+  if (primary->mirror_peer_uuids.count(mirror_peer_uuid) == 0) {
+    return -ENOENT;
+  }
+
+  if (primary->mirror_peer_uuids.size() == 1) {
+    // return a special error when trying to unlink the last peer
+    return -ERESTART;
+  }
+
+  primary->mirror_peer_uuids.erase(mirror_peer_uuid);
+
+  r = image::snapshot::write(hctx, snap_key, std::move(snap));
+  if (r < 0) {
+    return r;
+  }
+
+  return 0;
+}
+
 } // namespace mirror
 
 /**
@@ -6643,6 +6685,38 @@ int mirror_image_map_remove(cls_method_context_t hctx, bufferlist *in,
   return 0;
 }
 
+
+/**
+ * Input:
+ * @param snap_id: snapshot id
+ * @param mirror_peer_uuid: mirror peer uuid
+ *
+ * Output:
+ * @returns 0 on success, negative error code on failure
+ */
+int mirror_image_snapshot_unlink_peer(cls_method_context_t hctx, bufferlist *in,
+                                      bufferlist *out) {
+  uint64_t snap_id;
+  std::string mirror_peer_uuid;
+  try {
+    auto iter = in->cbegin();
+    decode(snap_id, iter);
+    decode(mirror_peer_uuid, iter);
+  } catch (const buffer::error &err) {
+    return -EINVAL;
+  }
+
+  CLS_LOG(20,
+          "mirror_image_snapshot_unlink_peer snap_id=%" PRIu64 " peer_uuid=%s",
+          snap_id, mirror_peer_uuid.c_str());
+
+  int r = mirror::image_snapshot_unlink_peer(hctx, snap_id, mirror_peer_uuid);
+  if (r < 0) {
+    return r;
+  }
+  return 0;
+}
+
 namespace group {
 
 /********************** methods for rbd_group_directory ***********************/
@@ -8011,6 +8085,7 @@ CLS_INIT(rbd)
   cls_method_handle_t h_mirror_image_map_list;
   cls_method_handle_t h_mirror_image_map_update;
   cls_method_handle_t h_mirror_image_map_remove;
+  cls_method_handle_t h_mirror_image_snapshot_unlink_peer;
   cls_method_handle_t h_group_dir_list;
   cls_method_handle_t h_group_dir_add;
   cls_method_handle_t h_group_dir_remove;
@@ -8356,6 +8431,10 @@ CLS_INIT(rbd)
   cls_register_cxx_method(h_class, "mirror_image_map_remove",
                           CLS_METHOD_WR, mirror_image_map_remove,
                           &h_mirror_image_map_remove);
+  cls_register_cxx_method(h_class, "mirror_image_snapshot_unlink_peer",
+                          CLS_METHOD_RD | CLS_METHOD_WR,
+                          mirror_image_snapshot_unlink_peer,
+                          &h_mirror_image_snapshot_unlink_peer);
 
   /* methods for the groups feature */
   cls_register_cxx_method(h_class, "group_dir_list",

diff --git a/src/cls/rbd/cls_rbd_client.cc b/src/cls/rbd/cls_rbd_client.cc
index 514acefeeca79fec5f04bcae91028549d5255d0b..03312d2f630c091100ba7390428b4ef1845a9875 100644 (file)
--- a/src/cls/rbd/cls_rbd_client.cc
+++ b/src/cls/rbd/cls_rbd_client.cc
@@ -2445,6 +2445,25 @@ void mirror_image_map_remove(librados::ObjectWriteOperation *op,
   op->exec("rbd", "mirror_image_map_remove", bl);
 }
 
+void mirror_image_snapshot_unlink_peer(librados::ObjectWriteOperation *op,
+                                       snapid_t snap_id,
+                                       const std::string &mirror_peer_uuid) {
+  bufferlist bl;
+  encode(snap_id, bl);
+  encode(mirror_peer_uuid, bl);
+
+  op->exec("rbd", "mirror_image_snapshot_unlink_peer", bl);
+}
+
+int mirror_image_snapshot_unlink_peer(librados::IoCtx *ioctx,
+                                      const std::string &oid,
+                                      snapid_t snap_id,
+                                      const std::string &mirror_peer_uuid) {
+  librados::ObjectWriteOperation op;
+  mirror_image_snapshot_unlink_peer(&op, snap_id, mirror_peer_uuid);
+  return ioctx->operate(oid, &op);
+}
+
 // Groups functions
 int group_dir_list(librados::IoCtx *ioctx, const std::string &oid,
                    const std::string &start, uint64_t max_return,

diff --git a/src/cls/rbd/cls_rbd_client.h b/src/cls/rbd/cls_rbd_client.h
index 5ff683737e1f8963270f0e615d70635f7c4b15a1..fcb0ebc862ec326285a941cd889b9266b4370d57 100644 (file)
--- a/src/cls/rbd/cls_rbd_client.h
+++ b/src/cls/rbd/cls_rbd_client.h
@@ -519,6 +519,13 @@ void mirror_image_map_update(librados::ObjectWriteOperation *op,
 void mirror_image_map_remove(librados::ObjectWriteOperation *op,
                              const std::string &global_image_id);
 
+void mirror_image_snapshot_unlink_peer(librados::ObjectWriteOperation *op,
+                                       snapid_t snap_id,
+                                       const std::string &mirror_peer_uuid);
+int mirror_image_snapshot_unlink_peer(librados::IoCtx *ioctx,
+                                      const std::string &oid,
+                                      snapid_t snap_id,
+                                      const std::string &mirror_peer_uuid);
 // Groups functions
 int group_dir_list(librados::IoCtx *ioctx, const std::string &oid,
                    const std::string &start, uint64_t max_return,

diff --git a/src/test/cls_rbd/test_cls_rbd.cc b/src/test/cls_rbd/test_cls_rbd.cc
index 39d84dbefeafe5ca6ff160f8494016d9bc1698d4..42522cfbb135cde4ff758aebf44f680d30e405ff 100644 (file)
--- a/src/test/cls_rbd/test_cls_rbd.cc
+++ b/src/test/cls_rbd/test_cls_rbd.cc
@@ -2229,6 +2229,63 @@ TEST_F(TestClsRbd, mirror_instances) {
   ASSERT_EQ(0U, instance_ids.size());
 }
 
+TEST_F(TestClsRbd, mirror_snapshot) {
+  librados::IoCtx ioctx;
+  ASSERT_EQ(0, _rados.ioctx_create(_pool_name.c_str(), ioctx));
+
+  string oid = get_temp_image_name();
+  ASSERT_EQ(0, create_image(&ioctx, oid, 10, 22, 0, oid, -1));
+
+  cls::rbd::MirrorPrimarySnapshotNamespace primary = {false,
+                                                      {"peer1", "peer2"}};
+  cls::rbd::MirrorNonPrimarySnapshotNamespace non_primary = {"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);
+  ASSERT_EQ(0, ioctx.operate(oid, &op));
+
+  cls::rbd::SnapshotInfo snap;
+  ASSERT_EQ(0, snapshot_get(&ioctx, oid, 1, &snap));
+  auto sn = boost::get<cls::rbd::MirrorPrimarySnapshotNamespace>(
+    &snap.snapshot_namespace);
+  ASSERT_NE(nullptr, sn);
+  ASSERT_EQ(primary, *sn);
+  ASSERT_EQ(2U, sn->mirror_peer_uuids.size());
+  ASSERT_EQ(1U, sn->mirror_peer_uuids.count("peer1"));
+  ASSERT_EQ(1U, sn->mirror_peer_uuids.count("peer2"));
+
+  ASSERT_EQ(-ENOENT, mirror_image_snapshot_unlink_peer(&ioctx, oid, 1, "peer"));
+  ASSERT_EQ(0, mirror_image_snapshot_unlink_peer(&ioctx, oid, 1, "peer1"));
+  ASSERT_EQ(-ENOENT, mirror_image_snapshot_unlink_peer(&ioctx, oid, 1,
+                                                       "peer1"));
+  ASSERT_EQ(0, snapshot_get(&ioctx, oid, 1, &snap));
+  sn = boost::get<cls::rbd::MirrorPrimarySnapshotNamespace>(
+    &snap.snapshot_namespace);
+  ASSERT_NE(nullptr, sn);
+  ASSERT_EQ(1U, sn->mirror_peer_uuids.size());
+  ASSERT_EQ(1U, sn->mirror_peer_uuids.count("peer2"));
+
+  ASSERT_EQ(-ERESTART,
+            mirror_image_snapshot_unlink_peer(&ioctx, oid, 1, "peer2"));
+  ASSERT_EQ(0, snapshot_get(&ioctx, oid, 1, &snap));
+  sn = boost::get<cls::rbd::MirrorPrimarySnapshotNamespace>(
+    &snap.snapshot_namespace);
+  ASSERT_NE(nullptr, sn);
+  ASSERT_EQ(1U, sn->mirror_peer_uuids.size());
+  ASSERT_EQ(1U, sn->mirror_peer_uuids.count("peer2"));
+
+  ASSERT_EQ(0, snapshot_get(&ioctx, oid, 2, &snap));
+  auto nsn = boost::get<cls::rbd::MirrorNonPrimarySnapshotNamespace>(
+    &snap.snapshot_namespace);
+  ASSERT_NE(nullptr, nsn);
+  ASSERT_EQ(non_primary, *nsn);
+  ASSERT_FALSE(nsn->copied);
+  ASSERT_EQ(nsn->last_copied_object_number, 0);
+
+  ASSERT_EQ(0, snapshot_remove(&ioctx, oid, 1));
+  ASSERT_EQ(0, snapshot_remove(&ioctx, oid, 2));
+}
+
 TEST_F(TestClsRbd, group_dir_list) {
   librados::IoCtx ioctx;
   ASSERT_EQ(0, _rados.ioctx_create(_pool_name.c_str(), ioctx));