cls_rgw: fix bucket listing when dealing with invisible entries

Fixes: #12913
The bucket listing logic was broken when dealing with invisible entries.
Need to read keys until we can fetch all the requested entries, or
determine that there are no more entries. Anything else requires
a change to the protocol.

Signed-off-by: Yehuda Sadeh <yehuda@redhat.com>
(cherry picked from commit ef9730515e001d77637a77eb89950f70fd5e7f5a)

diff --git a/src/cls/rgw/cls_rgw.cc b/src/cls/rgw/cls_rgw.cc
index db488e40fb62c7fc800f2260d196eea5263d9af2..980884a9f099b678785f39e0f23e4af3b91eba00 100644 (file)
--- a/src/cls/rgw/cls_rgw.cc
+++ b/src/cls/rgw/cls_rgw.cc
@@ -423,54 +423,65 @@ int rgw_bucket_list(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
   bufferlist bl;
 
   map<string, bufferlist> keys;
+  std::map<string, bufferlist>::iterator kiter;
   string start_key;
   encode_list_index_key(hctx, op.start_obj, &start_key);
-  rc = get_obj_vals(hctx, start_key, op.filter_prefix, op.num_entries + 1, &keys);
-  if (rc < 0)
-    return rc;
+  bool done = false;
+  uint32_t left_to_read = op.num_entries + 1;
 
-  std::map<string, struct rgw_bucket_dir_entry>& m = new_dir.m;
-  std::map<string, bufferlist>::iterator kiter = keys.begin();
-  uint32_t i;
+  do {
+    rc = get_obj_vals(hctx, start_key, op.filter_prefix, left_to_read, &keys);
+    if (rc < 0)
+      return rc;
 
-  bool done = false;
+    std::map<string, struct rgw_bucket_dir_entry>& m = new_dir.m;
 
-  for (i = 0; i < op.num_entries && kiter != keys.end(); ++i, ++kiter) {
-    struct rgw_bucket_dir_entry entry;
+    done = keys.empty();
 
-    if (!bi_is_objs_index(kiter->first)) {
-      done = true;
-      break;
-    }
+    for (kiter = keys.begin(); kiter != keys.end(); ++kiter) {
+      struct rgw_bucket_dir_entry entry;
 
-    bufferlist& entrybl = kiter->second;
-    bufferlist::iterator eiter = entrybl.begin();
-    try {
-      ::decode(entry, eiter);
-    } catch (buffer::error& err) {
-      CLS_LOG(1, "ERROR: rgw_bucket_list(): failed to decode entry, key=%s\n", kiter->first.c_str());
-      return -EINVAL;
-    }
+      if (!bi_is_objs_index(kiter->first)) {
+        done = true;
+        break;
+      }
 
-    cls_rgw_obj_key key;
-    uint64_t ver;
-    decode_list_index_key(kiter->first, &key, &ver);
+      bufferlist& entrybl = kiter->second;
+      bufferlist::iterator eiter = entrybl.begin();
+      try {
+        ::decode(entry, eiter);
+      } catch (buffer::error& err) {
+        CLS_LOG(1, "ERROR: rgw_bucket_list(): failed to decode entry, key=%s\n", kiter->first.c_str());
+        return -EINVAL;
+      }
 
-    if (!entry.is_valid()) {
-      CLS_LOG(20, "entry %s[%s] is not valid\n", key.name.c_str(), key.instance.c_str());
-      continue;
-    }
+      cls_rgw_obj_key key;
+      uint64_t ver;
+      decode_list_index_key(kiter->first, &key, &ver);
 
-    if (!op.list_versions && !entry.is_visible()) {
-      CLS_LOG(20, "entry %s[%s] is not visible\n", key.name.c_str(), key.instance.c_str());
-      continue;
-    }
-    m[kiter->first] = entry;
+      start_key = kiter->first;
+      CLS_LOG(20, "start_key=%s len=%d", start_key.c_str(), start_key.size());
 
-    CLS_LOG(20, "got entry %s[%s] m.size()=%d\n", key.name.c_str(), key.instance.c_str(), (int)m.size());
-  }
+      if (!entry.is_valid()) {
+        CLS_LOG(20, "entry %s[%s] is not valid\n", key.name.c_str(), key.instance.c_str());
+        continue;
+      }
+
+      if (!op.list_versions && !entry.is_visible()) {
+        CLS_LOG(20, "entry %s[%s] is not visible\n", key.name.c_str(), key.instance.c_str());
+        continue;
+      }
+      if (m.size() < op.num_entries) {
+        m[kiter->first] = entry;
+      }
+      left_to_read--;
+
+      CLS_LOG(20, "got entry %s[%s] m.size()=%d\n", key.name.c_str(), key.instance.c_str(), (int)m.size());
+    }
+  } while (left_to_read > 0 && !done);
 
-  ret.is_truncated = (kiter != keys.end() && !done);
+  ret.is_truncated = (left_to_read == 0) && /* we found more entries than we were requested, meaning response is truncated */
+                     !done;
 
   ::encode(ret, *out);
   return 0;