mds: add mds_dir_max_entries config option

fixes: https://tracker.ceph.com/issues/52491

Signed-off-by: Yongseok Oh <[email protected]>

src/common/options/mds.yaml.in

@@ -1392,3 +1392,14 @@ options:
   - mds
   flags:
   - runtime
+- name: mds_dir_max_entries
+  type: uint
+  level: advanced
+  desc: maximum number of entries per directory before new creat/links fail
+  long_desc: The maximum number of entries before any new entries
+    are rejected with ENOSPC.
+  default: 0
+  services:
+  - mds
+  flags:
+  - runtime

src/mds/MDSRank.cc

@@ -3702,6 +3702,7 @@ const char** MDSRankDispatcher::get_tracked_conf_keys() const
     "mds_session_max_caps_throttle_ratio",
     "mds_cap_acquisition_throttle_retry_request_time",
     "mds_alternate_name_max",
+    "mds_dir_max_entries",
     NULL
   };
   return KEYS;

src/mds/Server.cc

@@ -262,6 +262,7 @@ Server::Server(MDSRank *m, MetricsHandler *metrics_handler) :
   cap_acquisition_throttle = g_conf().get_val<uint64_t>("mds_session_cap_acquisition_throttle");
   max_caps_throttle_ratio = g_conf().get_val<double>("mds_session_max_caps_throttle_ratio");
   caps_throttle_retry_request_timeout = g_conf().get_val<double>("mds_cap_acquisition_throttle_retry_request_timeout");
+  dir_max_entries = g_conf().get_val<uint64_t>("mds_dir_max_entries");
   supported_features = feature_bitset_t(CEPHFS_FEATURES_MDS_SUPPORTED);
 }
 
@@ -1255,6 +1256,11 @@ void Server::handle_conf_change(const std::set<std::string>& changed) {
   if (changed.count("mds_alternate_name_max")) {
     alternate_name_max  = g_conf().get_val<Option::size_t>("mds_alternate_name_max");
   }
+  if (changed.count("mds_dir_max_entries")) {
+    dir_max_entries = g_conf().get_val<uint64_t>("mds_dir_max_entries");
+    dout(20) << __func__ << " max entries per directory changed to "
+            << dir_max_entries << dendl;
+  }
 }
 
 /*
@@ -3206,6 +3212,23 @@ bool Server::check_fragment_space(MDRequestRef &mdr, CDir *in)
   return true;
 }
 
+/**
+ * check whether entries in a dir reached maximum size
+ *
+ */
+bool Server::check_dir_max_entries(MDRequestRef &mdr, CDir *in)
+{
+  const uint64_t size = in->inode->get_projected_inode()->dirstat.nfiles +
+                   in->inode->get_projected_inode()->dirstat.nsubdirs;
+  if (dir_max_entries && size >= dir_max_entries) {
+    dout(10) << "entries per dir " << *in << " size exceeds " << dir_max_entries << " (ENOSPC)" << dendl;
+    respond_to_request(mdr, -ENOSPC);
+    return false;
+  }
+  return true;
+}
+
+
 CDentry* Server::prepare_stray_dentry(MDRequestRef& mdr, CInode *in)
 {
   string straydname;
@@ -4423,6 +4446,8 @@ void Server::handle_client_openc(MDRequestRef& mdr)
     return;
   if (!check_fragment_space(mdr, dir))
     return;
+  if (!check_dir_max_entries(mdr, dir))
+    return;
 
   if (mdr->dn[0].size() == 1)
     mds->locker->create_lock_cache(mdr, diri, &mdr->dir_layout);
@@ -6328,7 +6353,9 @@ void Server::handle_client_mknod(MDRequestRef& mdr)
   CInode *diri = dir->get_inode();
   if (!check_access(mdr, diri, MAY_WRITE))
     return;
-  if (!check_fragment_space(mdr, dn->get_dir()))
+  if (!check_fragment_space(mdr, dir))
+    return;
+  if (!check_dir_max_entries(mdr, dir))
     return;
 
   ceph_assert(dn->get_projected_linkage()->is_null());
@@ -6429,6 +6456,8 @@ void Server::handle_client_mkdir(MDRequestRef& mdr)
 
   if (!check_fragment_space(mdr, dir))
     return;
+  if (!check_dir_max_entries(mdr, dir))
+    return;
 
   ceph_assert(dn->get_projected_linkage()->is_null());
   if (req->get_alternate_name().size() > alternate_name_max) {
@@ -6520,6 +6549,8 @@ void Server::handle_client_symlink(MDRequestRef& mdr)
     return;
   if (!check_fragment_space(mdr, dir))
     return;
+  if (!check_dir_max_entries(mdr, dir))
+    return;
 
   ceph_assert(dn->get_projected_linkage()->is_null());
   if (req->get_alternate_name().size() > alternate_name_max) {
@@ -6659,6 +6690,9 @@ void Server::handle_client_link(MDRequestRef& mdr)
 
     if (!check_fragment_space(mdr, dir))
       return;
+
+    if (!check_dir_max_entries(mdr, dir))
+      return;
   }
 
   CInode* target_pin = targeti->get_projected_parent_dir()->inode;
@@ -8198,6 +8232,9 @@ void Server::handle_client_rename(MDRequestRef& mdr)
     if (!check_fragment_space(mdr, destdn->get_dir()))
       return;
 
+    if (!check_dir_max_entries(mdr, destdn->get_dir()))
+      return;
+
     if (!check_access(mdr, srci, MAY_WRITE))
       return;
   }

src/mds/Server.h

@@ -176,6 +176,7 @@ class Server {
 
   // some helpers
   bool check_fragment_space(MDRequestRef& mdr, CDir *in);
+  bool check_dir_max_entries(MDRequestRef& mdr, CDir *in);
   bool check_access(MDRequestRef& mdr, CInode *in, unsigned mask);
   bool _check_access(Session *session, CInode *in, unsigned mask, int caller_uid, int caller_gid, int setattr_uid, int setattr_gid);
   CDentry *prepare_stray_dentry(MDRequestRef& mdr, CInode *in);
@@ -459,6 +460,7 @@ class Server {
   double cap_revoke_eviction_timeout = 0;
   uint64_t max_snaps_per_dir = 100;
   unsigned delegate_inos_pct = 0;
+  uint64_t dir_max_entries = 0;
 
   DecayCounter recall_throttle;
   time last_recall_state;