branch-3.0: [opt](cloud) Set thread name for all background thread workers to improve observability #49366 (#50557)

Cherry-picked from #49366

Co-authored-by: Gavin Chou <[email protected]>

Author: github-actions[bot]

cloud/src/common/metric.cpp

@@ -259,6 +259,7 @@ int FdbMetricExporter::start() {
                                    [this]() { return !running_.load(std::memory_order_acquire); });
         }
     });
+    pthread_setname_np(thread_->native_handle(), "fdb_metrics_exporter");
     return 0;
 }
 

cloud/src/common/simple_thread_pool.h

@@ -17,11 +17,14 @@
 
 #pragma once
 
+#include <pthread.h>
+
 #include <atomic>
 #include <condition_variable>
 #include <iostream>
 #include <memory>
 #include <mutex>
+#include <string>
 #include <thread>
 #include <vector>
 
@@ -38,9 +41,11 @@ class SimpleThreadPool {
     std::vector<std::thread> _worker_thread_group; // multi thread pool
     std::atomic<bool> _is_running;
     size_t _pool_size;
+    std::string _pool_name;
 
 public:
-    SimpleThreadPool(size_t size) : _is_running(false), _pool_size(size) {
+    SimpleThreadPool(size_t size, const std::string& name = "")
+            : _is_running(false), _pool_size(size), _pool_name(name) {
         _job_queue = std::make_shared<SimpleSyncQueue<JobType>>(_pool_size * 2);
     }
 
@@ -86,8 +91,11 @@ class SimpleThreadPool {
     int start() {
         _is_running = true;
         _worker_thread_group.clear();
+        _pool_name = _pool_name.empty() ? "simple_thread_pool" : _pool_name;
         for (size_t i = 0; i < _pool_size; ++i) {
             _worker_thread_group.emplace_back(&SimpleThreadPool::work, this);
+            std::string name = _pool_name + "_" + std::to_string(i);
+            pthread_setname_np(_worker_thread_group.back().native_handle(), name.c_str());
         }
         return 0;
     }

cloud/src/main.cpp

@@ -310,6 +310,7 @@ int main(int argc, char** argv) {
             }
         };
         periodiccally_log_thread = std::thread {periodiccally_log};
+        pthread_setname_np(periodiccally_log_thread.native_handle(), "recycler_periodically_log");
     }
     // start service
     brpc::ServerOptions options;

cloud/src/meta-service/meta_server.cpp

@@ -176,6 +176,7 @@ MetaServerRegister::MetaServerRegister(std::shared_ptr<TxnKv> txn_kv)
         }
         LOG(INFO) << "register thread quits";
     }));
+    pthread_setname_np(register_thread_->native_handle(), "ms_register_thread");
 }
 
 MetaServerRegister::~MetaServerRegister() {

cloud/src/meta-service/txn_kv.cpp

@@ -259,6 +259,7 @@ int Network::init() {
                 bool expected = true;
                 Network::working.compare_exchange_strong(expected, false);
             });
+    pthread_setname_np(network_thread_->native_handle(), "fdb_network_thread");
 
     return 0;
 }

cloud/src/meta-service/txn_lazy_committer.cpp

@@ -499,7 +499,8 @@ std::pair<MetaServiceCode, std::string> TxnLazyCommitTask::wait() {
 }
 
 TxnLazyCommitter::TxnLazyCommitter(std::shared_ptr<TxnKv> txn_kv) : txn_kv_(txn_kv) {
-    worker_pool_ = std::make_unique<SimpleThreadPool>(config::txn_lazy_commit_num_threads);
+    worker_pool_ = std::make_unique<SimpleThreadPool>(config::txn_lazy_commit_num_threads,
+                                                      "txn_lazy_commiter");
     worker_pool_->start();
 }
 
@@ -541,4 +542,4 @@ void TxnLazyCommitter::remove(int64_t txn_id) {
     running_tasks_.erase(txn_id);
 }
 
-} // namespace doris::cloud
+} // namespace doris::cloud

cloud/src/recycler/recycler.cpp

@@ -174,12 +174,14 @@ static inline void check_recycle_task(const std::string& instance_id, const std:
 Recycler::Recycler(std::shared_ptr<TxnKv> txn_kv) : txn_kv_(std::move(txn_kv)) {
     ip_port_ = std::string(butil::my_ip_cstr()) + ":" + std::to_string(config::brpc_listen_port);
 
-    auto s3_producer_pool = std::make_shared<SimpleThreadPool>(config::recycle_pool_parallelism);
+    auto s3_producer_pool = std::make_shared<SimpleThreadPool>(config::recycle_pool_parallelism,
+                                                               "s3_producer_pool");
     s3_producer_pool->start();
-    auto recycle_tablet_pool = std::make_shared<SimpleThreadPool>(config::recycle_pool_parallelism);
+    auto recycle_tablet_pool = std::make_shared<SimpleThreadPool>(config::recycle_pool_parallelism,
+                                                                  "recycle_tablet_pool");
     recycle_tablet_pool->start();
-    auto group_recycle_function_pool =
-            std::make_shared<SimpleThreadPool>(config::recycle_pool_parallelism);
+    auto group_recycle_function_pool = std::make_shared<SimpleThreadPool>(
+            config::recycle_pool_parallelism, "group_recycle_function_pool");
     group_recycle_function_pool->start();
     _thread_pool_group =
             RecyclerThreadPoolGroup(std::move(s3_producer_pool), std::move(recycle_tablet_pool),
@@ -1914,8 +1916,8 @@ int InstanceRecycler::recycle_rowsets() {
     // Store keys of rowset recycled by background workers
     std::mutex async_recycled_rowset_keys_mutex;
     std::vector<std::string> async_recycled_rowset_keys;
-    auto worker_pool =
-            std::make_unique<SimpleThreadPool>(config::instance_recycler_worker_pool_size);
+    auto worker_pool = std::make_unique<SimpleThreadPool>(
+            config::instance_recycler_worker_pool_size, "recycle_rowsets");
     worker_pool->start();
     auto delete_rowset_data_by_prefix = [&](std::string key, const std::string& resource_id,
                                             int64_t tablet_id, const std::string& rowset_id) {

cloud/src/recycler/recycler_service.cpp

@@ -208,6 +208,7 @@ void recycle_copy_jobs(const std::shared_ptr<TxnKv>& txn_kv, const std::string&
         std::lock_guard lock(s_worker_mtx);
         s_worker.erase(instance_id);
     });
+    pthread_setname_np(worker.native_handle(), "recycler_worker");
     worker.detach();
 }
 

cloud/src/recycler/s3_accessor.cpp

@@ -240,7 +240,8 @@ int S3Accessor::init() {
     static std::once_flag log_annotated_tags_key_once;
     std::call_once(log_annotated_tags_key_once, [&]() {
         LOG_INFO("start s3 accessor parallel worker pool");
-        worker_pool = std::make_shared<SimpleThreadPool>(config::recycle_pool_parallelism);
+        worker_pool =
+                std::make_shared<SimpleThreadPool>(config::recycle_pool_parallelism, "s3_accessor");
         worker_pool->start();
     });
     switch (conf_.provider) {