Merge remote-tracking branch 'gh/wip-bobtail-vxattrs' into bobtail

10 files changed, 392 insertions, 30 deletions

diff --git a/qa/workunits/misc/layout_vxattrs.sh b/qa/workunits/misc/layout_vxattrs.sh
new file mode 100755
index 00000000000..d181e03212c
--- /dev/null
+++ b/qa/workunits/misc/layout_vxattrs.sh
@@ -0,0 +1,68 @@
+#!/bin/bash -x
+
+set -e
+
+# file
+rm -f file file2
+touch file file2
+
+getfattr -d -m - file | grep -q ceph.file.layout
+getfattr -d -m - file | grep -q ceph.file.layout.pool   && exit 1 || true
+
+getfattr -n ceph.file.layout file
+getfattr -n ceph.file.layout file | grep -q object_size=
+getfattr -n ceph.file.layout file | grep -q stripe_count=
+getfattr -n ceph.file.layout file | grep -q stripe_unit=
+getfattr -n ceph.file.layout file | grep -q pool=
+getfattr -n ceph.file.layout.pool file
+getfattr -n ceph.file.layout.stripe_unit file
+getfattr -n ceph.file.layout.stripe_count file
+getfattr -n ceph.file.layout.object_size file
+
+getfattr -n ceph.file.layout.bogus file   2>&1 | grep -q 'No such attribute'
+getfattr -n ceph.dir.layout file    2>&1 | grep -q 'No such attribute'
+
+setfattr -n ceph.file.layout.stripe_unit -v 1048576 file2
+setfattr -n ceph.file.layout.stripe_count -v 8 file2
+setfattr -n ceph.file.layout.object_size -v 10485760 file2
+setfattr -n ceph.file.layout.pool -v data file2
+setfattr -n ceph.file.layout.pool -v 0 file2
+getfattr -n ceph.file.layout.pool file2 | grep -q data
+getfattr -n ceph.file.layout.stripe_unit file2 | grep -q 1048576
+getfattr -n ceph.file.layout.stripe_count file2 | grep -q 8
+getfattr -n ceph.file.layout.object_size file2 | grep -q 10485760
+
+# dir
+rm -f dir/file || true
+rmdir dir || true
+mkdir -p dir
+
+getfattr -d -m - dir | grep -q ceph.dir.layout       && exit 1 || true
+getfattr -d -m - dir | grep -q ceph.file.layout      && exit 1 || true
+
+setfattr -n ceph.dir.layout.stripe_unit -v 1048576 dir
+setfattr -n ceph.dir.layout.stripe_count -v 8 dir
+setfattr -n ceph.dir.layout.object_size -v 10485760 dir
+setfattr -n ceph.dir.layout.pool -v data dir
+setfattr -n ceph.dir.layout.pool -v 0 dir
+getfattr -n ceph.dir.layout dir
+getfattr -n ceph.dir.layout dir | grep -q object_size=10485760
+getfattr -n ceph.dir.layout dir | grep -q stripe_count=8
+getfattr -n ceph.dir.layout dir | grep -q stripe_unit=1048576
+getfattr -n ceph.dir.layout dir | grep -q pool=data
+getfattr -n ceph.dir.layout.pool dir | grep -q data
+getfattr -n ceph.dir.layout.stripe_unit dir | grep -q 1048576
+getfattr -n ceph.dir.layout.stripe_count dir | grep -q 8
+getfattr -n ceph.dir.layout.object_size dir | grep -q 10485760
+
+touch dir/file
+getfattr -n ceph.file.layout.pool dir/file | grep -q data
+getfattr -n ceph.file.layout.stripe_unit dir/file | grep -q 1048576
+getfattr -n ceph.file.layout.stripe_count dir/file | grep -q 8
+getfattr -n ceph.file.layout.object_size dir/file | grep -q 10485760
+
+setfattr -x ceph.dir.layout dir
+getfattr -n ceph.dir.layout dir     2>&1 | grep -q 'No such attribute'
+
+echo OK
+
diff --git a/src/client/Client.cc b/src/client/Client.cc
index d876454f0f4..94b976c510e 100644
--- a/src/client/Client.cc
+++ b/src/client/Client.cc
@@ -6389,11 +6389,57 @@ int Client::lsetxattr(const char *path, const char *name, const void *value, siz
   return Client::_setxattr(ceph_inode, name, value, size, flags, getuid(), getgid());
 }
 
-
 int Client::_getxattr(Inode *in, const char *name, void *value, size_t size,
 		      int uid, int gid)
 {
-  int r = _getattr(in, CEPH_STAT_CAP_XATTR, uid, gid);
+  int r;
+
+  if (strncmp(name, "ceph.", 5) == 0) {
+    string n(name);
+    char buf[256];
+
+    r = -ENODATA;
+    if ((in->is_file() && n.find("ceph.file.layout") == 0) ||
+	(in->is_dir() && in->has_dir_layout() && n.find("ceph.dir.layout") == 0)) {
+      string rest = n.substr(n.find("layout"));
+      if (rest == "layout") {
+	r = snprintf(buf, sizeof(buf),
+		     "stripe_unit=%lu stripe_count=%lu object_size=%lu pool=",
+		     (long unsigned)in->layout.fl_stripe_unit,
+		     (long unsigned)in->layout.fl_stripe_count,
+		     (long unsigned)in->layout.fl_object_size);
+	if (osdmap->have_pg_pool(in->layout.fl_pg_pool))
+	  r += snprintf(buf + r, sizeof(buf) - r, "%s",
+			osdmap->get_pool_name(in->layout.fl_pg_pool));
+	else
+	  r += snprintf(buf + r, sizeof(buf) - r, "%lu",
+			(long unsigned)in->layout.fl_pg_pool);
+      } else if (rest == "layout.stripe_unit") {
+	r = snprintf(buf, sizeof(buf), "%lu", (long unsigned)in->layout.fl_stripe_unit);
+      } else if (rest == "layout.stripe_count") {
+	r = snprintf(buf, sizeof(buf), "%lu", (long unsigned)in->layout.fl_stripe_count);
+      } else if (rest == "layout.object_size") {
+	r = snprintf(buf, sizeof(buf), "%lu", (long unsigned)in->layout.fl_object_size);
+      } else if (rest == "layout.pool") {
+	if (osdmap->have_pg_pool(in->layout.fl_pg_pool))
+	  r = snprintf(buf, sizeof(buf), "%s",
+		       osdmap->get_pool_name(in->layout.fl_pg_pool));
+	else
+	  r = snprintf(buf, sizeof(buf), "%lu",
+		       (long unsigned)in->layout.fl_pg_pool);
+      }
+    }
+    if (size != 0) {
+      if (r > (int)size) {
+	r = -ERANGE;
+      } else if (r > 0) {
+	memcpy(value, buf, r);
+      }
+    }
+    goto out;
+  }
+
+  r = _getattr(in, CEPH_STAT_CAP_XATTR, uid, gid);
   if (r == 0) {
     string n(name);
     r = -ENODATA;
@@ -6407,6 +6453,7 @@ int Client::_getxattr(Inode *in, const char *name, void *value, size_t size,
       }
     }
   }
+ out:
   ldout(cct, 3) << "_getxattr(" << in->ino << ", \"" << name << "\", " << size << ") = " << r << dendl;
   return r;
 }
@@ -6425,13 +6472,19 @@ int Client::ll_getxattr(vinodeno_t vino, const char *name, void *value, size_t s
 
 int Client::_listxattr(Inode *in, char *name, size_t size, int uid, int gid)
 {
+  const char file_vxattrs[] = "ceph.file.layout";
+  const char dir_vxattrs[] = "ceph.dir.layout";
   int r = _getattr(in, CEPH_STAT_CAP_XATTR, uid, gid);
   if (r == 0) {
     for (map<string,bufferptr>::iterator p = in->xattrs.begin();
 	 p != in->xattrs.end();
 	 p++)
       r += p->first.length() + 1;
-    
+    if (in->is_file())
+      r += sizeof(file_vxattrs);
+    else if (in->is_dir() && in->has_dir_layout())
+      r += sizeof(dir_vxattrs);
+
     if (size != 0) {
       if (size >= (unsigned)r) {
 	for (map<string,bufferptr>::iterator p = in->xattrs.begin();
@@ -6442,6 +6495,13 @@ int Client::_listxattr(Inode *in, char *name, size_t size, int uid, int gid)
 	  *name = '\0';
 	  name++;
 	}
+	if (in->is_file()) {
+	  memcpy(name, file_vxattrs, sizeof(file_vxattrs));
+	  name += sizeof(file_vxattrs);
+	} else if (in->is_dir() && in->has_dir_layout()) {
+	  memcpy(name, dir_vxattrs, sizeof(dir_vxattrs));
+	  name += sizeof(dir_vxattrs);
+	}
       } else
 	r = -ERANGE;
     }
@@ -6469,6 +6529,13 @@ int Client::_setxattr(Inode *in, const char *name, const void *value, size_t siz
     return -EROFS;
   }
 
+  // same xattrs supported by kernel client
+  if (strncmp(name, "user.", 5) &&
+      strncmp(name, "security.", 9) &&
+      strncmp(name, "trusted.", 8) &&
+      strncmp(name, "ceph.", 5))
+    return -EOPNOTSUPP;
+
   MetaRequest *req = new MetaRequest(CEPH_MDS_OP_SETXATTR);
   filepath path;
   in->make_nosnap_relative_path(path);
@@ -6497,10 +6564,6 @@ int Client::ll_setxattr(vinodeno_t vino, const char *name, const void *value, si
   tout(cct) << vino.ino.val << std::endl;
   tout(cct) << name << std::endl;
 
-  // same xattrs supported by kernel client
-  if (strncmp(name, "user.", 5) && strncmp(name, "security.", 9) && strncmp(name, "trusted.", 8))
-    return -EOPNOTSUPP;
-
   Inode *in = _ll_get_inode(vino);
   return _setxattr(in, name, value, size, flags, uid, gid);
 }
@@ -6511,6 +6574,13 @@ int Client::_removexattr(Inode *in, const char *name, int uid, int gid)
     return -EROFS;
   }
 
+  // same xattrs supported by kernel client
+  if (strncmp(name, "user.", 5) &&
+      strncmp(name, "security.", 9) &&
+      strncmp(name, "trusted.", 8) &&
+      strncmp(name, "ceph.", 5))
+    return -EOPNOTSUPP;
+
   MetaRequest *req = new MetaRequest(CEPH_MDS_OP_RMXATTR);
   filepath path;
   in->make_nosnap_relative_path(path);
@@ -6534,10 +6604,6 @@ int Client::ll_removexattr(vinodeno_t vino, const char *name, int uid, int gid)
   tout(cct) << vino.ino.val << std::endl;
   tout(cct) << name << std::endl;
 
-  // only user xattrs, for now
-  if (strncmp(name, "user.", 5) && strncmp(name, "security.", 9) && strncmp(name, "trusted.", 8))
-    return -EOPNOTSUPP;
-
   Inode *in = _ll_get_inode(vino);
   return _removexattr(in, name, uid, gid);
 }
diff --git a/src/client/Inode.cc b/src/client/Inode.cc
index 0dfd1ebdce8..4b0c99d5764 100644
--- a/src/client/Inode.cc
+++ b/src/client/Inode.cc
@@ -40,6 +40,9 @@ ostream& operator<<(ostream &out, Inode &in)
   if (!in.dn_set.empty())
     out << " parents=" << in.dn_set;
 
+  if (in.is_dir() && in.has_dir_layout())
+    out << " has_dir_layout";
+
   out << ' ' << &in << ")";
   return out;
 }
diff --git a/src/client/Inode.h b/src/client/Inode.h
index d1387a85894..b33c38eb6f0 100644
--- a/src/client/Inode.h
+++ b/src/client/Inode.h
@@ -115,6 +115,13 @@ class Inode {
   bool is_dir()     const { return (mode & S_IFMT) == S_IFDIR; }
   bool is_file()    const { return (mode & S_IFMT) == S_IFREG; }
 
+  bool has_dir_layout() const {
+    for (unsigned c = 0; c < sizeof(layout); c++)
+      if (*((const char *)&layout + c))
+	return true;
+    return false;
+  }
+
   unsigned flags;
 
   // about the dir (if this is one!)
diff --git a/src/mds/CInode.cc b/src/mds/CInode.cc
index af70b681ffc..e72d419784c 100644
--- a/src/mds/CInode.cc
+++ b/src/mds/CInode.cc
@@ -2663,13 +2663,14 @@ int CInode::encode_inodestat(bufferlist& bl, Session *session,
   i = pfile ? pi:oi;
   if (is_file()) {
     e.layout = i->layout;
-  } else {
-    if (ppolicy && get_projected_dir_layout())
-      e.layout = *get_projected_dir_layout();
-    else if (default_layout)
-      e.layout = default_layout->layout;
+  } else if (is_dir()) {
+    ceph_file_layout *l = ppolicy ? get_projected_dir_layout() : ( default_layout ? &default_layout->layout : NULL );
+    if (l)
+      e.layout = *l;
     else
       memset(&e.layout, 0, sizeof(e.layout));
+  } else {
+    memset(&e.layout, 0, sizeof(e.layout));
   }
   e.size = i->size;
   e.truncate_seq = i->truncate_seq;
diff --git a/src/mds/CInode.h b/src/mds/CInode.h
index e43ecf50fa3..ceb551071a4 100644
--- a/src/mds/CInode.h
+++ b/src/mds/CInode.h
@@ -279,7 +279,8 @@ public:
   }
 
   ceph_file_layout *get_projected_dir_layout() {
-    if (!inode.is_dir()) return NULL;
+    if (!inode.is_dir())
+      return NULL;
     if (projected_nodes.empty()) {
       if (default_layout)
         return &default_layout->layout;
@@ -288,7 +289,8 @@ public:
     }
     else if (projected_nodes.back()->dir_layout)
       return &projected_nodes.back()->dir_layout->layout;
-    else return NULL;
+    else
+      return NULL;
   }
 
   version_t get_projected_version() {
diff --git a/src/mds/MDSMap.h b/src/mds/MDSMap.h
index dc8969c3cff..2cbbfa33f19 100644
--- a/src/mds/MDSMap.h
+++ b/src/mds/MDSMap.h
@@ -244,6 +244,10 @@ public:
   int64_t get_cas_pg_pool() const { return cas_pg_pool; }
   int64_t get_metadata_pg_pool() const { return metadata_pg_pool; }
 
+  bool is_data_pool(int64_t poolid) const {
+    return std::find(data_pg_pools.begin(), data_pg_pools.end(), poolid) != data_pg_pools.end();
+  }
+
   const map<uint64_t,mds_info_t>& get_mds_info() { return mds_info; }
   const mds_info_t& get_mds_info_gid(uint64_t gid) {
     assert(mds_info.count(gid));
diff --git a/src/mds/Server.cc b/src/mds/Server.cc
index ba436566dec..c17fc8d0050 100644
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -12,6 +12,8 @@
  * 
  */
 
+#include <boost/lexical_cast.hpp>
+
 #include "MDS.h"
 #include "Server.h"
 #include "Locker.h"
@@ -51,6 +53,8 @@
 #include "common/perf_counters.h"
 #include "include/compat.h"
 
+#include "osd/OSDMap.h"
+
 #include <errno.h>
 #include <fcntl.h>
 
@@ -3377,6 +3381,179 @@ void Server::handle_client_setdirlayout(MDRequest *mdr)
 
 // XATTRS
 
+int Server::parse_layout_vxattr(string name, string value, ceph_file_layout *layout)
+{
+  dout(20) << "parse_layout_vxattr name " << name << " value '" << value << "'" << dendl;
+  try {
+    if (name == "layout") {
+      // XXX implement me
+    } else if (name == "layout.object_size") {
+      layout->fl_object_size = boost::lexical_cast<unsigned>(value);
+    } else if (name == "layout.stripe_unit") {
+      layout->fl_stripe_unit = boost::lexical_cast<unsigned>(value);
+    } else if (name == "layout.stripe_count") {
+      layout->fl_stripe_count = boost::lexical_cast<unsigned>(value);
+    } else if (name == "layout.pool") {
+      try {
+	layout->fl_pg_pool = boost::lexical_cast<unsigned>(value);
+      } catch (boost::bad_lexical_cast const&) {
+	int64_t pool = mds->osdmap->lookup_pg_pool_name(value);
+	if (pool < 0) {
+	  dout(10) << " unknown pool " << value << dendl;
+	  return -EINVAL;
+	}
+	layout->fl_pg_pool = pool;
+      }
+    } else {
+      dout(10) << " unknown layout vxattr " << name << dendl;
+      return -EINVAL;
+    }
+  } catch (boost::bad_lexical_cast const&) {
+    dout(10) << "bad vxattr value, unable to parse int for " << name << dendl;
+    return -EINVAL;
+  }
+
+  if (!ceph_file_layout_is_valid(layout)) {
+    dout(10) << "bad layout" << dendl;
+    return -EINVAL;
+  }
+  if (!mds->mdsmap->is_data_pool(layout->fl_pg_pool)) {
+    dout(10) << " invalid data pool " << layout->fl_pg_pool << dendl;
+    return -EINVAL;
+  }
+  return 0;
+}
+
+void Server::handle_set_vxattr(MDRequest *mdr, CInode *cur,
+			       ceph_file_layout *dir_layout,
+			       set<SimpleLock*> rdlocks,
+			       set<SimpleLock*> wrlocks,
+			       set<SimpleLock*> xlocks)
+{
+  MClientRequest *req = mdr->client_request;
+  string name(req->get_path2());
+  bufferlist bl = req->get_data();
+  string value (bl.c_str(), bl.length());
+  dout(10) << "handle_set_vxattr " << name << " val " << value.length() << " bytes on " << *cur << dendl;
+
+  // layout?
+  if (name.find("ceph.file.layout") == 0 ||
+      name.find("ceph.dir.layout") == 0) {
+    inode_t *pi;
+    string rest;
+    if (name.find("ceph.dir.layout") == 0) {
+      if (!cur->is_dir()) {
+	reply_request(mdr, -EINVAL);
+	return;
+      }
+
+      default_file_layout *dlayout = new default_file_layout;
+      if (cur->get_projected_dir_layout())
+	dlayout->layout = *cur->get_projected_dir_layout();
+      else if (dir_layout)
+	dlayout->layout = *dir_layout;
+      else
+	dlayout->layout = mds->mdcache->default_file_layout;
+
+      rest = name.substr(name.find("layout"));
+      int r = parse_layout_vxattr(rest, value, &dlayout->layout);
+      if (r < 0) {
+	reply_request(mdr, r);
+	return;
+      }
+
+      xlocks.insert(&cur->policylock);
+      if (!mds->locker->acquire_locks(mdr, rdlocks, wrlocks, xlocks))
+	return;
+
+      pi = cur->project_inode();
+      cur->get_projected_node()->dir_layout = dlayout;
+    } else {
+      if (!cur->is_file()) {
+	reply_request(mdr, -EINVAL);
+	return;
+      }
+      ceph_file_layout layout = cur->get_projected_inode()->layout;
+      rest = name.substr(name.find("layout"));
+      int r = parse_layout_vxattr(rest, value, &layout);
+      if (r < 0) {
+	reply_request(mdr, r);
+	return;
+      }
+
+      xlocks.insert(&cur->filelock);
+      if (!mds->locker->acquire_locks(mdr, rdlocks, wrlocks, xlocks))
+	return;
+
+      pi = cur->project_inode();
+      pi->layout = layout;
+      pi->ctime = ceph_clock_now(g_ceph_context);
+    }
+
+    pi->version = cur->pre_dirty();
+
+    // log + wait
+    mdr->ls = mdlog->get_current_segment();
+    EUpdate *le = new EUpdate(mdlog, "set vxattr layout");
+    mdlog->start_entry(le);
+    le->metablob.add_client_req(req->get_reqid(), req->get_oldest_client_tid());
+    mdcache->predirty_journal_parents(mdr, &le->metablob, cur, 0, PREDIRTY_PRIMARY, false);
+    mdcache->journal_dirty_inode(mdr, &le->metablob, cur);
+
+    journal_and_reply(mdr, cur, 0, le, new C_MDS_inode_update_finish(mds, mdr, cur));
+    return;
+  }
+
+  dout(10) << " unknown vxattr " << name << dendl;
+  reply_request(mdr, -EINVAL);
+}
+
+void Server::handle_remove_vxattr(MDRequest *mdr, CInode *cur,
+				  set<SimpleLock*> rdlocks,
+				  set<SimpleLock*> wrlocks,
+				  set<SimpleLock*> xlocks)
+{
+  MClientRequest *req = mdr->client_request;
+  string name(req->get_path2());
+  if (name == "ceph.dir.layout") {
+    if (!cur->is_dir()) {
+      reply_request(mdr, -ENODATA);
+      return;
+    }
+    if (cur->is_root()) {
+      dout(10) << "can't remove layout policy on the root directory" << dendl;
+      reply_request(mdr, -EINVAL);
+      return;
+    }
+
+    if (!cur->get_projected_dir_layout()) {
+      reply_request(mdr, -ENODATA);
+      return;
+    }
+
+    xlocks.insert(&cur->policylock);
+    if (!mds->locker->acquire_locks(mdr, rdlocks, wrlocks, xlocks))
+      return;
+
+    cur->project_inode();
+    cur->get_projected_node()->dir_layout = NULL;
+    cur->get_projected_inode()->version = cur->pre_dirty();
+
+    // log + wait
+    mdr->ls = mdlog->get_current_segment();
+    EUpdate *le = new EUpdate(mdlog, "remove dir layout vxattr");
+    mdlog->start_entry(le);
+    le->metablob.add_client_req(req->get_reqid(), req->get_oldest_client_tid());
+    mdcache->predirty_journal_parents(mdr, &le->metablob, cur, 0, PREDIRTY_PRIMARY, false);
+    mdcache->journal_dirty_inode(mdr, &le->metablob, cur);
+
+    journal_and_reply(mdr, cur, 0, le, new C_MDS_inode_update_finish(mds, mdr, cur));
+    return;
+  }
+
+  reply_request(mdr, -ENODATA);
+}
+
 class C_MDS_inode_xattr_update_finish : public Context {
   MDS *mds;
   MDRequest *mdr;
@@ -3402,26 +3579,39 @@ public:
 void Server::handle_client_setxattr(MDRequest *mdr)
 {
   MClientRequest *req = mdr->client_request;
+  string name(req->get_path2());
   set<SimpleLock*> rdlocks, wrlocks, xlocks;
-  CInode *cur = rdlock_path_pin_ref(mdr, 0, rdlocks, true);
-  if (!cur) return;
+  CInode *cur;
+
+  ceph_file_layout *dir_layout = NULL;
+  if (name.find("ceph.dir.layout") == 0)
+    cur = rdlock_path_pin_ref(mdr, 0, rdlocks, true, false, &dir_layout);
+  else
+    cur = rdlock_path_pin_ref(mdr, 0, rdlocks, true);
+  if (!cur)
+    return;
 
   if (mdr->snapid != CEPH_NOSNAP) {
     reply_request(mdr, -EROFS);
     return;
   }
-    if (cur->is_base()) {
+  if (cur->is_base()) {
     reply_request(mdr, -EINVAL);   // for now
     return;
   }
 
+  int flags = req->head.args.setxattr.flags;
+
+  // magic ceph.* namespace?
+  if (name.find("ceph.") == 0) {
+    handle_set_vxattr(mdr, cur, dir_layout, rdlocks, wrlocks, xlocks);
+    return;
+  }
+
   xlocks.insert(&cur->xattrlock);
   if (!mds->locker->acquire_locks(mdr, rdlocks, wrlocks, xlocks))
     return;
 
-  string name(req->get_path2());
-  int flags = req->head.args.setxattr.flags;
-
   if ((flags & CEPH_XATTR_CREATE) && cur->xattrs.count(name)) {
     dout(10) << "setxattr '" << name << "' XATTR_CREATE and EEXIST on " << *cur << dendl;
     reply_request(mdr, -EEXIST);
@@ -3462,25 +3652,35 @@ void Server::handle_client_setxattr(MDRequest *mdr)
 void Server::handle_client_removexattr(MDRequest *mdr)
 {
   MClientRequest *req = mdr->client_request;
+  string name(req->get_path2());
   set<SimpleLock*> rdlocks, wrlocks, xlocks;
-  CInode *cur = rdlock_path_pin_ref(mdr, 0, rdlocks, true);
-  if (!cur) return;
+  ceph_file_layout *dir_layout = NULL;
+  CInode *cur;
+  if (name == "ceph.dir.layout")
+    cur = rdlock_path_pin_ref(mdr, 0, rdlocks, true, false, &dir_layout);
+  else
+    cur = rdlock_path_pin_ref(mdr, 0, rdlocks, true);
+  if (!cur)
+    return;
 
   if (mdr->snapid != CEPH_NOSNAP) {
     reply_request(mdr, -EROFS);
     return;
   }
-    if (cur->is_base()) {
+  if (cur->is_base()) {
     reply_request(mdr, -EINVAL);   // for now
     return;
   }
 
+  if (name.find("ceph.") == 0) {
+    handle_remove_vxattr(mdr, cur, rdlocks, wrlocks, xlocks);
+    return;
+  }
+
   xlocks.insert(&cur->xattrlock);
   if (!mds->locker->acquire_locks(mdr, rdlocks, wrlocks, xlocks))
     return;
 
-  string name(req->get_path2());
-
   map<string, bufferptr> *pxattrs = cur->get_projected_xattrs();
   if (pxattrs->count(name) == 0) {
     dout(10) << "removexattr '" << name << "' and ENODATA on " << *cur << dendl;
diff --git a/src/mds/Server.h b/src/mds/Server.h
index 0611e3bbf26..d5ee05253a9 100644
--- a/src/mds/Server.h
+++ b/src/mds/Server.h
@@ -152,6 +152,17 @@ public:
   void handle_client_setattr(MDRequest *mdr);
   void handle_client_setlayout(MDRequest *mdr);
   void handle_client_setdirlayout(MDRequest *mdr);
+
+  int parse_layout_vxattr(string name, string value, ceph_file_layout *layout);
+  void handle_set_vxattr(MDRequest *mdr, CInode *cur,
+			 ceph_file_layout *dir_layout,
+			 set<SimpleLock*> rdlocks,
+			 set<SimpleLock*> wrlocks,
+			 set<SimpleLock*> xlocks);
+  void handle_remove_vxattr(MDRequest *mdr, CInode *cur,
+			    set<SimpleLock*> rdlocks,
+			    set<SimpleLock*> wrlocks,
+			    set<SimpleLock*> xlocks);
   void handle_client_setxattr(MDRequest *mdr);
   void handle_client_removexattr(MDRequest *mdr);
 
diff --git a/src/osd/OSDMap.h b/src/osd/OSDMap.h
index 4d4bbd0b031..a9ddcbb172f 100644
--- a/src/osd/OSDMap.h
+++ b/src/osd/OSDMap.h
@@ -444,7 +444,7 @@ public:
   void pg_to_raw_up(pg_t pg, vector<int>& up) const;
   void pg_to_up_acting_osds(pg_t pg, vector<int>& up, vector<int>& acting) const;
 
-  int64_t lookup_pg_pool_name(const char *name) {
+  int64_t lookup_pg_pool_name(const string& name) {
     if (name_pool.count(name))
       return name_pool[name];
     return -ENOENT;