librados,Objecter,PG: list objects now includes the locator key

Previously, there was no way to recover the locator key used to create
and object.  Now, rados_objects_list_next and ObjectIterator will return
the key as well as the object name.

Signed-off-by: Samuel Just <samuel.just@dreamhost.com>

diff --git a/src/include/rados/librados.h b/src/include/rados/librados.h
index 0a37e741388dd425549278f0937fc0f5c50bb0d8..843cad36d087d6aadd4411dea4b3305f938c021b 100644 (file)
--- a/src/include/rados/librados.h
+++ b/src/include/rados/librados.h
@@ -147,7 +147,7 @@ int rados_ioctx_get_id(rados_ioctx_t io);
 
 /* objects */
 int rados_objects_list_open(rados_ioctx_t io, rados_list_ctx_t *ctx);
-int rados_objects_list_next(rados_list_ctx_t ctx, const char **entry);
+int rados_objects_list_next(rados_list_ctx_t ctx, const char **entry, const char **key);
 void rados_objects_list_close(rados_list_ctx_t ctx);
 
 /* snapshots */

diff --git a/src/include/rados/librados.hpp b/src/include/rados/librados.hpp
index 2bb62e55c13a4f185d347e60ef423e31051ec7c6..be1298e4021c66d75ab6424148c83dbc67166aa6 100644 (file)
--- a/src/include/rados/librados.hpp
+++ b/src/include/rados/librados.hpp
@@ -7,6 +7,7 @@
 #include <map>
 #include <tr1/memory>
 #include <vector>
+#include <utility>
 #include "buffer.h"
 
 #include "librados.h"
@@ -64,14 +65,15 @@ namespace librados
     ~ObjectIterator();
     bool operator==(const ObjectIterator& rhs) const;
     bool operator!=(const ObjectIterator& rhs) const;
-    const std::string& operator*() const;
+    const std::pair<std::string, std::string>& operator*() const;
+    const std::pair<std::string, std::string>* operator->() const;
     ObjectIterator &operator++(); // Preincrement
     ObjectIterator operator++(int); // Postincrement
     friend class IoCtx;
   private:
     void get_next();
     std::tr1::shared_ptr < ObjListCtx > ctx;
-    std::string cur_obj;
+    std::pair<std::string, std::string> cur_obj;
   };
 
   class WatchCtx {

diff --git a/src/librados.cc b/src/librados.cc
index aa746c267e0910d04b216cd8ec5e865d750ec693..bd118d8d70b9cef3826cf8bd8c25f6079b1a7e44 100644 (file)
--- a/src/librados.cc
+++ b/src/librados.cc
@@ -2585,10 +2585,14 @@ bool librados::ObjectIterator::operator!=(const librados::ObjectIterator& rhs) c
   return (ctx.get() != rhs.ctx.get());
 }
 
-const std::string& librados::ObjectIterator::operator*() const {
+const pair<std::string, std::string>& librados::ObjectIterator::operator*() const {
   return cur_obj;
 }
 
+const pair<std::string, std::string>* librados::ObjectIterator::operator->() const {
+  return &cur_obj;
+}
+
 librados::ObjectIterator& librados::ObjectIterator::operator++()
 {
   get_next();
@@ -2604,8 +2608,8 @@ librados::ObjectIterator librados::ObjectIterator::operator++(int)
 
 void librados::ObjectIterator::get_next()
 {
-  const char *entry;
-  int ret = rados_objects_list_next(ctx.get(), &entry);
+  const char *entry, *key;
+  int ret = rados_objects_list_next(ctx.get(), &entry, &key);
   if (ret == -ENOENT) {
     ctx.reset();
     *this = __EndObjectIterator;
@@ -2617,7 +2621,7 @@ void librados::ObjectIterator::get_next()
     throw std::runtime_error(oss.str());
   }
 
-  cur_obj = entry;
+  cur_obj = make_pair(entry, key ? key : string());
 }
 
 const librados::ObjectIterator librados::ObjectIterator::__EndObjectIterator(NULL);
@@ -3993,7 +3997,7 @@ extern "C" void rados_objects_list_close(rados_list_ctx_t h)
   delete lh;
 }
 
-extern "C" int rados_objects_list_next(rados_list_ctx_t listctx, const char **entry)
+extern "C" int rados_objects_list_next(rados_list_ctx_t listctx, const char **entry, const char **key)
 {
   librados::ObjListCtx *lh = (librados::ObjListCtx *)listctx;
   Objecter::ListContext *h = lh->lc;
@@ -4012,7 +4016,14 @@ extern "C" int rados_objects_list_next(rados_list_ctx_t listctx, const char **en
       return -ENOENT;
   }
 
-  *entry = h->list.front().name.c_str();
+  *entry = h->list.front().first.name.c_str();
+
+  if (key) {
+    if (h->list.front().second.size())
+      *key = h->list.front().second.c_str();
+    else
+      *key = NULL;
+  }
   return 0;
 }
 

diff --git a/src/osd/PGLS.h b/src/osd/PGLS.h
index 8953613a47bf5a68ba263b9c2747701a96e38b3c..f405a6751af1c97b05b850fc9832ad5f176116d4 100644 (file)
--- a/src/osd/PGLS.h
+++ b/src/osd/PGLS.h
@@ -7,17 +7,18 @@
 
 struct PGLSResponse {
   collection_list_handle_t handle; 
-  list<object_t> entries;
+  list<pair<object_t, string> > entries;
 
   void encode(bufferlist& bl) const {
-               __u8 v = 0;
-               ::encode(v, bl);
+    __u8 v = 1;
+    ::encode(v, bl);
     ::encode(handle, bl);
     ::encode(entries, bl);
   }
   void decode(bufferlist::iterator& bl) {
-               __u8 v;
-               ::decode(v, bl);
+    __u8 v;
+    ::decode(v, bl);
+    assert(v == 1);
     ::decode(handle, bl);
     ::decode(entries, bl);
   }

diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index 7f475eeb494bd3b3d96e5f55cd590106dd81d4ef..d5cacd7d2ca6beddf7ebda84b4303cae526b32e6 100644 (file)
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -50,6 +50,7 @@ static ostream& _prefix(std::ostream *_dout, PG *pg, int whoami, OSDMapRef osdma
 
 
 #include <sstream>
+#include <utility>
 
 #include <errno.h>
 
@@ -291,7 +292,7 @@ void ReplicatedPG::do_pg_op(MOSDOp *op)
        }
        if (sentries.size() < p->op.pgls.count &&
            !response.handle.in_missing_set) {
-         // it's a readdir cookie
+         // it's a filestore cookie
          dout(10) << " handle high/missing " << response.handle << dendl;
          osr.flush();  // order wrt preceeding writes
          result = osd->store->collection_list_partial(coll, snapid,
@@ -307,6 +308,9 @@ void ReplicatedPG::do_pg_op(MOSDOp *op)
            // skip snapdir objects
            if (iter->snap == CEPH_SNAPDIR)
              continue;
+           bufferlist attr_bl;
+           osd->store->getattr(coll, *iter, OI_ATTR, attr_bl);
+           object_info_t oi(attr_bl);
 
            if (snapid != CEPH_NOSNAP) {
              // skip items not defined for this snapshot
@@ -317,9 +321,6 @@ void ReplicatedPG::do_pg_op(MOSDOp *op)
                if (snapid <= snapset.seq)
                  continue;
              } else {
-               bufferlist bl;
-               osd->store->getattr(coll, *iter, OI_ATTR, bl);
-               object_info_t oi(bl);
                bool exists = false;
                for (vector<snapid_t>::iterator i = oi.snaps.begin(); i != oi.snaps.end(); ++i)
                  if (*i == snapid) {
@@ -335,7 +336,7 @@ void ReplicatedPG::do_pg_op(MOSDOp *op)
              keep = pgls_filter(filter, *iter, filter_out);
 
             if (keep)
-             response.entries.push_back(iter->oid);
+             response.entries.push_back(make_pair(iter->oid, oi.oloc.key));
           }
          ::encode(response, outdata);
           if (filter)

diff --git a/src/osdc/Objecter.h b/src/osdc/Objecter.h
index 6a7cf293ac9ceae9960c3c67c3edd63caa2d3c1e..bad166c550b64fab74c7ce39ff275fbca6ce2c72 100644 (file)
--- a/src/osdc/Objecter.h
+++ b/src/osdc/Objecter.h
@@ -466,7 +466,7 @@ public:
     int64_t pool_id;
     int pool_snap_seq;
     int max_entries;
-    std::list<object_t> list;
+    std::list<pair<object_t, string> > list;
 
     bufferlist filter;
 

diff --git a/src/rados_export.cc b/src/rados_export.cc
index b138e490dde01157f400227d0a2c4d976d218978..15e22b9cf369bd8af5d8416ed3e1a2fa87283968 100644 (file)
--- a/src/rados_export.cc
+++ b/src/rados_export.cc
@@ -189,7 +189,7 @@ int do_rados_export(ThreadPool *tp, IoCtx& io_ctx,
   ExportLocalFileWQ export_object_wq(io_ctx_dist, time(NULL),
                                     tp, export_dir.get(), force);
   for (; oi != oi_end; ++oi) {
-    export_object_wq.queue(new std::string(*oi));
+    export_object_wq.queue(new std::string((*oi).first));
   }
   export_object_wq.drain();
 

diff --git a/src/rados_import.cc b/src/rados_import.cc
index c21801f0b374fd9c49128162aca08315150c8cef..da968a7cc503b92d5bcc18e48f0313d09bd66d6c 100644 (file)
--- a/src/rados_import.cc
+++ b/src/rados_import.cc
@@ -230,7 +230,7 @@ int do_rados_import(ThreadPool *tp, IoCtx &io_ctx, IoCtxDistributor* io_ctx_dist
     librados::ObjectIterator oi = io_ctx.objects_begin();
     librados::ObjectIterator oi_end = io_ctx.objects_end();
     for (; oi != oi_end; ++oi) {
-      import_val_wq.queue(new std::string(*oi));
+      import_val_wq.queue(new std::string((*oi).first));
     }
     import_val_wq.drain();
   }

diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index 146050b56af12aa2fedd73fa50488c1b3f26062f..038db3a27ae108bac0a80d5a54e8758b9a18b293 100644 (file)
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -198,7 +198,7 @@ int RGWRados::list_buckets_next(RGWObjEnt& obj, RGWAccessHandle *handle)
       return -ENOENT;
     }
 
-    obj.name = **state;
+    obj.name = (*state)->first;
     (*state)++;
   } while (obj.name[0] == '.');
 
@@ -237,11 +237,11 @@ int RGWRados::log_list_next(RGWAccessHandle handle, string *name)
       return -ENOENT;
     }
     if (state->prefix.length() &&
-       (*state->obit).find(state->prefix) != 0) {
+       state->obit->first.find(state->prefix) != 0) {
       state->obit++;
       continue;
     }
-    *name = *state->obit;
+    *name = state->obit->first;
     state->obit++;
     break;
   }
@@ -2019,7 +2019,7 @@ int RGWRados::pool_list(rgw_bucket& bucket, string start, uint32_t num, map<stri
 
   librados::ObjectIterator iter = io_ctx.objects_begin();
   while (iter != io_ctx.objects_end()) {
-    const std::string& oid = *iter;
+    const std::string& oid = iter->first;
     if (oid.compare(start) >= 0)
       break;
 
@@ -2032,7 +2032,7 @@ int RGWRados::pool_list(rgw_bucket& bucket, string start, uint32_t num, map<stri
 
   for (i = 0; i < num && iter != io_ctx.objects_end(); ++i, ++iter) {
     RGWObjEnt e;
-    const string &oid = *iter;
+    const string &oid = iter->first;
 
     // fill it in with initial values; we may correct later
     e.name = oid;

diff --git a/src/scratchtool.c b/src/scratchtool.c
index e9b52459890823987eff3dab59dfe21c430bd537..6934e0ccd6df164e95a2a8704731a27db08c70a7 100644 (file)
--- a/src/scratchtool.c
+++ b/src/scratchtool.c
@@ -284,7 +284,7 @@ static int testrados(void)
        r = rados_objects_list_open(io_ctx, &h);
        printf("rados_list_objects_open = %d, h = %p\n", r, h);
        const char *poolname;
-       while (rados_objects_list_next(h, &poolname) == 0)
+       while (rados_objects_list_next(h, &poolname, NULL) == 0)
                printf("rados_list_objects_next got object '%s'\n", poolname);
        rados_objects_list_close(h);
 

diff --git a/src/test/rados-api/list.cc b/src/test/rados-api/list.cc
index 5e2ef636b43740fbebe87ff272df2129be272309..0eb6afde32f6a3c03faad4d0a54f8633b8063703 100644 (file)
--- a/src/test/rados-api/list.cc
+++ b/src/test/rados-api/list.cc
@@ -20,9 +20,9 @@ TEST(LibRadosList, ListObjects) {
   rados_list_ctx_t ctx;
   ASSERT_EQ(0, rados_objects_list_open(ioctx, &ctx));
   const char *entry;
-  ASSERT_EQ(0, rados_objects_list_next(ctx, &entry));
+  ASSERT_EQ(0, rados_objects_list_next(ctx, &entry, NULL));
   ASSERT_EQ(std::string(entry), "foo");
-  ASSERT_EQ(-ENOENT, rados_objects_list_next(ctx, &entry));
+  ASSERT_EQ(-ENOENT, rados_objects_list_next(ctx, &entry, NULL));
   rados_objects_list_close(ctx);
   rados_ioctx_destroy(ioctx);
   ASSERT_EQ(0, destroy_one_pool(pool_name, &cluster));
@@ -41,7 +41,7 @@ TEST(LibRadosList, ListObjectsPP) {
   ASSERT_EQ((int)sizeof(buf), ioctx.write("foo", bl1, sizeof(buf), 0));
   ObjectIterator iter(ioctx.objects_begin());
   ASSERT_EQ(false, (iter == ioctx.objects_end()));
-  ASSERT_EQ(*iter, "foo");
+  ASSERT_EQ((*iter).first, "foo");
   ++iter;
   ASSERT_EQ(true, (iter == ioctx.objects_end()));
   ioctx.close();

diff --git a/src/test/system/st_rados_list_objects.cc b/src/test/system/st_rados_list_objects.cc
index 4bca4fe4197643d5f3f60bc07332e9f039591392..3dde29796336001efd3ebdbf6daeafaf1450e5f5 100644 (file)
--- a/src/test/system/st_rados_list_objects.cc
+++ b/src/test/system/st_rados_list_objects.cc
@@ -68,7 +68,7 @@ run()
   printf("%s: listing objects.\n", get_id_str());
   RETURN1_IF_NONZERO(rados_objects_list_open(io_ctx, &h));
   while (true) {
-    ret = rados_objects_list_next(h, &obj_name);
+    ret = rados_objects_list_next(h, &obj_name, NULL);
     if (ret == -ENOENT) {
       break;
     }