cls_rbd, cls_rbd_client, test_cls_rbd: copyup method

Fixes: #2559
Signed-off-by: Dan Mick <dan.mick@inktank.com>

diff --git a/src/cls_rbd.cc b/src/cls_rbd.cc
index 6035a0f7eb22354fb97e0808d366a33d88384e03..2eb59dc22e596c1e9d5d5d6c877228e63c5c0adf 100644 (file)
--- a/src/cls_rbd.cc
+++ b/src/cls_rbd.cc
@@ -59,6 +59,7 @@ cls_method_handle_t h_get_snapshot_name;
 cls_method_handle_t h_snapshot_add;
 cls_method_handle_t h_snapshot_remove;
 cls_method_handle_t h_get_all_features;
+cls_method_handle_t h_copyup;
 cls_method_handle_t h_lock_image_exclusive;
 cls_method_handle_t h_lock_image_shared;
 cls_method_handle_t h_unlock_image;
@@ -1134,6 +1135,34 @@ int get_all_features(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
   return 0;
 }
 
+/**
+ * "Copy up" data from the parent of a clone to the clone's object(s).
+ * Used for implementing copy-on-write for a clone image.  Client
+ * will pass down a chunk of data that fits completely within one
+ * clone block (one object), and is aligned (starts at beginning of block),
+ * but may be shorter (for non-full parent blocks).  The class method
+ * can't know the object size to validate the requested length,
+ * so it just writes the data as given if the child object doesn't
+ * already exist, and returns success if it does.
+ *
+ * Input:
+ * @param in bufferlist of data to write
+ *
+ * Output:
+ * @returns 0 on success, or if block already exists in child
+ *  negative error code on other error
+ */
+
+int copyup(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
+{
+  // check for existence; if child object exists, just return success
+  if (cls_cxx_stat(hctx, NULL, NULL) == 0)
+    return 0;
+  CLS_LOG(20, "copyup: writing length %d\n", in->length());
+  return cls_cxx_write(hctx, 0, in->length(), in);
+}
+
+
 /************************ rbd_id object methods **************************/
 
 /**
@@ -1810,6 +1839,9 @@ void __cls_init()
   cls_register_cxx_method(h_class, "get_all_features",
                          CLS_METHOD_RD | CLS_METHOD_PUBLIC,
                          get_all_features, &h_get_all_features);
+  cls_register_cxx_method(h_class, "copyup",
+                         CLS_METHOD_RD | CLS_METHOD_WR | CLS_METHOD_PUBLIC,
+                         copyup, &h_copyup);
   cls_register_cxx_method(h_class, "lock_exclusive",
                           CLS_METHOD_RD | CLS_METHOD_WR | CLS_METHOD_PUBLIC,
                           lock_image_exclusive, &h_lock_image_exclusive);

diff --git a/src/librbd/cls_rbd_client.cc b/src/librbd/cls_rbd_client.cc
index 94b0bb0630572aa79b18f36fb6c3b1586700f975..543844782b72dcd55c7e00fb1390b686efa805d9 100644 (file)
--- a/src/librbd/cls_rbd_client.cc
+++ b/src/librbd/cls_rbd_client.cc
@@ -419,6 +419,12 @@ namespace librbd {
       return 0;
     }
 
+    int copyup(librados::IoCtx *ioctx, const std::string &oid,
+              bufferlist data) {
+      bufferlist out;
+      return ioctx->exec(oid, "rbd", "copyup", data, out);
+    }
+
     int lock_image_exclusive(librados::IoCtx *ioctx, const std::string &oid,
                             const std::string &cookie)
     {

diff --git a/src/librbd/cls_rbd_client.h b/src/librbd/cls_rbd_client.h
index 461b431b5496f2f500a66f258e5e5182cf036bd5..48335bb96461f4a6b85dca445aef043f0e15d489 100644 (file)
--- a/src/librbd/cls_rbd_client.h
+++ b/src/librbd/cls_rbd_client.h
@@ -74,6 +74,8 @@ namespace librbd {
     int list_locks(librados::IoCtx *ioctx, const std::string &oid,
                   std::set<std::pair<std::string, std::string> > &locks,
                   bool &exclusive);
+    int copyup(librados::IoCtx *ioctx, const std::string &oid,
+              bufferlist data);
     int lock_image_exclusive(librados::IoCtx *ioctx, const std::string &oid,
                             const std::string &cookie);
     int lock_image_shared(librados::IoCtx *ioctx, const std::string &oid,

diff --git a/src/test/rbd/test_cls_rbd.cc b/src/test/rbd/test_cls_rbd.cc
index 77738e9c8b797f3744f06b88bd91ca002abb3d98..c8c532b9b67d576d53768b5d1e1d15d182fc374c 100644 (file)
--- a/src/test/rbd/test_cls_rbd.cc
+++ b/src/test/rbd/test_cls_rbd.cc
@@ -33,6 +33,7 @@ using ::librbd::cls_client::lock_image_exclusive;
 using ::librbd::cls_client::lock_image_shared;
 using ::librbd::cls_client::unlock_image;
 using ::librbd::cls_client::break_lock;
+using ::librbd::cls_client::copyup;
 using ::librbd::cls_client::get_id;
 using ::librbd::cls_client::set_id;
 using ::librbd::cls_client::dir_get_id;
@@ -43,6 +44,58 @@ using ::librbd::cls_client::dir_remove_image;
 using ::librbd::cls_client::dir_rename_image;
 using ::librbd::cls_client::parent_info;
 
+static char *random_buf(size_t len)
+{
+  char *b = new char[len];
+  for (size_t i = 0; i < len; i++)
+    b[i] = (rand() % (128 - 32)) + 32;
+  return b;
+}
+
+TEST(cls_rbd, copyup)
+{
+  librados::Rados rados;
+  librados::IoCtx ioctx;
+  string pool_name = get_temp_pool_name();
+
+  ASSERT_EQ("", create_one_pool_pp(pool_name, rados));
+  ASSERT_EQ(0, rados.ioctx_create(pool_name.c_str(), ioctx));
+
+  string oid = "rbd_copyup_test";
+  size_t l = 4 << 20;
+  char *b = random_buf(l);
+  bufferlist inbl, outbl;
+  inbl.append(b, l);
+  delete b;
+  ASSERT_EQ(l, inbl.length());
+
+  // copyup to nonexistent object should create new object
+  ioctx.remove(oid);
+  ASSERT_EQ(-ENOENT, ioctx.remove(oid));
+  ASSERT_EQ(0, copyup(&ioctx, oid, inbl));
+  // and its contents should match
+  ASSERT_EQ(l, (size_t)ioctx.read(oid, outbl, l, 0));
+  ASSERT_TRUE(outbl.contents_equal(inbl));
+
+  // now send different data, but with a preexisting object
+  bufferlist inbl2;
+  b = random_buf(l);
+  inbl2.append(b, l);
+  delete b;
+  ASSERT_EQ(l, inbl2.length());
+
+  // should still succeed
+  ASSERT_EQ(0, copyup(&ioctx, oid, inbl));
+  ASSERT_EQ(l, (size_t)ioctx.read(oid, outbl, l, 0));
+  // but contents should not have changed
+  ASSERT_FALSE(outbl.contents_equal(inbl2));
+  ASSERT_TRUE(outbl.contents_equal(inbl));
+
+  ASSERT_EQ(0, ioctx.remove(oid));
+  ioctx.close();
+  ASSERT_EQ(0, destroy_one_pool_pp(pool_name, rados));
+}
+
 TEST(cls_rbd, get_and_set_id)
 {
   librados::Rados rados;
@@ -105,7 +158,7 @@ TEST(cls_rbd, directory_methods)
 
   map<string, string> images;
   ASSERT_EQ(-ENOENT, dir_list(&ioctx, oid, "", 30, &images));
-  
+
   ASSERT_EQ(0, ioctx.create(oid, true));
   ASSERT_EQ(0, dir_list(&ioctx, oid, "", 30, &images));
   ASSERT_EQ(0u, images.size());
@@ -433,7 +486,7 @@ TEST(cls_rbd, parents)
   uint64_t size;
 
   ASSERT_EQ(-ENOENT, get_parent(&ioctx, "doesnotexist", CEPH_NOSNAP, &pool, &parent, &snapid, &size));
-  
+
   // old image should fail
   ASSERT_EQ(0, create_image(&ioctx, "old", 33<<20, 22, 0, "old_blk."));
   ASSERT_EQ(-ENOEXEC, get_parent(&ioctx, "old", CEPH_NOSNAP, &pool, &parent, &snapid, &size));
@@ -592,7 +645,7 @@ TEST(cls_rbd, snapshots)
   vector<uint64_t> snap_features;
   SnapContext snapc;
   vector<parent_info> parents;
-  
+
   ASSERT_EQ(0, get_snapcontext(&ioctx, "foo", &snapc));
   ASSERT_EQ(0u, snapc.snaps.size());
   ASSERT_EQ(0u, snapc.seq);