[Metrics] Add some metrics for container size in BE (#3246)

We can observe the workload of BE, and also it's a way to check
whether there is any problem in BE, like some container increase
too large and lead to OOM.

This patch add the following metrics:
```
Name                                   Description
rowset_count_generated_and_in_use      The total count of rowset id generated and in use since BE last start
unused_rowsets_count                   The total count of unused rowset waiting to be GC
broker_count                           The total count of brokers in management
data_stream_receiver_count             The total count of data stream receivers in management
fragment_endpoint_count                The total count of fragment endpoints of data stream in management, should always equal to data_stream_receiver_count
active_scan_context_count              The total count of active scan contexts
plan_fragment_count                    The total count of plan fragments in executing
load_channel_count                     The total count of load channels in management
result_buffer_block_count              The total count of result buffer blocks for queries, each block has a limited queue size (default 1024)
result_block_queue_count               The total count of queues for fragments, each queue has a limited size (default 20, by config::max_memory_sink_batch_count)
routine_load_task_count                The total count of routine load tasks in executing
small_file_cache_count                 The total count of cached small files' digest info
stream_load_pipe_count                 The total count of stream load pipes, each pipe has a limited buffer size (default 1M)
tablet_writer_count                    The total count of tablet writers
brpc_endpoint_stub_count               The total count of brpc endpoints
```

be/src/olap/rowset/unique_rowset_id_generator.cpp

@@ -16,13 +16,20 @@
 // under the License.
 
 #include "olap/rowset/unique_rowset_id_generator.h"
+
+#include "util/doris_metrics.h"
 #include "util/spinlock.h"
 #include "util/uid_util.h"
 
 namespace doris {
 
 UniqueRowsetIdGenerator::UniqueRowsetIdGenerator(const UniqueId& backend_uid)
-        : _backend_uid(backend_uid), _inc_id(0) {}
+        : _backend_uid(backend_uid), _inc_id(0) {
+    REGISTER_GAUGE_DORIS_METRIC(rowset_count_generated_and_in_use, [this]() {
+        std::lock_guard<SpinLock> l(_lock);
+        return _valid_rowset_id_hi.size();
+    });
+}
 
 // generate a unique rowset id and save it in a set to check whether it is valid in the future
 RowsetId UniqueRowsetIdGenerator::next_id() {

be/src/olap/storage_engine.cpp

@@ -123,6 +123,10 @@ StorageEngine::StorageEngine(const EngineOptions& options)
     if (_s_instance == nullptr) {
         _s_instance = this;
     }
+    REGISTER_GAUGE_DORIS_METRIC(unused_rowsets_count, [this]() {
+        MutexLock lock(&_gc_mutex);
+        return _unused_rowsets.size();
+    });
 }
 
 StorageEngine::~StorageEngine() {

be/src/runtime/broker_mgr.cpp

@@ -25,12 +25,17 @@
 #include "service/backend_options.h"
 #include "runtime/exec_env.h"
 #include "runtime/client_cache.h"
+#include "util/doris_metrics.h"
 #include "util/thrift_util.h"
 
 namespace doris {
 
 BrokerMgr::BrokerMgr(ExecEnv* exec_env) : 
         _exec_env(exec_env), _thread_stop(false), _ping_thread(&BrokerMgr::ping_worker, this) {
+    REGISTER_GAUGE_DORIS_METRIC(broker_count, [this]() {
+        std::lock_guard<std::mutex> l(_mutex);
+        return _broker_set.size();
+    });
 }
 
 BrokerMgr::~BrokerMgr() {

be/src/runtime/data_stream_mgr.cpp

@@ -26,6 +26,7 @@
 #include "runtime/data_stream_recvr.h"
 #include "runtime/raw_value.h"
 #include "runtime/runtime_state.h"
+#include "util/doris_metrics.h"
 #include "util/uid_util.h"
 
 #include "gen_cpp/types.pb.h" // PUniqueId
@@ -40,6 +41,17 @@ using boost::unique_lock;
 using boost::try_mutex;
 using boost::lock_guard;
 
+DataStreamMgr::DataStreamMgr() {
+    REGISTER_GAUGE_DORIS_METRIC(data_stream_receiver_count, [this]() {
+        lock_guard<mutex> l(_lock);
+        return _receiver_map.size();
+    });
+    REGISTER_GAUGE_DORIS_METRIC(fragment_endpoint_count, [this]() {
+        lock_guard<mutex> l(_lock);
+        return _fragment_stream_set.size();
+    });
+}
+
 inline uint32_t DataStreamMgr::get_hash_value(
         const TUniqueId& fragment_instance_id, PlanNodeId node_id) {
     uint32_t value = RawValue::get_hash_value(&fragment_instance_id.lo, TYPE_BIGINT, 0);

be/src/runtime/data_stream_mgr.h

@@ -66,7 +66,7 @@ class PUniqueId;
 // per-query memory limits.
 class DataStreamMgr {
 public:
-    DataStreamMgr() {}
+    DataStreamMgr();
 
     // Create a receiver for a specific fragment_instance_id/node_id destination;
     // If is_merging is true, the receiver maintains a separate queue of incoming row

be/src/runtime/external_scan_context_mgr.cpp

@@ -23,6 +23,7 @@
 
 #include "runtime/fragment_mgr.h"
 #include "runtime/result_queue_mgr.h"
+#include "util/doris_metrics.h"
 #include "util/uid_util.h"
 
 namespace doris {
@@ -32,6 +33,10 @@ ExternalScanContextMgr::ExternalScanContextMgr(ExecEnv* exec_env) : _exec_env(ex
     _keep_alive_reaper.reset(
             new std::thread(
                     std::bind<void>(std::mem_fn(&ExternalScanContextMgr::gc_expired_context), this)));
+    REGISTER_GAUGE_DORIS_METRIC(active_scan_context_count, [this]() {
+        std::lock_guard<std::mutex> l(_lock);
+        return _active_contexts.size();
+    });
 }
 
 Status ExternalScanContextMgr::create_scan_context(std::shared_ptr<ScanContext>* p_context) {
@@ -121,4 +126,4 @@ void ExternalScanContextMgr::gc_expired_context() {
         }
     }
 }
-}
+}

be/src/runtime/external_scan_context_mgr.h

@@ -72,4 +72,4 @@ private:
     std::mutex _lock;
 };
 
-}
+}

be/src/runtime/fragment_mgr.cpp

@@ -365,6 +365,10 @@ FragmentMgr::FragmentMgr(ExecEnv* exec_env) :
         // TODO(zc): we need a better thread-pool
         // now one user can use all the thread pool, others have no resource.
         _thread_pool(config::fragment_pool_thread_num, config::fragment_pool_queue_size) {
+    REGISTER_GAUGE_DORIS_METRIC(plan_fragment_count, [this]() {
+        std::lock_guard<std::mutex> lock(_lock);
+        return _fragment_map.size();
+    });
 }
 
 FragmentMgr::~FragmentMgr() {

be/src/runtime/fragment_mgr.h

@@ -91,7 +91,6 @@ private:
     std::thread _cancel_thread;
     // every job is a pool
     PriorityThreadPool _thread_pool;
-
 };
 
 }

be/src/runtime/load_channel_mgr.cpp

@@ -22,6 +22,7 @@
 #include "runtime/load_channel.h"
 #include "runtime/mem_tracker.h"
 #include "service/backend_options.h"
+#include "util/doris_metrics.h"
 #include "util/stopwatch.hpp"
 
 namespace doris {
@@ -61,6 +62,10 @@ static int64_t calc_job_timeout_s(int64_t timeout_in_req_s) {
 }
 
 LoadChannelMgr::LoadChannelMgr() : _is_stopped(false) {
+    REGISTER_GAUGE_DORIS_METRIC(load_channel_count, [this]() {
+        std::lock_guard<std::mutex> l(_lock);
+        return _load_channels.size();
+    });
     _lastest_success_channel = new_lru_cache(1024);
 }
 

be/src/runtime/result_buffer_mgr.cpp

@@ -20,6 +20,7 @@
 #include "runtime/buffer_control_block.h"
 #include "runtime/raw_value.h"
 #include "util/debug_util.h"
+#include "util/doris_metrics.h"
 #include "gen_cpp/PaloInternalService_types.h"
 #include "gen_cpp/types.pb.h"
 
@@ -33,6 +34,12 @@ namespace doris {
 
 ResultBufferMgr::ResultBufferMgr()
     : _is_stop(false) {
+    // Each BufferControlBlock has a limited queue size of 1024, it's not needed to count the
+    // actual size of all BufferControlBlock.
+    REGISTER_GAUGE_DORIS_METRIC(result_buffer_block_count, [this]() {
+        boost::lock_guard<boost::mutex> l(_lock);
+        return _buffer_map.size();
+    });
 }
 
 ResultBufferMgr::~ResultBufferMgr() {

be/src/runtime/result_queue_mgr.cpp

@@ -24,11 +24,19 @@
 #include "gen_cpp/Types_types.h"
 #include "runtime/exec_env.h"
 #include "util/arrow/row_batch.h"
+#include "util/doris_metrics.h"
 
 namespace doris {
 
 ResultQueueMgr::ResultQueueMgr() {
+    // Each BlockingQueue has a limited size (default 20, by config::max_memory_sink_batch_count),
+    // it's not needed to count the actual size of all BlockingQueue.
+    REGISTER_GAUGE_DORIS_METRIC(result_block_queue_count, [this]() {
+        std::lock_guard<std::mutex> l(_lock);
+        return _fragment_queue_map.size();
+    });
 }
+
 ResultQueueMgr::~ResultQueueMgr() {
 }
 

be/src/runtime/routine_load/routine_load_task_executor.h

@@ -22,6 +22,7 @@
 #include <mutex>
 
 #include "runtime/routine_load/data_consumer_pool.h"
+#include "util/doris_metrics.h"
 #include "util/priority_thread_pool.hpp"
 #include "util/uid_util.h"
 
@@ -46,6 +47,10 @@ public:
         _exec_env(exec_env),
         _thread_pool(config::routine_load_thread_pool_size, 1),
         _data_consumer_pool(10) {
+        REGISTER_GAUGE_DORIS_METRIC(routine_load_task_count, [this]() {
+            std::lock_guard<std::mutex> l(_lock);
+            return _task_map.size();
+        });
 
         _data_consumer_pool.start_bg_worker();
     }

be/src/runtime/small_file_mgr.cpp

@@ -31,6 +31,7 @@
 #include "gen_cpp/HeartbeatService.h"
 #include "http/http_client.h"
 #include "runtime/exec_env.h"
+#include "util/doris_metrics.h"
 #include "util/file_utils.h"
 #include "util/md5.h"
 
@@ -42,6 +43,10 @@ SmallFileMgr::SmallFileMgr(
         const std::string& local_path) :
     _exec_env(env),
     _local_path(local_path) {
+    REGISTER_GAUGE_DORIS_METRIC(small_file_cache_count, [this]() {
+        std::lock_guard<std::mutex> l(_lock);
+        return _file_cache.size();
+    });
 }
 
 SmallFileMgr::~SmallFileMgr() {

be/src/runtime/stream_load/load_stream_mgr.h

@@ -22,6 +22,7 @@
 #include <unordered_map>
 
 #include "runtime/stream_load/stream_load_pipe.h" // for StreamLoadPipe
+#include "util/doris_metrics.h"
 #include "util/uid_util.h" // for std::hash for UniqueId
 
 namespace doris {
@@ -29,7 +30,14 @@ namespace doris {
 // used to register all streams in process so that other module can get this stream
 class LoadStreamMgr {
 public:
-    LoadStreamMgr() { }
+    LoadStreamMgr() {
+        // Each StreamLoadPipe has a limited buffer size (default 1M), it's not needed to count the
+        // actual size of all StreamLoadPipe.
+        REGISTER_GAUGE_DORIS_METRIC(stream_load_pipe_count, [this]() {
+            std::lock_guard<std::mutex> l(_lock);
+            return _stream_map.size();
+        });
+    }
     ~LoadStreamMgr() { }
 
     Status put(const UniqueId& id,

be/src/runtime/tablets_channel.cpp

@@ -23,15 +23,25 @@
 #include "olap/memtable.h"
 #include "runtime/row_batch.h"
 #include "runtime/tuple_row.h"
+#include "util/doris_metrics.h"
 
 namespace doris {
 
+std::atomic<uint64_t> TabletsChannel::_s_tablet_writer_count;
+
 TabletsChannel::TabletsChannel(const TabletsChannelKey& key, MemTracker* mem_tracker):
         _key(key), _state(kInitialized), _closed_senders(64) {
     _mem_tracker.reset(new MemTracker(-1, "tablets channel", mem_tracker));
+    static std::once_flag once_flag;
+    std::call_once(once_flag, [] {
+        REGISTER_GAUGE_DORIS_METRIC(tablet_writer_count, [&]() {
+            return _s_tablet_writer_count.load();
+        });
+    });
 }
 
 TabletsChannel::~TabletsChannel() {
+    _s_tablet_writer_count -= _tablet_writers.size();
     for (auto& it : _tablet_writers) {
         delete it.second;
     }
@@ -237,6 +247,7 @@ Status TabletsChannel::_open_all_writers(const PTabletWriterOpenRequest& params)
         }
         _tablet_writers.emplace(tablet.tablet_id(), writer);
     }
+    _s_tablet_writer_count += _tablet_writers.size();
     DCHECK(_tablet_writers.size() == params.tablets_size());
     return Status::OK();
 }

be/src/runtime/tablets_channel.h

@@ -124,6 +124,8 @@ private:
     std::unordered_set<int64_t> _partition_ids;
 
     std::unique_ptr<MemTracker> _mem_tracker;
+
+    static std::atomic<uint64_t> _s_tablet_writer_count;
 };
 
 

be/src/util/brpc_stub_cache.h

@@ -24,6 +24,7 @@
 #include "gen_cpp/internal_service.pb.h"
 #include "gen_cpp/palo_internal_service.pb.h"
 #include "service/brpc.h"
+#include "util/doris_metrics.h"
 #include "util/spinlock.h"
 
 namespace doris {
@@ -33,6 +34,10 @@ class BrpcStubCache {
 public:
     BrpcStubCache() {
         _stub_map.init(239);
+        REGISTER_GAUGE_DORIS_METRIC(brpc_endpoint_stub_count, [this]() {
+            std::lock_guard<SpinLock> l(_lock);
+            return _stub_map.size();
+        });
     }
     ~BrpcStubCache() {
         for (auto& stub : _stub_map) {

be/src/util/doris_metrics.cpp

@@ -136,6 +136,22 @@ IntGauge DorisMetrics::blocks_open_writing;
 
 IntCounter DorisMetrics::blocks_push_remote_duration_us;
 
+UIntGauge DorisMetrics::rowset_count_generated_and_in_use;
+UIntGauge DorisMetrics::unused_rowsets_count;
+UIntGauge DorisMetrics::broker_count;
+UIntGauge DorisMetrics::data_stream_receiver_count;
+UIntGauge DorisMetrics::fragment_endpoint_count;
+UIntGauge DorisMetrics::active_scan_context_count;
+UIntGauge DorisMetrics::plan_fragment_count;
+UIntGauge DorisMetrics::load_channel_count;
+UIntGauge DorisMetrics::result_buffer_block_count;
+UIntGauge DorisMetrics::result_block_queue_count;
+UIntGauge DorisMetrics::routine_load_task_count;
+UIntGauge DorisMetrics::small_file_cache_count;
+UIntGauge DorisMetrics::stream_load_pipe_count;
+UIntGauge DorisMetrics::brpc_endpoint_stub_count;
+UIntGauge DorisMetrics::tablet_writer_count;
+
 DorisMetrics::~DorisMetrics() {
     delete _system_metrics;
     delete _metrics;

be/src/util/doris_metrics.h

@@ -47,6 +47,12 @@ private:
     std::unordered_map<std::string, IntGauge> metrics;
 };
 
+#define REGISTER_GAUGE_DORIS_METRIC(name, func) \
+  DorisMetrics::metrics()->register_metric(#name, &DorisMetrics::name); \
+  DorisMetrics::metrics()->register_hook(#name, [&]() { \
+      DorisMetrics::name.set_value(func());  \
+});
+
 class DorisMetrics {
 public:
     // counters
@@ -166,6 +172,23 @@ public:
 
     static IntCounter blocks_push_remote_duration_us;
 
+    // Size of some global containers
+    static UIntGauge rowset_count_generated_and_in_use;
+    static UIntGauge unused_rowsets_count;
+    static UIntGauge broker_count;
+    static UIntGauge data_stream_receiver_count;
+    static UIntGauge fragment_endpoint_count;
+    static UIntGauge active_scan_context_count;
+    static UIntGauge plan_fragment_count;
+    static UIntGauge load_channel_count;
+    static UIntGauge result_buffer_block_count;
+    static UIntGauge result_block_queue_count;
+    static UIntGauge routine_load_task_count;
+    static UIntGauge small_file_cache_count;
+    static UIntGauge stream_load_pipe_count;
+    static UIntGauge brpc_endpoint_stub_count;
+    static UIntGauge tablet_writer_count;
+
     ~DorisMetrics();
     // not thread-safe, call before calling metrics
     void initialize(

be/test/exec/tablet_sink_test.cpp

@@ -33,6 +33,7 @@
 #include "util/brpc_stub_cache.h"
 #include "util/cpu_info.h"
 #include "util/debug/leakcheck_disabler.h"
+#include "util/doris_metrics.h"
 #include "runtime/descriptor_helper.h"
 #include "runtime/bufferpool/reservation_tracker.h"
 #include "runtime/exec_env.h"
@@ -49,6 +50,7 @@ public:
     OlapTableSinkTest() { }
     virtual ~OlapTableSinkTest() { }
     void SetUp() override {
+        DorisMetrics::instance()->initialize("ut");
         k_add_batch_status = Status::OK();
         _env = ExecEnv::GetInstance();
         _env->_thread_mgr = new ThreadResourceMgr();

be/test/http/stream_load_test.cpp

@@ -78,6 +78,7 @@ public:
         k_response_str = "";
         config::streaming_load_max_mb = 1;
 
+        DorisMetrics::instance()->initialize("ut");
         _env._thread_mgr = new ThreadResourceMgr();
         _env._master_info = new TMasterInfo();
         _env._load_stream_mgr = new LoadStreamMgr();
@@ -108,7 +109,6 @@ private:
 };
 
 TEST_F(StreamLoadActionTest, no_auth) {
-    DorisMetrics::instance()->initialize("StreamLoadActionTest");
     StreamLoadAction action(&_env);
 
     HttpRequest request(_evhttp_req);
@@ -123,7 +123,6 @@ TEST_F(StreamLoadActionTest, no_auth) {
 
 #if 0
 TEST_F(StreamLoadActionTest, no_content_length) {
-    DorisMetrics::instance()->initialize("StreamLoadActionTest");
     StreamLoadAction action(&__env);
 
     HttpRequest request(_evhttp_req);
@@ -138,7 +137,6 @@ TEST_F(StreamLoadActionTest, no_content_length) {
 }
 
 TEST_F(StreamLoadActionTest, unknown_encoding) {
-    DorisMetrics::instance()->initialize("StreamLoadActionTest");
     StreamLoadAction action(&_env);
 
     HttpRequest request(_evhttp_req);
@@ -155,7 +153,6 @@ TEST_F(StreamLoadActionTest, unknown_encoding) {
 #endif
 
 TEST_F(StreamLoadActionTest, normal) {
-    DorisMetrics::instance()->initialize("StreamLoadActionTest");
     StreamLoadAction action(&_env);
 
     HttpRequest request(_evhttp_req);
@@ -176,7 +173,6 @@ TEST_F(StreamLoadActionTest, normal) {
 }
 
 TEST_F(StreamLoadActionTest, put_fail) {
-    DorisMetrics::instance()->initialize("StreamLoadActionTest");
     StreamLoadAction action(&_env);
 
     HttpRequest request(_evhttp_req);
@@ -199,7 +195,6 @@ TEST_F(StreamLoadActionTest, put_fail) {
 }
 
 TEST_F(StreamLoadActionTest, commit_fail) {
-    DorisMetrics::instance()->initialize("StreamLoadActionTest");
     StreamLoadAction action(&_env);
 
     HttpRequest request(_evhttp_req);
@@ -220,7 +215,6 @@ TEST_F(StreamLoadActionTest, commit_fail) {
 }
 
 TEST_F(StreamLoadActionTest, begin_fail) {
-    DorisMetrics::instance()->initialize("StreamLoadActionTest");
     StreamLoadAction action(&_env);
 
     HttpRequest request(_evhttp_req);
@@ -242,7 +236,6 @@ TEST_F(StreamLoadActionTest, begin_fail) {
 
 #if 0
 TEST_F(StreamLoadActionTest, receive_failed) {
-    DorisMetrics::instance()->initialize("StreamLoadActionTest");
     StreamLoadAction action(&_env);
 
     HttpRequest request(_evhttp_req);
@@ -259,7 +252,6 @@ TEST_F(StreamLoadActionTest, receive_failed) {
 #endif
 
 TEST_F(StreamLoadActionTest, plan_fail) {
-    DorisMetrics::instance()->initialize("StreamLoadActionTest");
     StreamLoadAction action(&_env);
 
     HttpRequest request(_evhttp_req);

be/test/olap/delete_handler_test.cpp

@@ -32,8 +32,9 @@
 #include "olap/push_handler.h"
 #include "olap/utils.h"
 #include "olap/options.h"
-#include "util/logging.h"
+#include "util/doris_metrics.h"
 #include "util/file_utils.h"
+#include "util/logging.h"
 
 using namespace std;
 using namespace doris;
@@ -46,6 +47,7 @@ static const uint32_t MAX_PATH_LEN = 1024;
 static StorageEngine* k_engine = nullptr;
 
 void set_up() {
+    DorisMetrics::instance()->initialize("ut");
     char buffer[MAX_PATH_LEN];
     getcwd(buffer, MAX_PATH_LEN);
     config::storage_root_path = string(buffer) + "/data_test";

be/test/olap/delta_writer_test.cpp

@@ -33,6 +33,7 @@
 #include "runtime/exec_env.h"
 #include "runtime/mem_pool.h"
 #include "runtime/mem_tracker.h"
+#include "util/doris_metrics.h"
 #include "util/logging.h"
 #include "util/file_utils.h"
 #include "olap/options.h"
@@ -58,6 +59,7 @@ void set_up() {
     std::vector<StorePath> paths;
     paths.emplace_back(config::storage_root_path, -1);
 
+    DorisMetrics::instance()->initialize("ut");
     doris::EngineOptions options;
     options.store_paths = paths;
     doris::StorageEngine::open(options, &k_engine);

be/test/olap/olap_snapshot_converter_test.cpp

@@ -34,6 +34,7 @@
 #include "olap/rowset/rowset_meta_manager.h"
 #include "olap/storage_engine.h"
 #include "olap/txn_manager.h"
+#include "util/doris_metrics.h"
 #include "util/file_utils.h"
 
 #ifndef BE_TEST
@@ -55,6 +56,7 @@ public:
         config::tablet_map_shard_size = 1;
         config::txn_map_shard_size = 1;
         config::txn_shard_size = 1;
+        DorisMetrics::instance()->initialize("ut");
         std::vector<StorePath> paths;
         paths.emplace_back("_engine_data_path", -1);
         EngineOptions options;

be/test/olap/rowset/alpha_rowset_test.cpp

@@ -34,6 +34,7 @@
 #include "olap/rowset/alpha_rowset_reader.h"
 #include "olap/data_dir.h"
 #include "olap/storage_engine.h"
+#include "util/doris_metrics.h"
 
 #ifndef BE_TEST
 #define BE_TEST
@@ -49,6 +50,7 @@ namespace doris {
 static const uint32_t MAX_PATH_LEN = 1024;
 
 void set_up() {
+    DorisMetrics::instance()->initialize("ut");
     config::path_gc_check = false;
     char buffer[MAX_PATH_LEN];
     getcwd(buffer, MAX_PATH_LEN);

be/test/olap/rowset/beta_rowset_test.cpp

@@ -34,8 +34,9 @@
 #include "runtime/exec_env.h"
 #include "runtime/mem_tracker.h"
 #include "runtime/mem_pool.h"
-#include "util/slice.h"
+#include "util/doris_metrics.h"
 #include "util/file_utils.h"
+#include "util/slice.h"
 
 using std::string;
 
@@ -52,6 +53,7 @@ protected:
         config::tablet_map_shard_size = 1;
         config::txn_map_shard_size = 1;
         config::txn_shard_size = 1;
+        DorisMetrics::instance()->initialize("ut");
         char buffer[MAX_PATH_LEN];
         getcwd(buffer, MAX_PATH_LEN);
         config::storage_root_path = std::string(buffer) + "/data_test";

be/test/olap/rowset/rowset_converter_test.cpp

@@ -37,6 +37,7 @@
 #include "olap/storage_engine.h"
 #include "olap/olap_cond.h"
 #include "runtime/exec_env.h"
+#include "util/doris_metrics.h"
 
 #ifndef BE_TEST
 #define BE_TEST
@@ -153,6 +154,7 @@ public:
         config::tablet_map_shard_size = 1;
         config::txn_map_shard_size = 1;
         config::txn_shard_size = 1;
+        DorisMetrics::instance()->initialize("ut");
         config::path_gc_check = false;
         char buffer[MAX_PATH_LEN];
         getcwd(buffer, MAX_PATH_LEN);

be/test/olap/rowset/rowset_meta_manager_test.cpp

@@ -26,6 +26,7 @@
 #include "olap/storage_engine.h"
 #include "boost/filesystem.hpp"
 #include "json2pb/json_to_pb.h"
+#include "util/doris_metrics.h"
 
 #ifndef BE_TEST
 #define BE_TEST
@@ -48,6 +49,7 @@ public:
         config::tablet_map_shard_size = 1;
         config::txn_map_shard_size = 1;
         config::txn_shard_size = 1;
+        DorisMetrics::instance()->initialize("ut");
         std::vector<StorePath> paths;
         paths.emplace_back("_engine_data_path", -1);
         EngineOptions options;

be/test/olap/rowset/unique_rowset_id_generator_test.cpp

@@ -20,6 +20,7 @@
 #include <gtest/gtest.h>
 #include <iostream>
 
+#include "util/doris_metrics.h"
 #include "util/runtime_profile.h"
 #include "util/threadpool.h"
 #include "util/pretty_printer.h"
@@ -30,6 +31,10 @@ public:
     UniqueRowsetIdGeneratorTest() { }
     virtual ~UniqueRowsetIdGeneratorTest() {
     }
+
+    virtual void SetUp() {
+        DorisMetrics::instance()->initialize("ut");
+    }
 };
 
 TEST_F(UniqueRowsetIdGeneratorTest, RowsetIdFormatTest) {

be/test/olap/tablet_mgr_test.cpp

@@ -31,6 +31,7 @@
 #include "boost/filesystem.hpp"
 #include "json2pb/json_to_pb.h"
 #include "util/file_utils.h"
+#include "util/doris_metrics.h"
 
 #ifndef BE_TEST
 #define BE_TEST
@@ -51,6 +52,7 @@ public:
         config::tablet_map_shard_size = 1;
         config::txn_map_shard_size = 1;
         config::txn_shard_size = 1;
+        DorisMetrics::instance()->initialize("ut");
         string test_engine_data_path = "./be/test/olap/test_data/converter_test_data/data";
         _engine_data_path = "./be/test/olap/test_data/converter_test_data/tmp";
         boost::filesystem::remove_all(_engine_data_path);

be/test/olap/txn_manager_test.cpp

@@ -30,6 +30,7 @@
 #include "olap/txn_manager.h"
 #include "boost/filesystem.hpp"
 #include "json2pb/json_to_pb.h"
+#include "util/doris_metrics.h"
 
 #ifndef BE_TEST
 #define BE_TEST
@@ -98,6 +99,7 @@ public:
         config::txn_shard_size = 1;
         config::max_runnings_transactions_per_txn_map = 500;
         _txn_mgr.reset(new TxnManager(64, 1024));
+        DorisMetrics::instance()->initialize("ut");
         std::vector<StorePath> paths;
         paths.emplace_back("_engine_data_path", -1);
         EngineOptions options;

be/test/runtime/external_scan_context_mgr_test.cpp

@@ -24,12 +24,14 @@
 #include "runtime/fragment_mgr.h"
 #include "runtime/result_queue_mgr.h"
 #include "runtime/thread_resource_mgr.h"
+#include "util/doris_metrics.h"
 
 namespace doris {
 
 class ExternalScanContextMgrTest : public testing::Test {
 public:
     ExternalScanContextMgrTest() {
+        DorisMetrics::instance()->initialize("ut");
         FragmentMgr* fragment_mgr = new FragmentMgr(&_exec_env);
         ThreadResourceMgr* thread_mgr = new ThreadResourceMgr();
         ResultQueueMgr* result_queue_mgr = new  ResultQueueMgr();

be/test/runtime/fragment_mgr_test.cpp

@@ -23,6 +23,7 @@
 #include "exec/data_sink.h"
 #include "runtime/plan_fragment_executor.h"
 #include "runtime/row_batch.h"
+#include "util/doris_metrics.h"
 #include "util/monotime.h"
 
 namespace doris {
@@ -61,6 +62,7 @@ protected:
                   << ", pool_size=" << config::fragment_pool_queue_size;
         config::fragment_pool_thread_num = 32;
         config::fragment_pool_queue_size = 1024;
+        DorisMetrics::instance()->initialize("ut");
     }
     virtual void TearDown() {}
 };

be/test/runtime/load_channel_mgr_test.cpp

@@ -30,6 +30,7 @@
 #include "runtime/row_batch.h"
 #include "runtime/tuple_row.h"
 #include "runtime/descriptor_helper.h"
+#include "util/doris_metrics.h"
 #include "util/thrift_util.h"
 #include "olap/delta_writer.h"
 #include "olap/memtable_flush_executor.h"
@@ -103,6 +104,7 @@ public:
         add_status = OLAP_SUCCESS;
         close_status = OLAP_SUCCESS;
         config::streaming_load_rpc_max_alive_time_sec = 120;
+        DorisMetrics::instance()->initialize("ut");
     }
 private:
 };

be/test/runtime/memory_scratch_sink_test.cpp

@@ -41,6 +41,7 @@
 #include "runtime/thread_resource_mgr.h"
 #include "runtime/tuple_row.h"
 #include "util/blocking_queue.hpp"
+#include "util/doris_metrics.h"
 #include "util/logging.h"
 #include "testutil/desc_tbl_builder.h"
 
@@ -77,10 +78,12 @@ public:
     virtual void SetUp() {
         config::periodic_counter_update_period_ms = 500;
         config::storage_root_path = "./data";
+        DorisMetrics::instance()->initialize("ut");
 
         system("mkdir -p ./test_run/output/");
         system("pwd");
         system("cp -r ./be/test/runtime/test_data/ ./test_run/.");
+
         init();
     }
 

be/test/runtime/result_queue_mgr_test.cpp

@@ -25,6 +25,7 @@
 #include "gen_cpp/DorisExternalService_types.h"
 #include "runtime/result_queue_mgr.h"
 #include "util/blocking_queue.hpp"
+#include "util/doris_metrics.h"
 
 namespace doris {
 
@@ -37,6 +38,7 @@ public:
 
 protected:
     virtual void SetUp() {
+        DorisMetrics::instance()->initialize("ut");
     }
 
 };

be/test/runtime/routine_load_task_executor_test.cpp

@@ -21,6 +21,7 @@
 #include "runtime/stream_load/load_stream_mgr.h"
 #include "runtime/stream_load/stream_load_executor.h"
 #include "util/cpu_info.h"
+#include "util/doris_metrics.h"
 #include "util/logging.h"
 
 #include <gtest/gtest.h>
@@ -49,6 +50,7 @@ public:
         k_stream_load_rollback_result = TLoadTxnRollbackResult();
         k_stream_load_put_result = TStreamLoadPutResult();
 
+        DorisMetrics::instance()->initialize("ut");
         _env._master_info = new TMasterInfo();
         _env._load_stream_mgr = new LoadStreamMgr();
         _env._stream_load_executor = new StreamLoadExecutor(&_env);

be/test/runtime/small_file_mgr_test.cpp

@@ -29,7 +29,7 @@
 #include "http/http_request.h"
 #include "gen_cpp/HeartbeatService_types.h"
 #include "runtime/exec_env.h"
-
+#include "util/doris_metrics.h"
 
 int main(int argc, char* argv[]);
 
@@ -87,6 +87,7 @@ public:
 
     static void SetUpTestCase() {
         s_server = new EvHttpServer(0);
+        DorisMetrics::instance()->initialize("ut");
         s_server->register_handler(GET, "/api/get_small_file", &s_test_handler);
         s_server->start();
         real_port = s_server->get_real_port();

be/test/util/arrow/arrow_work_flow_test.cpp

@@ -40,6 +40,7 @@
 #include "util/arrow/row_batch.h"
 #include "util/debug_util.h"
 #include "util/disk_info.h"
+#include "util/doris_metrics.h"
 #include "util/cpu_info.h"
 #include "util/logging.h"
 
@@ -54,10 +55,12 @@ protected:
     virtual void SetUp() {
         config::periodic_counter_update_period_ms = 500;
         config::storage_root_path = "./data";
+        DorisMetrics::instance()->initialize("ut");
 
         system("mkdir -p ./test_run/output/");
         system("pwd");
         system("cp -r ./be/test/util/test_data/ ./test_run/.");
+
         init();
     }
     virtual void TearDown() {