Merge pull request ceph#57538 from ivancich/wip-shrinky-dink

rgw: add shard reduction ability to dynamic resharding

Reviewed-by: Matt Benjamin <[email protected]>

Author: ivancich

src/cls/rgw/cls_rgw.cc

@@ -4422,15 +4422,31 @@ static int rgw_reshard_add(cls_method_context_t hctx, bufferlist *in, bufferlist
     return -EINVAL;
   }
 
-
-  string key;
+  std::string key;
   op.entry.get_key(&key);
 
+  int ret;
   bufferlist bl;
+
+  if (op.create_only) {
+    ret = cls_cxx_map_get_val(hctx, key, &bl);
+    if (ret == 0) {
+      // entry already exists; make no changes
+      return -EEXIST;
+    } else if (ret != -ENOENT) {
+      CLS_ERR("error accessing reshard queue for %s with key %s",
+	      op.entry.bucket_name.c_str(), key.c_str());
+      return ret;
+    }
+
+    // we got a -ENOENT and can just fall through...
+  }
+
   encode(op.entry, bl);
-  int ret = cls_cxx_map_set_val(hctx, key, &bl);
+  ret = cls_cxx_map_set_val(hctx, key, &bl);
   if (ret < 0) {
-    CLS_ERR("error adding reshard job for bucket %s with key %s",op.entry.bucket_name.c_str(), key.c_str());
+    CLS_ERR("error adding reshard job for bucket %s with key %s",
+	    op.entry.bucket_name.c_str(), key.c_str());
     return ret;
   }
 

src/cls/rgw/cls_rgw_client.cc

@@ -1083,11 +1083,14 @@ void cls_rgw_mp_upload_part_info_update(librados::ObjectWriteOperation& op,
   op.exec(RGW_CLASS, RGW_MP_UPLOAD_PART_INFO_UPDATE, in);
 }
 
-void cls_rgw_reshard_add(librados::ObjectWriteOperation& op, const cls_rgw_reshard_entry& entry)
+void cls_rgw_reshard_add(librados::ObjectWriteOperation& op,
+			 const cls_rgw_reshard_entry& entry,
+			 const bool create_only)
 {
   bufferlist in;
   cls_rgw_reshard_add_op call;
   call.entry = entry;
+  call.create_only = create_only;
   encode(call, in);
   op.exec(RGW_CLASS, RGW_RESHARD_ADD, in);
 }

src/cls/rgw/cls_rgw_client.h

@@ -626,7 +626,9 @@ int cls_rgw_lc_list(librados::IoCtx& io_ctx, const std::string& oid,
 void cls_rgw_mp_upload_part_info_update(librados::ObjectWriteOperation& op, const std::string& part_key, const RGWUploadPartInfo& info);
 
 /* resharding */
-void cls_rgw_reshard_add(librados::ObjectWriteOperation& op, const cls_rgw_reshard_entry& entry);
+void cls_rgw_reshard_add(librados::ObjectWriteOperation& op,
+			 const cls_rgw_reshard_entry& entry,
+			 const bool create_only);
 void cls_rgw_reshard_remove(librados::ObjectWriteOperation& op, const cls_rgw_reshard_entry& entry);
 // these overloads which call io_ctx.operate() should not be called in the rgw.
 // rgw_rados_operate() should be called after the overloads w/o calls to io_ctx.operate()

src/cls/rgw/cls_rgw_ops.h

@@ -1480,19 +1480,27 @@ struct cls_rgw_mp_upload_part_info_update_op {
 WRITE_CLASS_ENCODER(cls_rgw_mp_upload_part_info_update_op)
 
 struct cls_rgw_reshard_add_op {
- cls_rgw_reshard_entry entry;
+  cls_rgw_reshard_entry entry;
+
+  // true -> will not overwrite existing entry
+  bool create_only {false};
 
   cls_rgw_reshard_add_op() {}
 
   void encode(ceph::buffer::list& bl) const {
-    ENCODE_START(1, 1, bl);
+    ENCODE_START(2, 1, bl);
     encode(entry, bl);
+    encode(create_only, bl);
     ENCODE_FINISH(bl);
   }
 
   void decode(ceph::buffer::list::const_iterator& bl) {
-    DECODE_START(1, bl);
+    DECODE_START(2, bl);
     decode(entry, bl);
+    create_only = false;
+    if (struct_v >= 2) {
+      decode(create_only, bl);
+    }
     DECODE_FINISH(bl);
   }
   static void generate_test_instances(std::list<cls_rgw_reshard_add_op*>& o);

src/cls/rgw/cls_rgw_types.cc

@@ -814,6 +814,19 @@ void rgw_usage_log_entry::generate_test_instances(list<rgw_usage_log_entry *> &o
   o.push_back(new rgw_usage_log_entry);
 }
 
+std::string to_string(cls_rgw_reshard_initiator i) {
+  switch (i) {
+  case cls_rgw_reshard_initiator::Unknown:
+    return "unknown";
+  case cls_rgw_reshard_initiator::Admin:
+    return "administrator";
+  case cls_rgw_reshard_initiator::Dynamic:
+    return "dynamic resharding";
+  default:
+    return "error";
+  }
+}
+
 void cls_rgw_reshard_entry::generate_key(const string& tenant, const string& bucket_name, string *key)
 {
   *key = tenant + ":" + bucket_name;
@@ -827,12 +840,13 @@ void cls_rgw_reshard_entry::get_key(string *key) const
 void cls_rgw_reshard_entry::dump(Formatter *f) const
 {
   utime_t ut(time);
-  encode_json("time",ut, f);
+  encode_json("time", ut, f);
   encode_json("tenant", tenant, f);
   encode_json("bucket_name", bucket_name, f);
   encode_json("bucket_id", bucket_id, f);
   encode_json("old_num_shards", old_num_shards, f);
   encode_json("tentative_new_num_shards", new_num_shards, f);
+  encode_json("initiator", to_string(initiator), f);
 }
 
 void cls_rgw_reshard_entry::generate_test_instances(list<cls_rgw_reshard_entry*>& ls)

src/cls/rgw/cls_rgw_types.h

@@ -1325,25 +1325,40 @@ struct cls_rgw_lc_entry {
 };
 WRITE_CLASS_ENCODER(cls_rgw_lc_entry);
 
+
+// used to track the initiator of a reshard entry on the reshard queue (log)
+enum class cls_rgw_reshard_initiator : uint8_t {
+  Unknown = 0,
+  Admin = 1,
+  Dynamic = 2,
+};
+std::string to_string(cls_rgw_reshard_initiator i);
+inline std::ostream& operator<<(std::ostream& out, cls_rgw_reshard_initiator i) {
+  return out << to_string(i);
+}
+
+
 struct cls_rgw_reshard_entry
 {
   ceph::real_time time;
   std::string tenant;
   std::string bucket_name;
   std::string bucket_id;
-  uint32_t old_num_shards{0};
-  uint32_t new_num_shards{0};
+  uint32_t old_num_shards {0};
+  uint32_t new_num_shards {0};
+  cls_rgw_reshard_initiator initiator {cls_rgw_reshard_initiator::Unknown};
 
   cls_rgw_reshard_entry() {}
 
   void encode(ceph::buffer::list& bl) const {
-    ENCODE_START(2, 1, bl);
+    ENCODE_START(3, 1, bl);
     encode(time, bl);
     encode(tenant, bl);
     encode(bucket_name, bl);
     encode(bucket_id, bl);
     encode(old_num_shards, bl);
     encode(new_num_shards, bl);
+    encode(initiator, bl);
     ENCODE_FINISH(bl);
   }
 
@@ -1359,6 +1374,11 @@ struct cls_rgw_reshard_entry
     }
     decode(old_num_shards, bl);
     decode(new_num_shards, bl);
+    if (struct_v >= 3) {
+      decode(initiator, bl);
+    } else {
+      initiator = cls_rgw_reshard_initiator::Unknown;
+    }
     DECODE_FINISH(bl);
   }
 

src/common/options/rgw.yaml.in

@@ -3236,6 +3236,36 @@ options:
   see_also:
   - rgw_max_objs_per_shard
   - rgw_max_dynamic_shards
+- name: rgw_dynamic_resharding_may_reduce
+  type: bool
+  level: advanced
+  desc: Whether dynamic resharding can reduce the number of shards
+  long_desc: If true, RGW's dynamic resharding ability is allowed to
+    reduce the number of shards if it appears there are too many.
+  default: true
+  services:
+  - rgw
+  see_also:
+  - rgw_dynamic_resharding
+- name: rgw_dynamic_resharding_reduction_wait
+  type: uint
+  level: advanced
+  desc: Number of hours to delay bucket index shard reduction.
+  long_desc: >-
+    In order to avoid resharding buckets with object
+    counts that fluctuate up and down regularly, we implemement a delay
+    between noting a shard reduction might be appropriate and when it's
+    actually done. This allows us to cancel the reshard operation if the
+    number of object significantly increases during this delay.
+    WARNING: Setting this value too low could result in significantly reduced
+    cluster performance.
+  default: 120
+  min: 0
+  services:
+  - rgw
+  see_also:
+  - rgw_dynamic_resharding
+  - rgw_dynamic_resharding_may_reduce
 - name: rgw_max_objs_per_shard
   type: uint
   level: basic
@@ -3271,6 +3301,23 @@ options:
   services:
   - rgw
   min: 10
+- name: rgw_reshard_debug_interval
+  type: int
+  level: dev
+  desc: The number of seconds that simulate one "day" in order to debug RGW dynamic resharding.
+    Do *not* modify for a production cluster.
+  long_desc: For debugging RGW dynamic resharding, the number of seconds that are equivalent to
+    one simulated "day". Values less than 1 are ignored and do not change dynamic resharding behavior.
+    For example, during debugging if one wanted every 10 minutes to be equivalent to one day,
+    then this would be set to 600, the number of seconds in 10 minutes.
+  default: -1
+  services:
+  - rgw
+  with_legacy: true
+  see_also:
+  - rgw_dynamic_resharding
+  - rgw_reshard_thread_interval
+  - rgw_dynamic_resharding_reduction_wait
 - name: rgw_cache_expiry_interval
   type: uint
   level: advanced

src/rgw/driver/rados/rgw_rados.cc

@@ -10324,66 +10324,96 @@ int RGWRados::cls_bucket_head_async(const DoutPrefixProvider *dpp, const RGWBuck
   return 0;
 }
 
+
+// uses information that the store has easy access to transition to the shard calculatoin logic
+void RGWRados::calculate_preferred_shards(const DoutPrefixProvider* dpp,
+					  const uint64_t num_objs,
+					  const uint32_t num_source_shards,
+					  bool& need_resharding,
+					  uint32_t* suggested_num_shards)
+{
+  const uint32_t max_dynamic_shards =
+    uint32_t(cct->_conf.get_val<uint64_t>("rgw_max_dynamic_shards"));
+  const uint64_t max_objs_per_shard =
+    cct->_conf.get_val<uint64_t>("rgw_max_objs_per_shard");
+  const bool is_multisite = svc.zone->need_to_log_data();
+
+  RGWBucketReshard::calculate_preferred_shards(dpp,
+					       max_dynamic_shards,
+					       max_objs_per_shard,
+					       is_multisite,
+					       num_objs,
+					       num_source_shards,
+					       need_resharding,
+					       suggested_num_shards);
+}
+
+
+// Check whether a bucket is a candidate for dynamic resharding and if
+// so, add it to the reshard queue (log).
+//
+// We implement dynamic reshard reduction (where the number of shards
+// can be reduced) in the following manner. In addition to the maximum
+// number of desired entries per shard, we now set a minimum
 int RGWRados::check_bucket_shards(const RGWBucketInfo& bucket_info,
 				  uint64_t num_objs,
-                                  const DoutPrefixProvider *dpp, optional_yield y)
+                                  const DoutPrefixProvider* dpp, optional_yield y)
 {
   if (! cct->_conf.get_val<bool>("rgw_dynamic_resharding")) {
-      return 0;
+    return 0;
   }
 
   if (! is_layout_reshardable(bucket_info.layout)) {
     return 0;
   }
 
-  bool need_resharding = false;
-  uint32_t num_source_shards = rgw::current_num_shards(bucket_info.layout);
-  const uint32_t max_dynamic_shards =
-    uint32_t(cct->_conf.get_val<uint64_t>("rgw_max_dynamic_shards"));
-
-  if (num_source_shards >= max_dynamic_shards) {
-    return 0;
-  }
+  // TODO: consider per-bucket sync policy here?
 
+  bool need_resharding = false;
   uint32_t suggested_num_shards = 0;
-  const uint64_t max_objs_per_shard =
-    cct->_conf.get_val<uint64_t>("rgw_max_objs_per_shard");
+  const uint32_t num_source_shards =
+    rgw::current_num_shards(bucket_info.layout);
 
-  // TODO: consider per-bucket sync policy here?
-  const bool is_multisite = svc.zone->need_to_log_data();
-
-  quota_handler->check_bucket_shards(dpp, max_objs_per_shard, num_source_shards,
-				     num_objs, is_multisite, need_resharding,
-				     &suggested_num_shards);
+  calculate_preferred_shards(dpp, num_objs, num_source_shards,
+			     need_resharding, &suggested_num_shards);
   if (! need_resharding) {
     return 0;
   }
 
-  const uint32_t final_num_shards =
-    RGWBucketReshard::get_preferred_shards(suggested_num_shards,
-					   max_dynamic_shards);
   // final verification, so we don't reduce number of shards
-  if (final_num_shards <= num_source_shards) {
+  const bool may_reduce =
+    uint32_t(cct->_conf.get_val<bool>("rgw_dynamic_resharding_may_reduce"));
+  if (! may_reduce && suggested_num_shards <= num_source_shards) {
     return 0;
   }
 
-  ldpp_dout(dpp, 1) << "RGWRados::" << __func__ << " bucket " << bucket_info.bucket.name <<
-    " needs resharding; current num shards " << bucket_info.layout.current_index.layout.normal.num_shards <<
-    "; new num shards " << final_num_shards << " (suggested " <<
-    suggested_num_shards << ")" << dendl;
+  ldpp_dout(dpp, 1) << "RGWRados::" << __func__ <<
+    " bucket " << bucket_info.bucket.name <<
+    " needs resharding; current num shards " << num_source_shards <<
+    "; new num shards " << suggested_num_shards << dendl;
 
-  return add_bucket_to_reshard(dpp, bucket_info, final_num_shards, y);
+  return add_bucket_to_reshard(dpp, bucket_info, suggested_num_shards, y);
 }
 
-int RGWRados::add_bucket_to_reshard(const DoutPrefixProvider *dpp, const RGWBucketInfo& bucket_info, uint32_t new_num_shards, optional_yield y)
+int RGWRados::add_bucket_to_reshard(const DoutPrefixProvider *dpp,
+				    const RGWBucketInfo& bucket_info,
+				    uint32_t new_num_shards,
+				    optional_yield y)
 {
   RGWReshard reshard(this->driver, dpp);
 
-  uint32_t num_source_shards = rgw::current_num_shards(bucket_info.layout);
-
+  const uint32_t num_source_shards = rgw::current_num_shards(bucket_info.layout);
+  const bool may_reduce =
+    uint32_t(cct->_conf.get_val<bool>("rgw_dynamic_resharding_may_reduce"));
   new_num_shards = std::min(new_num_shards, get_max_bucket_shards());
-  if (new_num_shards <= num_source_shards) {
-    ldpp_dout(dpp, 20) << "not resharding bucket name=" << bucket_info.bucket.name << ", orig_num=" << num_source_shards << ", new_num_shards=" << new_num_shards << dendl;
+
+  if ((! may_reduce && new_num_shards < num_source_shards) ||
+      new_num_shards == num_source_shards) {
+    ldpp_dout(dpp, 10) << "WARNING: " << __func__ <<
+      ": rejecting resharding request for bucket name=" <<
+      bucket_info.bucket.name << ", shard count=" << num_source_shards <<
+      ", new shard count=" << new_num_shards <<
+      ", rgw_dynamic_resharding_may_reduce=" << may_reduce << dendl;
     return 0;
   }
 
@@ -10394,6 +10424,7 @@ int RGWRados::add_bucket_to_reshard(const DoutPrefixProvider *dpp, const RGWBuck
   entry.bucket_id = bucket_info.bucket.bucket_id;
   entry.old_num_shards = num_source_shards;
   entry.new_num_shards = new_num_shards;
+  entry.initiator = cls_rgw_reshard_initiator::Dynamic;
 
   return reshard.add(dpp, entry, y);
 }

src/rgw/driver/rados/rgw_rados.h

@@ -1586,6 +1586,12 @@ class RGWRados
                   RGWQuota& quota, uint64_t obj_size,
 		  optional_yield y, bool check_size_only = false);
 
+  void calculate_preferred_shards(const DoutPrefixProvider* dpp,
+				  const uint64_t num_objs,
+				  const uint32_t current_shard_count,
+				  bool& need_resharding,
+				  uint32_t* suggested_num_shard_count = nullptr);
+
   int check_bucket_shards(const RGWBucketInfo& bucket_info, uint64_t num_objs,
                           const DoutPrefixProvider *dpp, optional_yield y);