diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index 4414291cb..8eb8deec5 100644 --- a/crates/io-metrics/src/lib.rs +++ b/crates/io-metrics/src/lib.rs @@ -294,6 +294,18 @@ pub const GET_OBJECT_SIZE_BUCKET_LE_256_KIB: &str = "le_256kib"; pub const GET_OBJECT_SIZE_BUCKET_LE_512_KIB: &str = "le_512kib"; pub const GET_OBJECT_SIZE_BUCKET_LE_1_MIB: &str = "le_1mib"; pub const GET_OBJECT_SIZE_BUCKET_GT_1_MIB: &str = "gt_1mib"; +pub const GET_OBJECT_SIZE_BUCKET_UNKNOWN: &str = "unknown"; + +pub struct GetObjectStreamingBodyFailure { + pub stage: &'static str, + pub reason: &'static str, + pub error_class: &'static str, + pub strategy: &'static str, + pub buffer_source: &'static str, + pub size_bucket: &'static str, + pub emitted_bytes: usize, + pub remaining_bytes: usize, +} /// Return the bounded size bucket used by small-object GET diagnostics. #[inline(always)] @@ -592,6 +604,44 @@ pub fn record_get_object_reader_stream_poll( .record(duration_secs); } +/// Record a GET response body failure with bounded attribution labels. +#[inline(always)] +pub fn record_get_object_streaming_body_failure(failure: GetObjectStreamingBodyFailure) { + if !metrics_enabled() { + return; + } + counter!( + "rustfs_io_get_object_streaming_body_failure_total", + "stage" => failure.stage, + "reason" => failure.reason, + "error_class" => failure.error_class, + "strategy" => failure.strategy, + "buffer_source" => failure.buffer_source, + "size_bucket" => failure.size_bucket + ) + .increment(1); + histogram!( + "rustfs_io_get_object_streaming_body_failure_emitted_bytes", + "stage" => failure.stage, + "reason" => failure.reason, + "error_class" => failure.error_class, + "strategy" => failure.strategy, + "buffer_source" => failure.buffer_source, + "size_bucket" => failure.size_bucket + ) + .record(usize_to_f64(failure.emitted_bytes)); + histogram!( + "rustfs_io_get_object_streaming_body_failure_remaining_bytes", + "stage" => failure.stage, + "reason" => failure.reason, + "error_class" => failure.error_class, + "strategy" => failure.strategy, + "buffer_source" => failure.buffer_source, + "size_bucket" => failure.size_bucket + ) + .record(usize_to_f64(failure.remaining_bytes)); +} + /// Record a poll of the single-chunk in-memory GetObject handoff stream. #[inline(always)] pub fn record_get_object_memory_body_stream_poll(source: &'static str, outcome: &'static str, bytes: usize, duration_secs: f64) { @@ -2914,6 +2964,16 @@ mod tests { record_list_objects(50.0, 100, false); record_error("get_object", "timeout"); record_cpu_usage(25.5); + record_get_object_streaming_body_failure(GetObjectStreamingBodyFailure { + stage: "reader_stream", + reason: "short_eof", + error_class: "short_eof", + strategy: "standard", + buffer_source: "selected", + size_bucket: GET_OBJECT_SIZE_BUCKET_GT_1_MIB, + emitted_bytes: 1024, + remaining_bytes: 512, + }); // Enabled: the same recorders run their emission bodies without panicking. set_metrics_enabled(true); @@ -2922,6 +2982,16 @@ mod tests { record_list_objects(50.0, 100, false); record_error("get_object", "timeout"); record_cpu_usage(25.5); + record_get_object_streaming_body_failure(GetObjectStreamingBodyFailure { + stage: "reader_stream", + reason: "reader_error", + error_class: "timeout", + strategy: "standard", + buffer_source: "selected", + size_bucket: GET_OBJECT_SIZE_BUCKET_GT_1_MIB, + emitted_bytes: 2048, + remaining_bytes: 256, + }); set_metrics_enabled(false); } diff --git a/docs/operations/hotpath-warp-abba-runbook.md b/docs/operations/hotpath-warp-abba-runbook.md index 5d73b59d1..7b0e290cf 100644 --- a/docs/operations/hotpath-warp-abba-runbook.md +++ b/docs/operations/hotpath-warp-abba-runbook.md @@ -184,6 +184,22 @@ perf record -F 99 -g -- sleep 180 perf report --stdio > target/hotpath-abba/cluster-pr-XXXX/telemetry/perf-report.txt ``` +When using samply against an already-running RustFS service, attach through the +bounded helper instead of calling `samply record -p` directly: + +```bash +scripts/run_samply_attach_window.sh \ + --pid "$RUSTFS_PID" \ + --duration-secs 180 \ + --output target/hotpath-abba/cluster-pr-XXXX/telemetry/samply-A1-get-4mib.json.gz +``` + +Run one attach window per ABBA leg or focused verification cell. After every +window, confirm that the `.json.gz` profile and `.syms.json` sidecar are +non-empty, that no `samply` process is still attached to the RustFS PID, and +that any temporary `perf_event_paranoid` change has been restored before the +next cell starts. + For allocation profiling, build the candidate with: ```bash @@ -267,11 +283,14 @@ The AI agent should execute this sequence: CPU model, memory size, disk layout, and whether the run is local or cluster. 2. Run `scripts/run_hotpath_warp_abba.sh --dry-run` with the final arguments. 3. Run the real ABBA command with `--rounds >= 3`. -4. Preserve the full output directory without editing generated CSV files. -5. Read `summary.md`, `candidate_gate.md`, and `baseline_drift_gate.md`. -6. Summarize only measured facts: candidate deltas, baseline drift, CPU or +4. For samply CPU attribution, use `scripts/run_samply_attach_window.sh` for + each bounded attach window and reject the cell if the profile is empty or a + stale `samply` process remains. +5. Preserve the full output directory without editing generated CSV files. +6. Read `summary.md`, `candidate_gate.md`, and `baseline_drift_gate.md`. +7. Summarize only measured facts: candidate deltas, baseline drift, CPU or memory saturation, and any failed workloads. -7. Post the summary and artifact location to the tracking issue or PR. +8. Post the summary and artifact location to the tracking issue or PR. Do not report a performance win or loss when the baseline drift gate failed on the same workload and no rerun was collected. diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 865e38037..48cba8ccb 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -719,6 +719,9 @@ const GET_READER_STREAM_POLL_PENDING: &str = "pending"; const GET_READER_STREAM_POLL_READY_DATA: &str = "ready_data"; const GET_READER_STREAM_POLL_READY_EMPTY: &str = "ready_empty"; const GET_READER_STREAM_POLL_READY_ERROR: &str = "ready_error"; +const GET_STREAMING_BODY_FAILURE_STAGE_READER_STREAM: &str = "reader_stream"; +const GET_STREAMING_BODY_FAILURE_REASON_READER_ERROR: &str = "reader_error"; +const GET_STREAMING_BODY_FAILURE_REASON_SHORT_EOF: &str = "short_eof"; const GET_MEMORY_BODY_SOURCE_BUFFERED_BODY: &str = "buffered_body"; const GET_MEMORY_BODY_SOURCE_OBJECT_DATA_CACHE: &str = "object_data_cache"; const GET_MEMORY_BODY_SOURCE_OBJECT_DATA_CACHE_MATERIALIZED: &str = "object_data_cache_materialized"; @@ -774,6 +777,66 @@ fn tune_reader_stream_buffer_size( selected_size } +fn get_object_stream_size_bucket(expected: usize) -> &'static str { + rustfs_io_metrics::get_object_size_bucket(i64::try_from(expected).unwrap_or(i64::MAX)) +} + +fn classify_get_object_stream_read_error(err: &std::io::Error) -> &'static str { + if let Some(inner) = err.get_ref() { + if inner.is::() { + return "short_eof"; + } + + if inner.is::() { + return "bitrot"; + } + + let error_msg = inner.to_string().to_lowercase(); + if error_msg.contains("bitrot") { + return "bitrot"; + } + if error_msg.contains("read quorum") || error_msg.contains("insufficient read quorum") || error_msg.contains("erasure") { + return "read_quorum"; + } + } + + match err.kind() { + std::io::ErrorKind::UnexpectedEof => "short_eof", + std::io::ErrorKind::TimedOut => "timeout", + std::io::ErrorKind::InvalidInput | std::io::ErrorKind::InvalidData => "range_or_length_invalid", + _ => "io", + } +} + +fn get_object_stream_failure_reason(error_class: &'static str) -> &'static str { + if error_class == "short_eof" { + GET_STREAMING_BODY_FAILURE_REASON_SHORT_EOF + } else { + GET_STREAMING_BODY_FAILURE_REASON_READER_ERROR + } +} + +fn record_get_object_reader_stream_failure( + reason: &'static str, + error_class: &'static str, + strategy: &'static str, + buffer_source: &'static str, + expected: usize, + emitted: usize, + remaining: usize, +) { + rustfs_io_metrics::record_get_object_streaming_body_failure(rustfs_io_metrics::GetObjectStreamingBodyFailure { + stage: GET_STREAMING_BODY_FAILURE_STAGE_READER_STREAM, + reason, + error_class, + strategy, + buffer_source, + size_bucket: get_object_stream_size_bucket(expected), + emitted_bytes: emitted, + remaining_bytes: remaining, + }); +} + pin_project! { struct ExtractArchiveEtagReader { #[pin] @@ -1358,9 +1421,9 @@ where Poll::Ready(Ok(bytes_read)) if bytes_read > 0 => { let bytes = buf.freeze(); *this.remaining -= bytes.len(); + *this.emitted += bytes.len(); #[cfg(feature = "tracing-chunk-debug")] { - *this.emitted += bytes.len(); tracing::debug!( emitted = *this.emitted, expected = *this.expected, @@ -1378,6 +1441,15 @@ where this.reader.set(None); let remaining = i64::try_from(*this.remaining).unwrap_or(i64::MAX); let err = std::io::Error::new(std::io::ErrorKind::UnexpectedEof, rustfs_rio::IncompleteBody { remaining }); + record_get_object_reader_stream_failure( + GET_STREAMING_BODY_FAILURE_REASON_SHORT_EOF, + "short_eof", + this.strategy, + this.buffer_source, + *this.expected, + *this.emitted, + *this.remaining, + ); #[cfg(feature = "tracing-chunk-debug")] tracing::error!( emitted = *this.emitted, @@ -1389,6 +1461,16 @@ where } Poll::Ready(Err(err)) => { this.reader.set(None); + let error_class = classify_get_object_stream_read_error(&err); + record_get_object_reader_stream_failure( + get_object_stream_failure_reason(error_class), + error_class, + this.strategy, + this.buffer_source, + *this.expected, + *this.emitted, + *this.remaining, + ); #[cfg(feature = "tracing-chunk-debug")] tracing::error!( emitted = *this.emitted, @@ -1445,8 +1527,6 @@ where struct GetObjectStreamingReader { inner: Option, - bucket: String, - key: String, // request_id + optional content_range are only used for diagnostic correlation and // failure bucketing; they do not alter stream behavior. request_id: String, @@ -1467,8 +1547,8 @@ impl GetObjectStreamingReader { #[allow(clippy::too_many_arguments)] fn new( inner: R, - bucket: &str, - key: &str, + _bucket: &str, + _key: &str, request_id: &str, content_range: Option, expected: usize, @@ -1478,8 +1558,6 @@ impl GetObjectStreamingReader { ) -> Self { Self { inner: Some(inner), - bucket: bucket.to_string(), - key: key.to_string(), request_id: request_id.to_string(), content_range, expected, @@ -1503,33 +1581,7 @@ impl GetObjectStreamingReader { // distinguish truncated upstream bodies, corruption, quorum issues, and // genuine downstream-close disconnects. fn classify_read_error(err: &std::io::Error) -> &'static str { - if let Some(inner) = err.get_ref() { - if inner.is::() { - return "short_eof"; - } - - if inner.is::() { - return "bitrot"; - } - - let error_msg = inner.to_string().to_lowercase(); - if error_msg.contains("bitrot") { - return "bitrot"; - } - if error_msg.contains("read quorum") - || error_msg.contains("insufficient read quorum") - || error_msg.contains("erasure") - { - return "read_quorum"; - } - } - - match err.kind() { - std::io::ErrorKind::UnexpectedEof => "short_eof", - std::io::ErrorKind::TimedOut => "timeout", - std::io::ErrorKind::InvalidInput | std::io::ErrorKind::InvalidData => "range_or_length_invalid", - _ => "io", - } + classify_get_object_stream_read_error(err) } fn finish_ok(&mut self) { @@ -1646,10 +1698,9 @@ impl GetObjectStreamingReader { event = EVENT_GET_OBJECT_STREAM_BODY, component = LOG_COMPONENT_APP, subsystem = LOG_SUBSYSTEM_OBJECT, - bucket = %self.bucket, - object = %self.key, request_id = %self.request_id, range = %self.content_range.as_deref().unwrap_or("full"), + size_bucket = get_object_stream_size_bucket(self.expected), expected = self.expected, emitted = self.emitted, elapsed_ms = self.elapsed().as_millis(), @@ -1683,10 +1734,9 @@ impl AsyncRead for GetObjectStreamingReader { event = EVENT_GET_OBJECT_STREAM_BODY, component = LOG_COMPONENT_APP, subsystem = LOG_SUBSYSTEM_OBJECT, - bucket = %self.bucket, - object = %self.key, request_id = %self.request_id, range = %self.content_range.as_deref().unwrap_or("full"), + size_bucket = get_object_stream_size_bucket(self.expected), expected = self.expected, emitted = self.emitted, resume_attempts = attempts, @@ -1706,10 +1756,9 @@ impl AsyncRead for GetObjectStreamingReader { event = EVENT_GET_OBJECT_STREAM_BODY, component = LOG_COMPONENT_APP, subsystem = LOG_SUBSYSTEM_OBJECT, - bucket = %self.bucket, - object = %self.key, request_id = %self.request_id, range = %self.content_range.as_deref().unwrap_or("full"), + size_bucket = get_object_stream_size_bucket(self.expected), expected = self.expected, emitted = self.emitted, elapsed_ms = self.elapsed().as_millis(), @@ -1748,10 +1797,9 @@ impl AsyncRead for GetObjectStreamingReader { event = EVENT_GET_OBJECT_STREAM_BODY, component = LOG_COMPONENT_APP, subsystem = LOG_SUBSYSTEM_OBJECT, - bucket = %self.bucket, - object = %self.key, request_id = %self.request_id, range = %self.content_range.as_deref().unwrap_or("full"), + size_bucket = get_object_stream_size_bucket(self.expected), expected = self.expected, emitted = self.emitted, elapsed_ms = elapsed.as_millis(), @@ -1793,10 +1841,9 @@ impl AsyncRead for GetObjectStreamingReader { event = EVENT_GET_OBJECT_STREAM_BODY, component = LOG_COMPONENT_APP, subsystem = LOG_SUBSYSTEM_OBJECT, - bucket = %self.bucket, - object = %self.key, request_id = %self.request_id, range = %self.content_range.as_deref().unwrap_or("full"), + size_bucket = get_object_stream_size_bucket(self.expected), expected = self.expected, emitted = self.emitted, elapsed_ms = self.elapsed().as_millis(), @@ -1828,10 +1875,9 @@ impl AsyncRead for GetObjectStreamingReader { event = EVENT_GET_OBJECT_STREAM_BODY, component = LOG_COMPONENT_APP, subsystem = LOG_SUBSYSTEM_OBJECT, - bucket = %self.bucket, - object = %self.key, request_id = %self.request_id, range = %self.content_range.as_deref().unwrap_or("full"), + size_bucket = get_object_stream_size_bucket(self.expected), expected = self.expected, emitted = self.emitted, elapsed_ms = self.elapsed().as_millis(), @@ -1864,10 +1910,9 @@ impl Drop for GetObjectStreamingReader { event = EVENT_GET_OBJECT_STREAM_BODY, component = LOG_COMPONENT_APP, subsystem = LOG_SUBSYSTEM_OBJECT, - bucket = %self.bucket, - object = %self.key, request_id = %self.request_id, range = %self.content_range.as_deref().unwrap_or("full"), + size_bucket = get_object_stream_size_bucket(self.expected), expected = self.expected, emitted = self.emitted, elapsed_ms = self.elapsed().as_millis(), @@ -15306,6 +15351,19 @@ mod tests { ); } + #[test] + fn get_object_stream_failure_labels_are_low_cardinality() { + assert_eq!(get_object_stream_failure_reason("short_eof"), GET_STREAMING_BODY_FAILURE_REASON_SHORT_EOF); + assert_eq!( + get_object_stream_failure_reason("timeout"), + GET_STREAMING_BODY_FAILURE_REASON_READER_ERROR + ); + assert_eq!( + get_object_stream_size_bucket(4 * 1024 * 1024), + rustfs_io_metrics::GET_OBJECT_SIZE_BUCKET_GT_1_MIB + ); + } + #[tokio::test] async fn disk_read_permit_reader_releases_permit_at_eof() { use tokio::io::AsyncReadExt; diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index c45c25ab4..2711501ce 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -88,6 +88,7 @@ use tonic::{Request, Status}; use tower::{Service, ServiceBuilder}; use tower_http::add_extension::AddExtensionLayer; use tower_http::catch_panic::CatchPanicLayer; +use tower_http::classify::ServerErrorsFailureClass; use tower_http::compression::CompressionLayer; use tower_http::request_id::{MakeRequestUuid, PropagateRequestIdLayer, SetRequestIdLayer}; use tower_http::trace::TraceLayer; @@ -108,6 +109,10 @@ const METRIC_HTTP_SERVER_RESPONSE_BODY_CHUNK_SIZE_BYTES: &str = "rustfs_http_ser const METRIC_HTTP_SERVER_RESPONSE_BODY_CHUNK_LATENCY_SECONDS: &str = "rustfs_http_server_response_body_chunk_latency_seconds"; const METRIC_HTTP_SERVER_RESPONSE_BODY_STREAM_DURATION_SECONDS: &str = "rustfs_http_server_response_body_stream_duration_seconds"; const METRIC_HTTP_SERVER_CONNECTION_CAP_SATURATED_TOTAL: &str = "rustfs_http_server_connection_cap_saturated_total"; +const HTTP_STREAMING_BODY_FAILURE_STAGE_TRANSPORT: &str = "http_transport"; +const HTTP_STREAMING_BODY_FAILURE_REASON_TRANSPORT: &str = "transport_failure"; +const HTTP_STREAMING_BODY_FAILURE_CLASS_TRANSPORT: &str = "transport"; +const HTTP_STREAMING_BODY_FAILURE_UNKNOWN: &str = "unknown"; /// Cached handle for the per-response-body-chunk byte counter. A streamed GET /// emits many chunks, so resolving the `counter!` registry entry once — the @@ -357,6 +362,27 @@ fn record_active_http_requests(delta: i64) { gauge!(METRIC_HTTP_SERVER_ACTIVE_REQUESTS).set(next as f64); } +#[inline] +fn record_http_transport_streaming_body_failure() { + rustfs_io_metrics::record_get_object_streaming_body_failure(rustfs_io_metrics::GetObjectStreamingBodyFailure { + stage: HTTP_STREAMING_BODY_FAILURE_STAGE_TRANSPORT, + reason: HTTP_STREAMING_BODY_FAILURE_REASON_TRANSPORT, + error_class: HTTP_STREAMING_BODY_FAILURE_CLASS_TRANSPORT, + strategy: HTTP_STREAMING_BODY_FAILURE_UNKNOWN, + buffer_source: HTTP_STREAMING_BODY_FAILURE_UNKNOWN, + size_bucket: rustfs_io_metrics::GET_OBJECT_SIZE_BUCKET_UNKNOWN, + emitted_bytes: 0, + remaining_bytes: 0, + }); +} + +#[inline] +fn record_http_transport_failure_if_body_error(error: &ServerErrorsFailureClass) { + if matches!(error, ServerErrorsFailureClass::Error(_)) { + record_http_transport_streaming_body_failure(); + } +} + pub(crate) fn active_http_requests() -> u64 { ACTIVE_HTTP_REQUESTS.load(Ordering::Relaxed) } @@ -1591,6 +1617,7 @@ fn process_connection( LABEL_HTTP_STATUS_CLASS => "transport" ) .increment(1); + record_http_transport_failure_if_body_error(&error); trace!(error = ?error, duration_ms = duration_ms(latency), "HTTP request failure captured by trace layer"); }), ) @@ -1688,6 +1715,7 @@ fn process_connection( LABEL_HTTP_STATUS_CLASS => "transport" ) .increment(1); + record_http_transport_failure_if_body_error(&error); trace!(error = ?error, duration_ms = duration_ms(latency), "HTTP request failure captured by trace layer"); }), )