FileStore: add support for omap xattrs

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

diff --git a/src/os/FileStore.cc b/src/os/FileStore.cc
index 5b5f72d8b673b34b1c7f1df722a50aba6aa263f0..45eccb457909c20016418f2196ded60f713fb3c3 100644 (file)
--- a/src/os/FileStore.cc
+++ b/src/os/FileStore.cc
@@ -1235,6 +1235,26 @@ int FileStore::_detect_fs()
     return -ENOTSUP;
   }
 
+  char buf[1000];
+  do_setxattr(fn, "user.test", &buf, sizeof(buf));
+  do_setxattr(fn, "user.test2", &buf, sizeof(buf));
+  do_setxattr(fn, "user.test3", &buf, sizeof(buf));
+  do_setxattr(fn, "user.test4", &buf, sizeof(buf));
+  ret = do_setxattr(fn, "user.test5", &buf, sizeof(buf));
+  if (ret == -ENOSPC) {
+    if (!g_conf->filestore_xattr_use_omap) {
+      derr << "limited size xattrs -- enable filestore_xattr_use_omap" << dendl;
+      return -ENOTSUP;
+    } else {
+      derr << "limited size xattrs -- filestore_xattr_use_omap enabled" << dendl;
+    }
+  }
+  do_removexattr(fn, "user.test");
+  do_removexattr(fn, "user.test2");
+  do_removexattr(fn, "user.test3");
+  do_removexattr(fn, "user.test4");
+  do_removexattr(fn, "user.test5");
+
   int fd = ::open(basedir.c_str(), O_RDONLY);
   if (fd < 0)
     return -errno;
@@ -2518,8 +2538,11 @@ unsigned FileStore::_do_transaction(Transaction& t, uint64_t op_seq, int trans_n
        string name = i.get_attrname();
        bufferlist bl;
        i.get_bl(bl);
-       if (_check_replay_guard(cid, oid, spos))
-         r = _setattr(cid, oid, name.c_str(), bl.c_str(), bl.length());
+       if (!_check_replay_guard(cid, oid, spos))
+         break;
+       map<string, bufferptr> to_set;
+       to_set[name] = bufferptr(bl.c_str(), bl.length());
+       r = _setattrs(cid, oid, to_set);
        if (r == -ENOSPC)
          dout(0) << " ENOSPC on setxattr on " << cid << "/" << oid
                  << " name " << name << " size " << bl.length() << dendl;
@@ -3810,23 +3833,35 @@ int FileStore::_getattrs(const char *fn, map<string,bufferptr>& aset, bool user_
 
 // objects
 
-int FileStore::getattr(coll_t cid, const hobject_t& oid, const char *name,
-                      void *value, size_t size) 
-{
-  dout(15) << "getattr " << cid << "/" << oid << " '" << name << "' len " << size << dendl;
-  char n[ATTR_MAX_NAME_LEN];
-  get_attrname(name, n, ATTR_MAX_NAME_LEN);
-  int r = lfn_getxattr(cid, oid, n, value, size);
-  dout(10) << "getattr " << cid << "/" << oid << " '" << name << "' len " << size << " = " << r << dendl;
-  return r;
-}
-
 int FileStore::getattr(coll_t cid, const hobject_t& oid, const char *name, bufferptr &bp)
 {
   dout(15) << "getattr " << cid << "/" << oid << " '" << name << "'" << dendl;
   char n[ATTR_MAX_NAME_LEN];
   get_attrname(name, n, ATTR_MAX_NAME_LEN);
   int r = _getattr(cid, oid, n, bp);
+  if (r == -ENODATA && g_conf->filestore_xattr_use_omap) {
+    map<string, bufferlist> got;
+    set<string> to_get;
+    to_get.insert(string(name));
+    Index index;
+    r = get_index(cid, &index);
+    if (r < 0) {
+      dout(10) << __func__ << " could not get index r = " << r << dendl;
+      return r;
+    }
+    r = object_map->get_xattrs(oid, index, to_get, &got);
+    if (r < 0 && r != -ENOENT) {
+      dout(10) << __func__ << " get_xattrs err r =" << r << dendl;
+      return r;
+    }
+    if (!got.size()) {
+      dout(10) << __func__ << " got.size() is 0" << dendl;
+      return -ENODATA;
+    }
+    bp = bufferptr(got.begin()->second.c_str(),
+                  got.begin()->second.length());
+    r = 0;
+  }
   dout(10) << "getattr " << cid << "/" << oid << " '" << name << "' = " << r << dendl;
   return r;
 }
@@ -3835,23 +3870,54 @@ int FileStore::getattrs(coll_t cid, const hobject_t& oid, map<string,bufferptr>&
 {
   dout(15) << "getattrs " << cid << "/" << oid << dendl;
   int r = _getattrs(cid, oid, aset, user_only);
+  if (g_conf->filestore_xattr_use_omap) {
+    set<string> omap_attrs;
+    map<string, bufferlist> omap_aset;
+    Index index;
+    int r = get_index(cid, &index);
+    if (r < 0) {
+      dout(10) << __func__ << " could not get index r = " << r << dendl;
+      return r;
+    }
+    r = object_map->get_all_xattrs(oid, index, &omap_attrs);
+    if (r < 0) {
+      dout(10) << __func__ << " could not get omap_attrs r = " << r << dendl;
+      return r;
+    }
+    r = object_map->get_xattrs(oid, index, omap_attrs, &omap_aset);
+    if (r < 0) {
+      dout(10) << __func__ << " could not get omap_attrs r = " << r << dendl;
+      return r;
+    }
+    assert(omap_attrs.size() == omap_aset.size());
+    for (map<string, bufferlist>::iterator i = omap_aset.begin();
+        i != omap_aset.end();
+        ++i) {
+      string key;
+      if (user_only) {
+       if (i->first[0] != '_')
+         continue;
+       if (i->first == "_")
+         continue;
+       key = i->first.substr(1, i->first.size());
+      } else {
+       key = i->first;
+      }
+      aset.insert(make_pair(key,
+                           bufferptr(i->second.c_str(), i->second.length())));
+    }
+  }
   dout(10) << "getattrs " << cid << "/" << oid << " = " << r << dendl;
   return r;
 }
 
-int FileStore::_setattr(coll_t cid, const hobject_t& oid, const char *name,
-                       const void *value, size_t size) 
-{
-  dout(15) << "setattr " << cid << "/" << oid << " '" << name << "' len " << size << dendl;
-  char n[ATTR_MAX_NAME_LEN];
-  get_attrname(name, n, ATTR_MAX_NAME_LEN);
-  int r = lfn_setxattr(cid, oid, n, value, size);
-  dout(10) << "setattr " << cid << "/" << oid << " '" << name << "' len " << size << " = " << r << dendl;
-  return r;
-}
-
 int FileStore::_setattrs(coll_t cid, const hobject_t& oid, map<string,bufferptr>& aset) 
 {
+  map<string, bufferlist> omap_set;
+  set<string> omap_remove;
+  map<string, bufferptr> inline_set;
+  if (g_conf->filestore_xattr_use_omap)
+    _getattrs(cid, oid, inline_set);
   dout(15) << "setattrs " << cid << "/" << oid << dendl;
   int r = 0;
   for (map<string,bufferptr>::iterator p = aset.begin();
@@ -3859,6 +3925,33 @@ int FileStore::_setattrs(coll_t cid, const hobject_t& oid, map<string,bufferptr>
        ++p) {
     char n[ATTR_MAX_NAME_LEN];
     get_attrname(p->first.c_str(), n, ATTR_MAX_NAME_LEN);
+    if (g_conf->filestore_xattr_use_omap) {
+      if (p->second.length() > g_conf->filestore_max_inline_xattr_size) {
+       if (inline_set.count(p->first)) {
+         inline_set.erase(p->first);
+         r = lfn_removexattr(cid, oid, n);
+         if (r < 0)
+           return r;
+       }
+       omap_set[p->first].push_back(p->second);
+       continue;
+      }
+
+      if (!inline_set.count(p->first) &&
+         inline_set.size() >= g_conf->filestore_max_inline_xattrs) {
+       if (inline_set.count(p->first)) {
+         inline_set.erase(p->first);
+         r = lfn_removexattr(cid, oid, n);
+         if (r < 0)
+           return r;
+       }
+       omap_set[p->first].push_back(p->second);
+       continue;
+      }
+      omap_remove.insert(p->first);
+      inline_set.insert(*p);
+    }
+
     const char *val;
     if (p->second.length())
       val = p->second.c_str();
@@ -3871,6 +3964,25 @@ int FileStore::_setattrs(coll_t cid, const hobject_t& oid, map<string,bufferptr>
       break;
     }
   }
+
+  if (g_conf->filestore_xattr_use_omap) {
+    Index index;
+    int r = get_index(cid, &index);
+    if (r < 0) {
+      dout(10) << __func__ << " could not get index r = " << r << dendl;
+      return r;
+    }
+    r = object_map->remove_xattrs(oid, index, omap_remove);
+    if (r < 0 && r != -ENOENT) {
+      dout(10) << __func__ << " could not remove_xattrs r = " << r << dendl;
+      return r;
+    }
+    r = object_map->set_xattrs(oid, index, omap_set);
+    if (r < 0) {
+      dout(10) << __func__ << " could not set_xattrs r = " << r << dendl;
+      return r;
+    }
+  }
   dout(10) << "setattrs " << cid << "/" << oid << " = " << r << dendl;
   return r;
 }
@@ -3882,6 +3994,21 @@ int FileStore::_rmattr(coll_t cid, const hobject_t& oid, const char *name)
   char n[ATTR_MAX_NAME_LEN];
   get_attrname(name, n, ATTR_MAX_NAME_LEN);
   int r = lfn_removexattr(cid, oid, n);
+  if (r == -ENODATA && g_conf->filestore_xattr_use_omap) {
+    Index index;
+    r = get_index(cid, &index);
+    if (r < 0) {
+      dout(10) << __func__ << " could not get index r = " << r << dendl;
+      return r;
+    }
+    set<string> to_remove;
+    to_remove.insert(string(name));
+    r = object_map->remove_xattrs(oid, index, to_remove);
+    if (r < 0 && r != -ENOENT) {
+      dout(10) << __func__ << " could not remove_xattrs index r = " << r << dendl;
+      return r;
+    }
+  }
   dout(10) << "rmattr " << cid << "/" << oid << " '" << name << "' = " << r << dendl;
   return r;
 }
@@ -3901,6 +4028,25 @@ int FileStore::_rmattrs(coll_t cid, const hobject_t& oid)
        break;
     }
   }
+  if (g_conf->filestore_xattr_use_omap) {
+    set<string> omap_attrs;
+    Index index;
+    r = get_index(cid, &index);
+    if (r < 0) {
+      dout(10) << __func__ << " could not get index r = " << r << dendl;
+      return r;
+    }
+    r = object_map->get_all_xattrs(oid, index, &omap_attrs);
+    if (r < 0) {
+      dout(10) << __func__ << " could not get omap_attrs r = " << r << dendl;
+      return r;
+    }
+    r = object_map->remove_xattrs(oid, index, omap_attrs);
+    if (r < 0) {
+      dout(10) << __func__ << " could not remove omap_attrs r = " << r << dendl;
+      return r;
+    }
+  }
   dout(10) << "rmattrs " << cid << "/" << oid << " = " << r << dendl;
   return r;
 }

diff --git a/src/os/FileStore.h b/src/os/FileStore.h
index e29419af112da1341255f762a7854e6fb13e0901..77b289023fc2deaad8f237ad9810f50503e76448 100644 (file)
--- a/src/os/FileStore.h
+++ b/src/os/FileStore.h
@@ -370,7 +370,6 @@ public:
   int snapshot(const string& name);
 
   // attrs
-  int getattr(coll_t cid, const hobject_t& oid, const char *name, void *value, size_t size);
   int getattr(coll_t cid, const hobject_t& oid, const char *name, bufferptr &bp);
   int getattrs(coll_t cid, const hobject_t& oid, map<string,bufferptr>& aset, bool user_only = false);
 
@@ -379,7 +378,6 @@ public:
   int _getattr(const char *fn, const char *name, bufferptr& bp);
   int _getattrs(const char *fn, map<string,bufferptr>& aset, bool user_only = false);
 
-  int _setattr(coll_t cid, const hobject_t& oid, const char *name, const void *value, size_t size);
   int _setattrs(coll_t cid, const hobject_t& oid, map<string,bufferptr>& aset);
   int _rmattr(coll_t cid, const hobject_t& oid, const char *name);
   int _rmattrs(coll_t cid, const hobject_t& oid);

diff --git a/src/os/ObjectStore.h b/src/os/ObjectStore.h
index ba1ac1ca6c0bf87217e5eb725a5856e65af41a69..95c08b40744abe25f0d38c0f3e885de4f796a184 100644 (file)
--- a/src/os/ObjectStore.h
+++ b/src/os/ObjectStore.h
@@ -625,7 +625,6 @@ public:
   virtual int read(coll_t cid, const hobject_t& oid, uint64_t offset, size_t len, bufferlist& bl) = 0;
   virtual int fiemap(coll_t cid, const hobject_t& oid, uint64_t offset, size_t len, bufferlist& bl) = 0;
 
-  virtual int getattr(coll_t cid, const hobject_t& oid, const char *name, void *value, size_t size) = 0;
   virtual int getattr(coll_t cid, const hobject_t& oid, const char *name, bufferptr& value) = 0;
   int getattr(coll_t cid, const hobject_t& oid, const char *name, bufferlist& value) {
     bufferptr bp;

diff --git a/src/test/store_test.cc b/src/test/store_test.cc
index ea7926a95609ac1a0d7052eb5c84f2f38978c083..d884e4c944ce31fb13e33a414fd004f35ee49bca 100644 (file)
--- a/src/test/store_test.cc
+++ b/src/test/store_test.cc
@@ -638,6 +638,86 @@ TEST_F(StoreTest, OMapTest) {
   store->apply_transaction(t);
 }
 
+TEST_F(StoreTest, XattrTest) {
+  coll_t cid("blah");
+  hobject_t hoid("tesomap", "", CEPH_NOSNAP, 0);
+  bufferlist big;
+  for (unsigned i = 0; i < 10000; ++i) {
+    big.append('\0');
+  }
+  bufferlist small;
+  for (unsigned i = 0; i < 10; ++i) {
+    small.append('\0');
+  }
+  int r;
+  {
+    ObjectStore::Transaction t;
+    t.create_collection(cid);
+    t.touch(cid, hoid);
+    r = store->apply_transaction(t);
+    ASSERT_EQ(r, 0);
+  }
+
+  map<string, bufferlist> attrs;
+  {
+    ObjectStore::Transaction t;
+    t.setattr(cid, hoid, "attr1", small);
+    attrs["attr1"] = small;
+    t.setattr(cid, hoid, "attr2", big);
+    attrs["attr2"] = big;
+    t.setattr(cid, hoid, "attr3", small);
+    attrs["attr3"] = small;
+    t.setattr(cid, hoid, "attr1", small);
+    attrs["attr1"] = small;
+    t.setattr(cid, hoid, "attr4", big);
+    attrs["attr4"] = big;
+    t.setattr(cid, hoid, "attr3", big);
+    attrs["attr3"] = big;
+    r = store->apply_transaction(t);
+    ASSERT_EQ(r, 0);
+  }
+
+  map<string, bufferptr> aset;
+  store->getattrs(cid, hoid, aset);
+  ASSERT_EQ(aset.size(), attrs.size());
+  for (map<string, bufferptr>::iterator i = aset.begin();
+       i != aset.end();
+       ++i) {
+    bufferlist bl;
+    bl.push_back(i->second);
+    ASSERT_TRUE(attrs[i->first] == bl);
+  }
+
+  {
+    ObjectStore::Transaction t;
+    t.rmattr(cid, hoid, "attr2");
+    attrs.erase("attr2");
+    r = store->apply_transaction(t);
+    ASSERT_EQ(r, 0);
+  }
+
+  aset.clear();
+  store->getattrs(cid, hoid, aset);
+  ASSERT_EQ(aset.size(), attrs.size());
+  for (map<string, bufferptr>::iterator i = aset.begin();
+       i != aset.end();
+       ++i) {
+    bufferlist bl;
+    bl.push_back(i->second);
+    ASSERT_TRUE(attrs[i->first] == bl);
+  }
+
+  bufferptr bp;
+  r = store->getattr(cid, hoid, "attr2", bp);
+  ASSERT_EQ(r, -ENODATA);
+
+  r = store->getattr(cid, hoid, "attr3", bp);
+  ASSERT_EQ(r, 0);
+  bufferlist bl2;
+  bl2.push_back(bp);
+  ASSERT_TRUE(bl2 == attrs["attr3"]);
+}
+
 int main(int argc, char **argv) {
   vector<const char*> args;
   argv_to_vec(argc, (const char **)argv, args);