mds: add configurable snapshot limit

Added config option "mds_max_snaps_per_dir" defaulting to 100

Fixes: https://tracker.ceph.com/issues/41209
Signed-off-by: Milind Changire <mchangir@redhat.com>

diff --git a/qa/tasks/cephfs/test_snapshots.py b/qa/tasks/cephfs/test_snapshots.py
index 5f18d32cbc5647fe75c61885a87944771a308a54..b6a1a3ef5ea2c05675026b641567d59140413493 100644 (file)
--- a/qa/tasks/cephfs/test_snapshots.py
+++ b/qa/tasks/cephfs/test_snapshots.py
@@ -1,6 +1,8 @@
+import sys
 import logging
 import signal
 import time
+import errno
 from textwrap import dedent
 from tasks.cephfs.fuse_mount import FuseMount
 from tasks.cephfs.cephfs_test_case import CephFSTestCase
@@ -12,6 +14,7 @@ MDS_RESTART_GRACE = 60
 
 class TestSnapshots(CephFSTestCase):
     MDSS_REQUIRED = 3
+    LOAD_SETTINGS = ["mds_max_snaps_per_dir"]
 
     def _check_subtree(self, rank, path, status=None):
         got_subtrees = self.fs.rank_asok(["get", "subtrees"], rank=rank, status=status)
@@ -442,3 +445,89 @@ class TestSnapshots(CephFSTestCase):
 
         self.mount_a.run_shell(["rmdir", "d1/.snap/s1"])
         self.mount_a.run_shell(["rm", "-rf", "d0", "d1"])
+
+    class SnapLimitViolationException(Exception):
+        failed_snapshot_number = -1
+
+        def __init__(self, num):
+            self.failed_snapshot_number = num
+
+    def get_snap_name(self, dir_name, sno):
+            sname = "{dir_name}/.snap/s_{sno}".format(dir_name=dir_name, sno=sno)
+            return sname
+
+    def create_snap_dir(self, sname):
+        self.mount_a.run_shell(["mkdir", sname])
+
+    def delete_dir_and_snaps(self, dir_name, snaps):
+        for sno in range(1, snaps+1, 1):
+            sname = self.get_snap_name(dir_name, sno)
+            self.mount_a.run_shell(["rmdir", sname])
+        self.mount_a.run_shell(["rmdir", dir_name])
+
+    def create_dir_and_snaps(self, dir_name, snaps):
+        self.mount_a.run_shell(["mkdir", dir_name])
+        
+        for sno in range(1, snaps+1, 1):
+            sname = self.get_snap_name(dir_name, sno)
+            try:
+                self.create_snap_dir(sname)
+            except CommandFailedError as e:
+                # failing at the last mkdir beyond the limit is expected
+                if sno == snaps:
+                    log.info("failed while creating snap #{}: {}".format(sno, repr(e)))
+                    sys.exc_clear()
+                    raise TestSnapshots.SnapLimitViolationException(sno)
+
+    def test_mds_max_snaps_per_dir_default_limit(self):
+        """
+        Test the newly introudced option named mds_max_snaps_per_dir
+        Default snaps limit is 100
+        Test if the default number of snapshot directories can be created
+        """
+        self.create_dir_and_snaps("accounts", int(self.mds_max_snaps_per_dir))
+        self.delete_dir_and_snaps("accounts", int(self.mds_max_snaps_per_dir))
+
+    def test_mds_max_snaps_per_dir_with_increased_limit(self):
+        """
+        Test the newly introudced option named mds_max_snaps_per_dir
+        First create 101 directories and ensure that the 101st directory 
+        creation fails. Then increase the default by one and see if the
+        additional directory creation succeeds
+        """
+        # first test the default limit
+        new_limit = int(self.mds_max_snaps_per_dir)
+        self.fs.rank_asok(['config', 'set', 'mds_max_snaps_per_dir', repr(new_limit)])
+        try:
+            self.create_dir_and_snaps("accounts", new_limit + 1)
+        except TestSnapshots.SnapLimitViolationException as e:
+            if e.failed_snapshot_number == (new_limit + 1):
+                sys.exc_clear()
+                pass
+        # then increase the limit by one and test
+        new_limit = new_limit + 1
+        self.fs.rank_asok(['config', 'set', 'mds_max_snaps_per_dir', repr(new_limit)])
+        sname = self.get_snap_name("accounts", new_limit)
+        self.create_snap_dir(sname)
+        self.delete_dir_and_snaps("accounts", new_limit)
+
+    def test_mds_max_snaps_per_dir_with_reduced_limit(self):
+        """
+        Test the newly introudced option named mds_max_snaps_per_dir
+        First create 99 directories. Then reduce the limit to 98. Then try
+        creating another directory and ensure that additional directory
+        creation fails.
+        """
+        # first test the new limit
+        new_limit = int(self.mds_max_snaps_per_dir) - 1
+        self.create_dir_and_snaps("accounts", new_limit)
+        sname = self.get_snap_name("accounts", new_limit + 1)
+        # then reduce the limit by one and test
+        new_limit = new_limit - 1
+        self.fs.rank_asok(['config', 'set', 'mds_max_snaps_per_dir', repr(new_limit)])
+        try:
+            self.create_snap_dir(sname)
+        except CommandFailedError as e:
+            # after reducing limit we expect the new snapshot creation to fail
+            pass
+        self.delete_dir_and_snaps("accounts", new_limit + 1)

diff --git a/src/common/options.cc b/src/common/options.cc
index 77d04ed6231d5890b447c249c59c6cf5966c7d99..7395671a289657eaae897b6812d126a7cebca311 100644 (file)
--- a/src/common/options.cc
+++ b/src/common/options.cc
@@ -8285,6 +8285,13 @@ std::vector<Option> get_mds_options() {
      .set_default(2.0)
      .set_description("task status update interval to manager")
      .set_long_description("interval (in seconds) for sending mds task status to ceph manager"),
+
+    Option("mds_max_snaps_per_dir", Option::TYPE_UINT, Option::LEVEL_ADVANCED)
+     .set_default(100)
+     .set_min_max(0, 4096)
+     .set_flag(Option::FLAG_RUNTIME)
+     .set_description("max snapshots per directory")
+     .set_long_description("maximum number of snapshots that can be created per directory"),
   });
 }
 

diff --git a/src/mds/MDSRank.cc b/src/mds/MDSRank.cc
index cc5e749a8d69e815cd3a7af49c3d642f8292432c..bd046ad75d889e4dcd5d83070c24a9a6d61b40d1 100644 (file)
--- a/src/mds/MDSRank.cc
+++ b/src/mds/MDSRank.cc
@@ -3668,6 +3668,7 @@ const char** MDSRankDispatcher::get_tracked_conf_keys() const
     "mds_max_purge_files",
     "mds_max_purge_ops",
     "mds_max_purge_ops_per_pg",
+    "mds_max_snaps_per_dir",
     "mds_op_complaint_time",
     "mds_op_history_duration",
     "mds_op_history_size",

diff --git a/src/mds/Server.cc b/src/mds/Server.cc
index d5756fbf0bf7184c4ddabfa43a59c14b556339ba..2e50ccbdb7dadd4a2d32b9f826ea302b55f4dca7 100644 (file)
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -234,6 +234,7 @@ Server::Server(MDSRank *m) :
   recall_throttle(g_conf().get_val<double>("mds_recall_max_decay_rate"))
 {
   cap_revoke_eviction_timeout = g_conf().get_val<double>("mds_cap_revoke_eviction_timeout");
+  max_snaps_per_dir = g_conf().get_val<uint64_t>("mds_max_snaps_per_dir");
   supported_features = feature_bitset_t(CEPHFS_FEATURES_MDS_SUPPORTED);
 }
 
@@ -1133,6 +1134,11 @@ void Server::handle_conf_change(const std::set<std::string>& changed) {
   if (changed.count("mds_recall_max_decay_rate")) {
     recall_throttle = DecayCounter(g_conf().get_val<double>("mds_recall_max_decay_rate"));
   }
+  if (changed.count("mds_max_snaps_per_dir")) {
+    max_snaps_per_dir = g_conf().get_val<uint64_t>("mds_max_snaps_per_dir");
+    dout(20) << __func__ << " max snapshots per directory changed to "
+            << max_snaps_per_dir << dendl;
+  }
 }
 
 /*
@@ -9778,6 +9784,14 @@ void Server::handle_client_mksnap(MDRequestRef& mdr)
   if (!check_access(mdr, diri, MAY_WRITE|MAY_SNAPSHOT))
     return;
 
+  // check if we can create any more snapshots
+  // we don't allow any more if we are already at or beyond the limit
+  if (diri->snaprealm &&
+      diri->snaprealm->get_snaps().size() >= max_snaps_per_dir) {
+    respond_to_request(mdr, -EMLINK);
+    return;
+  }
+
   // make sure name is unique
   if (diri->snaprealm &&
       diri->snaprealm->exists(snapname)) {

diff --git a/src/mds/Server.h b/src/mds/Server.h
index f47627c99f7a0bf3025284540d26ddcb9cd7d10c..d6e97ef0fd126e04664a0ce434ae6507c15992b5 100644 (file)
--- a/src/mds/Server.h
+++ b/src/mds/Server.h
@@ -106,6 +106,7 @@ private:
   feature_bitset_t required_client_features;
 
   double cap_revoke_eviction_timeout = 0;
+  uint64_t max_snaps_per_dir = 100;
 
   friend class MDSContinuation;
   friend class ServerContext;