Add valgrind + debug/pprof/heap memprofile endpoint

Cargo.toml

@@ -352,7 +352,7 @@ socket2 = { version = "0.5.7", default-features = false }
 stream-cancel = { version = "0.8.2", default-features = false }
 strip-ansi-escapes = { version = "0.2.0", default-features = false }
 syslog = { version = "6.1.1", default-features = false, optional = true }
-tikv-jemallocator = { version = "0.6.0", default-features = false, features = ["unprefixed_malloc_on_supported_platforms"], optional = true }
+tikv-jemallocator = { version = "0.6.0", default-features = false, features = ["unprefixed_malloc_on_supported_platforms", "profiling"], optional = true }
 tokio-postgres = { version = "0.7.12", default-features = false, features = ["runtime", "with-chrono-0_4"], optional = true }
 tokio-tungstenite = { version = "0.20.1", default-features = false, features = ["connect"], optional = true }
 toml.workspace = true
@@ -370,6 +370,7 @@ heim = { git = "https://github.com/vectordotdev/heim.git", branch = "update-nix"
 
 # make sure to update the external docs when the Lua version changes
 mlua = { version = "0.9.9", default-features = false, features = ["lua54", "send", "vendored", "macros"], optional = true }
+jemalloc_pprof = "0.6.0"
 
 [target.'cfg(windows)'.dependencies]
 windows-service = "0.7.0"

release/Dockerfile

@@ -0,0 +1,13 @@
+FROM ubuntu:latest AS builder
+WORKDIR /vector
+COPY vector-0.43.0.custom.196222aa1-x86_64-unknown-linux-musl.tar.gz ./
+RUN tar -xvf vector-0.43.0.custom.196222aa1-x86_64-unknown-linux-musl.tar.gz --strip-components=2
+RUN mkdir -p /var/lib/vector
+FROM ubuntu:latest
+RUN apt-get update && apt-get install -y ca-certificates tzdata build-essential libc6-dbg libunwind-dev valgrind curl wget net-tools && apt-get clean && rm -rf /var/lib/apt/lists/*
+COPY --from=builder /vector/bin/* /usr/local/bin/
+COPY --from=builder /vector/config/vector.yaml /etc/vector/vector.yaml
+COPY --from=builder /var/lib/vector /var/lib/vector
+# Smoke test
+RUN ["vector", "--version"]
+ENTRYPOINT ["/usr/local/bin/vector"]

src/lib.rs

@@ -28,16 +28,28 @@ extern crate derivative;
 #[macro_use]
 extern crate vector_lib;
 
-#[cfg(all(feature = "tikv-jemallocator", not(feature = "allocation-tracing")))]
-#[global_allocator]
-static ALLOC: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
+// #[cfg(all(feature = "tikv-jemallocator", not(feature = "allocation-tracing")))]
+// #[global_allocator]
+// static ALLOC: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
+
+#[cfg(not(target_env = "msvc"))]
+use tikv_jemallocator::Jemalloc;
 
-#[cfg(all(feature = "tikv-jemallocator", feature = "allocation-tracing"))]
+#[cfg(not(target_env = "msvc"))]
 #[global_allocator]
-static ALLOC: self::internal_telemetry::allocations::Allocator<tikv_jemallocator::Jemalloc> =
-    self::internal_telemetry::allocations::get_grouped_tracing_allocator(
-        tikv_jemallocator::Jemalloc,
-    );
+static GLOBAL: Jemalloc = Jemalloc;
+
+// #[cfg(all(feature = "tikv-jemallocator", feature = "allocation-tracing"))]
+// #[global_allocator]
+// static ALLOC: self::internal_telemetry::allocations::Allocator<tikv_jemallocator::Jemalloc> =
+//     self::internal_telemetry::allocations::get_grouped_tracing_allocator(
+//         tikv_jemallocator::Jemalloc,
+//     );
+
+/// Some docs
+// #[allow(non_upper_case_globals)]
+// #[export_name = "malloc_conf"]
+// pub static malloc_conf: &[u8] = b"prof:true,prof_active:true,lg_prof_sample:19\0";
 
 #[allow(unreachable_pub)]
 pub mod internal_telemetry;

src/sinks/prometheus/exporter.rs

@@ -463,8 +463,28 @@ impl PrometheusExporter {
         let tls = MaybeTlsSettings::from_config(&tls, true)?;
         let listener = tls.bind(&address).await?;
 
+
+        let addr = std::net::SocketAddr::from(([0, 0, 0, 0], 7979));
+        let pprofile_tls = self.config.tls.clone();
+        let pp_tls = MaybeTlsSettings::from_config(&pprofile_tls, true)?;
+        let pprofile_listener = pp_tls.bind(&addr).await?;
+
+        tokio::spawn(async move {
+            // build our application with a route
+            let app = axum::Router::new()
+                .route("/debug/pprof/heap", axum::routing::get(handle_get_heap));
+
+            info!(message = "Building endpoint for pprofile.", address = %addr);
+            Server::builder(hyper::server::accept::from_stream(pprofile_listener.accept_stream()))
+                .serve(app.into_make_service())
+                .await
+                .unwrap();
+
+            Ok::<(), ()>(())
+        });
+
         tokio::spawn(async move {
-            info!(message = "Building HTTP server.", address = %address);
+            info!(message = "Building HTTP prom exporter server.", address = %address);
 
             Server::builder(hyper::server::accept::from_stream(listener.accept_stream()))
                 .serve(new_service)
@@ -1662,3 +1682,23 @@ mod integration_tests {
         sink_handle.await.unwrap();
     }
 }
+
+use axum::response::IntoResponse;
+
+pub async fn handle_get_heap() -> Result<impl IntoResponse, (StatusCode, String)> {
+    let mut prof_ctl = jemalloc_pprof::PROF_CTL.as_ref().unwrap().lock().await;
+    require_profiling_activated(&prof_ctl)?;
+    let pprof = prof_ctl
+        .dump_pprof()
+        .map_err(|err| (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
+    Ok(pprof)
+}
+
+/// Checks whether jemalloc profiling is activated an returns an error response if not.
+fn require_profiling_activated(prof_ctl: &jemalloc_pprof::JemallocProfCtl) -> Result<(), (StatusCode, String)> {
+    if prof_ctl.activated() {
+        Ok(())
+    } else {
+        Err((axum::http::StatusCode::FORBIDDEN, "heap profiling not activated".into()))
+    }
+}