cls/rbd: trash_list should be iterable

Fixes: http://tracker.ceph.com/issues/20643
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 3eb64a45a5c88bd0b96be5fc7e65dd95c46e57de..53a0d4efae97670c86393984032e40a5a7a568fc 100644 (file)
--- a/src/cls/rbd/cls_rbd.cc
+++ b/src/cls/rbd/cls_rbd.cc
@@ -5016,6 +5016,11 @@ int trash_remove(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
  * Returns the list of trash spec entries registered in the rbd_trash
  * object.
  *
+ * Input:
+ * @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 list
+ *
  * Output:
  * @param data the map between image id and trash spec info
  *
@@ -5023,18 +5028,30 @@ int trash_remove(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
  */
 int trash_list(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 {
+  string start_after;
+  uint64_t max_return;
+
+  try {
+    bufferlist::iterator iter = in->begin();
+    ::decode(start_after, iter);
+    ::decode(max_return, iter);
+  } catch (const buffer::error &err) {
+    return -EINVAL;
+  }
+
   map<string, cls::rbd::TrashImageSpec> data;
-  string last_read = trash::image_key("");
-  int max_read = RBD_MAX_KEYS_READ;
+  string last_read = trash::image_key(start_after);
 
   CLS_LOG(20, "trash_get_images");
-
-  do {
+  while (data.size() < max_return) {
     map<string, bufferlist> raw_data;
+    int max_read = std::min<int32_t>(RBD_MAX_KEYS_READ,
+                                     max_return - data.size());
     int r = cls_cxx_map_get_vals(hctx, last_read, trash::IMAGE_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());
+      CLS_ERR("failed to read the vals off of disk: %s",
+              cpp_strerror(r).c_str());
       return r;
     }
     if (raw_data.empty()) {
@@ -5051,10 +5068,9 @@ int trash_list(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
     }
 
     last_read = raw_data.rbegin()->first;
-  } while (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 88b5b3d0340e8aa7a90b0c5631ee6047d4e5d32b..0a3ee5555d570c9208ef35693056b770211060d6 100644 (file)
--- a/src/cls/rbd/cls_rbd_client.cc
+++ b/src/cls/rbd/cls_rbd_client.cc
@@ -2046,9 +2046,12 @@ namespace librbd {
       return ioctx->operate(RBD_TRASH, &op);
     }
 
-    void trash_list_start(librados::ObjectReadOperation *op)
+    void trash_list_start(librados::ObjectReadOperation *op,
+                          const std::string &start, uint64_t max_return)
     {
       bufferlist bl;
+      ::encode(start, bl);
+      ::encode(max_return, bl);
       op->exec("rbd", "trash_list", bl);
     }
 
@@ -2067,10 +2070,11 @@ namespace librbd {
     }
 
     int trash_list(librados::IoCtx *ioctx,
+                   const std::string &start, uint64_t max_return,
                    map<string, cls::rbd::TrashImageSpec> *entries)
     {
       librados::ObjectReadOperation op;
-      trash_list_start(&op);
+      trash_list_start(&op, start, max_return);
 
       bufferlist out_bl;
       int r = ioctx->operate(RBD_TRASH, &op, &out_bl);

diff --git a/src/cls/rbd/cls_rbd_client.h b/src/cls/rbd/cls_rbd_client.h
index 754017f2deaf856091c4d9719875b6999c1525e9..ed2c02c714e10c1a40921ae8d9f3a90bd0c12831 100644 (file)
--- a/src/cls/rbd/cls_rbd_client.h
+++ b/src/cls/rbd/cls_rbd_client.h
@@ -418,10 +418,12 @@ namespace librbd {
     void trash_remove(librados::ObjectWriteOperation *op,
                       const std::string &id);
     int trash_remove(librados::IoCtx *ioctx, const std::string &id);
-    void trash_list_start(librados::ObjectReadOperation *op);
+    void trash_list_start(librados::ObjectReadOperation *op,
+                          const std::string &start, uint64_t max_return);
     int trash_list_finish(bufferlist::iterator *it,
                           map<string, cls::rbd::TrashImageSpec> *entries);
     int trash_list(librados::IoCtx *ioctx,
+                   const std::string &start, uint64_t max_return,
                    map<string, cls::rbd::TrashImageSpec> *entries);
     void trash_get_start(librados::ObjectReadOperation *op,
                          const std::string &id);

diff --git a/src/librbd/internal.cc b/src/librbd/internal.cc
index a712851c8c388449a4f19e91d7165bcc0120f283..70c706c6d4beec42e45430c0e2f894f157ad6363 100644 (file)
--- a/src/librbd/internal.cc
+++ b/src/librbd/internal.cc
@@ -1465,25 +1465,36 @@ int validate_pool(IoCtx &io_ctx, CephContext *cct) {
     CephContext *cct((CephContext *)io_ctx.cct());
     ldout(cct, 20) << "trash_list " << &io_ctx << dendl;
 
-    map<string, cls::rbd::TrashImageSpec> trash_entries;
-    int r = cls_client::trash_list(&io_ctx,  &trash_entries);
-    if (r < 0) {
-      if (r != -ENOENT) {
-        lderr(cct) << "error listing rbd_trash entries: " << cpp_strerror(r)
+    bool more_entries;
+    uint32_t max_read = 1024;
+    std::string last_read = "";
+    do {
+      map<string, cls::rbd::TrashImageSpec> trash_entries;
+      int r = cls_client::trash_list(&io_ctx, last_read, max_read,
+                                     &trash_entries);
+      if (r < 0 && r != -ENOENT) {
+        lderr(cct) << "error listing rbd trash entries: " << cpp_strerror(r)
                    << dendl;
-      } else {
-        r = 0;
+        return r;
+      } else if (r == -ENOENT) {
+        break;
       }
-      return r;
-    }
 
-    for (const auto &entry : trash_entries) {
-      rbd_trash_image_source_t source =
-          static_cast<rbd_trash_image_source_t>(entry.second.source);
-      entries.push_back({entry.first, entry.second.name, source,
-                         entry.second.deletion_time.sec(),
-                         entry.second.deferment_end_time.sec()});
-    }
+      if (trash_entries.empty()) {
+        break;
+      }
+
+      for (const auto &entry : trash_entries) {
+        rbd_trash_image_source_t source =
+            static_cast<rbd_trash_image_source_t>(entry.second.source);
+        entries.push_back({entry.first, entry.second.name, source,
+                           entry.second.deletion_time.sec(),
+                           entry.second.deferment_end_time.sec()});
+      }
+      last_read = trash_entries.rbegin()->first;
+      more_entries = (trash_entries.size() >= max_read);
+    } while (more_entries);
+
     return 0;
   }
 

diff --git a/src/test/cls_rbd/test_cls_rbd.cc b/src/test/cls_rbd/test_cls_rbd.cc
index 8ec0165cbb1e193767b2e063110daa764c8432de..fb274391471ae0ac4c7fccf90b681fc5bf42e94f 100644 (file)
--- a/src/test/cls_rbd/test_cls_rbd.cc
+++ b/src/test/cls_rbd/test_cls_rbd.cc
@@ -2248,7 +2248,7 @@ TEST_F(TestClsRbd, trash_methods)
   string id2 = "123456780";
 
   std::map<string, cls::rbd::TrashImageSpec> entries;
-  ASSERT_EQ(-ENOENT, trash_list(&ioctx, &entries));
+  ASSERT_EQ(-ENOENT, trash_list(&ioctx, "", 1024, &entries));
 
   utime_t now1 = ceph_clock_now();
   utime_t now1_delay = now1;
@@ -2267,27 +2267,28 @@ TEST_F(TestClsRbd, trash_methods)
   ASSERT_EQ(0, trash_remove(&ioctx, id));
   ASSERT_EQ(-ENOENT, trash_remove(&ioctx, id));
 
-  ASSERT_EQ(0, trash_list(&ioctx, &entries));
+  ASSERT_EQ(0, trash_list(&ioctx, "", 1024, &entries));
   ASSERT_TRUE(entries.empty());
 
   ASSERT_EQ(0, trash_add(&ioctx, id, trash_spec2));
   ASSERT_EQ(0, trash_add(&ioctx, id2, trash_spec));
 
-  ASSERT_EQ(0, trash_list(&ioctx, &entries));
-
-  for (auto& entry : entries) {
-    if (entry.first == id) {
-      ASSERT_EQ(entry.second.source, cls::rbd::TRASH_IMAGE_SOURCE_MIRRORING);
-      ASSERT_EQ(entry.second.name, "name2");
-      ASSERT_EQ(entry.second.deletion_time, now2);
-      ASSERT_EQ(entry.second.deferment_end_time, now2_delay);
-    } else if (entry.first == id2) {
-      ASSERT_EQ(entry.second.source, cls::rbd::TRASH_IMAGE_SOURCE_USER);
-      ASSERT_EQ(entry.second.name, "name");
-      ASSERT_EQ(entry.second.deletion_time, now1);
-      ASSERT_EQ(entry.second.deferment_end_time, now1_delay);
-    }
-  }
+  ASSERT_EQ(0, trash_list(&ioctx, "", 1, &entries));
+  ASSERT_TRUE(entries.find(id2) != entries.end());
+  ASSERT_EQ(cls::rbd::TRASH_IMAGE_SOURCE_USER, entries[id2].source);
+  ASSERT_EQ(std::string("name"), entries[id2].name);
+  ASSERT_EQ(now1, entries[id2].deletion_time);
+  ASSERT_EQ(now1_delay, entries[id2].deferment_end_time);
+
+  ASSERT_EQ(0, trash_list(&ioctx, id2, 1, &entries));
+  ASSERT_TRUE(entries.find(id) != entries.end());
+  ASSERT_EQ(cls::rbd::TRASH_IMAGE_SOURCE_MIRRORING, entries[id].source);
+  ASSERT_EQ(std::string("name2"), entries[id].name);
+  ASSERT_EQ(now2, entries[id].deletion_time);
+  ASSERT_EQ(now2_delay, entries[id].deferment_end_time);
+
+  ASSERT_EQ(0, trash_list(&ioctx, id, 1, &entries));
+  ASSERT_TRUE(entries.empty());
 
   cls::rbd::TrashImageSpec spec_res1;
   ASSERT_EQ(0, trash_get(&ioctx, id, &spec_res1));