[pick](branch-2.1) pick #42059

be/CMakeLists.txt

@@ -750,6 +750,7 @@ install(DIRECTORY DESTINATION ${OUTPUT_DIR}/conf)
 install(FILES
     ${BASE_DIR}/../bin/start_be.sh
     ${BASE_DIR}/../bin/stop_be.sh
+    ${BASE_DIR}/../tools/jeprof
     PERMISSIONS OWNER_READ OWNER_WRITE OWNER_EXECUTE
     GROUP_READ GROUP_WRITE GROUP_EXECUTE
     WORLD_READ WORLD_EXECUTE

be/src/http/action/compaction_action.cpp

@@ -54,7 +54,7 @@ const static std::string HEADER_JSON = "application/json";
 
 CompactionAction::CompactionAction(CompactionActionType ctype, ExecEnv* exec_env,
                                    TPrivilegeHier::type hier, TPrivilegeType::type ptype)
-        : HttpHandlerWithAuth(exec_env, hier, ptype), _type(ctype) {}
+        : HttpHandlerWithAuth(exec_env, hier, ptype), _compaction_type(ctype) {}
 Status CompactionAction::_check_param(HttpRequest* req, uint64_t* tablet_id, uint64_t* table_id) {
     // req tablet id and table id, we have to set only one of them.
     std::string req_tablet_id = req->param(TABLET_ID_KEY);
@@ -341,15 +341,15 @@ Status CompactionAction::_execute_compaction_callback(TabletSharedPtr tablet,
 void CompactionAction::handle(HttpRequest* req) {
     req->add_output_header(HttpHeaders::CONTENT_TYPE, HEADER_JSON.c_str());
 
-    if (_type == CompactionActionType::SHOW_INFO) {
+    if (_compaction_type == CompactionActionType::SHOW_INFO) {
         std::string json_result;
         Status st = _handle_show_compaction(req, &json_result);
         if (!st.ok()) {
             HttpChannel::send_reply(req, HttpStatus::OK, st.to_json());
         } else {
             HttpChannel::send_reply(req, HttpStatus::OK, json_result);
         }
-    } else if (_type == CompactionActionType::RUN_COMPACTION) {
+    } else if (_compaction_type == CompactionActionType::RUN_COMPACTION) {
         std::string json_result;
         Status st = _handle_run_compaction(req, &json_result);
         if (!st.ok()) {

be/src/http/action/compaction_action.h

@@ -74,7 +74,7 @@ class CompactionAction : public HttpHandlerWithAuth {
     Status _check_param(HttpRequest* req, uint64_t* id_param, const std::string param_name);
 
 private:
-    CompactionActionType _type;
+    CompactionActionType _compaction_type;
 };
 
 } // end namespace doris

be/src/http/action/jeprofile_actions.cpp

@@ -18,67 +18,101 @@
 #include "http/action/jeprofile_actions.h"
 
 #include <jemalloc/jemalloc.h>
-#include <stdlib.h>
 #include <unistd.h>
 
-#include <ctime>
-#include <fstream>
-#include <memory>
-#include <mutex>
 #include <string>
 
-#include "common/config.h"
-#include "common/object_pool.h"
 #include "http/ev_http_server.h"
 #include "http/http_channel.h"
 #include "http/http_handler.h"
-#include "http/http_method.h"
-#include "io/fs/local_file_system.h"
+#include "http/http_handler_with_auth.h"
+#include "http/http_headers.h"
+#include "http/http_request.h"
+#include "runtime/memory/heap_profiler.h"
 
 namespace doris {
-class HttpRequest;
 
-static std::mutex kJeprofileActionMutex;
-class JeHeapAction : public HttpHandler {
-public:
-    JeHeapAction() = default;
-    virtual ~JeHeapAction() = default;
+const static std::string HEADER_JSON = "application/json";
 
-    virtual void handle(HttpRequest* req) override;
-};
-
-void JeHeapAction::handle(HttpRequest* req) {
-    std::lock_guard<std::mutex> lock(kJeprofileActionMutex);
-#ifndef USE_JEMALLOC
-    std::string str = "jemalloc heap dump is not available without setting USE_JEMALLOC";
-    HttpChannel::send_reply(req, str);
+static bool compile_check(HttpRequest* req) {
+#if defined(ADDRESS_SANITIZER) || defined(LEAK_SANITIZER) || defined(THREAD_SANITIZER)
+    HttpChannel::send_reply(
+            req, HttpStatus::INTERNAL_SERVER_ERROR,
+            "Jemalloc heap dump is not available with ASAN(address sanitizer) builds.\n");
+    return false;
+#elif !defined(USE_JEMALLOC)
+    HttpChannel::send_reply(req, HttpStatus::INTERNAL_SERVER_ERROR,
+                            "jemalloc heap dump is not available without setting USE_JEMALLOC.\n");
+    return false;
 #else
-    std::stringstream tmp_jeprof_file_name;
-    std::time_t now = std::time(nullptr);
-    // Build a temporary file name that is hopefully unique.
-    tmp_jeprof_file_name << config::jeprofile_dir << "/jeheap_dump." << now << "." << getpid()
-                         << "." << rand() << ".heap";
-    const std::string& tmp_file_name_str = tmp_jeprof_file_name.str();
-    const char* file_name_ptr = tmp_file_name_str.c_str();
-    int result = jemallctl("prof.dump", nullptr, nullptr, &file_name_ptr, sizeof(const char*));
-    std::stringstream response;
-    if (result == 0) {
-        response << "Jemalloc heap dump success, dump file path: " << tmp_jeprof_file_name.str()
-                 << "\n";
-    } else {
-        response << "Jemalloc heap dump failed, je_mallctl return: " << result << "\n";
-    }
-    HttpChannel::send_reply(req, response.str());
+    return true;
 #endif
 }
 
-Status JeprofileActions::setup(doris::ExecEnv* exec_env, doris::EvHttpServer* http_server,
-                               doris::ObjectPool& pool) {
-    if (!config::jeprofile_dir.empty()) {
-        RETURN_IF_ERROR(io::global_local_filesystem()->create_directory(config::jeprofile_dir));
+void SetJeHeapProfileActiveActions::handle(HttpRequest* req) {
+    req->add_output_header(HttpHeaders::CONTENT_TYPE, HEADER_JSON.c_str());
+    if (compile_check(req)) {
+        if (req->param("prof_value") == "true") {
+            HeapProfiler::instance()->heap_profiler_start();
+            HttpChannel::send_reply(
+                    req, HttpStatus::OK,
+                    "heap profiler started\nJemalloc will only track and sample the memory "
+                    "allocated and freed after the heap profiler started, it cannot analyze the "
+                    "memory allocated and freed before. Therefore, dumping the heap profile "
+                    "immediately after start heap profiler may prompt `No nodes to print`. If you "
+                    "want to analyze the memory that has been allocated in the past, you can only "
+                    "restart the BE process and start heap profiler immediately.\n");
+        } else {
+            HeapProfiler::instance()->heap_profiler_stop();
+            HttpChannel::send_reply(req, HttpStatus::OK, "heap profiler stoped\n");
+        }
+    }
+}
+
+void DumpJeHeapProfileToDotActions::handle(HttpRequest* req) {
+    req->add_output_header(HttpHeaders::CONTENT_TYPE, HEADER_JSON.c_str());
+    if (compile_check(req)) {
+        if (!HeapProfiler::instance()->check_heap_profiler()) {
+            HttpChannel::send_reply(
+                    req, HttpStatus::INTERNAL_SERVER_ERROR,
+                    "`curl http://be_host:be_webport/jeheap/prof/true` to start heap profiler\n");
+        }
+        std::string dot = HeapProfiler::instance()->dump_heap_profile_to_dot();
+        if (dot.empty()) {
+            HttpChannel::send_reply(req, HttpStatus::INTERNAL_SERVER_ERROR,
+                                    "dump heap profile to dot failed, see be.INFO\n");
+        } else {
+            dot += "\n-------------------------------------------------------\n";
+            dot += "Copy the text after `digraph` in the above output to "
+                   "http://www.webgraphviz.com to generate a dot graph.\n"
+                   "after start heap profiler, if there is no operation, will print `No nodes to "
+                   "print`."
+                   "If there are many errors: `addr2line: Dwarf Error`,"
+                   "or other FAQ, reference doc: "
+                   "https://doris.apache.org/community/developer-guide/debug-tool/#4-qa\n";
+            HttpChannel::send_reply(req, HttpStatus::OK, dot);
+        }
+    }
+}
+
+void DumpJeHeapProfileActions::handle(HttpRequest* req) {
+    req->add_output_header(HttpHeaders::CONTENT_TYPE, HEADER_JSON.c_str());
+    if (compile_check(req)) {
+        if (!HeapProfiler::instance()->check_heap_profiler()) {
+            HttpChannel::send_reply(
+                    req, HttpStatus::INTERNAL_SERVER_ERROR,
+                    "`curl http://be_host:be_webport/jeheap/prof/true` to start heap profiler\n");
+        }
+        std::string profile_file_name = HeapProfiler::instance()->dump_heap_profile();
+        if (profile_file_name.empty()) {
+            HttpChannel::send_reply(req, HttpStatus::INTERNAL_SERVER_ERROR,
+                                    "jemalloc heap dump failed\n");
+        } else {
+            HttpChannel::send_reply(req, HttpStatus::OK,
+                                    fmt::format("jemalloc heap dump success, dump file path: {}\n",
+                                                profile_file_name));
+        }
     }
-    http_server->register_handler(HttpMethod::GET, "/jeheap/dump", pool.add(new JeHeapAction()));
-    return Status::OK();
 }
 
 } // namespace doris

be/src/http/action/jeprofile_actions.h

@@ -15,17 +15,35 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#ifndef DORIS_JEPROFILE_ACTIONS_H
-#define DORIS_JEPROFILE_ACTIONS_H
-#include "common/status.h"
+#pragma once
+
+#include "http/http_handler.h"
+#include "http/http_handler_with_auth.h"
+
 namespace doris {
-class EvHttpServer;
+
+class HttpRequest;
 class ExecEnv;
-class ObjectPool;
-class JeprofileActions {
+
+class SetJeHeapProfileActiveActions final : public HttpHandlerWithAuth {
+public:
+    SetJeHeapProfileActiveActions(ExecEnv* exec_env) : HttpHandlerWithAuth(exec_env) {}
+    ~SetJeHeapProfileActiveActions() override = default;
+    void handle(HttpRequest* req) override;
+};
+
+class DumpJeHeapProfileToDotActions final : public HttpHandlerWithAuth {
+public:
+    DumpJeHeapProfileToDotActions(ExecEnv* exec_env) : HttpHandlerWithAuth(exec_env) {}
+    ~DumpJeHeapProfileToDotActions() override = default;
+    void handle(HttpRequest* req) override;
+};
+
+class DumpJeHeapProfileActions final : public HttpHandlerWithAuth {
 public:
-    static Status setup(ExecEnv* exec_env, EvHttpServer* http_server, ObjectPool& pool);
+    DumpJeHeapProfileActions(ExecEnv* exec_env) : HttpHandlerWithAuth(exec_env) {}
+    ~DumpJeHeapProfileActions() override = default;
+    void handle(HttpRequest* req) override;
 };
 
 } // namespace doris
-#endif //DORIS_JEPROFILE_ACTIONS_H

be/src/http/http_handler_with_auth.h

@@ -37,6 +37,7 @@ class HttpHandlerWithAuth : public HttpHandler {
 public:
     HttpHandlerWithAuth(ExecEnv* exec_env, TPrivilegeHier::type hier, TPrivilegeType::type type);
 
+    HttpHandlerWithAuth(ExecEnv* exec_env) : _exec_env(exec_env) {}
     ~HttpHandlerWithAuth() override = default;
 
     // return 0 if auth pass, otherwise -1.
@@ -53,10 +54,8 @@ class HttpHandlerWithAuth : public HttpHandler {
 
 protected:
     ExecEnv* _exec_env;
-
-private:
-    TPrivilegeHier::type _hier;
-    TPrivilegeType::type _type;
+    TPrivilegeHier::type _hier = TPrivilegeHier::GLOBAL;
+    TPrivilegeType::type _type = TPrivilegeType::ADMIN;
 };
 
 } // namespace doris

be/src/runtime/exec_env.h

@@ -109,6 +109,7 @@ class LookupConnectionCache;
 class RowCache;
 class DummyLRUCache;
 class CacheManager;
+class HeapProfiler;
 class WalManager;
 class DNSCache;
 
@@ -310,6 +311,7 @@ class ExecEnv {
     LookupConnectionCache* get_lookup_connection_cache() { return _lookup_connection_cache; }
     RowCache* get_row_cache() { return _row_cache; }
     CacheManager* get_cache_manager() { return _cache_manager; }
+    HeapProfiler* get_heap_profiler() { return _heap_profiler; }
     segment_v2::InvertedIndexSearcherCache* get_inverted_index_searcher_cache() {
         return _inverted_index_searcher_cache;
     }
@@ -452,6 +454,7 @@ class ExecEnv {
     LookupConnectionCache* _lookup_connection_cache = nullptr;
     RowCache* _row_cache = nullptr;
     CacheManager* _cache_manager = nullptr;
+    HeapProfiler* _heap_profiler = nullptr;
     segment_v2::InvertedIndexSearcherCache* _inverted_index_searcher_cache = nullptr;
     segment_v2::InvertedIndexQueryCache* _inverted_index_query_cache = nullptr;
     std::shared_ptr<DummyLRUCache> _dummy_lru_cache = nullptr;

be/src/runtime/exec_env_init.cpp

@@ -64,6 +64,7 @@
 #include "runtime/load_path_mgr.h"
 #include "runtime/load_stream_mgr.h"
 #include "runtime/memory/cache_manager.h"
+#include "runtime/memory/heap_profiler.h"
 #include "runtime/memory/mem_tracker.h"
 #include "runtime/memory/mem_tracker_limiter.h"
 #include "runtime/memory/thread_mem_tracker_mgr.h"
@@ -389,6 +390,7 @@ Status ExecEnv::_init_mem_env() {
     bool is_percent = false;
     std::stringstream ss;
     // 1. init mem tracker
+    _heap_profiler = HeapProfiler::create_global_instance();
     init_mem_tracker();
     thread_context()->thread_mem_tracker_mgr->init();
 #if defined(USE_MEM_TRACKER) && !defined(__SANITIZE_ADDRESS__) && !defined(ADDRESS_SANITIZER) && \
@@ -701,6 +703,8 @@ void ExecEnv::destroy() {
     // dns cache is a global instance and need to be released at last
     SAFE_DELETE(_dns_cache);
 
+    SAFE_DELETE(_heap_profiler);
+
     _s_tracking_memory = false;
     LOG(INFO) << "Doris exec envorinment is destoried.";
 }