From d5c6ba99d5b5ec961fc6ca075ef695245cf338d0 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 1 Aug 2026 16:49:48 +0800 Subject: [PATCH] fix(ecstore): harden HotPath profiling boundaries (#5555) * test(hotpath): gate mimalloc heap test by platform Co-Authored-By: heihutu * fix(ecstore): attribute HotPath CPU measurements to impls Co-Authored-By: heihutu * feat(ecstore): trace raw shard I/O with HotPath Co-Authored-By: heihutu * fix(obs): redact profiler and OPA endpoint diagnostics Co-Authored-By: heihutu * fix(ecstore): settle encoded queue accounting Co-Authored-By: heihutu * fix(ecstore): abort encoder producer on cancellation Co-Authored-By: heihutu --------- Co-authored-by: heihutu --- Cargo.lock | 3 + crates/ecstore/src/disk/local.rs | 4 +- crates/ecstore/src/erasure/coding/bitrot.rs | 4 +- crates/ecstore/src/erasure/coding/decode.rs | 4 +- crates/ecstore/src/erasure/coding/encode.rs | 590 ++++++++++++++++-- crates/ecstore/src/erasure/coding/erasure.rs | 10 +- crates/ecstore/src/io_support/bitrot.rs | 431 ++++++++++++- crates/ecstore/src/set_disk/read.rs | 12 +- crates/ecstore/src/store/multipart.rs | 2 +- crates/ecstore/src/store/object.rs | 4 +- .../hotpath_cpu_attribution_contract_test.rs | 95 +++ crates/iam/src/sys.rs | 8 +- crates/io-metrics/src/lib.rs | 10 +- crates/obs/Cargo.toml | 2 + crates/obs/src/telemetry/filter.rs | 21 +- crates/obs/src/telemetry/local.rs | 4 +- crates/obs/src/telemetry/otel.rs | 185 +++++- crates/policy/Cargo.toml | 2 +- crates/policy/src/policy/opa.rs | 229 ++++++- 19 files changed, 1480 insertions(+), 140 deletions(-) create mode 100644 crates/ecstore/tests/hotpath_cpu_attribution_contract_test.rs diff --git a/Cargo.lock b/Cargo.lock index 9d70a14e8..8efefa02f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9813,6 +9813,7 @@ dependencies = [ "hotpath", "jiff", "libc", + "log", "metrics", "num_cpus", "nvml-wrapper", @@ -9848,6 +9849,7 @@ dependencies = [ "tracing-error", "tracing-opentelemetry", "tracing-subscriber", + "url", "zstd", ] @@ -9879,6 +9881,7 @@ dependencies = [ "time", "tokio", "tracing", + "tracing-subscriber", ] [[package]] diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 00993e432..42b37f9e8 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -5078,7 +5078,7 @@ impl LocalDisk { Ok((buf, mtime)) } - #[hotpath::measure] + #[hotpath::measure(impl_type = "LocalDisk")] async fn read_metadata_with_dmtime(&self, file_path: impl AsRef) -> Result<(Vec, Option)> { check_path_length(file_path.as_ref().to_string_lossy().as_ref())?; @@ -5121,7 +5121,7 @@ impl LocalDisk { Ok((data, modtime)) } - #[hotpath::measure] + #[hotpath::measure(impl_type = "LocalDisk")] async fn read_all_data(&self, volume: &str, volume_dir: impl AsRef, file_path: impl AsRef) -> Result> { // TODO: timeout support let (data, _) = self.read_all_data_with_dmtime(volume, volume_dir, file_path).await?; diff --git a/crates/ecstore/src/erasure/coding/bitrot.rs b/crates/ecstore/src/erasure/coding/bitrot.rs index 006c9dea3..49b1f611c 100644 --- a/crates/ecstore/src/erasure/coding/bitrot.rs +++ b/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. - #[hotpath::measure] + #[hotpath::measure(impl_type = "BitrotReader")] pub async fn read(&mut self, out: &mut [u8]) -> std::io::Result { 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. - #[hotpath::measure(label = "BitrotWriter::write")] + #[hotpath::measure(label = "BitrotWriter::write", impl_type = "BitrotWriter")] pub async fn write(&mut self, buf: &[u8]) -> std::io::Result { if buf.is_empty() { return Ok(0); diff --git a/crates/ecstore/src/erasure/coding/decode.rs b/crates/ecstore/src/erasure/coding/decode.rs index daf27c93a..20eb535c4 100644 --- a/crates/ecstore/src/erasure/coding/decode.rs +++ b/crates/ecstore/src/erasure/coding/decode.rs @@ -691,7 +691,7 @@ impl ParallelReader where R: crate::erasure::coding::ShardSource, { - #[hotpath::measure] + #[hotpath::measure(impl_type = "ParallelReader")] pub async fn read(&mut self) -> (Vec>>, Vec>) { // 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 { - #[hotpath::measure] + #[hotpath::measure(impl_type = "Erasure")] pub async fn decode( &self, writer: &mut W, diff --git a/crates/ecstore/src/erasure/coding/encode.rs b/crates/ecstore/src/erasure/coding/encode.rs index c4414723c..db960c3ac 100644 --- a/crates/ecstore/src/erasure/coding/encode.rs +++ b/crates/ecstore/src/erasure/coding/encode.rs @@ -28,8 +28,11 @@ use std::vec; use tokio::io::AsyncRead; use tokio::runtime::RuntimeFlavor; use tokio::sync::mpsc; +use tokio::task::{JoinError, JoinHandle}; use tracing::error; +/// Queue-capacity input for encoded blocks awaiting shard writers; it is not a +/// per-PUT or process-RSS memory limit. const ENV_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES: &str = "RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES"; const ENV_RUSTFS_ERASURE_ENCODE_BATCH_BLOCKS: &str = "RUSTFS_ERASURE_ENCODE_BATCH_BLOCKS"; const ENV_RUSTFS_ERASURE_ENCODE_BYTESMUT_INGEST: &str = "RUSTFS_ERASURE_ENCODE_BYTESMUT_INGEST"; @@ -87,6 +90,33 @@ fn use_bytesmut_ingest() -> bool { rustfs_utils::get_env_bool(ENV_RUSTFS_ERASURE_ENCODE_BYTESMUT_INGEST, DEFAULT_RUSTFS_ERASURE_ENCODE_BYTESMUT_INGEST) }) } + +/// Keeps the encoder producer scoped to its parent future. Tokio detaches a +/// task when its `JoinHandle` is dropped, so the producer must be aborted when +/// an upload is cancelled before the encode pipeline finishes. +struct AbortOnDropTask(JoinHandle); + +impl AbortOnDropTask { + fn new(task: JoinHandle) -> Self { + Self(task) + } + + async fn abort_and_wait(&mut self) { + self.0.abort(); + let _ = (&mut self.0).await; + } + + async fn join(&mut self) -> Result { + (&mut self.0).await + } +} + +impl Drop for AbortOnDropTask { + fn drop(&mut self) { + self.0.abort(); + } +} + /// Read up to `limit` bytes into `buf`'s uninitialized spare capacity, appending after its /// current length, and distinguish a clean EOF from a short read. /// @@ -133,22 +163,65 @@ fn queued_block_bytes(block: &[Bytes]) -> usize { block.iter().map(Bytes::len).sum() } -async fn drain_queued_inflight_bytes(rx: &mut mpsc::Receiver>) { - while let Some(block) = rx.recv().await { - rustfs_io_metrics::remove_ec_encode_inflight_bytes(queued_block_bytes(&block)); +/// Owns an encoded queue entry's gauge contribution until its consumer takes it. +struct QueuedInflightBytes { + bytes: usize, +} + +impl QueuedInflightBytes { + fn new(bytes: usize) -> Self { + rustfs_io_metrics::add_ec_encode_inflight_bytes(bytes); + Self { bytes } } + + fn settle(&mut self) { + let bytes = std::mem::take(&mut self.bytes); + if bytes != 0 { + rustfs_io_metrics::remove_ec_encode_inflight_bytes(bytes); + } + } +} + +impl Drop for QueuedInflightBytes { + fn drop(&mut self) { + self.settle(); + } +} + +/// Couples an encoded block with its queue gauge contribution. Keeping the +/// guard in the queue entry also covers Tokio sends that complete through a +/// permit after the receiver has closed. +struct InflightEntry { + entry: T, + accounting: QueuedInflightBytes, +} + +impl InflightEntry { + fn new(entry: T, bytes: usize) -> Self { + Self { + entry, + accounting: QueuedInflightBytes::new(bytes), + } + } + + fn into_inner(mut self) -> T { + self.accounting.settle(); + self.entry + } +} + +async fn send_queued( + sender: &mpsc::Sender>, + entry: T, + bytes: usize, +) -> Result<(), mpsc::error::SendError>> { + sender.send(InflightEntry::new(entry, bytes)).await } fn queued_batch_bytes(batch: &[Vec]) -> usize { batch.iter().map(|block| queued_block_bytes(block)).sum() } -async fn drain_queued_batched_inflight_bytes(rx: &mut mpsc::Receiver>>) { - while let Some(batch) = rx.recv().await { - rustfs_io_metrics::remove_ec_encode_inflight_bytes(queued_batch_bytes(&batch)); - } -} - fn dominant_error_summary_label(summary: &WriteQuorumFailureSummary) -> &'static str { summary.dominant_error_label } @@ -504,7 +577,7 @@ impl Erasure { Ok((reader, total)) } - #[hotpath::measure] + #[hotpath::measure(impl_type = "Erasure")] pub async fn encode( self: Arc, reader: R, @@ -540,13 +613,14 @@ impl Erasure { )); } - // Bound queued encoded blocks by memory budget to avoid per-request spikes. + // Bound queued encoded blocks by a queue budget; this does not bound + // reader, encoder, writer, allocator, or process-RSS memory. let expanded_block_bytes = self.shard_size().saturating_mul(self.total_shard_count()); let max_inflight_bytes = erasure_encode_max_inflight_bytes(); let inflight_blocks = encode_channel_capacity(expanded_block_bytes, max_inflight_bytes); - let (tx, mut rx) = mpsc::channel::>(inflight_blocks); + let (tx, mut rx) = mpsc::channel::>>(inflight_blocks); - let task = tokio::spawn(async move { + let mut task = AbortOnDropTask::new(tokio::spawn(async move { let block_size = self.block_size; let mut total = 0; if use_bytesmut_ingest { @@ -567,10 +641,8 @@ impl Erasure { let res = self.clone().encode_block_bytes_mut(encode_buf, n).await?; buf = BytesMut::with_capacity(ingest_capacity); let queued_bytes = queued_block_bytes(&res); - rustfs_io_metrics::add_ec_encode_inflight_bytes(queued_bytes); let send_wait_stage_start = stage_timer_if_enabled(); - if let Err(err) = tx.send(res).await { - rustfs_io_metrics::remove_ec_encode_inflight_bytes(queued_bytes); + if let Err(err) = send_queued(&tx, res, queued_bytes).await { return Err(std::io::Error::other(format!("Failed to send encoded data : {err}"))); } record_internal_stage_if_enabled("erasure_encode_send_wait", send_wait_stage_start); @@ -598,10 +670,8 @@ impl Erasure { let (res, returned_buf) = self.clone().encode_block(encode_buf, n).await?; buf = returned_buf; let queued_bytes = queued_block_bytes(&res); - rustfs_io_metrics::add_ec_encode_inflight_bytes(queued_bytes); let send_wait_stage_start = stage_timer_if_enabled(); - if let Err(err) = tx.send(res).await { - rustfs_io_metrics::remove_ec_encode_inflight_bytes(queued_bytes); + if let Err(err) = send_queued(&tx, res, queued_bytes).await { return Err(std::io::Error::other(format!("Failed to send encoded data : {err}"))); } record_internal_stage_if_enabled("erasure_encode_send_wait", send_wait_stage_start); @@ -626,7 +696,7 @@ impl Erasure { } Ok((reader, total)) - }); + })); let mut writers = MultiWriter::new(writers, quorum); @@ -638,11 +708,10 @@ impl Erasure { break; }; record_internal_stage_if_enabled("erasure_encode_recv_wait", recv_wait_stage_start); + let block = block.into_inner(); if block.is_empty() { break; } - let queued_bytes = queued_block_bytes(&block); - rustfs_io_metrics::remove_ec_encode_inflight_bytes(queued_bytes); let write_stage_start = stage_timer_if_enabled(); if let Err(err) = writers.write(block).await { write_err = Some(err); @@ -652,9 +721,8 @@ impl Erasure { } if let Some(err) = write_err { - task.abort(); - let _ = task.await; - drain_queued_inflight_bytes(&mut rx).await; + task.abort_and_wait().await; + drop(rx); let shutdown_stage_start = stage_timer_if_enabled(); if let Err(shutdown_err) = writers.shutdown().await { error!("failed to shutdown erasure writers after write error: {:?}", shutdown_err); @@ -663,14 +731,14 @@ impl Erasure { return Err(err); } - let (reader, total) = task.await??; + let (reader, total) = task.join().await??; let shutdown_stage_start = stage_timer_if_enabled(); writers.shutdown().await?; record_internal_stage_if_enabled("erasure_encode_shutdown", shutdown_stage_start); Ok((reader, total)) } - #[hotpath::measure] + #[hotpath::measure(impl_type = "Erasure")] pub async fn encode_batched( self: Arc, mut reader: R, @@ -692,9 +760,9 @@ impl Erasure { let inflight_blocks = encode_channel_capacity(expanded_block_bytes, max_inflight_bytes); let batch_blocks = encode_batch_block_count().min(inflight_blocks); let channel_capacity = inflight_blocks.div_ceil(batch_blocks).max(1); - let (tx, mut rx) = mpsc::channel::>>(channel_capacity); + let (tx, mut rx) = mpsc::channel::>>>(channel_capacity); - let task = tokio::spawn(async move { + let mut task = AbortOnDropTask::new(tokio::spawn(async move { let block_size = self.block_size; let mut total = 0; let mut buf = vec![0u8; block_size]; @@ -713,10 +781,8 @@ impl Erasure { pending_batch.push(res); if pending_batch.len() >= batch_blocks { - rustfs_io_metrics::add_ec_encode_inflight_bytes(pending_batch_bytes); let send_wait_stage_start = stage_timer_if_enabled(); - if let Err(err) = tx.send(pending_batch).await { - rustfs_io_metrics::remove_ec_encode_inflight_bytes(pending_batch_bytes); + if let Err(err) = send_queued(&tx, pending_batch, pending_batch_bytes).await { return Err(std::io::Error::other(format!("Failed to send encoded data : {err}"))); } record_internal_stage_if_enabled("erasure_encode_batched_send_wait", send_wait_stage_start); @@ -742,17 +808,15 @@ impl Erasure { } if !pending_batch.is_empty() { - rustfs_io_metrics::add_ec_encode_inflight_bytes(pending_batch_bytes); let send_wait_stage_start = stage_timer_if_enabled(); - if let Err(err) = tx.send(pending_batch).await { - rustfs_io_metrics::remove_ec_encode_inflight_bytes(pending_batch_bytes); + if let Err(err) = send_queued(&tx, pending_batch, pending_batch_bytes).await { return Err(std::io::Error::other(format!("Failed to send encoded data : {err}"))); } record_internal_stage_if_enabled("erasure_encode_batched_send_wait", send_wait_stage_start); } Ok((reader, total)) - }); + })); let mut writers = MultiWriter::new(writers, quorum); let mut write_err = None; @@ -763,7 +827,7 @@ impl Erasure { break; }; record_internal_stage_if_enabled("erasure_encode_batched_recv_wait", recv_wait_stage_start); - rustfs_io_metrics::remove_ec_encode_inflight_bytes(queued_batch_bytes(&batch)); + let batch = batch.into_inner(); let write_stage_start = stage_timer_if_enabled(); for block in batch { if let Err(err) = writers.write(block).await { @@ -778,9 +842,8 @@ impl Erasure { } if let Some(err) = write_err { - task.abort(); - let _ = task.await; - drain_queued_batched_inflight_bytes(&mut rx).await; + task.abort_and_wait().await; + drop(rx); let shutdown_stage_start = stage_timer_if_enabled(); if let Err(shutdown_err) = writers.shutdown().await { error!("failed to shutdown erasure writers after write error: {:?}", shutdown_err); @@ -789,7 +852,7 @@ impl Erasure { return Err(err); } - let (reader, total) = task.await??; + let (reader, total) = task.join().await??; let shutdown_stage_start = stage_timer_if_enabled(); writers.shutdown().await?; record_internal_stage_if_enabled("erasure_encode_batched_shutdown", shutdown_stage_start); @@ -798,7 +861,7 @@ impl Erasure { /// Fast path for small inline objects: skip tokio::spawn + mpsc channel. /// Reads all data, encodes directly, writes shards sequentially. - #[hotpath::measure] + #[hotpath::measure(impl_type = "Erasure")] pub async fn encode_inline_small( self: Arc, reader: R, @@ -813,7 +876,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. - #[hotpath::measure] + #[hotpath::measure(impl_type = "Erasure")] pub async fn encode_single_block_non_inline( self: Arc, reader: R, @@ -839,7 +902,104 @@ mod tests { use std::sync::{Arc, Mutex}; use std::task::{Context, Poll}; use std::time::Duration; - use tokio::io::{AsyncWrite, AsyncWriteExt}; + use tokio::io::{AsyncWrite, AsyncWriteExt, ReadBuf}; + use tokio::sync::oneshot; + + struct PendingReader { + entered: Option>, + dropped: Option>, + } + + impl PendingReader { + fn new() -> (Self, oneshot::Receiver<()>, oneshot::Receiver<()>) { + let (entered_tx, entered_rx) = oneshot::channel(); + let (dropped_tx, dropped_rx) = oneshot::channel(); + ( + Self { + entered: Some(entered_tx), + dropped: Some(dropped_tx), + }, + entered_rx, + dropped_rx, + ) + } + } + + impl AsyncRead for PendingReader { + fn poll_read(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &mut ReadBuf<'_>) -> Poll> { + if let Some(entered) = self.entered.take() { + let _ = entered.send(()); + } + Poll::Pending + } + } + + impl Drop for PendingReader { + fn drop(&mut self) { + if let Some(dropped) = self.dropped.take() { + let _ = dropped.send(()); + } + } + } + + struct BlocksThenPendingReader { + blocks_remaining: usize, + block: Vec, + blocked: Option>, + dropped: Option>, + final_block: Option>, + } + + impl BlocksThenPendingReader { + fn new( + blocks_remaining: usize, + block_size: usize, + ) -> (Self, oneshot::Receiver<()>, oneshot::Receiver<()>, oneshot::Receiver<()>) { + let (blocked_tx, blocked_rx) = oneshot::channel(); + let (dropped_tx, dropped_rx) = oneshot::channel(); + let (final_block_tx, final_block_rx) = oneshot::channel(); + ( + Self { + blocks_remaining, + block: vec![0x5a; block_size], + blocked: Some(blocked_tx), + dropped: Some(dropped_tx), + final_block: Some(final_block_tx), + }, + blocked_rx, + dropped_rx, + final_block_rx, + ) + } + } + + impl AsyncRead for BlocksThenPendingReader { + fn poll_read(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { + if self.blocks_remaining == 0 { + if let Some(blocked) = self.blocked.take() { + let _ = blocked.send(()); + } + return Poll::Pending; + } + + if self.blocks_remaining == 1 + && let Some(final_block) = self.final_block.take() + { + let _ = final_block.send(()); + } + self.blocks_remaining -= 1; + buf.put_slice(&self.block); + Poll::Ready(Ok(())) + } + } + + impl Drop for BlocksThenPendingReader { + fn drop(&mut self) { + if let Some(dropped) = self.dropped.take() { + let _ = dropped.send(()); + } + } + } fn erasure_with_zero_block_size() -> Erasure { let mut erasure = Erasure::default(); @@ -897,6 +1057,54 @@ mod tests { } } + struct FailAfterReaderBlocksWriter { + reader_blocked: oneshot::Receiver<()>, + writes: Arc, + } + + impl AsyncWrite for FailAfterReaderBlocksWriter { + fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, _buf: &[u8]) -> Poll> { + match Pin::new(&mut self.reader_blocked).poll(cx) { + Poll::Pending => Poll::Pending, + Poll::Ready(_) => { + self.writes.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Poll::Ready(Err(std::io::Error::other("injected write failure after producer blocks"))) + } + } + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + } + + struct StallOnWriteWithSignal { + entered: Option>, + writes: Arc, + } + + impl AsyncWrite for StallOnWriteWithSignal { + fn poll_write(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &[u8]) -> Poll> { + self.writes.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + if let Some(entered) = self.entered.take() { + let _ = entered.send(()); + } + Poll::Pending + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + } + #[derive(Clone, Default)] struct ShortWriteWriter; @@ -1046,6 +1254,202 @@ mod tests { BitrotWriterWrapper::new(CustomWriter::new_tokio_writer(writer), shard_size, HashAlgorithm::None) } + #[derive(Clone, Copy)] + enum EncodePipeline { + Vec, + BytesMut, + Batched, + } + + async fn aborting_encode_drops_blocked_producer(pipeline: EncodePipeline) { + const BLOCK_SIZE: usize = 16; + + let gauge_baseline = rustfs_io_metrics::current_ec_encode_inflight_bytes(); + let committed = Arc::new(Mutex::new(Vec::new())); + let mut writers = vec![Some(bitrot_writer(DeferredCommitWriter::new(committed.clone()), BLOCK_SIZE))]; + let (reader, entered, dropped) = PendingReader::new(); + let erasure = Arc::new(Erasure::new(1, 0, BLOCK_SIZE)); + + let encode = match pipeline { + EncodePipeline::Vec => { + tokio::spawn(async move { erasure.encode_with_ingest_mode(reader, &mut writers, 1, false).await }) + } + EncodePipeline::BytesMut => { + tokio::spawn(async move { erasure.encode_with_ingest_mode(reader, &mut writers, 1, true).await }) + } + EncodePipeline::Batched => tokio::spawn(async move { erasure.encode_batched(reader, &mut writers, 1).await }), + }; + + tokio::time::timeout(Duration::from_secs(1), entered) + .await + .expect("producer should enter the blocked reader before cancellation") + .expect("blocked reader should signal entry"); + encode.abort(); + assert!(matches!(encode.await, Err(err) if err.is_cancelled()), "encode task should be cancelled"); + tokio::time::timeout(Duration::from_secs(1), dropped) + .await + .expect("cancelling encode should drop the producer reader") + .expect("blocked reader should signal producer drop"); + assert!( + committed.lock().expect("committed buffer should be lockable").is_empty(), + "cancelling before the first encoded block must not make data visible" + ); + assert_eq!( + rustfs_io_metrics::current_ec_encode_inflight_bytes(), + gauge_baseline, + "cancelling the encode pipeline must preserve the inflight queue gauge" + ); + } + + async fn writer_error_aborts_blocked_producer(pipeline: EncodePipeline) { + const BLOCK_SIZE: usize = 16; + + let gauge_baseline = rustfs_io_metrics::current_ec_encode_inflight_bytes(); + let blocks_before_pending = match pipeline { + EncodePipeline::Batched => encode_batch_block_count(), + EncodePipeline::Vec | EncodePipeline::BytesMut => 1, + }; + let (reader, reader_blocked, reader_dropped, _final_block) = + BlocksThenPendingReader::new(blocks_before_pending, BLOCK_SIZE); + let writes = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let mut writers = vec![Some(bitrot_writer( + FailAfterReaderBlocksWriter { + reader_blocked, + writes: writes.clone(), + }, + BLOCK_SIZE, + ))]; + let erasure = Arc::new(Erasure::new(1, 0, BLOCK_SIZE)); + + let result = match pipeline { + EncodePipeline::Vec => erasure.encode_with_ingest_mode(reader, &mut writers, 1, false).await, + EncodePipeline::BytesMut => erasure.encode_with_ingest_mode(reader, &mut writers, 1, true).await, + EncodePipeline::Batched => erasure.encode_batched(reader, &mut writers, 1).await, + }; + + let err = match result { + Ok(_) => panic!("writer quorum failure should fail the encode pipeline"), + Err(err) => err, + }; + assert!(err.to_string().contains("Failed to write data")); + tokio::time::timeout(Duration::from_secs(1), reader_dropped) + .await + .expect("writer failure should abort the blocked producer") + .expect("blocked producer should signal reader drop"); + assert_eq!( + writes.load(std::sync::atomic::Ordering::SeqCst), + 1, + "writer failure must stop the pipeline before any additional shard write" + ); + assert_eq!( + rustfs_io_metrics::current_ec_encode_inflight_bytes(), + gauge_baseline, + "writer failure must settle all queued and pending encoded bytes" + ); + } + + async fn aborting_full_queue_settles_pending_send() { + const BLOCK_SIZE: usize = 16; + + let gauge_baseline = rustfs_io_metrics::current_ec_encode_inflight_bytes(); + let erasure = Arc::new(Erasure::new(1, 0, BLOCK_SIZE)); + let inflight_blocks = encode_channel_capacity( + erasure.shard_size().saturating_mul(erasure.total_shard_count()), + erasure_encode_max_inflight_bytes(), + ); + let (reader, _reader_blocked, reader_dropped, final_block) = + BlocksThenPendingReader::new(inflight_blocks + 2, BLOCK_SIZE); + let (writer_entered_tx, writer_entered) = oneshot::channel(); + let writes = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let mut writers = vec![Some(bitrot_writer( + StallOnWriteWithSignal { + entered: Some(writer_entered_tx), + writes: writes.clone(), + }, + BLOCK_SIZE, + ))]; + let erasure_for_task = erasure.clone(); + let encode = tokio::spawn(async move { erasure_for_task.encode_with_ingest_mode(reader, &mut writers, 1, false).await }); + + tokio::time::timeout(Duration::from_secs(1), writer_entered) + .await + .expect("consumer should start the first writer call") + .expect("stalling writer should signal entry"); + tokio::time::timeout(Duration::from_secs(1), final_block) + .await + .expect("producer should supply the block whose send fills the queue") + .expect("reader should signal final block"); + + let expected_queued_bytes = + u64::try_from((inflight_blocks + 1) * BLOCK_SIZE).expect("queued byte count should fit the gauge"); + tokio::time::timeout(Duration::from_secs(1), async { + while rustfs_io_metrics::current_ec_encode_inflight_bytes() < gauge_baseline + expected_queued_bytes { + tokio::task::yield_now().await; + } + }) + .await + .expect("producer should account for the pending send after the queue fills"); + + encode.abort(); + assert!(matches!(encode.await, Err(err) if err.is_cancelled()), "encode task should be cancelled"); + tokio::time::timeout(Duration::from_secs(1), reader_dropped) + .await + .expect("cancelling a full queue should abort its producer") + .expect("full-queue producer should signal reader drop"); + assert_eq!( + writes.load(std::sync::atomic::Ordering::SeqCst), + 1, + "cancellation must not resume the stalled writer" + ); + assert_eq!( + rustfs_io_metrics::current_ec_encode_inflight_bytes(), + gauge_baseline, + "cancelling a full queue must settle queued and pending bytes" + ); + } + + #[tokio::test] + #[serial_test::serial] + async fn cancelling_vec_encode_drops_blocked_producer() { + aborting_encode_drops_blocked_producer(EncodePipeline::Vec).await; + } + + #[tokio::test] + #[serial_test::serial] + async fn cancelling_bytesmut_encode_drops_blocked_producer() { + aborting_encode_drops_blocked_producer(EncodePipeline::BytesMut).await; + } + + #[tokio::test] + #[serial_test::serial] + async fn cancelling_batched_encode_drops_blocked_producer() { + aborting_encode_drops_blocked_producer(EncodePipeline::Batched).await; + } + + #[tokio::test] + #[serial_test::serial] + async fn vec_writer_error_aborts_blocked_producer() { + writer_error_aborts_blocked_producer(EncodePipeline::Vec).await; + } + + #[tokio::test] + #[serial_test::serial] + async fn bytesmut_writer_error_aborts_blocked_producer() { + writer_error_aborts_blocked_producer(EncodePipeline::BytesMut).await; + } + + #[tokio::test] + #[serial_test::serial] + async fn batched_writer_error_aborts_blocked_producer() { + writer_error_aborts_blocked_producer(EncodePipeline::Batched).await; + } + + #[tokio::test] + #[serial_test::serial] + async fn cancelling_full_queue_settles_pending_send() { + aborting_full_queue_settles_pending_send().await; + } + #[tokio::test] async fn helper_writers_cover_flush_and_shutdown_paths() { let mut failing_write = FailingWriteWriter; @@ -1329,25 +1733,99 @@ mod tests { } #[tokio::test] - async fn drain_queued_inflight_bytes_consumes_pending_blocks() { - let (tx, mut rx) = mpsc::channel(2); - tx.send(vec![Bytes::from_static(b"queued")]).await.unwrap(); - drop(tx); + #[serial_test::serial] + async fn queued_inflight_bytes_are_settled_on_all_queue_exit_paths() { + let baseline = rustfs_io_metrics::current_ec_encode_inflight_bytes(); + let (tx, rx) = mpsc::channel(1); + let queued = vec![Bytes::from_static(b"queued")]; + let queued_bytes = queued_block_bytes(&queued); - drain_queued_inflight_bytes(&mut rx).await; + send_queued(&tx, queued, queued_bytes) + .await + .expect("first queue entry should fit"); - assert!(rx.recv().await.is_none()); + let blocked = vec![Bytes::from_static(b"blocked")]; + let blocked_bytes = queued_block_bytes(&blocked); + { + let pending_send = send_queued(&tx, blocked, blocked_bytes); + tokio::pin!(pending_send); + assert!( + futures::poll!(pending_send.as_mut()).is_pending(), + "full queue must suspend producer send" + ); + } + assert_eq!( + rustfs_io_metrics::current_ec_encode_inflight_bytes(), + baseline + u64::try_from(queued_bytes).expect("queue bytes fit the gauge"), + "dropping a pending producer send must compensate its bytes" + ); + + drop(rx); + assert_eq!( + rustfs_io_metrics::current_ec_encode_inflight_bytes(), + baseline, + "dropping the receiver must settle every buffered entry" + ); + + let rejected = vec![Bytes::from_static(b"rejected")]; + let rejected_bytes = queued_block_bytes(&rejected); + assert!( + send_queued(&tx, rejected, rejected_bytes).await.is_err(), + "closed receiver must reject a new send" + ); + assert_eq!( + rustfs_io_metrics::current_ec_encode_inflight_bytes(), + baseline, + "failed sends must compensate their bytes" + ); } #[tokio::test] - async fn drain_queued_batched_inflight_bytes_consumes_pending_batches() { - let (tx, mut rx) = mpsc::channel(2); - tx.send(vec![vec![Bytes::from_static(b"queued")]]).await.unwrap(); - drop(tx); + #[serial_test::serial] + async fn queued_batch_entry_settles_bytes_before_handoff() { + let baseline = rustfs_io_metrics::current_ec_encode_inflight_bytes(); + let (tx, rx) = mpsc::channel(2); + let mut rx = rx; + let batch = vec![vec![Bytes::from_static(b"queued")], vec![Bytes::from_static(b"batch")]]; + let batch_bytes = queued_batch_bytes(&batch); - drain_queued_batched_inflight_bytes(&mut rx).await; + send_queued(&tx, batch, batch_bytes).await.expect("batch should be queued"); + let batch = rx.recv().await.expect("queued batch should be received").into_inner(); + assert_eq!(batch_bytes, queued_batch_bytes(&batch)); - assert!(rx.recv().await.is_none()); + assert_eq!( + rustfs_io_metrics::current_ec_encode_inflight_bytes(), + baseline, + "receiving a batch must settle all contained block bytes before shard writes" + ); + } + + #[tokio::test] + #[serial_test::serial] + async fn queued_entry_settles_when_a_closed_receiver_accepts_an_outstanding_permit() { + let baseline = rustfs_io_metrics::current_ec_encode_inflight_bytes(); + let (tx, mut rx) = mpsc::channel(1); + let permit = tx + .clone() + .reserve_owned() + .await + .expect("open receiver should reserve queue capacity"); + rx.close(); + assert!( + matches!(rx.try_recv(), Err(mpsc::error::TryRecvError::Empty)), + "an outstanding permit must leave the closed queue observably empty" + ); + + let block = vec![Bytes::from_static(b"late-permit")]; + let block_bytes = queued_block_bytes(&block); + permit.send(InflightEntry::new(block, block_bytes)); + drop(rx); + + assert_eq!( + rustfs_io_metrics::current_ec_encode_inflight_bytes(), + baseline, + "a queue entry sent through an outstanding permit must settle when Tokio drops it" + ); } #[tokio::test] diff --git a/crates/ecstore/src/erasure/coding/erasure.rs b/crates/ecstore/src/erasure/coding/erasure.rs index 5607a2302..50910f47b 100644 --- a/crates/ecstore/src/erasure/coding/erasure.rs +++ b/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()))] - #[hotpath::measure] + #[hotpath::measure(impl_type = "Erasure")] pub fn encode_data(&self, data: &[u8]) -> io::Result> { 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. - #[hotpath::measure] + #[hotpath::measure(impl_type = "Erasure")] pub fn encode_data_owned(&self, data: Vec) -> io::Result> { 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. - #[hotpath::measure] + #[hotpath::measure(impl_type = "Erasure")] pub fn encode_data_bytes_mut(&self, mut data_buffer: BytesMut, data_len: usize) -> io::Result> { 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. - #[hotpath::measure] + #[hotpath::measure(impl_type = "Erasure")] pub fn decode_data(&self, shards: &mut [Option>]) -> 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. - #[hotpath::measure] + #[hotpath::measure(impl_type = "Erasure")] pub fn decode_data_and_parity(&self, shards: &mut [Option>]) -> io::Result<()> { if self.parity_shards > 0 { if self.uses_legacy { diff --git a/crates/ecstore/src/io_support/bitrot.rs b/crates/ecstore/src/io_support/bitrot.rs index aec9eb79e..f48635f1a 100644 --- a/crates/ecstore/src/io_support/bitrot.rs +++ b/crates/ecstore/src/io_support/bitrot.rs @@ -19,6 +19,8 @@ use crate::diagnostics::get::{ GET_STAGE_READER_MMAP_PATH_RESOLVE, GET_STAGE_READER_OPEN_MMAP_COPY_FALLBACK, GET_STAGE_READER_OPEN_MMAP_COPY_SUCCESS, GET_STAGE_READER_OPEN_STREAM, GET_STAGE_READER_STREAM_FIRST_READ, record_get_stage_duration_if_enabled, }; +#[cfg(feature = "hotpath")] +use crate::disk::FileWriter; use crate::disk::{self, DiskAPI as _, DiskStore, FileReader, MmapCopyStageMetrics, error::DiskError}; use crate::erasure::coding::{BitrotReader, BitrotWriterWrapper, CustomWriter}; use bytes::Bytes; @@ -36,6 +38,11 @@ use std::time::Instant; use tokio::io::{AsyncRead, ReadBuf}; use tracing::debug; +#[cfg(all(test, feature = "hotpath"))] +tokio::task_local! { + static FORCE_MMAP_COPY_FAILURE_FOR_TEST: (); +} + /// A shard source for the bitrot reader. /// /// `InMemory` keeps the `Bytes` concrete instead of erasing it behind @@ -360,10 +367,21 @@ async fn open_disk_reader( mmap_copy_stage: GET_STAGE_READER_MMAP_COPY_BUFFER, direct_read_copy_stage: GET_STAGE_READER_MMAP_DIRECT_READ_COPY, }); - match disk - .read_file_mmap_copy_with_metrics(bucket, path, offset, length, mmap_metrics) - .await - { + let mmap_result = { + #[cfg(all(test, feature = "hotpath"))] + if FORCE_MMAP_COPY_FAILURE_FOR_TEST.try_with(|_| ()).is_ok() { + Err(DiskError::other("forced mmap-copy failure for test")) + } else { + disk.read_file_mmap_copy_with_metrics(bucket, path, offset, length, mmap_metrics) + .await + } + #[cfg(not(all(test, feature = "hotpath")))] + { + disk.read_file_mmap_copy_with_metrics(bucket, path, offset, length, mmap_metrics) + .await + } + }; + match mmap_result { Ok(bytes) => { let duration_ms = zero_copy_start.elapsed().as_secs_f64() * 1000.0; @@ -398,7 +416,11 @@ async fn open_disk_reader( } return match stream_result { - Ok(reader) => Ok(wrap_first_read_metrics(reader, metrics_path)), + Ok(reader) => { + #[cfg(feature = "hotpath")] + let reader = instrument_raw_shard_reader(reader, disk.is_local()); + Ok(wrap_first_read_metrics(reader, metrics_path)) + } Err(_) => Err(err), }; } @@ -407,6 +429,8 @@ async fn open_disk_reader( let stream_start = stage_metrics_enabled.then(Instant::now); let reader = disk.read_file_stream(bucket, path, offset, length).await?; + #[cfg(feature = "hotpath")] + let reader = instrument_raw_shard_reader(reader, disk.is_local()); if let Some(metrics_path) = metrics_path { record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READER_OPEN_STREAM, stream_start); } @@ -427,6 +451,37 @@ fn wrap_first_read_metrics(reader: FileReader, metrics_path: Option<&'static str ShardReader::Stream(reader) } +// The labels are deliberately fixed: object keys, disk paths, and remote hosts +// are all high-cardinality or sensitive and belong nowhere in a profiling report. +#[cfg(feature = "hotpath")] +const RAW_SHARD_READ_LOCAL_LABEL: &str = "EC raw shard read local"; +#[cfg(feature = "hotpath")] +const RAW_SHARD_READ_REMOTE_LABEL: &str = "EC raw shard read remote"; +#[cfg(feature = "hotpath")] +const RAW_SHARD_WRITE_LOCAL_LABEL: &str = "EC raw shard write local"; +#[cfg(feature = "hotpath")] +const RAW_SHARD_WRITE_REMOTE_LABEL: &str = "EC raw shard write remote"; + +#[cfg(feature = "hotpath")] +fn instrument_raw_shard_reader(reader: FileReader, is_local: bool) -> FileReader { + // `io!` aggregates by call site, so the local and remote branches must remain distinct. + if is_local { + Box::new(hotpath::io!(reader, label = RAW_SHARD_READ_LOCAL_LABEL)) + } else { + Box::new(hotpath::io!(reader, label = RAW_SHARD_READ_REMOTE_LABEL)) + } +} + +#[cfg(feature = "hotpath")] +fn instrument_raw_shard_writer(writer: FileWriter, is_local: bool) -> FileWriter { + // `io!` aggregates by call site, so the local and remote branches must remain distinct. + if is_local { + Box::new(hotpath::io!(writer, label = RAW_SHARD_WRITE_LOCAL_LABEL)) + } else { + Box::new(hotpath::io!(writer, label = RAW_SHARD_WRITE_REMOTE_LABEL)) + } +} + fn bitrot_encoded_range(offset: usize, length: usize, shard_size: usize, checksum_algo: HashAlgorithm) -> (usize, usize) { ( offset.div_ceil(shard_size) * checksum_algo.size() + offset, @@ -686,6 +741,8 @@ pub async fn create_bitrot_writer( }; let file = disk.create_file("", volume, path, length).await?; + #[cfg(feature = "hotpath")] + let file = instrument_raw_shard_writer(file, disk.is_local()); CustomWriter::new_tokio_writer(file) } else { return Err(DiskError::DiskNotFound); @@ -698,6 +755,182 @@ pub async fn create_bitrot_writer( mod tests { use super::*; + #[cfg(feature = "hotpath")] + use crate::cluster::rpc::RemoteDisk; + #[cfg(feature = "hotpath")] + use crate::cluster::rpc::internode_data_transport::{ + InternodeDataTransport, InternodeDataTransportCapabilities, ReadStreamRequest, WalkDirStreamRequest, WriteStreamRequest, + }; + #[cfg(feature = "hotpath")] + use crate::disk::{Disk, DiskOption, error::Result}; + + #[cfg(feature = "hotpath")] + #[derive(Debug, Clone, Default)] + struct TestRemoteDataTransport { + bytes: Arc>>, + } + + #[cfg(feature = "hotpath")] + impl TestRemoteDataTransport { + fn bytes(&self) -> Vec { + self.bytes + .lock() + .expect("test remote transport bytes lock should not be poisoned") + .clone() + } + } + + #[cfg(feature = "hotpath")] + #[derive(Debug)] + struct TestRemoteWriter { + bytes: Arc>>, + } + + #[cfg(feature = "hotpath")] + impl tokio::io::AsyncWrite for TestRemoteWriter { + fn poll_write(self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + self.bytes + .lock() + .expect("test remote transport bytes lock should not be poisoned") + .extend_from_slice(buf); + Poll::Ready(Ok(buf.len())) + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + } + + #[cfg(feature = "hotpath")] + #[async_trait::async_trait] + impl InternodeDataTransport for TestRemoteDataTransport { + async fn open_read(&self, _request: ReadStreamRequest) -> Result { + Ok(Box::new(Cursor::new(self.bytes()))) + } + + async fn open_write(&self, _request: WriteStreamRequest) -> Result { + Ok(Box::new(TestRemoteWriter { + bytes: Arc::clone(&self.bytes), + })) + } + + async fn open_walk_dir(&self, _request: WalkDirStreamRequest) -> Result { + panic!("open_walk_dir must not be used by the raw shard I/O test") + } + + fn name(&self) -> &'static str { + "bitrot-test-remote" + } + + fn capabilities(&self) -> InternodeDataTransportCapabilities { + InternodeDataTransportCapabilities::tcp_http() + } + } + + async fn local_test_disk() -> (DiskStore, tempfile::TempDir) { + use crate::disk::endpoint::Endpoint; + use crate::disk::{DiskOption, new_disk}; + + let dir = tempfile::tempdir().expect("tempdir should be created"); + let mut endpoint = + Endpoint::try_from(dir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse"); + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(0); + let disk = new_disk( + &endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await + .expect("local disk should be created"); + + (disk, dir) + } + + #[cfg(feature = "hotpath")] + async fn remote_test_disk() -> (DiskStore, TestRemoteDataTransport) { + use crate::disk::endpoint::Endpoint; + + let endpoint = Endpoint { + url: url::Url::parse("http://remote-node:9000/data/rustfs0").expect("test remote endpoint should parse"), + is_local: false, + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }; + let transport = TestRemoteDataTransport::default(); + let remote = RemoteDisk::new( + &endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + Arc::new(transport.clone()), + ) + .await + .expect("test remote disk should be created"); + + (Arc::new(Disk::Remote(Box::new(remote))), transport) + } + + async fn round_trip_disk_bitrot(disk: &DiskStore, bucket: &str, path: &str, payload: &[u8], shard_size: usize) -> Vec { + disk.make_volume(bucket).await.expect("volume should be created"); + let mut writer = create_bitrot_writer( + false, + Some(disk), + bucket, + path, + i64::try_from(payload.len()).expect("test payload length should fit i64"), + shard_size, + HashAlgorithm::None, + ) + .await + .expect("disk bitrot writer should open the raw shard file"); + for chunk in payload.chunks(shard_size) { + writer + .write(chunk) + .await + .expect("disk bitrot writer should preserve each shard block"); + } + writer.shutdown().await.expect("disk bitrot writer should close cleanly"); + + let mut reader = create_bitrot_reader( + None, + Some(disk), + bucket, + path, + 0, + payload.len(), + shard_size, + HashAlgorithm::None, + false, + false, + ) + .await + .expect("disk bitrot reader should open the raw shard file") + .expect("disk bitrot reader should exist"); + let mut actual = Vec::with_capacity(payload.len()); + while actual.len() < payload.len() { + let remaining = payload.len() - actual.len(); + let mut chunk = vec![0; remaining.min(shard_size)]; + let read = reader + .read(&mut chunk) + .await + .expect("disk bitrot reader should return the complete shard body"); + assert!(read > 0, "disk bitrot reader must not end before the expected shard body is complete"); + actual.extend_from_slice(&chunk[..read]); + } + + actual + } + #[test] fn object_mmap_read_enabled_accepts_legacy_zero_copy_alias() { temp_env::with_vars( @@ -724,6 +957,165 @@ mod tests { ); } + #[cfg(feature = "hotpath")] + #[test] + fn raw_shard_io_wrappers_report_fixed_labels_and_preserve_bytes() { + const CHILD_ENV: &str = "RUSTFS_HOTPATH_RAW_SHARD_IO_TEST_CHILD"; + if std::env::var_os(CHILD_ENV).is_none() { + let status = std::process::Command::new(std::env::current_exe().expect("test executable path should be available")) + .arg("--exact") + .arg("io_support::bitrot::tests::raw_shard_io_wrappers_report_fixed_labels_and_preserve_bytes") + .arg("--nocapture") + .env(CHILD_ENV, "1") + .status() + .expect("isolated HotPath I/O test process should start"); + assert!(status.success(), "isolated HotPath I/O test process should pass"); + return; + } + + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("test runtime should be created") + .block_on(raw_shard_io_wrappers_report_fixed_labels_and_preserve_bytes_in_isolated_process()); + } + + #[cfg(feature = "hotpath")] + async fn raw_shard_io_wrappers_report_fixed_labels_and_preserve_bytes_in_isolated_process() { + use hotpath::{Format, HotpathGuardBuilder, Section}; + use tokio::io::AsyncReadExt; + + let report_dir = tempfile::tempdir().expect("report tempdir should be created"); + let report_path = report_dir.path().join("hotpath-io.json"); + let guard = HotpathGuardBuilder::new("raw_shard_io_test") + .format(Format::Json) + .output_path(&report_path) + .sections(vec![Section::Io]) + .build(); + + let (disk, _dir) = local_test_disk().await; + let bucket = "test-bucket"; + let path = "obj/hotpath-part.1"; + let payload = b"local shard bytes"; + let shard_size = 4; + let local_read = round_trip_disk_bitrot(&disk, bucket, path, payload, shard_size).await; + assert_eq!(local_read, payload, "local raw shard I/O instrumentation must not alter stored bytes"); + + let fallback_path = "obj/hotpath-mmap-fallback-part.1"; + let fallback_payload = b"mmap fallback shard bytes"; + disk.write_all("test-bucket", fallback_path, Bytes::from_static(fallback_payload)) + .await + .expect("fallback shard file should be written"); + let mut fallback_reader = FORCE_MMAP_COPY_FAILURE_FOR_TEST + .scope((), open_disk_reader(&disk, bucket, fallback_path, 0, fallback_payload.len(), true, None)) + .await + .expect("mmap-copy failure should fall back to a raw shard stream"); + assert!( + matches!(fallback_reader, ShardReader::Stream(_)), + "forced mmap-copy failure must use the streaming fallback" + ); + let mut fallback_read = Vec::new(); + fallback_reader + .read_to_end(&mut fallback_read) + .await + .expect("mmap-copy fallback stream should preserve bytes"); + assert_eq!(fallback_read, fallback_payload); + + let (remote_disk, remote_transport) = remote_test_disk().await; + let remote_payload = b"remote shard bytes"; + let mut remote_writer = create_bitrot_writer( + false, + Some(&remote_disk), + bucket, + "obj/hotpath-remote-part.1", + i64::try_from(remote_payload.len()).expect("remote payload length should fit i64"), + shard_size, + HashAlgorithm::None, + ) + .await + .expect("remote bitrot writer should use the production raw writer path"); + for chunk in remote_payload.chunks(shard_size) { + remote_writer + .write(chunk) + .await + .expect("remote bitrot writer should preserve bytes"); + } + remote_writer + .shutdown() + .await + .expect("remote bitrot writer should close cleanly"); + assert_eq!(remote_transport.bytes(), remote_payload); + + let mut reader = create_bitrot_reader( + None, + Some(&remote_disk), + bucket, + "obj/hotpath-remote-part.1", + 0, + remote_payload.len(), + shard_size, + HashAlgorithm::None, + false, + true, + ) + .await + .expect("remote bitrot reader should use the production raw reader path") + .expect("remote bitrot reader should exist"); + let mut remote_read = Vec::with_capacity(remote_payload.len()); + while remote_read.len() < remote_payload.len() { + let remaining = remote_payload.len() - remote_read.len(); + let mut chunk = vec![0; remaining.min(shard_size)]; + let read = reader + .read(&mut chunk) + .await + .expect("remote bitrot reader should preserve bytes"); + assert!(read > 0, "remote bitrot reader must not end before the expected shard body is complete"); + remote_read.extend_from_slice(&chunk[..read]); + } + assert_eq!(remote_read, remote_payload); + + drop(guard); + let report = std::fs::read_to_string(&report_path).expect("HotPath I/O report should be written"); + let report: serde_json::Value = serde_json::from_str(&report).expect("HotPath I/O report should be valid JSON"); + let entries = report["io"]["data"] + .as_array() + .expect("HotPath I/O report should include data rows"); + let io_bytes = |label: &str, direction: &str| { + let byte_count: u64 = entries + .iter() + .filter(|entry| entry["label"].as_str().is_some_and(|entry_label| entry_label == label)) + .filter_map(|entry| entry[direction]["bytes"].as_u64()) + .sum(); + assert!(byte_count > 0, "report must include fixed label {label}"); + byte_count + }; + let payload_len = u64::try_from(payload.len()).expect("test payload length should fit u64"); + assert_eq!( + io_bytes(RAW_SHARD_READ_LOCAL_LABEL, "read"), + payload_len + u64::try_from(fallback_payload.len()).expect("fallback payload length should fit u64") + ); + assert_eq!(io_bytes(RAW_SHARD_WRITE_LOCAL_LABEL, "write"), payload_len); + assert_eq!( + io_bytes(RAW_SHARD_READ_REMOTE_LABEL, "read"), + u64::try_from(remote_payload.len()).expect("remote payload length should fit u64") + ); + assert_eq!( + io_bytes(RAW_SHARD_WRITE_REMOTE_LABEL, "write"), + u64::try_from(remote_payload.len()).expect("remote payload length should fit u64") + ); + for label in [ + RAW_SHARD_READ_LOCAL_LABEL, + RAW_SHARD_READ_REMOTE_LABEL, + RAW_SHARD_WRITE_LOCAL_LABEL, + RAW_SHARD_WRITE_REMOTE_LABEL, + ] { + assert!( + !label.contains(['/', ':', '?', '@']), + "raw shard I/O labels must not carry a path, host, query, or credential delimiter: {label}" + ); + } + } + #[test] fn object_mmap_read_max_length_defaults_and_env_override() { temp_env::with_var(ENV_OBJECT_MMAP_READ_MAX_LENGTH, None::<&str>, || { @@ -741,25 +1133,9 @@ mod tests { // be materialized in memory by the mmap-copy path; over-cap reads stream. #[tokio::test] async fn open_disk_reader_streams_when_length_exceeds_mmap_cap() { - use crate::disk::endpoint::Endpoint; - use crate::disk::{DiskOption, new_disk}; use tokio::io::AsyncReadExt; - let dir = tempfile::tempdir().expect("tempdir should be created"); - let mut endpoint = - Endpoint::try_from(dir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse"); - endpoint.set_pool_index(0); - endpoint.set_set_index(0); - endpoint.set_disk_index(0); - let disk = new_disk( - &endpoint, - &DiskOption { - cleanup: false, - health_check: false, - }, - ) - .await - .expect("local disk should be created"); + let (disk, _dir) = local_test_disk().await; let payload = vec![7u8; 4096]; disk.make_volume("test-bucket").await.expect("volume should be created"); @@ -814,6 +1190,17 @@ mod tests { .await; } + #[tokio::test] + async fn disk_bitrot_reader_and_writer_preserve_full_shard_body() { + let (disk, _dir) = local_test_disk().await; + let bucket = "test-bucket"; + let path = "obj/wrapped-part.1"; + let payload = b"wrapped shard body"; + let actual = round_trip_disk_bitrot(&disk, bucket, path, payload, 4).await; + + assert_eq!(actual, payload, "raw shard I/O instrumentation must not alter stored bytes"); + } + #[tokio::test] async fn test_create_bitrot_reader_with_inline_data() { let test_data = b"hello world test data"; diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index aaec15180..1d31c0e27 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -199,7 +199,7 @@ impl SetDisks { ); } - #[hotpath::measure] + #[hotpath::measure(impl_type = "SetDisks")] pub async fn read_version_optimized( &self, bucket: &str, @@ -238,7 +238,7 @@ impl SetDisks { } #[tracing::instrument(level = "debug", skip(self))] - #[hotpath::measure] + #[hotpath::measure(impl_type = "SetDisks")] pub(super) async fn get_object_fileinfo( &self, bucket: &str, @@ -410,7 +410,7 @@ impl SetDisks { Ok((fi, parts_metadata, op_online_disks)) } - #[hotpath::measure] + #[hotpath::measure(impl_type = "SetDisks")] pub(super) async fn get_object_info_and_quorum( &self, bucket: &str, @@ -605,7 +605,7 @@ impl SetDisks { } #[allow(clippy::too_many_arguments)] - #[hotpath::measure] + #[hotpath::measure(impl_type = "SetDisks")] pub(super) async fn get_object_with_fileinfo( // &self, bucket: &str, @@ -1140,7 +1140,7 @@ impl SetDisks { } #[allow(clippy::too_many_arguments)] - #[hotpath::measure] + #[hotpath::measure(impl_type = "SetDisks")] 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)] - #[hotpath::measure] + #[hotpath::measure(impl_type = "SetDisks")] async fn build_codec_streaming_part_reader( bucket: &str, object: &str, diff --git a/crates/ecstore/src/store/multipart.rs b/crates/ecstore/src/store/multipart.rs index 32b21c11e..d595d1885 100644 --- a/crates/ecstore/src/store/multipart.rs +++ b/crates/ecstore/src/store/multipart.rs @@ -238,7 +238,7 @@ impl ECStore { } #[instrument(skip(self, data))] - #[hotpath::measure] + #[hotpath::measure(impl_type = "ECStore")] pub(super) async fn handle_put_object_part( &self, bucket: &str, diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index 6bfa27ed6..cacc1e94d 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -815,7 +815,7 @@ impl ECStore { } #[instrument(level = "debug", skip(self))] - #[hotpath::measure] + #[hotpath::measure(impl_type = "ECStore")] pub(super) async fn handle_get_object_reader( &self, bucket: &str, @@ -849,7 +849,7 @@ impl ECStore { } #[instrument(level = "debug", skip(self, data))] - #[hotpath::measure] + #[hotpath::measure(impl_type = "ECStore")] pub(super) async fn handle_put_object( &self, bucket: &str, diff --git a/crates/ecstore/tests/hotpath_cpu_attribution_contract_test.rs b/crates/ecstore/tests/hotpath_cpu_attribution_contract_test.rs new file mode 100644 index 000000000..38c82329b --- /dev/null +++ b/crates/ecstore/tests/hotpath_cpu_attribution_contract_test.rs @@ -0,0 +1,95 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +const STORE_MULTIPART: &str = include_str!("../src/store/multipart.rs"); +const STORE_OBJECT: &str = include_str!("../src/store/object.rs"); +const ERASURE: &str = include_str!("../src/erasure/coding/erasure.rs"); +const DECODE: &str = include_str!("../src/erasure/coding/decode.rs"); +const ENCODE: &str = include_str!("../src/erasure/coding/encode.rs"); +const LOCAL_DISK: &str = include_str!("../src/disk/local.rs"); +const BITROT: &str = include_str!("../src/erasure/coding/bitrot.rs"); +const SET_DISK_READ: &str = include_str!("../src/set_disk/read.rs"); + +fn assert_measured_as(source: &str, impl_type: &str, function: &str) { + let attribute = format!("#[hotpath::measure(impl_type = \"{impl_type}\")]\n"); + let function_start = format!("fn {function}"); + let mut remaining = source; + + while let Some(attribute_offset) = remaining.find(&attribute) { + let measured_source = &remaining[attribute_offset + attribute.len()..]; + if measured_source.find(&function_start).is_some_and(|offset| offset < 256) { + return; + } + remaining = measured_source; + } + + panic!("missing HotPath impl_type={impl_type} for {function}"); +} + +#[test] +fn inherent_hotpath_measurements_have_cpu_attribution_types() { + assert_measured_as(STORE_MULTIPART, "ECStore", "handle_put_object_part"); + for function in ["handle_get_object_reader", "handle_put_object"] { + assert_measured_as(STORE_OBJECT, "ECStore", function); + } + for function in [ + "encode_data", + "encode_data_owned", + "encode_data_bytes_mut", + "decode_data", + "decode_data_and_parity", + ] { + assert_measured_as(ERASURE, "Erasure", function); + } + assert_measured_as(DECODE, "ParallelReader", "read"); + assert_measured_as(DECODE, "Erasure", "decode"); + for function in [ + "encode", + "encode_batched", + "encode_inline_small", + "encode_single_block_non_inline", + ] { + assert_measured_as(ENCODE, "Erasure", function); + } + for function in ["read_metadata_with_dmtime", "read_all_data"] { + assert_measured_as(LOCAL_DISK, "LocalDisk", function); + } + assert_measured_as(BITROT, "BitrotReader", "read"); + assert!( + BITROT.contains("#[hotpath::measure(label = \"BitrotWriter::write\", impl_type = \"BitrotWriter\")]"), + "BitrotWriter::write must retain its stable label and CPU impl_type", + ); + for function in [ + "read_version_optimized", + "get_object_fileinfo", + "get_object_info_and_quorum", + "get_object_with_fileinfo", + "get_object_decode_reader_with_fileinfo", + "build_codec_streaming_part_reader", + ] { + assert_measured_as(SET_DISK_READ, "SetDisks", function); + } +} + +#[test] +fn module_hotpath_measurements_remain_untyped() { + assert!( + BITROT.contains("#[hotpath::measure]\npub async fn bitrot_verify"), + "bitrot_verify is a module function and must not claim an impl type", + ); + assert!( + SET_DISK_READ.contains("#[hotpath::measure]\nasync fn setup_multipart_part_readers"), + "setup_multipart_part_readers is a module function and must not claim an impl type", + ); +} diff --git a/crates/iam/src/sys.rs b/crates/iam/src/sys.rs index 263e1f727..ab6f40a2d 100644 --- a/crates/iam/src/sys.rs +++ b/crates/iam/src/sys.rs @@ -90,7 +90,13 @@ fn get_policy_plugin_state() -> Arc> { } Ok(_) => PolicyPluginState::Failed, Err(e) => { - error!("Error loading OPA configuration err:{}", e); + error!( + component = "iam", + subsystem = "policy_plugin", + result = "configuration_load_failed", + error_kind = e.kind(), + "OPA plugin configuration load failed" + ); PolicyPluginState::Failed } }; diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index 9fd1dbe71..85e7519bf 100644 --- a/crates/io-metrics/src/lib.rs +++ b/crates/io-metrics/src/lib.rs @@ -2195,18 +2195,22 @@ pub fn record_allocator_memory_observation(backend: &'static str, observation: A } } -/// Track encoded bytes currently queued between erasure encode and disk writers. +/// Track encoded bytes in the queue hand-off between erasure encode and disk writers. +/// +/// This is queue occupancy, not a per-request or process-RSS memory limit. The +/// erasure encoder settles it on failed/cancelled sends, receiver drop, and +/// consumer hand-off before shard writes begin. #[inline(always)] pub fn add_ec_encode_inflight_bytes(bytes: usize) { let next = EC_ENCODE_INFLIGHT_BYTES.fetch_add(bytes as u64, Ordering::Relaxed) + bytes as u64; - gauge!("rustfs_ec_encode_inflight_bytes_current").set(next as f64); + gauge_set_cached!("rustfs_ec_encode_inflight_bytes_current", next as f64); } /// Remove encoded bytes from the tracked erasure encode in-flight gauge. #[inline(always)] pub fn remove_ec_encode_inflight_bytes(bytes: usize) { let next = saturating_sub_atomic(&EC_ENCODE_INFLIGHT_BYTES, bytes as u64); - gauge!("rustfs_ec_encode_inflight_bytes_current").set(next as f64); + gauge_set_cached!("rustfs_ec_encode_inflight_bytes_current", next as f64); } /// Return the current tracked EC encode in-flight bytes. diff --git a/crates/obs/Cargo.toml b/crates/obs/Cargo.toml index 8b627c353..eeec4ee02 100644 --- a/crates/obs/Cargo.toml +++ b/crates/obs/Cargo.toml @@ -146,6 +146,7 @@ tracing-opentelemetry = { workspace = true } tracing-subscriber = { workspace = true, features = ["fmt", "env-filter", "tracing-log", "local-time", "json"] } tokio = { workspace = true, features = ["sync", "rt-multi-thread", "time", "macros"] } tokio-util = { workspace = true, features = ["io", "compat"] } +url.workspace = true dial9-tokio-telemetry = { workspace = true, optional = true } thiserror = { workspace = true } zstd = { workspace = true, features = ["zstdmt"] } @@ -162,3 +163,4 @@ libc = { workspace = true } [dev-dependencies] tempfile = { workspace = true } temp-env = { workspace = true } +log = "0.4" diff --git a/crates/obs/src/telemetry/filter.rs b/crates/obs/src/telemetry/filter.rs index d0e9c1df7..9ef639eb0 100644 --- a/crates/obs/src/telemetry/filter.rs +++ b/crates/obs/src/telemetry/filter.rs @@ -18,7 +18,26 @@ //! used across different logging backends (stdout, file, OpenTelemetry). use std::env; -use tracing_subscriber::{EnvFilter, filter::LevelFilter}; +use tracing::Metadata; +use tracing_subscriber::{ + EnvFilter, + filter::{FilterFn, LevelFilter, filter_fn}, +}; + +/// Pyroscope emits raw reqwest errors from its background session manager. +/// Those errors can include the configured endpoint, so they must never reach +/// a RustFS logging sink even when an operator enables verbose dependency logs. +pub(super) fn is_pyroscope_dependency_target(target: &str) -> bool { + target == "pyroscope" || target.starts_with("pyroscope::") || target.starts_with("Pyroscope::") +} + +fn allows_non_pyroscope_dependency_event(metadata: &Metadata<'_>) -> bool { + !is_pyroscope_dependency_target(metadata.target()) +} + +pub(super) fn pyroscope_log_filter() -> FilterFn { + filter_fn(allows_non_pyroscope_dependency_event) +} /// Build an `EnvFilter` from the given log level string. /// diff --git a/crates/obs/src/telemetry/local.rs b/crates/obs/src/telemetry/local.rs index b0a9d947b..fedc28dbf 100644 --- a/crates/obs/src/telemetry/local.rs +++ b/crates/obs/src/telemetry/local.rs @@ -36,7 +36,7 @@ use crate::cleaner::LogCleaner; use crate::cleaner::types::{CompressionAlgorithm, FileMatchMode}; use crate::config::OtelConfig; use crate::global::{METRIC_LOG_CLEANER_RUN_FAILURES_TOTAL, METRIC_LOG_CLEANER_RUNS_TOTAL, set_observability_metric_enabled}; -use crate::telemetry::filter::build_env_filter; +use crate::telemetry::filter::{build_env_filter, pyroscope_log_filter}; use crate::telemetry::rolling::{RollingAppender, Rotation}; use metrics::counter; use rustfs_config::observability::{ @@ -319,6 +319,7 @@ fn init_stdout_only(_config: &OtelConfig, logger_level: &str, is_production: boo tracing_subscriber::registry() .with(env_filter) + .with(pyroscope_log_filter()) .with(ErrorLayer::default()) .with(fmt_layer) .try_init() @@ -431,6 +432,7 @@ fn init_file_logging_internal( tracing_subscriber::registry() .with(env_filter) + .with(pyroscope_log_filter()) .with(ErrorLayer::default()) .with(file_layer) .with(stdout_layer) diff --git a/crates/obs/src/telemetry/otel.rs b/crates/obs/src/telemetry/otel.rs index f452a0c26..e03d450b3 100644 --- a/crates/obs/src/telemetry/otel.rs +++ b/crates/obs/src/telemetry/otel.rs @@ -39,7 +39,7 @@ use crate::cleaner::types::FileMatchMode; use crate::config::OtelConfig; use crate::global::set_observability_metric_enabled; -use crate::telemetry::filter::build_env_filter; +use crate::telemetry::filter::{build_env_filter, pyroscope_log_filter}; use crate::telemetry::guard::{OtelGuard, ProfilingAgent}; use crate::telemetry::local::{build_json_log_layer, spawn_cleanup_task}; use crate::telemetry::recorder::{Recorder, install_process_global_recorder}; @@ -68,7 +68,7 @@ use std::{fs, io::IsTerminal, time::Duration}; use tracing::{info, warn}; use tracing_error::ErrorLayer; use tracing_opentelemetry::{MetricsLayer, OpenTelemetryLayer}; -use tracing_subscriber::{Layer, fmt::format::FmtSpan, layer::SubscriberExt, util::SubscriberInitExt}; +use tracing_subscriber::{fmt::format::FmtSpan, layer::SubscriberExt, util::SubscriberInitExt}; const GET_OBJECT_DURATION_HISTOGRAM_METRICS: &[&str] = &[ "rustfs_io_get_object_request_duration_seconds", @@ -83,6 +83,26 @@ const GET_OBJECT_DURATION_HISTOGRAM_BUCKETS: &[f64] = &[ 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, ]; +#[cfg(all( + feature = "pyroscope", + any(target_os = "macos", all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")) +))] +const REDACTED_PROFILING_ENDPOINT: &str = "[redacted]"; + +#[cfg(all( + feature = "pyroscope", + any(target_os = "macos", all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")) +))] +fn log_profiler_failure(result: &'static str, error_kind: &'static str) { + warn!( + backend = "pyroscope", + endpoint = REDACTED_PROFILING_ENDPOINT, + result, + error_kind, + "Profiling export agent initialization failed" + ); +} + /// Initialize the full OpenTelemetry HTTP pipeline (traces + metrics + logs). /// /// This function is invoked when at least one OTLP endpoint has been @@ -158,9 +178,7 @@ pub(super) fn init_observability_http( logger_provider = build_logger_provider(&log_ep, config, res, use_stdout)?; // Build bridge to capture `tracing` events. - otel_bridge = logger_provider - .as_ref() - .map(|p| OpenTelemetryTracingBridge::new(p).with_filter(build_env_filter(logger_level, None))); + otel_bridge = logger_provider.as_ref().map(OpenTelemetryTracingBridge::new); // No separate formatting layer is added here; when OTLP logging is // active, the OpenTelemetry bridge is the authoritative sink for @@ -256,6 +274,7 @@ pub(super) fn init_observability_http( let filter = build_env_filter(logger_level, None); tracing_subscriber::registry() .with(filter) + .with(pyroscope_log_filter()) .with(ErrorLayer::default()) .with(file_layer_opt) .with(stdout_layer_opt) @@ -531,6 +550,11 @@ pub(super) fn init_profiler(config: &OtelConfig) -> Option { return None; }; + if url::Url::parse(endpoint).is_err() { + log_profiler_failure("profiling_endpoint_invalid", "invalid_endpoint"); + return None; + } + // Configure Pyroscope Agent let backend = pprof_backend(PprofConfig::default(), BackendConfig::default()); let service_name = config.service_name.as_deref().unwrap_or(APP_NAME); @@ -542,28 +566,16 @@ pub(super) fn init_profiler(config: &OtelConfig) -> Option { .build() { Ok(agent) => agent, - Err(err) => { - warn!( - backend = "pyroscope", - endpoint, - result = "profiling_agent_build_failed", - error = %err, - "Profiling export agent initialization failed" - ); + Err(_) => { + log_profiler_failure("profiling_agent_build_failed", "build"); return None; } }; match agent.start() { Ok(agent) => Some(agent), - Err(err) => { - warn!( - backend = "pyroscope", - endpoint, - result = "profiling_agent_start_failed", - error = ?err, - "Profiling export agent failed to start" - ); + Err(_) => { + log_profiler_failure("profiling_agent_start_failed", "start"); None } } @@ -650,6 +662,34 @@ fn resolve_signal_timeout(common_timeout_millis: Option, signal_timeout_mil #[cfg(test)] mod tests { use super::*; + use std::env; + use std::io::{self, Write}; + use std::process::Command; + use std::sync::{Arc, Mutex}; + + const LOG_TRACER_CHILD_ENV: &str = "RUSTFS_OBS_LOG_TRACER_CHILD"; + + #[derive(Clone)] + struct TestWriter(Arc>>); + + impl Write for TestWriter { + fn write(&mut self, bytes: &[u8]) -> io::Result { + self.0.lock().expect("lock test log buffer").extend_from_slice(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } + } + + impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for TestWriter { + type Writer = Self; + + fn make_writer(&'a self) -> Self::Writer { + self.clone() + } + } #[test] /// Valid ratios should produce trace-id-ratio sampling. @@ -766,4 +806,107 @@ mod tests { assert_eq!(GET_OBJECT_DURATION_HISTOGRAM_BUCKETS.first(), Some(&0.0001)); assert_eq!(GET_OBJECT_DURATION_HISTOGRAM_BUCKETS.last(), Some(&10.0)); } + + #[cfg(all( + feature = "pyroscope", + any(target_os = "macos", all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")) + ))] + #[test] + fn test_init_profiler_invalid_endpoint_redacts_sensitive_components() { + let endpoint = "https://profile-user:profile-token@10.24.0.5:invalid/private/profiles?access_token=query-secret"; + let buffer = Arc::new(Mutex::new(Vec::new())); + let writer = TestWriter(Arc::clone(&buffer)); + let subscriber = tracing_subscriber::fmt() + .with_ansi(false) + .without_time() + .with_writer(writer) + .finish(); + + let config = OtelConfig { + profiling_export_enabled: Some(true), + profiling_endpoint: Some(endpoint.to_string()), + ..OtelConfig::default() + }; + + tracing::subscriber::with_default(subscriber, || assert!(init_profiler(&config).is_none())); + + let rendered = String::from_utf8(buffer.lock().expect("lock test log buffer").clone()).expect("decode test log"); + assert!(rendered.contains(REDACTED_PROFILING_ENDPOINT)); + assert!(rendered.contains("error_kind=\"invalid_endpoint\"")); + for leaked in [ + "profile-user", + "profile-token", + "10.24.0.5", + "private/profiles", + "query-secret", + ] { + assert!(!rendered.contains(leaked), "profiling log leaked {leaked}: {rendered}"); + } + } + + #[test] + fn test_pyroscope_upload_failure_targets_stay_filtered_when_env_filter_enables_them() { + let endpoint = "https://profile-user:profile-token@10.24.0.5/private/profiles?access_token=query-secret"; + let buffer = Arc::new(Mutex::new(Vec::new())); + let writer = TestWriter(Arc::clone(&buffer)); + let env_filter = tracing_subscriber::EnvFilter::new("pyroscope=trace") + .add_directive("trace".parse().expect("parse trace filter directive")) + .add_directive("Pyroscope::Session=trace".parse().expect("parse Pyroscope filter directive")); + let subscriber = tracing_subscriber::registry() + .with(env_filter) + .with(pyroscope_log_filter()) + .with( + tracing_subscriber::fmt::layer() + .with_ansi(false) + .without_time() + .with_writer(writer), + ); + + tracing::subscriber::with_default(subscriber, || { + tracing::error!(target: "pyroscope::session", "SessionManager - Failed to send session: {endpoint}"); + tracing::error!(target: "Pyroscope::Session", "SessionManager - Failed to send session: {endpoint}"); + }); + + let rendered = String::from_utf8(buffer.lock().expect("lock test log buffer").clone()).expect("decode test log"); + assert!(rendered.is_empty(), "filtered Pyroscope upload failure reached a log sink: {rendered}"); + } + + #[test] + fn test_pyroscope_log_facade_upload_failure_is_filtered() { + let endpoint = "https://profile-user:profile-token@10.24.0.5/private/profiles?access_token=query-secret"; + if env::var_os(LOG_TRACER_CHILD_ENV).is_some() { + tracing_subscriber::registry() + .with(build_env_filter("info", None)) + .with(pyroscope_log_filter()) + .with(tracing_subscriber::fmt::layer().with_ansi(false).without_time()) + .try_init() + .expect("install isolated LogTracer subscriber"); + log::error!(target: "pyroscope::session", "SessionManager - Failed to send session: {endpoint}"); + log::error!(target: "Pyroscope::Session", "SessionManager - Failed to send session: {endpoint}"); + return; + } + + let output = Command::new(env::current_exe().expect("resolve test executable")) + .args([ + "--exact", + "telemetry::otel::tests::test_pyroscope_log_facade_upload_failure_is_filtered", + "--nocapture", + ]) + .env(LOG_TRACER_CHILD_ENV, "1") + .env("RUST_LOG", "trace,pyroscope=trace,Pyroscope::Session=trace") + .output() + .expect("run isolated LogTracer test process"); + assert!(output.status.success(), "isolated LogTracer test failed: {output:?}"); + + let rendered = format!("{}{}", String::from_utf8_lossy(&output.stdout), String::from_utf8_lossy(&output.stderr)); + for leaked in [ + "profile-user", + "profile-token", + "10.24.0.5", + "private/profiles", + "query-secret", + ] { + assert!(!rendered.contains(leaked), "LogTracer path leaked {leaked}: {rendered}"); + } + } } diff --git a/crates/policy/Cargo.toml b/crates/policy/Cargo.toml index 1732045fb..c06452ec7 100644 --- a/crates/policy/Cargo.toml +++ b/crates/policy/Cargo.toml @@ -32,7 +32,6 @@ workspace = true default = [] hotpath = [ "hotpath/hotpath", - "hotpath/reqwest-0-13", "rustfs-config/hotpath", "rustfs-credentials/hotpath", "rustfs-crypto/hotpath", @@ -80,6 +79,7 @@ pollster.workspace = true proptest = "1" test-case.workspace = true temp-env = { workspace = true } +tracing-subscriber = { workspace = true, features = ["fmt"] } [lib] doctest = false diff --git a/crates/policy/src/policy/opa.rs b/crates/policy/src/policy/opa.rs index c1f70bcbb..8fd1ab149 100644 --- a/crates/policy/src/policy/opa.rs +++ b/crates/policy/src/policy/opa.rs @@ -36,15 +36,10 @@ pub fn is_configured() -> bool { #[derive(Debug, Clone)] pub struct AuthZPlugin { - client: OpaHttpClient, + client: reqwest::Client, args: Args, } -#[cfg(feature = "hotpath")] -type OpaHttpClient = hotpath::wrap::reqwest::Client; -#[cfg(not(feature = "hotpath"))] -type OpaHttpClient = reqwest::Client; - #[derive(Debug, thiserror::Error)] pub enum OpaConfigError { #[error("Missing required env var: {0}")] @@ -63,6 +58,44 @@ pub enum OpaConfigError { Connection(reqwest::Error), } +impl OpaConfigError { + pub fn kind(&self) -> &'static str { + match self { + Self::MissingRequiredEnv(_) => "missing_required_env", + Self::InvalidEnvVars(_) => "invalid_env_vars", + Self::EnvRead { .. } => "env_read_failed", + Self::InvalidStatus(_) => "invalid_status", + Self::Connection(_) => "connection_failed", + } + } +} + +fn redact_opa_connection_error(error: reqwest::Error) -> OpaConfigError { + OpaConfigError::Connection(error.without_url()) +} + +fn opa_request_error_kind(error: &reqwest::Error) -> &'static str { + if error.is_timeout() { + "timeout" + } else if error.is_connect() { + "connect" + } else if error.is_request() { + "request" + } else if error.is_body() { + "body" + } else if error.is_decode() { + "decode" + } else if error.is_status() { + "status" + } else if error.is_builder() { + "builder" + } else if error.is_redirect() { + "redirect" + } else { + "other" + } +} + fn check() -> Result<(), OpaConfigError> { let env_list = env::vars(); let mut candidate = HashMap::new(); @@ -90,7 +123,7 @@ async fn validate(config: &Args) -> Result<(), OpaConfigError> { .timeout(Duration::from_secs(5)) .connect_timeout(Duration::from_secs(1)) .build() - .map_err(OpaConfigError::Connection)?; + .map_err(redact_opa_connection_error)?; let mut request = client.post(&config.url); if !config.auth_token.is_empty() { @@ -109,7 +142,7 @@ async fn validate(config: &Args) -> Result<(), OpaConfigError> { }; } Err(err) => { - return Err(OpaConfigError::Connection(err)); + return Err(redact_opa_connection_error(err)); } }; Ok(()) @@ -155,9 +188,6 @@ impl AuthZPlugin { reqwest::Client::new() }); - #[cfg(feature = "hotpath")] - let client = hotpath::http!(client, label = "Policy::OPA"); - Self { client, args: config } } @@ -173,7 +203,13 @@ impl AuthZPlugin { Ok(resp) => { let status = resp.status(); if !status.is_success() { - error!("OPA returned non-success status: {}", status); + error!( + component = "policy", + subsystem = "opa", + result = "request_rejected", + status = %status, + "OPA request rejected" + ); return false; } @@ -183,13 +219,25 @@ impl AuthZPlugin { OpaResponseEnum::AllowResult(response) => response.result.allow, }, Err(err) => { - error!("Error parsing OPA response: {:?}", err); + error!( + component = "policy", + subsystem = "opa", + result = "response_decode_failed", + error_kind = opa_request_error_kind(&err), + "OPA response decoding failed" + ); false } } } Err(err) => { - error!("Error sending request to OPA: {:?}", err); + error!( + component = "policy", + subsystem = "opa", + result = "request_failed", + error_kind = opa_request_error_kind(&err), + "OPA request failed" + ); false } } @@ -258,8 +306,35 @@ enum OpaResponseEnum { #[cfg(test)] mod tests { use super::*; + use std::io::{Read, Write}; + use std::net::TcpListener; + use std::sync::{Arc, Mutex}; use temp_env; + #[derive(Clone)] + struct TestWriter(Arc>>); + + impl Write for TestWriter { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0.lock().expect("lock test log buffer").extend_from_slice(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for TestWriter { + type Writer = Self; + + fn make_writer(&'a self) -> Self::Writer { + self.clone() + } + } + + fn assert_reqwest_client(_: &reqwest::Client) {} + #[test] fn test_check_valid_config() { // Use temp_env to temporarily set environment variables @@ -345,4 +420,130 @@ mod tests { }; assert!(!args_disabled.enable()); } + + #[test] + fn test_sensitive_opa_endpoints_never_use_hotpath_http_wrapper() { + let endpoints = [ + "https://profile-user:profile-token@10.24.0.5/private/opa?access_token=query-secret", + "https://service.internal.example/private/path?token=another-secret", + ]; + + for endpoint in endpoints { + let plugin = AuthZPlugin::new(Args { + url: endpoint.to_string(), + auth_token: "opa-auth-token".to_string(), + }); + + assert_reqwest_client(&plugin.client); + assert_eq!(plugin.args.url, endpoint); + } + } + + #[test] + fn test_opa_connection_error_removes_sensitive_endpoint() { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind local test listener"); + let port = listener.local_addr().expect("read local test listener address").port(); + let server = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().expect("accept local test request"); + let mut request = [0_u8; 1024]; + let read = stream.read(&mut request).expect("read local test request"); + assert!(read > 0, "local test client must send an HTTP request"); + stream + .write_all(b"HTTP/1.1 500 Internal Server Error\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") + .expect("write local test response"); + }); + let endpoint = format!("http://profile-user:profile-token@127.0.0.1:{port}/private/opa?access_token=query-secret"); + let runtime = tokio::runtime::Runtime::new().expect("build tokio test runtime"); + let error = runtime.block_on(async { + reqwest::Client::builder() + .no_proxy() + .build() + .expect("build local test client") + .get(&endpoint) + .send() + .await + .expect("send local test request") + .error_for_status() + .expect_err("test response should be an HTTP error") + }); + server.join().expect("join local test server"); + assert!(error.url().is_some(), "fixture must carry the endpoint before redaction"); + + let OpaConfigError::Connection(error) = redact_opa_connection_error(error) else { + panic!("expected a redacted OPA connection error"); + }; + let rendered = error.to_string(); + assert!(error.url().is_none()); + for leaked in ["profile-user", "profile-token", "127.0.0.1", "private/opa", "query-secret"] { + assert!(!rendered.contains(leaked), "redacted error leaked {leaked}: {rendered}"); + } + } + + #[test] + fn test_is_allowed_rejection_log_redacts_sensitive_endpoint() { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind local test listener"); + let port = listener.local_addr().expect("read local test listener address").port(); + let server = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().expect("accept local test request"); + let mut request = [0_u8; 1024]; + let read = stream.read(&mut request).expect("read local test request"); + assert!(read > 0, "local test client must send an HTTP request"); + stream + .write_all(b"HTTP/1.1 500 Internal Server Error\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") + .expect("write local test response"); + }); + let endpoint = format!("http://opa-user:opa-token@127.0.0.1:{port}/private/allow?access_token=query-secret"); + let plugin = AuthZPlugin::new(Args { + url: endpoint, + auth_token: "authorization-secret".to_string(), + }); + let groups = None; + let conditions = HashMap::new(); + let claims = HashMap::new(); + let args = PArgs { + account: "account", + groups: &groups, + action: crate::policy::action::Action::None, + bucket: "bucket", + conditions: &conditions, + is_owner: false, + object: "object", + claims: &claims, + deny_only: false, + }; + let buffer = Arc::new(Mutex::new(Vec::new())); + let writer = TestWriter(Arc::clone(&buffer)); + let subscriber = tracing_subscriber::fmt() + .with_ansi(false) + .without_time() + .with_writer(writer) + .finish(); + + tracing::subscriber::with_default(subscriber, || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build tokio test runtime"); + assert!(!runtime.block_on(plugin.is_allowed(&args))); + }); + server.join().expect("join local test server"); + + let rendered = String::from_utf8(buffer.lock().expect("lock test log buffer").clone()).expect("decode test log"); + assert!(!rendered.is_empty(), "OPA request failure did not reach the test log sink"); + assert!( + rendered.contains("request_rejected"), + "OPA request failure did not retain its stable result category" + ); + assert!(rendered.contains("500 Internal Server Error")); + for leaked in [ + "opa-user", + "opa-token", + "127.0.0.1", + "private/allow", + "query-secret", + "authorization-secret", + ] { + assert!(!rendered.contains(leaked), "OPA request log leaked {leaked}: {rendered}"); + } + } }