cls_rbd: implement get_parent, set_parent, remove_parent

Implement methods to get/set/remove the parent pointer on child images.

Signed-off-by: Sage Weil <sage@inktank.com>

diff --git a/src/cls_rbd.cc b/src/cls_rbd.cc
index 1aaee76c17333eb9d23e9850a1c2b8d9c9085a33..8685ae85163206cc2722cb8bdd6d9ef2863cbb5f 100644 (file)
--- a/src/cls_rbd.cc
+++ b/src/cls_rbd.cc
@@ -48,6 +48,9 @@ cls_method_handle_t h_create;
 cls_method_handle_t h_get_features;
 cls_method_handle_t h_get_size;
 cls_method_handle_t h_set_size;
+cls_method_handle_t h_get_parent;
+cls_method_handle_t h_set_parent;
+cls_method_handle_t h_remove_parent;
 cls_method_handle_t h_get_snapcontext;
 cls_method_handle_t h_get_object_prefix;
 cls_method_handle_t h_get_snapshot_name;
@@ -72,32 +75,77 @@ cls_method_handle_t h_assign_bid;
 #define RBD_LOCK_EXCLUSIVE "exclusive"
 #define RBD_LOCK_SHARED "shared"
 
+
+/// information about our parent image, if any
+struct cls_rbd_parent {
+  int64_t pool;        ///< parent pool id
+  string id;           ///< parent image id
+  snapid_t snapid;     ///< parent snapid we refer to
+  uint64_t size;       ///< portion of this image mapped onto parent
+
+  /// true if our parent pointer information is defined
+  bool exists() const {
+    return snapid != CEPH_NOSNAP && pool >= 0 && id.length() > 0;
+  }
+
+  cls_rbd_parent() : pool(-1), snapid(CEPH_NOSNAP), size(0) {}
+
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    ::encode(pool, bl);
+    ::encode(id, bl);
+    ::encode(snapid, bl);
+    ::encode(size, bl);
+    ENCODE_FINISH(bl);
+  }
+  void decode(bufferlist::iterator& bl) {
+    DECODE_START(1, bl);
+    ::decode(pool, bl);
+    ::decode(id, bl);
+    ::decode(snapid, bl);
+    ::decode(size, bl);
+    DECODE_FINISH(bl);
+  }
+};
+WRITE_CLASS_ENCODER(cls_rbd_parent)
+
 struct cls_rbd_snap {
   snapid_t id;
   string name;
   uint64_t image_size;
   uint64_t features;
+  cls_rbd_parent parent;
+
+  /// true if we have a parent
+  bool has_parent() const {
+    return parent.exists();
+  }
 
   cls_rbd_snap() : id(CEPH_NOSNAP), image_size(0), features(0) {}
   void encode(bufferlist& bl) const {
-    ENCODE_START(1, 1, bl);
+    ENCODE_START(2, 1, bl);
     ::encode(id, bl);
     ::encode(name, bl);
     ::encode(image_size, bl);
     ::encode(features, bl);
+    ::encode(parent, bl);
     ENCODE_FINISH(bl);
   }
   void decode(bufferlist::iterator& p) {
-    DECODE_START(1, p);
+    DECODE_START(2, p);
     ::decode(id, p);
     ::decode(name, p);
     ::decode(image_size, p);
     ::decode(features, p);
+    if (struct_v >= 2) {
+      ::decode(parent, p);
+    }
     DECODE_FINISH(p);
   }
 };
 WRITE_CLASS_ENCODER(cls_rbd_snap)
 
+
 static int snap_read_header(cls_method_context_t hctx, bufferlist& bl)
 {
   unsigned snap_count = 0;
@@ -628,6 +676,163 @@ int list_locks(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
   return r;
 }
 
+
+/**
+ * verify that the header object exists
+ *
+ * @return 0 if the object exists, -ENOENT if it does not, or other error
+ */
+int check_exists(cls_method_context_t hctx)
+{
+  uint64_t size;
+  time_t mtime;
+  return cls_cxx_stat(hctx, &size, &mtime);
+}
+
+/**
+ * Input:
+ * @param snap_id which snapshot to query, or CEPH_NOSNAP (uint64_t)
+ *
+ * Output:
+ * @param pool parent pool id
+ * @param image parent image id
+ * @param snapid parent snapid
+ * @param size portion of parent mapped under the child
+ * @returns 0 on success, negative error code on failure
+ */
+int get_parent(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
+{
+  uint64_t snap_id;
+
+  bufferlist::iterator iter = in->begin();
+  try {
+    ::decode(snap_id, iter);
+  } catch (const buffer::error &err) {
+    return -EINVAL;
+  }
+
+  int r = check_exists(hctx);
+  if (r < 0)
+    return r;
+
+  CLS_LOG(20, "get_parent snap_id=%llu", snap_id);
+
+  cls_rbd_parent parent;
+  if (snap_id == CEPH_NOSNAP) {
+    r = read_key(hctx, "parent", &parent);
+    if (r < 0)
+      return r;
+  } else {
+    cls_rbd_snap snap;
+    string snapshot_key;
+    key_from_snap_id(snap_id, &snapshot_key);
+    r = read_key(hctx, snapshot_key, &snap);
+    if (r < 0)
+      return r;
+    parent = snap.parent;
+  }
+
+  if (!parent.exists())
+    return -ENOENT;
+
+  ::encode(parent.pool, *out);
+  ::encode(parent.id, *out);
+  ::encode(parent.snapid, *out);
+  ::encode(parent.size, *out);
+  return 0;
+}
+
+/**
+ * set the image parent
+ *
+ * @param pool parent pool
+ * @param id parent image id
+ * @param snapid parent snapid
+ * @param size parent size
+ * @returns 0 on success, or negative error code
+ */
+int set_parent(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
+{
+  int64_t pool;
+  string id;
+  snapid_t snapid;
+  uint64_t size;
+
+  bufferlist::iterator iter = in->begin();
+  try {
+    ::decode(pool, iter);
+    ::decode(id, iter);
+    ::decode(snapid, iter);
+    ::decode(size, iter);
+  } catch (const buffer::error &err) {
+    return -EINVAL;
+  }
+
+  int r = check_exists(hctx);
+  if (r < 0)
+    return r;
+
+  CLS_LOG(20, "set_parent pool=%lld id=%s snapid=%llu size=%llu",
+         pool, id.c_str(), snapid.val, size);
+
+  if (pool < 0 || id.length() == 0 || snapid == CEPH_NOSNAP || size == 0) {
+    return -EINVAL;
+  }
+
+  // make sure there isn't already a parent
+  cls_rbd_parent parent;
+  r = read_key(hctx, "parent", &parent);
+  if (r == 0) {
+    CLS_LOG(20, "set_parent existing parent pool=%lld id=%s snapid=%llu size=%llu",
+           parent.pool, parent.id.c_str(), parent.snapid.val,
+           parent.size);
+    return -EEXIST;
+  }
+
+  bufferlist parentbl;
+  parent.pool = pool;
+  parent.id = id;
+  parent.snapid = snapid;
+  parent.size = size;
+  ::encode(parent, parentbl);
+  r = cls_cxx_map_set_val(hctx, "parent", &parentbl);
+  if (r < 0) {
+    CLS_ERR("error writing parent: %d", r);
+    return r;
+  }
+
+  return 0;
+}
+
+
+/**
+ * remove the parent pointer
+ *
+ * This can only happen on the head, not on a snapshot.  No arguments.
+ *
+ * @returns 0 on success, negative error code on failure.
+ */
+int remove_parent(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
+{
+  int r = check_exists(hctx);
+  if (r < 0)
+    return r;
+
+  cls_rbd_parent parent;
+  r = read_key(hctx, "parent", &parent);
+  if (r < 0)
+    return r;
+
+  r = cls_cxx_map_remove_key(hctx, "parent");
+  if (r < 0) {
+    CLS_ERR("error removing parent: %d", r);
+    return r;
+  }
+
+  return 0;
+}
+
+
 /**
  * Get the information needed to create a rados snap context for doing
  * I/O to the data objects. This must include all snapshots.
@@ -1171,6 +1376,15 @@ void __cls_init()
   cls_register_cxx_method(h_class, "list_locks",
                           CLS_METHOD_RD | CLS_METHOD_PUBLIC,
                           list_locks, &h_list_locks);
+  cls_register_cxx_method(h_class, "get_parent",
+                         CLS_METHOD_RD | CLS_METHOD_PUBLIC,
+                         get_parent, &h_get_parent);
+  cls_register_cxx_method(h_class, "set_parent",
+                         CLS_METHOD_RD | CLS_METHOD_WR | CLS_METHOD_PUBLIC,
+                         set_parent, &h_set_parent);
+  cls_register_cxx_method(h_class, "remove_parent",
+                         CLS_METHOD_RD | CLS_METHOD_WR | CLS_METHOD_PUBLIC,
+                         remove_parent, &h_remove_parent);
 
   /* methods for the old format */
   cls_register_cxx_method(h_class, "snap_list",

diff --git a/src/librbd/cls_rbd_client.cc b/src/librbd/cls_rbd_client.cc
index e6d7c814d7bf018b3794fe9433fca114a27c919c..b03b202907fea6aa0ecc6eea05e02ed4b4ed3ec0 100644 (file)
--- a/src/librbd/cls_rbd_client.cc
+++ b/src/librbd/cls_rbd_client.cc
@@ -162,6 +162,50 @@ namespace librbd {
       return ioctx->exec(oid, "rbd", "set_size", bl, bl2);
     }
 
+    int get_parent(librados::IoCtx *ioctx, const std::string &oid,
+                  snapid_t snap_id, int64_t *parent_pool,
+                  string *parent_image, snapid_t *parent_snap_id,
+                  uint64_t *parent_size)
+    {
+      bufferlist inbl, outbl;
+      ::encode(snap_id, inbl);
+
+      int r = ioctx->exec(oid, "rbd", "get_parent", inbl, outbl);
+      if (r < 0)
+       return r;
+
+      try {
+       bufferlist::iterator iter = outbl.begin();
+       ::decode(*parent_pool, iter);
+       ::decode(*parent_image, iter);
+       ::decode(*parent_snap_id, iter);
+       ::decode(*parent_size, iter);
+      } catch (const buffer::error &err) {
+       return -EBADMSG;
+      }
+
+      return 0;
+    }
+
+    int set_parent(librados::IoCtx *ioctx, const std::string &oid,
+                  int64_t parent_pool, const string& parent_image,
+                  snapid_t parent_snap_id, uint64_t parent_size)
+    {
+      bufferlist inbl, outbl;
+      ::encode(parent_pool, inbl);
+      ::encode(parent_image, inbl);
+      ::encode(parent_snap_id, inbl);
+      ::encode(parent_size, inbl);
+
+      return ioctx->exec(oid, "rbd", "set_parent", inbl, outbl);
+    }
+
+    int remove_parent(librados::IoCtx *ioctx, const std::string &oid)
+    {
+      bufferlist inbl, outbl;
+      return ioctx->exec(oid, "rbd", "remove_parent", inbl, outbl);
+    }
+
     int snapshot_add(librados::IoCtx *ioctx, const std::string &oid,
                     snapid_t snap_id, const std::string &snap_name)
     {

diff --git a/src/librbd/cls_rbd_client.h b/src/librbd/cls_rbd_client.h
index e801d6343738bffc689d31d1cdc6951fcba0162c..f2263ece778fd3a2f0eb48f2c0342fb603090207 100644 (file)
--- a/src/librbd/cls_rbd_client.h
+++ b/src/librbd/cls_rbd_client.h
@@ -37,6 +37,16 @@ namespace librbd {
                 snapid_t snap_id, uint64_t *size, uint8_t *order);
     int set_size(librados::IoCtx *ioctx, const std::string &oid,
                 uint64_t size);
+    int set_size(librados::IoCtx *ioctx, const std::string &oid,
+                uint64_t size);
+    int get_parent(librados::IoCtx *ioctx, const std::string &oid,
+                  snapid_t snap_id, int64_t *parent_pool,
+                  std::string *parent_image, snapid_t *parent_snap_id,
+                  uint64_t *parent_size);
+    int set_parent(librados::IoCtx *ioctx, const std::string &oid,
+                  int64_t parent_pool, const std::string& parent_image, snapid_t parent_snap_id,
+                  uint64_t parent_size);
+    int remove_parent(librados::IoCtx *ioctx, const std::string &oid);
     int snapshot_add(librados::IoCtx *ioctx, const std::string &oid,
                     snapid_t snap_id, const std::string &snap_name);
     int snapshot_remove(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 62ac0c38a798d1b22212b58a519ecae6eaba6a93..d94bda8ccb68cb3bf66148cc7c297e55d02275e6 100644 (file)
--- a/src/test/rbd/test_cls_rbd.cc
+++ b/src/test/rbd/test_cls_rbd.cc
@@ -21,6 +21,9 @@ using ::librbd::cls_client::get_features;
 using ::librbd::cls_client::get_size;
 using ::librbd::cls_client::get_object_prefix;
 using ::librbd::cls_client::set_size;
+using ::librbd::cls_client::get_parent;
+using ::librbd::cls_client::set_parent;
+using ::librbd::cls_client::remove_parent;
 using ::librbd::cls_client::snapshot_add;
 using ::librbd::cls_client::snapshot_remove;
 using ::librbd::cls_client::get_snapcontext;
@@ -267,6 +270,49 @@ TEST(cls_rbd, set_size)
   ASSERT_EQ(0, destroy_one_pool_pp(pool_name, rados));
 }
 
+TEST(cls_rbd, parents)
+{
+  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));
+
+  int64_t pool;
+  string parent;
+  snapid_t snapid;
+  uint64_t size;
+
+  ASSERT_EQ(-ENOENT, get_parent(&ioctx, "doesnotexist", CEPH_NOSNAP, &pool, &parent, &snapid, &size));
+
+  ASSERT_EQ(0, create_image(&ioctx, "foo", 1000, 22, 0, "foo."));
+
+  ASSERT_EQ(-ENOENT, get_parent(&ioctx, "foo", CEPH_NOSNAP, &pool, &parent, &snapid, &size));
+  ASSERT_EQ(-ENOENT, get_parent(&ioctx, "foo", 123, &pool, &parent, &snapid, &size));
+
+  ASSERT_EQ(-EINVAL, set_parent(&ioctx, "foo", -1, "parent", 3, 10<<20));
+  ASSERT_EQ(-EINVAL, set_parent(&ioctx, "foo", 1, "", 3, 10<<20));
+  ASSERT_EQ(-EINVAL, set_parent(&ioctx, "foo", 1, "parent", CEPH_NOSNAP, 10<<20));
+  ASSERT_EQ(-EINVAL, set_parent(&ioctx, "foo", 1, "parent", 3, 0));
+
+  ASSERT_EQ(0, set_parent(&ioctx, "foo", 1, "parent", 3, 10<<20));
+  ASSERT_EQ(-EEXIST, set_parent(&ioctx, "foo", 1, "parent", 3, 10<<20));
+  ASSERT_EQ(-EEXIST, set_parent(&ioctx, "foo", 2, "parent", 34, 10<<20));
+
+  ASSERT_EQ(0, get_parent(&ioctx, "foo", CEPH_NOSNAP, &pool, &parent, &snapid, &size));
+  ASSERT_EQ(pool, 1);
+  ASSERT_EQ(parent, "parent");
+  ASSERT_EQ(snapid, snapid_t(3));
+
+  ASSERT_EQ(0, remove_parent(&ioctx, "foo"));
+  ASSERT_EQ(-ENOENT, remove_parent(&ioctx, "foo"));
+  ASSERT_EQ(-ENOENT, get_parent(&ioctx, "foo", CEPH_NOSNAP, &pool, &parent, &snapid, &size));
+
+  ioctx.close();
+  ASSERT_EQ(0, destroy_one_pool_pp(pool_name, rados));
+}
+
 TEST(cls_rbd, snapshots)
 {
   librados::Rados rados;