feat(hotspot): add a cluster level perf_counter to display hotspot

src/server/hotspot_partition_calculator.cpp

@@ -82,10 +82,18 @@ void hotspot_partition_calculator::init_perf_counter(int partition_count)
             _hot_points[i][data_type].init_app_counter(
                 "app.pegasus", counter_name.c_str(), COUNTER_TYPE_NUMBER, counter_desc.c_str());
         }
+
+        string total_desc =
+            _app_name + '.' +
+            (data_type == partition_qps_type::WRITE_HOTSPOT_DATA ? "write.total" : "read.total");
+        std::string counter_name = fmt::format("app.stat.hotspots.{}", total_desc);
+        std::string counter_desc = fmt::format("statistic the hotspots of app {}", total_desc);
+        _total_hotspot_cnt[data_type].init_app_counter(
+            "app.pegasus", counter_name.c_str(), COUNTER_TYPE_NUMBER, counter_desc.c_str());
     }
 }
 
-void hotspot_partition_calculator::stat_histories_analyse(int data_type,
+void hotspot_partition_calculator::stat_histories_analyse(uint32_t data_type,
                                                           std::vector<int> &hot_points)
 {
     double table_qps_sum = 0, standard_deviation = 0, table_qps_avg = 0;
@@ -121,13 +129,19 @@ void hotspot_partition_calculator::stat_histories_analyse(int data_type,
     }
 }
 
-void hotspot_partition_calculator::update_hot_point(int data_type, std::vector<int> &hot_points)
+void hotspot_partition_calculator::update_hot_point(uint32_t data_type,
+                                                    const std::vector<int> &hot_points)
 {
     dcheck_eq(_hot_points.size(), hot_points.size());
     int size = hot_points.size();
+    uint32_t hotspot_count = 0;
     for (int i = 0; i < size; i++) {
         _hot_points[i][data_type].get()->set(hot_points[i]);
+        if (hot_points[i] >= FLAGS_hot_partition_threshold) {
+            hotspot_count++;
+        }
     }
+    _total_hotspot_cnt[data_type].get()->set(hotspot_count);
 }
 
 void hotspot_partition_calculator::data_analyse()
@@ -136,7 +150,8 @@ void hotspot_partition_calculator::data_analyse()
             "The number of partitions in this table has changed, and hotspot analysis cannot be "
             "performed,in %s",
             _app_name.c_str());
-    for (int data_type = 0; data_type <= 1; data_type++) {
+
+    for (uint32_t data_type = 0; data_type <= 1; data_type++) {
         // data_type 0: READ_HOTSPOT_DATA; 1: WRITE_HOTSPOT_DATA
         std::vector<int> hot_points;
         stat_histories_analyse(data_type, hot_points);

src/server/hotspot_partition_calculator.h

@@ -59,15 +59,19 @@ class hotspot_partition_calculator
 private:
     // empirical rule to calculate hot point of each partition
     // ref: https://en.wikipedia.org/wiki/68%E2%80%9395%E2%80%9399.7_rule
-    void stat_histories_analyse(int data_type, std::vector<int> &hot_points);
+    void stat_histories_analyse(uint32_t data_type, std::vector<int> &hot_points);
     // set hot_point to corresponding perf_counter
-    void update_hot_point(int data_type, std::vector<int> &hot_points);
+    void update_hot_point(uint32_t data_type, std::vector<int> &hot_points);
     void detect_hotkey_in_hotpartition(int data_type);
 
     const std::string _app_name;
     void init_perf_counter(int perf_counter_count);
     // usually a partition with "hot-point value" >= 3 can be considered as a hotspot partition.
     hot_partition_counters _hot_points;
+    // hotspot_cnt c[type_of_read(0)/write(1)_stat] = number of hot partition count in one table
+    // per data_analyse
+    std::array<dsn::perf_counter_wrapper, 2> _total_hotspot_cnt;
+
     // saving historical data can improve accuracy
     stat_histories _partitions_stat_histories;
     std::shared_ptr<shell_context> _shell_context;

src/server/test/hotspot_partition_test.cpp

@@ -72,13 +72,26 @@ class hotspot_partition_test : public pegasus_server_test_base
         return result;
     }
 
+    std::array<uint32_t, 2>
+    get_calculator_total_hotspot_cnt(const std::array<dsn::perf_counter_wrapper, 2> &cnts)
+    {
+        std::array<uint32_t, 2> result;
+        result[READ_HOTSPOT_DATA] = cnts[READ_HOTSPOT_DATA].get()->get_value();
+        result[WRITE_HOTSPOT_DATA] = cnts[WRITE_HOTSPOT_DATA].get()->get_value();
+        return result;
+    }
+
     void test_policy_in_scenarios(std::vector<row_data> scenario,
-                                  std::vector<std::vector<double>> &expect_result)
+                                  std::vector<std::vector<double>> &expect_result,
+                                  std::array<uint32_t, 2> expect_cnt)
     {
         calculator.data_aggregate(std::move(scenario));
         calculator.data_analyse();
         std::vector<std::vector<double>> result = get_calculator_result(calculator._hot_points);
+        auto cnt = get_calculator_total_hotspot_cnt(calculator._total_hotspot_cnt);
+
         ASSERT_EQ(result, expect_result);
+        ASSERT_EQ(cnt, expect_cnt);
     }
 
     void aggregate_analyse_data(std::vector<row_data> scenario,
@@ -101,7 +114,9 @@ TEST_F(hotspot_partition_test, hotspot_partition_policy)
     std::vector<row_data> test_rows = generate_row_data();
     std::vector<std::vector<double>> expect_vector = {{0, 0, 0, 0, 0, 0, 0, 0},
                                                       {0, 0, 0, 0, 0, 0, 0, 0}};
-    test_policy_in_scenarios(test_rows, expect_vector);
+
+    std::array<uint32_t, 2> expect_hotspot_cnt = {0, 0};
+    test_policy_in_scenarios(test_rows, expect_vector, expect_hotspot_cnt);
 
     // Insert hotspot scenario_0 data to test
     test_rows = generate_row_data();
@@ -110,14 +125,16 @@ TEST_F(hotspot_partition_test, hotspot_partition_policy)
     test_rows[HOT_SCENARIO_0_READ_HOT_PARTITION].get_qps = 5000.0;
     test_rows[HOT_SCENARIO_0_WRITE_HOT_PARTITION].put_qps = 5000.0;
     expect_vector = {{0, 0, 0, 0, 0, 0, 0, 4}, {4, 0, 0, 0, 0, 0, 0, 0}};
-    test_policy_in_scenarios(test_rows, expect_vector);
+    expect_hotspot_cnt = {1, 1};
+    test_policy_in_scenarios(test_rows, expect_vector, expect_hotspot_cnt);
 
     // Insert hotspot scenario_0 data to test again
     test_rows = generate_row_data();
     test_rows[HOT_SCENARIO_0_READ_HOT_PARTITION].get_qps = 5000.0;
     test_rows[HOT_SCENARIO_0_WRITE_HOT_PARTITION].put_qps = 5000.0;
     expect_vector = {{0, 0, 0, 0, 0, 0, 0, 4}, {4, 0, 0, 0, 0, 0, 0, 0}};
-    test_policy_in_scenarios(test_rows, expect_vector);
+    expect_hotspot_cnt = {1, 1};
+    test_policy_in_scenarios(test_rows, expect_vector, expect_hotspot_cnt);
 
     // Insert hotspot scenario_1 data to test again
     test_rows = generate_row_data();
@@ -126,7 +143,21 @@ TEST_F(hotspot_partition_test, hotspot_partition_policy)
     test_rows[HOT_SCENARIO_1_READ_HOT_PARTITION].get_qps = 5000.0;
     test_rows[HOT_SCENARIO_1_WRITE_HOT_PARTITION].put_qps = 5000.0;
     expect_vector = {{0, 0, 0, 4, 0, 0, 0, 0}, {0, 0, 4, 0, 0, 0, 0, 0}};
-    test_policy_in_scenarios(test_rows, expect_vector);
+    expect_hotspot_cnt = {1, 1};
+    test_policy_in_scenarios(test_rows, expect_vector, expect_hotspot_cnt);
+
+    test_rows = generate_row_data();
+    const int HOT_SCENARIO_2_READ_HOT_PARTITION_0 = 3;
+    const int HOT_SCENARIO_2_READ_HOT_PARTITION_1 = 5;
+    const int HOT_SCENARIO_2_WRITE_HOT_PARTITION = 2;
+
+    test_rows[HOT_SCENARIO_2_READ_HOT_PARTITION_0].get_qps = 7000.0;
+    test_rows[HOT_SCENARIO_2_READ_HOT_PARTITION_1].get_qps = 8000.0;
+    test_rows[HOT_SCENARIO_2_WRITE_HOT_PARTITION].put_qps = 7000.0;
+
+    expect_vector = {{0, 0, 0, 4, 0, 4, 0, 0}, {0, 0, 4, 0, 0, 0, 0, 0}};
+    expect_hotspot_cnt = {2, 1};
+    test_policy_in_scenarios(test_rows, expect_vector, expect_hotspot_cnt);
     clear_calculator_histories();
 }