From ed73952cb635fac9c42104f279d990fbb2ba4556 Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 8 Jun 2026 17:45:56 +0800 Subject: [PATCH] perf(ecstore): improve erasure write diagnostics and single-block performance (#3280) * docs(object-capacity): add localized crate docs * fix(ecstore): improve quorum and transport diagnostics * perf(ecstore): add safe single-block write fast path * refactor(ecstore): collapse layered small write paths * chore(docs): keep issue 662 design note tracked * fix(docs): restore issue 662 design note * chore(docs): keep issue 662 design local only * feat(obs): add internode reliability metrics and dashboard * feat(obs): extend internode diagnostics and service logging * fix(docs): use AGENTS guide filename * perf(ecstore): reuse owned buffer in small encode * fix(ecstore): tighten small write diagnostics --------- Co-authored-by: cxymds --- .../grafana/dashboards/rustfs.json | 410 ++++++++++++++++++ .gitignore | 1 + crates/ecstore/Cargo.toml | 4 + .../single_block_non_inline_benchmark.rs | 126 ++++++ crates/ecstore/src/disk/error.rs | 26 +- crates/ecstore/src/disk/error_reduce.rs | 118 +++++ crates/ecstore/src/erasure_coding/encode.rs | 240 ++++++++-- crates/ecstore/src/rpc/remote_disk.rs | 189 +++++++- crates/ecstore/src/set_disk.rs | 108 ++++- crates/io-metrics/README.md | 19 + crates/io-metrics/src/internode_metrics.rs | 152 +++++-- crates/object-capacity/AGENTS.md | 87 ++++ crates/object-capacity/README.md | 353 +++++++++++++++ crates/object-capacity/README_ZH.md | 353 +++++++++++++++ crates/rio/src/http_reader.rs | 344 ++++++++++++++- rustfs/src/storage/rpc/http_service.rs | 48 +- 16 files changed, 2455 insertions(+), 123 deletions(-) create mode 100644 crates/ecstore/benches/single_block_non_inline_benchmark.rs create mode 100644 crates/object-capacity/AGENTS.md create mode 100644 crates/object-capacity/README.md create mode 100644 crates/object-capacity/README_ZH.md diff --git a/.docker/observability/grafana/dashboards/rustfs.json b/.docker/observability/grafana/dashboards/rustfs.json index 2830693ee..04ae9eed4 100644 --- a/.docker/observability/grafana/dashboards/rustfs.json +++ b/.docker/observability/grafana/dashboards/rustfs.json @@ -9627,6 +9627,416 @@ ], "title": "Cluster Config / API Rejected (All)", "type": "timeseries" + }, + { + "collapsed": false, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 262 + }, + "id": 507, + "panels": [], + "title": "Internode Reliability", + "type": "row" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": true, + "axisColorMode": "text", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "lineInterpolation": "smooth", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + } + ] + }, + "unit": "ops" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 263 + }, + "id": 508, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "right", + "showLegend": true + }, + "tooltip": { + "hideZeros": false, + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "12.3.2", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by (operation, classification) (rate(rustfs_system_network_internode_operation_classified_errors_total{job=~\"$job\", operation=~\"put_file_stream|walk_dir|read_file_stream\"}[$__rate_interval]))", + "legendFormat": "{{operation}} | {{classification}}", + "range": true, + "refId": "A" + } + ], + "title": "Internode Classified Errors", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": true, + "axisColorMode": "text", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "lineInterpolation": "smooth", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + } + ] + }, + "unit": "ops" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 263 + }, + "id": 509, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "right", + "showLegend": true + }, + "tooltip": { + "hideZeros": false, + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "12.3.2", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by (operation, classification) (rate(rustfs_system_network_internode_operation_retries_total{job=~\"$job\", operation=\"put_file_stream\"}[$__rate_interval]))", + "legendFormat": "retry | {{classification}}", + "range": true, + "refId": "A" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by (operation, classification) (rate(rustfs_system_network_internode_operation_retry_successes_total{job=~\"$job\", operation=\"put_file_stream\"}[$__rate_interval]))", + "legendFormat": "retry success | {{classification}}", + "range": true, + "refId": "B" + } + ], + "title": "Internode Open Write Retries", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "lineInterpolation": "smooth", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "max": 1, + "min": 0, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + }, + { + "color": "yellow", + "value": 0.25 + }, + { + "color": "red", + "value": 0.5 + } + ] + }, + "unit": "percentunit" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 271 + }, + "id": 510, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "right", + "showLegend": true + }, + "tooltip": { + "hideZeros": false, + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "12.3.2", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by (classification) (rate(rustfs_system_network_internode_operation_retry_successes_total{job=~\"$job\", operation=\"put_file_stream\"}[$__rate_interval])) / clamp_min(sum by (classification) (rate(rustfs_system_network_internode_operation_retries_total{job=~\"$job\", operation=\"put_file_stream\"}[$__rate_interval])), 1e-9)", + "legendFormat": "{{classification}}", + "range": true, + "refId": "A" + } + ], + "title": "Internode Retry Success Ratio", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": true, + "axisColorMode": "text", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "lineInterpolation": "smooth", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + } + ] + }, + "unit": "ops" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 271 + }, + "id": 511, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "right", + "showLegend": true + }, + "tooltip": { + "hideZeros": false, + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "12.3.2", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by (stage, dominant_error) (rate(rustfs_system_storage_erasure_write_quorum_failures_total{job=~\"$job\"}[$__rate_interval]))", + "legendFormat": "{{stage}} | {{dominant_error}}", + "range": true, + "refId": "A" + } + ], + "title": "Erasure Write Quorum Failures", + "type": "timeseries" } ], "preload": false, diff --git a/.gitignore b/.gitignore index a94697c58..d4acd5af1 100644 --- a/.gitignore +++ b/.gitignore @@ -57,3 +57,4 @@ rustfs-webdav.code-workspace # Local test/benchmark artifacts benchmarks.logs tmp/ +crates/*/docs diff --git a/crates/ecstore/Cargo.toml b/crates/ecstore/Cargo.toml index bb40c626c..fe6ca2b9f 100644 --- a/crates/ecstore/Cargo.toml +++ b/crates/ecstore/Cargo.toml @@ -153,5 +153,9 @@ harness = false name = "rename_data_meta_benchmark" harness = false +[[bench]] +name = "single_block_non_inline_benchmark" +harness = false + [lib] doctest = false diff --git a/crates/ecstore/benches/single_block_non_inline_benchmark.rs b/crates/ecstore/benches/single_block_non_inline_benchmark.rs new file mode 100644 index 000000000..627bdbdbf --- /dev/null +++ b/crates/ecstore/benches/single_block_non_inline_benchmark.rs @@ -0,0 +1,126 @@ +// 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. + +use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use rustfs_ecstore::erasure_coding::{BitrotWriterWrapper, CustomWriter, Erasure}; +use rustfs_utils::HashAlgorithm; +use std::io::Cursor; +use std::sync::Arc; +use std::time::Duration; + +#[derive(Clone, Debug)] +struct BenchConfig { + name: String, + payload_size: usize, + data_shards: usize, + parity_shards: usize, + block_size: usize, +} + +impl BenchConfig { + fn new(payload_size: usize, data_shards: usize, parity_shards: usize, block_size: usize) -> Self { + Self { + name: format!( + "payload={}KB_{}+{}_block={}KB", + payload_size / 1024, + data_shards, + parity_shards, + block_size / 1024 + ), + payload_size, + data_shards, + parity_shards, + block_size, + } + } +} + +fn generate_payload(size: usize) -> Vec { + (0..size).map(|i| (i % 251) as u8).collect() +} + +fn build_non_inline_writers(config: &BenchConfig) -> Vec> { + let erasure = Erasure::new(config.data_shards, config.parity_shards, config.block_size); + let total_shards = config.data_shards + config.parity_shards; + let shard_size = erasure.shard_size(); + (0..total_shards) + .map(|_| { + Some(BitrotWriterWrapper::new( + CustomWriter::new_tokio_writer(tokio::io::sink()), + shard_size, + HashAlgorithm::HighwayHash256S, + )) + }) + .collect() +} + +fn bench_single_block_non_inline_fast_path(c: &mut Criterion) { + let configs = vec![ + BenchConfig::new(4 * 1024, 4, 2, 128 * 1024), + BenchConfig::new(64 * 1024, 4, 2, 128 * 1024), + BenchConfig::new(128 * 1024, 4, 2, 128 * 1024), + ]; + + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("benchmark runtime"); + + let mut group = c.benchmark_group("single_block_non_inline_write_path"); + group.sample_size(20); + group.measurement_time(Duration::from_secs(6)); + + for config in configs { + let payload = generate_payload(config.payload_size); + let erasure = Arc::new(Erasure::new(config.data_shards, config.parity_shards, config.block_size)); + group.throughput(Throughput::Bytes(config.payload_size as u64)); + + group.bench_with_input(BenchmarkId::new("encode_pipeline", &config.name), &config, |b, config| { + let payload = payload.clone(); + let erasure = Arc::clone(&erasure); + b.iter(|| { + let mut writers = build_non_inline_writers(config); + let reader = tokio::io::BufReader::new(Cursor::new(payload.clone())); + rt.block_on(async { + erasure + .clone() + .encode(reader, &mut writers, config.data_shards) + .await + .expect("encode pipeline benchmark"); + }); + }); + }); + + group.bench_with_input(BenchmarkId::new("single_block_candidate", &config.name), &config, |b, config| { + let payload = payload.clone(); + let erasure = Arc::clone(&erasure); + b.iter(|| { + let mut writers = build_non_inline_writers(config); + let reader = tokio::io::BufReader::new(Cursor::new(payload.clone())); + rt.block_on(async { + erasure + .clone() + .encode_single_block_non_inline(reader, &mut writers, config.data_shards) + .await + .expect("single block candidate benchmark"); + }); + }); + }); + } + + group.finish(); +} + +criterion_group!(benches, bench_single_block_non_inline_fast_path); +criterion_main!(benches); diff --git a/crates/ecstore/src/disk/error.rs b/crates/ecstore/src/disk/error.rs index 245c671dc..630802ac3 100644 --- a/crates/ecstore/src/disk/error.rs +++ b/crates/ecstore/src/disk/error.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use rustfs_rio::{InternodeHttpError, InternodeHttpErrorKind}; use std::hash::{Hash, Hasher}; use std::io::{self}; use std::path::PathBuf; @@ -181,6 +182,26 @@ impl DiskError { matches!(err, &DiskError::FileVersionNotFound) } + pub fn is_retryable_internode_write_failure(&self) -> bool { + match self { + DiskError::Io(io_error) => io_error + .get_ref() + .and_then(|source| source.downcast_ref::()) + .is_some_and(|err| err.kind().is_retryable()), + _ => false, + } + } + + pub fn internode_http_error_kind(&self) -> Option { + match self { + DiskError::Io(io_error) => io_error + .get_ref() + .and_then(|source| source.downcast_ref::()) + .map(InternodeHttpError::kind), + _ => None, + } + } + // /// If all errors are of the same fatal disk error type, returns the corresponding error. // /// Otherwise, returns Ok. // pub fn check_disk_fatal_errs(errs: &[Option]) -> Result<()> { @@ -241,7 +262,10 @@ impl From for DiskError { impl From for DiskError { fn from(e: std::io::Error) -> Self { - e.downcast::().unwrap_or_else(DiskError::Io) + match e.downcast::() { + Ok(disk_error) => disk_error, + Err(io_error) => DiskError::Io(io_error), + } } } diff --git a/crates/ecstore/src/disk/error_reduce.rs b/crates/ecstore/src/disk/error_reduce.rs index 1a6aeb549..3e6617296 100644 --- a/crates/ecstore/src/disk/error_reduce.rs +++ b/crates/ecstore/src/disk/error_reduce.rs @@ -14,6 +14,19 @@ use crate::disk::error::Error; +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WriteQuorumFailureSummary { + pub required: usize, + pub achieved: usize, + pub failed: usize, + pub total: usize, + pub offline_disks: usize, + pub ignored_failures: usize, + pub retryable_failures: usize, + pub dominant_error: Option, + pub dominant_error_label: &'static str, +} + pub static OBJECT_OP_IGNORED_ERRS: &[Error] = &[ Error::DiskNotFound, Error::FaultyDisk, @@ -77,6 +90,58 @@ pub fn reduce_errs(errors: &[Option], ignored_errs: &[Error]) -> (usize, } } +pub fn build_write_quorum_failure_summary( + errors: &[Option], + ignored_errs: &[Error], + quorum: usize, +) -> WriteQuorumFailureSummary { + let total = errors.len(); + let achieved = errors.iter().filter(|err| err.is_none()).count(); + let failed = total.saturating_sub(achieved); + let offline_disks = count_errs(errors, &Error::DiskNotFound); + let ignored_failures = errors + .iter() + .filter_map(|err| err.as_ref()) + .filter(|err| is_ignored_err(ignored_errs, err)) + .count(); + let retryable_failures = count_retryable_failures(errors); + let (_, dominant_error) = reduce_errs(errors, ignored_errs); + let dominant_error_label = dominant_error_label(errors, ignored_errs, dominant_error.as_ref()); + + WriteQuorumFailureSummary { + required: quorum, + achieved, + failed, + total, + offline_disks, + ignored_failures, + retryable_failures, + dominant_error, + dominant_error_label, + } +} + +fn dominant_error_label(errors: &[Option], ignored_errs: &[Error], dominant_error: Option<&Error>) -> &'static str { + let Some(dominant_error) = dominant_error else { + return "nil_dominated"; + }; + + if dominant_error == &Error::DiskNotFound { + return "disk_not_found"; + } + if dominant_error == &Error::ShortWrite { + return "short_write"; + } + + errors + .iter() + .filter_map(|err| err.as_ref()) + .find(|err| !is_ignored_err(ignored_errs, err) && *err == dominant_error) + .and_then(Error::internode_http_error_kind) + .map(|kind| kind.metric_label()) + .unwrap_or("other_error") +} + pub fn is_ignored_err(ignored_errs: &[Error], err: &Error) -> bool { ignored_errs.iter().any(|e| e == err) } @@ -85,6 +150,14 @@ pub fn count_errs(errors: &[Option], err: &Error) -> usize { errors.iter().filter(|&e| e.as_ref() == Some(err)).count() } +pub fn count_retryable_failures(errors: &[Option]) -> usize { + errors + .iter() + .filter_map(|err| err.as_ref()) + .filter(|err| err.is_retryable_internode_write_failure()) + .count() +} + pub fn is_all_buckets_not_found(errs: &[Option]) -> bool { for err in errs.iter() { if let Some(err) = err { @@ -163,6 +236,51 @@ mod tests { assert!(!is_ignored_err(&ignored, &e2)); } + #[test] + fn test_build_write_quorum_failure_summary() { + let retryable = Error::from(rustfs_rio::new_test_internode_http_io_error( + rustfs_rio::InternodeHttpErrorKind::ConnectionReset, + )); + let non_retryable = err_io("other"); + let errors = vec![ + None, + None, + None, + None, + None, + Some(retryable), + Some(non_retryable), + Some(Error::DiskNotFound), + ]; + + let summary = build_write_quorum_failure_summary(&errors, OBJECT_OP_IGNORED_ERRS, 6); + assert_eq!(summary.required, 6); + assert_eq!(summary.achieved, 5); + assert_eq!(summary.failed, 3); + assert_eq!(summary.total, 8); + assert_eq!(summary.offline_disks, 1); + assert_eq!(summary.ignored_failures, 1); + assert_eq!(summary.retryable_failures, 1); + assert_eq!(summary.dominant_error, None); + assert_eq!(summary.dominant_error_label, "nil_dominated"); + } + + #[test] + fn test_build_write_quorum_failure_summary_preserves_internode_label() { + let retryable_a = Error::from(rustfs_rio::new_test_internode_http_io_error( + rustfs_rio::InternodeHttpErrorKind::ConnectionReset, + )); + let retryable_b = Error::from(rustfs_rio::new_test_internode_http_io_error( + rustfs_rio::InternodeHttpErrorKind::ConnectionReset, + )); + let errors = vec![Some(retryable_a), Some(retryable_b)]; + + let summary = build_write_quorum_failure_summary(&errors, OBJECT_OP_IGNORED_ERRS, 2); + + assert_eq!(summary.retryable_failures, 2); + assert_eq!(summary.dominant_error_label, "connection_reset"); + } + #[test] fn test_reduce_errs_nil_tiebreak() { // Error::Nil and another error have the same count, should prefer Nil diff --git a/crates/ecstore/src/erasure_coding/encode.rs b/crates/ecstore/src/erasure_coding/encode.rs index 958222405..5d74fcf7a 100644 --- a/crates/ecstore/src/erasure_coding/encode.rs +++ b/crates/ecstore/src/erasure_coding/encode.rs @@ -13,8 +13,9 @@ // limitations under the License. use crate::disk::error::Error; -use crate::disk::error_reduce::count_errs; -use crate::disk::error_reduce::{OBJECT_OP_IGNORED_ERRS, reduce_write_quorum_errs}; +use crate::disk::error_reduce::{ + OBJECT_OP_IGNORED_ERRS, WriteQuorumFailureSummary, build_write_quorum_failure_summary, reduce_write_quorum_errs, +}; use crate::erasure_coding::BitrotWriterWrapper; use crate::erasure_coding::Erasure; use bytes::Bytes; @@ -50,6 +51,28 @@ async fn drain_queued_inflight_bytes(rx: &mut mpsc::Receiver>) { } } +fn dominant_error_summary_label(summary: &WriteQuorumFailureSummary) -> &'static str { + summary.dominant_error_label +} + +fn format_write_quorum_failure(summary: &WriteQuorumFailureSummary) -> String { + format!( + "erasure write quorum (required={}, achieved={}, failed={}, total={}, offline-disks={}/{}, retryable-failures={}, dominant-error={})", + summary.required, + summary.achieved, + summary.failed, + summary.total, + summary.offline_disks, + summary.total, + summary.retryable_failures, + dominant_error_summary_label(summary) + ) +} + +fn quorum_dominant_error_metric_label(summary: &WriteQuorumFailureSummary) -> &'static str { + dominant_error_summary_label(summary) +} + pub(crate) struct MultiWriter<'a> { writers: &'a mut [Option], write_quorum: usize, @@ -109,25 +132,18 @@ impl<'a> MultiWriter<'a> { } if let Some(write_err) = reduce_write_quorum_errs(&self.errs, OBJECT_OP_IGNORED_ERRS, self.write_quorum) { - error!( - "reduce_write_quorum_errs: {:?}, offline-disks={}/{}, errs={:?}", - write_err, - count_errs(&self.errs, &Error::DiskNotFound), - self.writers.len(), - self.errs - ); - return Err(std::io::Error::other(format!( - "Failed to write data: {} (offline-disks={}/{})", - write_err, - count_errs(&self.errs, &Error::DiskNotFound), - self.writers.len() - ))); + let summary = build_write_quorum_failure_summary(&self.errs, OBJECT_OP_IGNORED_ERRS, self.write_quorum); + let summary_text = format_write_quorum_failure(&summary); + rustfs_io_metrics::internode_metrics::global_internode_metrics() + .record_erasure_write_quorum_failure("write", quorum_dominant_error_metric_label(&summary)); + error!("reduce_write_quorum_errs: {:?}, {}, errs={:?}", write_err, summary_text, self.errs); + return Err(std::io::Error::other(format!("Failed to write data: {summary_text}"))); } + let summary = build_write_quorum_failure_summary(&self.errs, OBJECT_OP_IGNORED_ERRS, self.write_quorum); Err(std::io::Error::other(format!( - "Failed to write data: (offline-disks={}/{}): {}", - count_errs(&self.errs, &Error::DiskNotFound), - self.writers.len(), + "Failed to write data: {}: {}", + format_write_quorum_failure(&summary), self.errs .iter() .map(|e| e.as_ref().map_or_else(|| "".to_string(), |e| e.to_string())) @@ -171,25 +187,21 @@ impl<'a> MultiWriter<'a> { } if let Some(write_err) = reduce_write_quorum_errs(&self.errs, OBJECT_OP_IGNORED_ERRS, self.write_quorum) { + let summary = build_write_quorum_failure_summary(&self.errs, OBJECT_OP_IGNORED_ERRS, self.write_quorum); + let summary_text = format_write_quorum_failure(&summary); + rustfs_io_metrics::internode_metrics::global_internode_metrics() + .record_erasure_write_quorum_failure("shutdown", quorum_dominant_error_metric_label(&summary)); error!( - "reduce_write_quorum_errs during shutdown: {:?}, offline-disks={}/{}, errs={:?}", - write_err, - count_errs(&self.errs, &Error::DiskNotFound), - self.writers.len(), - self.errs + "reduce_write_quorum_errs during shutdown: {:?}, {}, errs={:?}", + write_err, summary_text, self.errs ); - return Err(std::io::Error::other(format!( - "Failed to shutdown writers: {} (offline-disks={}/{})", - write_err, - count_errs(&self.errs, &Error::DiskNotFound), - self.writers.len() - ))); + return Err(std::io::Error::other(format!("Failed to shutdown writers: {summary_text}"))); } + let summary = build_write_quorum_failure_summary(&self.errs, OBJECT_OP_IGNORED_ERRS, self.write_quorum); Err(std::io::Error::other(format!( - "Failed to shutdown writers: (offline-disks={}/{}): {}", - count_errs(&self.errs, &Error::DiskNotFound), - self.writers.len(), + "Failed to shutdown writers: {}: {}", + format_write_quorum_failure(&summary), self.errs .iter() .map(|e| e.as_ref().map_or_else(|| "".to_string(), |e| e.to_string())) @@ -200,6 +212,49 @@ impl<'a> MultiWriter<'a> { } impl Erasure { + async fn encode_small_direct( + self: Arc, + mut reader: R, + writers: &mut [Option], + quorum: usize, + require_single_block: bool, + ) -> std::io::Result<(R, usize)> + where + R: AsyncRead + Send + Sync + Unpin, + { + use tokio::io::AsyncReadExt; + + let mut buf = Vec::with_capacity(self.block_size); + let total = if require_single_block { + let read_limit = self + .block_size + .checked_add(1) + .ok_or_else(|| std::io::Error::new(std::io::ErrorKind::InvalidInput, "erasure block_size is too large"))?; + let read_limit = u64::try_from(read_limit) + .map_err(|_| std::io::Error::new(std::io::ErrorKind::InvalidInput, "erasure block_size exceeds u64"))?; + (&mut reader).take(read_limit).read_to_end(&mut buf).await? + } else { + reader.read_to_end(&mut buf).await? + }; + + if total == 0 { + return Ok((reader, 0)); + } + + if require_single_block && total > self.block_size { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "single-block non-inline fast path expects total <= block_size", + )); + } + + let shards = self.encode_data_owned(buf)?; + let mut mw = MultiWriter::new(writers, quorum); + mw.write(shards).await?; + mw.shutdown().await?; + Ok((reader, total)) + } + pub async fn encode( self: Arc, mut reader: R, @@ -307,27 +362,28 @@ impl Erasure { /// Reads all data, encodes directly, writes shards sequentially. pub async fn encode_inline_small( self: Arc, - mut reader: R, + reader: R, writers: &mut [Option], quorum: usize, ) -> std::io::Result<(R, usize)> where R: AsyncRead + Send + Sync + Unpin, { - use tokio::io::AsyncReadExt; + self.encode_small_direct(reader, writers, quorum, false).await + } - let mut buf = Vec::with_capacity(self.block_size); - let total = reader.read_to_end(&mut buf).await?; - - if total == 0 { - return Ok((reader, 0)); - } - - let shards = self.encode_data(&buf)?; - let mut mw = MultiWriter::new(writers, quorum); - mw.write(shards).await?; - mw.shutdown().await?; - Ok((reader, total)) + /// Fast path for single-block non-inline objects: avoids the producer/consumer + /// pipeline in `encode()` while keeping the same writer/quorum/shutdown semantics. + pub async fn encode_single_block_non_inline( + self: Arc, + reader: R, + writers: &mut [Option], + quorum: usize, + ) -> std::io::Result<(R, usize)> + where + R: AsyncRead + Send + Sync + Unpin, + { + self.encode_small_direct(reader, writers, quorum, true).await } } @@ -491,6 +547,75 @@ mod tests { } } + #[tokio::test] + async fn encode_single_block_non_inline_payload_writes_all_shards() { + const DATA_SHARDS: usize = 2; + const PARITY_SHARDS: usize = 2; + const TOTAL_SHARDS: usize = DATA_SHARDS + PARITY_SHARDS; + const BLOCK_SIZE: usize = 64; + + let committed: Vec>>> = (0..TOTAL_SHARDS).map(|_| Arc::new(Mutex::new(Vec::new()))).collect(); + + let mut writers: Vec> = committed + .iter() + .map(|c| { + Some(BitrotWriterWrapper::new( + CustomWriter::new_tokio_writer(DeferredCommitWriter::new(c.clone())), + BLOCK_SIZE / DATA_SHARDS, + HashAlgorithm::HighwayHash256S, + )) + }) + .collect(); + + let payload = b"hello single block"; + let erasure = Arc::new(Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE)); + let reader = tokio::io::BufReader::new(std::io::Cursor::new(payload.to_vec())); + let (_reader, total) = erasure + .encode_single_block_non_inline(reader, &mut writers, DATA_SHARDS) + .await + .unwrap(); + + assert_eq!(total, payload.len()); + for (i, c) in committed.iter().enumerate() { + assert!(!c.lock().unwrap().is_empty(), "shard {i} should have received data"); + } + } + + #[tokio::test] + async fn encode_single_block_non_inline_rejects_multi_block_payload() { + const DATA_SHARDS: usize = 2; + const PARITY_SHARDS: usize = 2; + const TOTAL_SHARDS: usize = DATA_SHARDS + PARITY_SHARDS; + const BLOCK_SIZE: usize = 64; + + let committed: Vec>>> = (0..TOTAL_SHARDS).map(|_| Arc::new(Mutex::new(Vec::new()))).collect(); + + let mut writers: Vec> = committed + .iter() + .map(|c| { + Some(BitrotWriterWrapper::new( + CustomWriter::new_tokio_writer(DeferredCommitWriter::new(c.clone())), + BLOCK_SIZE / DATA_SHARDS, + HashAlgorithm::HighwayHash256S, + )) + }) + .collect(); + + let payload = vec![1u8; BLOCK_SIZE + 1]; + let erasure = Arc::new(Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE)); + let reader = tokio::io::BufReader::new(std::io::Cursor::new(payload)); + let err = erasure + .encode_single_block_non_inline(reader, &mut writers, DATA_SHARDS) + .await + .expect_err("single-block fast path must reject oversized readers"); + + assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput); + assert!(err.to_string().contains("single-block non-inline fast path")); + for c in committed { + assert!(c.lock().unwrap().is_empty()); + } + } + #[test] fn encode_channel_capacity_never_returns_zero() { assert_eq!(encode_channel_capacity(0, 1024), 1); @@ -498,6 +623,29 @@ mod tests { assert_eq!(encode_channel_capacity(4096, 1024), 1); } + #[test] + fn write_quorum_failure_summary_uses_stable_dominant_error_label() { + let err = Error::from(rustfs_rio::new_test_internode_http_io_error( + rustfs_rio::InternodeHttpErrorKind::ConnectionReset, + )); + let summary = WriteQuorumFailureSummary { + required: 2, + achieved: 0, + failed: 2, + total: 2, + offline_disks: 0, + ignored_failures: 0, + retryable_failures: 2, + dominant_error: Some(err), + dominant_error_label: "connection_reset", + }; + let text = format_write_quorum_failure(&summary); + + assert!(text.contains("dominant-error=connection_reset")); + assert!(!text.contains("/rustfs/rpc/put_file_stream")); + assert!(!text.contains("PUT ")); + } + #[test] fn encode_channel_capacity_respects_budget_and_hard_cap() { assert_eq!(encode_channel_capacity(4 * 1024 * 1024, 32 * 1024 * 1024), 8); diff --git a/crates/ecstore/src/rpc/remote_disk.rs b/crates/ecstore/src/rpc/remote_disk.rs index f9c91cd91..829e323ac 100644 --- a/crates/ecstore/src/rpc/remote_disk.rs +++ b/crates/ecstore/src/rpc/remote_disk.rs @@ -36,7 +36,7 @@ use metrics::counter; use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo}; use rustfs_io_metrics::internode_metrics::{ INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC, - global_internode_metrics, + INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, global_internode_metrics, }; use rustfs_protos::evict_failed_connection; use rustfs_protos::proto_gen::node_service::RenamePartRequest; @@ -74,6 +74,9 @@ enum FailureHealthAction { IgnoreFailure, } +const REMOTE_DISK_OPEN_WRITE_MAX_ATTEMPTS: usize = 2; +const REMOTE_DISK_OPEN_WRITE_RETRY_BACKOFF: Duration = Duration::from_millis(20); + async fn copy_stream_with_buffer(reader: &mut R, writer: &mut W, buffer_size: usize) -> io::Result where R: AsyncRead + Unpin, @@ -119,6 +122,10 @@ impl RemoteDisk { err_text.contains("httpreader stream error") || err_text.contains("error decoding response body") } + fn is_retryable_open_write_error(err: &DiskError) -> bool { + err.is_retryable_internode_write_failure() + } + pub(crate) async fn new(ep: &Endpoint, opt: &DiskOption, data_transport: Arc) -> Result { let addr = if let Some(port) = ep.url.port() { format!("{}://{}:{}", ep.url.scheme(), ep.url.host_str().unwrap(), port) @@ -156,6 +163,50 @@ impl RemoteDisk { self.health.last_capacity_snapshot() } + async fn open_write_with_retry(&self, request: WriteStreamRequest) -> Result { + let mut attempt = 1; + let mut last_retry_classification = None; + loop { + match self.data_transport.open_write(request.clone()).await { + Ok(writer) => { + if attempt > 1 + && let Some(classification) = last_retry_classification + { + global_internode_metrics().record_retry_success_for_operation_and_backend( + rustfs_io_metrics::internode_metrics::INTERNODE_OPERATION_PUT_FILE_STREAM, + INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, + classification, + ); + } + return Ok(writer); + } + Err(err) if attempt < REMOTE_DISK_OPEN_WRITE_MAX_ATTEMPTS && Self::is_retryable_open_write_error(&err) => { + if let Some(classification) = err.internode_http_error_kind() { + let classification = classification.metric_label(); + global_internode_metrics().record_retry_for_operation_and_backend( + rustfs_io_metrics::internode_metrics::INTERNODE_OPERATION_PUT_FILE_STREAM, + INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, + classification, + ); + last_retry_classification = Some(classification); + } + debug!( + endpoint = %request.endpoint, + volume = %request.volume, + path = %request.path, + append = request.append, + size = request.size, + attempt, + "retrying remote open_write after retryable transport error" + ); + tokio::time::sleep(REMOTE_DISK_OPEN_WRITE_RETRY_BACKOFF).await; + attempt += 1; + } + Err(err) => return Err(err), + } + } + } + pub fn record_capacity_probe(&self, total: u64, used: u64, free: u64) { self.health.record_capacity_probe(total, used, free); } @@ -1342,16 +1393,15 @@ impl DiskAPI for RemoteDisk { return Err(DiskError::FaultyDisk); } let disk = self.disk_ref().await; - self.data_transport - .open_write(WriteStreamRequest { - endpoint: self.endpoint.grid_host(), - disk, - volume: volume.to_string(), - path: path.to_string(), - append: true, - size: 0, - }) - .await + self.open_write_with_retry(WriteStreamRequest { + endpoint: self.endpoint.grid_host(), + disk, + volume: volume.to_string(), + path: path.to_string(), + append: true, + size: 0, + }) + .await } #[tracing::instrument(level = "debug", skip(self))] @@ -1368,16 +1418,15 @@ impl DiskAPI for RemoteDisk { return Err(DiskError::FaultyDisk); } let disk = self.disk_ref().await; - self.data_transport - .open_write(WriteStreamRequest { - endpoint: self.endpoint.grid_host(), - disk, - volume: volume.to_string(), - path: path.to_string(), - append: false, - size: file_size, - }) - .await + self.open_write_with_retry(WriteStreamRequest { + endpoint: self.endpoint.grid_host(), + disk, + volume: volume.to_string(), + path: path.to_string(), + append: false, + size: file_size, + }) + .await } #[tracing::instrument(level = "debug", skip(self))] @@ -1832,6 +1881,35 @@ mod tests { } } + #[derive(Debug, Clone)] + enum OpenWriteTestStep { + Error(DiskError), + Success, + } + + #[derive(Debug, Clone, Default)] + struct RetryingOpenWriteInternodeDataTransport { + calls: Arc>>, + steps: Arc>>, + } + + impl RetryingOpenWriteInternodeDataTransport { + fn with_steps(steps: Vec) -> Self { + Self { + calls: Arc::new(StdMutex::new(Vec::new())), + steps: Arc::new(StdMutex::new(steps)), + } + } + + fn calls(&self) -> Vec { + self.calls.lock().expect("recorded transport calls lock poisoned").clone() + } + + fn record(&self, call: RecordedTransportCall) { + self.calls.lock().expect("recorded transport calls lock poisoned").push(call); + } + } + #[derive(Debug, Default)] struct EmptyTestReader; @@ -1916,6 +1994,34 @@ mod tests { } } + #[async_trait::async_trait] + impl InternodeDataTransport for RetryingOpenWriteInternodeDataTransport { + async fn open_read(&self, _request: ReadStreamRequest) -> Result { + panic!("open_read should not be used in open_write retry test"); + } + + async fn open_write(&self, request: WriteStreamRequest) -> Result { + self.record(RecordedTransportCall::Write(request)); + let step = self.steps.lock().expect("open_write retry steps lock poisoned").remove(0); + match step { + OpenWriteTestStep::Error(err) => Err(err), + OpenWriteTestStep::Success => Ok(Box::new(SinkTestWriter)), + } + } + + async fn open_walk_dir(&self, _request: WalkDirStreamRequest) -> Result { + panic!("open_walk_dir should not be used in open_write retry test"); + } + + fn name(&self) -> &'static str { + "retrying-open-write" + } + + fn capabilities(&self) -> InternodeDataTransportCapabilities { + InternodeDataTransportCapabilities::tcp_http() + } + } + async fn new_remote_disk_with_transport(data_transport: Arc) -> RemoteDisk { let endpoint = Endpoint { url: url::Url::parse("http://remote-node:9000/data/rustfs0").unwrap(), @@ -2283,6 +2389,47 @@ mod tests { } } + #[tokio::test] + async fn test_remote_disk_create_file_retries_once_on_retryable_open_write_error() { + let transport = RetryingOpenWriteInternodeDataTransport::with_steps(vec![ + OpenWriteTestStep::Error(DiskError::from(rustfs_rio::new_test_internode_http_io_error( + rustfs_rio::InternodeHttpErrorKind::ConnectionReset, + ))), + OpenWriteTestStep::Success, + ]); + let remote_disk = new_remote_disk_with_transport(Arc::new(transport.clone())).await; + rustfs_io_metrics::internode_metrics::global_internode_metrics().reset_for_test(); + + let _created = remote_disk + .create_file("orig-bucket", "bucket", "object/part.1", 4096) + .await + .expect("retryable open_write error should recover"); + + let calls = transport.calls(); + assert_eq!(calls.len(), 2, "create_file should retry exactly once"); + let snapshot = rustfs_io_metrics::internode_metrics::global_internode_metrics().snapshot(); + assert_eq!(snapshot.outgoing_requests_total, 0); + } + + #[tokio::test] + async fn test_remote_disk_append_file_does_not_retry_non_retryable_open_write_error() { + let transport = RetryingOpenWriteInternodeDataTransport::with_steps(vec![OpenWriteTestStep::Error(DiskError::from( + rustfs_rio::new_test_internode_http_io_error(rustfs_rio::InternodeHttpErrorKind::DnsResolutionFailed), + ))]); + let remote_disk = new_remote_disk_with_transport(Arc::new(transport.clone())).await; + + let err = match remote_disk.append_file("bucket", "object/part.2").await { + Ok(_) => panic!("non-retryable open_write error should be returned directly"), + Err(err) => err, + }; + + assert_eq!( + err.internode_http_error_kind(), + Some(rustfs_rio::InternodeHttpErrorKind::DnsResolutionFailed) + ); + assert_eq!(transport.calls().len(), 1, "append_file should not retry non-retryable errors"); + } + #[tokio::test] async fn test_remote_disk_walk_dir_uses_configured_data_transport() { let transport = RecordingInternodeDataTransport::default(); diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index 74ac90d92..c86c368d0 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -773,6 +773,37 @@ fn delete_file_info_version_id(version_id: Option) -> Option { } } +fn object_fits_single_block(object_size: i64, block_size: usize) -> bool { + match usize::try_from(object_size) { + Ok(size) => size > 0 && size <= block_size, + Err(_) => false, + } +} + +fn should_use_inline_small_fast_path(is_inline_buffer: bool, object_size: i64, block_size: usize) -> bool { + is_inline_buffer && object_fits_single_block(object_size, block_size) +} + +fn should_use_single_block_non_inline_fast_path(is_inline_buffer: bool, object_size: i64, block_size: usize) -> bool { + !is_inline_buffer && object_fits_single_block(object_size, block_size) +} + +enum SmallWritePath { + Inline, + SingleBlockNonInline, + Pipeline, +} + +fn classify_small_write_path(is_inline_buffer: bool, object_size: i64, block_size: usize) -> SmallWritePath { + if should_use_inline_small_fast_path(is_inline_buffer, object_size, block_size) { + SmallWritePath::Inline + } else if should_use_single_block_non_inline_fast_path(is_inline_buffer, object_size, block_size) { + SmallWritePath::SingleBlockNonInline + } else { + SmallWritePath::Pipeline + } +} + #[async_trait::async_trait] impl ObjectIO for SetDisks { #[tracing::instrument(level = "debug", skip(self))] @@ -1065,10 +1096,10 @@ impl ObjectIO for SetDisks { HashReader::from_stream(Cursor::new(Vec::new()), 0, 0, None, None, false)?, ); - let use_fast_path = is_inline_buffer && data.size() <= fi.erasure.block_size as i64; + let write_path = classify_small_write_path(is_inline_buffer, data.size(), fi.erasure.block_size); - let (reader, w_size) = if use_fast_path { - match Arc::new(erasure) + let (reader, w_size) = match write_path { + SmallWritePath::Inline => match Arc::new(erasure) .encode_inline_small(stream, &mut writers, write_quorum) .await { @@ -1077,15 +1108,24 @@ impl ObjectIO for SetDisks { error!("encode_inline_small err {:?}", e); return Err(e.into()); } - } - } else { - match Arc::new(erasure).encode(stream, &mut writers, write_quorum).await { + }, + SmallWritePath::SingleBlockNonInline => match Arc::new(erasure) + .encode_single_block_non_inline(stream, &mut writers, write_quorum) + .await + { + Ok((r, w)) => (r, w), + Err(e) => { + error!("encode_single_block_non_inline err {:?}", e); + return Err(e.into()); + } + }, + SmallWritePath::Pipeline => match Arc::new(erasure).encode(stream, &mut writers, write_quorum).await { Ok((r, w)) => (r, w), Err(e) => { error!("encode err {:?}", e); return Err(e.into()); } - } + }, }; let _ = mem::replace(&mut data.stream, reader); @@ -2971,7 +3011,18 @@ impl MultipartOperations for SetDisks { HashReader::from_stream(Cursor::new(Vec::new()), 0, 0, None, None, false)?, ); - let (reader, w_size) = Arc::new(erasure).encode(stream, &mut writers, write_quorum).await?; // TODO: delete temporary directory on error + let write_path = classify_small_write_path(false, data.size(), fi.erasure.block_size); + + let (reader, w_size) = match write_path { + SmallWritePath::SingleBlockNonInline => { + Arc::new(erasure) + .encode_single_block_non_inline(stream, &mut writers, write_quorum) + .await? + } + SmallWritePath::Inline | SmallWritePath::Pipeline => { + Arc::new(erasure).encode(stream, &mut writers, write_quorum).await? + } + }; // TODO: delete temporary directory on error let _ = mem::replace(&mut data.stream, reader); @@ -6716,6 +6767,47 @@ mod tests { assert_eq!(delete_file_info_version_id(None), None); } + #[test] + fn put_object_fast_path_selection_prefers_inline_only_when_inline_buffer_and_single_block() { + assert!(should_use_inline_small_fast_path(true, 1024, 4096)); + assert!(!should_use_single_block_non_inline_fast_path(true, 1024, 4096)); + assert!(matches!(classify_small_write_path(true, 1024, 4096), SmallWritePath::Inline)); + + assert!(!should_use_inline_small_fast_path(false, 1024, 4096)); + assert!(should_use_single_block_non_inline_fast_path(false, 1024, 4096)); + assert!(matches!( + classify_small_write_path(false, 1024, 4096), + SmallWritePath::SingleBlockNonInline + )); + } + + #[test] + fn put_object_fast_path_selection_rejects_zero_and_multi_block_payloads() { + assert!(!should_use_inline_small_fast_path(true, 0, 4096)); + assert!(!should_use_single_block_non_inline_fast_path(false, 0, 4096)); + assert!(matches!(classify_small_write_path(true, 0, 4096), SmallWritePath::Pipeline)); + + assert!(!should_use_inline_small_fast_path(true, -1, 4096)); + assert!(!should_use_single_block_non_inline_fast_path(false, -1, 4096)); + assert!(matches!(classify_small_write_path(false, -1, 4096), SmallWritePath::Pipeline)); + + assert!(!should_use_inline_small_fast_path(true, 8192, 4096)); + assert!(!should_use_single_block_non_inline_fast_path(false, 8192, 4096)); + assert!(matches!(classify_small_write_path(false, 8192, 4096), SmallWritePath::Pipeline)); + } + + #[test] + fn put_object_part_fast_path_selection_matches_single_block_non_inline_rules() { + assert!(should_use_single_block_non_inline_fast_path(false, 4096, 4096)); + assert!(should_use_single_block_non_inline_fast_path(false, 2048, 4096)); + assert!(!should_use_single_block_non_inline_fast_path(false, 4097, 4096)); + assert!(!should_use_single_block_non_inline_fast_path(false, 0, 4096)); + assert!(matches!( + classify_small_write_path(false, 4096, 4096), + SmallWritePath::SingleBlockNonInline + )); + } + #[test] fn test_is_cold_storage_class() { // Test cold storage classes diff --git a/crates/io-metrics/README.md b/crates/io-metrics/README.md index 8da89cef0..4cbb75410 100644 --- a/crates/io-metrics/README.md +++ b/crates/io-metrics/README.md @@ -167,6 +167,10 @@ Operation-level metrics use the same low-cardinality label set: | `rustfs_system_network_internode_operation_requests_outgoing_total` | `operation`, `backend` | Outgoing request attempts for an internode operation. | | `rustfs_system_network_internode_operation_requests_incoming_total` | `operation`, `backend` | Incoming request attempts for an internode operation. | | `rustfs_system_network_internode_operation_errors_total` | `operation`, `backend` | Failed internode operation attempts. | +| `rustfs_system_network_internode_operation_classified_errors_total` | `operation`, `backend`, `classification` | Classified internode transport failures. | +| `rustfs_system_network_internode_operation_retries_total` | `operation`, `backend`, `classification` | Retry attempts for retryable internode transport failures. | +| `rustfs_system_network_internode_operation_retry_successes_total` | `operation`, `backend`, `classification` | Successful recoveries after retryable internode transport failures. | +| `rustfs_system_storage_erasure_write_quorum_failures_total` | `stage`, `dominant_error` | Erasure write quorum failures grouped by failure stage and dominant error class. | Current `operation` values are `read_file_stream`, `put_file_stream`, `walk_dir`, `grpc_read_all`, and `grpc_write_all`. Current `backend` values are @@ -180,6 +184,21 @@ successful completions are not emitted as a dedicated result-labeled metric. Adding completion/result labels is a follow-up once stream completion semantics are defined consistently for request setup, body transfer, and shutdown. +Current low-cardinality `classification` values come from the TCP/HTTP internode +path and include: + +- `connect_timeout` +- `connection_refused` +- `dns_resolution_failed` +- `connection_reset` +- `body_stream_aborted` +- `http_429` +- `http_502` +- `http_503` +- `http_504` +- `http_status_other` +- `unknown` + `scripts/run_internode_transport_baseline.sh --metrics-url ...` records metric deltas with `operation` and `backend` columns, so the TCP baseline can attribute bytes and request/error counts to `tcp-http` transport operations. diff --git a/crates/io-metrics/src/internode_metrics.rs b/crates/io-metrics/src/internode_metrics.rs index bf26133c9..2482f140b 100644 --- a/crates/io-metrics/src/internode_metrics.rs +++ b/crates/io-metrics/src/internode_metrics.rs @@ -30,11 +30,18 @@ pub const INTERNODE_TRANSPORT_BACKEND_UNKNOWN: &str = "unknown"; const OPERATION_LABEL: &str = "operation"; const BACKEND_LABEL: &str = "backend"; +const CLASSIFICATION_LABEL: &str = "classification"; +const STAGE_LABEL: &str = "stage"; +const DOMINANT_ERROR_LABEL: &str = "dominant_error"; const INTERNODE_OPERATION_SENT_BYTES_TOTAL: &str = "rustfs_system_network_internode_operation_sent_bytes_total"; const INTERNODE_OPERATION_RECV_BYTES_TOTAL: &str = "rustfs_system_network_internode_operation_recv_bytes_total"; const INTERNODE_OPERATION_REQUESTS_OUTGOING_TOTAL: &str = "rustfs_system_network_internode_operation_requests_outgoing_total"; const INTERNODE_OPERATION_REQUESTS_INCOMING_TOTAL: &str = "rustfs_system_network_internode_operation_requests_incoming_total"; const INTERNODE_OPERATION_ERRORS_TOTAL: &str = "rustfs_system_network_internode_operation_errors_total"; +const INTERNODE_OPERATION_CLASSIFIED_ERRORS_TOTAL: &str = "rustfs_system_network_internode_operation_classified_errors_total"; +const INTERNODE_OPERATION_RETRIES_TOTAL: &str = "rustfs_system_network_internode_operation_retries_total"; +const INTERNODE_OPERATION_RETRY_SUCCESSES_TOTAL: &str = "rustfs_system_network_internode_operation_retry_successes_total"; +const ERASURE_WRITE_QUORUM_FAILURES_TOTAL: &str = "rustfs_system_storage_erasure_write_quorum_failures_total"; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct InternodeOperationMetricDescriptor { @@ -43,6 +50,8 @@ pub struct InternodeOperationMetricDescriptor { } const OPERATION_BACKEND_LABELS: &[&str] = &[OPERATION_LABEL, BACKEND_LABEL]; +const OPERATION_BACKEND_CLASSIFICATION_LABELS: &[&str] = &[OPERATION_LABEL, BACKEND_LABEL, CLASSIFICATION_LABEL]; +const QUORUM_FAILURE_LABELS: &[&str] = &[STAGE_LABEL, DOMINANT_ERROR_LABEL]; pub const INTERNODE_OPERATION_METRICS: &[InternodeOperationMetricDescriptor] = &[ InternodeOperationMetricDescriptor { @@ -65,6 +74,22 @@ pub const INTERNODE_OPERATION_METRICS: &[InternodeOperationMetricDescriptor] = & name: INTERNODE_OPERATION_ERRORS_TOTAL, labels: OPERATION_BACKEND_LABELS, }, + InternodeOperationMetricDescriptor { + name: INTERNODE_OPERATION_CLASSIFIED_ERRORS_TOTAL, + labels: OPERATION_BACKEND_CLASSIFICATION_LABELS, + }, + InternodeOperationMetricDescriptor { + name: INTERNODE_OPERATION_RETRIES_TOTAL, + labels: OPERATION_BACKEND_CLASSIFICATION_LABELS, + }, + InternodeOperationMetricDescriptor { + name: INTERNODE_OPERATION_RETRY_SUCCESSES_TOTAL, + labels: OPERATION_BACKEND_CLASSIFICATION_LABELS, + }, + InternodeOperationMetricDescriptor { + name: ERASURE_WRITE_QUORUM_FAILURES_TOTAL, + labels: QUORUM_FAILURE_LABELS, + }, ]; #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] @@ -183,6 +208,60 @@ impl InternodeMetrics { counter!(INTERNODE_OPERATION_ERRORS_TOTAL, OPERATION_LABEL => operation, BACKEND_LABEL => backend).increment(1); } + pub fn record_classified_error_for_operation_and_backend( + &self, + operation: &'static str, + backend: &'static str, + classification: &'static str, + ) { + counter!( + INTERNODE_OPERATION_CLASSIFIED_ERRORS_TOTAL, + OPERATION_LABEL => operation, + BACKEND_LABEL => backend, + CLASSIFICATION_LABEL => classification + ) + .increment(1); + } + + pub fn record_retry_for_operation_and_backend( + &self, + operation: &'static str, + backend: &'static str, + classification: &'static str, + ) { + counter!( + INTERNODE_OPERATION_RETRIES_TOTAL, + OPERATION_LABEL => operation, + BACKEND_LABEL => backend, + CLASSIFICATION_LABEL => classification + ) + .increment(1); + } + + pub fn record_retry_success_for_operation_and_backend( + &self, + operation: &'static str, + backend: &'static str, + classification: &'static str, + ) { + counter!( + INTERNODE_OPERATION_RETRY_SUCCESSES_TOTAL, + OPERATION_LABEL => operation, + BACKEND_LABEL => backend, + CLASSIFICATION_LABEL => classification + ) + .increment(1); + } + + pub fn record_erasure_write_quorum_failure(&self, stage: &'static str, dominant_error: &'static str) { + counter!( + ERASURE_WRITE_QUORUM_FAILURES_TOTAL, + STAGE_LABEL => stage, + DOMINANT_ERROR_LABEL => dominant_error + ) + .increment(1); + } + pub fn record_dial_result(&self, duration: Duration, success: bool) { let elapsed_nanos = duration.as_nanos().min(u128::from(u64::MAX)) as u64; self.dial_total_time_nanos.fetch_add(elapsed_nanos, Ordering::Relaxed); @@ -303,10 +382,14 @@ mod tests { #[test] fn operation_metric_descriptors_include_backend_and_operation_labels() { - assert_eq!(INTERNODE_OPERATION_METRICS.len(), 5); - for metric in INTERNODE_OPERATION_METRICS { + assert_eq!(INTERNODE_OPERATION_METRICS.len(), 9); + for metric in &INTERNODE_OPERATION_METRICS[..5] { assert_eq!(metric.labels, &[OPERATION_LABEL, BACKEND_LABEL]); } + for metric in &INTERNODE_OPERATION_METRICS[5..8] { + assert_eq!(metric.labels, &[OPERATION_LABEL, BACKEND_LABEL, CLASSIFICATION_LABEL]); + } + assert_eq!(INTERNODE_OPERATION_METRICS[8].labels, &[STAGE_LABEL, DOMINANT_ERROR_LABEL]); } #[test] @@ -322,29 +405,48 @@ mod tests { assert_eq!(INTERNODE_TRANSPORT_BACKEND_UNKNOWN, "unknown"); assert_eq!( - INTERNODE_OPERATION_METRICS, - &[ - InternodeOperationMetricDescriptor { - name: "rustfs_system_network_internode_operation_sent_bytes_total", - labels: &[OPERATION_LABEL, BACKEND_LABEL], - }, - InternodeOperationMetricDescriptor { - name: "rustfs_system_network_internode_operation_recv_bytes_total", - labels: &[OPERATION_LABEL, BACKEND_LABEL], - }, - InternodeOperationMetricDescriptor { - name: "rustfs_system_network_internode_operation_requests_outgoing_total", - labels: &[OPERATION_LABEL, BACKEND_LABEL], - }, - InternodeOperationMetricDescriptor { - name: "rustfs_system_network_internode_operation_requests_incoming_total", - labels: &[OPERATION_LABEL, BACKEND_LABEL], - }, - InternodeOperationMetricDescriptor { - name: "rustfs_system_network_internode_operation_errors_total", - labels: &[OPERATION_LABEL, BACKEND_LABEL], - }, - ] + INTERNODE_OPERATION_METRICS[5].name, + "rustfs_system_network_internode_operation_classified_errors_total" + ); + assert_eq!( + INTERNODE_OPERATION_METRICS[6].name, + "rustfs_system_network_internode_operation_retries_total" + ); + assert_eq!( + INTERNODE_OPERATION_METRICS[7].name, + "rustfs_system_network_internode_operation_retry_successes_total" + ); + assert_eq!( + INTERNODE_OPERATION_METRICS[8].name, + "rustfs_system_storage_erasure_write_quorum_failures_total" ); } + + #[test] + fn classified_and_retry_metrics_update_counters() { + let metrics = InternodeMetrics::default(); + + metrics.record_classified_error_for_operation_and_backend( + INTERNODE_OPERATION_PUT_FILE_STREAM, + INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, + "connection_reset", + ); + metrics.record_retry_for_operation_and_backend( + INTERNODE_OPERATION_PUT_FILE_STREAM, + INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, + "connection_reset", + ); + metrics.record_retry_success_for_operation_and_backend( + INTERNODE_OPERATION_PUT_FILE_STREAM, + INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, + "connection_reset", + ); + metrics.record_erasure_write_quorum_failure("write", "connection_reset"); + + let snapshot = metrics.snapshot(); + assert_eq!(snapshot.sent_bytes_total, 0); + assert_eq!(snapshot.recv_bytes_total, 0); + assert_eq!(snapshot.outgoing_requests_total, 0); + assert_eq!(snapshot.incoming_requests_total, 0); + } } diff --git a/crates/object-capacity/AGENTS.md b/crates/object-capacity/AGENTS.md new file mode 100644 index 000000000..f42cb634f --- /dev/null +++ b/crates/object-capacity/AGENTS.md @@ -0,0 +1,87 @@ +# rustfs-object-capacity Agent Guide + +This document is a lightweight maintainer-oriented guide for collaborators working in `crates/object-capacity`. + +## Purpose + +`rustfs-object-capacity` is responsible for estimating and refreshing RustFS object-data usage, not general filesystem capacity. Changes in this crate can affect: + +- admin `used_capacity` responses +- refresh latency and background work +- dirty-disk propagation after writes, heal, and data movement +- capacity-related metrics and operational observability + +Keep changes narrow and source-driven. + +## Source of Truth + +Before changing behavior, read these files first: + +- `src/scan.rs` +- `src/capacity_manager.rs` +- `src/capacity_scope.rs` +- `src/types.rs` +- `../../rustfs/src/capacity/service.rs` +- `../config/src/constants/capacity.rs` + +Do not treat shell scripts or old docs as the authoritative behavior definition when the Rust code says otherwise. + +## Change Priorities + +Prefer the following order: + +1. correctness of returned capacity values +2. safety of degraded behavior under timeout, stall, or partial failure +3. consistency of dirty-subset refresh and per-disk cache state +4. bounded scan cost and runtime overhead +5. clarity of metrics and logs + +## Important Behavioral Constraints + +- This crate intentionally prefers a usable degraded result over hard failure when partial progress is available. +- Dirty-subset refresh is safe only after a complete per-disk cache has been established. +- Subset refresh failures with partial errors must not silently corrupt aggregate cache state. +- Environment-derived configuration is cached behind `OnceLock` in non-test builds, so runtime env changes usually require a restart. +- Symlink following is disabled by default for safety and determinism. +- Multi-disk scans are concurrent, so avoid changes that introduce hidden shared-state coupling. + +## Editing Guidance + +- Keep scan-path changes local and easy to audit. +- Avoid refactoring the control flow unless it is necessary for correctness. +- Preserve the current exact-vs-estimated semantics unless the task explicitly changes them. +- Reuse existing configuration constants from `rustfs-config` instead of introducing duplicate literals. +- Treat metric names and labels as compatibility-sensitive unless the task explicitly allows a metric contract change. + +## Validation Guidance + +For code changes, prefer focused verification close to the changed behavior: + +- scan correctness tests in `src/scan.rs` +- manager/refresh-state tests in `src/capacity_manager.rs` +- scope propagation tests in `src/capacity_scope.rs` +- integration behavior in `../../rustfs/src/capacity/service.rs` or related app tests when needed + +Useful commands: + +```bash +cargo test -p rustfs-object-capacity +cargo test -p rustfs-object-capacity --doc +cargo bench -p rustfs-object-capacity --bench capacity_scan +``` + +If the change is documentation-only, lightweight verification is enough. + +## Documentation Expectations + +- Keep `README.md` in English. +- Keep `README_ZH.md` aligned when behavior or public usage changes. +- Prefer describing real behavior from code instead of aspirational architecture. + +## Common Pitfalls + +- Assuming `used_capacity` means total disk usage instead of object-data usage. +- Forgetting that the initial full refresh is required before safe dirty-subset refresh. +- Forgetting that partial errors can still produce a returned result. +- Overlooking the fallback path that stores externally supplied disk-used capacity as `DataSource::Fallback`. +- Changing timeout or sampling behavior without reviewing the observable effects on degraded results and metrics. diff --git a/crates/object-capacity/README.md b/crates/object-capacity/README.md new file mode 100644 index 000000000..621f176d9 --- /dev/null +++ b/crates/object-capacity/README.md @@ -0,0 +1,353 @@ +# rustfs-object-capacity + +`rustfs-object-capacity` is the core object-capacity statistics component in RustFS. It scans local data directories, maintains a capacity cache, triggers incremental refreshes after writes, and provides the admin layer with a used-capacity result that is as inexpensive and resilient as possible. + +This crate is not meant to measure total filesystem capacity. Its job is to answer: "How many bytes are currently occupied by RustFS object data?" It makes practical tradeoffs between accuracy, freshness, and scan cost. + +## Core Responsibilities + +- Scan one or more local data-disk roots and aggregate used bytes and file counts. +- Reduce scan cost on large directories with an "exact prefix + sampled overflow" strategy. +- Return usable degraded results when scans time out, traversal stalls, or some directories fail, instead of failing the entire request immediately. +- Maintain a global `HybridCapacityManager` cache with scheduled refresh, write-triggered refresh, foreground blocking refresh, and background refresh. +- Track which disks were affected by writes so the system can refresh only the dirty subset after a complete per-disk cache is available. +- Emit capacity-related metrics for observability and benchmarks. + +## Module Layout + +- `src/lib.rs` + Re-exports `scan_used_capacity_disks`, `CapacityDiskRef`, and `CapacityScanSummary`. +- `src/types.rs` + Defines scan input/output types, including `CapacityDiskRef`, the internal `CapacityScanResult`, and the public `CapacityScanSummary`. +- `src/scan.rs` + Implements directory traversal, sampled estimation, timeout/stall detection, multi-disk concurrent scans, and conversion into `CapacityUpdate`. +- `src/capacity_manager.rs` + Owns caching, write-frequency tracking, singleflight refresh coordination, background tasks, dirty-subset merge logic, and the global singleton manager. +- `src/capacity_scope.rs` + Tracks "which disks were touched by a write", including token-bound local scopes and the global dirty-scope registry. +- `benches/capacity_scan.rs` + Exercises the public scan API with benchmark scenarios for exact, sampled, and multi-disk scans. + +## Data Model + +### `CapacityDiskRef` + +```rust +pub struct CapacityDiskRef { + pub endpoint: String, + pub drive_path: String, +} +``` + +This is the minimal unit required for a scan: + +- `endpoint` is used to distinguish metrics and logs. +- `drive_path` is the local disk root path. + +### `CapacityScanSummary` + +```rust +pub struct CapacityScanSummary { + pub used_bytes: u64, + pub file_count: usize, + pub sampled_count: usize, + pub is_estimated: bool, + pub had_partial_errors: bool, + pub scan_duration: Duration, +} +``` + +Field meanings: + +- `used_bytes`: the computed or estimated used capacity. +- `file_count`: the number of regular files traversed. +- `sampled_count`: the number of overflow files sampled after crossing the threshold. +- `is_estimated`: whether the result is estimated instead of exact. +- `had_partial_errors`: whether traversal encountered local errors while still producing a result. +- `scan_duration`: total scan duration. + +## Scan Algorithm + +The directory scan lives in `scan.rs::get_dir_size_async` and works as follows: + +1. Wrap blocking directory traversal in `tokio::task::spawn_blocking` so the async runtime is not blocked. +2. Walk the directory tree with `WalkDir` and count only regular files. +3. If the file count stays below `DEFAULT_MAX_FILES_THRESHOLD` (default `200_000`), add every file size exactly. +4. After crossing the threshold: + - Keep the first `max_files_threshold` files as an exact prefix. + - Sample every `sample_rate` file after that and estimate the overflow portion from sampled bytes. +5. Periodically perform progress checks: + - If total elapsed time exceeds the timeout, attempt to fall back to a sampled estimate. + - If no file progress is observed within `stall_timeout`, treat the traversal as stalled. +6. If some directory entries or metadata reads fail: + - As long as at least one disk scan succeeds, return a partial-success result. + - Mark the result with `had_partial_errors = true`. + +### Scan Concurrency + +- Multi-disk scans run concurrently through `buffer_unordered`. +- The current hard-coded maximum concurrency is `4` disks. +- A failure on one disk does not immediately stop scans for the others. + +### Timeout and Estimation Fallback + +This crate is intentionally not "timeout means hard failure": + +- If enough sampled data has already been collected, a timeout or stall produces an estimated result. +- Only when no usable estimate is available does the scan return an error. +- This keeps capacity queries useful for large directories, slow disks, and temporary I/O stalls. + +### Symlink Handling + +- Symlinks are not followed by default: `RUSTFS_CAPACITY_FOLLOW_SYMLINKS=false`. +- If enabled, the scan applies circular-reference detection and a maximum follow depth. +- The default maximum depth is `3`. + +## Capacity Cache and Refresh Strategy + +`HybridCapacityManager` is the state center of this crate. + +### Cached State + +- Latest total capacity value `total_used` +- Last refresh time `last_update` +- File count `file_count` +- Estimated/exact flag `is_estimated` +- Data source `DataSource` +- Per-disk cache `disk_cache` +- Dirty-disk set +- Recent 60-second write buckets + +### `DataSource` + +- `RealTime` + Foreground real-time refresh when no cache exists yet. +- `Scheduled` + Background refresh triggered by the scheduled task. +- `WriteTriggered` + Refresh triggered when write frequency is high and the cache is old enough. +- `Fallback` + Fallback to externally supplied disk-used capacity when all scans fail. + +### Refresh Entry Points + +- `refresh_or_join` + A singleflight foreground refresh. If another refresh is already running, callers join and wait for the shared result. +- `spawn_refresh_if_needed` + A background refresh. If another refresh is already running, it is skipped. +- `start_background_task` + Starts two background tasks: + - the scheduled capacity refresh task + - the runtime summary logging task + +### Singleflight Semantics + +`refresh_or_join` and `spawn_refresh_if_needed` use a `watch` channel to coordinate refresh cycles: + +- Only one leader performs the actual refresh at a time. +- Joiners share the same published result after the leader completes. +- Panics inside the refresh function are caught and converted into errors so callers do not crash with the leader. + +## Dirty Scope and Subset Refresh + +One of the main optimizations in this crate is "refresh only the disks dirtied by writes". + +### Scope Propagation + +`capacity_scope.rs` provides two ways to propagate dirty disks: + +- token scope + - The caller first binds a write operation to a disk set with `record_capacity_scope(token, scope)`. + - Later, `record_write_operation_with_scope_token(Some(token))` consumes that scope and marks the disks dirty. +- global dirty scope + - `record_global_dirty_scope(scope)` records dirty disks directly in the global registry. + - The manager drains and merges them during `get_dirty_disks()`. + +### When Dirty-Subset Refresh Is Allowed + +Refreshing only dirty disks is safe only when: + +- `disk_cache_complete == true` +- which means the system has already completed at least one full refresh without partial errors +- and the per-disk cache is fully populated + +If the per-disk cache is incomplete, or there are no dirty disks, the system falls back to a full refresh. + +### Merge Rules After a Subset Refresh + +- On a successful full refresh, `per_disk` replaces the entire `disk_cache`. +- On a successful dirty-subset refresh, only the affected per-disk entries are updated. +- The total capacity is recomputed from the updated `disk_cache` instead of trusting the subset sum directly. +- If a dirty-subset refresh reports partial errors, that cycle fails and the caller falls back to a full refresh to recover consistency. + +## Relationship to the RustFS Main Flow + +This crate provides capacity primitives only. The actual RustFS integration lives in `rustfs/src/capacity/service.rs`. + +The high-level flow is: + +1. Startup calls `init_capacity_management_for_local_disks()`. +2. It collects all local disks and calls `capacity_manager::start_background_task(...)`. +3. Admin used-capacity queries first try the `HybridCapacityManager` cache. +4. If the cache is fresh enough, the cached value is returned directly. +5. If the cache is stale but still acceptable, the stale value is served and a background refresh is triggered. +6. If the cache is very stale and the write rate is high, the request blocks on a foreground refresh. +7. If the initial real-time scan fails, the service falls back to externally supplied disk-used capacity and stores it as `Fallback`. + +`crates/ecstore/src/set_disk.rs` is responsible for recording capacity scopes during object writes, heal operations, data movement, and related flows, so this crate can learn which disks were affected. + +## Public API + +### 1. Direct Scan + +This is useful for benchmarks, operational tooling, or isolated validation. + +```rust +use rustfs_object_capacity::{CapacityDiskRef, scan_used_capacity_disks}; + +let disks = vec![ + CapacityDiskRef { + endpoint: "node-a".to_string(), + drive_path: "/data/disk1".to_string(), + }, +]; + +let summary = scan_used_capacity_disks(&disks).await?; +println!( + "used={} files={} estimated={}", + summary.used_bytes, summary.file_count, summary.is_estimated +); +# Ok::<(), Box>(()) +``` + +### 2. Use the Global Manager + +This is useful for in-service caching and refresh orchestration. + +```rust +use rustfs_object_capacity::capacity_manager::{DataSource, get_capacity_manager}; + +let manager = get_capacity_manager(); + +if let Some(cached) = manager.get_capacity().await { + println!("cached bytes={}", cached.total_used); +} + +manager.record_write_operation().await; + +let _ = manager + .refresh_or_join(DataSource::Scheduled, || async { + rustfs_object_capacity::scan::refresh_capacity_with_scope( + vec![rustfs_object_capacity::CapacityDiskRef { + endpoint: "node-a".to_string(), + drive_path: "/data/disk1".to_string(), + }], + false, + ) + .await + }) + .await; +``` + +### 3. Propagate a Dirty Scope + +```rust +use rustfs_object_capacity::capacity_scope::{ + CapacityScope, CapacityScopeDisk, record_capacity_scope, +}; +use rustfs_object_capacity::capacity_manager::get_capacity_manager; +use uuid::Uuid; + +let token = Uuid::new_v4(); +record_capacity_scope( + token, + CapacityScope { + disks: vec![CapacityScopeDisk { + endpoint: "node-a".to_string(), + drive_path: "/data/disk1".to_string(), + }], + }, +); + +get_capacity_manager() + .record_write_operation_with_scope_token(Some(token)) + .await; +``` + +## Environment Variables and Defaults + +The configuration constants are defined in `crates/config/src/constants/capacity.rs`. + +| Environment Variable | Default | Description | +| --- | --- | --- | +| `RUSTFS_CAPACITY_SCHEDULED_INTERVAL` | `120s` | Scheduled refresh interval | +| `RUSTFS_CAPACITY_WRITE_TRIGGER_DELAY` | `5s` | Debounce delay after writes | +| `RUSTFS_CAPACITY_WRITE_FREQUENCY_THRESHOLD` | `5` | Recent 60-second write-frequency threshold | +| `RUSTFS_CAPACITY_FAST_UPDATE_THRESHOLD` | `30s` | Cache age required before fast refresh is considered | +| `RUSTFS_CAPACITY_MAX_FILES_THRESHOLD` | `200000` | Exact-count file threshold | +| `RUSTFS_CAPACITY_STAT_TIMEOUT` | `3s` | Base scan timeout | +| `RUSTFS_CAPACITY_SAMPLE_RATE` | `200` | Overflow-file sampling interval | +| `RUSTFS_CAPACITY_METRICS_INTERVAL` | `600s` | Runtime summary emission interval | +| `RUSTFS_CAPACITY_FOLLOW_SYMLINKS` | `false` | Whether to follow symlinks | +| `RUSTFS_CAPACITY_MAX_SYMLINK_DEPTH` | `3` | Maximum symlink follow depth | +| `RUSTFS_CAPACITY_ENABLE_DYNAMIC_TIMEOUT` | `true` | Whether to enable dynamic timeout scaling | +| `RUSTFS_CAPACITY_MIN_TIMEOUT` | `2s` | Dynamic-timeout lower bound | +| `RUSTFS_CAPACITY_MAX_TIMEOUT` | `15s` | Dynamic-timeout upper bound | +| `RUSTFS_CAPACITY_STALL_TIMEOUT` | `20s` | Stall-detection threshold | + +### Configuration-Caching Note + +In non-test builds, configuration is cached behind `OnceLock`: + +- Environment variables are effectively read once on first access. +- Updating `RUSTFS_CAPACITY_*` during runtime usually does not take effect immediately. +- A process restart is normally required to apply configuration changes reliably. + +## Metrics + +This crate reports multiple metric families to `rustfs-io-metrics::capacity_metrics`, including: + +- cache hit / miss / served state +- refresh inflight, joiners, and success / error outcomes +- current capacity bytes +- write frequency +- dirty-disk count +- per-disk scan duration, sampling mode, timeout fallback, stall detection, and symlink statistics + +So this crate is both a capacity-calculation component and an important producer of runtime observability data. + +## Benchmarks + +Run the benchmark suite with: + +```bash +cargo bench -p rustfs-object-capacity --bench capacity_scan +``` + +Current benchmark scenarios: + +- `capacity_scan_exact` + Single-disk exact scan over 10k files. +- `capacity_scan_sampled` + Single-disk scan over 202,048 files that triggers sampled estimation. +- `capacity_scan_multi_disk` + Four-disk exact scan with mixed directory sizes. + +## Known Boundaries and Tradeoffs + +- It sums file sizes under RustFS object-data directories; it is not a full replacement for filesystem-level `du`. +- Estimated mode prioritizes bounded cost and usable results over perfect per-run precision. +- Dirty-subset refresh is safe only after a complete per-disk cache has been established. +- Partial errors intentionally try to return a degraded result, which improves availability but means callers should pay attention to `had_partial_errors`. +- Symlink following is disabled by default for safety and determinism. + +## Relevant Source Entry Points + +- [src/lib.rs](./src/lib.rs) +- [src/scan.rs](./src/scan.rs) +- [src/capacity_manager.rs](./src/capacity_manager.rs) +- [src/capacity_scope.rs](./src/capacity_scope.rs) +- [src/types.rs](./src/types.rs) +- [benches/capacity_scan.rs](./benches/capacity_scan.rs) +- [../../rustfs/src/capacity/service.rs](../../rustfs/src/capacity/service.rs) +- [../config/src/constants/capacity.rs](../config/src/constants/capacity.rs) diff --git a/crates/object-capacity/README_ZH.md b/crates/object-capacity/README_ZH.md new file mode 100644 index 000000000..a97b8f253 --- /dev/null +++ b/crates/object-capacity/README_ZH.md @@ -0,0 +1,353 @@ +# rustfs-object-capacity + +`rustfs-object-capacity` 是 RustFS 的对象容量统计核心组件，负责扫描本地数据目录、维护容量缓存、在写入后触发增量刷新，并为上层管理接口提供尽量便宜且可恢复的 used-capacity 结果。 + +这个 crate 的目标不是做“磁盘总容量”探测，而是回答“RustFS 当前对象数据大约占用了多少字节”，并在精确性、实时性、扫描成本之间做工程化折中。 + +## 核心职责 + +- 扫描一个或多个本地数据盘目录，汇总对象数据占用字节数与文件数。 +- 在目录规模较大时使用“前缀精确统计 + 尾部采样估算”降低扫描成本。 +- 在扫描超时、遍历卡住、部分目录失败时尽量保留可用结果，而不是直接让上层完全失效。 +- 维护全局 `HybridCapacityManager` 缓存，支持定时刷新、写触发刷新、前台阻塞刷新和后台异步刷新。 +- 记录写入涉及的脏盘范围，在完整磁盘缓存可用时只刷新 dirty subset，而不是每次全量扫描所有盘。 +- 输出容量相关 metrics，供运行时观测与基准测试使用。 + +## 模块划分 + +- `src/lib.rs` + 对外导出 `scan_used_capacity_disks`、`CapacityDiskRef`、`CapacityScanSummary`。 +- `src/types.rs` + 定义扫描输入输出类型，包括 `CapacityDiskRef`、内部 `CapacityScanResult` 和公开 `CapacityScanSummary`。 +- `src/scan.rs` + 负责真实目录遍历、采样估算、超时/卡顿检测、多盘并发扫描，以及把扫描结果转换成 `CapacityUpdate`。 +- `src/capacity_manager.rs` + 负责缓存、写频率统计、singleflight 刷新协调、后台定时任务、dirty subset 合并和全局单例管理。 +- `src/capacity_scope.rs` + 负责“写操作影响了哪些磁盘”的范围传播，包括 token 绑定的局部 scope 和全局 dirty scope 注册表。 +- `benches/capacity_scan.rs` + 使用公开扫描 API 做基准，覆盖单盘精确扫描、单盘采样扫描和多盘扫描。 + +## 数据模型 + +### `CapacityDiskRef` + +```rust +pub struct CapacityDiskRef { + pub endpoint: String, + pub drive_path: String, +} +``` + +它是扫描入口的最小描述单元： + +- `endpoint` 用于指标标签和日志区分。 +- `drive_path` 是本地磁盘根目录。 + +### `CapacityScanSummary` + +```rust +pub struct CapacityScanSummary { + pub used_bytes: u64, + pub file_count: usize, + pub sampled_count: usize, + pub is_estimated: bool, + pub had_partial_errors: bool, + pub scan_duration: Duration, +} +``` + +字段语义： + +- `used_bytes`：本次扫描或估算得到的容量。 +- `file_count`：遍历到的普通文件数量。 +- `sampled_count`：超过阈值后被抽样统计的 overflow 文件数。 +- `is_estimated`：是否为估算值。 +- `had_partial_errors`：遍历中是否出现局部错误但整体仍返回了结果。 +- `scan_duration`：扫描耗时。 + +## 扫描算法 + +目录扫描实现在 `scan.rs::get_dir_size_async`，核心逻辑如下： + +1. 用 `tokio::task::spawn_blocking` 包裹阻塞型目录遍历，避免阻塞 async runtime。 +2. 通过 `WalkDir` 遍历目录树，只统计普通文件大小。 +3. 当文件数未超过 `DEFAULT_MAX_FILES_THRESHOLD`（默认 `200_000`）时，逐文件精确累加。 +4. 超过阈值后： + - 前 `max_files_threshold` 个文件继续作为精确前缀保留。 + - 之后每隔 `sample_rate` 个文件采样一次，基于 sampled bytes 估算 overflow 部分。 +5. 周期性做进度检查： + - 若总耗时超过 timeout，则尝试退化为采样估算结果。 + - 若在 `stall_timeout` 内没有任何新文件进展，则判定为 stall。 +6. 若遍历中部分目录或元数据读取失败： + - 只要仍有至少一个磁盘成功，就返回部分成功结果。 + - 同时设置 `had_partial_errors = true`。 + +### 扫描并发 + +- 多盘扫描使用 `buffer_unordered` 并发执行。 +- 当前硬编码最大并发为 `4` 个磁盘。 +- 单次磁盘扫描失败不会立即中断其它磁盘。 + +### 超时与估算退化 + +crate 不是“超时就直接失败”的设计： + +- 如果已经收集到足够的采样数据，超时或 stall 时会返回估算值。 +- 如果还没有可用采样，才会真正报错。 +- 这保证了大目录、慢盘、暂时抖动场景下，上层依然能拿到近似可用的容量值。 + +### 符号链接处理 + +- 默认不跟随符号链接：`RUSTFS_CAPACITY_FOLLOW_SYMLINKS=false`。 +- 开启后会做循环引用检测和最大深度限制。 +- 默认最大深度是 `3`。 + +## 容量缓存与刷新策略 + +`HybridCapacityManager` 是这个 crate 的状态中心。 + +### 缓存内容 + +- 最近一次容量值 `total_used` +- 更新时间 `last_update` +- 文件数 `file_count` +- 是否估算值 `is_estimated` +- 数据来源 `DataSource` +- 每盘缓存 `disk_cache` +- dirty disk 集合 +- 最近 60 秒写入桶统计 + +### `DataSource` + +- `RealTime` + 首次无缓存时的前台实时刷新。 +- `Scheduled` + 定时后台刷新。 +- `WriteTriggered` + 写入频率高且缓存偏旧时触发的刷新。 +- `Fallback` + 全部扫描失败时，回退到外部传入的磁盘 used capacity。 + +### 刷新入口 + +- `refresh_or_join` + singleflight 前台刷新。若已有刷新进行中，调用方加入等待，不重复扫描。 +- `spawn_refresh_if_needed` + 后台异步刷新。若已有刷新在进行，则直接跳过。 +- `start_background_task` + 启动两个后台任务： + - 定时容量刷新任务 + - 定时 runtime summary 日志任务 + +### singleflight 语义 + +`refresh_or_join` / `spawn_refresh_if_needed` 通过 `watch` channel 协调刷新： + +- 同一时刻只允许一个 leader 真正执行 refresh。 +- joiner 在 leader 完成后共享同一份结果。 +- refresh panic 会被捕获并转换为错误，避免把调用者一起打崩。 + +## Dirty Scope 与子集刷新 + +这个 crate 的一个关键优化是“写后只刷新脏盘”。 + +### Scope 传播 + +`capacity_scope.rs` 提供两种脏盘传播方式： + +- token scope + - 调用方先用 `record_capacity_scope(token, scope)` 把一次写操作关联到一组磁盘。 + - 后续 `record_write_operation_with_scope_token(Some(token))` 会取出该 scope，并把磁盘标记为 dirty。 +- global dirty scope + - 通过 `record_global_dirty_scope(scope)` 直接记录全局脏盘。 + - manager 在 `get_dirty_disks()` 时会 drain 这些全局脏盘并合并。 + +### 何时允许 dirty subset refresh + +不是所有时候都能只刷脏盘。前提是： + +- `disk_cache_complete == true` +- 也就是系统已经完成过一次“无部分错误”的全盘刷新 +- 并且成功拿到了每盘缓存 + +若当前还没有完整 per-disk cache，或者脏盘集合为空，就会回退到全盘刷新。 + +### 子集刷新后的合并规则 + +- 全盘刷新成功时，`per_disk` 会完整替换 `disk_cache`。 +- dirty subset 刷新成功时，只更新对应脏盘的缓存条目。 +- 总容量会基于更新后的 `disk_cache` 重新求和，而不是盲信子集扫描返回的局部和。 +- 若 dirty subset 刷新出现 partial errors，则当前轮次失败，并回退到全盘刷新恢复一致性。 + +## 与 RustFS 主流程的关系 + +这个 crate 本身只提供容量能力，真正把它接到 RustFS 主流程的是 `rustfs/src/capacity/service.rs`。 + +上层使用方式大致如下： + +1. 启动时调用 `init_capacity_management_for_local_disks()`。 +2. 它收集所有本地盘，调用 `capacity_manager::start_background_task(...)`。 +3. 管理接口查询 used capacity 时，优先读 `HybridCapacityManager` 缓存。 +4. 缓存足够新时直接返回。 +5. 缓存过旧但还可容忍时，先返回 stale cache，再后台刷新。 +6. 缓存极旧且写入频率高时，前台阻塞刷新。 +7. 若首次实时扫描失败，则退回外部已有的磁盘 used capacity，并写入 `Fallback` 缓存。 + +`crates/ecstore/src/set_disk.rs` 中则负责在对象写入、heal、data movement 等流程里记录 capacity scope，把“这次写影响了哪些盘”传播给本 crate。 + +## 公开 API + +### 1. 直接扫描 + +适合 benchmark、运维工具或独立验证路径。 + +```rust +use rustfs_object_capacity::{CapacityDiskRef, scan_used_capacity_disks}; + +let disks = vec![ + CapacityDiskRef { + endpoint: "node-a".to_string(), + drive_path: "/data/disk1".to_string(), + }, +]; + +let summary = scan_used_capacity_disks(&disks).await?; +println!( + "used={} files={} estimated={}", + summary.used_bytes, summary.file_count, summary.is_estimated +); +# Ok::<(), Box>(()) +``` + +### 2. 使用全局 manager + +适合服务内缓存与刷新控制。 + +```rust +use rustfs_object_capacity::capacity_manager::{DataSource, get_capacity_manager}; + +let manager = get_capacity_manager(); + +if let Some(cached) = manager.get_capacity().await { + println!("cached bytes={}", cached.total_used); +} + +manager.record_write_operation().await; + +let _ = manager + .refresh_or_join(DataSource::Scheduled, || async { + rustfs_object_capacity::scan::refresh_capacity_with_scope( + vec![rustfs_object_capacity::CapacityDiskRef { + endpoint: "node-a".to_string(), + drive_path: "/data/disk1".to_string(), + }], + false, + ) + .await + }) + .await; +``` + +### 3. 传播 dirty scope + +```rust +use rustfs_object_capacity::capacity_scope::{ + CapacityScope, CapacityScopeDisk, record_capacity_scope, +}; +use rustfs_object_capacity::capacity_manager::get_capacity_manager; +use uuid::Uuid; + +let token = Uuid::new_v4(); +record_capacity_scope( + token, + CapacityScope { + disks: vec![CapacityScopeDisk { + endpoint: "node-a".to_string(), + drive_path: "/data/disk1".to_string(), + }], + }, +); + +get_capacity_manager() + .record_write_operation_with_scope_token(Some(token)) + .await; +``` + +## 环境变量与默认值 + +配置常量定义在 `crates/config/src/constants/capacity.rs`。 + +| 环境变量 | 默认值 | 说明 | +| --- | --- | --- | +| `RUSTFS_CAPACITY_SCHEDULED_INTERVAL` | `120s` | 定时刷新间隔 | +| `RUSTFS_CAPACITY_WRITE_TRIGGER_DELAY` | `5s` | 写后防抖延迟 | +| `RUSTFS_CAPACITY_WRITE_FREQUENCY_THRESHOLD` | `5` | 最近 60 秒写频率阈值 | +| `RUSTFS_CAPACITY_FAST_UPDATE_THRESHOLD` | `30s` | 缓存超过该年龄后才考虑快速刷新 | +| `RUSTFS_CAPACITY_MAX_FILES_THRESHOLD` | `200000` | 精确统计文件数阈值 | +| `RUSTFS_CAPACITY_STAT_TIMEOUT` | `3s` | 基础扫描超时 | +| `RUSTFS_CAPACITY_SAMPLE_RATE` | `200` | overflow 文件采样间隔 | +| `RUSTFS_CAPACITY_METRICS_INTERVAL` | `600s` | runtime summary 打点间隔 | +| `RUSTFS_CAPACITY_FOLLOW_SYMLINKS` | `false` | 是否跟随符号链接 | +| `RUSTFS_CAPACITY_MAX_SYMLINK_DEPTH` | `3` | 符号链接最大跟随深度 | +| `RUSTFS_CAPACITY_ENABLE_DYNAMIC_TIMEOUT` | `true` | 是否启用动态超时 | +| `RUSTFS_CAPACITY_MIN_TIMEOUT` | `2s` | 动态超时下界 | +| `RUSTFS_CAPACITY_MAX_TIMEOUT` | `15s` | 动态超时上界 | +| `RUSTFS_CAPACITY_STALL_TIMEOUT` | `20s` | 无进展 stall 判定阈值 | + +### 配置缓存注意事项 + +在非测试构建中，配置通过 `OnceLock` 缓存： + +- 环境变量只在首次读取时生效。 +- 运行中修改 `RUSTFS_CAPACITY_*` 通常不会即时生效。 +- 需要重启进程才能稳定应用新配置。 + +## Metrics + +这个 crate 会向 `rustfs-io-metrics::capacity_metrics` 上报多类指标，包括但不限于： + +- cache hit / miss / served 状态 +- refresh inflight、joiner、success / error +- 当前容量字节数 +- 写频率 +- dirty disk 数量 +- 单盘扫描耗时、采样模式、timeout fallback、stall、symlink 统计 + +因此它既是容量计算模块，也是容量观测数据的重要生产者。 + +## 基准测试 + +运行基准： + +```bash +cargo bench -p rustfs-object-capacity --bench capacity_scan +``` + +当前 bench 场景： + +- `capacity_scan_exact` + 单盘 10k 文件精确扫描。 +- `capacity_scan_sampled` + 单盘 202,048 文件，触发采样估算。 +- `capacity_scan_multi_disk` + 四盘混合规模精确扫描。 + +## 已知边界与设计取舍 + +- 它统计的是对象数据目录中文件大小之和，不是文件系统 `du` 的完全等价替代。 +- 估算模式优先保证成本可控和结果可用，不保证逐次完全精确。 +- dirty subset refresh 只有在完整 per-disk cache 已建立后才安全。 +- 部分错误会尽量返回 degraded result，这对可用性更友好，但也意味着调用方需要识别 `had_partial_errors`。 +- symlink 默认关闭，是出于安全性和结果确定性考虑。 + +## 相关源码入口 + +- [src/lib.rs](./src/lib.rs) +- [src/scan.rs](./src/scan.rs) +- [src/capacity_manager.rs](./src/capacity_manager.rs) +- [src/capacity_scope.rs](./src/capacity_scope.rs) +- [src/types.rs](./src/types.rs) +- [benches/capacity_scan.rs](./benches/capacity_scan.rs) +- [../../rustfs/src/capacity/service.rs](../../rustfs/src/capacity/service.rs) +- [../config/src/constants/capacity.rs](../config/src/constants/capacity.rs) diff --git a/crates/rio/src/http_reader.rs b/crates/rio/src/http_reader.rs index 956b21dc7..ade420996 100644 --- a/crates/rio/src/http_reader.rs +++ b/crates/rio/src/http_reader.rs @@ -47,6 +47,162 @@ const READ_FILE_STREAM_PATH: &str = "/rustfs/rpc/read_file_stream"; const PUT_FILE_STREAM_PATH: &str = "/rustfs/rpc/put_file_stream"; const WALK_DIR_PATH: &str = "/rustfs/rpc/walk_dir"; +#[derive(Debug, Clone, Copy, Eq, PartialEq)] +pub enum InternodeHttpErrorKind { + ConnectTimeout, + ConnectionRefused, + DnsResolutionFailed, + ConnectionReset, + BodyStreamAborted, + HttpStatus(reqwest::StatusCode), + Unknown, +} + +impl InternodeHttpErrorKind { + pub fn is_retryable(self) -> bool { + match self { + Self::ConnectTimeout | Self::ConnectionRefused | Self::ConnectionReset | Self::BodyStreamAborted => true, + Self::HttpStatus(status) => matches!( + status, + reqwest::StatusCode::TOO_MANY_REQUESTS + | reqwest::StatusCode::BAD_GATEWAY + | reqwest::StatusCode::SERVICE_UNAVAILABLE + | reqwest::StatusCode::GATEWAY_TIMEOUT + ), + Self::DnsResolutionFailed | Self::Unknown => false, + } + } + + fn io_error_kind(self) -> io::ErrorKind { + match self { + Self::ConnectTimeout => io::ErrorKind::TimedOut, + Self::ConnectionRefused => io::ErrorKind::ConnectionRefused, + Self::DnsResolutionFailed => io::ErrorKind::AddrNotAvailable, + Self::ConnectionReset => io::ErrorKind::ConnectionReset, + Self::BodyStreamAborted => io::ErrorKind::BrokenPipe, + Self::HttpStatus(_) | Self::Unknown => io::ErrorKind::Other, + } + } + + pub fn metric_label(self) -> &'static str { + match self { + Self::ConnectTimeout => "connect_timeout", + Self::ConnectionRefused => "connection_refused", + Self::DnsResolutionFailed => "dns_resolution_failed", + Self::ConnectionReset => "connection_reset", + Self::BodyStreamAborted => "body_stream_aborted", + Self::HttpStatus(status) => match status.as_u16() { + 429 => "http_429", + 502 => "http_502", + 503 => "http_503", + 504 => "http_504", + _ => "http_status_other", + }, + Self::Unknown => "unknown", + } + } +} + +impl std::fmt::Display for InternodeHttpErrorKind { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::ConnectTimeout => write!(f, "internode connect timeout"), + Self::ConnectionRefused => write!(f, "internode connection refused"), + Self::DnsResolutionFailed => write!(f, "internode dns resolution failed"), + Self::ConnectionReset => write!(f, "internode connection reset"), + Self::BodyStreamAborted => write!(f, "internode body stream aborted"), + Self::HttpStatus(status) => write!(f, "internode http status {status}"), + Self::Unknown => write!(f, "internode request failed"), + } + } +} + +#[derive(Debug, Clone, Eq, PartialEq)] +pub struct InternodeHttpRequestContext { + method: String, + target: String, + operation: Option<&'static str>, +} + +impl InternodeHttpRequestContext { + pub fn method(&self) -> &str { + &self.method + } + + pub fn target(&self) -> &str { + &self.target + } + + pub fn operation(&self) -> Option<&'static str> { + self.operation + } +} + +impl std::fmt::Display for InternodeHttpRequestContext { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{} {}", self.method, self.target) + } +} + +#[derive(Debug, thiserror::Error)] +#[error("{kind}: {context}")] +pub struct InternodeHttpError { + kind: InternodeHttpErrorKind, + context: InternodeHttpRequestContext, + #[source] + source: Option>, +} + +impl InternodeHttpError { + pub fn kind(&self) -> InternodeHttpErrorKind { + self.kind + } + + pub fn context(&self) -> &InternodeHttpRequestContext { + &self.context + } + + fn new(kind: InternodeHttpErrorKind, context: InternodeHttpRequestContext) -> Self { + Self { + kind, + context, + source: None, + } + } + + #[doc(hidden)] + pub fn new_for_test(kind: InternodeHttpErrorKind) -> Self { + Self::new( + kind, + InternodeHttpRequestContext { + method: "PUT".to_string(), + target: "/rustfs/rpc/put_file_stream".to_string(), + operation: Some(INTERNODE_OPERATION_PUT_FILE_STREAM), + }, + ) + } + + fn with_source(kind: InternodeHttpErrorKind, context: InternodeHttpRequestContext, source: E) -> Self + where + E: std::error::Error + Send + Sync + 'static, + { + Self { + kind, + context, + source: Some(Box::new(source)), + } + } + + fn into_io_error(self) -> io::Error { + io::Error::new(self.kind.io_error_kind(), self) + } +} + +#[doc(hidden)] +pub fn new_test_internode_http_io_error(kind: InternodeHttpErrorKind) -> io::Error { + InternodeHttpError::new_for_test(kind).into_io_error() +} + fn add_root_certificates_from_der(builder: reqwest::ClientBuilder, certs_der: &[Vec]) -> reqwest::ClientBuilder { let mut b = builder; for der in certs_der { @@ -189,6 +345,75 @@ async fn get_http_client(url: &str) -> Client { return_client } +fn internode_request_context(method: &Method, url: &str, operation: Option<&'static str>) -> InternodeHttpRequestContext { + let target = reqwest::Url::parse(url) + .ok() + .map(|parsed| match parsed.query() { + Some(query) => format!("{}?{query}", parsed.path()), + None => parsed.path().to_string(), + }) + .unwrap_or_else(|| url.to_string()); + InternodeHttpRequestContext { + method: method.to_string(), + target, + operation, + } +} + +fn classify_reqwest_error(err: &reqwest::Error) -> InternodeHttpErrorKind { + if err.is_timeout() { + return InternodeHttpErrorKind::ConnectTimeout; + } + + let message = err.to_string().to_ascii_lowercase(); + if err.is_connect() { + if message.contains("dns") + || message.contains("name or service not known") + || message.contains("failed to lookup address") + { + return InternodeHttpErrorKind::DnsResolutionFailed; + } + if message.contains("refused") { + return InternodeHttpErrorKind::ConnectionRefused; + } + } + + if message.contains("connection reset") || message.contains("broken pipe") || message.contains("connection aborted") { + return InternodeHttpErrorKind::ConnectionReset; + } + if message.contains("body") || message.contains("stream") { + return InternodeHttpErrorKind::BodyStreamAborted; + } + + InternodeHttpErrorKind::Unknown +} + +fn classify_http_status(status: reqwest::StatusCode) -> InternodeHttpErrorKind { + InternodeHttpErrorKind::HttpStatus(status) +} + +fn internode_reqwest_error(method: &Method, url: &str, operation: Option<&'static str>, err: reqwest::Error) -> io::Error { + let context = internode_request_context(method, url, operation); + let classified = classify_reqwest_error(&err); + InternodeHttpError::with_source(classified, context, err).into_io_error() +} + +fn internode_classified_error( + method: &Method, + url: &str, + operation: Option<&'static str>, + kind: InternodeHttpErrorKind, +) -> io::Error { + let context = internode_request_context(method, url, operation); + InternodeHttpError::new(kind, context).into_io_error() +} + +fn internode_status_error(method: &Method, url: &str, operation: Option<&'static str>, status: reqwest::StatusCode) -> io::Error { + let context = internode_request_context(method, url, operation); + let classified = classify_http_status(status); + InternodeHttpError::new(classified, context).into_io_error() +} + pin_project! { pub struct HttpReader { url:String, @@ -248,15 +473,14 @@ impl HttpReader { let resp = request.send().await.map_err(|e| { record_internode_error(track_internode_metrics, internode_operation); - Error::other(format!("HttpReader HTTP request error for {method} {url}: {e}")) + record_internode_classified_error(track_internode_metrics, internode_operation, classify_reqwest_error(&e)); + internode_reqwest_error(&method, &url, internode_operation, e) })?; if resp.status().is_success().not() { record_internode_error(track_internode_metrics, internode_operation); - return Err(Error::other(format!( - "HttpReader HTTP request failed for {method} {url} with non-200 status {}", - resp.status(), - ))); + record_internode_classified_error(track_internode_metrics, internode_operation, classify_http_status(resp.status())); + return Err(internode_status_error(&method, &url, internode_operation, resp.status())); } record_internode_outgoing_request(track_internode_metrics, internode_operation); @@ -265,7 +489,8 @@ impl HttpReader { let stream_error_method = method.clone(); let stream = resp.bytes_stream().map_err(move |e| { record_internode_error(track_internode_metrics, internode_operation); - Error::other(format!("HttpReader stream error for {stream_error_method} {stream_error_url}: {e}")) + record_internode_classified_error(track_internode_metrics, internode_operation, classify_reqwest_error(&e)); + internode_reqwest_error(&stream_error_method, &stream_error_url, internode_operation, e) }); Ok(Self { @@ -426,7 +651,7 @@ impl HttpWriter { let client = get_http_client(&url_clone).await; let request = client - .request(method_clone, url_clone.clone()) + .request(method_clone.clone(), url_clone.clone()) .headers(headers_clone.clone()) .body(body); @@ -438,18 +663,24 @@ impl HttpWriter { // http_log!("[HttpWriter::spawn] got response: status={}", resp.status()); if !resp.status().is_success() { record_internode_error(track_internode_metrics, internode_operation); - let _ = err_tx.send(Error::other(format!( - "HttpWriter HTTP request failed with non-200 status {}", - resp.status() - ))); - return Err(Error::other(format!("HTTP request failed with non-200 status {}", resp.status()))); + record_internode_classified_error( + track_internode_metrics, + internode_operation, + classify_http_status(resp.status()), + ); + let status = resp.status(); + let io_err = internode_status_error(&method_clone, &url_clone, internode_operation, status); + let _ = err_tx.send(internode_status_error(&method_clone, &url_clone, internode_operation, status)); + return Err(io_err); } } Err(e) => { record_internode_error(track_internode_metrics, internode_operation); - // http_log!("[HttpWriter::spawn] HTTP request error: {e}"); - let _ = err_tx.send(Error::other(format!("HTTP request failed: {e}"))); - return Err(Error::other(format!("HTTP request failed: {e}"))); + let classified = classify_reqwest_error(&e); + record_internode_classified_error(track_internode_metrics, internode_operation, classified); + let _ = err_tx.send(internode_classified_error(&method_clone, &url_clone, internode_operation, classified)); + let io_err = internode_reqwest_error(&method_clone, &url_clone, internode_operation, e); + return Err(io_err); } } @@ -553,6 +784,20 @@ fn record_internode_error(track: bool, operation: Option<&'static str>) { } } +fn record_internode_classified_error(track: bool, operation: Option<&'static str>, classification: InternodeHttpErrorKind) { + if !track { + return; + } + + if let Some(operation) = operation { + global_internode_metrics().record_classified_error_for_operation_and_backend( + operation, + INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, + classification.metric_label(), + ); + } +} + fn poll_send_error_to_io(err: tokio_util::sync::PollSendError, context: &str) -> io::Error { Error::other(format!("{context}: {err}")) } @@ -915,9 +1160,72 @@ mod tests { Err(err) => err, }; - let err_text = err.to_string(); - assert!(err_text.contains("HttpReader HTTP request error for GET")); - assert!(err_text.contains(&url)); + let source = err + .get_ref() + .and_then(|source| source.downcast_ref::()) + .expect("expected InternodeHttpError source"); + assert_eq!(source.context().method(), "GET"); + assert!(source.context().target().contains("/stream")); + } + + #[test] + fn classify_http_status_marks_retryable_gateway_errors() { + let unavailable = classify_http_status(reqwest::StatusCode::SERVICE_UNAVAILABLE); + let bad_gateway = classify_http_status(reqwest::StatusCode::BAD_GATEWAY); + let bad_request = classify_http_status(reqwest::StatusCode::BAD_REQUEST); + + assert!(unavailable.is_retryable()); + assert!(bad_gateway.is_retryable()); + assert!(!bad_request.is_retryable()); + } + + #[test] + fn dns_resolution_error_uses_network_io_kind() { + let err = internode_classified_error( + &Method::GET, + "http://missing.invalid/rustfs/rpc/read_file_stream", + Some(INTERNODE_OPERATION_READ_FILE_STREAM), + InternodeHttpErrorKind::DnsResolutionFailed, + ); + + assert_eq!(err.kind(), io::ErrorKind::AddrNotAvailable); + assert!(err.to_string().contains("internode dns resolution failed")); + assert!(err.to_string().contains("GET /rustfs/rpc/read_file_stream")); + } + + #[test] + fn internode_status_error_preserves_classification_source_and_context() { + let err = internode_status_error( + &Method::PUT, + "http://node:9000/rustfs/rpc/put_file_stream?disk=disk-a", + Some(INTERNODE_OPERATION_PUT_FILE_STREAM), + reqwest::StatusCode::SERVICE_UNAVAILABLE, + ); + + let source = err + .get_ref() + .and_then(|source| source.downcast_ref::()) + .expect("expected status error to carry InternodeHttpError source"); + assert_eq!( + source.kind(), + InternodeHttpErrorKind::HttpStatus(reqwest::StatusCode::SERVICE_UNAVAILABLE) + ); + assert_eq!(source.context().method(), "PUT"); + assert!(source.context().target().contains(PUT_FILE_STREAM_PATH)); + } + + #[test] + fn test_internode_http_error_test_helper_is_retryable() { + let err = new_test_internode_http_io_error(InternodeHttpErrorKind::ConnectionReset); + let source = err + .get_ref() + .and_then(|source| source.downcast_ref::()) + .expect("expected test helper to carry InternodeHttpError source"); + + assert_eq!(source.kind(), InternodeHttpErrorKind::ConnectionReset); + assert!(source.kind().is_retryable()); + assert_eq!(source.context().method(), "PUT"); + assert!(source.context().target().contains(PUT_FILE_STREAM_PATH)); } #[test] diff --git a/rustfs/src/storage/rpc/http_service.rs b/rustfs/src/storage/rpc/http_service.rs index dfbe5a0e0..8a6205d64 100644 --- a/rustfs/src/storage/rpc/http_service.rs +++ b/rustfs/src/storage/rpc/http_service.rs @@ -280,18 +280,30 @@ async fn handle_put_file(req: Request) -> Response { let mut file = if query.append { match disk.append_file(&query.volume, &query.path).await { Ok(file) => file, - Err(e) => return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("append file err {e}")), + Err(e) => { + let message = put_file_stage_error_message("append", &query, &e); + warn!("{message}"); + return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, message); + } } } else { match disk.create_file("", &query.volume, &query.path, query.size).await { Ok(file) => file, - Err(e) => return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("create file err {e}")), + Err(e) => { + let message = put_file_stage_error_message("create", &query, &e); + warn!("{message}"); + return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, message); + } } }; let copied = match write_body_chunks_to_writer(req.into_body().into_data_stream(), &mut file).await { Ok(copied) => copied, - Err(e) => return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("write file err {e}")), + Err(e) => { + let message = put_file_stage_error_message("write_body", &query, &e); + warn!("{message}"); + return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, message); + } }; global_internode_metrics().record_incoming_request_for_operation_and_backend( @@ -305,7 +317,9 @@ async fn handle_put_file(req: Request) -> Response { ); if let Err(e) = file.flush().await { - return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("write file err {e}")); + let message = put_file_stage_error_message("flush", &query, &e); + warn!("{message}"); + return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, message); } empty_ok() @@ -364,6 +378,13 @@ fn response_with_status(status: StatusCode, message: impl Into) -> Respo .expect("failed to build rpc error response") } +fn put_file_stage_error_message(stage: &str, query: &PutFileQuery, err: &dyn std::fmt::Display) -> String { + format!( + "{stage} file err {err} [disk={}, volume={}, path={}, append={}, size={}]", + query.disk, query.volume, query.path, query.append, query.size + ) +} + #[cfg(test)] mod tests { use super::*; @@ -404,6 +425,25 @@ mod tests { assert_eq!(response.status(), StatusCode::FORBIDDEN); } + #[test] + fn put_file_stage_error_message_includes_stage_and_request_context() { + let query = PutFileQuery { + disk: "disk-a".to_string(), + volume: ".rustfs.sys/tmp".to_string(), + path: "tmp/object/part.1".to_string(), + append: false, + size: 1024, + }; + + let msg = put_file_stage_error_message("write_body", &query, &"connection reset"); + assert!(msg.contains("write_body")); + assert!(msg.contains("disk=disk-a")); + assert!(msg.contains("volume=.rustfs.sys/tmp")); + assert!(msg.contains("path=tmp/object/part.1")); + assert!(msg.contains("append=false")); + assert!(msg.contains("size=1024")); + } + #[tokio::test] async fn write_body_chunks_to_writer_streams_all_chunks() { let (mut reader, mut writer) = tokio::io::duplex(64);