Merge 'Respect per-shard tablet goal and 10x default per-shard tablet count' from Tomasz Grabiec

This series achieves two things:

1) changes default number of tablet replicas per shard to be 10 in order to reduce load imbalance between shards

    This will result in new tables having at least 10 tablet replicas per
    shard by default.

    We want this to reduce tablet load imbalance due to differences in
    tablet count per shard, where some shards have 1 tablet and some
    shards have 2 tablets. With higher tablet count per shard, this
    difference-by-one is less relevant.

    Fixes scylladb#21967

2) introduces a global goal for tablet replica count per shard and adds logic to tablet scheduler to respect it by controlling per-table tablet count

    The per-shard goal is enforced by controlling average per-shard tablet replica
    count in a given DC, which is controlled by per-table tablet
    count. This is effective in respecting the limit on individual shards
    as long as tablet replicas are distributed evenly between shards.
    There is no attempt to move tablets around in order to enforce limits
    on individual shards in case of imbalance between shards.

    If the average per-shard tablet count exceeds the limit, all tables
    which contribute to it (have replicas in the DC) are scaled down
    by the same factor. Due to rounding up to the nearest power of 2,
    we may overshoot the per-shard goal by at most a factor of 2.

    The scaling is applied after computing desired tablet count due to
    all other factors: per-table tablet count hints, defaults, average tablet size.

    If different DCs want different scale factors of a given table, the
    lowest scale factor is chosen for a given table.

    When creating a new table, its tablet count is determined by tablet
    scheduler using the scheduler logic, as if the table was already created.
    So any scaling due to per-shard tablet count goal is reflected immediately
    when creating a table. It may however still take some time for the system
    to shrink existing tables. We don't reject requests to create new tables.

    Fixes scylladb#21458

Closes scylladb#22522

* github.com:scylladb/scylladb:
  config, tablets: Allow tablets_initial_scale_factor to be a fraction
  test: tablets_test: Test scaling when creating lots of tables
  test: tablets_test: Test tablet count changes on per-table option and config changes
  test: tablets_test: Add support for auto-split mode
  test: cql_test_env: Expose db config
  config: Make tablets_initial_scale_factor live-updateable
  tablets: load_balancer: Pick initial_scale_factor from config
  tablets, load_balancer: Fix and improve logging of resize decisions
  tablets, load_balancer: Log reason for target tablet count
  tablets: load_balancer: Move hints processing to tablet scheduler
  tablets: load_balancer: Scale down tablet count to respect per-shard tablet count goal
  tablets: Use scheduler's make_sizing_plan() to decide about tablet count of a new table
  tablets: load_balancer: Determine desired count from size separately from count from options
  tablets: load_balancer: Determine resize decision from target tablet count
  tablets: load_balancer: Allow splits even if table stats not available
  tablets: load_balancer: Extract make_sizing_plan()
  tablets: Add formatter for resize_decision::way_type
  tablets: load_balancer: Simplify resize_urgency_cmp()
  tablets: load_balancer: Keep config items as instance members
  locator: network_topology_strategy: Simplify calculate_initial_tablets_from_topology()
  tablets: Change the meaning of initial_scale to mean min-avg-tablets-per-shard
  tablets: Set default initial tablet count scale to 10
  tablets: network_topology_stragy: Coroutinize calculate_initial_tablets_from_topology()
  tablets: load_balancer: Extract get_schema_and_rs()
  tablets: load_balancer: Drop test_mode

Author: avikivity

db/config.cc

@@ -1334,7 +1334,10 @@ db::config::config(std::shared_ptr<db::extensions> exts)
     , minimum_replication_factor_warn_threshold(this, "minimum_replication_factor_warn_threshold", liveness::LiveUpdate, value_status::Used,  3, "")
     , maximum_replication_factor_warn_threshold(this, "maximum_replication_factor_warn_threshold", liveness::LiveUpdate, value_status::Used, -1, "")
     , maximum_replication_factor_fail_threshold(this, "maximum_replication_factor_fail_threshold", liveness::LiveUpdate, value_status::Used, -1, "")
-    , tablets_initial_scale_factor(this, "tablets_initial_scale_factor", value_status::Used, 1, "Calculated initial tablets are multiplied by this number")
+    , tablets_initial_scale_factor(this, "tablets_initial_scale_factor", liveness::LiveUpdate, value_status::Used, 10,
+         "Minimum average number of tablet replicas per shard per table. Suppressed by tablet options in table's schema: min_per_shard_tablet_count and min_tablet_count")
+    , tablets_per_shard_goal(this, "tablets_per_shard_goal", liveness::LiveUpdate, value_status::Used, 100,
+         "The goal for the maximum number of tablet replicas per shard. Tablet allocator tries to keep this goal.")
     , target_tablet_size_in_bytes(this, "target_tablet_size_in_bytes", liveness::LiveUpdate, value_status::Used, service::default_target_tablet_size,
          "Allows target tablet size to be configured. Defaults to 5G (in bytes). Maintaining tablets at reasonable sizes is important to be able to " \
          "redistribute load. A higher value means tablet migration throughput can be reduced. A lower value may cause number of tablets to increase significantly, " \

db/config.hh

@@ -502,7 +502,8 @@ public:
     named_value<int> maximum_replication_factor_warn_threshold;
     named_value<int> maximum_replication_factor_fail_threshold;
 
-    named_value<int> tablets_initial_scale_factor;
+    named_value<double> tablets_initial_scale_factor;
+    named_value<unsigned> tablets_per_shard_goal;
     named_value<uint64_t> target_tablet_size_in_bytes;
 
     named_value<std::vector<enum_option<replication_strategy_restriction_t>>> replication_strategy_warn_list;

locator/network_topology_strategy.cc

@@ -296,64 +296,13 @@ effective_replication_map_ptr network_topology_strategy::make_replication_map(ta
     return do_make_replication_map(table, shared_from_this(), std::move(tm), _rep_factor);
 }
 
-//
-// Try to use as many tablets initially, so that all shards in the current topology
-// are covered with at least `min_per_shard_tablet_count` tablets. In other words, the value is
-//
-//    initial_tablets = max(nr_shards_in(dc) / RF_in(dc) for dc in datacenters)
-//
-
-static unsigned calculate_initial_tablets_from_topology(const schema& s, token_metadata_ptr tm, const std::unordered_map<sstring, size_t>& rf, double min_per_shard_tablet_count = 0) {
-    unsigned initial_tablets = std::numeric_limits<unsigned>::min();
-    std::unordered_map<sstring, unsigned> shards_per_dc_map;
-    tm->for_each_token_owner([&] (const node& node) {
-        if (node.is_normal()) {
-            shards_per_dc_map[node.dc_rack().dc] += node.get_shard_count();
-        }
-    });
-    for (const auto& [dc, rf_in_dc] : rf) {
-        if (!rf_in_dc) {
-            continue;
-        }
-        unsigned shards_in_dc = shards_per_dc_map[dc];
-        unsigned tablets_in_dc = (shards_in_dc + rf_in_dc - 1) / rf_in_dc;
-        if (min_per_shard_tablet_count) {
-            auto min_tablets_in_dc = std::ceil((double)(min_per_shard_tablet_count * shards_in_dc) / rf_in_dc);
-            tablets_in_dc = std::max<unsigned>(tablets_in_dc, min_tablets_in_dc);
-        }
-        initial_tablets = std::max(initial_tablets, tablets_in_dc);
-    }
-    rslogger.debug("Estimated {} initial tablets for table {}.{}", initial_tablets, s.ks_name(), s.cf_name());
-    return initial_tablets;
-}
-
-size_t network_topology_strategy::calculate_min_tablet_count(schema_ptr s, token_metadata_ptr tm, uint64_t target_tablet_size, std::optional<unsigned> initial_scale) const {
-    size_t tablet_count = get_initial_tablets();
-    const auto& tablet_options = s->tablet_options();
-    if (tablet_options.min_tablet_count) {
-        tablet_count = std::max<size_t>(tablet_count, tablet_options.min_tablet_count.value());
-    }
-    if (tablet_options.expected_data_size_in_gb) {
-        tablet_count = std::max<size_t>(tablet_count, (tablet_options.expected_data_size_in_gb.value() << 30) / target_tablet_size);
-    }
-    if (tablet_options.min_per_shard_tablet_count) {
-        tablet_count = std::max<size_t>(tablet_count, calculate_initial_tablets_from_topology(*s, tm, _dc_rep_factor, tablet_options.min_per_shard_tablet_count.value()));
-    }
-    if (tablet_count == 0) {
-        tablet_count = calculate_initial_tablets_from_topology(*s, tm, _dc_rep_factor) * initial_scale.value_or(1);
-    }
-    return tablet_count;
-}
-
-future<tablet_map> network_topology_strategy::allocate_tablets_for_new_table(schema_ptr s, token_metadata_ptr tm, uint64_t target_tablet_size, std::optional<unsigned> initial_scale) const {
-    size_t tablet_count = calculate_min_tablet_count(s, tm, target_tablet_size, initial_scale);
+future<tablet_map> network_topology_strategy::allocate_tablets_for_new_table(schema_ptr s, token_metadata_ptr tm, size_t tablet_count) const {
     auto aligned_tablet_count = 1ul << log2ceil(tablet_count);
     if (tablet_count != aligned_tablet_count) {
         rslogger.info("Rounding up tablet count from {} to {} for table {}.{}", tablet_count, aligned_tablet_count, s->ks_name(), s->cf_name());
         tablet_count = aligned_tablet_count;
     }
-
-    return reallocate_tablets(std::move(s), std::move(tm), tablet_map(tablet_count));
+    co_return co_await reallocate_tablets(std::move(s), std::move(tm), tablet_map(tablet_count));
 }
 
 future<tablet_map> network_topology_strategy::reallocate_tablets(schema_ptr s, token_metadata_ptr tm, tablet_map tablets) const {

locator/network_topology_strategy.hh

@@ -29,7 +29,7 @@ public:
         return _rep_factor;
     }
 
-    size_t get_replication_factor(const sstring& dc) const {
+    size_t get_replication_factor(const sstring& dc) const override {
         auto dc_factor = _dc_rep_factor.find(dc);
         return (dc_factor == _dc_rep_factor.end()) ? 0 : dc_factor->second;
     }
@@ -46,8 +46,7 @@ public:
 
 public: // tablet_aware_replication_strategy
     virtual effective_replication_map_ptr make_replication_map(table_id, token_metadata_ptr) const override;
-    virtual size_t calculate_min_tablet_count(schema_ptr s, token_metadata_ptr tm, uint64_t target_tablet_size, std::optional<unsigned> initial_scale) const override;
-    virtual future<tablet_map> allocate_tablets_for_new_table(schema_ptr, token_metadata_ptr, uint64_t target_tablet_size, std::optional<unsigned> initial_scale = std::nullopt) const override;
+    virtual future<tablet_map> allocate_tablets_for_new_table(schema_ptr, token_metadata_ptr, size_t tablet_count) const override;
     virtual future<tablet_map> reallocate_tablets(schema_ptr, token_metadata_ptr, tablet_map cur_tablets) const override;
 protected:
     /**

locator/tablet_replication_strategy.hh

@@ -29,26 +29,29 @@ private:
 protected:
     void validate_tablet_options(const abstract_replication_strategy&, const gms::feature_service&, const replication_strategy_config_options&) const;
     void process_tablet_options(abstract_replication_strategy&, replication_strategy_config_options&, replication_strategy_params);
-    size_t get_initial_tablets() const { return _initial_tablets; }
     effective_replication_map_ptr do_make_replication_map(table_id,
                                                           replication_strategy_ptr,
                                                           token_metadata_ptr,
                                                           size_t replication_factor) const;
 
 public:
-    /// Calculate the minimum tablet_count for a table, given the target_tablet_size, the per-table hints,
-    /// the network topology, and the configured replication factors.
-    virtual size_t calculate_min_tablet_count(schema_ptr s, token_metadata_ptr tm, uint64_t target_tablet_size, std::optional<unsigned> initial_scale) const = 0;
+    size_t get_initial_tablets() const { return _initial_tablets; }
 
     /// Generates tablet_map for a new table.
     /// Runs under group0 guard.
-    virtual future<tablet_map> allocate_tablets_for_new_table(schema_ptr, token_metadata_ptr, uint64_t target_tablet_size, std::optional<unsigned> initial_scale = std::nullopt) const = 0;
+    virtual future<tablet_map> allocate_tablets_for_new_table(schema_ptr, token_metadata_ptr, size_t tablet_count) const = 0;
 
     /// Generates tablet_map for a new table or when increasing replication factor.
     /// For a new table, cur_tablets is initialized with the tablet_count,
     /// otherwise, cur_tablets is a copy of the current tablet_map.
     /// Runs under group0 guard.
     virtual future<tablet_map> reallocate_tablets(schema_ptr, token_metadata_ptr, tablet_map cur_tablets) const = 0;
+
+    /// Returns replication factor in a given DC.
+    /// Note that individual tablets may lag behind desired replication factor in their
+    /// current replica list, as replication factor changes involve table rebuilding transitions
+    /// which are not instantaneous.
+    virtual size_t get_replication_factor(const sstring& dc) const = 0;
 };
 
 } // namespace locator

locator/tablets.cc

@@ -638,13 +638,7 @@ resize_decision::resize_decision(sstring decision, uint64_t seq_number)
 }
 
 sstring resize_decision::type_name() const {
-    static const std::array<sstring, 3> index_to_string = {
-        "none",
-        "split",
-        "merge",
-    };
-    static_assert(std::variant_size_v<decltype(way)> == index_to_string.size());
-    return index_to_string[way.index()];
+    return fmt::format("{}", way);
 }
 
 resize_decision::seq_number_t resize_decision::next_sequence_number() const {
@@ -1036,6 +1030,17 @@ void tablet_metadata_guard::subscribe() {
 
 }
 
+auto fmt::formatter<locator::resize_decision_way>::format(const locator::resize_decision_way& way, fmt::format_context& ctx) const
+        -> decltype(ctx.out()) {
+    static const std::array<sstring, 3> index_to_string = {
+        "none",
+        "split",
+        "merge",
+    };
+    static_assert(std::variant_size_v<locator::resize_decision_way> == index_to_string.size());
+    return fmt::format_to(ctx.out(), "{}", index_to_string[way.index()]);
+}
+
 auto fmt::formatter<locator::global_tablet_id>::format(const locator::global_tablet_id& id, fmt::format_context& ctx) const
         -> decltype(ctx.out()) {
     return fmt::format_to(ctx.out(), "{}:{}", id.table, id.tablet);

locator/tablets.hh

@@ -325,10 +325,12 @@ struct resize_decision {
     struct merge {
         auto operator<=>(const merge&) const = default;
     };
+    using way_type = std::variant<none, split, merge>;
 
     using seq_number_t = int64_t;
 
-    std::variant<none, split, merge> way;
+    way_type way;
+
     // The sequence number globally identifies a resize decision.
     // It's monotonically increasing, globally.
     // Needed to distinguish stale decision from latest one, in case coordinator
@@ -354,6 +356,8 @@ struct resize_decision {
     seq_number_t next_sequence_number() const;
 };
 
+using resize_decision_way = resize_decision::way_type;
+
 struct table_load_stats {
     uint64_t size_in_bytes = 0;
     // Stores the minimum seq number among all replicas, as coordinator wants to know if
@@ -673,6 +677,11 @@ struct tablet_metadata_change_hint {
 
 }
 
+template <>
+struct fmt::formatter<locator::resize_decision_way> : fmt::formatter<string_view> {
+    auto format(const locator::resize_decision_way&, fmt::format_context& ctx) const -> decltype(ctx.out());
+};
+
 template <>
 struct fmt::formatter<locator::tablet_transition_stage> : fmt::formatter<string_view> {
     auto format(const locator::tablet_transition_stage&, fmt::format_context& ctx) const -> decltype(ctx.out());

main.cc

@@ -1658,7 +1658,6 @@ To start the scylla server proper, simply invoke as: scylla server (or just scyl
                     gossiper.local(), feature_service.local(), sys_ks.local(), group0_client, dbcfg.gossip_scheduling_group};
 
             service::tablet_allocator::config tacfg;
-            tacfg.initial_tablets_scale = cfg->tablets_initial_scale_factor();
             distributed<service::tablet_allocator> tablet_allocator;
             tablet_allocator.start(tacfg, std::ref(mm_notifier), std::ref(db)).get();
             auto stop_tablet_allocator = defer_verbose_shutdown("tablet allocator", [&tablet_allocator] {
@@ -1693,6 +1692,11 @@ To start the scylla server proper, simply invoke as: scylla server (or just scyl
             auto stop_tsm = defer_verbose_shutdown("topology_state_machine", [&tsm] {
                 tsm.stop().get();
             });
+            auto notify_topology = [&tsm] (auto) {
+                tsm.local().event.broadcast();
+            };
+            auto tablets_per_shard_goal_observer = cfg->tablets_per_shard_goal.observe(notify_topology);
+            auto tablets_initial_scale_factor_observer = cfg->tablets_initial_scale_factor.observe(notify_topology);
 
             auto compression_dict_updated_callback = [] () -> future<> {
                 auto dict = co_await sys_ks.local().query_dict();