feat: add opt-in hotpath profiling (#5488)

* feat: add opt-in hotpath profiling

Co-Authored-By: heihutu <heihutu@gmail.com>

* test: fix vault kms client construction

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>

Cargo.lock

@@ -4916,19 +4916,28 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "66750a77f4f6b408a148be5102ef1f3ba7172def7ee92b1cfc75d9f7a3870453"
 dependencies = [
  "arc-swap",
+ "async-channel",
+ "async-trait",
  "cfg-if",
  "crossbeam-channel",
+ "futures-channel",
  "futures-util",
  "hdrhistogram",
  "hotpath-macros",
+ "hotpath-meta",
+ "http 1.5.0",
  "libc",
+ "parking_lot",
  "pin-project-lite",
  "prettytable-rs",
  "quanta",
  "regex",
+ "reqwest",
+ "reqwest-middleware",
  "serde",
  "serde_json",
  "tiny_http",
+ "tokio",
 ]
 
 [[package]]
@@ -4942,6 +4951,21 @@ dependencies = [
  "syn 2.0.119",
 ]
 
+[[package]]
+name = "hotpath-macros-meta"
+version = "0.22.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "21f2f70b29b6f42311acd2fb0b2f91a34d9a573c76fb8a7f51970670a1673a49"
+
+[[package]]
+name = "hotpath-meta"
+version = "0.22.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b771f77b2f409086bb40ff029f850ce99d17118d4e207df0f28cb32bd7568c7"
+dependencies = [
+ "hotpath-macros-meta",
+]
+
 [[package]]
 name = "htmlescape"
 version = "0.3.1"
@@ -8514,6 +8538,21 @@ dependencies = [
  "web-sys",
 ]
 
+[[package]]
+name = "reqwest-middleware"
+version = "0.5.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "07bc3f1384cffa4f274dad2d4ddd73aed32fed8f786d96c6be8aa4e5fd3c3b58"
+dependencies = [
+ "anyhow",
+ "async-trait",
+ "http 1.5.0",
+ "reqwest",
+ "serde",
+ "thiserror 2.0.19",
+ "tower-service",
+]
+
 [[package]]
 name = "resolv-conf"
 version = "0.7.6"

Cargo.toml

@@ -348,7 +348,7 @@ dav-server = "0.11.0"
 
 # Performance Analysis and Memory Profiling
 mimalloc = "0.1.52"
-hotpath = "0.22.0"
+hotpath = { version = "0.22.0", default-features = false }
 # Snapshot testing for output format regression detection
 insta = { version = "1.48" }
 

crates/ecstore/Cargo.toml

@@ -33,14 +33,28 @@ workspace = true
 [features]
 default = []
 rio-v2 = ["dep:rustfs-rio-v2"]
-hotpath = ["dep:hotpath", "hotpath/hotpath", "rustfs-filemeta/hotpath", "rustfs-rio/hotpath"]
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "hotpath/futures",
+    "hotpath/async-channel",
+    "hotpath/parking_lot",
+    "hotpath/reqwest-0-13",
+    "rustfs-filemeta/hotpath",
+    "rustfs-rio/hotpath",
+]
+hotpath-alloc = [
+    "hotpath/hotpath-alloc",
+    "rustfs-filemeta/hotpath-alloc",
+    "rustfs-rio/hotpath-alloc",
+]
 # Exposes shared lifecycle/tier test utilities (MockWarmBackend, fault
 # injection, xl.meta transition assertions) via `api::tier::test_util`.
 # Enable only from `[dev-dependencies]` (rustfs/backlog#1148 ilm-6).
 test-util = []
 
 [dependencies]
-hotpath = { workspace = true, optional = true }
+hotpath.workspace = true
 rustfs-filemeta.workspace = true
 rustfs-utils = { workspace = true, features = ["full"] }
 rustfs-rio.workspace = true

crates/ecstore/src/disk/local.rs

@@ -5078,7 +5078,7 @@ impl LocalDisk {
         Ok((buf, mtime))
     }
 
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     async fn read_metadata_with_dmtime(&self, file_path: impl AsRef<Path>) -> Result<(Vec<u8>, Option<OffsetDateTime>)> {
         check_path_length(file_path.as_ref().to_string_lossy().as_ref())?;
 
@@ -5121,7 +5121,7 @@ impl LocalDisk {
         Ok((data, modtime))
     }
 
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     async fn read_all_data(&self, volume: &str, volume_dir: impl AsRef<Path>, file_path: impl AsRef<Path>) -> Result<Vec<u8>> {
         // TODO: timeout support
         let (data, _) = self.read_all_data_with_dmtime(volume, volume_dir, file_path).await?;

crates/ecstore/src/erasure/coding/bitrot.rs

@@ -103,7 +103,7 @@ where
     /// or `out` is larger than one shard. On error `out`'s contents are
     /// unspecified but never contain bytes that failed the hash check — the copy
     /// happens only after verification.
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     pub async fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
         let want = out.len();
         self.begin_read(want)?;
@@ -303,7 +303,7 @@ where
 
     /// Write a (hash+data) block. Returns the number of data bytes written.
     /// Returns an error if called after a short write or if data exceeds shard_size.
-    #[cfg_attr(feature = "hotpath", hotpath::measure(label = "BitrotWriter::write"))]
+    #[hotpath::measure(label = "BitrotWriter::write")]
     pub async fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
         if buf.is_empty() {
             return Ok(0);
@@ -455,7 +455,7 @@ pub fn bitrot_shard_file_size(size: usize, shard_size: usize, algo: HashAlgorith
 /// stores those as whole-file bitrot with no interleaved hash, so the size guard
 /// on the next line would reject a genuinely healthy part. Reading legacy V1
 /// whole-file-bitrot objects would need a separate verification path.
-#[cfg_attr(feature = "hotpath", hotpath::measure)]
+#[hotpath::measure]
 pub async fn bitrot_verify<R: AsyncRead + Unpin + Send>(
     mut r: R,
     want_size: usize,

crates/ecstore/src/erasure/coding/decode.rs

@@ -691,7 +691,7 @@ impl<R> ParallelReader<R>
 where
     R: crate::erasure::coding::ShardSource,
 {
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     pub async fn read(&mut self) -> (Vec<Option<Vec<u8>>>, Vec<Option<Error>>) {
         // On the reconstruction-verifying GET path, read every live shard reader
         // in lockstep so all readers advance one block per stripe and stay
@@ -1505,7 +1505,7 @@ where
 }
 
 impl Erasure {
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     pub async fn decode<W, R>(
         &self,
         writer: &mut W,

crates/ecstore/src/erasure/coding/encode.rs

@@ -504,7 +504,7 @@ impl Erasure {
         Ok((reader, total))
     }
 
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     pub async fn encode<R>(
         self: Arc<Self>,
         reader: R,
@@ -670,7 +670,7 @@ impl Erasure {
         Ok((reader, total))
     }
 
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     pub async fn encode_batched<R>(
         self: Arc<Self>,
         mut reader: R,
@@ -798,7 +798,7 @@ impl Erasure {
 
     /// Fast path for small inline objects: skip tokio::spawn + mpsc channel.
     /// Reads all data, encodes directly, writes shards sequentially.
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     pub async fn encode_inline_small<R>(
         self: Arc<Self>,
         reader: R,
@@ -813,7 +813,7 @@ impl Erasure {
 
     /// Fast path for single-block non-inline objects: avoids the producer/consumer
     /// pipeline in `encode()` while keeping the same writer/quorum/shutdown semantics.
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     pub async fn encode_single_block_non_inline<R>(
         self: Arc<Self>,
         reader: R,

crates/ecstore/src/erasure/coding/erasure.rs

@@ -640,7 +640,7 @@ impl Erasure {
     /// # Returns
     /// A vector of encoded shards as `Bytes`.
     #[tracing::instrument(level = "debug", skip_all, fields(data_len=data.len()))]
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     pub fn encode_data(&self, data: &[u8]) -> io::Result<Vec<Bytes>> {
         let shard_size_fn = if self.uses_legacy {
             calc_shard_size_legacy
@@ -688,7 +688,7 @@ impl Erasure {
 
     /// Encode owned data, avoiding a copy when the caller already has a heap buffer.
     /// Falls back to copying into a new buffer if zero-copy conversion fails.
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     pub fn encode_data_owned(&self, data: Vec<u8>) -> io::Result<Vec<Bytes>> {
         let shard_size_fn = if self.uses_legacy {
             calc_shard_size_legacy
@@ -752,7 +752,7 @@ impl Erasure {
     /// block), the `resize(need_total_size)` below stays within capacity for every
     /// `data_len <= block_size` — both shard-size formulas are monotone in
     /// `data_len` — so this function never reallocates the buffer.
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     pub fn encode_data_bytes_mut(&self, mut data_buffer: BytesMut, data_len: usize) -> io::Result<Vec<Bytes>> {
         let shard_size_fn = if self.uses_legacy {
             calc_shard_size_legacy
@@ -805,7 +805,7 @@ impl Erasure {
     ///
     /// # Returns
     /// Ok if reconstruction succeeds, error otherwise.
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     pub fn decode_data(&self, shards: &mut [Option<Vec<u8>>]) -> io::Result<()> {
         if self.parity_shards > 0 {
             if self.uses_legacy {
@@ -825,7 +825,7 @@ impl Erasure {
     }
 
     /// Decode and reconstruct missing data shards, then regenerate parity shards.
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     pub fn decode_data_and_parity(&self, shards: &mut [Option<Vec<u8>>]) -> io::Result<()> {
         if self.parity_shards > 0 {
             if self.uses_legacy {

crates/ecstore/src/lib.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 /// Scope-based hotpath measurement for `#[async_trait]` methods, where
-/// `#[cfg_attr(feature = "hotpath", hotpath::measure)]` would only time the boxed-future construction.
+/// `#[hotpath::measure]` would only time the boxed-future construction.
 /// The guard records wall time from this statement until the enclosing
 /// (desugared) async block completes, including early returns via `?`.
 #[cfg(feature = "hotpath")]

crates/ecstore/src/set_disk/read.rs

@@ -199,7 +199,7 @@ impl SetDisks {
         );
     }
 
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     pub async fn read_version_optimized(
         &self,
         bucket: &str,
@@ -238,7 +238,7 @@ impl SetDisks {
     }
 
     #[tracing::instrument(level = "debug", skip(self))]
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     pub(super) async fn get_object_fileinfo(
         &self,
         bucket: &str,
@@ -410,7 +410,7 @@ impl SetDisks {
         Ok((fi, parts_metadata, op_online_disks))
     }
 
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     pub(super) async fn get_object_info_and_quorum(
         &self,
         bucket: &str,
@@ -605,7 +605,7 @@ impl SetDisks {
     }
 
     #[allow(clippy::too_many_arguments)]
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     pub(super) async fn get_object_with_fileinfo<W>(
         // &self,
         bucket: &str,
@@ -1140,7 +1140,7 @@ impl SetDisks {
     }
 
     #[allow(clippy::too_many_arguments)]
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     pub(super) async fn get_object_decode_reader_with_fileinfo(
         bucket: &str,
         object: &str,
@@ -1296,7 +1296,7 @@ impl SetDisks {
     }
 
     #[allow(clippy::too_many_arguments)]
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     async fn build_codec_streaming_part_reader(
         bucket: &str,
         object: &str,
@@ -1469,7 +1469,7 @@ fn multipart_part_checksum_algo(fi: &FileInfo, part_number: usize) -> HashAlgori
 /// `get_object_with_fileinfo` (backlog#870) so both report the same
 /// stage-duration semantics.
 #[allow(clippy::too_many_arguments)]
-#[cfg_attr(feature = "hotpath", hotpath::measure)]
+#[hotpath::measure]
 async fn setup_multipart_part_readers(
     files: &[FileInfo],
     disks: &[Option<DiskStore>],

crates/ecstore/src/store/multipart.rs

@@ -238,7 +238,7 @@ impl ECStore {
     }
 
     #[instrument(skip(self, data))]
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     pub(super) async fn handle_put_object_part(
         &self,
         bucket: &str,

crates/ecstore/src/store/object.rs

@@ -815,7 +815,7 @@ impl ECStore {
     }
 
     #[instrument(level = "debug", skip(self))]
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     pub(super) async fn handle_get_object_reader(
         &self,
         bucket: &str,
@@ -849,7 +849,7 @@ impl ECStore {
     }
 
     #[instrument(level = "debug", skip(self, data))]
-    #[cfg_attr(feature = "hotpath", hotpath::measure)]
+    #[hotpath::measure]
     pub(super) async fn handle_put_object(
         &self,
         bucket: &str,

crates/filemeta/Cargo.toml

@@ -27,10 +27,11 @@ documentation = "https://docs.rs/rustfs-filemeta/latest/rustfs_filemeta/"
 
 [features]
 default = []
-hotpath = ["dep:hotpath", "hotpath/hotpath"]
+hotpath = ["hotpath/hotpath", "hotpath/tokio"]
+hotpath-alloc = ["hotpath/hotpath-alloc"]
 
 [dependencies]
-hotpath = { workspace = true, optional = true }
+hotpath.workspace = true
 crc-fast = { workspace = true }
 rmp.workspace = true
 rmp-serde.workspace = true

crates/filemeta/src/filemeta/codec.rs

@@ -19,7 +19,7 @@ impl FileMeta {
         !matches!(Self::check_xl2_v1(buf), Err(_e))
     }
 
-    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "FileMeta"))]
+    #[hotpath::measure(impl_type = "FileMeta")]
     pub fn load(buf: &[u8]) -> Result<FileMeta> {
         let mut xl = FileMeta::default();
         xl.unmarshal_msg(buf)?;
@@ -112,7 +112,7 @@ impl FileMeta {
         Ok((bin_len, remaining))
     }
 
-    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "FileMeta"))]
+    #[hotpath::measure(impl_type = "FileMeta")]
     pub fn unmarshal_msg(&mut self, buf: &[u8]) -> Result<u64> {
         let i = buf.len() as u64;
 
@@ -326,7 +326,7 @@ impl FileMeta {
         }
     }
 
-    #[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "FileMeta"))]
+    #[hotpath::measure(impl_type = "FileMeta")]
     pub fn marshal_msg(&self) -> Result<Vec<u8>> {
         let mut wr = Vec::new();
 

crates/rio/Cargo.toml

@@ -30,10 +30,11 @@ workspace = true
 
 [features]
 default = []
-hotpath = ["dep:hotpath", "hotpath/hotpath"]
+hotpath = ["hotpath/hotpath", "hotpath/tokio", "hotpath/futures", "hotpath/reqwest-0-13"]
+hotpath-alloc = ["hotpath/hotpath-alloc"]
 
 [dependencies]
-hotpath = { workspace = true, optional = true }
+hotpath.workspace = true
 arc-swap.workspace = true
 tokio = { workspace = true, features = ["io-util", "macros", "net", "rt-multi-thread", "sync", "time"] }
 rand = { workspace = true, features = ["serde"] }

rustfs/Cargo.toml

@@ -53,18 +53,29 @@ pyroscope = ["rustfs-obs/pyroscope"]
 # Tokio runtime telemetry. Requires `--cfg tokio_unstable`; use `make build-profiling`.
 dial9 = ["rustfs-obs/dial9"]
 hotpath = [
-    "dep:hotpath",
     "hotpath/hotpath",
+    "hotpath/tokio",
+    "hotpath/futures",
+    "hotpath/async-channel",
+    "hotpath/crossbeam",
+    "hotpath/parking_lot",
+    "hotpath/reqwest-0-13",
     "rustfs-ecstore/hotpath",
     "rustfs-filemeta/hotpath",
     "rustfs-rio/hotpath",
 ]
+hotpath-alloc = [
+    "hotpath/hotpath-alloc",
+    "rustfs-ecstore/hotpath-alloc",
+    "rustfs-filemeta/hotpath-alloc",
+    "rustfs-rio/hotpath-alloc",
+]
 
 [lints]
 workspace = true
 
 [dependencies]
-hotpath = { workspace = true, optional = true }
+hotpath.workspace = true
 # RustFS Internal Crates
 rustfs-heal = { workspace = true }
 rustfs-audit = { workspace = true }

rustfs/src/main.rs

@@ -12,23 +12,16 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#[cfg(all(feature = "hotpath", feature = "hotpath-alloc"))]
+#[global_allocator]
+static GLOBAL: hotpath::CountingAllocator = hotpath::CountingAllocator::new();
+
+#[cfg(not(all(feature = "hotpath", feature = "hotpath-alloc")))]
 #[global_allocator]
 static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
 
 fn main() {
-    // Wall-time profiling guard for `--features hotpath` builds. The guard
-    // lives for the whole process; dropping it on normal exit (SIGINT/SIGTERM
-    // unwind back through run_process) prints the timing report.
-    //
-    // hotpath's alloc mode is intentionally NOT wired up: hotpath 0.21.x
-    // panics when a MeasurementGuardSync drops on a different thread than it
-    // was created on (TLS index mismatch, alloc/core.rs:151), which tokio's
-    // work-stealing runtime does constantly. See rustfs/backlog#935.
-    #[cfg(feature = "hotpath")]
-    let _hotpath_guard = hotpath::HotpathGuardBuilder::new("main")
-        .percentiles(&[50.0, 95.0, 99.0])
-        .functions_limit(0)
-        .build();
+    let _hotpath_guard = hotpath::HotpathGuardBuilder::new("main").build();
 
     rustfs::startup_entrypoint::run_process();
 }

rustfs/src/startup_entrypoint.rs

@@ -62,6 +62,8 @@ fn emit_fatal_stderr(context: &str, error: impl std::fmt::Display) {
 }
 
 async fn async_main() -> Result<()> {
+    hotpath::tokio_runtime!();
+
     let env_compat_report = bootstrap_external_prefix_compat()?;
 
     // Parse command line arguments