Librbd: Add "start" and "max" arguments to metadata_list

Signed-off-by: Haomai Wang <haomaiwang@gmail.com>

diff --git a/src/cls/rbd/cls_rbd.cc b/src/cls/rbd/cls_rbd.cc
index 11b1685c1df55a9cb8d321760ca4f27890481a01..1c79473fe8456a265f5ad218ab9914079d67ebe0 100644 (file)
--- a/src/cls/rbd/cls_rbd.cc
+++ b/src/cls/rbd/cls_rbd.cc
@@ -105,7 +105,6 @@ cls_method_handle_t h_old_snapshot_add;
 cls_method_handle_t h_old_snapshot_remove;
 
 #define RBD_MAX_KEYS_READ 64
-#define RBD_MAX_METAADATA_KEYS 1024
 #define RBD_SNAP_KEY_PREFIX "snapshot_"
 #define RBD_DIR_ID_KEY_PREFIX "id_"
 #define RBD_DIR_NAME_KEY_PREFIX "name_"
@@ -2128,29 +2127,54 @@ static const string metadata_name_from_key(const string &key)
 
 /**
  * Input:
- * @param in ignored
- *
+ * @param start_after which name to begin listing after
+ *        (use the empty string to start at the beginning)
+ * @param max_return the maximum number of names to lis(if 0 means no limit)
+
  * Output:
  * @param value
  * @returns 0 on success, negative error code on failure
  */
 int metadata_list(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 {
-  map<string, bufferlist> raw_data, data;
-  int r = cls_cxx_map_get_vals(hctx, RBD_METADATA_KEY_PREFIX, RBD_METADATA_KEY_PREFIX,
-                               RBD_MAX_METAADATA_KEYS, &raw_data);
-  if (r < 0) {
-    CLS_ERR("failed to read the vals off of disk: %s", cpp_strerror(r).c_str());
-    return r;
-  }
+  string start_after;
+  uint64_t max_return;
 
-  bufferlist::iterator iter;
-  string v;
-  for (map<string, bufferlist>::iterator it = raw_data.begin(); it != raw_data.end(); ++it) {
-    assert(it->first.find(RBD_METADATA_KEY_PREFIX, 0) == 0);
-    data[metadata_name_from_key(it->first)].swap(it->second);
+  try {
+    bufferlist::iterator iter = in->begin();
+    ::decode(start_after, iter);
+    ::decode(max_return, iter);
+  } catch (const buffer::error &err) {
+    return -EINVAL;
   }
 
+  map<string, bufferlist> data;
+  string last_read = metadata_key_for_name(start_after);
+  int max_read = max_return ? MIN(RBD_MAX_KEYS_READ, max_return) : RBD_MAX_KEYS_READ;
+  int r;
+
+  do {
+    map<string, bufferlist> raw_data;
+    r = cls_cxx_map_get_vals(hctx, last_read, RBD_METADATA_KEY_PREFIX,
+                             max_read, &raw_data);
+    if (r < 0) {
+      CLS_ERR("failed to read the vals off of disk: %s", cpp_strerror(r).c_str());
+      return r;
+    }
+    if (raw_data.empty())
+      break;
+
+    map<string, bufferlist>::iterator it = raw_data.begin();
+    if (metadata_name_from_key(it->first) == last_read)
+        ++it;
+    for (; it != raw_data.end(); ++it)
+      data[metadata_name_from_key(it->first)].swap(it->second);
+
+    last_read = raw_data.rbegin()->first;
+    if (max_return)
+      max_read = MIN(RBD_MAX_KEYS_READ, max_return-data.size());
+  } while (max_return && max_read);
+
   ::encode(data, *out);
   return 0;
 }

diff --git a/src/cls/rbd/cls_rbd_client.cc b/src/cls/rbd/cls_rbd_client.cc
index 9a3561ca522d5d1878f59a251e30eff7ef992731..9849881b1eb326c8e9f9f28a6f344a882b05fb15 100644 (file)
--- a/src/cls/rbd/cls_rbd_client.cc
+++ b/src/cls/rbd/cls_rbd_client.cc
@@ -768,10 +768,13 @@ namespace librbd {
     }
 
     int metadata_list(librados::IoCtx *ioctx, const std::string &oid,
-                     map<string, bufferlist> *pairs)
+                      const std::string &start, uint64_t max_return,
+                      map<string, bufferlist> *pairs)
     {
       assert(pairs);
       bufferlist in, out;
+      ::encode(start, in);
+      ::encode(max_return, in);
       int r = ioctx->exec(oid, "rbd", "metadata_list", in, out);
       if (r < 0)
         return r;

diff --git a/src/cls/rbd/cls_rbd_client.h b/src/cls/rbd/cls_rbd_client.h
index 3ab7b8f70bbe5c0d740413ed06289e1843ad40d2..cd6f187ff7c1516e6d72be9b72aef6d58b2c0e6a 100644 (file)
--- a/src/cls/rbd/cls_rbd_client.h
+++ b/src/cls/rbd/cls_rbd_client.h
@@ -88,6 +88,7 @@ namespace librbd {
     int set_stripe_unit_count(librados::IoCtx *ioctx, const std::string &oid,
                              uint64_t stripe_unit, uint64_t stripe_count);
     int metadata_list(librados::IoCtx *ioctx, const std::string &oid,
+                      const std::string &start, uint64_t max_return,
                       map<string, bufferlist> *pairs);
     int metadata_set(librados::IoCtx *ioctx, const std::string &oid,
                      const map<std::string, bufferlist> &data);

diff --git a/src/include/rbd/librbd.h b/src/include/rbd/librbd.h
index c578840553b519872cb791d4a54b849abd897ae8..506be78766130adf50c22253f36dae49937665cb 100644 (file)
--- a/src/include/rbd/librbd.h
+++ b/src/include/rbd/librbd.h
@@ -458,14 +458,18 @@ CEPH_RBD_API int rbd_metadata_remove(rbd_image_t image, const char *key);
  * of the image, with a '\0' after each.
  *
  * @param image which image (and implicitly snapshot) to list clones of
- * @param pools buffer in which to store pool names
- * @param pools_len number of bytes in pools buffer
- * @param images buffer in which to store image names
- * @param images_len number of bytes in images buffer
+ * @param start_after which name to begin listing after
+ *        (use the empty string to start at the beginning)
+ * @param max the maximum number of names to lis(if 0 means no limit)
+ * @param keys buffer in which to store pool names
+ * @param keys_len number of bytes in pools buffer
+ * @param values buffer in which to store image names
+ * @param vals_len number of bytes in images buffer
  * @returns number of children on success, negative error code on failure
  * @returns -ERANGE if either buffer is too short
  */
-CEPH_RBD_API int rbd_metadata_list(rbd_image_t image,  char *key, size_t *key_len, char *value, size_t *val_len);
+CEPH_RBD_API int rbd_metadata_list(rbd_image_t image, const char *start, uint64_t max,
+    char *keys, size_t *key_len, char *values, size_t *vals_len);
 
 
 #ifdef __cplusplus

diff --git a/src/include/rbd/librbd.hpp b/src/include/rbd/librbd.hpp
index 6e5df6d182c5508be7260a7e68da21ef79d3934d..67c9c74a72fe65d3dc8fa228cc095c7dbe519577 100644 (file)
--- a/src/include/rbd/librbd.hpp
+++ b/src/include/rbd/librbd.hpp
@@ -247,7 +247,7 @@ public:
   /**
    * Returns a pair of key/value for this image
    */
-  int metadata_list(std::map<std::string, ceph::bufferlist> *pairs);
+  int metadata_list(const std::string &start, uint64_t max, std::map<std::string, ceph::bufferlist> *pairs);
 
 private:
   friend class RBD;

diff --git a/src/librbd/internal.cc b/src/librbd/internal.cc
index e48da6fe0c2ec271d7c7abfb571c140be2de44fa..52f0f9e5e9a554bbcbdacf1d3e7465fafef06dad 100644 (file)
--- a/src/librbd/internal.cc
+++ b/src/librbd/internal.cc
@@ -1144,7 +1144,7 @@ reprotect_and_return_err:
       goto err_close_child;
     }
 
-    r = cls_client::metadata_list(&p_ioctx, p_imctx->header_oid, &pairs);
+    r = cls_client::metadata_list(&p_ioctx, p_imctx->header_oid, "", 0, &pairs);
     if (r < 0) {
       lderr(cct) << "couldn't list metadata: " << r << dendl;
       goto err_close_child;
@@ -2276,7 +2276,7 @@ reprotect_and_return_err:
     int r;
     map<string, bufferlist> pairs;
 
-    r = cls_client::metadata_list(&src->md_ctx, src->header_oid, &pairs);
+    r = cls_client::metadata_list(&src->md_ctx, src->header_oid, "", 0, &pairs);
     if (r < 0) {
       lderr(cct) << "couldn't list metadata: " << r << dendl;
       return r;
@@ -3422,7 +3422,7 @@ reprotect_and_return_err:
     return cls_client::metadata_remove(&ictx->md_ctx, ictx->header_oid, key);
   }
 
-  int metadata_list(ImageCtx *ictx, map<string, bufferlist> *pairs)
+  int metadata_list(ImageCtx *ictx, const string &start, uint64_t max, map<string, bufferlist> *pairs)
   {
     CephContext *cct = ictx->cct;
     ldout(cct, 20) << "metadata_list " << ictx << dendl;
@@ -3432,7 +3432,7 @@ reprotect_and_return_err:
       return r;
     }
 
-    return cls_client::metadata_list(&ictx->md_ctx, ictx->header_oid, pairs);
+    return cls_client::metadata_list(&ictx->md_ctx, ictx->header_oid, start, max, pairs);
   }
   
   int aio_discard(ImageCtx *ictx, uint64_t off, uint64_t len, AioCompletion *c)

diff --git a/src/librbd/internal.h b/src/librbd/internal.h
index 024b91cd1221f3a6c5b57e4fc4cfb9af03344c82..bbd785e58a6aed776eb93b39b11f489e9854ae61 100644 (file)
--- a/src/librbd/internal.h
+++ b/src/librbd/internal.h
@@ -207,7 +207,7 @@ namespace librbd {
   int flush(ImageCtx *ictx);
   int _flush(ImageCtx *ictx);
   int invalidate_cache(ImageCtx *ictx);
-  int metadata_list(ImageCtx *ictx, map<string, bufferlist> *pairs);
+  int metadata_list(ImageCtx *ictx, const string &last, uint64_t max, map<string, bufferlist> *pairs);
   int metadata_get(ImageCtx *ictx, const std::string &key, std::string *value);
   int metadata_set(ImageCtx *ictx, const std::string &key, const std::string &value);
   int metadata_remove(ImageCtx *ictx, const std::string &key);

diff --git a/src/librbd/librbd.cc b/src/librbd/librbd.cc
index 88734f4d629eb2271a5bd3d456406d7cb6c94f23..cdba2cc3655d4860ee5017167032122fc6d25bfd 100644 (file)
--- a/src/librbd/librbd.cc
+++ b/src/librbd/librbd.cc
@@ -823,11 +823,11 @@ namespace librbd {
     return r;
   }
 
-  int Image::metadata_list(map<string, bufferlist> *pairs)
+  int Image::metadata_list(const std::string &start, uint64_t max, map<string, bufferlist> *pairs)
   {
     ImageCtx *ictx = (ImageCtx *)ctx;
     tracepoint(librbd, metadata_list_enter, ictx);
-    int r = librbd::metadata_list(ictx, pairs);
+    int r = librbd::metadata_list(ictx, start, max, pairs);
     if (r >= 0) {
       for (map<string, bufferlist>::iterator it = pairs->begin();
            it != pairs->end(); ++it) {
@@ -1782,12 +1782,13 @@ extern "C" int rbd_metadata_remove(rbd_image_t image, const char *key)
   return r;
 }
 
-extern "C" int rbd_metadata_list(rbd_image_t image,  char *key, size_t *key_len, char *value, size_t *val_len)
+extern "C" int rbd_metadata_list(rbd_image_t image, const char *start, uint64_t max,
+                                 char *key, size_t *key_len, char *value, size_t *val_len)
 {
   librbd::ImageCtx *ictx = (librbd::ImageCtx *)image;
   tracepoint(librbd, metadata_list_enter, ictx);
   map<string, bufferlist> pairs;
-  int r = librbd::metadata_list(ictx, &pairs);
+  int r = librbd::metadata_list(ictx, start, max, &pairs);
   size_t key_total_len = 0, val_total_len = 0;
   bool too_short = false;
   for (map<string, bufferlist>::iterator it = pairs.begin();

diff --git a/src/rbd.cc b/src/rbd.cc
index 8335e2cbcf6155aec9049b2814dce4f533e83749..55f29400c6c6c4b744cfe086ce91d64cad13defd 100644 (file)
--- a/src/rbd.cc
+++ b/src/rbd.cc
@@ -2108,7 +2108,7 @@ static int do_metadata_list(librbd::Image& image, Formatter *f)
   int r;
   TextTable tbl;
 
-  r = image.metadata_list(&pairs);
+  r = image.metadata_list("", 0, &pairs);
   if (r < 0)
     return r;
 

diff --git a/src/test/cls_rbd/test_cls_rbd.cc b/src/test/cls_rbd/test_cls_rbd.cc
index c97d55bed4b83fb22c6799e4df61e308df1e30f8..f095469e02bcad4ec70f5a8dcb4272455312fdfa 100644 (file)
--- a/src/test/cls_rbd/test_cls_rbd.cc
+++ b/src/test/cls_rbd/test_cls_rbd.cc
@@ -1069,7 +1069,6 @@ TEST_F(TestClsRbd, flags)
   ioctx.close();
 }
 
-
 TEST_F(TestClsRbd, metadata)
 {
   librados::IoCtx ioctx;
@@ -1080,7 +1079,7 @@ TEST_F(TestClsRbd, metadata)
 
   map<string, bufferlist> pairs;
   string value;
-  ASSERT_EQ(0, metadata_list(&ioctx, oid, &pairs));
+  ASSERT_EQ(0, metadata_list(&ioctx, oid, "", 0, &pairs));
   ASSERT_TRUE(pairs.empty());
 
   pairs["key1"].append("value1");
@@ -1089,7 +1088,7 @@ TEST_F(TestClsRbd, metadata)
   ASSERT_EQ(0, metadata_get(&ioctx, oid, "key1", &value));
   ASSERT_EQ(0, strcmp("value1", value.c_str()));
   pairs.clear();
-  ASSERT_EQ(0, metadata_list(&ioctx, oid, &pairs));
+  ASSERT_EQ(0, metadata_list(&ioctx, oid, "", 0, &pairs));
   ASSERT_EQ(2U, pairs.size());
   ASSERT_EQ(0, strncmp("value1", pairs["key1"].c_str(), 6));
   ASSERT_EQ(0, strncmp("value2", pairs["key2"].c_str(), 6));
@@ -1098,9 +1097,38 @@ TEST_F(TestClsRbd, metadata)
   ASSERT_EQ(0, metadata_remove(&ioctx, oid, "key1"));
   ASSERT_EQ(0, metadata_remove(&ioctx, oid, "key3"));
   ASSERT_TRUE(metadata_get(&ioctx, oid, "key1", &value) < 0);
-  ASSERT_EQ(0, metadata_list(&ioctx, oid, &pairs));
+  ASSERT_EQ(0, metadata_list(&ioctx, oid, "", 0, &pairs));
   ASSERT_EQ(1U, pairs.size());
   ASSERT_EQ(0, strncmp("value2", pairs["key2"].c_str(), 6));
 
+  pairs.clear();
+  char key[10], val[20];
+  for (int i = 0; i < 1024; i++) {
+    sprintf(key, "key%d", i);
+    sprintf(val, "value%d", i);
+    pairs[key].append(val, strlen(val));
+  }
+  ASSERT_EQ(0, metadata_set(&ioctx, oid, pairs));
+
+  string last_read = "";
+  uint64_t max_read = 48, r;
+  uint64_t size = 0;
+  map<string, bufferlist> data;
+  do {
+    map<string, bufferlist> cur;
+    metadata_list(&ioctx, oid, last_read, max_read, &cur);
+    size += cur.size();
+    for (map<string, bufferlist>::iterator it = cur.begin();
+         it != cur.end(); ++it)
+      data[it->first] = it->second;
+    last_read = cur.rbegin()->first;
+    r = cur.size();
+  } while (r == max_read);
+  ASSERT_EQ(size, 1024U);
+  for (map<string, bufferlist>::iterator it = data.begin();
+       it != data.end(); ++it) {
+    ASSERT_TRUE(it->second.contents_equal(pairs[it->first]));
+  }
+
   ioctx.close();
 }

diff --git a/src/test/librbd/test_librbd.cc b/src/test/librbd/test_librbd.cc
index 4d3c575d846a30a8fbcbc8474ea764c8d1fac2ec..50ef913ea889c91d64ee93f0192db3cb70b4d84c 100644 (file)
--- a/src/test/librbd/test_librbd.cc
+++ b/src/test/librbd/test_librbd.cc
@@ -2614,25 +2614,26 @@ TEST_F(TestLibRBD, Metadata)
   librbd::Image image1;
   ASSERT_EQ(0, rbd.open(ioctx, image1, name.c_str(), NULL));
   map<string, bufferlist> pairs;
-  ASSERT_EQ(0, image1.metadata_list(&pairs));
+  ASSERT_EQ(0, image1.metadata_list("", 0, &pairs));
   ASSERT_TRUE(pairs.empty());
 
   ASSERT_EQ(0, image1.metadata_set("key1", "value1"));
   ASSERT_EQ(0, image1.metadata_set("key2", "value2"));
   ASSERT_EQ(0, image1.metadata_get("key1", &value));
   ASSERT_EQ(0, strcmp("value1", value.c_str()));
-  ASSERT_EQ(0, image1.metadata_list(&pairs));
+  ASSERT_EQ(0, image1.metadata_list("", 0, &pairs));
   ASSERT_EQ(2U, pairs.size());
-  ASSERT_EQ(0, strcmp("value1", pairs["key1"].c_str()));
-  ASSERT_EQ(0, strcmp("value2", pairs["key2"].c_str()));
+  const char * ddd = pairs["key1"].c_str();
+  ASSERT_EQ(0, strncmp("value1", pairs["key1"].c_str(), 6));
+  ASSERT_EQ(0, strncmp("value2", pairs["key2"].c_str(), 6));
 
   pairs.clear();
   ASSERT_EQ(0, image1.metadata_remove("key1"));
   ASSERT_EQ(0, image1.metadata_remove("key3"));
   ASSERT_TRUE(image1.metadata_get("key3", &value) < 0);
-  ASSERT_EQ(0, image1.metadata_list(&pairs));
+  ASSERT_EQ(0, image1.metadata_list("", 0, &pairs));
   ASSERT_EQ(1U, pairs.size());
-  ASSERT_EQ(0, strcmp("value2", pairs["key2"].c_str()));
+  ASSERT_EQ(0, strncmp("value2", pairs["key2"].c_str(), 6));
 
   // test metadata with snapshot adding
   ASSERT_EQ(0, image1.snap_create("snap1"));
@@ -2642,18 +2643,18 @@ TEST_F(TestLibRBD, Metadata)
   pairs.clear();
   ASSERT_EQ(0, image1.metadata_set("key1", "value1"));
   ASSERT_EQ(0, image1.metadata_set("key3", "value3"));
-  ASSERT_EQ(0, image1.metadata_list(&pairs));
+  ASSERT_EQ(0, image1.metadata_list("", 0, &pairs));
   ASSERT_EQ(3U, pairs.size());
-  ASSERT_EQ(0, strcmp("value1", pairs["key1"].c_str()));
-  ASSERT_EQ(0, strcmp("value2", pairs["key2"].c_str()));
-  ASSERT_EQ(0, strcmp("value3", pairs["key3"].c_str()));
+  ASSERT_EQ(0, strncmp("value1", pairs["key1"].c_str(), 6));
+  ASSERT_EQ(0, strncmp("value2", pairs["key2"].c_str(), 6));
+  ASSERT_EQ(0, strncmp("value3", pairs["key3"].c_str(), 6));
 
   ASSERT_EQ(0, image1.snap_set(NULL));
-  ASSERT_EQ(0, image1.metadata_list(&pairs));
+  ASSERT_EQ(0, image1.metadata_list("", 0, &pairs));
   ASSERT_EQ(3U, pairs.size());
-  ASSERT_EQ(0, strcmp("value1", pairs["key1"].c_str()));
-  ASSERT_EQ(0, strcmp("value2", pairs["key2"].c_str()));
-  ASSERT_EQ(0, strcmp("value3", pairs["key3"].c_str()));
+  ASSERT_EQ(0, strncmp("value1", pairs["key1"].c_str(), 6));
+  ASSERT_EQ(0, strncmp("value2", pairs["key2"].c_str(), 6));
+  ASSERT_EQ(0, strncmp("value3", pairs["key3"].c_str(), 6));
 
   // test metadata with cloning
   string cname = get_temp_image_name();
@@ -2664,10 +2665,10 @@ TEST_F(TestLibRBD, Metadata)
   ASSERT_EQ(0, rbd.open(ioctx, image2, cname.c_str(), NULL));
   ASSERT_EQ(0, image2.metadata_set("key4", "value4"));
   pairs.clear();
-  ASSERT_EQ(0, image2.metadata_list(&pairs));
+  ASSERT_EQ(0, image2.metadata_list("", 0, &pairs));
   ASSERT_EQ(4U, pairs.size());
   pairs.clear();
-  ASSERT_EQ(0, image1.metadata_list(&pairs));
+  ASSERT_EQ(0, image1.metadata_list("", 0, &pairs));
   ASSERT_EQ(3U, pairs.size());
   ASSERT_EQ(-ENOENT, image1.metadata_get("key4", &value));
 }