[yugabyte#20118] docdb: Add load balancer metrics for blacklist and data move progress

Summary:
This diff adds the following three metrics to the load balancer:
 - `tablets_in_wrong_placement`: the number of tablets in invalid or blacklisted locations
 - `blacklisted_leaders`: the number of tablet leaders in leader blacklisted locations
 - `tablet_load_variance`: the sum across tables of the tablet load variance (difference between the number of running or starting tablets on the highest and lowest load tablet servers)
Jira: DB-9054

Test Plan:
```
ybd --cxx-test load_balancer_mini_cluster-test --gtest_filter LoadBalancerMiniClusterTest.TabletsInWrongPlacementMetric
ybd --cxx-test load_balancer_mini_cluster-test --gtest_filter LoadBalancerMiniClusterTest.BlacklistedLeadersMetric
ybd --cxx-test load_balancer_mini_cluster-test --gtest_filter LoadBalancerMiniClusterTest.TabletLoadVarianceMetric
```

Reviewers: mlillibridge

Reviewed By: mlillibridge

Subscribers: ybase, bogdan

Tags: #jenkins-ready

Differential Revision: https://phorge.dev.yugabyte.com/D30650

Author: SrivastavaAnubhav

src/yb/integration-tests/load_balancer_mini_cluster-test.cc

@@ -29,24 +29,31 @@
 
 #include "yb/tools/yb-admin_client.h"
 
+#include "yb/tablet/tablet_peer.h"
+
 #include "yb/tserver/mini_tablet_server.h"
 #include "yb/tserver/tablet_server.h"
 #include "yb/tserver/tablet_server_options.h"
 
 #include "yb/util/backoff_waiter.h"
+#include "yb/util/metrics.h"
 #include "yb/util/monotime.h"
 #include "yb/util/multi_drive_test_env.h"
 
+METRIC_DECLARE_gauge_uint32(blacklisted_leaders);
 METRIC_DECLARE_event_stats(load_balancer_duration);
+METRIC_DECLARE_gauge_uint32(tablets_in_wrong_placement);
+METRIC_DECLARE_gauge_uint32(total_table_load_difference);
 
+DECLARE_int32(catalog_manager_bg_task_wait_ms);
 DECLARE_bool(enable_load_balancing);
 DECLARE_bool(load_balancer_drive_aware);
-DECLARE_int32(catalog_manager_bg_task_wait_ms);
+DECLARE_int32(load_balancer_max_concurrent_moves);
+DECLARE_int32(replication_factor);
 DECLARE_int32(TEST_slowdown_master_async_rpc_tasks_by_ms);
 DECLARE_int32(TEST_load_balancer_wait_ms);
 DECLARE_int32(TEST_load_balancer_wait_after_count_pending_tasks_ms);
 DECLARE_bool(tserver_heartbeat_metrics_add_drive_data);
-DECLARE_int32(load_balancer_max_concurrent_moves);
 DECLARE_int32(tserver_heartbeat_metrics_interval_ms);
 
 using namespace std::literals;
@@ -238,6 +245,20 @@ class LoadBalancerMiniClusterTest : public LoadBalancerMiniClusterTestBase {
   int num_tablets() override {
     return 4;
   }
+
+  Status AddTserverToBlacklist(size_t idx, bool leader_blacklist) {
+    HostPort ts_hostport(mini_cluster()->mini_tablet_server(idx)->bound_rpc_addr());
+    RETURN_NOT_OK(
+        yb_admin_client_->ChangeBlacklist({ts_hostport}, true /* add */, leader_blacklist));
+    return Status::OK();
+  }
+
+  Status RemoveTserverFromBlacklist(size_t idx, bool leader_blacklist) {
+    HostPort ts_hostport(mini_cluster()->mini_tablet_server(idx)->bound_rpc_addr());
+    RETURN_NOT_OK(
+        yb_admin_client_->ChangeBlacklist({ts_hostport}, false /* add */, leader_blacklist));
+    return Status::OK();
+  }
 };
 
 class LoadBalancerMiniClusterRf3Test : public LoadBalancerMiniClusterTest {
@@ -271,6 +292,122 @@ TEST_F(LoadBalancerMiniClusterRf3Test, DurationMetric) {
   }, 10s, "load_balancer_duration value resets"));
 }
 
+TEST_F(LoadBalancerMiniClusterTest, TabletsInWrongPlacementMetric) {
+  const int ts_idx = 0;
+  auto* mini_master = ASSERT_RESULT(mini_cluster()->GetLeaderMiniMaster());
+  auto cluster_metric_entity = mini_master->master()->metric_entity_cluster();
+  auto tablets_in_wrong_placement =
+      cluster_metric_entity->FindOrNull<AtomicGauge<uint32_t>>(METRIC_tablets_in_wrong_placement);
+  ASSERT_EQ(tablets_in_wrong_placement->value(), 0);
+
+  // Prevent moves so we can reliably read the metric and get a non-zero value.
+  FLAGS_load_balancer_max_concurrent_adds = 0;
+
+  unsigned int peers_on_ts = 0;
+  auto ts_uuid = mini_cluster_->mini_tablet_server(ts_idx)->server()->permanent_uuid();
+  for (const auto& peer : ListTabletPeers(mini_cluster(), ListPeersFilter::kAll)) {
+    if (peer->permanent_uuid() == ts_uuid) {
+      ++peers_on_ts;
+    }
+  }
+  ASSERT_GT(peers_on_ts, 0);
+
+  ASSERT_EQ(tablets_in_wrong_placement->value(), 0);
+
+  // Blacklist first tserver.
+  ASSERT_OK(AddTserverToBlacklist(ts_idx, false /* leader_blacklist */));
+  SleepFor(FLAGS_catalog_manager_bg_task_wait_ms * 2ms);
+  ASSERT_EQ(tablets_in_wrong_placement->value(), peers_on_ts);
+
+  // Change placement info to make first tserver invalid.
+  // The invalid and blacklisted tablets should not be double-counted.
+  ASSERT_OK(yb_admin_client_->ModifyPlacementInfo("cloud1.rack2.zone,cloud2.rack3.zone", 3, ""));
+  SleepFor(FLAGS_catalog_manager_bg_task_wait_ms * 2ms);
+  ASSERT_EQ(tablets_in_wrong_placement->value(), peers_on_ts);
+
+  // Unblacklist first tserver.
+  // The tablets are still invalid, so the metric should not change.
+  ASSERT_OK(RemoveTserverFromBlacklist(ts_idx, false /* leader_blacklist */));
+  SleepFor(FLAGS_catalog_manager_bg_task_wait_ms * 2ms);
+  ASSERT_EQ(tablets_in_wrong_placement->value(), peers_on_ts);
+
+  // Change placement info to make first tserver valid again.
+  ASSERT_OK(yb_admin_client_->ModifyPlacementInfo(
+      "cloud1.rack1.zone,cloud1.rack2.zone,cloud2.rack3.zone", 3, ""));
+  ASSERT_OK(WaitFor([&] {
+    return tablets_in_wrong_placement->value() == 0;
+  }, 5s, "Wait for tablets_in_wrong_placement to be 0"));
+}
+
+TEST_F(LoadBalancerMiniClusterTest, BlacklistedLeadersMetric) {
+  auto* mini_master = ASSERT_RESULT(mini_cluster()->GetLeaderMiniMaster());
+  auto cluster_metric_entity = mini_master->master()->metric_entity_cluster();
+  auto blacklisted_leaders =
+      cluster_metric_entity->FindOrNull<AtomicGauge<uint32_t>>(METRIC_blacklisted_leaders);
+  ASSERT_EQ(blacklisted_leaders->value(), 0);
+
+  // Prevent leader moves so we can reliably read the metric and get a non-zero value.
+  FLAGS_load_balancer_max_concurrent_moves = 0;
+
+  // Leader blacklist first tserver.
+  ASSERT_OK(AddTserverToBlacklist(0 /* idx */, true /* leader_blacklist */));
+  ASSERT_OK(WaitFor([&] {
+    return blacklisted_leaders->value() > 0;
+  }, 5s, "Wait for blacklisted_leaders to reflect blacklisted leader"));
+}
+
+TEST_F(LoadBalancerMiniClusterTest, TableLoadDifferenceMetric) {
+  const auto kNumTablets = static_cast<uint32_t>(num_tablets());
+  auto* mini_master = ASSERT_RESULT(mini_cluster()->GetLeaderMiniMaster());
+  auto cluster_metric_entity = mini_master->master()->metric_entity_cluster();
+  auto load_difference_metric = cluster_metric_entity->FindOrNull<AtomicGauge<uint32_t>>(
+      METRIC_total_table_load_difference);
+
+  ASSERT_EQ(load_difference_metric->value(), 0);
+
+  // Prevent moves temporarily so we can reliably read the metric and get a non-zero value.
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_load_balancer_max_concurrent_adds) = 0;
+
+  // Add a new node and wait for load difference to equal kNumTablets (each existing node has
+  // kNumTablets tablets, and the new node has 0).
+  auto new_ts_index = mini_cluster()->num_tablet_servers();
+  ASSERT_OK(mini_cluster()->AddTabletServer());
+  ASSERT_OK(mini_cluster()->WaitForTabletServerCount(new_ts_index + 1));
+  ASSERT_OK(WaitFor([&] {
+    return load_difference_metric->value() == kNumTablets;
+  }, 5s, "load_difference reflects new node"));
+
+  // Enable moves and verify that load_difference monotonically decreases to 0.
+  auto load_difference_low_water_mark = load_difference_metric->value();
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_load_balancer_max_concurrent_adds) = 1;
+  ASSERT_OK(WaitFor([&]() -> Result<bool> {
+    auto load_difference = load_difference_metric->value();
+    if (load_difference > load_difference_low_water_mark) {
+      return STATUS_FORMAT(
+          IllegalState, "load_difference unexpectedly increased from $0 to $1",
+          load_difference_low_water_mark, load_difference);
+    }
+    load_difference_low_water_mark = load_difference;
+    return load_difference == 0 && VERIFY_RESULT(client_->IsLoadBalancerIdle());
+  }, 30s, "Wait for tablet moves after adding ts-3"));
+
+  // Blacklist first tserver.
+  // The replicas should move off the blacklisted tserver and load_difference should increase to
+  // kNumTablets again.
+  auto load_difference_high_water_mark = load_difference_metric->value();
+  ASSERT_OK(AddTserverToBlacklist(0 /* idx */, false /* leader_blacklist */));
+  ASSERT_OK(WaitFor([&]() -> Result<bool> {
+    auto load_difference = load_difference_metric->value();
+    if (load_difference < load_difference_high_water_mark) {
+      return STATUS_FORMAT(
+          IllegalState, "load_difference unexpectedly decreased from $0 to $1",
+          load_difference_high_water_mark, load_difference);
+    }
+    load_difference_high_water_mark = load_difference;
+    return load_difference == kNumTablets && VERIFY_RESULT(client_->IsLoadBalancerIdle());
+  }, 30s, "Wait for tablet moves after blacklisting ts-0"));
+}
+
 // See issue #6278. This test tests the segfault that used to occur during a rare race condition,
 // where we would have an uninitialized TSDescriptor that we try to access.
 // To trigger the race condition, we need a pending add task that gets completed after

src/yb/master/cluster_balance.cc

@@ -134,6 +134,33 @@ METRIC_DEFINE_gauge_int64(cluster,
                           "Is load balancing enabled in the cluster where "
                           "1 indicates it is enabled.");
 
+METRIC_DEFINE_gauge_uint32(cluster,
+                           tablets_in_wrong_placement,
+                           "Tablets in Wrong/Blacklisted Placement",
+                           yb::MetricUnit::kUnits,
+                           "Number of tablet peers in invalid or blacklisted locations.");
+
+METRIC_DEFINE_gauge_uint32(cluster,
+                           blacklisted_leaders,
+                           "Blacklisted Leaders",
+                           yb::MetricUnit::kUnits,
+                           "Number of tablet leaders in locations from which leaders are "
+                           "blacklisted.");
+
+METRIC_DEFINE_gauge_uint32(cluster,
+                           total_table_load_difference,
+                           "Sum of Table Load Difference",
+                           yb::MetricUnit::kUnits,
+                           "This metric is the sum of every table's load difference, where a "
+                           "table's load difference is the maximum load difference for that table "
+                           "between all pairs of TServers that are valid hosts for its tablets. "
+                           "Here, the load difference for a table between a pair of TServers is "
+                           "the positive difference between the number of running/starting tablet "
+                           "peers belonging to that table hosted on those TServers. Exception: "
+                           "the load difference is defined as 0 if it otherwise would be 1 because "
+                           "the load balancer cannot fix a load difference of 1 by moving a tablet "
+                           "peer from one TServer to another.");
+
 namespace yb {
 namespace master {
 
@@ -255,6 +282,10 @@ size_t ClusterLoadBalancer::get_total_wrong_placement() const {
   return state_->tablets_wrong_placement_.size();
 }
 
+size_t ClusterLoadBalancer::get_badly_placed_leaders() const {
+  return state_->tablets_with_badly_placed_leaders_.size();
+}
+
 size_t ClusterLoadBalancer::get_total_blacklisted_servers() const {
   return global_state_->blacklisted_servers_.size();
 }
@@ -312,6 +343,12 @@ ClusterLoadBalancer::~ClusterLoadBalancer() = default;
 void ClusterLoadBalancer::InitMetrics() {
   is_load_balancing_enabled_metric_ = METRIC_is_load_balancing_enabled.Instantiate(
       catalog_manager_->master_->metric_entity_cluster(), 0);
+  tablets_in_wrong_placement_metric_ = METRIC_tablets_in_wrong_placement.Instantiate(
+      catalog_manager_->master_->metric_entity_cluster(), 0);
+  blacklisted_leaders_metric_ = METRIC_blacklisted_leaders.Instantiate(
+      catalog_manager_->master_->metric_entity_cluster(), 0);
+  total_table_load_difference_metric_ = METRIC_total_table_load_difference.Instantiate(
+      catalog_manager_->master_->metric_entity_cluster(), 0);
 }
 
 // This function uses the following stratification of vlog levels:
@@ -452,6 +489,10 @@ void ClusterLoadBalancer::RunLoadBalancerWithOptions(Options* options) {
   // At the start of the run, report LB state that might prevent it from running smoothly.
   ReportUnusualLoadBalancerState();
 
+  uint32_t total_tablets_in_wrong_placement = 0;
+  uint32_t total_blacklisted_leaders = 0;
+  uint32_t total_table_load_difference = 0;
+
   // Loop over all tables to analyze the global and per-table load.
   for (const auto& table : GetTables()) {
     if (SkipLoadBalancing(*table)) {
@@ -474,8 +515,29 @@ void ClusterLoadBalancer::RunLoadBalancerWithOptions(Options* options) {
       per_table_states_.erase(table->id());
       master_errors++;
     }
+
+    if (!state_->sorted_load_.empty()) {
+      // Only report values greater than 1 since the LB cannot fix a load difference of 1.
+      // This makes the total table load difference metric more interpretable since we would
+      // otherwise add 1 for every table has a difference of 1 between its most/least loaded
+      // tservers, even though the table load is balanced.
+      const auto& low_load_uuid = state_->sorted_load_.front();
+      const auto& high_load_uuid = state_->sorted_load_.back();
+      const int actual_load_difference =
+          narrow_cast<int>(state_->GetLoad(high_load_uuid) - state_->GetLoad(low_load_uuid));
+      if (actual_load_difference > 1) {
+        total_table_load_difference += actual_load_difference;
+      }
+    }
+    total_tablets_in_wrong_placement += get_total_wrong_placement();
+    total_blacklisted_leaders += get_badly_placed_leaders();
   }
 
+  // Update metrics.
+  tablets_in_wrong_placement_metric_->set_value(total_tablets_in_wrong_placement);
+  blacklisted_leaders_metric_->set_value(total_blacklisted_leaders);
+  total_table_load_difference_metric_->set_value(total_table_load_difference);
+
   VLOG(1) << "Global state after analyzing all tablets: " << global_state_->ToString();
 
   bool task_added = false;

src/yb/master/cluster_balance.h

@@ -348,6 +348,7 @@ class ClusterLoadBalancer {
   int get_total_running_tablets() const;
 
   size_t get_total_wrong_placement() const;
+  size_t get_badly_placed_leaders() const;
   size_t get_total_blacklisted_servers() const;
   size_t get_total_leader_blacklisted_servers() const;
 
@@ -362,8 +363,10 @@ class ClusterLoadBalancer {
   // managed by this class, but by the Master's unique_ptr.
   CatalogManager* catalog_manager_;
 
-  // Info about if load balancing is enabled in the cluster.
   scoped_refptr<AtomicGauge<int64_t>> is_load_balancing_enabled_metric_;
+  scoped_refptr<AtomicGauge<uint32_t>> tablets_in_wrong_placement_metric_;
+  scoped_refptr<AtomicGauge<uint32_t>> blacklisted_leaders_metric_;
+  scoped_refptr<AtomicGauge<uint32_t>> total_table_load_difference_metric_;
 
   std::shared_ptr<YsqlTablespaceManager> tablespace_manager_;
 

src/yb/master/cluster_balance_util.cc

@@ -380,6 +380,9 @@ Status PerTableLoadState::UpdateTablet(TabletInfo *tablet) {
   if (tablet_meta.has_wrong_placements()) {
     tablets_wrong_placement_.insert(tablet_id);
   }
+  if (tablet_meta.has_badly_placed_leader()) {
+    tablets_with_badly_placed_leaders_.insert(tablet_id);
+  }
 
   return Status::OK();
 }

src/yb/master/cluster_balance_util.h

@@ -57,7 +57,7 @@ struct CBTabletMetadata {
     return !wrong_placement_tablet_servers.empty() || !blacklisted_tablet_servers.empty();
   }
 
-  bool has_blacklisted_leader() {
+  bool has_badly_placed_leader() {
     return !leader_blacklisted_tablet_servers.empty();
   }
 
@@ -427,7 +427,8 @@ class PerTableLoadState {
     out << Format("tablets_missing_replicas: $0, ", tablets_missing_replicas_);
     out << Format("tablets_over_replicated: $0, ", tablets_over_replicated_);
     out << Format("tablets_wrong_placement: $0, ", tablets_wrong_placement_);
-    out << Format("tablets_added: $0, ", tablets_wrong_placement_);
+    out << Format("tablets_with_badly_placed_leaders: $0, ", tablets_with_badly_placed_leaders_);
+    out << Format("tablets_added: $0, ", tablets_added_);
     out << Format("leader_balance_threshold: $0, ", leader_balance_threshold_);
     out << Format("sorted_leader_load: $0, ", sorted_leader_load_);
     out << Format("use_preferred_zones: $0, ", use_preferred_zones_);
@@ -482,6 +483,9 @@ class PerTableLoadState {
   // Set of tablet ids that have been determined to have replicas in incorrect placements.
   std::set<TabletId> tablets_wrong_placement_;
 
+  // Set of tablet ids that have leaders in leader blacklisted placements.
+  std::set<TabletId> tablets_with_badly_placed_leaders_;
+
   // List of tablet ids that have been added to a new tablet server.
   std::set<TabletId> tablets_added_;
 

src/yb/util/metric_entity.cc

@@ -176,11 +176,6 @@ void MetricEntity::CheckInstantiation(const MetricPrototype* proto) const {
       << "Metric name is not compatible with Prometheus: " << proto->name();
 }
 
-scoped_refptr<Metric> MetricEntity::FindOrNull(const MetricPrototype& prototype) const {
-  std::lock_guard l(lock_);
-  return FindPtrOrNull(metric_map_, &prototype);
-}
-
 bool MetricEntity::TEST_ContainMetricName(const std::string& metric_name) const {
   std::lock_guard l(lock_);
   for (const MetricMap::value_type& val : metric_map_) {

src/yb/util/metric_entity.h

@@ -149,6 +149,7 @@ class MetricEntity : public RefCountedThreadSafe<MetricEntity> {
 
   // Return the metric instantiated from the given prototype, or NULL if none has been
   // instantiated. Primarily used by tests trying to read metric values.
+  template<typename Metric>
   scoped_refptr<Metric> FindOrNull(const MetricPrototype& prototype) const;
 
   const std::string& id() const { return id_; }
@@ -248,6 +249,12 @@ scoped_refptr<Metric> MetricEntity::FindOrCreateMetric(PrototypePtr proto, Args&
   return m;
 }
 
+template<typename Metric>
+scoped_refptr<Metric> MetricEntity::FindOrNull(const MetricPrototype& prototype) const {
+  std::lock_guard l(lock_);
+  return down_cast<Metric*>(FindPtrOrNull(metric_map_, &prototype).get());
+}
+
 void WriteRegistryAsJson(JsonWriter* writer);
 
 } // namespace yb