cls_rbd: mirroring directory reverse lookup from global id to image id

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 67f0aa87e57117e5d834855260ba8957e64e42a4..b3c7fc696b380e55a326bbb66d86fb477c0e9c52 100644 (file)
--- a/src/cls/rbd/cls_rbd.cc
+++ b/src/cls/rbd/cls_rbd.cc
@@ -121,6 +121,7 @@ cls_method_handle_t h_mirror_peer_remove;
 cls_method_handle_t h_mirror_peer_set_client;
 cls_method_handle_t h_mirror_peer_set_cluster;
 cls_method_handle_t h_mirror_image_list;
+cls_method_handle_t h_mirror_image_get_image_id;
 cls_method_handle_t h_mirror_image_get;
 cls_method_handle_t h_mirror_image_set;
 cls_method_handle_t h_mirror_image_remove;
@@ -2960,6 +2961,7 @@ static const std::string UUID("mirror_uuid");
 static const std::string MODE("mirror_mode");
 static const std::string PEER_KEY_PREFIX("mirror_peer_");
 static const std::string IMAGE_KEY_PREFIX("image_");
+static const std::string GLOBAL_KEY_PREFIX("global_");
 
 std::string peer_key(const std::string &uuid) {
   return PEER_KEY_PREFIX + uuid;
@@ -2969,6 +2971,10 @@ std::string image_key(const string &image_id) {
   return IMAGE_KEY_PREFIX + image_id;
 }
 
+std::string global_key(const string &global_id) {
+  return GLOBAL_KEY_PREFIX + global_id;
+}
+
 int uuid_get(cls_method_context_t hctx, std::string *mirror_uuid) {
   bufferlist mirror_uuid_bl;
   int r = cls_cxx_map_get_val(hctx, mirror::UUID, &mirror_uuid_bl);
@@ -3074,19 +3080,24 @@ int image_set(cls_method_context_t hctx, const string &image_id,
   bufferlist bl;
   ::encode(mirror_image, bl);
 
-  // don't overwrite the key if it already exists with a different
-  // global_image_id
   cls::rbd::MirrorImage existing_mirror_image;
   int r = image_get(hctx, image_id, &existing_mirror_image);
-  if (r < 0 && r != -ENOENT) {
+  if (r == -ENOENT) {
+    // make sure global id doesn't already exist
+    std::string global_id_key = global_key(mirror_image.global_image_id);
+    std::string image_id;
+    r = read_key(hctx, global_id_key, &image_id);
+    if (r != -ENOENT) {
+      return -EEXIST;
+    }
+  } else if (r < 0) {
     CLS_ERR("error reading mirrored image '%s': '%s'", image_id.c_str(),
            cpp_strerror(r).c_str());
     return r;
-  }
-
-  if (r != -ENOENT &&
-      existing_mirror_image.global_image_id != mirror_image.global_image_id) {
-    return -EEXIST;
+  } else if (existing_mirror_image.global_image_id !=
+                mirror_image.global_image_id) {
+    // cannot change the global id
+    return -EINVAL;
   }
 
   r = cls_cxx_map_set_val(hctx, image_key(image_id), &bl);
@@ -3095,6 +3106,16 @@ int image_set(cls_method_context_t hctx, const string &image_id,
             cpp_strerror(r).c_str());
     return r;
   }
+
+  bufferlist image_id_bl;
+  ::encode(image_id, image_id_bl);
+  r = cls_cxx_map_set_val(hctx, global_key(mirror_image.global_image_id),
+                          &image_id_bl);
+  if (r < 0) {
+    CLS_ERR("error adding global id for image '%s': %s", image_id.c_str(),
+            cpp_strerror(r).c_str());
+    return r;
+  }
   return 0;
 }
 
@@ -3120,6 +3141,13 @@ int image_remove(cls_method_context_t hctx, const string &image_id) {
             cpp_strerror(r).c_str());
     return r;
   }
+
+  r = cls_cxx_map_remove_key(hctx, global_key(mirror_image.global_image_id));
+  if (r < 0 && r != -ENOENT) {
+    CLS_ERR("error removing global id for image '%s': %s", image_id.c_str(),
+           cpp_strerror(r).c_str());
+    return r;
+  }
   return 0;
 }
 
@@ -3527,6 +3555,36 @@ int mirror_image_list(cls_method_context_t hctx, bufferlist *in,
   return 0;
 }
 
+/**
+ * Input:
+ * @param global_id (std::string)
+ *
+ * Output:
+ * @param std::string - image id
+ * @returns 0 on success, negative error code on failure
+ */
+int mirror_image_get_image_id(cls_method_context_t hctx, bufferlist *in,
+                              bufferlist *out) {
+  std::string global_id;
+  try {
+    bufferlist::iterator it = in->begin();
+    ::decode(global_id, it);
+  } catch (const buffer::error &err) {
+    return -EINVAL;
+  }
+
+  std::string image_id;
+  int r = read_key(hctx, mirror::global_key(global_id), &image_id);
+  if (r < 0) {
+    CLS_ERR("error retrieving image id for global id '%s': %s",
+            global_id.c_str(), cpp_strerror(r).c_str());
+    return r;
+  }
+
+  ::encode(image_id, *out);
+  return 0;
+}
+
 /**
  * Input:
  * @param image_id (std::string)
@@ -3791,6 +3849,9 @@ void __cls_init()
                           mirror_peer_set_cluster, &h_mirror_peer_set_cluster);
   cls_register_cxx_method(h_class, "mirror_image_list", CLS_METHOD_RD,
                           mirror_image_list, &h_mirror_image_list);
+  cls_register_cxx_method(h_class, "mirror_image_get_image_id", CLS_METHOD_RD,
+                          mirror_image_get_image_id,
+                          &h_mirror_image_get_image_id);
   cls_register_cxx_method(h_class, "mirror_image_get", CLS_METHOD_RD,
                           mirror_image_get, &h_mirror_image_get);
   cls_register_cxx_method(h_class, "mirror_image_set",

diff --git a/src/cls/rbd/cls_rbd_client.cc b/src/cls/rbd/cls_rbd_client.cc
index a3dc011503416d421396ddd2b6aa53ecd2db3a83..5fe8452028a9fe0f49eea24aea9c488a82e7f6e6 100644 (file)
--- a/src/cls/rbd/cls_rbd_client.cc
+++ b/src/cls/rbd/cls_rbd_client.cc
@@ -1154,6 +1154,39 @@ namespace librbd {
       return 0;
     }
 
+    void mirror_image_get_image_id_start(librados::ObjectReadOperation *op,
+                                         const std::string &global_image_id) {
+      bufferlist in_bl;
+      ::encode(global_image_id, in_bl);
+      op->exec( "rbd", "mirror_image_get_image_id", in_bl);
+    }
+
+    int mirror_image_get_image_id_finish(bufferlist::iterator *it,
+                                         std::string *image_id) {
+      try {
+       ::decode(*image_id, *it);
+      } catch (const buffer::error &err) {
+       return -EBADMSG;
+      }
+      return 0;
+    }
+
+    int mirror_image_get_image_id(librados::IoCtx *ioctx,
+                                  const std::string &global_image_id,
+                                  std::string *image_id) {
+      librados::ObjectReadOperation op;
+      mirror_image_get_image_id_start(&op, global_image_id);
+
+      bufferlist out_bl;
+      int r = ioctx->operate(RBD_MIRRORING, &op, &out_bl);
+      if (r < 0) {
+        return r;
+      }
+
+      bufferlist::iterator it = out_bl.begin();
+      return mirror_image_get_image_id_finish(&it, image_id);
+    }
+
     int mirror_image_get(librados::IoCtx *ioctx, const std::string &image_id,
                         cls::rbd::MirrorImage *mirror_image) {
       bufferlist in_bl;

diff --git a/src/cls/rbd/cls_rbd_client.h b/src/cls/rbd/cls_rbd_client.h
index de9ac7cf8189c60081f3fda30ec3186c8f283ec8..8bc8ee2b98a733f5fe0a827a04a0ce7e283cff1b 100644 (file)
--- a/src/cls/rbd/cls_rbd_client.h
+++ b/src/cls/rbd/cls_rbd_client.h
@@ -230,6 +230,13 @@ namespace librbd {
     int mirror_image_list(librados::IoCtx *ioctx,
                          const std::string &start, uint64_t max_return,
                           std::map<std::string, std::string> *mirror_image_ids);
+    void mirror_image_get_image_id_start(librados::ObjectReadOperation *op,
+                                         const std::string &global_image_id);
+    int mirror_image_get_image_id_finish(bufferlist::iterator *it,
+                                         std::string *image_id);
+    int mirror_image_get_image_id(librados::IoCtx *ioctx,
+                                  const std::string &global_image_id,
+                                  std::string *image_id);
     int mirror_image_get(librados::IoCtx *ioctx, const std::string &image_id,
                         cls::rbd::MirrorImage *mirror_image);
     int mirror_image_set(librados::IoCtx *ioctx, const std::string &image_id,

diff --git a/src/test/cls_rbd/test_cls_rbd.cc b/src/test/cls_rbd/test_cls_rbd.cc
index a2055ecc3b364f6a8d7055de3554e09f535419d7..64f5b54f54de616e266b596c38ccec4e1a21adf6 100644 (file)
--- a/src/test/cls_rbd/test_cls_rbd.cc
+++ b/src/test/cls_rbd/test_cls_rbd.cc
@@ -1382,10 +1382,14 @@ TEST_F(TestClsRbd, mirror_image) {
 
   ASSERT_EQ(0, mirror_image_set(&ioctx, "image_id1", image1));
   ASSERT_EQ(0, mirror_image_set(&ioctx, "image_id2", image2));
-  ASSERT_EQ(-EEXIST, mirror_image_set(&ioctx, "image_id1", image2));
-  ASSERT_EQ(-EEXIST, mirror_image_set(&ioctx, "image_id2", image3));
+  ASSERT_EQ(-EINVAL, mirror_image_set(&ioctx, "image_id1", image2));
+  ASSERT_EQ(-EEXIST, mirror_image_set(&ioctx, "image_id3", image2));
   ASSERT_EQ(0, mirror_image_set(&ioctx, "image_id3", image3));
 
+  std::string image_id;
+  ASSERT_EQ(0, mirror_image_get_image_id(&ioctx, "uuid2", &image_id));
+  ASSERT_EQ("image_id2", image_id);
+
   cls::rbd::MirrorImage read_image;
   ASSERT_EQ(0, mirror_image_get(&ioctx, "image_id1", &read_image));
   ASSERT_EQ(read_image, image1);
@@ -1404,6 +1408,7 @@ TEST_F(TestClsRbd, mirror_image) {
   ASSERT_EQ(expected_mirror_image_ids, mirror_image_ids);
 
   ASSERT_EQ(0, mirror_image_remove(&ioctx, "image_id2"));
+  ASSERT_EQ(-ENOENT, mirror_image_get_image_id(&ioctx, "uuid2", &image_id));
   ASSERT_EQ(-EBUSY, mirror_image_remove(&ioctx, "image_id1"));
 
   ASSERT_EQ(0, mirror_image_list(&ioctx, "", 3, &mirror_image_ids));