cephfs: make vxattr ceph.dir.rctime writable

It is observed that clients have files with invalid ctimes on cephfs.
Even if those files can be fixed, the parent directories up to the root have their rctime set to an invalid value.
Make rctime modifiable and let sysadmins fix those invalid rctimes recusively (bottom-up).

Signed-off-by: Theofilos Mouratidis <t.mour@cern.ch>

qa/workunits/fs/misc/rstats.sh

@@ -75,6 +75,11 @@ touch d1/d2/f3 # create new file
 wait_until_changed rctime
 check_rctime
 
+old_value=`getfattr --only-value -n ceph.dir.rctime .`
+setfattr -n "ceph.dir.rctime" -v `date -d 'now+1hour' +%s` .
+wait_until_changed rctime
+check_rctime
+
 cd ..
 rm -rf rstats_testdir
 echo OK

src/client/Client.cc

@@ -12918,7 +12918,13 @@ const Client::VXattr Client::_dir_vxattrs[] = {
   XATTR_NAME_CEPH(dir, rsubdirs, VXATTR_RSTAT),
   XATTR_NAME_CEPH(dir, rsnaps, VXATTR_RSTAT),
   XATTR_NAME_CEPH(dir, rbytes, VXATTR_RSTAT),
-  XATTR_NAME_CEPH(dir, rctime, VXATTR_RSTAT),
+  {
+    name: "ceph.dir.rctime",
+    getxattr_cb: &Client::_vxattrcb_dir_rctime,
+    readonly: false,
+    exists_cb: NULL,
+    flags: VXATTR_RSTAT,
+  },
   {
     name: "ceph.quota",
     getxattr_cb: &Client::_vxattrcb_quota,

src/mds/Server.cc

@@ -5324,6 +5324,21 @@ bool Server::xlock_policylock(MDRequestRef& mdr, CInode *in, bool want_layout, b
   return true;
 }
 
+bool Server::wrlock_nestlock(MDRequestRef& mdr, CInode *in)
+{
+  if (mdr->locking_state & MutationImpl::ALL_LOCKED)
+    return true;
+
+  MutationImpl::LockOpVec lov;
+  lov.add_wrlock(&in->nestlock);
+
+  if (!mds->locker->acquire_locks(mdr, lov))
+    return false;
+
+  mdr->locking_state |= MutationImpl::ALL_LOCKED;
+  return true;
+}
+
 CInode* Server::try_get_auth_inode(MDRequestRef& mdr, inodeno_t ino)
 {
   CInode *in = mdcache->get_inode(ino);
@@ -5861,6 +5876,34 @@ void Server::handle_set_vxattr(MDRequestRef& mdr, CInode *cur)
     auto pi = cur->project_inode(mdr);
     cur->setxattr_ephemeral_dist(val);
     pip = pi.inode.get();
+  } else if (name == "ceph.dir.rctime") {
+    if (!cur->is_dir()) {
+      respond_to_request(mdr, -EINVAL);
+      return;
+    }
+
+    uint64_t sec, nsec = 0;
+    try {
+      if (auto pos = value.find('.', 0); pos != string::npos) {
+        sec  = boost::lexical_cast<uint64_t>(value.substr(0, pos));
+        nsec = boost::lexical_cast<uint64_t>(value.substr(pos+1));
+      } else {
+        sec = boost::lexical_cast<uint64_t>(value);
+      }
+    } catch (boost::bad_lexical_cast const&) {
+      dout(10) << "bad vxattr value, unable to parse utime_t for " << name << dendl;
+      respond_to_request(mdr, -EINVAL);
+      return;
+    }
+    utime_t val(sec, nsec);
+
+    if (!wrlock_nestlock(mdr, cur))
+      return;
+
+    auto pi = cur->project_inode(mdr);
+    pi.inode->rstat.rctime = val;
+    mdr->no_early_reply = true;
+    pip = pi.inode.get();
   } else {
     dout(10) << " unknown vxattr " << name << dendl;
     respond_to_request(mdr, -CEPHFS_EINVAL);

src/mds/Server.h

@@ -206,6 +206,7 @@ class Server {
 
   bool xlock_policylock(MDRequestRef& mdr, CInode *in,
 			bool want_layout=false, bool xlock_snaplock=false);
+  bool wrlock_nestlock(MDRequestRef& mdr, CInode *in);
   CInode* try_get_auth_inode(MDRequestRef& mdr, inodeno_t ino);
   void handle_client_setattr(MDRequestRef& mdr);
   void handle_client_setlayout(MDRequestRef& mdr);
@@ -418,7 +419,8 @@ class Server {
            xattr_name == "ceph.dir.subvolume" ||
            xattr_name == "ceph.dir.pin" ||
            xattr_name == "ceph.dir.pin.random" ||
-           xattr_name == "ceph.dir.pin.distributed";
+           xattr_name == "ceph.dir.pin.distributed" ||
+           xattr_name == "ceph.dir.rctime";
   }
 
   static bool is_allowed_ceph_xattr(std::string_view xattr_name) {