refactor(collector): sort out the structure of partition hotspot detection

src/server/hotspot_partition_calculator.cpp

@@ -0,0 +1,53 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "hotspot_partition_calculator.h"
+
+#include <dsn/dist/fmt_logging.h>
+
+namespace pegasus {
+namespace server {
+
+void hotspot_partition_calculator::aggregate(const std::vector<row_data> &partitions)
+{
+    while (_app_data.size() > kMaxQueueSize - 1) {
+        _app_data.pop();
+    }
+    std::vector<hotspot_partition_data> temp(partitions.size());
+    for (int i = 0; i < partitions.size(); i++) {
+        temp[i] = std::move(hotspot_partition_data(partitions[i]));
+    }
+    _app_data.emplace(temp);
+}
+
+void hotspot_partition_calculator::init_perf_counter(const int perf_counter_count)
+{
+    std::string counter_name;
+    std::string counter_desc;
+    for (int i = 0; i < perf_counter_count; i++) {
+        string paritition_desc = _app_name + '.' + std::to_string(i);
+        counter_name = fmt::format("app.stat.hotspots@{}", paritition_desc);
+        counter_desc = fmt::format("statistic the hotspots of app {}", paritition_desc);
+        _points[i].init_app_counter(
+            "app.pegasus", counter_name.c_str(), COUNTER_TYPE_NUMBER, counter_desc.c_str());
+    }
+}
+
+void hotspot_partition_calculator::start_alg() { _policy->analysis(_app_data, _points); }
+
+} // namespace server
+} // namespace pegasus

src/server/hotspot_partition_calculator.h

@@ -0,0 +1,58 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include "hotspot_partition_data.h"
+
+#include <algorithm>
+#include <gtest/gtest_prod.h>
+#include <math.h>
+
+#include <dsn/perf_counter/perf_counter.h>
+#include "hotspot_partition_policy.h"
+
+namespace pegasus {
+namespace server {
+
+// hotspot_partition_calculator is used to find the hotspot in Pegasus
+class hotspot_partition_calculator
+{
+public:
+    hotspot_partition_calculator(const std::string &app_name,
+                                 const int partition_num,
+                                 std::unique_ptr<hotspot_partition_policy> policy)
+        : _app_name(app_name), _points(partition_num), _policy(std::move(policy))
+    {
+        init_perf_counter(partition_num);
+    }
+    void aggregate(const std::vector<row_data> &partitions);
+    void start_alg();
+    void init_perf_counter(const int perf_counter_count);
+
+private:
+    const std::string _app_name;
+    std::vector<::dsn::perf_counter_wrapper> _points;
+    std::queue<std::vector<hotspot_partition_data>> _app_data;
+    std::unique_ptr<hotspot_partition_policy> _policy;
+    static const int kMaxQueueSize = 100;
+
+    FRIEND_TEST(hotspot_partition_calculator, hotspot_partition_policy);
+};
+
+} // namespace server
+} // namespace pegasus

src/server/hotspot_partition_policy.cpp

@@ -0,0 +1,67 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "hotspot_partition_policy.h"
+
+namespace pegasus {
+namespace server {
+
+void hotspot_partition_policy::analysis(
+    const std::queue<std::vector<hotspot_partition_data>> &hotspot_app_data,
+    std::vector<::dsn::perf_counter_wrapper> &perf_counters)
+{
+    dassert(hotspot_app_data.back().size() == perf_counters.size(),
+            "partition counts error, please check");
+    std::vector<double> data_samples;
+    data_samples.reserve(hotspot_app_data.size() * perf_counters.size());
+    auto temp_data = hotspot_app_data;
+    double total = 0, sd = 0, avg = 0;
+    int sample_count = 0;
+    // avg: Average number
+    // sd: Standard deviation
+    // sample_count: Number of samples
+    while (!temp_data.empty()) {
+        for (auto partition_data : temp_data.front()) {
+            if (partition_data.total_qps - 1.00 > 0) {
+                data_samples.push_back(partition_data.total_qps);
+                total += partition_data.total_qps;
+                sample_count++;
+            }
+        }
+        temp_data.pop();
+    }
+    if (sample_count == 0) {
+        ddebug("hotspot_app_data size == 0");
+        return;
+    }
+    avg = total / sample_count;
+    for (auto data_sample : data_samples) {
+        sd += pow((data_sample - avg), 2);
+    }
+    sd = sqrt(sd / sample_count);
+    const auto &anly_data = hotspot_app_data.back();
+    for (int i = 0; i < perf_counters.size(); i++) {
+        double hot_point = (anly_data[i].total_qps - avg) / sd;
+        // perf_counter->set can only be unsigned __int64
+        // use ceil to guarantee conversion results
+        hot_point = ceil(std::max(hot_point, double(0)));
+        perf_counters[i]->set(hot_point);
+    }
+}
+
+} // namespace server
+} // namespace pegasus

src/server/hotspot_partition_policy.h

@@ -0,0 +1,39 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include "hotspot_partition_data.h"
+
+#include <algorithm>
+#include <gtest/gtest_prod.h>
+#include <math.h>
+
+#include <dsn/perf_counter/perf_counter.h>
+
+namespace pegasus {
+namespace server {
+// PauTa Criterion
+class hotspot_partition_policy
+{
+public:
+    void analysis(const std::queue<std::vector<hotspot_partition_data>> &hotspot_app_data,
+                  std::vector<::dsn::perf_counter_wrapper> &perf_counters);
+};
+
+} // namespace server
+} // namespace pegasus

src/server/info_collector.cpp

@@ -78,10 +78,6 @@ info_collector::info_collector()
                                               "storage_size_fetch_interval_seconds",
                                               3600, // default value 1h
                                               "storage size fetch interval seconds");
-    _hotspot_detect_algorithm = dsn_config_get_value_string("pegasus.collector",
-                                                            "hotspot_detect_algorithm",
-                                                            "hotspot_algo_qps_variance",
-                                                            "hotspot_detect_algorithm");
     // _storage_size_retry_wait_seconds is in range of [1, 60]
     _storage_size_retry_wait_seconds =
         std::min(60u, std::max(1u, _storage_size_fetch_interval_seconds / 10));
@@ -150,15 +146,15 @@ void info_collector::on_app_stat()
         // get row data statistics for all of the apps
         all_stats.merge(app_stats);
 
-        // hotspot_calculator is to detect hotspots
-        hotspot_calculator *hotspot_calculator =
+        // hotspot_partition_calculator is to detect hotspots
+        hotspot_partition_calculator *hotspot_partition_calculator =
             get_hotspot_calculator(app_rows.first, app_rows.second.size());
-        if (!hotspot_calculator) {
+        if (!hotspot_partition_calculator) {
             continue;
         }
-        hotspot_calculator->aggregate(app_rows.second);
+        hotspot_partition_calculator->aggregate(app_rows.second);
         // new policy can be designed by strategy pattern in hotspot_partition_data.h
-        hotspot_calculator->start_alg();
+        hotspot_partition_calculator->start_alg();
     }
     get_app_counters(all_stats.app_name)->set(all_stats);
 
@@ -302,25 +298,17 @@ void info_collector::on_storage_size_stat(int remaining_retry_count)
     _result_writer->set_result(st_stat.timestamp, "ss", st_stat.dump_to_json());
 }
 
-hotspot_calculator *info_collector::get_hotspot_calculator(const std::string &app_name,
-                                                           const int partition_num)
+hotspot_partition_calculator *info_collector::get_hotspot_calculator(const std::string &app_name,
+                                                                     const int partition_num)
 {
     // use appname+partition_num as a key can prevent the impact of dynamic partition changes
     std::string app_name_pcount = fmt::format("{}.{}", app_name, partition_num);
     auto iter = _hotspot_calculator_store.find(app_name_pcount);
     if (iter != _hotspot_calculator_store.end()) {
         return iter->second;
     }
-    std::unique_ptr<hotspot_policy> policy;
-    if (_hotspot_detect_algorithm == "hotspot_algo_qps_variance") {
-        policy.reset(new hotspot_algo_qps_variance());
-    } else {
-        dwarn("hotspot detection is disabled");
-        _hotspot_calculator_store[app_name_pcount] = nullptr;
-        return nullptr;
-    }
-    hotspot_calculator *calculator =
-        new hotspot_calculator(app_name, partition_num, std::move(policy));
+    hotspot_partition_calculator *calculator = new hotspot_partition_calculator(
+        app_name, partition_num, std::make_unique<hotspot_partition_policy>());
     _hotspot_calculator_store[app_name_pcount] = calculator;
     return calculator;
 }

src/server/info_collector.h

@@ -19,7 +19,8 @@
 
 #include "../shell/commands.h"
 #include "table_stats.h"
-#include "table_hotspot_policy.h"
+#include "hotspot_partition_calculator.h"
+#include "hotspot_partition_policy.h"
 
 namespace pegasus {
 namespace server {
@@ -177,15 +178,14 @@ class info_collector
     uint32_t _storage_size_fetch_interval_seconds;
     uint32_t _storage_size_retry_wait_seconds;
     uint32_t _storage_size_retry_max_count;
-    std::string _hotspot_detect_algorithm;
     ::dsn::task_ptr _storage_size_stat_timer_task;
     ::dsn::utils::ex_lock_nr _capacity_unit_update_info_lock;
     // mapping 'node address' --> 'last updated timestamp'
     std::map<std::string, string> _capacity_unit_update_info;
-    std::map<std::string, hotspot_calculator *> _hotspot_calculator_store;
+    std::map<std::string, hotspot_partition_calculator *> _hotspot_calculator_store;
 
-    hotspot_calculator *get_hotspot_calculator(const std::string &app_name,
-                                               const int partition_num);
+    hotspot_partition_calculator *get_hotspot_calculator(const std::string &app_name,
+                                                         const int partition_num);
 };
 
 } // namespace server