rbd: not allowed to restore an image when it is being deleted

Add a state for trash image.
Fixes: http://tracker.ceph.com/issues/25346
Signed-off-by: songweibin <song.weibin@zte.com.cn>
(cherry picked from commit 94140df624f33406c709dea270f07ed6ad6edac4)

diff --git a/src/cls/rbd/cls_rbd.cc b/src/cls/rbd/cls_rbd.cc
index 38cfb7ca97c3b71b76e8ff62e6c1e04fd2ba92c6..2c4c13ea00832ed7911743222156d7b6747d603e 100644 (file)
--- a/src/cls/rbd/cls_rbd.cc
+++ b/src/cls/rbd/cls_rbd.cc
@@ -6893,6 +6893,62 @@ int trash_get(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
   return r;
 }
 
+/**
+ * Set state of an image in the rbd_trash object.
+ *
+ * Input:
+ * @param id the id of the image
+ * @param trash_state the state of the image to be set
+ * @param expect_state the expected state of the image
+ *
+ * Output:
+ * @returns 0 on success, negative error code on failure
+ */
+int trash_state_set(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
+{
+  string id;
+  cls::rbd::TrashImageState trash_state;
+  cls::rbd::TrashImageState expect_state;
+  try {
+    bufferlist::const_iterator iter = in->begin();
+    decode(id, iter);
+    decode(trash_state, iter);
+    decode(expect_state, iter);
+  } catch (const buffer::error &err) {
+    return -EINVAL;
+  }
+
+  CLS_LOG(20, "trash_state_set id=%s", id.c_str());
+
+  string key = trash::image_key(id);
+  cls::rbd::TrashImageSpec trash_spec;
+  int r = read_key(hctx, key, &trash_spec);
+  if (r < 0) {
+    if (r != -ENOENT) {
+      CLS_ERR("Could not read trash image spec off disk: %s",
+              cpp_strerror(r).c_str());
+    }
+    return r;
+  }
+
+  if (trash_spec.state == expect_state) {
+    trash_spec.state = trash_state;
+    r = write_key(hctx, key, trash_spec);
+    if (r < 0) {
+      CLS_ERR("error setting trash image state: %s", cpp_strerror(r).c_str());
+      return r;
+    }
+
+    return 0;
+  } else if (trash_spec.state == trash_state) {
+    return 0;
+  } else {
+    CLS_ERR("Current trash state: %d do not match expected: %d or set: %d",
+            trash_spec.state, expect_state, trash_state);
+    return -ESTALE;
+  }
+}
+
 namespace nspace {
 
 const std::string NAME_KEY_PREFIX("name_");
@@ -7158,6 +7214,7 @@ CLS_INIT(rbd)
   cls_method_handle_t h_trash_remove;
   cls_method_handle_t h_trash_list;
   cls_method_handle_t h_trash_get;
+  cls_method_handle_t h_trash_state_set;
   cls_method_handle_t h_namespace_add;
   cls_method_handle_t h_namespace_remove;
   cls_method_handle_t h_namespace_list;
@@ -7532,6 +7589,9 @@ CLS_INIT(rbd)
   cls_register_cxx_method(h_class, "trash_get",
                           CLS_METHOD_RD,
                           trash_get, &h_trash_get);
+  cls_register_cxx_method(h_class, "trash_state_set",
+                          CLS_METHOD_RD | CLS_METHOD_WR,
+                          trash_state_set, &h_trash_state_set);
 
   /* rbd_namespace object methods */
   cls_register_cxx_method(h_class, "namespace_add",

diff --git a/src/cls/rbd/cls_rbd_client.cc b/src/cls/rbd/cls_rbd_client.cc
index b4835551209f867d166185b426faebdc1735bbc4..8dca588fd7f833b9b2a0e8ea7e595ca952e02ca1 100644 (file)
--- a/src/cls/rbd/cls_rbd_client.cc
+++ b/src/cls/rbd/cls_rbd_client.cc
@@ -2595,7 +2595,6 @@ int trash_get_finish(bufferlist::const_iterator *it,
   return 0;
 }
 
-
 int trash_get(librados::IoCtx *ioctx, const std::string &id,
               cls::rbd::TrashImageSpec *trash_spec)
 {
@@ -2612,6 +2611,28 @@ int trash_get(librados::IoCtx *ioctx, const std::string &id,
   return trash_get_finish(&it, trash_spec);
 }
 
+void trash_state_set(librados::ObjectWriteOperation *op,
+                     const std::string &id,
+                     const cls::rbd::TrashImageState &trash_state,
+                     const cls::rbd::TrashImageState &expect_state)
+{
+  bufferlist bl;
+  encode(id, bl);
+  encode(trash_state, bl);
+  encode(expect_state, bl);
+  op->exec("rbd", "trash_state_set", bl);
+}
+
+int trash_state_set(librados::IoCtx *ioctx, const std::string &id,
+                    const cls::rbd::TrashImageState &trash_state,
+                    const cls::rbd::TrashImageState &expect_state)
+{
+  librados::ObjectWriteOperation op;
+  trash_state_set(&op, id, trash_state, expect_state);
+
+  return ioctx->operate(RBD_TRASH, &op);
+}
+
 void namespace_add(librados::ObjectWriteOperation *op,
                    const std::string &name)
 {

diff --git a/src/cls/rbd/cls_rbd_client.h b/src/cls/rbd/cls_rbd_client.h
index c7a8e22badd028365ee3e1796dff6bc7db1ab975..a38ad24d0c9f02897ba5997e7e6afe394a21274c 100644 (file)
--- a/src/cls/rbd/cls_rbd_client.h
+++ b/src/cls/rbd/cls_rbd_client.h
@@ -565,6 +565,13 @@ int trash_get_finish(bufferlist::const_iterator *it,
                      cls::rbd::TrashImageSpec *trash_spec);
 int trash_get(librados::IoCtx *ioctx, const std::string &id,
               cls::rbd::TrashImageSpec *trash_spec);
+void trash_state_set(librados::ObjectWriteOperation *op,
+                     const std::string &id,
+                     const cls::rbd::TrashImageState &trash_state,
+                     const cls::rbd::TrashImageState &expect_state);
+int trash_state_set(librados::IoCtx *ioctx, const std::string &id,
+                    const cls::rbd::TrashImageState &trash_state,
+                    const cls::rbd::TrashImageState &expect_state);
 
 // operations on rbd_namespace object
 void namespace_add(librados::ObjectWriteOperation *op,

diff --git a/src/cls/rbd/cls_rbd_types.cc b/src/cls/rbd/cls_rbd_types.cc
index fc733bb6f223199de57f13cccbb6014f2d75fe58..0c779a812d827a2d6377ef93c33c661a78df77c0 100644 (file)
--- a/src/cls/rbd/cls_rbd_types.cc
+++ b/src/cls/rbd/cls_rbd_types.cc
@@ -689,20 +689,24 @@ void GroupSnapshot::generate_test_instances(std::list<GroupSnapshot *> &o) {
   o.push_back(new GroupSnapshot("1018643c9869", "groupsnapshot2", GROUP_SNAPSHOT_STATE_COMPLETE));
 }
 void TrashImageSpec::encode(bufferlist& bl) const {
-  ENCODE_START(1, 1, bl);
+  ENCODE_START(2, 1, bl);
   encode(source, bl);
   encode(name, bl);
   encode(deletion_time, bl);
   encode(deferment_end_time, bl);
+  encode(state, bl);
   ENCODE_FINISH(bl);
 }
 
 void TrashImageSpec::decode(bufferlist::const_iterator &it) {
-  DECODE_START(1, it);
+  DECODE_START(2, it);
   decode(source, it);
   decode(name, it);
   decode(deletion_time, it);
   decode(deferment_end_time, it);
+  if (struct_v >= 2) {
+    decode(state, it);
+  }
   DECODE_FINISH(it);
 }
 

diff --git a/src/cls/rbd/cls_rbd_types.h b/src/cls/rbd/cls_rbd_types.h
index 60042e6097969a3870d719cd095bc59c828cc92f..ac914081ed9f36b3f3d0ff45f4a839326d9f1cd5 100644 (file)
--- a/src/cls/rbd/cls_rbd_types.h
+++ b/src/cls/rbd/cls_rbd_types.h
@@ -595,11 +595,33 @@ inline void decode(TrashImageSource &source, bufferlist::const_iterator& it)
   source = static_cast<TrashImageSource>(int_source);
 }
 
+enum TrashImageState {
+  TRASH_IMAGE_STATE_NORMAL    = 0,
+  TRASH_IMAGE_STATE_MOVING    = 1,
+  TRASH_IMAGE_STATE_REMOVING  = 2,
+  TRASH_IMAGE_STATE_RESTORING = 3
+};
+
+inline void encode(const TrashImageState &state, bufferlist &bl)
+{
+  using ceph::encode;
+  encode(static_cast<uint8_t>(state), bl);
+}
+
+inline void decode(TrashImageState &state, bufferlist::const_iterator &it)
+{
+  uint8_t int_state;
+  using ceph::decode;
+  decode(int_state, it);
+  state = static_cast<TrashImageState>(int_state);
+}
+
 struct TrashImageSpec {
   TrashImageSource source = TRASH_IMAGE_SOURCE_USER;
   std::string name;
   utime_t deletion_time; // time of deletion
   utime_t deferment_end_time;
+  TrashImageState state = TRASH_IMAGE_STATE_NORMAL;
 
   TrashImageSpec() {}
   TrashImageSpec(TrashImageSource source, const std::string &name,

diff --git a/src/librbd/internal.cc b/src/librbd/internal.cc
index f7eec3c4d69946811b1a3da3c77b21345b3ba450..c73f8c2872cad24078684132424f2d699707b99f 100644 (file)
--- a/src/librbd/internal.cc
+++ b/src/librbd/internal.cc
@@ -1443,6 +1443,7 @@ bool compare_by_name(const child_info_t& c1, const child_info_t& c2)
       static_cast<cls::rbd::TrashImageSource>(source), ictx->name,
       delete_time, deferment_end_time};
 
+    trash_image_spec.state = cls::rbd::TRASH_IMAGE_STATE_MOVING;
     C_SaferCond ctx;
     auto req = trash::MoveRequest<>::create(io_ctx, ictx->id, trash_image_spec,
                                             &ctx);
@@ -1450,6 +1451,15 @@ bool compare_by_name(const child_info_t& c1, const child_info_t& c2)
 
     r = ctx.wait();
     ictx->state->close();
+    trash_image_spec.state = cls::rbd::TRASH_IMAGE_STATE_NORMAL;
+    int ret = cls_client::trash_state_set(&io_ctx, image_id,
+                                          trash_image_spec.state,
+                                          cls::rbd::TRASH_IMAGE_STATE_MOVING);
+    if (ret < 0 && ret != -EOPNOTSUPP) {
+      lderr(cct) << "error setting trash image state: "
+                 << cpp_strerror(ret) << dendl;
+      return ret;
+    }
     if (r < 0) {
       return r;
     }
@@ -1544,11 +1554,32 @@ bool compare_by_name(const child_info_t& c1, const child_info_t& c2)
       lderr(cct) << "error: deferment time has not expired." << dendl;
       return -EPERM;
     }
+    if (trash_spec.state != cls::rbd::TRASH_IMAGE_STATE_NORMAL &&
+        trash_spec.state != cls::rbd::TRASH_IMAGE_STATE_REMOVING) {
+      lderr(cct) << "error: image is pending restoration." << dendl;
+      return -EBUSY;
+    }
+
+    r = cls_client::trash_state_set(&io_ctx, image_id,
+                                    cls::rbd::TRASH_IMAGE_STATE_REMOVING,
+                                    cls::rbd::TRASH_IMAGE_STATE_NORMAL);
+    if (r < 0 && r != -EOPNOTSUPP) {
+      lderr(cct) << "error setting trash image state: "
+                 << cpp_strerror(r) << dendl;
+      return r;
+    }
 
     r = remove(io_ctx, "", image_id, prog_ctx, false, true);
     if (r < 0) {
       lderr(cct) << "error removing image " << image_id
                  << ", which is pending deletion" << dendl;
+      int ret = cls_client::trash_state_set(&io_ctx, image_id,
+                                            cls::rbd::TRASH_IMAGE_STATE_NORMAL,
+                                            cls::rbd::TRASH_IMAGE_STATE_REMOVING);
+      if (ret < 0 && ret != -EOPNOTSUPP) {
+        lderr(cct) << "error setting trash image state: "
+                   << cpp_strerror(ret) << dendl;
+      }
       return r;
     }
     r = cls_client::trash_remove(&io_ctx, image_id);
@@ -1584,6 +1615,21 @@ bool compare_by_name(const child_info_t& c1, const child_info_t& c2)
     }
 
     std::string image_name = image_new_name;
+    if (trash_spec.state != cls::rbd::TRASH_IMAGE_STATE_NORMAL &&
+        trash_spec.state != cls::rbd::TRASH_IMAGE_STATE_RESTORING) {
+      lderr(cct) << "error restoring image id " << image_id
+                 << ", which is pending deletion" << dendl;
+      return -EBUSY;
+    }
+    r = cls_client::trash_state_set(&io_ctx, image_id,
+                                    cls::rbd::TRASH_IMAGE_STATE_RESTORING,
+                                    cls::rbd::TRASH_IMAGE_STATE_NORMAL);
+    if (r < 0 && r != -EOPNOTSUPP) {
+      lderr(cct) << "error setting trash image state: "
+                 << cpp_strerror(r) << dendl;
+      return r;
+    }
+
     if (image_name.empty()) {
       // if user didn't specify a new name, let's try using the old name
       image_name = trash_spec.name;
@@ -1598,11 +1644,25 @@ bool compare_by_name(const child_info_t& c1, const child_info_t& c2)
     if (r < 0 && r != -ENOENT) {
       lderr(cct) << "error checking if image " << image_name << " exists: "
                  << cpp_strerror(r) << dendl;
+      int ret = cls_client::trash_state_set(&io_ctx, image_id,
+                                            cls::rbd::TRASH_IMAGE_STATE_NORMAL,
+                                            cls::rbd::TRASH_IMAGE_STATE_RESTORING);
+      if (ret < 0 && ret != -EOPNOTSUPP) {
+        lderr(cct) << "error setting trash image state: "
+                   << cpp_strerror(ret) << dendl;
+      }
       return r;
     } else if (r != -ENOENT){
       // checking if we are recovering from an incomplete restore
       if (existing_id != image_id) {
         ldout(cct, 2) << "an image with the same name already exists" << dendl;
+        int r2 = cls_client::trash_state_set(&io_ctx, image_id,
+                                             cls::rbd::TRASH_IMAGE_STATE_NORMAL,
+                                             cls::rbd::TRASH_IMAGE_STATE_RESTORING);
+        if (r2 < 0 && r2 != -EOPNOTSUPP) {
+        lderr(cct) << "error setting trash image state: "
+                   << cpp_strerror(r2) << dendl;
+       }
         return -EEXIST;
       }
       create_id_obj = false;