[feat](metrics) Unify metrics of thread pool

be/src/runtime/exec_env.cpp

@@ -38,7 +38,9 @@
 
 namespace doris {
 
-ExecEnv::ExecEnv() = default;
+ExecEnv::ExecEnv() {
+    _doris_metrics = std::make_unique<DorisMetrics>();
+}
 
 ExecEnv::~ExecEnv() {
     destroy();

be/src/runtime/exec_env.h

@@ -45,6 +45,7 @@ class MemoryPool;
 }
 
 namespace doris {
+class DorisMetrics;
 namespace vectorized {
 class VDataStreamMgr;
 class ScannerScheduler;
@@ -349,6 +350,8 @@ class ExecEnv {
 
     bool check_auth_token(const std::string& auth_token);
 
+    DorisMetrics* doris_metrics() { return _doris_metrics.get(); }
+
 private:
     ExecEnv();
 
@@ -360,9 +363,6 @@ class ExecEnv {
     Status _init_mem_env();
     Status _check_deploy_mode();
 
-    void _register_metrics();
-    void _deregister_metrics();
-
     inline static std::atomic_bool _s_ready {false};
     inline static std::atomic_bool _s_tracking_memory {false};
     std::vector<StorePath> _store_paths;
@@ -488,6 +488,8 @@ class ExecEnv {
 
     orc::MemoryPool* _orc_memory_pool = nullptr;
     arrow::MemoryPool* _arrow_memory_pool = nullptr;
+
+    std::unique_ptr<DorisMetrics> _doris_metrics;
 };
 
 template <>

be/src/runtime/exec_env_init.cpp

@@ -139,10 +139,6 @@ namespace doris {
 class PBackendService_Stub;
 class PFunctionService_Stub;
 
-DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(scanner_thread_pool_queue_size, MetricUnit::NOUNIT);
-DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(send_batch_thread_pool_thread_num, MetricUnit::NOUNIT);
-DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(send_batch_thread_pool_queue_size, MetricUnit::NOUNIT);
-
 static void init_doris_metrics(const std::vector<StorePath>& store_paths) {
     bool init_system_metrics = config::enable_system_metrics;
     std::set<std::string> disk_devices;
@@ -336,7 +332,6 @@ Status ExecEnv::_init(const std::vector<StorePath>& store_paths,
     RETURN_IF_ERROR(_load_channel_mgr->init(MemInfo::mem_limit()));
     RETURN_IF_ERROR(_wal_manager->init());
     _heartbeat_flags = new HeartbeatFlags();
-    _register_metrics();
 
     _tablet_schema_cache =
             TabletSchemaCache::create_global_schema_cache(config::tablet_schema_cache_capacity);
@@ -668,20 +663,6 @@ Status ExecEnv::_check_deploy_mode() {
     return Status::OK();
 }
 
-void ExecEnv::_register_metrics() {
-    REGISTER_HOOK_METRIC(send_batch_thread_pool_thread_num,
-                         [this]() { return _send_batch_thread_pool->num_threads(); });
-
-    REGISTER_HOOK_METRIC(send_batch_thread_pool_queue_size,
-                         [this]() { return _send_batch_thread_pool->get_queue_size(); });
-}
-
-void ExecEnv::_deregister_metrics() {
-    DEREGISTER_HOOK_METRIC(scanner_thread_pool_queue_size);
-    DEREGISTER_HOOK_METRIC(send_batch_thread_pool_thread_num);
-    DEREGISTER_HOOK_METRIC(send_batch_thread_pool_queue_size);
-}
-
 // TODO(zhiqiang): Need refactor all thread pool. Each thread pool must have a Stop method.
 // We need to stop all threads before releasing resource.
 void ExecEnv::destroy() {
@@ -736,7 +717,6 @@ void ExecEnv::destroy() {
     SAFE_SHUTDOWN(_s3_file_system_thread_pool);
     SAFE_SHUTDOWN(_send_batch_thread_pool);
 
-    _deregister_metrics();
     SAFE_DELETE(_load_channel_mgr);
 
     SAFE_DELETE(_spill_stream_mgr);
@@ -825,6 +805,9 @@ void ExecEnv::destroy() {
     SAFE_DELETE(_process_profile);
     SAFE_DELETE(_heap_profiler);
 
+    // DorisMetrics should be destoried after all other thread stopped.
+    _doris_metrics.reset(nullptr);
+
     _s_tracking_memory = false;
 
     LOG(INFO) << "Doris exec envorinment is destoried.";

be/src/runtime/fragment_mgr.cpp

@@ -97,8 +97,7 @@ namespace doris {
 
 DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(fragment_instance_count, MetricUnit::NOUNIT);
 DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(timeout_canceled_fragment_count, MetricUnit::NOUNIT);
-DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(fragment_thread_pool_queue_size, MetricUnit::NOUNIT);
-DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(fragment_thread_pool_num_active_threads, MetricUnit::NOUNIT);
+
 bvar::LatencyRecorder g_fragmentmgr_prepare_latency("doris_FragmentMgr", "prepare");
 
 bvar::Adder<uint64_t> g_fragment_executing_count("fragment_executing_count");
@@ -243,20 +242,13 @@ FragmentMgr::FragmentMgr(ExecEnv* exec_env)
                 .set_max_threads(config::fragment_mgr_asynic_work_pool_thread_num_max)
                 .set_max_queue_size(config::fragment_mgr_asynic_work_pool_queue_size)
                 .build(&_thread_pool);
-
-    REGISTER_HOOK_METRIC(fragment_thread_pool_queue_size,
-                         [this]() { return _thread_pool->get_queue_size(); });
-    REGISTER_HOOK_METRIC(fragment_thread_pool_num_active_threads,
-                         [this]() { return _thread_pool->num_active_threads(); });
     CHECK(s.ok()) << s.to_string();
 }
 
 FragmentMgr::~FragmentMgr() = default;
 
 void FragmentMgr::stop() {
     DEREGISTER_HOOK_METRIC(fragment_instance_count);
-    DEREGISTER_HOOK_METRIC(fragment_thread_pool_queue_size);
-    DEREGISTER_HOOK_METRIC(fragment_thread_pool_num_active_threads);
     _stop_background_threads_latch.count_down();
     if (_cancel_thread) {
         _cancel_thread->join();

be/src/runtime/workload_group/workload_group.cpp

@@ -487,7 +487,7 @@ void WorkloadGroup::upsert_thread_pool_no_lock(WorkloadGroupInfo* wg_info,
     if (_scan_task_sched == nullptr) {
         std::unique_ptr<vectorized::SimplifiedScanScheduler> scan_scheduler =
                 std::make_unique<vectorized::SimplifiedScanScheduler>("Scan_" + wg_name,
-                                                                      cg_cpu_ctl_ptr);
+                                                                      cg_cpu_ctl_ptr, wg_name);
         Status ret = scan_scheduler->start(config::doris_scanner_thread_pool_thread_num,
                                            config::doris_scanner_thread_pool_thread_num,
                                            config::doris_scanner_thread_pool_queue_size);
@@ -507,7 +507,7 @@ void WorkloadGroup::upsert_thread_pool_no_lock(WorkloadGroupInfo* wg_info,
                 vectorized::ScannerScheduler::get_remote_scan_thread_queue_size();
         std::unique_ptr<vectorized::SimplifiedScanScheduler> remote_scan_scheduler =
                 std::make_unique<vectorized::SimplifiedScanScheduler>("RScan_" + wg_name,
-                                                                      cg_cpu_ctl_ptr);
+                                                                      cg_cpu_ctl_ptr, wg_name);
         Status ret = remote_scan_scheduler->start(remote_max_thread_num,
                                                   config::doris_scanner_min_thread_pool_thread_num,
                                                   remote_scan_thread_queue_size);

be/src/service/doris_main.cpp

@@ -425,6 +425,11 @@ int main(int argc, char** argv) {
                    << doris::config::spill_storage_root_path;
         exit(-1);
     }
+
+    // ExecEnv must be created before calleing doris::check_datapath_rw
+    // because will create metrics, it need an initialized DorisMetrics.
+    doris::ExecEnv* exec_env(doris::ExecEnv::GetInstance());
+
     std::set<std::string> broken_paths;
     doris::parse_conf_broken_store_paths(doris::config::broken_storage_path, &broken_paths);
 
@@ -526,7 +531,6 @@ int main(int argc, char** argv) {
     doris::ThreadLocalHandle::create_thread_local_if_not_exits();
 
     // init exec env
-    auto* exec_env(doris::ExecEnv::GetInstance());
     status = doris::ExecEnv::init(doris::ExecEnv::GetInstance(), paths, spill_paths, broken_paths);
     if (status != Status::OK()) {
         std::cerr << "failed to init doris storage engine, res=" << status;

be/src/util/doris_metrics.cpp

@@ -191,9 +191,6 @@ DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(query_ctx_cnt, MetricUnit::NOUNIT);
 DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(scanner_ctx_cnt, MetricUnit::NOUNIT);
 DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(scanner_cnt, MetricUnit::NOUNIT);
 DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(scanner_task_cnt, MetricUnit::NOUNIT);
-DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(scanner_task_queued, MetricUnit::NOUNIT);
-DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(scanner_task_running, MetricUnit::NOUNIT);
-DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(scanner_task_submit_failed, MetricUnit::NOUNIT);
 
 const std::string DorisMetrics::_s_registry_name = "doris_be";
 const std::string DorisMetrics::_s_hook_name = "doris_metrics";
@@ -319,9 +316,6 @@ DorisMetrics::DorisMetrics() : _metric_registry(_s_registry_name) {
     INT_COUNTER_METRIC_REGISTER(_server_metric_entity, scanner_ctx_cnt);
     INT_COUNTER_METRIC_REGISTER(_server_metric_entity, scanner_cnt);
     INT_COUNTER_METRIC_REGISTER(_server_metric_entity, scanner_task_cnt);
-    INT_COUNTER_METRIC_REGISTER(_server_metric_entity, scanner_task_queued);
-    INT_COUNTER_METRIC_REGISTER(_server_metric_entity, scanner_task_running);
-    INT_COUNTER_METRIC_REGISTER(_server_metric_entity, scanner_task_submit_failed);
 }
 
 void DorisMetrics::initialize(bool init_system_metrics, const std::set<std::string>& disk_devices,

be/src/util/doris_metrics.h

@@ -24,6 +24,7 @@
 #include <string>
 #include <vector>
 
+#include "runtime/exec_env.h"
 #include "util/jvm_metrics.h"
 #include "util/metrics.h"
 #include "util/system_metrics.h"
@@ -47,6 +48,10 @@ namespace doris {
 
 class DorisMetrics {
 public:
+    DorisMetrics();
+
+    static DorisMetrics* instance() { return ExecEnv::GetInstance()->doris_metrics(); }
+
     IntCounter* fragment_requests_total = nullptr;
     IntCounter* fragment_request_duration_us = nullptr;
     IntCounter* query_scan_bytes = nullptr;
@@ -197,10 +202,7 @@ class DorisMetrics {
     UIntGauge* query_cache_sql_total_count = nullptr;
     UIntGauge* query_cache_partition_total_count = nullptr;
 
-    UIntGauge* scanner_thread_pool_queue_size = nullptr;
     UIntGauge* add_batch_task_queue_size = nullptr;
-    UIntGauge* send_batch_thread_pool_thread_num = nullptr;
-    UIntGauge* send_batch_thread_pool_queue_size = nullptr;
     UIntGauge* fragment_thread_pool_queue_size = nullptr;
     UIntGauge* fragment_thread_pool_num_active_threads = nullptr;
 
@@ -244,14 +246,6 @@ class DorisMetrics {
     IntCounter* scanner_ctx_cnt = nullptr;
     IntCounter* scanner_cnt = nullptr;
     IntCounter* scanner_task_cnt = nullptr;
-    IntCounter* scanner_task_queued = nullptr;
-    IntCounter* scanner_task_submit_failed = nullptr;
-    IntCounter* scanner_task_running = nullptr;
-
-    static DorisMetrics* instance() {
-        static DorisMetrics instance;
-        return &instance;
-    }
 
     // not thread-safe, call before calling metrics
     void initialize(
@@ -266,14 +260,10 @@ class DorisMetrics {
     void init_jvm_metrics(JNIEnv* env);
 
 private:
-    // Don't allow constructor
-    DorisMetrics();
-
     void _update();
     void _update_process_thread_num();
     void _update_process_fd_num();
 
-private:
     static const std::string _s_registry_name;
     static const std::string _s_hook_name;
 

be/src/util/interval_histogram.cpp

@@ -0,0 +1,80 @@
+// 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 "util/interval_histogram.h"
+
+#include <algorithm>
+#include <mutex>
+#include <numeric>
+#include <vector>
+
+#include "gutil/integral_types.h"
+
+namespace doris {
+
+template <typename T>
+IntervalHistogramStat<T>::IntervalHistogramStat(size_t N) : window(N) {}
+
+template <typename T>
+void IntervalHistogramStat<T>::add(T value) {
+    std::unique_lock<std::shared_mutex> lock(mutex);
+    if (window.full()) {
+        window.pop_front();
+    }
+    window.push_back(value);
+}
+
+template <typename T>
+T IntervalHistogramStat<T>::mean() {
+    std::shared_lock<std::shared_mutex> lock(mutex);
+    if (window.empty()) {
+        return T();
+    }
+    T sum = std::accumulate(window.begin(), window.end(), T());
+    return sum / window.size();
+}
+
+template <typename T>
+T IntervalHistogramStat<T>::median() {
+    std::shared_lock<std::shared_mutex> lock(mutex);
+    if (window.empty()) {
+        return T();
+    }
+
+    std::vector<T> sorted(window.begin(), window.end());
+    std::sort(sorted.begin(), sorted.end());
+
+    size_t mid = sorted.size() / 2;
+    return sorted.size() % 2 == 0 ? (sorted[mid - 1] + sorted[mid]) / 2 : sorted[mid];
+}
+
+template <typename T>
+T IntervalHistogramStat<T>::max() {
+    std::shared_lock<std::shared_mutex> lock(mutex);
+    return *std::max_element(window.begin(), window.end());
+}
+
+template <typename T>
+T IntervalHistogramStat<T>::min() {
+    std::shared_lock<std::shared_mutex> lock(mutex);
+    return *std::min_element(window.begin(), window.end());
+}
+
+template class doris::IntervalHistogramStat<int64>;
+template class doris::IntervalHistogramStat<int32>;
+
+} // namespace doris

be/src/util/interval_histogram.h

@@ -0,0 +1,46 @@
+// 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 <boost/circular_buffer.hpp>
+#include <shared_mutex>
+
+namespace doris {
+
+// A thread-safe interval histogram stat class.
+// IntervalHistogramStat will keep a FIXED-SIZE window of values and provide
+// statistics like mean, median, max, min.
+
+template <typename T>
+class IntervalHistogramStat {
+public:
+    explicit IntervalHistogramStat(size_t N);
+
+    void add(T value);
+
+    T mean();
+    T median();
+    T max();
+    T min();
+
+private:
+    boost::circular_buffer<T> window;
+    mutable std::shared_mutex mutex;
+};
+
+} // namespace doris

be/src/util/metrics.cpp

@@ -169,6 +169,7 @@ std::string HistogramMetric::to_string() const {
 std::string HistogramMetric::to_prometheus(const std::string& display_name,
                                            const Labels& entity_labels,
                                            const Labels& metric_labels) const {
+    // TODO: Use std::string concate for better performance.
     std::stringstream ss;
     for (const auto& percentile : _s_output_percentiles) {
         auto quantile_lable = Labels({{"quantile", percentile.first}});