From 8c0c629af3dca89c9154cb2e8395139de758e487 Mon Sep 17 00:00:00 2001
From: Yongseok Oh <yongseok.oh@linecorp.com>
Date: Thu, 2 Sep 2021 09:33:44 +0000
Subject: [PATCH] mds: add mds_dir_max_entries config option

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

Signed-off-by: Yongseok Oh <yongseok.oh@linecorp.com>
(cherry picked from commit 49270bc6d2fcab2daaa729ca2b14e692152c82ff)

 Conflicts:
	src/common/options/mds.yaml.in
Resolved by converting the yaml config change to src/common/options.cc
---
 src/common/options.cc |  8 +++++++-
 src/mds/MDSRank.cc    |  1 +
 src/mds/Server.cc     | 39 ++++++++++++++++++++++++++++++++++++++-
 src/mds/Server.h      |  2 ++
 4 files changed, 48 insertions(+), 2 deletions(-)

diff --git a/src/common/options.cc b/src/common/options.cc
index 6a2115e15a0ab..fc9fb07f95b92 100644
--- a/src/common/options.cc
+++ b/src/common/options.cc
@@ -8878,7 +8878,13 @@ std::vector<Option> get_mds_options() {
      .set_default(2)
      .set_flag(Option::FLAG_RUNTIME)
      .set_description("interval in seconds for metrics data update.")
-     .set_long_description("interval in seconds after which active MDSs send client metrics data to rank 0.")
+     .set_long_description("interval in seconds after which active MDSs send client metrics data to rank 0."),
+
+    Option("mds_dir_max_entries", Option::TYPE_UINT, Option::LEVEL_ADVANCED)
+     .set_default(0)
+     .set_flag(Option::FLAG_RUNTIME)
+     .set_description("maximum number of entries per directory before new creat/links fail")
+     .set_long_description("The maximum number of entries before any new entries are rejected with ENOSPC.")
   });
 }
 
diff --git a/src/mds/MDSRank.cc b/src/mds/MDSRank.cc
index 3d5ed6e2c59b1..71a8167a5cef7 100644
--- a/src/mds/MDSRank.cc
+++ b/src/mds/MDSRank.cc
@@ -3703,6 +3703,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;
diff --git a/src/mds/Server.cc b/src/mds/Server.cc
index 618f1160de569..72bb86c1707d8 100644
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -260,6 +260,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;
@@ -4420,6 +4443,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);
@@ -6322,7 +6347,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());
@@ -6423,6 +6450,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) {
@@ -6514,6 +6543,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) {
@@ -6653,6 +6684,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;
@@ -8192,6 +8226,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;
   }
diff --git a/src/mds/Server.h b/src/mds/Server.h
index 7297fd56b6231..e378959c180e3 100644
--- a/src/mds/Server.h
+++ b/src/mds/Server.h
@@ -176,6 +176,7 @@ public:
 
   // 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 @@ private:
   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;
-- 
2.39.5