readyset-server: Add HTTP API endpoints for jemalloc profiling

Currently can only be enabled, disabled, and manually dumped; other
options like sample rate and dumping to a file at an interval can still
only be configured via the `MALLOC_CONF` environment variable (or
`_RJEM_MALLOC_CONF` on macOS).

Can be used like so:

```sh
$ curl -X POST localhost:6033/jemalloc/profiling/activate
Memory profiling activated
$ curl localhost:6033/jemalloc/profiling/dump > jeprof.heap
$ jeprof --svg target/release/readyset jeprof.heap > jeprof.heap.svg
$ curl -X POST localhost:6033/jemalloc/profiling/deactivate
Memory profiling deactivated
```

Change-Id: Ifc8efa3ccd67d01123b42e0588a45f2c26c947cd
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/8503
Reviewed-by: Jason Brown <[email protected]>
Tested-by: Buildkite CI

Author: mvzink

readyset-alloc/Cargo.toml

@@ -19,6 +19,7 @@ futures = { workspace = true, features = ["executor", "thread-pool"] }
 fxhash = { workspace = true }
 lazy_static = { workspace = true }
 libc = { workspace = true }
+tempfile = { workspace = true }
 tikv-jemalloc-ctl = { workspace = true }
 tikv-jemalloc-sys = { workspace = true, features = ["stats"] }
 tikv-jemallocator = { workspace = true, features = [

readyset-alloc/src/jemalloc.rs

@@ -131,7 +131,7 @@ pub fn remove_thread_memory_accessor() {
 
 use std::thread::ThreadId;
 
-pub use self::profiling::{activate_prof, deactivate_prof, dump_prof};
+pub use self::profiling::{activate_prof, deactivate_prof, dump_prof, dump_prof_to_string};
 
 /// Returns a very verbose output of jemalloc stats as well as per-thread stats
 pub fn dump_stats() -> Result<String, Error> {
@@ -325,7 +325,7 @@ mod tests {
 
 #[cfg(feature = "mem-profiling")]
 mod profiling {
-    use std::ffi::CString;
+    use std::{ffi::CString, os::unix::ffi::OsStrExt, path::Path};
 
     use libc::c_char;
 
@@ -360,26 +360,32 @@ mod profiling {
     }
 
     /// Dump the profile to the `path`.
-    pub fn dump_prof(path: &str) -> ProfResult<()> {
-        let mut bytes = CString::new(path)?.into_bytes_with_nul();
+    pub fn dump_prof(path: impl AsRef<Path>) -> ProfResult<()> {
+        let mut bytes = CString::new(path.as_ref().as_os_str().as_bytes())?.into_bytes_with_nul();
         let ptr = bytes.as_mut_ptr() as *mut c_char;
         unsafe {
             if let Err(e) = tikv_jemalloc_ctl::raw::write(PROF_DUMP, ptr) {
                 return Err(ProfError::JemallocError(format!(
                     "failed to dump the profile to {:?}: {}",
-                    path, e
+                    path.as_ref(),
+                    e
                 )));
             }
         }
         Ok(())
     }
 
+    pub async fn dump_prof_to_string() -> ProfResult<String> {
+        let tempdir = tempfile::Builder::new().prefix("jeprof").tempdir()?;
+        let path = tempdir.path().join("jeprof.out");
+        dump_prof(&path)?;
+        Ok(tokio::fs::read_to_string(path).await?)
+    }
+
     #[cfg(test)]
     mod tests {
         use std::fs;
 
-        use tempfile::Builder;
-
         const OPT_PROF: &[u8] = b"opt.prof\0";
 
         fn is_profiling_on() -> bool {
@@ -407,17 +413,15 @@ mod profiling {
             // Make sure somebody has turned on profiling
             assert!(is_profiling_on(), "set MALLOC_CONF=prof:true");
 
-            let dir = Builder::new()
+            let dir = tempfile::Builder::new()
                 .prefix("test_profiling_memory")
                 .tempdir()
                 .unwrap();
 
-            let os_path = dir.path().to_path_buf().join("test1.dump").into_os_string();
-            let path = os_path.into_string().unwrap();
+            let path = dir.path().join("test1.dump");
             super::dump_prof(&path).unwrap();
 
-            let os_path = dir.path().to_path_buf().join("test2.dump").into_os_string();
-            let path = os_path.into_string().unwrap();
+            let path = dir.path().join("test2.dump");
             super::dump_prof(&path).unwrap();
 
             let files = fs::read_dir(dir.path()).unwrap().count();
@@ -445,13 +449,16 @@ mod profiling {
 mod profiling {
     use super::{ProfError, ProfResult};
 
-    pub fn dump_prof(_path: &str) -> ProfResult<()> {
-        Err(ProfError::MemProfilingNotEnabled)
-    }
     pub fn activate_prof() -> ProfResult<()> {
         Err(ProfError::MemProfilingNotEnabled)
     }
     pub fn deactivate_prof() -> ProfResult<()> {
         Err(ProfError::MemProfilingNotEnabled)
     }
+    pub fn dump_prof(path: impl AsRef<Path>) -> ProfResult<()> {
+        Err(ProfError::MemProfilingNotEnabled)
+    }
+    pub async fn dump_prof_to_string() -> ProfResult<String> {
+        Err(ProfError::MemProfilingNotEnabled)
+    }
 }

readyset-server/src/http_router.rs

@@ -11,7 +11,10 @@ use health_reporter::{HealthReporter, State};
 use hyper::header::CONTENT_TYPE;
 use hyper::service::make_service_fn;
 use hyper::{self, Body, Method, Request, Response, StatusCode};
-use readyset_alloc::{dump_stats, print_memory_and_per_thread_stats};
+use readyset_alloc::{
+    activate_prof, deactivate_prof, dump_prof_to_string, dump_stats,
+    print_memory_and_per_thread_stats,
+};
 use readyset_client::metrics::recorded;
 use readyset_errors::ReadySetError;
 use readyset_util::shutdown::ShutdownReceiver;
@@ -112,7 +115,7 @@ impl Service<Request<Body>> for NoriaServerHttpRouter {
                     let contents = match bincode::deserialize(&body) {
                         Err(_) => {
                             return Ok(res
-                                .status(400)
+                                .status(StatusCode::BAD_REQUEST)
                                 .header(CONTENT_TYPE, "text/plain")
                                 .body(hyper::Body::from(
                                     "body cannot be deserialized into failpoint name and action",
@@ -123,7 +126,7 @@ impl Service<Request<Body>> for NoriaServerHttpRouter {
                     };
                     let (name, action): (String, String) = contents;
                     let resp = res
-                        .status(200)
+                        .status(StatusCode::OK)
                         .header(CONTENT_TYPE, "text/plain")
                         .body(hyper::Body::from(
                             ::bincode::serialize(&fail::cfg(name, &action)).unwrap(),
@@ -147,7 +150,7 @@ impl Service<Request<Body>> for NoriaServerHttpRouter {
                 let res = match render {
                     Some(metrics) => res.body(hyper::Body::from(metrics)),
                     None => res
-                        .status(404)
+                        .status(StatusCode::NOT_FOUND)
                         .body(hyper::Body::from("Prometheus metrics were not enabled. To fix this, run Noria with --prometheus-metrics".to_string())),
                 };
                 Box::pin(async move { Ok(res.unwrap()) })
@@ -158,11 +161,11 @@ impl Service<Request<Body>> for NoriaServerHttpRouter {
                     let body = format!("Server is in {} state", &state).into();
                     let res = match state {
                         State::Healthy | State::ShuttingDown => res
-                            .status(200)
+                            .status(StatusCode::OK)
                             .header(CONTENT_TYPE, "text/plain")
                             .body(body),
                         _ => res
-                            .status(500)
+                            .status(StatusCode::INTERNAL_SERVER_ERROR)
                             .header(CONTENT_TYPE, "text/plain")
                             .body(body),
                     };
@@ -177,7 +180,7 @@ impl Service<Request<Body>> for NoriaServerHttpRouter {
                         .header(CONTENT_TYPE, "application/json")
                         .body(hyper::Body::from(metrics)),
                     None => res
-                        .status(404)
+                        .status(StatusCode::NOT_FOUND)
                         .header(CONTENT_TYPE, "text/plain")
                         .body(hyper::Body::from("Noria metrics were not enabled. To fix this, run Noria with --noria-metrics".to_string())),
                 };
@@ -242,35 +245,87 @@ impl Service<Request<Body>> for NoriaServerHttpRouter {
             }
             // Returns a summary of memory usage for the entire process and per-thread memory usage
             (&Method::POST, "/memory_stats") => {
-                let res =
-                    match print_memory_and_per_thread_stats() {
-                        Ok(stats) => res
-                            .status(200)
-                            .header(CONTENT_TYPE, "text/plain")
-                            .body(hyper::Body::from(stats)),
-                        Err(e) => res.status(500).header(CONTENT_TYPE, "text/plain").body(
-                            hyper::Body::from(format!("Error fetching memory stats: {e}")),
-                        ),
-                    };
+                let res = match print_memory_and_per_thread_stats() {
+                    Ok(stats) => res
+                        .status(StatusCode::OK)
+                        .header(CONTENT_TYPE, "text/plain")
+                        .body(hyper::Body::from(stats)),
+                    Err(e) => res
+                        .status(StatusCode::INTERNAL_SERVER_ERROR)
+                        .header(CONTENT_TYPE, "text/plain")
+                        .body(hyper::Body::from(format!(
+                            "Error fetching memory stats: {e}"
+                        ))),
+                };
 
                 Box::pin(async move { Ok(res.unwrap()) })
             }
             // Returns a large dump of jemalloc debugging information along with per-thread
             // memory stats
             (&Method::POST, "/memory_stats_verbose") => {
-                let res =
-                    match dump_stats() {
-                        Ok(stats) => res
-                            .status(200)
-                            .header(CONTENT_TYPE, "text/plain")
-                            .body(hyper::Body::from(stats)),
-                        Err(e) => res.status(500).header(CONTENT_TYPE, "text/plain").body(
-                            hyper::Body::from(format!("Error fetching memory stats: {e}")),
-                        ),
-                    };
+                let res = match dump_stats() {
+                    Ok(stats) => res
+                        .status(StatusCode::OK)
+                        .header(CONTENT_TYPE, "text/plain")
+                        .body(hyper::Body::from(stats)),
+                    Err(e) => res
+                        .status(StatusCode::INTERNAL_SERVER_ERROR)
+                        .header(CONTENT_TYPE, "text/plain")
+                        .body(hyper::Body::from(format!(
+                            "Error fetching memory stats: {e}"
+                        ))),
+                };
 
                 Box::pin(async move { Ok(res.unwrap()) })
             }
+            // Turns on jemalloc's profiler
+            (&Method::POST, "/jemalloc/profiling/activate") => {
+                let res = match activate_prof() {
+                    Ok(_) => res
+                        .status(StatusCode::OK)
+                        .header(CONTENT_TYPE, "text/plain")
+                        .body(hyper::Body::from("Memory profiling activated")),
+                    Err(e) => res
+                        .status(StatusCode::INTERNAL_SERVER_ERROR)
+                        .header(CONTENT_TYPE, "text/plain")
+                        .body(hyper::Body::from(format!(
+                            "Error activating memory profiling: {e}"
+                        ))),
+                };
+                Box::pin(async move { Ok(res.unwrap()) })
+            }
+            // Disables jemalloc's profiler
+            (&Method::POST, "/jemalloc/profiling/deactivate") => {
+                let res = match deactivate_prof() {
+                    Ok(_) => res
+                        .status(StatusCode::OK)
+                        .header(CONTENT_TYPE, "text/plain")
+                        .body(hyper::Body::from("Memory profiling deactivated")),
+                    Err(e) => res
+                        .status(StatusCode::INTERNAL_SERVER_ERROR)
+                        .header(CONTENT_TYPE, "text/plain")
+                        .body(hyper::Body::from(format!(
+                            "Error deactivating memory profiling: {e}"
+                        ))),
+                };
+                Box::pin(async move { Ok(res.unwrap()) })
+            }
+            // Returns the current jemalloc profiler output
+            (&Method::GET, "/jemalloc/profiling/dump") => Box::pin(async move {
+                let res = match dump_prof_to_string().await {
+                    Ok(dump) => res
+                        .status(StatusCode::OK)
+                        .header(CONTENT_TYPE, "text/plain")
+                        .body(hyper::Body::from(dump)),
+                    Err(e) => res
+                        .status(StatusCode::INTERNAL_SERVER_ERROR)
+                        .header(CONTENT_TYPE, "text/plain")
+                        .body(hyper::Body::from(format!(
+                            "Error dumping profiling output: {e}"
+                        ))),
+                };
+                Ok(res.unwrap())
+            }),
             _ => {
                 metrics::counter!(recorded::SERVER_CONTROLLER_REQUESTS).increment(1);