quincy: mds: prevent clients from exceeding the xattrs key/value limits

PendingReleaseNotes

@@ -707,3 +707,9 @@ Relevant tracker: https://tracker.ceph.com/issues/55521
   (``auth_client_required=cephx`` and ``ms_mon_client_mode=secure``).
   If you have cephx authentication disabled on your cluster, you may
   need to adjust these settings for RGW.
+
+* New MDSMap field `max_xattr_size` which can be set using the `fs set` command.
+  This MDSMap field allows to configure the maximum size allowed for the full
+  key/value set for a filesystem extended attributes.  It effectively replaces
+  the old per-MDS `max_xattr_pairs_size` setting, which is now dropped.
+  Relevant tracker: https://tracker.ceph.com/issues/55725

qa/tasks/mgr/dashboard/test_health.py

@@ -28,6 +28,7 @@ class HealthTest(DashboardTestCase):
         'in': JList(int),
         'last_failure': int,
         'max_file_size': int,
+        'max_xattr_size': int,
         'explicitly_allowed_features': int,
         'damaged': JList(int),
         'tableserver': int,

src/common/options/mds.yaml.in

@@ -55,15 +55,6 @@ options:
   - mds
   flags:
   - runtime
-# max xattr kv pairs size for each dir/file
-- name: mds_max_xattr_pairs_size
-  type: size
-  level: advanced
-  desc: maximum aggregate size of extended attributes on a file
-  default: 64_K
-  services:
-  - mds
-  with_legacy: true
 - name: mds_cache_trim_interval
   type: secs
   level: advanced

src/mds/Locker.cc

@@ -3567,6 +3567,36 @@ void Locker::kick_cap_releases(MDRequestRef& mdr)
   }
 }
 
+__u32 Locker::get_xattr_total_length(CInode::mempool_xattr_map &xattr)
+{
+  __u32 total = 0;
+
+  for (const auto &p : xattr)
+    total += (p.first.length() + p.second.length());
+  return total;
+}
+
+void Locker::decode_new_xattrs(CInode::mempool_inode *inode,
+			       CInode::mempool_xattr_map *px,
+			       const cref_t<MClientCaps> &m)
+{
+  CInode::mempool_xattr_map tmp;
+
+  auto p = m->xattrbl.cbegin();
+  decode_noshare(tmp, p);
+  __u32 total = get_xattr_total_length(tmp);
+  inode->xattr_version = m->head.xattr_version;
+  if (total > mds->mdsmap->get_max_xattr_size()) {
+    dout(1) << "Maximum xattr size exceeded: " << total
+	    << " max size: " << mds->mdsmap->get_max_xattr_size() << dendl;
+    // Ignore new xattr (!!!) but increase xattr version
+    // XXX how to force the client to drop cached xattrs?
+    inode->xattr_version++;
+  } else {
+    *px = std::move(tmp);
+  }
+}
+
 /**
  * m and ack might be NULL, so don't dereference them unless dirty != 0
  */
@@ -3637,10 +3667,8 @@ void Locker::_do_snap_update(CInode *in, snapid_t snap, int dirty, snapid_t foll
   // xattr
   if (xattrs) {
     dout(7) << " xattrs v" << i->xattr_version << " -> " << m->head.xattr_version
-	    << " len " << m->xattrbl.length() << dendl;
-    i->xattr_version = m->head.xattr_version;
-    auto p = m->xattrbl.cbegin();
-    decode(*px, p);
+            << " len " << m->xattrbl.length() << dendl;
+    decode_new_xattrs(i, px, m);
   }
 
   {
@@ -3928,9 +3956,7 @@ bool Locker::_do_cap_update(CInode *in, Capability *cap,
   // xattrs update?
   if (xattr) {
     dout(7) << " xattrs v" << pi.inode->xattr_version << " -> " << m->head.xattr_version << dendl;
-    pi.inode->xattr_version = m->head.xattr_version;
-    auto p = m->xattrbl.cbegin();
-    decode_noshare(*pi.xattrs, p);
+    decode_new_xattrs(pi.inode.get(), pi.xattrs.get(), m);
     wrlock_force(&in->xattrlock, mut);
   }
 

src/mds/Locker.h

@@ -262,6 +262,10 @@ class Locker {
 
   bool any_late_revoking_caps(xlist<Capability*> const &revoking, double timeout) const;
   uint64_t calc_new_max_size(const CInode::inode_const_ptr& pi, uint64_t size);
+  __u32 get_xattr_total_length(CInode::mempool_xattr_map &xattr);
+  void decode_new_xattrs(CInode::mempool_inode *inode,
+			 CInode::mempool_xattr_map *px,
+			 const cref_t<MClientCaps> &m);
 
   MDSRank *mds;
   MDCache *mdcache;

src/mds/MDSMap.cc

@@ -177,6 +177,7 @@ void MDSMap::dump(Formatter *f) const
   cephfs_dump_features(f, required_client_features);
   f->close_section();
   f->dump_int("max_file_size", max_file_size);
+  f->dump_int("max_xattr_size", max_xattr_size);
   f->dump_int("last_failure", last_failure);
   f->dump_int("last_failure_osd_epoch", last_failure_osd_epoch);
   f->open_object_section("compat");
@@ -266,6 +267,7 @@ void MDSMap::print(ostream& out) const
   out << "session_timeout\t" << session_timeout << "\n"
       << "session_autoclose\t" << session_autoclose << "\n";
   out << "max_file_size\t" << max_file_size << "\n";
+  out << "max_xattr_size\t" << max_xattr_size << "\n";
   out << "required_client_features\t" << cephfs_stringify_features(required_client_features) << "\n";
   out << "last_failure\t" << last_failure << "\n"
       << "last_failure_osd_epoch\t" << last_failure_osd_epoch << "\n";
@@ -785,6 +787,7 @@ void MDSMap::encode(bufferlist& bl, uint64_t features) const
     encode(min_compat_client, bl);
   }
   encode(required_client_features, bl);
+  encode(max_xattr_size, bl);
   ENCODE_FINISH(bl);
 }
 
@@ -932,6 +935,10 @@ void MDSMap::decode(bufferlist::const_iterator& p)
     }
   }
 
+  if (ev >= 17) {
+    decode(max_xattr_size, p);
+  }
+
   /* All MDS since at least v14.0.0 understand INLINE */
   /* TODO: remove after R is released */
   compat.incompat.insert(MDS_FEATURE_INCOMPAT_INLINE);

src/mds/MDSMap.h

@@ -50,6 +50,12 @@ static inline const auto MDS_FEATURE_INCOMPAT_SNAPREALM_V2 = CompatSet::Feature(
 
 #define MDS_FS_NAME_DEFAULT "cephfs"
 
+/*
+ * Maximum size of xattrs the MDS can handle per inode by default.  This
+ * includes the attribute name and 4+4 bytes for the key/value sizes.
+ */
+#define MDS_MAX_XATTR_SIZE (1<<16) /* 64K */
+
 class health_check_map_t;
 
 class MDSMap {
@@ -196,6 +202,9 @@ class MDSMap {
   uint64_t get_max_filesize() const { return max_file_size; }
   void set_max_filesize(uint64_t m) { max_file_size = m; }
 
+  uint64_t get_max_xattr_size() const { return max_xattr_size; }
+  void set_max_xattr_size(uint64_t m) { max_xattr_size = m; }
+
   void set_min_compat_client(ceph_release_t version);
 
   void add_required_client_feature(size_t bit) {
@@ -605,6 +614,8 @@ class MDSMap {
   __u32 session_autoclose = 300;
   uint64_t max_file_size = 1ULL<<40; /* 1TB */
 
+  uint64_t max_xattr_size = MDS_MAX_XATTR_SIZE;
+
   feature_bitset_t required_client_features;
 
   std::vector<int64_t> data_pools;  // file data pools available to clients (via an ioctl).  first is the default.

src/mds/Server.cc

@@ -4835,7 +4835,7 @@ void Server::handle_client_readdir(MDRequestRef& mdr)
   unsigned max_bytes = req->head.args.readdir.max_bytes;
   if (!max_bytes)
     // make sure at least one item can be encoded
-    max_bytes = (512 << 10) + g_conf()->mds_max_xattr_pairs_size;
+    max_bytes = (512 << 10) + mds->mdsmap->get_max_xattr_size();
 
   // start final blob
   bufferlist dirbl;
@@ -6456,22 +6456,22 @@ void Server::handle_client_setxattr(MDRequestRef& mdr)
 
   auto handler = Server::get_xattr_or_default_handler(name);
   const auto& pxattrs = cur->get_projected_xattrs();
+  size_t cur_xattrs_size = 0;
   if (pxattrs) {
     // check xattrs kv pairs size
-    size_t cur_xattrs_size = 0;
     for (const auto& p : *pxattrs) {
       if ((flags & CEPH_XATTR_REPLACE) && name.compare(p.first) == 0) {
 	continue;
       }
       cur_xattrs_size += p.first.length() + p.second.length();
     }
 
-    if (((cur_xattrs_size + inc) > g_conf()->mds_max_xattr_pairs_size)) {
-      dout(10) << "xattr kv pairs size too big. cur_xattrs_size "
-	<< cur_xattrs_size << ", inc " << inc << dendl;
-      respond_to_request(mdr, -CEPHFS_ENOSPC);
-      return;
-    }
+  }
+  if (((cur_xattrs_size + inc) > mds->mdsmap->get_max_xattr_size())) {
+    dout(10) << "xattr kv pairs size too big. cur_xattrs_size "
+	     << cur_xattrs_size << ", inc " << inc << dendl;
+    respond_to_request(mdr, -CEPHFS_ENOSPC);
+    return;
   }
 
   XattrOp xattr_op(CEPH_MDS_OP_SETXATTR, name, req->get_data(), flags);
@@ -10845,7 +10845,7 @@ void Server::handle_client_lssnap(MDRequestRef& mdr)
   int max_bytes = req->head.args.readdir.max_bytes;
   if (!max_bytes)
     // make sure at least one item can be encoded
-    max_bytes = (512 << 10) + g_conf()->mds_max_xattr_pairs_size;
+    max_bytes = (512 << 10) + mds->mdsmap->get_max_xattr_size();
 
   __u64 last_snapid = 0;
   string offset_str = req->get_path2();

src/mon/FSCommands.cc

@@ -437,6 +437,17 @@ class SetHandler : public FileSystemCommandHandler
       {
         fs->mds_map.set_max_filesize(n);
       });
+    } else if (var == "max_xattr_size") {
+      if (interr.length()) {
+	ss << var << " requires an integer value";
+	return -EINVAL;
+      }
+      fsmap.modify_filesystem(
+          fs->fscid,
+          [n](std::shared_ptr<Filesystem> fs)
+      {
+        fs->mds_map.set_max_xattr_size(n);
+      });
     } else if (var == "allow_new_snaps") {
       bool enable_snaps = false;
       int r = parse_bool(val, &enable_snaps, ss);

src/mon/MonCommands.h

@@ -377,7 +377,7 @@ COMMAND("fs set "
 	"name=var,type=CephChoices,strings=max_mds|max_file_size"
         "|allow_new_snaps|inline_data|cluster_down|allow_dirfrags|balancer"
         "|standby_count_wanted|session_timeout|session_autoclose"
-        "|allow_standby_replay|down|joinable|min_compat_client "
+        "|allow_standby_replay|down|joinable|min_compat_client|max_xattr_size "
 	"name=val,type=CephString "
 	"name=yes_i_really_mean_it,type=CephBool,req=false "
 	"name=yes_i_really_really_mean_it,type=CephBool,req=false",