mds: handle ceph.*.layout.* setxattr

Allow individual fields of file or dir layouts to be set via setxattr.

Signed-off-by: Sage Weil <sage@inktank.com>
(cherry picked from commit ebebf72f0993d028e795c78a986e1aee542ca5e0)

3 files changed, 159 insertions, 7 deletions

diff --git a/src/mds/Server.cc b/src/mds/Server.cc
index ba436566dec..0f74349f90e 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,134 @@ 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);
+}
+
+
 class C_MDS_inode_xattr_update_finish : public Context {
   MDS *mds;
   MDRequest *mdr;
@@ -3402,26 +3534,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);
diff --git a/src/mds/Server.h b/src/mds/Server.h
index 0611e3bbf26..d4732bd5da5 100644
--- a/src/mds/Server.h
+++ b/src/mds/Server.h
@@ -152,6 +152,13 @@ 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_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;