crimson/os/seastore/omap_manager: limit the max size of kv inserted to

the omap tree

Signed-off-by: Xuehan Xu <xuxuehan@qianxin.com>

diff --git a/src/crimson/os/seastore/omap_manager.h b/src/crimson/os/seastore/omap_manager.h
index e017f8d7be2d071d3ba00afea5d9f3b21cc3f8d1..85a3f89a96f75a77a75d788f5de6c90f25c3664e 100644 (file)
--- a/src/crimson/os/seastore/omap_manager.h
+++ b/src/crimson/os/seastore/omap_manager.h
@@ -68,7 +68,8 @@ public:
    * @param string &key, omap string key
    * @param string &value, mapped value corresponding key
    */
-  using omap_set_key_iertr = base_iertr;
+  using omap_set_key_iertr = base_iertr::extend<
+    crimson::ct_error::value_too_large>;
   using omap_set_key_ret = omap_set_key_iertr::future<>;
   virtual omap_set_key_ret omap_set_key(
     omap_root_t &omap_root,
@@ -76,7 +77,7 @@ public:
     const std::string &key,
     const ceph::bufferlist &value) = 0;
 
-  using omap_set_keys_iertr = base_iertr;
+  using omap_set_keys_iertr = omap_set_key_iertr;
   using omap_set_keys_ret = omap_set_keys_iertr::future<>;
   virtual omap_set_keys_ret omap_set_keys(
     omap_root_t &omap_root,

diff --git a/src/crimson/os/seastore/omap_manager/btree/omap_btree_node.h b/src/crimson/os/seastore/omap_manager/btree/omap_btree_node.h
index 6a455fc882b843b2f37eb8b184d20f76e484da45..cef4830258dd1b97b7fee73c10d319adbbd0eda2 100644 (file)
--- a/src/crimson/os/seastore/omap_manager/btree/omap_btree_node.h
+++ b/src/crimson/os/seastore/omap_manager/btree/omap_btree_node.h
@@ -66,7 +66,8 @@ struct OMapNode : LogicalChildNode {
     omap_context_t oc,
     const std::string &key) = 0;
 
-  using insert_iertr = base_iertr;
+  using insert_iertr = base_iertr::extend<
+    crimson::ct_error::value_too_large>;
   using insert_ret = insert_iertr::future<mutation_result_t>;
   virtual insert_ret insert(
     omap_context_t oc,
@@ -116,6 +117,10 @@ struct OMapNode : LogicalChildNode {
 
   virtual ~OMapNode() = default;
 
+  virtual bool exceeds_max_kv_limit(
+    const std::string &key,
+    const ceph::bufferlist &value) const = 0;
+
   void init_range(std::string _begin, std::string _end) {
     assert(begin.empty());
     assert(end.empty());

diff --git a/src/crimson/os/seastore/omap_manager/btree/omap_btree_node_impl.cc b/src/crimson/os/seastore/omap_manager/btree/omap_btree_node_impl.cc
index 9ed808f551841b40f0fcf02dade7aaeb9b37ccd5..9705d198cef2e3a4bc5bd9fcbec18297ce20424f 100644 (file)
--- a/src/crimson/os/seastore/omap_manager/btree/omap_btree_node_impl.cc
+++ b/src/crimson/os/seastore/omap_manager/btree/omap_btree_node_impl.cc
@@ -166,11 +166,14 @@ OMapInnerNode::insert(
   LOG_PREFIX(OMapInnerNode::insert);
   DEBUGT("{}->{}, this: {}",  oc.t, key, value, *this);
   auto child_pt = get_containing_child(key);
+  if (exceeds_max_kv_limit(key, value)) {
+    return crimson::ct_error::value_too_large::make();
+  }
   return get_child_node(oc, child_pt).si_then(
     [oc, &key, &value] (auto extent) {
     ceph_assert(!extent->is_btree_root());
     return extent->insert(oc, key, value);
-  }).si_then([this, oc, child_pt] (auto mresult) {
+  }).si_then([this, oc, child_pt] (auto mresult) -> insert_ret {
     if (mresult.status == mutation_status_t::SUCCESS) {
       return insert_iertr::make_ready_future<mutation_result_t>(mresult);
     } else if (mresult.status == mutation_status_t::WAS_SPLIT) {
@@ -208,7 +211,11 @@ OMapInnerNode::rm_key(omap_context_t oc, const std::string &key)
                                      std::nullopt, std::nullopt));
         }
         case mutation_status_t::WAS_SPLIT:
-          return handle_split(oc, child_pt, mresult);
+          return handle_split(oc, child_pt, mresult
+         ).handle_error_interruptible(
+           rm_key_iertr::pass_further{},
+           crimson::ct_error::assert_all{"unexpected error"}
+         );
         default:
           return rm_key_iertr::make_ready_future<mutation_result_t>(mresult);
       }
@@ -620,6 +627,9 @@ OMapLeafNode::insert(
 {
   LOG_PREFIX(OMapLeafNode::insert);
   DEBUGT("{} -> {}, this: {}", oc.t, key, value, *this);
+  if (exceeds_max_kv_limit(key, value)) {
+    return crimson::ct_error::value_too_large::make();
+  }
   bool overflow = extent_will_overflow(key.size(), value.length());
   if (!overflow) {
     if (!is_mutable()) {

diff --git a/src/crimson/os/seastore/omap_manager/btree/omap_btree_node_impl.h b/src/crimson/os/seastore/omap_manager/btree/omap_btree_node_impl.h
index 3cd311610292a4e3b9f22a81ed3169f6bc619297..46d8f1bc7f3a11ec6a9e34a8e25aa8b23a5e548b 100644 (file)
--- a/src/crimson/os/seastore/omap_manager/btree/omap_btree_node_impl.h
+++ b/src/crimson/os/seastore/omap_manager/btree/omap_btree_node_impl.h
@@ -152,6 +152,12 @@ struct OMapInnerNode
     const std::string &key,
     const ceph::bufferlist &value) final;
 
+  bool exceeds_max_kv_limit(
+    const std::string &key,
+    const ceph::bufferlist &value) const final {
+    return (key.length() + sizeof(laddr_le_t)) > (capacity() / 4);
+  }
+
   rm_key_ret rm_key(
     omap_context_t oc,
     const std::string &key) final;
@@ -187,7 +193,8 @@ struct OMapInnerNode
     omap_context_t oc,
     internal_const_iterator_t iter, OMapNodeRef entry);
 
-  using handle_split_iertr = base_iertr;
+  using handle_split_iertr = base_iertr::extend<
+    crimson::ct_error::value_too_large>;
   using handle_split_ret = handle_split_iertr::future<mutation_result_t>;
   handle_split_ret handle_split(
     omap_context_t oc, internal_const_iterator_t iter,
@@ -365,6 +372,12 @@ struct OMapLeafNode
     const std::string &key,
     const ceph::bufferlist &value) final;
 
+  bool exceeds_max_kv_limit(
+    const std::string &key,
+    const ceph::bufferlist &value) const final {
+    return (key.length() + value.length()) > (capacity() / 4);
+  }
+
   rm_key_ret rm_key(
     omap_context_t oc, const std::string &key) final;
 

diff --git a/src/crimson/os/seastore/seastore.cc b/src/crimson/os/seastore/seastore.cc
index 3029118a560fb4bf78fa4e45752efc388e6b0096..0fc5e3bcaf0806b6fb3072f8c6270d604383e794 100644 (file)
--- a/src/crimson/os/seastore/seastore.cc
+++ b/src/crimson/os/seastore/seastore.cc
@@ -2747,13 +2747,16 @@ SeaStore::Shard::omaptree_clone(
              seastar::stop_iteration>(
                seastar::stop_iteration::no);
          }
-       });
+       }).handle_error_interruptible(
+         base_iertr::pass_further{},
+         crimson::ct_error::assert_all{"unexpected value_too_large"}
+       );
       });
     });
   });
 }
 
-SeaStore::base_iertr::future<>
+SeaStore::Shard::omaptree_set_keys_iertr::future<>
 SeaStore::Shard::omaptree_set_keys(
   Transaction& t,
   omap_root_t&& root,

diff --git a/src/crimson/os/seastore/seastore.h b/src/crimson/os/seastore/seastore.h
index 76227af6edc8d1969a7be807a301ee5782a47688..98a3095a68d77d1490a4699f1e8fbf7b84d64788 100644 (file)
--- a/src/crimson/os/seastore/seastore.h
+++ b/src/crimson/os/seastore/seastore.h
@@ -381,7 +381,8 @@ public:
       uint64_t off,
       uint64_t len) const;
 
-    using tm_iertr = base_iertr;
+    using tm_iertr = base_iertr::extend<
+      crimson::ct_error::value_too_large>;
     using tm_ret = tm_iertr::future<>;
     tm_ret _do_transaction_step(
       internal_context_t &ctx,
@@ -502,7 +503,9 @@ public:
       omap_root_t&& root,
       const std::optional<std::string>& start) const;
 
-    base_iertr::future<> omaptree_set_keys(
+    using omaptree_set_keys_iertr = base_iertr::extend<
+      crimson::ct_error::value_too_large>;
+    omaptree_set_keys_iertr::future<> omaptree_set_keys(
       Transaction& t,
       omap_root_t&& root,
       Onode& onode,