You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by mg...@apache.org on 2026/07/10 08:36:16 UTC
(kudu) branch master updated: KUDU-3767 Add global leader awareness to auto leader rebalancer (Part 2)
This is an automated email from the ASF dual-hosted git repository.
martongreber pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
The following commit(s) were added to refs/heads/master by this push:
new 6124d83bc KUDU-3767 Add global leader awareness to auto leader rebalancer (Part 2)
6124d83bc is described below
commit 6124d83bc1f97b426a8efe47969845003266540d
Author: Gabriella Lotz <lo...@gmail.com>
AuthorDate: Mon Jun 15 16:29:50 2026 +0200
KUDU-3767 Add global leader awareness to auto leader rebalancer (Part 2)
Part 1 threaded a global leader count map through the per-table loop to
break ties when scoring destinations. That steers individual transfers
but can't fix skew that spans tables: since each table is balanced
independently, a tserver can keep getting the ceiling ('+1') allocation
across many tables and end up holding far more leaders than its peers,
even though every table is locally balanced. A typical case is many
single-tablet tables whose lone leaders all land on the same tserver.
This patch adds the post-loop global pass promised in Part 1. Once every
table is balanced on its own, it moves leaders off tservers that are above
the global ceiling onto ones with room below it, but only when doing so
keeps the affected table within its own floor/ceil. The pass runs only in
a round where per-table balancing scheduled nothing (to avoid racing its
in-flight, asynchronous moves) and is best-effort, logging and skipping
transient step-down failures.
Change-Id: I5d44761f6444ca1d1bf4fcee405d65d029e41980
Reviewed-on: http://gerrit.cloudera.org:8080/24476
Reviewed-by: Alexey Serbin <al...@apache.org>
Reviewed-by: Marton Greber <gr...@gmail.com>
Tested-by: Marton Greber <gr...@gmail.com>
---
src/kudu/master/auto_leader_rebalancer-test.cc | 362 +++++++++++++++++++++++++
src/kudu/master/auto_leader_rebalancer.cc | 333 ++++++++++++++++++++++-
src/kudu/master/auto_leader_rebalancer.h | 25 ++
3 files changed, 705 insertions(+), 15 deletions(-)
diff --git a/src/kudu/master/auto_leader_rebalancer-test.cc b/src/kudu/master/auto_leader_rebalancer-test.cc
index 78558e8df..cfa4262cf 100644
--- a/src/kudu/master/auto_leader_rebalancer-test.cc
+++ b/src/kudu/master/auto_leader_rebalancer-test.cc
@@ -17,6 +17,7 @@
#include "kudu/master/auto_leader_rebalancer.h"
#include <algorithm>
+#include <atomic>
#include <cmath>
#include <cstdint>
#include <functional>
@@ -25,6 +26,7 @@
#include <memory>
#include <numeric>
#include <string>
+#include <tuple>
#include <type_traits>
#include <unordered_map>
#include <utility>
@@ -265,6 +267,103 @@ class LeaderRebalancerTest : public KuduTest {
return Status::OK();
}
+ // Creates 'num_tables' single-tablet RF=3 tables (named '<table_prefix><i>')
+ // and forces every leader onto ts0, so all leaders pile up on a single
+ // tserver. Each single-tablet table is balanced on its own (one leader equals
+ // the per-table target), so only the global pass can fix the cross-table skew.
+ // Appends the created table names to 'table_names'.
+ void CreateSingleTabletTablesOnTs0(int num_tables,
+ const string& table_prefix,
+ vector<string>* table_names) {
+ const int num_tservers = cluster_->num_tablet_servers();
+ for (int i = 0; i < num_tables; i++) {
+ const string name = strings::Substitute("$0$1", table_prefix, i);
+ table_names->emplace_back(name);
+ workload_.reset(new TestWorkload(cluster_.get()));
+ workload_->set_table_name(name);
+ workload_->set_num_tablets(1);
+ workload_->set_num_replicas(3);
+ workload_->Setup();
+ vector<int32_t> leader_distribution(num_tservers, 0);
+ leader_distribution[0] = 1;
+ ASSERT_OK(MakeLeaderDistribution(leader_distribution, name));
+ }
+ }
+
+ // Number of leader transfers the last rebalancer round scheduled.
+ int MovesScheduledLastRound() {
+ return cluster_->mini_master()->master()->catalog_manager()
+ ->auto_leader_rebalancer()->moves_scheduled_this_round_for_test_.load();
+ }
+
+ // Sums each tserver's leader count across all of 'table_names'.
+ Status GlobalLeaderDistribution(const vector<string>& table_names,
+ std::map<string, int32_t>* totals) {
+ totals->clear();
+ for (const auto& name : table_names) {
+ std::map<string, int32_t> dist;
+ RETURN_NOT_OK(GetLeaderDistribution(&dist, name));
+ for (const auto& [uuid, count] : dist) {
+ (*totals)[uuid] += count;
+ }
+ }
+ return Status::OK();
+ }
+
+ // Creates 'num_tables' single-tablet tables with every leader on ts0, then
+ // runs the full rebalancer repeatedly until the leaders are spread within
+ // floor/ceil of the average across all tservers.
+ void CheckGlobalLeaderBalanceConverges(int num_tables, const string& table_prefix) {
+ const int num_tservers = cluster_->num_tablet_servers();
+
+ vector<string> table_names;
+ NO_FATALS(CreateSingleTabletTablesOnTs0(num_tables, table_prefix, &table_names));
+
+ const string ts0_uuid = cluster_->mini_tablet_server(0)->uuid();
+ using LeaderMap = std::map<string, int32_t>;
+
+ // Wait for the forced leader moves to settle: every leader should be on ts0.
+ ASSERT_EVENTUALLY([&] {
+ LeaderMap totals;
+ ASSERT_OK(GlobalLeaderDistribution(table_names, &totals));
+ ASSERT_EQ(num_tables, totals.at(ts0_uuid));
+ });
+
+ master::Master* master = cluster_->mini_master()->master();
+ master::AutoLeaderRebalancerTask* leader_rebalancer =
+ master->catalog_manager()->auto_leader_rebalancer();
+
+ // Run the full rebalancer (per-table loop followed by the global pass) a few
+ // times, letting the transfers settle between rounds, until every tserver
+ // holds floor or ceil of the average.
+ constexpr const int32_t kRetries = 10;
+ const double expected_leader_num = static_cast<double>(num_tables) / num_tservers;
+ LeaderMap totals;
+ for (int i = 0; i < kRetries; i++) {
+ SCOPED_TRACE(strings::Substitute("try $0", i));
+ ASSERT_OK(leader_rebalancer->RunLeaderRebalancer());
+ SleepFor(MonoDelta::FromMilliseconds(FLAGS_heartbeat_interval_ms));
+ ASSERT_OK(GlobalLeaderDistribution(table_names, &totals));
+ bool balanced = true;
+ for (const auto& entry : totals) {
+ if (entry.second < std::floor(expected_leader_num) ||
+ entry.second > std::ceil(expected_leader_num)) {
+ balanced = false;
+ break;
+ }
+ }
+ if (balanced) {
+ break;
+ }
+ }
+
+ ASSERT_OK(GlobalLeaderDistribution(table_names, &totals));
+ for (const auto& [uuid, count] : totals) {
+ ASSERT_GE(count, std::floor(expected_leader_num)) << uuid;
+ ASSERT_LE(count, std::ceil(expected_leader_num)) << uuid;
+ }
+ }
+
protected:
unique_ptr<InternalMiniCluster> cluster_;
InternalMiniClusterOptions cluster_opts_;
@@ -691,6 +790,269 @@ TEST_F(LeaderRebalancerTest, MultiTableLeaderBalance) {
NO_PENDING_FATALS();
}
+// Check that the post-loop global pass corrects cross-table skew that per-table
+// balancing leaves behind.
+//
+// Setup: three single-tablet RF=3 tables on a 3 tserver cluster, with every
+// leader forced onto ts0 ({3,0,0} globally). The per-table loop schedules
+// nothing (each single-tablet table is already balanced), yet the cluster is
+// badly skewed. The global pass should notice ts0 is above the global ceiling
+// (ceil(3/3)=1) and spread the leaders to one per tserver. Without the global
+// pass the per-table loop alone leaves ts0 holding all three, so this guards
+// the global pass from regressing.
+TEST_F(LeaderRebalancerTest, GlobalLeaderBalanceAcrossTables) {
+ cluster_opts_.num_tablet_servers = 3;
+ ASSERT_OK(CreateAndStartCluster());
+ // 3 tables / 3 tservers divides evenly, so the ideal spread is {1,1,1}.
+ NO_FATALS(CheckGlobalLeaderBalanceConverges(3, "global_leader_balance_table"));
+}
+
+// Like GlobalLeaderBalanceAcrossTables, but with a leader count that does not
+// divide evenly across the tservers, which is the case the global pass used to
+// get stuck on.
+//
+// Setup: five single-tablet RF=3 tables on a 3 tserver cluster, every leader
+// forced onto ts0 ({5,0,0} globally). The ideal spread is {2,2,1} (floor(5/3)=1,
+// ceil(5/3)=2).
+//
+// The point of this test is the ceiling-vs-floor destination bar. When the
+// global pass only sends leaders to tservers strictly below the floor, ts0 can
+// fill ts1 and ts2 up to one leader each and then stall at {3,1,1}, leaving ts0
+// permanently above the ceiling. Allowing a destination anywhere below the
+// ceiling lets ts0 hand a second leader to one of them, reaching {2,2,1}. So
+// this test fails if the destination bar regresses back to the floor.
+TEST_F(LeaderRebalancerTest, GlobalLeaderBalanceUneven) {
+ cluster_opts_.num_tablet_servers = 3;
+ ASSERT_OK(CreateAndStartCluster());
+ NO_FATALS(CheckGlobalLeaderBalanceConverges(5, "global_leader_uneven_table"));
+}
+
+// The global pass must never send a leader to a tserver in maintenance mode,
+// and must leave such a tserver out of the average it balances toward.
+//
+// Setup: four single-tablet RF=3 tables with every leader on ts0, then ts2 is
+// put into maintenance mode. Only the global pass can rebalance these tables (a
+// single-tablet table is balanced on its own), and with ts2 excluded it should
+// even the leaders across ts0 and ts1 only, ending at {2,2,0} and never placing
+// a leader on the draining ts2.
+TEST_F(LeaderRebalancerTest, GlobalLeaderBalanceSkipsMaintenanceMode) {
+ cluster_opts_.num_tablet_servers = 3;
+ ASSERT_OK(CreateAndStartCluster());
+
+ constexpr const int kNumTables = 4;
+ vector<string> table_names;
+ NO_FATALS(CreateSingleTabletTablesOnTs0(
+ kNumTables, "global_leader_maint_table", &table_names));
+
+ const string ts0_uuid = cluster_->mini_tablet_server(0)->uuid();
+ const string ts1_uuid = cluster_->mini_tablet_server(1)->uuid();
+ const string ts2_uuid = cluster_->mini_tablet_server(2)->uuid();
+ using LeaderMap = std::map<string, int32_t>;
+
+ // Wait for all leaders to land on ts0.
+ ASSERT_EVENTUALLY([&] {
+ LeaderMap totals;
+ ASSERT_OK(GlobalLeaderDistribution(table_names, &totals));
+ ASSERT_EQ(kNumTables, totals.at(ts0_uuid));
+ });
+
+ master::Master* master = cluster_->mini_master()->master();
+ // Put ts2 into maintenance mode so the global pass must avoid it both as a
+ // destination and in the average it balances toward.
+ ASSERT_OK(master->ts_manager()->SetTServerState(
+ ts2_uuid,
+ TServerStatePB::MAINTENANCE_MODE,
+ ChangeTServerStateRequestPB::ALLOW_MISSING_TSERVER,
+ master->catalog_manager()->sys_catalog()));
+
+ master::AutoLeaderRebalancerTask* leader_rebalancer =
+ master->catalog_manager()->auto_leader_rebalancer();
+
+ constexpr const int32_t kRetries = 10;
+ LeaderMap totals;
+ for (int i = 0; i < kRetries; i++) {
+ SCOPED_TRACE(strings::Substitute("try $0", i));
+ ASSERT_OK(leader_rebalancer->RunLeaderRebalancer());
+ SleepFor(MonoDelta::FromMilliseconds(FLAGS_heartbeat_interval_ms));
+ ASSERT_OK(GlobalLeaderDistribution(table_names, &totals));
+ if (totals[ts2_uuid] == 0 &&
+ totals[ts0_uuid] == kNumTables / 2 &&
+ totals[ts1_uuid] == kNumTables / 2) {
+ break;
+ }
+ }
+
+ ASSERT_OK(GlobalLeaderDistribution(table_names, &totals));
+ // ts2 must hold no leaders, and the rest are split evenly over ts0 and ts1.
+ ASSERT_EQ(0, totals[ts2_uuid]);
+ ASSERT_EQ(kNumTables / 2, totals[ts0_uuid]);
+ ASSERT_EQ(kNumTables / 2, totals[ts1_uuid]);
+}
+
+// The global pass must honor leader_rebalancing_max_moves_per_round. With the
+// cap set to 1 and five single-tablet tables all led by ts0 (so per-table
+// balancing is a no-op and only the global pass schedules anything), a single
+// round should schedule exactly one move even though more are needed to balance.
+TEST_F(LeaderRebalancerTest, GlobalLeaderBalanceRespectsMaxMovesPerRound) {
+ const auto saved_max_moves = FLAGS_leader_rebalancing_max_moves_per_round;
+ FLAGS_leader_rebalancing_max_moves_per_round = 1;
+
+ cluster_opts_.num_tablet_servers = 3;
+ ASSERT_OK(CreateAndStartCluster());
+
+ constexpr const int kNumTables = 5;
+ vector<string> table_names;
+ NO_FATALS(CreateSingleTabletTablesOnTs0(
+ kNumTables, "global_leader_cap_table", &table_names));
+
+ const string ts0_uuid = cluster_->mini_tablet_server(0)->uuid();
+ using LeaderMap = std::map<string, int32_t>;
+ ASSERT_EVENTUALLY([&] {
+ LeaderMap totals;
+ ASSERT_OK(GlobalLeaderDistribution(table_names, &totals));
+ ASSERT_EQ(kNumTables, totals.at(ts0_uuid));
+ });
+
+ master::Master* master = cluster_->mini_master()->master();
+ master::AutoLeaderRebalancerTask* leader_rebalancer =
+ master->catalog_manager()->auto_leader_rebalancer();
+
+ // A single round: the scheduled-move count reflects only the global pass
+ // (per-table balancing does nothing for single-tablet tables), and the cap
+ // must hold it to one even though four moves are needed to reach {2,2,1}.
+ ASSERT_OK(leader_rebalancer->RunLeaderRebalancer());
+ const int scheduled = MovesScheduledLastRound();
+
+ // Restore the flag before asserting so a failure here doesn't leak it into
+ // later tests.
+ FLAGS_leader_rebalancing_max_moves_per_round = saved_max_moves;
+ ASSERT_EQ(1, scheduled);
+}
+
+// The global pass is deferred to a later round whenever per-table balancing
+// still has moves to make, so the two never act on the cluster at the same
+// time (which would race on the same tablets).
+//
+// Setup: one 9-tablet table seeded badly skewed ({7,2,0}), which keeps the
+// per-table loop busy, plus two single-tablet tables holding a purely global
+// skew (all leaders on ts0, movable only by the global pass). In the round
+// where the per-table loop works the skewed table, the single-tablet leaders
+// must stay put because the global pass does not run.
+TEST_F(LeaderRebalancerTest, GlobalLeaderBalanceDeferredWhilePerTableBusy) {
+ cluster_opts_.num_tablet_servers = 3;
+ ASSERT_OK(CreateAndStartCluster());
+
+ // A 9-tablet table seeded {7,2,0}: per-table balancing has work to do here.
+ const string kSkewedTable = "global_leader_gate_skewed";
+ workload_.reset(new TestWorkload(cluster_.get()));
+ workload_->set_table_name(kSkewedTable);
+ workload_->set_num_tablets(9);
+ workload_->set_num_replicas(3);
+ workload_->Setup();
+ ASSERT_OK(MakeLeaderDistribution({7, 2, 0}, kSkewedTable));
+
+ // Single-tablet tables whose leaders are all on ts0: only the global pass can
+ // move these.
+ constexpr const int kNumGlobalTables = 2;
+ vector<string> global_tables;
+ NO_FATALS(CreateSingleTabletTablesOnTs0(
+ kNumGlobalTables, "global_leader_gate_table", &global_tables));
+
+ const string ts0_uuid = cluster_->mini_tablet_server(0)->uuid();
+ using LeaderMap = std::map<string, int32_t>;
+
+ // Wait for the seeded distributions to settle.
+ ASSERT_EVENTUALLY([&] {
+ LeaderMap skewed;
+ ASSERT_OK(GetLeaderDistribution(&skewed, kSkewedTable));
+ ASSERT_EQ(7, skewed.at(ts0_uuid));
+ LeaderMap global_totals;
+ ASSERT_OK(GlobalLeaderDistribution(global_tables, &global_totals));
+ ASSERT_EQ(kNumGlobalTables, global_totals.at(ts0_uuid));
+ });
+
+ master::Master* master = cluster_->mini_master()->master();
+ master::AutoLeaderRebalancerTask* leader_rebalancer =
+ master->catalog_manager()->auto_leader_rebalancer();
+
+ // One round. The per-table loop has work (the skewed table), so the global
+ // pass is gated off this round.
+ ASSERT_OK(leader_rebalancer->RunLeaderRebalancer());
+ // Give any erroneously-issued global transfers time to land before asserting
+ // that they did not happen.
+ SleepFor(MonoDelta::FromMilliseconds(5 * FLAGS_heartbeat_interval_ms));
+
+ // The global-skew tables must be untouched: their leaders are still all on
+ // ts0 because the global pass did not run this round.
+ LeaderMap global_totals;
+ ASSERT_OK(GlobalLeaderDistribution(global_tables, &global_totals));
+ ASSERT_EQ(kNumGlobalTables, global_totals[ts0_uuid]);
+
+ // Sanity check that the per-table loop actually had work this round (so the
+ // gate was exercised): the skewed table moves off {7,2,0}.
+ ASSERT_EVENTUALLY([&] {
+ LeaderMap skewed;
+ ASSERT_OK(GetLeaderDistribution(&skewed, kSkewedTable));
+ ASSERT_LT(skewed[ts0_uuid], 7);
+ });
+}
+
+// Single-replica (RF=1) tables have no follower to hand leadership to. Their
+// leaders are counted as load (so the pass won't pile movable leaders onto a
+// tserver already busy with them) but must never be moved. This exercises that
+// edge case: a mix of RF=1 and RF=3 tables where the RF=1 leaders stay put and
+// the pass keeps working normally.
+TEST_F(LeaderRebalancerTest, GlobalLeaderBalanceIgnoresSingleReplicaTables) {
+ cluster_opts_.num_tablet_servers = 3;
+ ASSERT_OK(CreateAndStartCluster());
+ // Disable replica rebalancing so RF=1 replicas (which are also their leaders)
+ // aren't relocated out from under us; we only want to observe leader moves.
+ const bool saved_auto_rebalancing = FLAGS_auto_rebalancing_enabled;
+ FLAGS_auto_rebalancing_enabled = false;
+
+ // A few RF=1 tables: the sole replica is the leader and cannot be moved.
+ constexpr const int kNumRf1Tables = 4;
+ vector<string> rf1_tables;
+ for (int i = 0; i < kNumRf1Tables; i++) {
+ const string name = strings::Substitute("global_leader_rf1_table$0", i);
+ rf1_tables.emplace_back(name);
+ workload_.reset(new TestWorkload(cluster_.get()));
+ workload_->set_table_name(name);
+ workload_->set_num_tablets(1);
+ workload_->set_num_replicas(1);
+ workload_->Setup();
+ }
+
+ // RF=3 tables with every leader piled on ts0: a global skew the pass can fix
+ // while the RF=1 tables are present.
+ vector<string> rf3_tables;
+ NO_FATALS(CreateSingleTabletTablesOnTs0(3, "global_leader_rf1_mixed_table", &rf3_tables));
+
+ using LeaderMap = std::map<string, int32_t>;
+ // Snapshot where each RF=1 leader lives; it must not change.
+ LeaderMap rf1_before;
+ ASSERT_OK(GlobalLeaderDistribution(rf1_tables, &rf1_before));
+
+ master::Master* master = cluster_->mini_master()->master();
+ master::AutoLeaderRebalancerTask* leader_rebalancer =
+ master->catalog_manager()->auto_leader_rebalancer();
+
+ // Run several rounds; the RF=1 tables must not trip up the pass.
+ for (int i = 0; i < 10; i++) {
+ ASSERT_OK(leader_rebalancer->RunLeaderRebalancer());
+ SleepFor(MonoDelta::FromMilliseconds(FLAGS_heartbeat_interval_ms));
+ }
+
+ LeaderMap rf1_after;
+ ASSERT_OK(GlobalLeaderDistribution(rf1_tables, &rf1_after));
+
+ // Restore before asserting so a failure doesn't leak the flag into later tests.
+ FLAGS_auto_rebalancing_enabled = saved_auto_rebalancing;
+
+ // RF=1 leaders are immovable, so their placement is unchanged.
+ ASSERT_EQ(rf1_before, rf1_after);
+}
+
class FilterSoftDeletedTableTest :
public LeaderRebalancerTest,
public ::testing::WithParamInterface<bool> {
diff --git a/src/kudu/master/auto_leader_rebalancer.cc b/src/kudu/master/auto_leader_rebalancer.cc
index 33a72bdce..b0224714c 100644
--- a/src/kudu/master/auto_leader_rebalancer.cc
+++ b/src/kudu/master/auto_leader_rebalancer.cc
@@ -49,6 +49,7 @@
#include "kudu/rpc/messenger.h"
#include "kudu/rpc/rpc_controller.h"
#include "kudu/security/init.h"
+#include "kudu/tserver/tserver.pb.h"
#include "kudu/util/cow_object.h"
#include "kudu/util/flag_tags.h"
#include "kudu/util/monotime.h"
@@ -66,7 +67,10 @@ using kudu::rpc::MessengerBuilder;
using kudu::rpc::RpcController;
using std::map;
using std::nullopt;
+using std::pair;
+using std::shared_ptr;
using std::string;
+using std::unordered_map;
using std::unordered_set;
using std::vector;
using strings::Substitute;
@@ -136,8 +140,9 @@ Status AutoLeaderRebalancerTask::RunLeaderRebalanceForTable(
const scoped_refptr<TableInfo>& table_info,
const vector<string>& tserver_uuids,
const unordered_set<string>& exclude_dest_uuids,
- std::unordered_map<string, int>* global_leader_count,
- AutoLeaderRebalancerTask::ExecuteMode mode) {
+ unordered_map<string, int>* global_leader_count,
+ AutoLeaderRebalancerTask::ExecuteMode mode,
+ int* num_scheduled_moves) {
LOG(INFO) << Substitute("leader rebalance for table $0", table_info->table_name());
TableMetadataLock table_l(table_info.get(), LockMode::READ);
const SysTablesEntryPB& table_data = table_info->metadata().state().pb;
@@ -230,7 +235,7 @@ Status AutoLeaderRebalancerTask::RunLeaderRebalanceForTable(
// step 2.
// pick the servers which number of leaders greater than 1/3 of number of all replicas
// <uuid, number of replica, number of leader>
- map<string, std::pair<int32_t, int32_t>> replica_and_leader_count_by_ts_uuid;
+ map<string, pair<int32_t, int32_t>> replica_and_leader_count_by_ts_uuid;
// uuid->leader should transfer count
map<string, int32_t> leader_transfer_source;
size_t remaining_tablets = tablet_infos.size();
@@ -246,7 +251,7 @@ Status AutoLeaderRebalancerTask::RunLeaderRebalanceForTable(
auto* leader_tablet_ids_ptr = FindOrNull(leader_tablet_ids_by_ts_uuid, uuid);
int32_t leader_count = leader_tablet_ids_ptr ? leader_tablet_ids_ptr->size() : 0;
replica_and_leader_count_by_ts_uuid.insert(
- {uuid, std::pair<int32_t, int32_t>(replica_count, leader_count)});
+ {uuid, pair<int32_t, int32_t>(replica_count, leader_count)});
VLOG(1) << Substitute(
"uuid: $0, replica_count: $1, leader_count: $2", uuid, replica_count, leader_count);
@@ -274,7 +279,7 @@ Status AutoLeaderRebalancerTask::RunLeaderRebalanceForTable(
// Step 3.
// Generate transfer task, <tablet_id, from_uuid, to_uuid>
- map<string, std::pair<string, string>> leader_transfer_tasks;
+ map<string, pair<string, string>> leader_transfer_tasks;
for (const auto& from_info : leader_transfer_source) {
string leader_uuid = from_info.first;
int32_t need_transfer_count = from_info.second;
@@ -310,7 +315,7 @@ Status AutoLeaderRebalancerTask::RunLeaderRebalanceForTable(
if (ContainsKey(exclude_dest_uuids, follower_uuid)) {
continue;
}
- std::pair<int32_t, int32_t>& replica_and_leader_count =
+ pair<int32_t, int32_t>& replica_and_leader_count =
replica_and_leader_count_by_ts_uuid[follower_uuid];
int32_t replica_count = replica_and_leader_count.first;
if (replica_count <= 0) {
@@ -344,7 +349,7 @@ Status AutoLeaderRebalancerTask::RunLeaderRebalanceForTable(
if (dest_invalid || dest_follower_uuid.empty()) {
continue;
}
- std::pair<int32_t, int32_t>& replica_and_leader_count =
+ pair<int32_t, int32_t>& replica_and_leader_count =
replica_and_leader_count_by_ts_uuid[leader_uuid];
int32_t replica_count = replica_and_leader_count.first;
int32_t leader_count = replica_and_leader_count.second;
@@ -357,7 +362,7 @@ Status AutoLeaderRebalancerTask::RunLeaderRebalanceForTable(
}
leader_transfer_tasks.insert(
- {tablet_id, std::pair<string, string>(leader_uuid, dest_follower_uuid)});
+ {tablet_id, pair<string, string>(leader_uuid, dest_follower_uuid)});
replica_and_leader_count_by_ts_uuid[leader_uuid].second--;
replica_and_leader_count_by_ts_uuid[dest_follower_uuid].second++;
if (global_leader_count) {
@@ -379,7 +384,7 @@ Status AutoLeaderRebalancerTask::RunLeaderRebalanceForTable(
}
}
- if (mode == AutoLeaderRebalancerTask::ExecuteMode::TEST) {
+ if (PREDICT_FALSE(mode == AutoLeaderRebalancerTask::ExecuteMode::TEST)) {
if (!leader_transfer_tasks.empty()) {
return Status::IllegalState(Substitute("leader_transfer_task size should be 0, but $0",
leader_transfer_tasks.size()));
@@ -388,6 +393,9 @@ Status AutoLeaderRebalancerTask::RunLeaderRebalanceForTable(
}
moves_scheduled_this_round_for_test_ = leader_transfer_tasks.size();
+ if (num_scheduled_moves) {
+ *num_scheduled_moves += leader_transfer_tasks.size();
+ }
VLOG(1) << Substitute("leader rebalance tasks, size: $0, leader_transfer_source, size: $1",
moves_scheduled_this_round_for_test_.load(),
leader_transfer_source.size());
@@ -413,7 +421,7 @@ Status AutoLeaderRebalancerTask::RunLeaderRebalanceForTable(
if (!host_port) {
continue;
}
- std::shared_ptr<TSDescriptor> leader_desc;
+ shared_ptr<TSDescriptor> leader_desc;
if (!ts_manager_->LookupTSByUUID(leader_uuid, &leader_desc)) {
continue;
}
@@ -443,6 +451,282 @@ Status AutoLeaderRebalancerTask::RunLeaderRebalanceForTable(
return Status::OK();
}
+Status AutoLeaderRebalancerTask::RunGlobalLeaderRebalance(
+ const vector<scoped_refptr<TableInfo>>& table_infos,
+ const vector<string>& tserver_uuids,
+ const unordered_set<string>& exclude_dest_uuids,
+ unordered_map<string, int>* global_leader_count,
+ AutoLeaderRebalancerTask::ExecuteMode mode) {
+ // Callers always pass a non-null map; it isn't optional like it is for the
+ // per-table method.
+ DCHECK(global_leader_count);
+
+ // Destinations in maintenance mode are excluded: such tservers should be
+ // shedding leaders, so they neither count towards the global average nor act
+ // as a move source here. A tserver that hosts no replicas yet (e.g. one that
+ // just joined) still counts: it can only pull the average down, and since it
+ // is a follower of nothing it can never be selected as a destination below,
+ // so it is harmless.
+ vector<string> eligible_uuids;
+ eligible_uuids.reserve(tserver_uuids.size());
+ for (const auto& uuid : tserver_uuids) {
+ if (!ContainsKey(exclude_dest_uuids, uuid)) {
+ eligible_uuids.emplace_back(uuid);
+ }
+ }
+ // At least two tservers are needed to move a leader between them.
+ if (eligible_uuids.size() < 2) {
+ return Status::OK();
+ }
+
+ // 'ceil_avg' is the ceiling of the average number of leaders per eligible
+ // tserver. A perfectly balanced cluster has every tserver at the floor or the
+ // ceiling of that average. A tserver above 'ceil_avg' is overloaded and must
+ // shed leaders; any tserver still below 'ceil_avg' has room to take one.
+ // Using the ceiling (rather than the floor) as the destination bar matters
+ // when the leaders don't divide evenly: it lets an overloaded tserver hand a
+ // leader to one currently sitting at the floor, which is exactly the move
+ // needed to reach e.g. {3,3,2,2} instead of getting stuck at {4,2,2,2}.
+ //
+ // A leader is only ever handed to a tserver that already hosts a follower of
+ // the tablet in question, so a tserver holding no replicas (e.g. one that
+ // just joined) can never receive leadership here. Such a tserver can drag the
+ // average down, but at worst the pass makes no move for it rather than
+ // shuffling leadership back and forth.
+ int64_t total_leaders = 0;
+ for (const auto& uuid : eligible_uuids) {
+ total_leaders += FindWithDefault(*global_leader_count, uuid, 0);
+ }
+ const int64_t num_eligible = static_cast<int64_t>(eligible_uuids.size());
+ const int64_t ceil_avg = (total_leaders + num_eligible - 1) / num_eligible;
+
+ // Fast path: if nobody is above the ceiling there is nothing to correct, so
+ // skip the relatively expensive tablet location gathering below. This is the
+ // common case once a cluster has converged.
+ bool any_overloaded = false;
+ for (const auto& uuid : eligible_uuids) {
+ if (FindWithDefault(*global_leader_count, uuid, 0) > ceil_avg) {
+ any_overloaded = true;
+ break;
+ }
+ }
+ if (!any_overloaded) {
+ return Status::OK();
+ }
+
+ // Per-tablet leadership for a single table, gathered fresh from the catalog.
+ struct TabletPlacement {
+ string tablet_id;
+ string leader_uuid;
+ vector<string> follower_uuids;
+ };
+
+ // Plan and (in NORMAL mode) execute corrective moves, table by table. The
+ // global counts decide who is over/underloaded, while the per-table leader
+ // counts keep each affected table within its own floor/ceil: a leader is
+ // only moved from a tserver holding more of this table's leaders than the
+ // destination, so the table itself does not become skewed.
+ int remaining_moves = FLAGS_leader_rebalancing_max_moves_per_round;
+ int global_moves_scheduled = 0;
+ for (const auto& table_info : table_infos) {
+ if (remaining_moves <= 0) {
+ break;
+ }
+ TableMetadataLock table_l(table_info.get(), LockMode::READ);
+ const SysTablesEntryPB& table_data = table_info->metadata().state().pb;
+ const int replication_factor = table_data.num_replicas();
+ // Removed tables aren't rebalanced, and a single replica tablet has no
+ // follower to hand leadership to.
+ if (table_data.state() == SysTablesEntryPB::REMOVED || replication_factor <= 1) {
+ continue;
+ }
+
+ vector<TabletPlacement> placements;
+ // tserver uuid -> number of this table's leaders it currently holds.
+ map<string, int> table_leader_count_by_ts;
+ map<string, HostPort> host_port_by_leader_ts_uuid;
+
+ vector<scoped_refptr<TabletInfo>> tablet_infos;
+ table_info->GetAllTablets(&tablet_infos);
+ for (const auto& tablet : tablet_infos) {
+ TabletMetadataLock tablet_l(tablet.get(), LockMode::READ);
+
+ TabletLocationsPB locs_pb;
+ CatalogManager::TSInfosDict ts_infos_dict;
+ {
+ CatalogManager::ScopedLeaderSharedLock leaderlock(catalog_manager_);
+ RETURN_NOT_OK(leaderlock.first_failed_status());
+ RETURN_NOT_OK(catalog_manager_->GetTabletLocations(
+ tablet->id(),
+ ReplicaTypeFilter::VOTER_REPLICA,
+ /*use_external_addr=*/false,
+ &locs_pb,
+ &ts_infos_dict,
+ nullopt));
+ }
+
+ TabletPlacement placement;
+ placement.tablet_id = tablet->id();
+ for (const auto& r : locs_pb.interned_replicas()) {
+ const TSInfoPB& ts_info = *(ts_infos_dict.ts_info_pbs()[r.ts_info_idx()]);
+ const string& uuid = ts_info.permanent_uuid();
+ if (r.role() == RaftPeerPB::LEADER) {
+ placement.leader_uuid = uuid;
+ table_leader_count_by_ts[uuid]++;
+ InsertIfNotPresent(
+ &host_port_by_leader_ts_uuid, uuid, HostPortFromPB(ts_info.rpc_addresses(0)));
+ } else if (r.role() == RaftPeerPB::FOLLOWER) {
+ placement.follower_uuids.emplace_back(uuid);
+ }
+ }
+ // A tablet with no observed leader can't be a move source.
+ if (!placement.leader_uuid.empty()) {
+ placements.emplace_back(std::move(placement));
+ }
+ }
+
+ // Plan this table's corrective transfers, <tablet_id, <from_uuid, to_uuid>>.
+ map<string, pair<string, string>> leader_transfer_tasks;
+ for (const auto& placement : placements) {
+ if (remaining_moves <= 0) {
+ break;
+ }
+ const string& source_uuid = placement.leader_uuid;
+ // Only drain tservers that are above the global ceiling.
+ if (FindWithDefault(*global_leader_count, source_uuid, 0) <= ceil_avg) {
+ continue;
+ }
+ // Skip degraded tablets that lack a full set of voters; per-table
+ // balancing avoids them too.
+ if (placement.follower_uuids.size() + 1 < replication_factor) {
+ continue;
+ }
+ const int source_table_count = FindWithDefault(table_leader_count_by_ts, source_uuid, 0);
+
+ // Pick the most underloaded follower that can take a leader without
+ // skewing this table. Ties are broken by the smaller uuid so the result
+ // does not depend on the order replicas happen to come back in.
+ string dest_uuid;
+ int dest_global_count = 0;
+ for (const auto& follower_uuid : placement.follower_uuids) {
+ if (ContainsKey(exclude_dest_uuids, follower_uuid)) {
+ continue;
+ }
+ const int follower_global_count =
+ FindWithDefault(*global_leader_count, follower_uuid, 0);
+ // The destination must have room below the global ceiling ...
+ if (follower_global_count >= ceil_avg) {
+ continue;
+ }
+ // ... and must hold strictly fewer of this table's leaders than the
+ // source, so handing it one keeps the table within its own floor/ceil.
+ if (FindWithDefault(table_leader_count_by_ts, follower_uuid, 0) >= source_table_count) {
+ continue;
+ }
+ if (dest_uuid.empty() ||
+ follower_global_count < dest_global_count ||
+ (follower_global_count == dest_global_count && follower_uuid < dest_uuid)) {
+ dest_uuid = follower_uuid;
+ dest_global_count = follower_global_count;
+ }
+ }
+ if (dest_uuid.empty()) {
+ continue;
+ }
+
+ if (PREDICT_FALSE(mode == AutoLeaderRebalancerTask::ExecuteMode::TEST)) {
+ // In test mode we only want to learn whether the cluster is globally
+ // balanced, so report the first corrective move that would be made
+ // without planning further or mutating the caller's counts.
+ return Status::IllegalState(Substitute(
+ "global leader rebalance would move a leader off $0 for table $1",
+ source_uuid, table_data.name()));
+ }
+
+ // Update the counts as if the move has already happened so later tablets
+ // in this round see a consistent picture. The execution below may end up
+ // skipping a transfer (e.g. the destination enters maintenance mode or
+ // the leader can no longer be looked up), so these counts can drift from
+ // what actually transferred; that is fine, the next round rebuilds the
+ // distribution from scratch.
+ leader_transfer_tasks.insert(
+ {placement.tablet_id, pair<string, string>(source_uuid, dest_uuid)});
+ (*global_leader_count)[source_uuid]--;
+ (*global_leader_count)[dest_uuid]++;
+ table_leader_count_by_ts[source_uuid]--;
+ table_leader_count_by_ts[dest_uuid]++;
+ remaining_moves--;
+ }
+
+ if (leader_transfer_tasks.empty()) {
+ continue;
+ }
+ global_moves_scheduled += leader_transfer_tasks.size();
+
+ for (const auto& task : leader_transfer_tasks) {
+ const string& tablet_id = task.first;
+ const string& leader_uuid = task.second.first;
+ const string& dest_uuid = task.second.second;
+ if (PREDICT_FALSE(TServerStatePB::MAINTENANCE_MODE ==
+ ts_manager_->GetTServerState(dest_uuid))) {
+ continue;
+ }
+ auto* host_port = FindOrNull(host_port_by_leader_ts_uuid, leader_uuid);
+ if (!host_port) {
+ continue;
+ }
+ shared_ptr<TSDescriptor> leader_desc;
+ if (!ts_manager_->LookupTSByUUID(leader_uuid, &leader_desc)) {
+ continue;
+ }
+
+ LeaderStepDownRequestPB request;
+ request.set_dest_uuid(leader_uuid);
+ request.set_tablet_id(tablet_id);
+ request.set_mode(LeaderStepDownMode::GRACEFUL);
+ request.set_new_leader_uuid(dest_uuid);
+
+ LeaderStepDownResponsePB response;
+ RpcController rpc;
+ rpc.set_timeout(MonoDelta::FromSeconds(FLAGS_auto_leader_rebalancing_rpc_timeout_seconds));
+
+ // This pass is best-effort: a transfer may fail transiently (for example
+ // "already in progress" if a previous round's move for this tablet has
+ // not settled yet). Log and move on rather than aborting the round, since
+ // the next round re-evaluates the distribution from scratch.
+ vector<Sockaddr> resolved;
+ Status s = host_port->ResolveAddresses(&resolved);
+ if (!s.ok()) {
+ WARN_NOT_OK(s, Substitute("global leader rebalance: cannot resolve $0", leader_uuid));
+ continue;
+ }
+ ConsensusServiceProxy proxy(messenger_, resolved[0], host_port->host());
+ s = proxy.LeaderStepDown(request, &response, &rpc);
+ if (!s.ok()) {
+ WARN_NOT_OK(s, Substitute(
+ "global leader rebalance: leader step down for tablet $0 failed", tablet_id));
+ continue;
+ }
+ if (!response.has_error()) {
+ VLOG(1) << Substitute(
+ "global leader rebalance transfer table: $0, tablet_id: $1, from: $2 to: $3",
+ table_data.name(), tablet_id, leader_uuid, dest_uuid);
+ } else {
+ LOG(WARNING) << Substitute(
+ "global leader rebalance: transfer for tablet $0 (from $1 to $2) failed: $3",
+ tablet_id, leader_uuid, dest_uuid, response.error().ShortDebugString());
+ }
+ }
+ }
+
+ // In TEST mode the method returns early above as soon as a move is found, so
+ // reaching here means nothing was scheduled and this is a no-op increment.
+ moves_scheduled_this_round_for_test_ += global_moves_scheduled;
+ VLOG(0) << Substitute("global leader rebalance finished, scheduled $0 move(s)",
+ global_moves_scheduled);
+ return Status::OK();
+}
+
Status AutoLeaderRebalancerTask::RunLeaderRebalancer() {
std::lock_guard guard(running_mutex_);
@@ -498,18 +782,37 @@ Status AutoLeaderRebalancerTask::RunLeaderRebalancer() {
// Start the global map at 0 for every live tserver. As each table is processed
// we add its real leader counts and apply any planned moves, so later tables
// can see the leader load across all tables when they need to break a tie.
- std::unordered_map<string, int> global_leader_count_by_ts_uuid;
+ unordered_map<string, int> global_leader_count_by_ts_uuid;
for (const auto& uuid : tserver_uuids) {
global_leader_count_by_ts_uuid[uuid] = 0;
}
+ int per_table_moves_scheduled = 0;
for (const auto& table_info : table_infos) {
RETURN_NOT_OK(RunLeaderRebalanceForTable(
- table_info, tserver_uuids, exclude_dest_uuids, &global_leader_count_by_ts_uuid));
+ table_info, tserver_uuids, exclude_dest_uuids, &global_leader_count_by_ts_uuid,
+ AutoLeaderRebalancerTask::ExecuteMode::NORMAL, &per_table_moves_scheduled));
+ }
+ // Every table is now balanced on its own, but the cluster can still be
+ // globally skewed if one tserver kept getting the ceiling allocation across
+ // tables (e.g. many single-tablet tables whose lone leader landed on it).
+ // Run a global pass to shed leaders from globally overloaded tservers onto
+ // underloaded ones without pushing any individual table out of balance.
+ //
+ // Only do this once per-table balancing has nothing left to do this round.
+ // While per-table moves are still being scheduled the cluster is mid-flight:
+ // running the global pass now would re-target tablets the per-table loop just
+ // moved (the leadership transfers are asynchronous), and a follow-up round
+ // does the global correction with a consistent view instead.
+ //
+ // TODO(gabriellalotz): if per-table balancing keeps finding work every round
+ // (e.g. from continuous leader elections), the global pass can be starved.
+ // Emit a metric when it's skipped for this reason so the situation is
+ // diagnosable.
+ if (per_table_moves_scheduled == 0) {
+ RETURN_NOT_OK(RunGlobalLeaderRebalance(
+ table_infos, tserver_uuids, exclude_dest_uuids, &global_leader_count_by_ts_uuid));
}
- // TODO(KUDU-3767): add a post-loop global pass here to detect and correct
- // tservers that are globally overloaded even when per-table balance is achieved
- // (e.g. many single-tablet tables whose leaders all land on the same tserver).
// @TODO(duyuqi)
// Enrich the log and add metrics for leader rebalancer.
LOG(INFO) << "All tables' leader rebalancing finished this round";
diff --git a/src/kudu/master/auto_leader_rebalancer.h b/src/kudu/master/auto_leader_rebalancer.h
index 9ff0e5be0..6577044e6 100644
--- a/src/kudu/master/auto_leader_rebalancer.h
+++ b/src/kudu/master/auto_leader_rebalancer.h
@@ -77,11 +77,36 @@ class AutoLeaderRebalancerTask {
// global_leader_count, when provided, is read as a tiebreaker during destination scoring and
// updated with every planned move so that successive table calls share a consistent view of
// in-round leader distribution.
+ // num_scheduled_moves, when provided, is incremented by the number of leader transfers planned
+ // for this table, letting the caller tell whether per-table balancing still has work to do.
Status RunLeaderRebalanceForTable(
const scoped_refptr<TableInfo>& table_info,
const std::vector<std::string>& tserver_uuids,
const std::unordered_set<std::string>& exclude_dest_uuids,
std::unordered_map<std::string, int>* global_leader_count = nullptr,
+ AutoLeaderRebalancerTask::ExecuteMode mode = AutoLeaderRebalancerTask::ExecuteMode::NORMAL,
+ int* num_scheduled_moves = nullptr);
+
+ // Runs a global corrective pass once every table has been balanced on its
+ // own by RunLeaderRebalanceForTable. Per-table balancing keeps each table
+ // within floor/ceil of its own average, but it can still leave the cluster
+ // globally skewed when the same tserver repeatedly receives the ceiling
+ // (the extra '+1') allocation across many tables, e.g. lots of single-tablet
+ // or otherwise indivisible tables whose lone leader keeps landing on it.
+ //
+ // 'global_leader_count' is the in-round leader distribution accumulated by
+ // the per-table calls. Tservers above the global ceiling are treated as move
+ // sources and tservers below the global ceiling as destinations. A leader is
+ // only transferred when the destination holds strictly fewer of that table's
+ // leaders than the source, so the move improves global balance without
+ // pushing the affected table out of its own floor/ceil range. The map is
+ // updated for every planned move so successive tables share a consistent
+ // view.
+ Status RunGlobalLeaderRebalance(
+ const std::vector<scoped_refptr<TableInfo>>& table_infos,
+ const std::vector<std::string>& tserver_uuids,
+ const std::unordered_set<std::string>& exclude_dest_uuids,
+ std::unordered_map<std::string, int>* global_leader_count,
AutoLeaderRebalancerTask::ExecuteMode mode = AutoLeaderRebalancerTask::ExecuteMode::NORMAL);
// Only one task can be scheduled at a time.