pingcap · ti-chi-bot · Aug 23, 2024 · Aug 8, 2024 · Aug 8, 2024 · Aug 8, 2024
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/proxy_components/proxy_server/Cargo.toml b/proxy_components/proxy_server/Cargo.toml
@@ -42,6 +42,7 @@ range_cache_memory_engine = { workspace = true }
 health_controller = { workspace = true }
 api_version = { workspace = true }
 async-stream = "0.2"
+backtrace = "0.3"
 backup = { workspace = true, default-features = false }
 backup-stream = { workspace = true, default-features = false }
 causal_ts = { workspace = true }

diff --git a/proxy_components/proxy_server/src/status_server/jeprof.in b/proxy_components/proxy_server/src/status_server/jeprof.in
@@ -3718,6 +3718,7 @@ BEGIN {
     my $slots = $self->{slots};
     my $str;
     read($self->{file}, $str, 8);
+
     # Set the global $address_length based on what we see here.
     # 8 is 32-bit (8 hexadecimal chars); 16 is 64-bit (16 hexadecimal chars).
     $address_length = ($str eq (chr(0)x8)) ? 16 : 8;

diff --git a/proxy_components/proxy_server/src/status_server/mod.rs b/proxy_components/proxy_server/src/status_server/mod.rs
@@ -4,6 +4,7 @@ pub mod profile;
 pub mod vendored_utils;
 
 use std::{
+    env::args,
     error::Error as StdError,
     marker::PhantomData,
     net::SocketAddr,
@@ -239,10 +240,19 @@ where
         let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();
 
         let use_jeprof = query_pairs.get("jeprof").map(|x| x.as_ref()) == Some("true");
+        let output_format = match query_pairs.get("text").map(|x| x.as_ref()) {
+            None => "--svg",
+            Some("svg") => "--svg",
+            Some("text") => "--text",
+            Some("raw") => "--raw",
+            Some("collapsed") => "--collapsed",
+            _ => "--svg",
+        }
+        .to_string();
 
         let result = if let Some(name) = query_pairs.get("name") {
             if use_jeprof {
-                jeprof_heap_profile(name)
+                jeprof_heap_profile(name, output_format)
             } else {
                 read_file(name)
             }
@@ -261,7 +271,7 @@ where
             let end = Compat01As03::new(timer)
                 .map_err(|_| TIMER_CANCELED.to_owned())
                 .into_future();
-            start_one_heap_profile(end, use_jeprof).await
+            start_one_heap_profile(end, use_jeprof, output_format).await
         };
 
         match result {
@@ -339,6 +349,83 @@ where
         })
     }
 
+    fn get_cmdline(_req: Request<Body>) -> hyper::Result<Response<Body>> {
+        let args = args().fold(String::new(), |mut a, b| {
+            a.push_str(&b);
+            a.push('\x00');
+            a
+        });
+        let response = Response::builder()
+            .header("Content-Type", mime::TEXT_PLAIN.to_string())
+            .header("X-Content-Type-Options", "nosniff")
+            .body(args.into())
+            .unwrap();
+        Ok(response)
+    }
+
+    fn get_symbol_count(req: Request<Body>) -> hyper::Result<Response<Body>> {
+        assert_eq!(req.method(), Method::GET);
+        // We don't know how many symbols we have, but we
+        // do have symbol information. pprof only cares whether
+        // this number is 0 (no symbols available) or > 0.
+        let text = "num_symbols: 1\n";
+        let response = Response::builder()
+            .header("Content-Type", mime::TEXT_PLAIN.to_string())
+            .header("X-Content-Type-Options", "nosniff")
+            .header("Content-Length", text.len())
+            .body(text.into())
+            .unwrap();
+        Ok(response)
+    }
+
+    // The request and response format follows pprof remote server
+    // https://gperftools.github.io/gperftools/pprof_remote_servers.html
+    // Here is the go pprof implementation:
+    // https://github.com/golang/go/blob/3857a89e7eb872fa22d569e70b7e076bec74ebbb/src/net/http/pprof/pprof.go#L191
+    async fn get_symbol(req: Request<Body>) -> hyper::Result<Response<Body>> {
+        assert_eq!(req.method(), Method::POST);
+        let mut text = String::new();
+        let body_bytes = hyper::body::to_bytes(req.into_body()).await?;
+        let body = String::from_utf8(body_bytes.to_vec()).unwrap();
+
+        // The request body is a list of addr to be resolved joined by '+'.
+        // Resolve addrs with addr2line and write the symbols each per line in
+        // response.
+        for pc in body.split('+') {
+            let addr = usize::from_str_radix(pc.trim_start_matches("0x"), 16).unwrap_or(0);
+            if addr == 0 {
+                info!("invalid addr: {}", addr);
+                continue;
+            }
+
+            // Would be multiple symbols if inlined.
+            let mut syms = vec![];
+            backtrace::resolve(addr as *mut std::ffi::c_void, |sym| {
+                let name = sym
+                    .name()
+                    .unwrap_or_else(|| backtrace::SymbolName::new(b"<unknown>"));
+                syms.push(name.to_string());
+            });
+
+            if !syms.is_empty() {
+                // join inline functions with '--'
+                let f = syms.join("--");
+                // should be <hex address> <function name>
+                text.push_str(format!("{:#x} {}\n", addr, f).as_str());
+            } else {
+                info!("can't resolve mapped addr: {:#x}", addr);
+                text.push_str(format!("{:#x} ??\n", addr).as_str());
+            }
+        }
+        let response = Response::builder()
+            .header("Content-Type", mime::TEXT_PLAIN.to_string())
+            .header("X-Content-Type-Options", "nosniff")
+            .header("Content-Length", text.len())
+            .body(text.into())
+            .unwrap();
+        Ok(response)
+    }
+
     async fn update_config(
         cfg_controller: ConfigController,
         req: Request<Body>,
@@ -712,6 +799,9 @@ where
                             (Method::GET, "/debug/pprof/profile") => {
                                 Self::dump_cpu_prof_to_resp(req).await
                             }
+                            (Method::GET, "/debug/pprof/cmdline") => Self::get_cmdline(req),
+                            (Method::GET, "/debug/pprof/symbol") => Self::get_symbol_count(req),
+                            (Method::POST, "/debug/pprof/symbol") => Self::get_symbol(req).await,
                             (Method::GET, "/debug/fail_point") => {
                                 info!("debug fail point API start");
                                 fail_point!("debug_fail_point");
@@ -731,7 +821,10 @@ where
                                 Self::handle_http_request(req, engine_store_server_helper).await
                             }
 
-                            _ => Ok(make_response(StatusCode::NOT_FOUND, "path not found")),
+                            _ => Ok(make_response(
+                                StatusCode::NOT_FOUND,
+                                format!("path not found, {:?}", req),
+                            )),
                         }
                     }
                 }))

diff --git a/proxy_components/proxy_server/src/status_server/profile.rs b/proxy_components/proxy_server/src/status_server/profile.rs
@@ -100,7 +100,11 @@ impl<'a, I, T> Future for ProfileGuard<'a, I, T> {
 
 /// Trigger a heap profie and return the content.
 #[allow(dead_code)]
-pub async fn start_one_heap_profile<F>(end: F, use_jeprof: bool) -> Result<Vec<u8>, String>
+pub async fn start_one_heap_profile<F>(
+    end: F,
+    use_jeprof: bool,
+    output_format: String,
+) -> Result<Vec<u8>, String>
 where
     F: Future<Output = Result<(), String>> + Send + 'static,
 {
@@ -112,8 +116,10 @@ where
         let path = f.path().to_str().unwrap();
         dump_prof(path).map_err(|e| format!("dump_prof: {}", e))?;
         if use_jeprof {
-            jeprof_heap_profile(path)
+            // Use jeprof to transform heap file into svg/raw/collapsed...
+            jeprof_heap_profile(path, output_format)
         } else {
+            // Juse return the heap file.
             read_file(path)
         }
     };
@@ -231,22 +237,50 @@ pub fn read_file(path: &str) -> Result<Vec<u8>, String> {
     Ok(buf)
 }
 
-pub fn jeprof_heap_profile(path: &str) -> Result<Vec<u8>, String> {
+pub fn jeprof_heap_profile(path: &str, output_format: String) -> Result<Vec<u8>, String> {
     let bin = std::env::current_exe().map_err(|e| format!("get current exe path fail: {}", e))?;
+    let mut bin_proxy = bin.clone();
+    let mut bin_proxy_str = String::from("");
+    if bin_proxy.pop() {
+        bin_proxy.push("libtiflash_proxy.so");
+        if let Some(s) = &bin_proxy.to_str() {
+            if std::path::Path::new(s).exists() {
+                bin_proxy_str = s.to_string();
+            }
+        }
+    }
     let bin_str = &bin.as_os_str().to_string_lossy();
     info!(
-        "using jeprof to process {} bin {} exist {}",
+        "using jeprof to process {} bin {} bin_proxy {} exist {}",
         path,
         bin_str,
+        bin_proxy_str,
         std::path::Path::new(path).exists()
     );
-    let mut jeprof = Command::new("perl")
-        .args(["/dev/stdin", "--show_bytes", bin_str, path, "--svg"])
-        .stdin(Stdio::piped())
-        .stderr(Stdio::piped())
-        .stdout(Stdio::piped())
-        .spawn()
-        .map_err(|e| format!("spawn jeprof fail: {}", e))?;
+    let mut jeprof = if !bin_proxy_str.is_empty() {
+        Command::new("perl")
+            .args([
+                "/dev/stdin",
+                "--show_bytes",
+                bin_str,
+                path,
+                &output_format,
+                &format!("--add_lib={}", bin_proxy_str),
+            ])
+            .stdin(Stdio::piped())
+            .stderr(Stdio::piped())
+            .stdout(Stdio::piped())
+            .spawn()
+            .map_err(|e| format!("spawn jeprof fail: {}", e))
+    } else {
+        Command::new("perl")
+            .args(["/dev/stdin", "--show_bytes", bin_str, path, &output_format])
+            .stdin(Stdio::piped())
+            .stderr(Stdio::piped())
+            .stdout(Stdio::piped())
+            .spawn()
+            .map_err(|e| format!("spawn jeprof fail: {}", e))
+    }?;
     jeprof
         .stdin
         .take()

diff --git a/proxy_tests/proxy/shared/fast_add_peer/fp.rs b/proxy_tests/proxy/shared/fast_add_peer/fp.rs
@@ -1007,6 +1007,8 @@ fn test_msgsnapshot_before_msgappend() {
 
     pd_client.must_add_peer(1, new_learner_peer(2, 2));
 
+    std::thread::sleep(std::time::Duration::from_secs(1));
+
     iter_ffi_helpers(&cluster, Some(vec![2]), &mut |_, ffi: &mut FFIHelperSet| {
         let mut x: u64 = 0;
         let mut y: u64 = 0;