KUDU-3476: Make replica placement range and table aware

Previously, the replica selection policy randomly selected
two tablet servers and placed the replica on the tserver
with less replicas. This could lead to hotspotting if
placing replicas from the same range on the same set of
tservers since the policy doesn't discriminate by range.

With this patch, the replica selection policy now ranks
the available tservers by range and table load and places
the replica accordingly. It prioritizes replicas by range
first, replicas by table are used as a tiebreaker, then
total replicas is used as the final tiebreaker. The range
and table load is determined by the existing number of
replicas before the placement begins and the number of
pending replicas placed on the tserver while placing replicas.

The flag --enable_range_replica_placement on the master side
controls whether or not this new policy is used.
For this feature to work, both the range start key and the
table id of the table the reange belongs to must be defined.
This is because multiple tables could have the same range
defined by the same range start key, so to
differentiate the ranges, the table id is required.

The link to the design doc is here:
https://docs.google.com/document/d/1r-p0GW8lj2iLA3VGvZWAem09ykCmR5jEe8npUhJ07G8/edit?usp=sharing

Change-Id: I9caeb8d5547e946bfeb152a99e1ec034c3fa0a0f
Reviewed-on: http://gerrit.cloudera.org:8080/19931
Tested-by: Alexey Serbin <alexey@apache.org>
Reviewed-by: Alexey Serbin <alexey@apache.org>

Author: mreddy7

src/kudu/integration-tests/create-table-itest.cc

@@ -40,6 +40,7 @@
 #include "kudu/client/shared_ptr.h" // IWYU pragma: keep
 #include "kudu/common/common.pb.h"
 #include "kudu/common/partial_row.h"
+#include "kudu/common/partition.h"
 #include "kudu/common/schema.h"
 #include "kudu/common/wire_protocol-test-util.h"
 #include "kudu/gutil/mathlimits.h"
@@ -53,7 +54,9 @@
 #include "kudu/mini-cluster/external_mini_cluster.h"
 #include "kudu/mini-cluster/mini_cluster.h"
 #include "kudu/rpc/rpc_controller.h"
+#include "kudu/tablet/tablet.pb.h"
 #include "kudu/tools/tool_test_util.h"
+#include "kudu/tserver/tserver.pb.h"
 #include "kudu/util/atomic.h"
 #include "kudu/util/metrics.h"
 #include "kudu/util/monotime.h"
@@ -199,7 +202,10 @@ TEST_F(CreateTableITest, TestCreateWhenMajorityOfReplicasFailCreation) {
 TEST_F(CreateTableITest, TestSpreadReplicasEvenly) {
   const int kNumServers = 10;
   const int kNumTablets = 20;
-  NO_FATALS(StartCluster({}, {}, kNumServers));
+  vector<string> master_flags = {
+      "--enable_range_replica_placement=false",
+  };
+  NO_FATALS(StartCluster({}, master_flags, kNumServers));
 
   unique_ptr<client::KuduTableCreator> table_creator(client_->NewTableCreator());
   auto client_schema = KuduSchema::FromSchema(GetSimpleTestSchema());
@@ -399,6 +405,99 @@ TEST_F(CreateTableITest, TestSpreadReplicasEvenlyWithDimension) {
   }
 }
 
+// Tests the range aware replica placement by adding multiple tables with multiple ranges
+// and checking the replica distribution.
+TEST_F(CreateTableITest, TestSpreadReplicas) {
+  const int kNumServers = 5;
+  const int kNumReplicas = 3;
+  NO_FATALS(StartCluster({ }, { }, kNumServers));
+
+  Schema schema = Schema({ ColumnSchema("key1", INT32),
+                           ColumnSchema("key2", INT32),
+                           ColumnSchema("int_val", INT32),
+                           ColumnSchema("string_val", STRING, true) }, 2);
+  auto client_schema = KuduSchema::FromSchema(schema);
+
+  auto create_table_func = [](KuduClient* client,
+                              KuduSchema* client_schema,
+                              const string& table_name,
+                              const vector<std::pair<int32_t, int32_t>> range_bounds,
+                              const int num_buckets) {
+    unique_ptr<client::KuduTableCreator> table_creator(client->NewTableCreator());
+    table_creator->table_name(table_name)
+        .schema(client_schema)
+        .add_hash_partitions({ "key1" }, num_buckets)
+        .set_range_partition_columns({ "key2" })
+        .num_replicas(kNumReplicas);
+    for (const auto& range_bound : range_bounds) {
+      unique_ptr<KuduPartialRow> lower_bound(client_schema->NewRow());
+      RETURN_NOT_OK(lower_bound->SetInt32("key2", range_bound.first));
+      unique_ptr<KuduPartialRow> upper_bound(client_schema->NewRow());
+      RETURN_NOT_OK(upper_bound->SetInt32("key2", range_bound.second));
+      table_creator->add_range_partition(lower_bound.release(), upper_bound.release());
+    }
+    return table_creator->Create();
+  };
+
+  vector<string> tables = {"table1", "table2", "table3", "table4"};
+  vector<std::pair<int32_t, int32_t>> range_bounds =
+      { {0, 100}, {100, 200}, {200, 300}, {300, 400}};
+  const int doubleNumBuckets = 10;
+  const int numBuckets = 5;
+  for (const auto& table : tables) {
+    if (table == "table1") {
+      ASSERT_OK(create_table_func(
+          client_.get(), &client_schema, table, range_bounds, doubleNumBuckets));
+    } else {
+      ASSERT_OK(create_table_func(
+          client_.get(), &client_schema, table, range_bounds, numBuckets));
+    }
+  }
+
+  // Stats of number of replicas per range per table per tserver.
+  typedef std::unordered_map<string, std::unordered_map<string, int>> replicas_per_range_per_table;
+  std::unordered_map<int, replicas_per_range_per_table> stats;
+  for (int ts_idx = 0; ts_idx < kNumServers; ts_idx++) {
+    rpc::RpcController rpc;
+    tserver::ListTabletsRequestPB req;
+    tserver::ListTabletsResponsePB resp;
+    cluster_->tserver_proxy(ts_idx)->ListTablets(req, &resp, &rpc);
+    for (auto i = 0; i < resp.status_and_schema_size(); ++i) {
+      auto tablet_status = resp.status_and_schema(i).tablet_status();
+      if (tablet_status.has_partition()) {
+        Partition partition;
+        Partition::FromPB(tablet_status.partition(), &partition);
+        auto range_start_key = partition.begin().range_key();
+        auto table_name = tablet_status.table_name();
+        ++stats[ts_idx][table_name][range_start_key];
+      }
+    }
+  }
+
+  ASSERT_EQ(kNumServers, stats.size());
+  for (const auto& stat : stats) {
+    int tserver_replicas = 0;
+    // Verifies that four tables exist on each tserver.
+    ASSERT_EQ(tables.size(), stat.second.size());
+    for (const auto& table : stat.second) {
+      // Verifies that the four ranges exist for each table on each tserver.
+      ASSERT_EQ(range_bounds.size(), table.second.size());
+      for (const auto& ranges : table.second) {
+        // Since there are ten buckets instead of five for table "table1",
+        // we expect twice as many replicas (6 instead of 3).
+        if (table.first == "table1") {
+          ASSERT_EQ(doubleNumBuckets * kNumReplicas / kNumServers, ranges.second);;
+        } else {
+          ASSERT_EQ(numBuckets * kNumReplicas / kNumServers, ranges.second);
+        }
+        tserver_replicas += ranges.second;
+      }
+    }
+    // Verifies that 60 replicas are placed on each tserver, 300 total across 5 tservers.
+    ASSERT_EQ(60, tserver_replicas);
+  }
+}
+
 static void LookUpRandomKeysLoop(const std::shared_ptr<master::MasterServiceProxy>& master,
                                  const char* table_name,
                                  AtomicBool* quit) {
@@ -472,7 +571,7 @@ TEST_F(CreateTableITest, TestCreateTableWithDeadTServers) {
   unique_ptr<client::KuduTableCreator> table_creator(client_->NewTableCreator());
 
   // Don't bother waiting for table creation to finish; it'll never happen
-  // because all of the tservers are dead.
+  // because all the tservers are dead.
   CHECK_OK(table_creator->table_name(kTableName)
            .schema(&client_schema)
            .set_range_partition_columns({ "key" })
@@ -688,7 +787,7 @@ TEST_P(NotEnoughHealthyTServersTest, TestNotEnoughHealthyTServers) {
     }
     // Wait the 3 tablet servers heartbeat timeout and unresponsive timeout. Then catalog
     // manager will take them as unavailable tablet servers. KSCK gets the status of tablet
-    // server from tablet serve interface. Here must wait the caltalog manager to take the
+    // server from tablet serve interface. Here must wait the catalog manager to take the
     // as unavailable.
     SleepFor(MonoDelta::FromMilliseconds(3*(kTSUnresponsiveTimeoutMs + kHeartbeatIntervalMs)));
   }
@@ -703,7 +802,7 @@ TEST_P(NotEnoughHealthyTServersTest, TestNotEnoughHealthyTServers) {
   {
     // Restart the first tablet server.
     NO_FATALS(cluster_->tablet_server(0)->Restart());
-    // Wait the restarted tablet server to send a heartbeat and be registered in catalog manaager.
+    // Wait the restarted tablet server to send a heartbeat and be registered in catalog manager.
     SleepFor(MonoDelta::FromMilliseconds(kHeartbeatIntervalMs));
   }
 
@@ -713,7 +812,7 @@ TEST_P(NotEnoughHealthyTServersTest, TestNotEnoughHealthyTServers) {
   {
     // Restart the second tablet server.
     NO_FATALS(cluster_->tablet_server(1)->Restart());
-    // Wait the restarted tablet server to send a heartbeat and be registered in catalog manaager.
+    // Wait the restarted tablet server to send a heartbeat and be registered in catalog manager.
     SleepFor(MonoDelta::FromMilliseconds(kHeartbeatIntervalMs));
   }
 
@@ -755,7 +854,7 @@ TEST_P(NotEnoughHealthyTServersTest, TestNotEnoughHealthyTServers) {
     // Add one new tablet server.
     NO_FATALS(cluster_->AddTabletServer());
   } else {
-    // Restart the stopped tablet server
+    // Restart the stopped tablet server.
     NO_FATALS(cluster_->tablet_server(2)->Restart());
   }
 

src/kudu/master/catalog_manager.cc

@@ -228,7 +228,7 @@ TAG_FLAG(allow_unsafe_replication_factor, runtime);
 
 DEFINE_int32(catalog_manager_bg_task_wait_ms, 1000,
              "Amount of time the catalog manager background task thread waits "
-             "between runs");
+             "between runs.");
 TAG_FLAG(catalog_manager_bg_task_wait_ms, hidden);
 
 DEFINE_int32(max_create_tablets_per_ts, 60,
@@ -357,11 +357,11 @@ TAG_FLAG(auto_leader_rebalancing_enabled, runtime);
 
 DEFINE_uint32(table_locations_cache_capacity_mb, 0,
               "Capacity for the table locations cache (in MiB); a value "
-              "of 0 means table locations are not be cached");
+              "of 0 means table locations are not be cached.");
 TAG_FLAG(table_locations_cache_capacity_mb, advanced);
 
 DEFINE_bool(enable_per_range_hash_schemas, true,
-            "Whether to support range-specific hash schemas for tables");
+            "Whether to support range-specific hash schemas for tables.");
 TAG_FLAG(enable_per_range_hash_schemas, advanced);
 TAG_FLAG(enable_per_range_hash_schemas, runtime);
 
@@ -372,6 +372,10 @@ DEFINE_bool(enable_table_write_limit, false,
 TAG_FLAG(enable_table_write_limit, experimental);
 TAG_FLAG(enable_table_write_limit, runtime);
 
+DEFINE_bool(enable_range_replica_placement, true,
+            "Whether to use range aware replica placement for newly created tablets.");
+TAG_FLAG(enable_range_replica_placement, runtime);
+
 DEFINE_int64(table_disk_size_limit, -1,
              "Set the target size in bytes of a table to write. "
              "This is a system wide configuration for every newly "
@@ -385,7 +389,7 @@ DEFINE_int64(table_row_count_limit, -1,
 TAG_FLAG(table_row_count_limit, experimental);
 
 DEFINE_double(table_write_limit_ratio, 0.95,
-              "Set the ratio of how much write limit can be reached");
+              "Set the ratio of how much write limit can be reached.");
 TAG_FLAG(table_write_limit_ratio, experimental);
 
 DEFINE_bool(enable_metadata_cleanup_for_deleted_tables_and_tablets, false,
@@ -410,7 +414,7 @@ TAG_FLAG(enable_chunked_tablet_writes, runtime);
 DEFINE_bool(require_new_spec_for_custom_hash_schema_range_bound, false,
             "Whether to require the client to use newer signature to specify "
             "range bounds when working with a table having custom hash schema "
-            "per range");
+            "per range.");
 TAG_FLAG(require_new_spec_for_custom_hash_schema_range_bound, experimental);
 TAG_FLAG(require_new_spec_for_custom_hash_schema_range_bound, runtime);
 
@@ -5088,16 +5092,29 @@ bool AsyncAddReplicaTask::SendRequest(int attempt) {
     TSDescriptorVector ts_descs;
     master_->ts_manager()->GetDescriptorsAvailableForPlacement(&ts_descs);
 
-    // Get the dimension of the tablet. Otherwise, it will be nullopt.
+    // Get the dimension, table id, and range start key of the tablet.
     optional<string> dimension = nullopt;
+    optional<string> table_id = nullopt;
+    optional<string> range_key_start = nullopt;
     {
       TabletMetadataLock l(tablet_.get(), LockMode::READ);
       if (tablet_->metadata().state().pb.has_dimension_label()) {
         dimension = tablet_->metadata().state().pb.dimension_label();
       }
+      if (FLAGS_enable_range_replica_placement) {
+        Partition partition;
+        if (tablet_->metadata().state().pb.has_partition()) {
+          const auto& tablet_partition = tablet_->metadata().state().pb.partition();
+          Partition::FromPB(tablet_partition, &partition);
+        }
+        range_key_start = partition.begin().range_key();
+        VLOG(1) << Substitute("range_key_start is set to $1", range_key_start.value());
+        table_id = tablet_->metadata().state().pb.table_id();
+        VLOG(1) << Substitute("table_id is set to $1", table_id.value());
+      }
     }
 
-    // Some of the tablet servers hosting the current members of the config
+    // Some tablet servers hosting the current members of the config
     // (see the 'existing' populated above) might be presumably dead.
     // Inclusion of a presumably dead tablet server into 'existing' is OK:
     // PlacementPolicy::PlaceExtraTabletReplica() does not require elements of
@@ -5106,7 +5123,8 @@ bool AsyncAddReplicaTask::SendRequest(int attempt) {
     // to host the extra replica is 'ts_descs' after blacklisting all elements
     // common with 'existing'.
     PlacementPolicy policy(std::move(ts_descs), rng_);
-    s = policy.PlaceExtraTabletReplica(std::move(existing), dimension, &extra_replica);
+    s = policy.PlaceExtraTabletReplica(
+        std::move(existing), dimension, range_key_start, table_id, &extra_replica);
   }
   if (PREDICT_FALSE(!s.ok())) {
     auto msg = Substitute("no extra replica candidate found for tablet $0: $1",
@@ -6153,15 +6171,29 @@ Status CatalogManager::SelectReplicasForTablet(const PlacementPolicy& policy,
   config->set_obsolete_local(nreplicas == 1);
   config->set_opid_index(consensus::kInvalidOpIdIndex);
 
-  // Get the dimension of the tablet. Otherwise, it will be nullopt.
+  // Get the dimension, table id, and range start key of the tablet.
   optional<string> dimension = nullopt;
+  optional<string> table_id = nullopt;
+  optional<string> range_key_start = nullopt;
   if (tablet->metadata().state().pb.has_dimension_label()) {
     dimension = tablet->metadata().state().pb.dimension_label();
   }
+  if (FLAGS_enable_range_replica_placement) {
+    Partition partition;
+    if (tablet->metadata().state().pb.has_partition()) {
+      const auto& tablet_partition = tablet->metadata().state().pb.partition();
+      Partition::FromPB(tablet_partition, &partition);
+    }
+    range_key_start = partition.begin().range_key();
+    VLOG(1) << Substitute("range_key_start is set to $1", range_key_start.value());
+    table_id = tablet->metadata().state().pb.table_id();
+    VLOG(1) << Substitute("table_id is set to $1", table_id.value());
+  }
 
   // Select the set of replicas for the tablet.
   TSDescriptorVector descriptors;
-  RETURN_NOT_OK_PREPEND(policy.PlaceTabletReplicas(nreplicas, dimension, &descriptors),
+  RETURN_NOT_OK_PREPEND(policy.PlaceTabletReplicas(
+      nreplicas, dimension, range_key_start, table_id, &descriptors),
                         Substitute("failed to place replicas for tablet $0 "
                                    "(table '$1')",
                                    tablet->id(), table_guard.data().name()));

src/kudu/master/master.proto

@@ -335,6 +335,17 @@ message TabletReportUpdatesPB {
   repeated ReportedTabletUpdatesPB tablets = 1;
 }
 
+// The number of tablets that are BOOTSTRAPPING or RUNNING per range.
+message TabletsByRangePB {
+  optional bytes range_start_key = 1;
+  optional int32 tablets = 2;
+}
+
+// The number of tablets that are BOOTSTRAPPING or RUNNING per each range for each table.
+message TabletsByRangePerTablePB {
+  repeated TabletsByRangePB num_live_tablets_by_range = 1;
+}
+
 // Heartbeat sent from the tablet-server to the master
 // to establish liveness and report back any status changes.
 message TSHeartbeatRequestPB {
@@ -370,9 +381,14 @@ message TSHeartbeatRequestPB {
   optional consensus.ReplicaManagementInfoPB replica_management_info = 7;
 
   // The number of tablets that are BOOTSTRAPPING or RUNNING in each dimension.
-  // Used by the master to determine load when creating new tablet replicas
+  // Used by the master to determine load when placing new tablet replicas
   // based on dimension.
   map<string, int32> num_live_tablets_by_dimension = 8;
+
+  // Per table, the number of tablets that are BOOTSTRAPPING or RUNNING
+  // in each range. Used by the master to determine load when placing
+  // new tablet replicas based on range and table.
+  map<string, TabletsByRangePerTablePB> num_live_tablets_by_range_per_table = 9;
 }
 
 message TSHeartbeatResponsePB {

src/kudu/master/master_runner.cc

@@ -348,9 +348,9 @@ void SetMasterFlagDefaults() {
       to_string(kDefaultRpcServiceQueueLength).c_str(),
       SET_FLAGS_DEFAULT));
   // Master always reads the latest data snapshot from the system catalog and
-  // never uses any specific timestatmp in past for a read snapshot. With that,
-  // here isn't much sense to keep long chain of UNDO deltas in addition to the
-  // latest version in the MVCC. Keeping short history of deltas frees CPU
+  // never uses any specific past timestamp for a read snapshot. With that,
+  // it doesn't make sense to keep a long chain of UNDO deltas in addition to the
+  // latest version in the MVCC. Keeping a short history of deltas frees CPU
   // cycles, memory, and IO bandwidth that otherwise would be consumed by
   // background maintenance jobs running compactions. In addition, less disk
   // space is consumed to store the system tablet's data.

src/kudu/master/master_service.cc

@@ -451,11 +451,21 @@ void MasterServiceImpl::TSHeartbeat(const TSHeartbeatRequestPB* req,
   }
 
   // 4. Update tserver soft state based on the heartbeat contents.
+  // TODO(mreddy) If --enable_range_replica_placement is set to false, don't populate ranges map.
   ts_desc->UpdateHeartbeatTime();
   ts_desc->set_num_live_replicas(req->num_live_tablets());
   ts_desc->set_num_live_replicas_by_dimension(
       TabletNumByDimensionMap(req->num_live_tablets_by_dimension().begin(),
                               req->num_live_tablets_by_dimension().end()));
+  for (auto it = req->num_live_tablets_by_range_per_table().begin();
+       it != req->num_live_tablets_by_range_per_table().end(); ++it) {
+    TabletNumByRangeMap ranges;
+    for (auto range = it->second.num_live_tablets_by_range().begin();
+         range != it->second.num_live_tablets_by_range().end(); ++range) {
+      ranges[range->range_start_key()] = range->tablets();
+    }
+    ts_desc->set_num_live_replicas_by_range_per_table(it->first, ranges);
+  }
 
   // 5. Only leaders handle tablet reports.
   if (is_leader_master && req->has_tablet_report()) {