// Copyright 2018 TiKV Project Authors. Licensed under Apache-2.0.

pub mod jemalloc_utils;
pub mod operator;
pub mod profile;
mod util;
pub mod vendored_utils;

use std::{
    collections::HashMap,
    env::args,
    error::Error as StdError,
    net::SocketAddr,
    path::PathBuf,
    sync::Arc,
    time::{Duration, Instant},
};

use anyhow::Context;
use futures::{
    compat::{Compat01As03, Stream01CompatExt},
    prelude::*,
};
use hyper::{
    Body, Method, Request, Response, Server, StatusCode,
    server::{Builder as HyperBuilder, accept::Accept, conn::AddrIncoming},
    service::{make_service_fn, service_fn},
};
use lazy_static::lazy_static;
use log::{error, info};
use profile::{
    activate_heap_profile, deactivate_heap_profile, dump_one_heap_profile, list_heap_profiles,
    start_one_cpu_profile,
};
use prometheus::TEXT_FORMAT;
use regex::Regex;
use tici_system::metrics::{dump, dump_to};
// use security::SecurityConfig;
use tici_system::timer::GLOBAL_TIMER_HANDLE;
use tokio::{
    io::{AsyncRead, AsyncWrite},
    runtime::Handle,
    sync::oneshot::{self, Receiver, Sender},
};
use util::*;
use vendored_utils::{jeprof_memory_status, jeprof_purge_arena};

use crate::{
    meta::MetaServiceCore,
    status_server::{
        operator::{MetaOperator, Operators},
        profile::set_prof_active,
    },
};

static TIMER_CANCELED: &str = "tokio timer canceled";

pub struct StatusServer {
    tx: Sender<()>,
    rx: Option<Receiver<()>>,
    addr: Option<SocketAddr>,
    store_path: PathBuf,
    op: Operators,
}

impl StatusServer {
    #[cfg_attr(not(target_arch = "x86_64"), allow(dead_code))]
    fn extract_thread_name(thread_name: &str) -> String {
        lazy_static! {
            static ref THREAD_NAME_RE: Regex =
                Regex::new(r"^(?P<thread_name>[a-z-_ :]+?)(-?\d)*$").unwrap();
            static ref THREAD_NAME_REPLACE_SEPERATOR_RE: Regex = Regex::new(r"[_ ]").unwrap();
        }

        THREAD_NAME_RE
            .captures(thread_name)
            .and_then(|cap| {
                cap.name("thread_name").map(|thread_name| {
                    THREAD_NAME_REPLACE_SEPERATOR_RE
                        .replace_all(thread_name.as_str(), "-")
                        .into_owned()
                })
            })
            .unwrap_or_else(|| thread_name.to_owned())
    }

    #[cfg(target_arch = "x86_64")]
    fn frames_post_processor() -> impl Fn(&mut pprof::Frames) {
        move |frames| {
            let name = Self::extract_thread_name(&frames.thread_name);
            frames.thread_name = name;
        }
    }
}

impl StatusServer {
    pub fn new(store_path: PathBuf, op: Operators) -> Self {
        let (tx, rx) = oneshot::channel::<()>();
        StatusServer {
            tx,
            rx: Some(rx),
            addr: None,
            store_path,
            op,
        }
    }

    pub fn new_for_meta(store_path: PathBuf, op: Arc<MetaServiceCore>) -> anyhow::Result<Self> {
        if !store_path.exists() {
            std::fs::create_dir_all(&store_path).with_context(|| {
                format!("Failed to create status server directory={:?}", store_path)
            })?;
        }
        Ok(StatusServer::new(
            store_path,
            Operators::Meta(MetaOperator::new(op)),
        ))
    }

    fn list_heap_prof(_req: Request<Body>) -> hyper::Result<Response<Body>> {
        let profiles = match list_heap_profiles() {
            Ok(s) => s,
            Err(e) => return Ok(make_response(StatusCode::INTERNAL_SERVER_ERROR, e)),
        };

        let text = profiles
            .into_iter()
            .map(|(f, ct)| format!("{}\t\t{}", f, ct))
            .collect::<Vec<_>>()
            .join("\n")
            .into_bytes();

        let response = Response::builder()
            .header("Content-Type", mime::TEXT_PLAIN.to_string())
            .header("Content-Length", text.len())
            .body(text.into())
            .unwrap();
        Ok(response)
    }

    async fn activate_heap_prof(
        req: Request<Body>,
        store_path: PathBuf,
    ) -> hyper::Result<Response<Body>> {
        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();

        let interval: u64 = match query_pairs.get("interval") {
            Some(val) => match val.parse() {
                Ok(val) => val,
                Err(err) => return Ok(make_response(StatusCode::BAD_REQUEST, err.to_string())),
            },
            None => 60,
        };

        let interval = Duration::from_secs(interval);
        let period = GLOBAL_TIMER_HANDLE
            .interval(Instant::now() + interval, interval)
            .compat()
            .map_ok(|_| ())
            .map_err(|_| TIMER_CANCELED.to_owned())
            .into_stream();
        let (tx, rx) = oneshot::channel();
        let callback = move || tx.send(()).unwrap_or_default();
        let res = Handle::current().spawn(activate_heap_profile(period, store_path, callback));
        if rx.await.is_ok() {
            let msg = "activate heap profile success";
            Ok(make_response(StatusCode::OK, msg))
        } else {
            let errmsg = format!("{:?}", res.await);
            Ok(make_response(StatusCode::INTERNAL_SERVER_ERROR, errmsg))
        }
    }

    fn deactivate_heap_prof(_req: Request<Body>) -> hyper::Result<Response<Body>> {
        let body = if deactivate_heap_profile() {
            "deactivate heap profile success"
        } else {
            "no heap profile is running"
        };
        Ok(make_response(StatusCode::OK, body))
    }

    fn set_profile_active(val: bool) -> hyper::Result<Response<Body>> {
        match set_prof_active(val) {
            Ok(()) => Ok(make_response(StatusCode::OK, "set prof.active succeed")),
            Err(err) => {
                info!("set_profile_active with err {}", &err);
                Ok(make_response(StatusCode::BAD_REQUEST, err.to_string()))
            }
        }
    }

    async fn arena_purge(_: Request<Body>) -> hyper::Result<Response<Body>> {
        jeprof_purge_arena();
        Ok(make_response(StatusCode::OK, "purge OK"))
    }

    async fn memory_status(_: Request<Body>) -> hyper::Result<Response<Body>> {
        let s = jeprof_memory_status();
        Ok(make_response(StatusCode::OK, s))
    }

    #[allow(dead_code)]
    async fn dump_heap_prof_to_resp(req: Request<Body>) -> hyper::Result<Response<Body>> {
        let query = req.uri().query().unwrap_or("");
        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 = dump_one_heap_profile(use_jeprof, output_format.clone());

        match result {
            Ok(body) => {
                info!("dump or get heap profile successfully");
                let mut response = Response::builder()
                    .header("X-Content-Type-Options", "nosniff")
                    .header("Content-Disposition", "attachment; filename=\"profile\"")
                    .header("Content-Length", body.len());
                response = if use_jeprof && output_format == "--svg" {
                    response.header("Content-Type", mime::IMAGE_SVG.to_string())
                } else {
                    response.header("Content-Type", mime::APPLICATION_OCTET_STREAM.to_string())
                };
                Ok(response.body(body.into()).unwrap())
            }
            Err(e) => {
                info!("dump or get heap profile fail, err={}", e);
                Ok(make_response(StatusCode::INTERNAL_SERVER_ERROR, e))
            }
        }
    }

    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)
    }

    #[cfg(target_arch = "x86_64")]
    pub async fn dump_rsprof(seconds: u64, frequency: i32) -> pprof::Result<pprof::Report> {
        let guard = pprof::ProfilerGuardBuilder::default()
            .frequency(frequency)
            .blocklist(&["libc", "libgcc", "pthread"])
            .build()?;
        info!(
            "start profiling {} seconds with frequency {} /s",
            seconds, frequency
        );
        let timer = GLOBAL_TIMER_HANDLE.clone();
        let _ = Compat01As03::new(timer.delay(Instant::now() + Duration::from_secs(seconds))).await;
        guard
            .report()
            .frames_post_processor(StatusServer::frames_post_processor())
            .build()
    }

    pub async fn dump_cpu_prof_to_resp(req: Request<Body>) -> hyper::Result<Response<Body>> {
        log::info!("start CPU profile {:?}", req);
        let query = req.uri().query().unwrap_or("");
        let query_pairs: HashMap<_, _> = url::form_urlencoded::parse(query.as_bytes()).collect();

        let seconds: u64 = match query_pairs.get("seconds") {
            Some(val) => match val.parse() {
                Ok(val) => val,
                Err(err) => return Ok(make_response(StatusCode::BAD_REQUEST, err.to_string())),
            },
            None => 10,
        };

        let frequency: i32 = match query_pairs.get("frequency") {
            Some(val) => match val.parse() {
                Ok(val) => val,
                Err(err) => return Ok(make_response(StatusCode::BAD_REQUEST, err.to_string())),
            },
            None => 99, /* Default frequency of sampling. 99Hz to avoid coincide with special
                         * periods */
        };

        let prototype_content_type: hyper::http::HeaderValue =
            hyper::http::HeaderValue::from_str("application/protobuf").unwrap();
        let output_protobuf = req.headers().get("Content-Type") == Some(&prototype_content_type);

        let timer = GLOBAL_TIMER_HANDLE.delay(Instant::now() + Duration::from_secs(seconds));
        let end = async move {
            Compat01As03::new(timer)
                .await
                .map_err(|_| TIMER_CANCELED.to_owned())
        };
        match start_one_cpu_profile(end, frequency, output_protobuf).await {
            Ok(body) => {
                info!("dump cpu profile successfully");
                let mut response = Response::builder()
                    .header(
                        "Content-Disposition",
                        "attachment; filename=\"cpu_profile\"",
                    )
                    .header("Content-Length", body.len());
                response = if output_protobuf {
                    response.header("Content-Type", mime::APPLICATION_OCTET_STREAM.to_string())
                } else {
                    response.header("Content-Type", mime::IMAGE_SVG.to_string())
                };
                Ok(response.body(body.into()).unwrap())
            }
            Err(e) => {
                info!("dump cpu profile fail, err={}", e);
                Ok(make_response(StatusCode::INTERNAL_SERVER_ERROR, e))
            }
        }
    }

    pub fn stop(self) {
        let _ = self.tx.send(());
    }

    // Return listening address, this may only be used for outer test
    // to get the real address because we may use "127.0.0.1:0"
    // in test to avoid port conflict.
    pub fn listening_addr(&self) -> SocketAddr {
        self.addr.unwrap()
    }
}

use flate2::{Compression, write::GzEncoder};
use http::header::{ACCEPT_ENCODING, CONTENT_ENCODING, CONTENT_TYPE, HeaderValue};

// check if the client allow return response with gzip compression
// the following logic is port from prometheus's golang:
// https://github.com/prometheus/client_golang/blob/24172847e35ba46025c49d90b8846b59eb5d9ead/prometheus/promhttp/http.go#L155-L176
fn client_accept_gzip(req: &Request<Body>) -> bool {
    let encoding = req
        .headers()
        .get(ACCEPT_ENCODING)
        .map(|enc| enc.to_str().unwrap_or_default())
        .unwrap_or_default();
    encoding
        .split(',')
        .map(|s| s.trim())
        .any(|s| s == "gzip" || s.starts_with("gzip;"))
}

impl StatusServer {
    fn handle_get_metrics(req: Request<Body>) -> hyper::Result<Response<Body>> {
        Self::metrics_to_resp(req, false)
    }

    fn metrics_to_resp(req: Request<Body>, should_simplify: bool) -> hyper::Result<Response<Body>> {
        let gz_encoding = client_accept_gzip(&req);
        let metrics = if gz_encoding {
            // gzip can reduce the body size to less than 1/10.
            let mut encoder = GzEncoder::new(vec![], Compression::default());
            dump_to(&mut encoder, should_simplify);
            encoder.finish().unwrap()
        } else {
            dump(should_simplify).into_bytes()
        };
        let mut resp = Response::new(metrics.into());
        resp.headers_mut()
            .insert(CONTENT_TYPE, HeaderValue::from_static(TEXT_FORMAT));
        if gz_encoding {
            resp.headers_mut()
                .insert(CONTENT_ENCODING, HeaderValue::from_static("gzip"));
        }

        Ok(resp)
    }

    fn start_serve<I, C>(&mut self, builder: HyperBuilder<I>)
    where
        I: Accept<Conn = C, Error = std::io::Error> + Send + 'static,
        I::Error: Into<Box<dyn StdError + Send + Sync>>,
        I::Conn: AsyncRead + AsyncWrite + Unpin + Send + 'static,
        C: ServerConnection,
    {
        log::info!("StatusServer: start_serve");
        let store_path = self.store_path.clone();
        let operator = self.op.clone();
        // Start to serve.
        let server = builder.serve(make_service_fn(move |_conn: &C| {
            let store_path = store_path.clone();
            let operator = operator.clone();
            async move {
                // Create a status service.
                Ok::<_, hyper::Error>(service_fn(move |req: Request<Body>| {
                    let store_path = store_path.clone();
                    let operator = operator.clone();
                    async move {
                        let path = req.uri().path().to_owned();
                        let method = req.method().to_owned();

                        #[cfg(feature = "failpoints")]
                        {
                            if path.starts_with(FAIL_POINTS_REQUEST_PATH) {
                                return handle_fail_points_request(req).await;
                            }
                        }

                        // 1. POST "/config" will modify the configuration of TiKV.
                        // 2. GET "/region" will get start key and end key. These keys could be
                        // actual user data since in some cases the data
                        // itself is stored in the key.
                        // let should_check_cert = !matches!(
                        //     (&method, path.as_ref()),
                        //     (&Method::GET, "/metrics")
                        //         | (&Method::GET, "/status")
                        //         | (&Method::GET, "/config")
                        //         | (&Method::GET, "/debug/pprof/profile")
                        // );

                        // if should_check_cert && !check_cert(security_config, x509) {
                        //     return Ok(make_response(
                        //         StatusCode::FORBIDDEN,
                        //         "certificate role error",
                        //     ));
                        // }

                        match (method, path.as_ref()) {
                            (Method::GET, "/metrics") => Self::handle_get_metrics(req),
                            (Method::GET, "/status") => Ok(Response::default()),
                            (Method::GET, "/debug/pprof/set_prof_active") => {
                                Self::set_profile_active(true)
                            }
                            (Method::GET, "/debug/pprof/set_prof_inactive") => {
                                Self::set_profile_active(false)
                            }
                            (Method::GET, "/debug/pprof/heap_list") => Self::list_heap_prof(req),
                            (Method::GET, "/debug/pprof/heap_activate") => {
                                Self::activate_heap_prof(req, store_path).await
                            }
                            (Method::GET, "/debug/pprof/heap_deactivate") => {
                                Self::deactivate_heap_prof(req)
                            }
                            // jeprof --svg "${host}:${status_port}/debug/pprof/heap" is fixed,
                            // check #127 or use curl
                            // "${host}:${status_port}/debug/pprof/heap?jeprof=true&text=svg"
                            (Method::GET, "/debug/pprof/heap") => {
                                Self::dump_heap_prof_to_resp(req).await
                            }
                            (Method::GET, "/debug/pprof/arena_purge") => {
                                Self::arena_purge(req).await
                            }
                            (Method::GET, "/debug/pprof/memory_status") => {
                                Self::memory_status(req).await
                            }
                            (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, "/config") => operator.get_config(req),
                            (Method::POST, "/config") => operator.update_config(req).await,
                            // (Method::GET, "/debug/fail_point") => {
                            //     info!("debug fail point API start");
                            //     fail_point!("debug_fail_point");
                            //     info!("debug fail point API finish");
                            //     Ok(Response::default())
                            // }
                            // (Method::PUT, path) if path.starts_with("/log-level") => {
                            //     Self::change_log_level(req).await
                            // }
                            (Method::GET, "/op/meta/compact_range") => {
                                operator.compact_range(req).await
                            }
                            (Method::GET, "/op/meta/compactions") => {
                                operator.compactions(req).await
                            }
                            (Method::GET, "/op/meta/shard_info") => operator.shard_info(req).await,
                            (Method::DELETE, "/op/meta/compactions") => {
                                operator.abort_compactions(req).await
                            }
                            (Method::GET, "/op/meta/reschedule_range") => {
                                operator.reschedule_range(req).await
                            }
                            (Method::GET, "/op/meta/split_shard") => {
                                operator.split_shard(req).await
                            }
                            (Method::GET, "/op/meta/gc") => operator.manual_gc(req).await,
                            (Method::GET, "/op/meta/reader/reschedule") => {
                                operator.reader_reschedule(req).await
                            }
                            (Method::GET, "/op/meta/reader/allocation/add") => {
                                operator.reader_allocation_add(req).await
                            }
                            (Method::GET, "/op/meta/reader/allocation/remove") => {
                                operator.reader_allocation_remove(req).await
                            }
                            (Method::GET, "/op/meta/reader/allocation/by_node") => {
                                operator.reader_allocation_by_node(req).await
                            }
                            (Method::GET, "/op/meta/worker/allocation/shards") => {
                                operator.worker_allocation_shards(req).await
                            }
                            (Method::GET, "/op/meta/import/inactive_tasks") => {
                                operator.inactive_import_tasks(req).await
                            }
                            (Method::POST, "/op/meta/worker/block") => {
                                operator.block_worker(req).await
                            }
                            (Method::GET, "/op/meta/worker/block") => {
                                operator.list_blocked_workers(req).await
                            }
                            (Method::DELETE, "/op/meta/worker/block") => {
                                operator.unblock_worker(req).await
                            }
                            (Method::POST, "/op/meta/reader/block") => {
                                operator.block_reader(req).await
                            }
                            (Method::GET, "/op/meta/reader/block") => {
                                operator.list_blocked_readers(req).await
                            }
                            (Method::DELETE, "/op/meta/reader/block") => {
                                operator.unblock_reader(req).await
                            }
                            _ => Ok(make_response(
                                StatusCode::NOT_FOUND,
                                format!("path not found, {:?}", req),
                            )),
                        }
                    }
                }))
            }
        }));

        let rx = self.rx.take().unwrap();
        let graceful = server
            .with_graceful_shutdown(async move {
                let _ = rx.await;
                log::info!("status server shutdown");
            })
            .map_err(|e| error!("Status server err={}", e));
        tokio::spawn(graceful);
    }

    pub fn start(&mut self, status_addr: SocketAddr) -> anyhow::Result<()> {
        let incoming = AddrIncoming::bind(&status_addr)?;
        self.addr = Some(incoming.local_addr());
        // TODO(P2) Support security
        let server = Server::builder(incoming);
        self.start_serve(server);
        Ok(())
    }
}

#[allow(unused)]
mod tests {
    use hyper::{Client, StatusCode, Uri, body};

    use super::*;
    async fn check_impl(authority: SocketAddr) -> Result<(), Box<dyn std::error::Error>> {
        let client = Client::new();
        let uri = Uri::builder()
            .scheme("http")
            .authority(authority.to_string().as_str())
            .path_and_query("/debug/pprof/profile?seconds=15")
            .build()?;
        let resp = client.get(uri.clone()).await?;
        let (parts, raw_body) = resp.into_parts();
        let body = body::to_bytes(raw_body).await?;
        assert_eq!(
            StatusCode::OK,
            parts.status,
            "{}",
            String::from_utf8(body.to_vec())?
        );
        assert_eq!("image/svg+xml", parts.headers["content-type"].to_str()?);
        {
            let xmldata = body.as_ref();
            let xmlstr = std::str::from_utf8(xmldata);
            assert!(xmlstr.is_ok());
        }
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn test_pprof() {
        let mut status_server = StatusServer::new(std::env::temp_dir(), Operators::None);

        status_server.start("127.0.0.1:0".parse().unwrap()).unwrap();
        let lis = status_server.listening_addr();
        let handle = tokio::task::spawn(async move {
            if let Err(e) = check_impl(lis).await {
                panic!("error in check_impl {:?}", e);
            }
        });
        let mut s: f64 = 0.0;
        use rand::Rng;
        let mut rng = rand::rng();
        for _ in 0..10000000 {
            s += rng.random::<i16>() as f64 * rng.random::<i16>() as f64;
        }
        log::info!("s={}", s);
        handle.await.unwrap();
        // status_server.stop();
    }

    async fn check_impl_heap(authority: SocketAddr) -> Result<(), Box<dyn std::error::Error>> {
        let client = Client::new();
        let uri = Uri::builder()
            .scheme("http")
            .authority(authority.to_string().as_str())
            .path_and_query("/debug/pprof/heap?jeprof=true&text=svg")
            .build()?;
        let resp = client.get(uri.clone()).await?;
        let (parts, raw_body) = resp.into_parts();
        let body = body::to_bytes(raw_body).await?;
        assert_eq!(
            StatusCode::OK,
            parts.status,
            "{}",
            String::from_utf8(body.to_vec())?
        );
        assert_eq!("image/svg+xml", parts.headers["content-type"].to_str()?);
        {
            let xmldata = body.as_ref();
            let xmlstr = std::str::from_utf8(xmldata);
            assert!(xmlstr.is_ok());
        }
        Ok(())
    }

    #[cfg(target_os = "linux")] // Only support linux for pprof heap test
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn test_heap() {
        let mut status_server = StatusServer::new(std::env::temp_dir(), Operators::None);

        assert_eq!(
            StatusServer::set_profile_active(true).unwrap().status(),
            StatusCode::OK
        );
        assert!(super::vendored_utils::has_activate_prof());
        status_server.start("127.0.0.1:0".parse().unwrap()).unwrap();
        std::thread::sleep(std::time::Duration::from_secs(1));
        let lis = status_server.listening_addr();
        let v = vec![1; 256 * 1024 * 1024];
        log::info!("v.len()={}", v.len());
        if let Err(e) = check_impl_heap(lis).await {
            panic!("error in check_impl {:?}", e);
        }
        StatusServer::set_profile_active(false);
    }
}