gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-27 16:48:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' of github.com:rustfs/rustfs into houseme/get-small-file-optimization

Browse Source 
* 'main' of github.com:rustfs/rustfs:
  fix(ecstore): replace unwrap() with proper error handling in api_get_object_attributes (#729 batch 11) (#3991)
  fix(ecstore): improve expect() messages in admin_server_info (#729 batch 9) (#3990)
  fix(server): improve expect() messages in layer.rs (#729 batch 8) (#3989)
  fix(ecstore): improve expect() messages in replication_resyncer (#729 batch 7) (#3988)
  fix(admin): improve expect() messages in remaining admin handlers (#729 batch 6) (#3987)
  fix(admin): improve expect() messages in policies handler (#729 batch 5) (#3986)
  fix(admin): improve expect() messages in user handler (#729 batch 4) (#3985)
  fix(admin): replace unwrap() with safe pattern in bucket_meta handler (#729 batch 3) (#3984)
  feat(get): harden codec streaming rollout (#3981)
  fix(admin): replace unwrap() with proper error handling in tier handler (#729 batch 2) (#3983)

# Conflicts:
#	crates/ecstore/src/set_disk/mod.rs
This commit is contained in:
houseme
2026-06-28 11:31:06 +08:00
parent e3e5f3c746 1f8a5bc095
commit 4c83669c30
22 changed files with 2946 additions and 444 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/replication/replication_resyncer.rs
+7 -7
View File
@@ -609,7 +609,7 @@ impl ReplicationResyncer {
let mut bucket_status = BucketReplicationResyncStatus::new();
bucket_status.id = 0;
status_map.insert(opts.bucket.clone(), bucket_status);
status_map.get_mut(&opts.bucket).unwrap()
status_map.get_mut(&opts.bucket).expect("bucket should be in status map")
};
let state = if let Some(state) = bucket_status.targets_map.get_mut(&opts.arn) {
@@ -617,7 +617,7 @@ impl ReplicationResyncer {
} else {
let state = TargetReplicationResyncStatus::new();
bucket_status.targets_map.insert(opts.arn.clone(), state);
bucket_status.targets_map.get_mut(&opts.arn).unwrap()
bucket_status.targets_map.get_mut(&opts.arn).expect("ARN should be in targets map")
};
if !resync_state_accepts_update(state, &opts) {
@@ -670,7 +670,7 @@ impl ReplicationResyncer {
let mut bucket_status = BucketReplicationResyncStatus::new();
bucket_status.id = 0;
status_map.insert(opts.bucket.clone(), bucket_status);
status_map.get_mut(&opts.bucket).unwrap()
status_map.get_mut(&opts.bucket).expect("bucket should be in status map")
};
let state = if let Some(state) = bucket_status.targets_map.get_mut(&opts.arn) {
@@ -678,7 +678,7 @@ impl ReplicationResyncer {
} else {
let state = TargetReplicationResyncStatus::new();
bucket_status.targets_map.insert(opts.arn.clone(), state);
bucket_status.targets_map.get_mut(&opts.arn).unwrap()
bucket_status.targets_map.get_mut(&opts.arn).expect("ARN should be in targets map")
};
if !resync_state_accepts_update(state, &opts) {
@@ -764,7 +764,7 @@ impl ReplicationResyncer {
"Failed to persist resync status"
);
} else {
last_update_times.insert(bucket.clone(), status.last_update.unwrap());
last_update_times.insert(bucket.clone(), status.last_update.expect("last_update should be set"));
}
}
}
@@ -2710,7 +2710,7 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli
&& !tgt_client.reset_id.is_empty()
&& dobj.op_type == ReplicationType::ExistingObject
{
rinfo.resync_timestamp = format!("{};{}", OffsetDateTime::now_utc().format(&Rfc3339).unwrap(), tgt_client.reset_id);
rinfo.resync_timestamp = format!("{};{}", OffsetDateTime::now_utc().format(&Rfc3339).unwrap_or_else(|_| "invalid-time".to_string()), tgt_client.reset_id);
}
rinfo
@@ -3474,7 +3474,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
&& !tgt_client.reset_id.is_empty()
{
rinfo.resync_timestamp =
format!("{};{}", OffsetDateTime::now_utc().format(&Rfc3339).unwrap(), tgt_client.reset_id);
format!("{};{}", OffsetDateTime::now_utc().format(&Rfc3339).unwrap_or_else(|_| "invalid-time".to_string()), tgt_client.reset_id);
rinfo.replication_resynced = true;
}
crates/ecstore/src/client/api_get_object_attributes.rs
+24 -9
View File
@@ -133,11 +133,23 @@ struct ObjectAttributePart {
impl ObjectAttributes {
pub async fn parse_response(&mut self, h: &HeaderMap, body_vec: Vec<u8>) -> Result<(), std::io::Error> {
let mod_time = OffsetDateTime::parse(h.get("Last-Modified").unwrap().to_str().unwrap(), ISO8601_DATEFORMAT).unwrap(); //RFC7231Time
let last_modified = h.get("Last-Modified")
.ok_or_else(|| std::io::Error::other("missing Last-Modified header"))?
.to_str()
.map_err(|e| std::io::Error::other(format!("invalid Last-Modified header: {e}")))?;
let mod_time = OffsetDateTime::parse(last_modified, ISO8601_DATEFORMAT)
.map_err(|e| std::io::Error::other(format!("invalid Last-Modified date: {e}")))?;
self.last_modified = mod_time;
self.version_id = h.get(X_AMZ_VERSION_ID).unwrap().to_str().unwrap().to_string();
let mut response = match quick_xml::de::from_str::<ObjectAttributesResponse>(&String::from_utf8(body_vec).unwrap()) {
let version_id = h.get(X_AMZ_VERSION_ID)
.ok_or_else(|| std::io::Error::other("missing version ID header"))?
.to_str()
.map_err(|e| std::io::Error::other(format!("invalid version ID header: {e}")))?;
self.version_id = version_id.to_string();
let body_str = String::from_utf8(body_vec)
.map_err(|e| std::io::Error::other(format!("invalid UTF-8 body: {e}")))?;
let mut response = match quick_xml::de::from_str::<ObjectAttributesResponse>(&body_str) {
Ok(result) => result,
Err(err) => {
return Err(std::io::Error::other(err.to_string()));
@@ -163,21 +175,21 @@ impl TransitionClient {
}
let mut headers = HeaderMap::new();
headers.insert(X_AMZ_OBJECT_ATTRIBUTES, HeaderValue::from_str(GET_OBJECT_ATTRIBUTES_TAGS).unwrap());
headers.insert(X_AMZ_OBJECT_ATTRIBUTES, HeaderValue::from_str(GET_OBJECT_ATTRIBUTES_TAGS).expect("valid header value"));
if opts.part_number_marker > 0 {
headers.insert(
X_AMZ_PART_NUMBER_MARKER,
HeaderValue::from_str(&opts.part_number_marker.to_string()).unwrap(),
HeaderValue::from_str(&opts.part_number_marker.to_string()).expect("valid header value"),
);
}
if opts.max_parts > 0 {
headers.insert(X_AMZ_MAX_PARTS, HeaderValue::from_str(&opts.max_parts.to_string()).unwrap());
headers.insert(X_AMZ_MAX_PARTS, HeaderValue::from_str(&opts.max_parts.to_string()).expect("valid header value"));
} else {
headers.insert(
X_AMZ_MAX_PARTS,
HeaderValue::from_str(&GET_OBJECT_ATTRIBUTES_MAX_PARTS.to_string()).unwrap(),
HeaderValue::from_str(&GET_OBJECT_ATTRIBUTES_MAX_PARTS.to_string()).expect("valid header value"),
);
}
@@ -210,7 +222,9 @@ impl TransitionClient {
let resp_status = resp.status();
let h = resp.headers().clone();
let has_etag = h.get("ETag").unwrap().to_str().unwrap();
let has_etag = h.get("ETag")
.and_then(|v| v.to_str().ok())
.unwrap_or("");
if !has_etag.is_empty() {
return Err(std::io::Error::other(
"get_object_attributes is not supported by the current endpoint version",
@@ -227,7 +241,8 @@ impl TransitionClient {
}
if resp_status != http::StatusCode::OK {
let err_body = String::from_utf8(body_vec).unwrap();
let err_body = String::from_utf8(body_vec)
.map_err(|e| std::io::Error::other(format!("invalid UTF-8 error body: {e}")))?;
let mut er = match quick_xml::de::from_str::<AccessControlPolicy>(&err_body) {
Ok(result) => result,
Err(err) => {
crates/ecstore/src/diagnostics/admin_server_info.rs
+6 -6
View File
@@ -78,8 +78,8 @@ async fn is_server_resolvable(endpoint: &Endpoint) -> Result<()> {
let addr = format!(
"{}://{}:{}",
endpoint.url.scheme(),
endpoint.url.host_str().unwrap(),
endpoint.url.port().unwrap()
endpoint.url.host_str().expect("URL should have host"),
endpoint.url.port().expect("URL should have port")
);
let ping_task = async {
@@ -309,10 +309,10 @@ fn get_online_offline_disks_stats(disks_info: &[Disk]) -> (BackendDisks, Backend
let ep = &disk.endpoint;
let state = &disk.state;
if *state != DriveState::Ok.to_string() && *state != DriveState::Unformatted.to_string() {
*offline_disks.get_mut(ep).unwrap() += 1;
*offline_disks.get_mut(ep).expect("endpoint should be in disk map") += 1;
continue;
}
*online_disks.get_mut(ep).unwrap() += 1;
*online_disks.get_mut(ep).expect("endpoint should be in disk map") += 1;
}
let mut root_disk_count = 0;
@@ -329,8 +329,8 @@ fn get_online_offline_disks_stats(disks_info: &[Disk]) -> (BackendDisks, Backend
for disk in disks_info {
let ep = &disk.endpoint;
if disk.root_disk {
*offline_disks.get_mut(ep).unwrap() += 1;
*online_disks.get_mut(ep).unwrap() -= 1;
*offline_disks.get_mut(ep).expect("endpoint should be in disk map") += 1;
*online_disks.get_mut(ep).expect("endpoint should be in disk map") -= 1;
}
}
crates/ecstore/src/diagnostics/get.rs
+13
View File
@@ -15,6 +15,7 @@
use crate::disk::error::DiskError;
use crate::error::StorageError;
use std::io;
use std::time::Instant;
pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING: &str = "codec_streaming";
pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE: &str = "codec_streaming_legacy_engine";
@@ -174,6 +175,18 @@ pub(crate) fn record_get_object_pipeline_failure_for_path(
rustfs_io_metrics::record_get_object_pipeline_failure_for_path(path, stage, reason.as_str());
}
#[inline(always)]
pub(crate) fn get_stage_timer_if_enabled(stage_metrics_enabled: bool) -> Option<Instant> {
stage_metrics_enabled.then(Instant::now)
}
#[inline(always)]
pub(crate) fn record_get_stage_duration_if_enabled(path: &'static str, stage: &'static str, started_at: Option<Instant>) {
if let Some(started_at) = started_at {
rustfs_io_metrics::record_get_object_stage_duration(path, stage, started_at.elapsed().as_secs_f64());
}
}
#[cfg(test)]
mod tests {
use super::*;
crates/ecstore/src/erasure/codec/bridge.rs
+137 -23
View File
@@ -16,9 +16,14 @@ use crate::erasure::codec::workspace::RustfsCodecDecodeWorkspace;
use crate::erasure::coding::Erasure;
use reed_solomon_erasure::galois_8::ReedSolomon;
use std::io;
use std::sync::{Arc, OnceLock};
pub(crate) const GET_CODEC_STREAMING_ENGINE_LEGACY: &str = "legacy";
pub(crate) const GET_CODEC_STREAMING_ENGINE_RUSTFS: &str = "rustfs";
pub(crate) const GET_RECONSTRUCT_OUTCOME_LEGACY_CALLED: &str = "legacy_called";
pub(crate) const GET_RECONSTRUCT_OUTCOME_RUSTFS_CALLED: &str = "rustfs_called";
pub(crate) const GET_RECONSTRUCT_OUTCOME_SKIP_DATA_COMPLETE: &str = "skip_data_complete";
pub(crate) const GET_RECONSTRUCT_OUTCOME_SKIP_EMPTY_PAYLOAD: &str = "skip_empty_payload";
pub(crate) trait DecodeWorkspace: Send + Sync + 'static {
fn shard_len(&self) -> usize;
@@ -30,13 +35,14 @@ pub(crate) trait ErasureDecodeEngine: Send + Sync + 'static {
fn data_shards(&self) -> usize;
fn parity_shards(&self) -> usize;
fn block_size(&self) -> usize;
fn engine_name(&self) -> &'static str;
fn supports_progressive_decode(&self) -> bool;
fn supports_aligned_shards(&self) -> bool;
fn prepare_workspace(&self, shard_len: usize) -> io::Result<Self::Workspace>;
fn reconstruct_into(&self, shards: &mut [Option<Vec<u8>>], workspace: &mut Self::Workspace) -> io::Result<()>;
fn reconstruct_into(&self, shards: &mut [Option<Vec<u8>>], workspace: &mut Self::Workspace) -> io::Result<&'static str>;
}
fn data_shards_complete(shards: &[Option<Vec<u8>>], data_shards: usize) -> bool {
@@ -126,6 +132,10 @@ impl ErasureDecodeEngine for LegacyEcDecodeEngine {
self.erasure.block_size
}
fn engine_name(&self) -> &'static str {
GET_CODEC_STREAMING_ENGINE_LEGACY
}
fn supports_progressive_decode(&self) -> bool {
false
}
@@ -138,12 +148,13 @@ impl ErasureDecodeEngine for LegacyEcDecodeEngine {
Ok(LegacyDecodeWorkspace::new(shard_len))
}
fn reconstruct_into(&self, shards: &mut [Option<Vec<u8>>], _workspace: &mut Self::Workspace) -> io::Result<()> {
fn reconstruct_into(&self, shards: &mut [Option<Vec<u8>>], _workspace: &mut Self::Workspace) -> io::Result<&'static str> {
if data_shards_complete(shards, self.erasure.data_shards) {
return Ok(());
return Ok(GET_RECONSTRUCT_OUTCOME_SKIP_DATA_COMPLETE);
}
self.erasure.decode_data_with_reconstruction_verification(shards)
self.erasure.decode_data_with_reconstruction_verification(shards)?;
Ok(GET_RECONSTRUCT_OUTCOME_LEGACY_CALLED)
}
}
@@ -152,26 +163,74 @@ pub(crate) struct RustfsCodecDecodeEngine {
data_shards: usize,
parity_shards: usize,
block_size: usize,
codec: Option<ReedSolomon>,
codec: OnceLock<Arc<ReedSolomon>>,
}
impl RustfsCodecDecodeEngine {
pub(crate) fn new(erasure: &Erasure) -> io::Result<Self> {
let codec = if erasure.parity_shards > 0 {
ReedSolomon::new(erasure.data_shards, erasure.parity_shards)
.map_err(|err| io::Error::other(format!("Failed to create RustFS codec decode engine: {err:?}")))
.map(Some)?
} else {
None
};
Ok(Self {
data_shards: erasure.data_shards,
parity_shards: erasure.parity_shards,
block_size: erasure.block_size,
codec,
codec: OnceLock::new(),
})
}
fn codec(&self) -> io::Result<Option<Arc<ReedSolomon>>> {
if self.parity_shards == 0 {
return Ok(None);
}
if let Some(codec) = self.codec.get() {
return Ok(Some(Arc::clone(codec)));
}
let codec = Arc::new(
ReedSolomon::new(self.data_shards, self.parity_shards)
.map_err(|err| io::Error::other(format!("Failed to create RustFS codec decode engine: {err:?}")))?,
);
if self.codec.set(Arc::clone(&codec)).is_err() {
return Ok(self.codec.get().map(Arc::clone).or(Some(codec)));
}
Ok(Some(codec))
}
fn needs_source_parity_verification(&self, shards: &[Option<Vec<u8>>]) -> bool {
let missing_data_source = shards.iter().take(self.data_shards).any(|shard| shard.is_none());
let available_shards = shards.iter().filter(|shard| shard.is_some()).count();
missing_data_source && available_shards > self.data_shards
}
fn verify_source_parity(&self, codec: &ReedSolomon, shards: &[Option<Vec<u8>>], needs_verification: bool) -> io::Result<()> {
if !needs_verification {
return Ok(());
}
let mut shard_refs = Vec::with_capacity(self.data_shards + self.parity_shards);
for (index, shard) in shards.iter().enumerate() {
let shard = shard.as_ref().ok_or_else(|| {
io::Error::new(
io::ErrorKind::InvalidData,
format!("missing shard {index} after RustFS codec reconstruction"),
)
})?;
shard_refs.push(shard.as_slice());
}
let valid = codec
.verify(&shard_refs)
.map_err(|err| io::Error::other(format!("RustFS codec verify failed: {err:?}")))?;
if !valid {
return Err(io::Error::new(io::ErrorKind::InvalidData, "inconsistent read source shards"));
}
Ok(())
}
#[cfg(test)]
fn codec_is_initialized(&self) -> bool {
self.codec.get().is_some()
}
}
impl ErasureDecodeEngine for RustfsCodecDecodeEngine {
@@ -189,6 +248,10 @@ impl ErasureDecodeEngine for RustfsCodecDecodeEngine {
self.block_size
}
fn engine_name(&self) -> &'static str {
GET_CODEC_STREAMING_ENGINE_RUSTFS
}
fn supports_progressive_decode(&self) -> bool {
false
}
@@ -201,21 +264,23 @@ impl ErasureDecodeEngine for RustfsCodecDecodeEngine {
Ok(RustfsCodecDecodeWorkspace::new(shard_len))
}
fn reconstruct_into(&self, shards: &mut [Option<Vec<u8>>], _workspace: &mut Self::Workspace) -> io::Result<()> {
fn reconstruct_into(&self, shards: &mut [Option<Vec<u8>>], _workspace: &mut Self::Workspace) -> io::Result<&'static str> {
if data_shards_complete(shards, self.data_shards) {
return Ok(());
return Ok(GET_RECONSTRUCT_OUTCOME_SKIP_DATA_COMPLETE);
}
if recover_empty_payload_data_shards(shards, self.data_shards, self.parity_shards)? {
return Ok(());
return Ok(GET_RECONSTRUCT_OUTCOME_SKIP_EMPTY_PAYLOAD);
}
if let Some(codec) = &self.codec {
if let Some(codec) = self.codec()? {
let needs_source_parity_verification = self.needs_source_parity_verification(shards);
codec
.reconstruct_data_opt(shards)
.map_err(|err| io::Error::other(format!("RustFS codec reconstruct failed: {err:?}")))
} else {
Ok(())
.map_err(|err| io::Error::other(format!("RustFS codec reconstruct failed: {err:?}")))?;
self.verify_source_parity(&codec, shards, needs_source_parity_verification)?;
}
Ok(GET_RECONSTRUCT_OUTCOME_RUSTFS_CALLED)
}
}
@@ -273,6 +338,13 @@ impl ErasureDecodeEngine for CodecStreamingDecodeEngine {
}
}
fn engine_name(&self) -> &'static str {
match self {
Self::Legacy(engine) => engine.engine_name(),
Self::Rustfs(engine) => engine.engine_name(),
}
}
fn supports_progressive_decode(&self) -> bool {
match self {
Self::Legacy(engine) => engine.supports_progressive_decode(),
@@ -294,7 +366,7 @@ impl ErasureDecodeEngine for CodecStreamingDecodeEngine {
}
}
fn reconstruct_into(&self, shards: &mut [Option<Vec<u8>>], workspace: &mut Self::Workspace) -> io::Result<()> {
fn reconstruct_into(&self, shards: &mut [Option<Vec<u8>>], workspace: &mut Self::Workspace) -> io::Result<&'static str> {
match (self, workspace) {
(Self::Legacy(engine), CodecStreamingDecodeWorkspace::Legacy(workspace)) => {
engine.reconstruct_into(shards, workspace)
@@ -325,7 +397,7 @@ mod tests {
E: ErasureDecodeEngine,
{
let mut workspace = engine.prepare_workspace(4)?;
engine.reconstruct_into(shards, &mut workspace)
engine.reconstruct_into(shards, &mut workspace).map(|_| ())
}
#[test]
@@ -393,8 +465,10 @@ mod tests {
let before = shards.clone();
let engine = RustfsCodecDecodeEngine::new(&erasure).expect("engine should be created");
assert!(!engine.codec_is_initialized());
reconstruct_with(&engine, &mut shards).expect("complete data shards should not reconstruct");
assert!(!engine.codec_is_initialized());
assert_eq!(shards, before);
}
@@ -446,6 +520,46 @@ mod tests {
assert!(reconstruct_with(&rustfs, &mut rustfs_shards).is_err());
}
#[test]
fn rustfs_codec_decode_engine_rejects_inconsistent_reconstruction_sources() {
let erasure = Erasure::new(2, 2, 32);
let encoded = erasure
.encode_data(&(0u8..64u8).collect::<Vec<_>>())
.expect("test stripe should encode");
let mut shards = encoded.into_iter().map(|shard| Some(shard.to_vec())).collect::<Vec<_>>();
shards[0] = None;
let Some(corrupt_parity) = shards[erasure.data_shards].as_mut() else {
panic!("test parity shard should be present");
};
corrupt_parity[0] ^= 0x80;
let engine = RustfsCodecDecodeEngine::new(&erasure).expect("engine should be created");
let err = reconstruct_with(&engine, &mut shards).expect_err("rustfs codec should reject inconsistent sources");
assert_eq!(err.kind(), io::ErrorKind::InvalidData);
assert!(err.to_string().contains("inconsistent read source shards"));
}
#[test]
fn rustfs_codec_decode_engine_rejects_stale_data_source() {
let erasure = Erasure::new(4, 2, 32);
let encoded = erasure
.encode_data(&(0u8..128u8).collect::<Vec<_>>())
.expect("test stripe should encode");
let mut shards = encoded.into_iter().map(|shard| Some(shard.to_vec())).collect::<Vec<_>>();
shards[0] = None;
let Some(stale_data) = shards[1].as_mut() else {
panic!("test data shard should be present");
};
stale_data[0] ^= 0x40;
let engine = RustfsCodecDecodeEngine::new(&erasure).expect("engine should be created");
let err = reconstruct_with(&engine, &mut shards).expect_err("rustfs codec should reject stale data sources");
assert_eq!(err.kind(), io::ErrorKind::InvalidData);
assert!(err.to_string().contains("inconsistent read source shards"));
}
#[test]
fn rustfs_codec_decode_engine_recovers_empty_data_shard() {
let erasure = Erasure::new(4, 2, 16);
crates/ecstore/src/erasure/coding/decode.rs
+47 -46
View File
@@ -16,7 +16,8 @@ use crate::diagnostics::get::{
GET_OBJECT_PATH_LEGACY_DUPLEX, GET_SHARD_READ_ERROR_MISSING, GET_SHARD_READ_ERROR_NONE, GET_SHARD_READ_OUTCOME_ERROR,
GET_SHARD_READ_OUTCOME_MISSING, GET_SHARD_READ_OUTCOME_SUCCESS, GET_SHARD_ROLE_DATA, GET_SHARD_ROLE_PARITY, GET_STAGE_EMIT,
GET_STAGE_RANGE, GET_STAGE_RECONSTRUCT, GET_STAGE_STRIPE_READ, GET_STAGE_STRIPE_READ_FIRST_SHARD,
GET_STAGE_STRIPE_READ_QUORUM, GetObjectFailureReason, classify_io_error, record_get_object_pipeline_failure,
GET_STAGE_STRIPE_READ_QUORUM, GetObjectFailureReason, classify_io_error, get_stage_timer_if_enabled,
record_get_object_pipeline_failure, record_get_stage_duration_if_enabled,
};
use crate::disk::disk_store::get_object_disk_read_timeout;
use crate::disk::error::Error;
@@ -182,7 +183,7 @@ where
Box::pin(async move {
let mut buf = recycled_buf.unwrap_or_else(|| vec![0; shard_size]);
debug_assert_eq!(buf.len(), shard_size);
let read_start = Instant::now();
let read_start = metrics_path.map(|_| Instant::now());
let read_result = if read_timeout.is_zero() {
reader.read(&mut buf).await
} else {
@@ -200,7 +201,7 @@ where
GET_SHARD_READ_OUTCOME_ERROR,
error_class,
0,
read_start.elapsed().as_secs_f64(),
read_start.map_or(0.0, |read_start| read_start.elapsed().as_secs_f64()),
reader.last_verify_duration().as_secs_f64(),
);
}
@@ -221,7 +222,7 @@ where
GET_SHARD_READ_OUTCOME_SUCCESS,
GET_SHARD_READ_ERROR_NONE,
n,
read_start.elapsed().as_secs_f64(),
read_start.map_or(0.0, |read_start| read_start.elapsed().as_secs_f64()),
reader.last_verify_duration().as_secs_f64(),
);
}
@@ -240,7 +241,7 @@ where
GET_SHARD_READ_OUTCOME_ERROR,
error_class,
0,
read_start.elapsed().as_secs_f64(),
read_start.map_or(0.0, |read_start| read_start.elapsed().as_secs_f64()),
verify_duration_secs,
);
}
@@ -450,6 +451,7 @@ where
read_timeout: Duration,
verify_reconstruction: bool,
) -> Self {
let metrics_path = metrics_path.filter(|_| rustfs_io_metrics::get_stage_metrics_enabled());
let shard_size = e.shard_size();
let shard_file_size = e.shard_file_size(total_length as i64) as usize;
@@ -615,7 +617,7 @@ where
let mut reader_iter = ReaderLaunchIter::new(&mut self.readers, read_costs, locality_preference_enabled);
let mut sets = FuturesUnordered::new();
let mut active_readers = vec![false; num_readers];
let stripe_read_start = Instant::now();
let stripe_read_start = self.metrics_path.map(|_| Instant::now());
let mut scheduled = 0usize;
for _ in 0..self.data_shards {
if let Some((i, reader)) = reader_iter.next() {
@@ -722,11 +724,7 @@ where
completed += 1;
if !first_shard_recorded {
if let Some(path) = self.metrics_path {
rustfs_io_metrics::record_get_object_stage_duration(
path,
GET_STAGE_STRIPE_READ_FIRST_SHARD,
stripe_read_start.elapsed().as_secs_f64(),
);
record_get_stage_duration_if_enabled(path, GET_STAGE_STRIPE_READ_FIRST_SHARD, stripe_read_start);
}
first_shard_recorded = true;
}
@@ -876,11 +874,7 @@ where
}
if let Some(path) = self.metrics_path {
rustfs_io_metrics::record_get_object_stage_duration(
path,
GET_STAGE_STRIPE_READ_QUORUM,
stripe_read_start.elapsed().as_secs_f64(),
);
record_get_stage_duration_if_enabled(path, GET_STAGE_STRIPE_READ_QUORUM, stripe_read_start);
rustfs_io_metrics::record_get_object_shard_read_fanout(path, scheduled, completed, success, failed);
if locality_preference_enabled {
let remote_avoided = remote_available.saturating_sub(remote_scheduled);
@@ -1038,9 +1032,11 @@ where
offset = 0;
let write_len = write_left.min(block_slice.len());
let write_stage_start = Instant::now();
let write_stage_start = get_stage_timer_if_enabled(rustfs_io_metrics::get_stage_metrics_enabled());
if let Err(e) = writer.write_all(&block_slice[..write_len]).await {
rustfs_io_metrics::record_get_object_duplex_backpressure_duration(write_stage_start.elapsed().as_secs_f64());
if let Some(write_stage_start) = write_stage_start {
rustfs_io_metrics::record_get_object_duplex_backpressure_duration(write_stage_start.elapsed().as_secs_f64());
}
let reason = classify_io_error(&e);
record_get_object_pipeline_failure(GET_STAGE_EMIT, reason);
error!(
@@ -1054,7 +1050,9 @@ where
);
return Err(e);
}
rustfs_io_metrics::record_get_object_duplex_backpressure_duration(write_stage_start.elapsed().as_secs_f64());
if let Some(write_stage_start) = write_stage_start {
rustfs_io_metrics::record_get_object_duplex_backpressure_duration(write_stage_start.elapsed().as_secs_f64());
}
total_written += write_len;
write_left -= write_len;
@@ -1180,13 +1178,10 @@ impl Erasure {
break;
}
let stripe_read_stage_start = Instant::now();
let stage_metrics_enabled = rustfs_io_metrics::get_stage_metrics_enabled();
let stripe_read_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let (mut shards, errs) = reader.read().await;
rustfs_io_metrics::record_get_object_stage_duration(
"legacy_duplex",
"stripe_read",
stripe_read_stage_start.elapsed().as_secs_f64(),
);
record_get_stage_duration_if_enabled(GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_STRIPE_READ, stripe_read_stage_start);
if ret_err.is_none()
&& let (_, Some(err)) = reduce_errs(&errs, &[])
@@ -1216,12 +1211,12 @@ impl Erasure {
// Decode the shards. If this stripe needed parity to reconstruct a
// missing data shard and an extra source shard was available, verify
// the reconstructed data against that source before streaming bytes.
let reconstruct_stage_start = Instant::now();
let reconstruct_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
if let Err(e) = self.decode_data_with_reconstruction_verification(&mut shards) {
rustfs_io_metrics::record_get_object_stage_duration(
"legacy_duplex",
"reconstruct",
reconstruct_stage_start.elapsed().as_secs_f64(),
record_get_stage_duration_if_enabled(
GET_OBJECT_PATH_LEGACY_DUPLEX,
GET_STAGE_RECONSTRUCT,
reconstruct_stage_start,
);
let reason = GetObjectFailureReason::DecodeError;
error!(
@@ -1238,28 +1233,16 @@ impl Erasure {
ret_err = Some(e);
break;
}
rustfs_io_metrics::record_get_object_stage_duration(
"legacy_duplex",
"reconstruct",
reconstruct_stage_start.elapsed().as_secs_f64(),
);
record_get_stage_duration_if_enabled(GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_RECONSTRUCT, reconstruct_stage_start);
let emit_stage_start = Instant::now();
let emit_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let n = match write_data_blocks(writer, &shards, self.data_shards, block_offset, block_length).await {
Ok(n) => {
rustfs_io_metrics::record_get_object_stage_duration(
"legacy_duplex",
"emit",
emit_stage_start.elapsed().as_secs_f64(),
);
record_get_stage_duration_if_enabled(GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_EMIT, emit_stage_start);
n
}
Err(e) => {
rustfs_io_metrics::record_get_object_stage_duration(
"legacy_duplex",
"emit",
emit_stage_start.elapsed().as_secs_f64(),
);
record_get_stage_duration_if_enabled(GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_EMIT, emit_stage_start);
error!(
block_offset,
block_length,
@@ -1624,6 +1607,24 @@ mod tests {
assert_eq!(shard_read_launch_order(&read_costs, read_costs.len(), true), vec![1, 3, 2, 0, 4]);
}
#[test]
#[serial_test::serial]
fn parallel_reader_drops_metrics_path_when_stage_metrics_disabled() {
let erasure = Erasure::new(2, 1, 32);
let readers: Vec<Option<BitrotReader<Cursor<Vec<u8>>>>> = vec![None, None, None];
rustfs_io_metrics::set_get_stage_metrics_enabled(false);
let reader = ParallelReader::new_with_metrics_path(readers, erasure.clone(), 0, 1, Some(GET_OBJECT_PATH_LEGACY_DUPLEX));
assert_eq!(reader.metrics_path, None);
let readers: Vec<Option<BitrotReader<Cursor<Vec<u8>>>>> = vec![None, None, None];
rustfs_io_metrics::set_get_stage_metrics_enabled(true);
let reader = ParallelReader::new_with_metrics_path(readers, erasure, 0, 1, Some(GET_OBJECT_PATH_LEGACY_DUPLEX));
assert_eq!(reader.metrics_path, Some(GET_OBJECT_PATH_LEGACY_DUPLEX));
rustfs_io_metrics::set_get_stage_metrics_enabled(false);
}
#[tokio::test]
#[serial_test::serial]
async fn test_parallel_reader_local_first_avoids_remote_when_local_quorum_exists() {
crates/ecstore/src/erasure/coding/decode_reader.rs
+512 -164
View File
@@ -17,10 +17,10 @@ use crate::diagnostics::get::{
GET_READER_POLL_READY_DATA, GET_READER_POLL_READY_EMPTY, GET_READER_POLL_READY_ERROR, GET_READER_PREFETCH_DIRECT,
GET_READER_PREFETCH_EOF, GET_READER_PREFETCH_ERROR_DEFERRED, GET_READER_PREFETCH_ERROR_IMMEDIATE, GET_READER_PREFETCH_STORED,
GET_STAGE_DECODE, GET_STAGE_EMIT, GET_STAGE_FILL, GET_STAGE_OUTPUT_LOCK_WAIT, GET_STAGE_OUTPUT_POLL, GET_STAGE_RECONSTRUCT,
GET_STAGE_STRIPE_READ,
GET_STAGE_STRIPE_READ, get_stage_timer_if_enabled, record_get_stage_duration_if_enabled,
};
use crate::disk::error::Error as DiskError;
use crate::erasure::codec::bridge::ErasureDecodeEngine;
use crate::erasure::codec::bridge::{ErasureDecodeEngine, GET_RECONSTRUCT_OUTCOME_SKIP_DATA_COMPLETE};
use crate::set_disk::shard_source::{ShardStripeSource, StripeReadState};
use std::collections::VecDeque;
use std::io;
@@ -30,6 +30,7 @@ use std::sync::Mutex;
use std::task::{Context, Poll, ready};
use std::time::Instant;
use tokio::io::{AsyncRead, ReadBuf};
use tokio::sync::{mpsc, oneshot};
use tokio::task::JoinHandle;
const ENV_RUSTFS_GET_CODEC_STREAMING_MAX_INFLIGHT: &str = "RUSTFS_GET_CODEC_STREAMING_MAX_INFLIGHT";
@@ -37,13 +38,23 @@ const DEFAULT_RUSTFS_GET_CODEC_STREAMING_MAX_INFLIGHT: usize = 2;
const FILL_POLICY_SINGLE_INFLIGHT: &str = "single_inflight";
const FILL_POLICY_DUAL_INFLIGHT: &str = "dual_inflight";
type FillTask<S, W> = JoinHandle<FillResult<S, W>>;
type FillTask = oneshot::Receiver<FillResult>;
struct FillResult<S, W> {
source: S,
workspace: W,
struct FillWorker {
tx: mpsc::Sender<FillRequest>,
task: JoinHandle<()>,
}
struct FillRequest {
remaining: usize,
reusable_buffers: Vec<Vec<u8>>,
response: oneshot::Sender<FillResult>,
}
struct FillResult {
result: io::Result<Option<Vec<u8>>>,
queued_buffers: VecDeque<Vec<u8>>,
reusable_buffers: Vec<Vec<u8>>,
deferred_error: Option<io::Error>,
}
@@ -88,19 +99,22 @@ where
E: ErasureDecodeEngine,
{
metrics_path: &'static str,
stage_metrics_enabled: bool,
fill_policy: FillPolicy,
source: Option<S>,
engine: E,
engine: Option<E>,
workspace: Option<E::Workspace>,
worker: Option<FillWorker>,
output_buf: Vec<u8>,
output_pos: usize,
reusable_output_bufs: Vec<Vec<u8>>,
prefetched_bufs: VecDeque<Vec<u8>>,
prefetch_error: Option<io::Error>,
prefetch_wait_started_at: Option<Instant>,
output_wait_started_at: Option<Instant>,
remaining: usize,
// Bounded lookahead controlled by `FillPolicy`.
fill: Option<FillTask<S, E::Workspace>>,
fill: Option<FillTask>,
}
impl<S, E> ErasureDecodeReader<S, E>
@@ -140,12 +154,15 @@ where
Ok(Self {
metrics_path,
stage_metrics_enabled: rustfs_io_metrics::get_stage_metrics_enabled(),
fill_policy,
source: Some(source),
engine,
engine: Some(engine),
workspace: Some(workspace),
worker: None,
output_buf: Vec::new(),
output_pos: 0,
reusable_output_bufs: Vec::new(),
prefetched_bufs: VecDeque::new(),
prefetch_error: None,
prefetch_wait_started_at: None,
@@ -155,6 +172,69 @@ where
})
}
fn max_reusable_output_bufs(&self) -> usize {
self.fill_policy.max_inflight() + 1
}
fn push_reusable_output_buf(&mut self, mut buf: Vec<u8>) {
if buf.capacity() == 0 || self.reusable_output_bufs.len() >= self.max_reusable_output_bufs() {
return;
}
buf.clear();
self.reusable_output_bufs.push(buf);
}
fn recycle_drained_output_buf(&mut self) {
if self.output_pos < self.output_buf.len() {
return;
}
let buf = std::mem::take(&mut self.output_buf);
self.output_pos = 0;
self.push_reusable_output_buf(buf);
}
fn extend_reusable_output_bufs(&mut self, bufs: Vec<Vec<u8>>) {
for buf in bufs {
self.push_reusable_output_buf(buf);
}
}
fn fill_worker_tx(&mut self) -> io::Result<mpsc::Sender<FillRequest>> {
if self.worker.is_none() {
let Some(source) = self.source.take() else {
return Err(io::Error::new(ErrorKind::BrokenPipe, "erasure reader source missing"));
};
let Some(engine) = self.engine.take() else {
self.source = Some(source);
return Err(io::Error::new(ErrorKind::BrokenPipe, "erasure reader engine missing"));
};
let Some(workspace) = self.workspace.take() else {
self.source = Some(source);
self.engine = Some(engine);
return Err(io::Error::new(ErrorKind::BrokenPipe, "erasure reader workspace missing"));
};
let (tx, rx) = mpsc::channel(1);
rustfs_io_metrics::record_get_object_fill_worker_started(self.metrics_path, self.fill_policy.as_str());
let task = tokio::spawn(run_fill_worker(
source,
engine,
workspace,
self.fill_policy,
self.metrics_path,
self.stage_metrics_enabled,
rx,
));
self.worker = Some(FillWorker { tx, task });
}
self.worker
.as_ref()
.map(|worker| worker.tx.clone())
.ok_or_else(|| io::Error::new(ErrorKind::BrokenPipe, "erasure reader fill worker missing"))
}
#[cfg(test)]
fn new_with_fill_policy(
source: S,
@@ -168,87 +248,31 @@ where
fn poll_fill_result(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<Option<Vec<u8>>>> {
if self.fill.is_none() {
let Some(mut source) = self.source.take() else {
return Poll::Ready(Err(io::Error::new(ErrorKind::BrokenPipe, "erasure reader source missing")));
let fill_worker_tx = match self.fill_worker_tx() {
Ok(tx) => tx,
Err(err) => return Poll::Ready(Err(err)),
};
let Some(mut workspace) = self.workspace.take() else {
self.source = Some(source);
return Poll::Ready(Err(io::Error::new(ErrorKind::BrokenPipe, "erasure reader workspace missing")));
};
let engine = self.engine.clone();
let metrics_path = self.metrics_path;
let fill_policy = self.fill_policy;
let remaining = self.remaining;
self.fill = Some(tokio::spawn(async move {
let mut queued_buffers = VecDeque::new();
let mut deferred_error = None;
let fill_stage_start = Instant::now();
let stripe_read_stage_start = Instant::now();
let state = source.read_next_stripe().await;
rustfs_io_metrics::record_get_object_stage_duration(
metrics_path,
GET_STAGE_STRIPE_READ,
stripe_read_stage_start.elapsed().as_secs_f64(),
);
let decode_stage_start = Instant::now();
let result = decode_stripe(metrics_path, &engine, &mut workspace, state, remaining);
rustfs_io_metrics::record_get_object_stage_duration(
metrics_path,
GET_STAGE_DECODE,
decode_stage_start.elapsed().as_secs_f64(),
);
if let Ok(Some(first_buf)) = result.as_ref() {
let mut remaining_after_first = remaining.saturating_sub(first_buf.len());
for _ in 0..fill_policy.additional_queued_buffers() {
if remaining_after_first == 0 {
break;
}
let stripe_read_stage_start = Instant::now();
let state = source.read_next_stripe().await;
rustfs_io_metrics::record_get_object_stage_duration(
metrics_path,
GET_STAGE_STRIPE_READ,
stripe_read_stage_start.elapsed().as_secs_f64(),
);
let decode_stage_start = Instant::now();
let queued_result = decode_stripe(metrics_path, &engine, &mut workspace, state, remaining_after_first);
rustfs_io_metrics::record_get_object_stage_duration(
metrics_path,
GET_STAGE_DECODE,
decode_stage_start.elapsed().as_secs_f64(),
);
match queued_result {
Ok(Some(buf)) => {
remaining_after_first = remaining_after_first.saturating_sub(buf.len());
queued_buffers.push_back(buf);
}
Ok(None) => {
if remaining_after_first > 0 {
deferred_error = Some(DiskError::LessData.into());
}
break;
}
Err(err) => {
deferred_error = Some(err);
break;
}
}
}
}
rustfs_io_metrics::record_get_object_stage_duration(
metrics_path,
GET_STAGE_FILL,
fill_stage_start.elapsed().as_secs_f64(),
);
FillResult {
source,
workspace,
result,
queued_buffers,
deferred_error,
}
}));
let reusable_buffers = std::mem::take(&mut self.reusable_output_bufs);
let (response, fill) = oneshot::channel();
let request = FillRequest {
remaining,
reusable_buffers,
response,
};
if let Err(err) = fill_worker_tx.try_send(request) {
self.extend_reusable_output_bufs(err.into_inner().reusable_buffers);
return Poll::Ready(Err(io::Error::new(
ErrorKind::BrokenPipe,
"erasure reader fill worker request queue is closed or full",
)));
}
rustfs_io_metrics::record_get_object_fill_started(metrics_path, fill_policy.as_str());
self.fill = Some(fill);
}
let fill = self
@@ -257,22 +281,20 @@ where
.ok_or_else(|| io::Error::new(ErrorKind::BrokenPipe, "erasure reader fill future missing"))?;
let fill_result = ready!(Pin::new(fill).poll(cx));
let FillResult {
source,
workspace,
result,
queued_buffers,
reusable_buffers,
deferred_error,
} = match fill_result {
Ok(result) => result,
Err(err) => {
self.fill = None;
return Poll::Ready(Err(io::Error::other(format!("erasure reader fill task failed: {err}"))));
return Poll::Ready(Err(io::Error::other(format!("erasure reader fill worker stopped: {err}"))));
}
};
self.source = Some(source);
self.workspace = Some(workspace);
self.fill = None;
self.extend_reusable_output_bufs(reusable_buffers);
if let Some(deferred_error) = deferred_error {
self.prefetch_error = Some(deferred_error);
}
@@ -304,13 +326,15 @@ where
return Poll::Ready(Ok(()));
}
if self.prefetch_wait_started_at.is_none() {
if self.stage_metrics_enabled && self.prefetch_wait_started_at.is_none() {
self.prefetch_wait_started_at = Some(Instant::now());
}
let fill = match self.poll_fill_result(cx) {
Poll::Ready(result) => {
if let Some(started_at) = self.prefetch_wait_started_at.take() {
if self.stage_metrics_enabled
&& let Some(started_at) = self.prefetch_wait_started_at.take()
{
rustfs_io_metrics::record_get_object_reader_prefetch_wait(
self.metrics_path,
started_at.elapsed().as_secs_f64(),
@@ -385,14 +409,146 @@ where
}
}
async fn run_fill_worker<S, E>(
mut source: S,
engine: E,
mut workspace: E::Workspace,
fill_policy: FillPolicy,
metrics_path: &'static str,
stage_metrics_enabled: bool,
mut rx: mpsc::Receiver<FillRequest>,
) where
S: ShardStripeSource + Send + 'static,
E: ErasureDecodeEngine + Send + Sync + 'static,
{
while let Some(request) = rx.recv().await {
let response = request.response;
let result = run_fill_request(FillRequestWork {
source: &mut source,
engine: &engine,
workspace: &mut workspace,
fill_policy,
metrics_path,
stage_metrics_enabled,
remaining: request.remaining,
reusable_buffers: request.reusable_buffers,
})
.await;
let _ = response.send(result);
}
}
struct FillRequestWork<'a, S, E>
where
E: ErasureDecodeEngine,
{
source: &'a mut S,
engine: &'a E,
workspace: &'a mut E::Workspace,
fill_policy: FillPolicy,
metrics_path: &'static str,
stage_metrics_enabled: bool,
remaining: usize,
reusable_buffers: Vec<Vec<u8>>,
}
async fn run_fill_request<S, E>(work: FillRequestWork<'_, S, E>) -> FillResult
where
S: ShardStripeSource + Send,
E: ErasureDecodeEngine,
{
let FillRequestWork {
source,
engine,
workspace,
fill_policy,
metrics_path,
stage_metrics_enabled,
remaining,
mut reusable_buffers,
} = work;
let mut queued_buffers = VecDeque::new();
let mut deferred_error = None;
let fill_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let stripe_read_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let state = source.read_next_stripe().await;
record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_STRIPE_READ, stripe_read_stage_start);
let decode_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let mut output_buf = reusable_buffers.pop().unwrap_or_default();
let result =
match decode_stripe_into(metrics_path, stage_metrics_enabled, engine, workspace, state, remaining, &mut output_buf) {
Ok(true) => Ok(Some(output_buf)),
Ok(false) => {
reusable_buffers.push(output_buf);
Ok(None)
}
Err(err) => {
reusable_buffers.push(output_buf);
Err(err)
}
};
record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_DECODE, decode_stage_start);
if let Ok(Some(first_buf)) = result.as_ref() {
let mut remaining_after_first = remaining.saturating_sub(first_buf.len());
for _ in 0..fill_policy.additional_queued_buffers() {
if remaining_after_first == 0 {
break;
}
let stripe_read_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let state = source.read_next_stripe().await;
record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_STRIPE_READ, stripe_read_stage_start);
let decode_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let mut queued_buf = reusable_buffers.pop().unwrap_or_default();
let queued_result = decode_stripe_into(
metrics_path,
stage_metrics_enabled,
engine,
workspace,
state,
remaining_after_first,
&mut queued_buf,
);
record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_DECODE, decode_stage_start);
match queued_result {
Ok(true) => {
remaining_after_first = remaining_after_first.saturating_sub(queued_buf.len());
queued_buffers.push_back(queued_buf);
}
Ok(false) => {
reusable_buffers.push(queued_buf);
if remaining_after_first > 0 {
deferred_error = Some(DiskError::LessData.into());
}
break;
}
Err(err) => {
reusable_buffers.push(queued_buf);
deferred_error = Some(err);
break;
}
}
}
}
record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_FILL, fill_stage_start);
FillResult {
result,
queued_buffers,
reusable_buffers,
deferred_error,
}
}
impl<S, E> Drop for ErasureDecodeReader<S, E>
where
E: ErasureDecodeEngine,
{
fn drop(&mut self) {
if let Some(fill) = self.fill.take() {
if self.fill.take().is_some() {
rustfs_io_metrics::record_get_object_fill_cancelled_on_drop(self.metrics_path, self.fill_policy.as_str());
fill.abort();
}
if let Some(worker) = self.worker.take() {
worker.task.abort();
}
}
}
@@ -421,10 +577,12 @@ where
let read_buf_remaining_before = buf.remaining();
let output_remaining_before = available.len();
let copy_len = available.len().min(buf.remaining());
let copy_start = Instant::now();
let copy_start = get_stage_timer_if_enabled(self.stage_metrics_enabled);
buf.put_slice(&available[..copy_len]);
self.output_pos += copy_len;
if copy_len > 0 {
if copy_len > 0
&& let Some(copy_start) = copy_start
{
rustfs_io_metrics::record_get_object_reader_copy(
self.metrics_path,
copy_len,
@@ -439,6 +597,8 @@ where
continue;
}
self.recycle_drained_output_buf();
if let Some(next_buf) = self.prefetched_bufs.pop_front() {
rustfs_io_metrics::record_get_object_reader_buffer(self.metrics_path, GET_READER_BUFFER_OUTPUT, next_buf.len());
self.output_buf = next_buf;
@@ -458,14 +618,16 @@ where
}
if self.output_wait_started_at.is_none() {
self.output_wait_started_at = Some(Instant::now());
self.output_wait_started_at = get_stage_timer_if_enabled(self.stage_metrics_enabled);
}
let prefetch = match self.poll_prefetch(cx) {
Poll::Ready(result) => result,
Poll::Pending if buf.filled().len() > filled_before_poll => return Poll::Ready(Ok(())),
Poll::Pending => return Poll::Pending,
};
if let Some(started_at) = self.output_wait_started_at.take() {
if self.stage_metrics_enabled
&& let Some(started_at) = self.output_wait_started_at.take()
{
rustfs_io_metrics::record_get_object_fill_waited_by_output(
self.metrics_path,
self.fill_policy.as_str(),
@@ -480,6 +642,7 @@ where
pub(crate) struct SyncErasureDecodeReader<R> {
inner: Mutex<R>,
metrics_path: &'static str,
stage_metrics_enabled: bool,
}
impl<R> SyncErasureDecodeReader<R> {
@@ -491,6 +654,7 @@ impl<R> SyncErasureDecodeReader<R> {
Self {
inner: Mutex::new(inner),
metrics_path,
stage_metrics_enabled: rustfs_io_metrics::get_stage_metrics_enabled(),
}
}
}
@@ -500,97 +664,90 @@ where
R: AsyncRead + Unpin + Send,
{
fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
let lock_wait_start = Instant::now();
let stage_metrics_enabled = self.stage_metrics_enabled;
let lock_wait_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let mut inner = match self.inner.lock() {
Ok(inner) => {
rustfs_io_metrics::record_get_object_stage_duration(
self.metrics_path,
GET_STAGE_OUTPUT_LOCK_WAIT,
lock_wait_start.elapsed().as_secs_f64(),
);
record_get_stage_duration_if_enabled(self.metrics_path, GET_STAGE_OUTPUT_LOCK_WAIT, lock_wait_start);
inner
}
Err(_) => {
rustfs_io_metrics::record_get_object_stage_duration(
self.metrics_path,
GET_STAGE_OUTPUT_LOCK_WAIT,
lock_wait_start.elapsed().as_secs_f64(),
);
record_get_stage_duration_if_enabled(self.metrics_path, GET_STAGE_OUTPUT_LOCK_WAIT, lock_wait_start);
return Poll::Ready(Err(io::Error::other("erasure decode reader lock poisoned")));
}
};
let read_buf_remaining_before = buf.remaining();
let filled_before = buf.filled().len();
let poll_start = Instant::now();
let read_buf_remaining_before = stage_metrics_enabled.then(|| buf.remaining());
let filled_before = stage_metrics_enabled.then(|| buf.filled().len());
let poll_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let result = Pin::new(&mut *inner).poll_read(cx, buf);
let poll_duration = poll_start.elapsed().as_secs_f64();
let filled_bytes = buf.filled().len().saturating_sub(filled_before);
let poll_outcome = match &result {
Poll::Ready(Ok(())) if filled_bytes > 0 => GET_READER_POLL_READY_DATA,
Poll::Ready(Ok(())) => GET_READER_POLL_READY_EMPTY,
Poll::Ready(Err(_)) => GET_READER_POLL_READY_ERROR,
Poll::Pending => GET_READER_POLL_PENDING,
};
rustfs_io_metrics::record_get_object_stage_duration(self.metrics_path, GET_STAGE_OUTPUT_POLL, poll_duration);
rustfs_io_metrics::record_get_object_reader_poll(
self.metrics_path,
poll_outcome,
read_buf_remaining_before,
filled_bytes,
poll_duration,
);
if let (Some(read_buf_remaining_before), Some(filled_before), Some(poll_start)) =
(read_buf_remaining_before, filled_before, poll_start)
{
let poll_duration = poll_start.elapsed().as_secs_f64();
let filled_bytes = buf.filled().len().saturating_sub(filled_before);
let poll_outcome = match &result {
Poll::Ready(Ok(())) if filled_bytes > 0 => GET_READER_POLL_READY_DATA,
Poll::Ready(Ok(())) => GET_READER_POLL_READY_EMPTY,
Poll::Ready(Err(_)) => GET_READER_POLL_READY_ERROR,
Poll::Pending => GET_READER_POLL_PENDING,
};
rustfs_io_metrics::record_get_object_stage_duration(self.metrics_path, GET_STAGE_OUTPUT_POLL, poll_duration);
rustfs_io_metrics::record_get_object_reader_poll(
self.metrics_path,
poll_outcome,
read_buf_remaining_before,
filled_bytes,
poll_duration,
);
}
result
}
}
fn decode_stripe<E>(
fn decode_stripe_into<E>(
metrics_path: &'static str,
stage_metrics_enabled: bool,
engine: &E,
workspace: &mut E::Workspace,
state: StripeReadState,
remaining: usize,
) -> io::Result<Option<Vec<u8>>>
output: &mut Vec<u8>,
) -> io::Result<bool>
where
E: ErasureDecodeEngine,
{
output.clear();
if state.slots().is_empty() {
return Ok(None);
return Ok(false);
}
if !state.can_decode() {
return Err(DiskError::ErasureReadQuorum.into());
}
let reconstruct_stage_start = Instant::now();
let reconstruct_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
if state.data_shards_complete(engine.data_shards()) {
rustfs_io_metrics::record_get_object_stage_duration(
rustfs_io_metrics::record_get_object_reconstruct_outcome(
metrics_path,
GET_STAGE_RECONSTRUCT,
reconstruct_stage_start.elapsed().as_secs_f64(),
engine.engine_name(),
GET_RECONSTRUCT_OUTCOME_SKIP_DATA_COMPLETE,
);
let emit_stage_start = Instant::now();
let output = emit_data_shards(&state, engine.data_shards(), engine.block_size(), remaining)?;
rustfs_io_metrics::record_get_object_stage_duration(
metrics_path,
GET_STAGE_EMIT,
emit_stage_start.elapsed().as_secs_f64(),
);
return Ok(Some(output));
record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_RECONSTRUCT, reconstruct_stage_start);
let emit_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
emit_data_shards_into(&state, engine.data_shards(), engine.block_size(), remaining, output)?;
record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_EMIT, emit_stage_start);
return Ok(true);
}
let (mut shards, _errs) = state.into_parts();
if let Err(err) = engine.reconstruct_into(&mut shards, workspace) {
rustfs_io_metrics::record_get_object_stage_duration(
metrics_path,
GET_STAGE_RECONSTRUCT,
reconstruct_stage_start.elapsed().as_secs_f64(),
);
return Err(err);
}
rustfs_io_metrics::record_get_object_stage_duration(
metrics_path,
GET_STAGE_RECONSTRUCT,
reconstruct_stage_start.elapsed().as_secs_f64(),
);
let reconstruct_outcome = match engine.reconstruct_into(&mut shards, workspace) {
Ok(outcome) => outcome,
Err(err) => {
record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_RECONSTRUCT, reconstruct_stage_start);
return Err(err);
}
};
rustfs_io_metrics::record_get_object_reconstruct_outcome(metrics_path, engine.engine_name(), reconstruct_outcome);
record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_RECONSTRUCT, reconstruct_stage_start);
if shards.len() < engine.data_shards() {
return Err(io::Error::new(
@@ -599,8 +756,8 @@ where
));
}
let emit_stage_start = Instant::now();
let mut output = Vec::with_capacity(engine.block_size().min(remaining));
let emit_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
reserve_output_capacity(output, engine.block_size().min(remaining));
for shard in shards.iter().take(engine.data_shards()) {
if output.len() >= remaining {
break;
@@ -611,13 +768,32 @@ where
let copy_len = shard.len().min(remaining - output.len());
output.extend_from_slice(&shard[..copy_len]);
}
rustfs_io_metrics::record_get_object_stage_duration(metrics_path, GET_STAGE_EMIT, emit_stage_start.elapsed().as_secs_f64());
record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_EMIT, emit_stage_start);
Ok(Some(output))
Ok(true)
}
fn emit_data_shards(state: &StripeReadState, data_shards: usize, block_size: usize, remaining: usize) -> io::Result<Vec<u8>> {
let mut output = Vec::with_capacity(block_size.min(remaining));
let mut output = Vec::new();
emit_data_shards_into(state, data_shards, block_size, remaining, &mut output)?;
Ok(output)
}
fn reserve_output_capacity(output: &mut Vec<u8>, target_capacity: usize) {
if output.capacity() < target_capacity {
output.reserve(target_capacity - output.capacity());
}
}
fn emit_data_shards_into(
state: &StripeReadState,
data_shards: usize,
block_size: usize,
remaining: usize,
output: &mut Vec<u8>,
) -> io::Result<()> {
output.clear();
reserve_output_capacity(output, block_size.min(remaining));
for index in 0..data_shards {
if output.len() >= remaining {
break;
@@ -631,7 +807,7 @@ fn emit_data_shards(state: &StripeReadState, data_shards: usize, block_size: usi
let copy_len = shard.len().min(remaining - output.len());
output.extend_from_slice(&shard[..copy_len]);
}
Ok(output)
Ok(())
}
#[cfg(test)]
@@ -641,7 +817,7 @@ mod tests {
CodecStreamingDecodeEngine, ErasureDecodeEngine, LegacyEcDecodeEngine, RustfsCodecDecodeEngine,
};
use crate::erasure::coding::decode::ParallelReader;
use crate::erasure::coding::{BitrotReader, Erasure};
use crate::erasure::coding::{BitrotReader, BitrotWriter, Erasure};
use crate::set_disk::shard_source::{ShardSlot, StripeReadState};
use rustfs_utils::HashAlgorithm;
use std::collections::VecDeque;
@@ -757,6 +933,40 @@ mod tests {
decode_all_with_engine(&erasure, engine, data, missing_indexes).await
}
async fn bitrot_readers_from_encoded(
erasure: &Erasure,
data: &[u8],
missing_indexes: &[usize],
corrupt_indexes: &[usize],
hash_algo: HashAlgorithm,
) -> Vec<Option<BitrotReader<Cursor<Vec<u8>>>>> {
let shard_size = erasure.shard_size();
let mut readers = Vec::with_capacity(erasure.data_shards + erasure.parity_shards);
for (index, shard) in erasure
.encode_data(data)
.expect("test stripe should encode")
.into_iter()
.enumerate()
{
if missing_indexes.contains(&index) {
readers.push(None);
continue;
}
let mut writer = BitrotWriter::new(Cursor::new(Vec::new()), shard_size, hash_algo.clone());
writer.write(&shard).await.expect("test shard should write with bitrot hash");
let mut encoded = writer.into_inner().into_inner();
if corrupt_indexes.contains(&index) {
let data_offset = hash_algo.size();
if let Some(byte) = encoded.get_mut(data_offset) {
*byte ^= 0x80;
}
}
readers.push(Some(BitrotReader::new(Cursor::new(encoded), shard_size, hash_algo.clone(), false)));
}
readers
}
#[test]
fn fill_policy_defaults_to_dual_inflight() {
with_var(ENV_RUSTFS_GET_CODEC_STREAMING_MAX_INFLIGHT, None::<&str>, || {
@@ -772,6 +982,41 @@ mod tests {
});
}
#[test]
#[serial_test::serial]
fn erasure_decode_reader_caches_stage_metrics_enabled_at_construction() {
let erasure = Erasure::new(4, 2, 32);
let data = b"metrics switch cache";
rustfs_io_metrics::set_get_stage_metrics_enabled(false);
let source = source_from_data(&erasure, data, &[]);
let engine = LegacyEcDecodeEngine::new(erasure.clone());
let reader = ErasureDecodeReader::new_with_fill_policy(
source,
engine,
data.len(),
GET_OBJECT_PATH_CODEC_STREAMING,
FillPolicy::SingleInFlight,
)
.expect("reader should be constructed");
assert!(!reader.stage_metrics_enabled);
rustfs_io_metrics::set_get_stage_metrics_enabled(true);
let source = source_from_data(&erasure, data, &[]);
let engine = LegacyEcDecodeEngine::new(erasure);
let reader = ErasureDecodeReader::new_with_fill_policy(
source,
engine,
data.len(),
GET_OBJECT_PATH_CODEC_STREAMING,
FillPolicy::SingleInFlight,
)
.expect("reader should be constructed");
assert!(reader.stage_metrics_enabled);
rustfs_io_metrics::set_get_stage_metrics_enabled(false);
}
#[tokio::test]
async fn erasure_decode_reader_reads_single_stripe() {
let erasure = Erasure::new(4, 2, 64);
@@ -1012,6 +1257,39 @@ mod tests {
assert!(decoded.is_empty());
}
#[tokio::test]
async fn erasure_decode_reader_rustfs_engine_recovers_after_bitrot_source_mismatch() {
let erasure = Erasure::new(4, 2, 64);
let data = (0..64u16)
.map(|value| value.wrapping_mul(29).to_le_bytes()[0])
.collect::<Vec<_>>();
let readers = bitrot_readers_from_encoded(&erasure, &data, &[0], &[1], HashAlgorithm::HighwayHash256).await;
let source = ParallelReader::new_with_metrics_path_and_reconstruction_verification(
readers,
erasure.clone(),
0,
data.len(),
Some(GET_OBJECT_PATH_CODEC_STREAMING),
);
let engine = RustfsCodecDecodeEngine::new(&erasure).expect("engine should be created");
let mut reader = ErasureDecodeReader::new_with_fill_policy(
source,
engine,
data.len(),
GET_OBJECT_PATH_CODEC_STREAMING,
FillPolicy::SingleInFlight,
)
.expect("reader should be constructed");
let mut decoded = Vec::new();
reader
.read_to_end(&mut decoded)
.await
.expect("rustfs reader should reconstruct from clean shards after bitrot mismatch");
assert_eq!(decoded, data);
}
#[tokio::test]
async fn erasure_decode_reader_verifying_parallel_source_rejects_inconsistent_reconstruction_sources() {
const DATA_SHARDS: usize = 2;
@@ -1176,4 +1454,74 @@ mod tests {
.await
.expect("reader should start reading the next stripe before the current output buffer is fully consumed");
}
#[tokio::test]
async fn erasure_decode_reader_reuses_output_buffers_after_drain() {
let erasure = Erasure::new(4, 2, 32);
let data = (0..128u16)
.map(|value| value.wrapping_mul(5).to_le_bytes()[0])
.collect::<Vec<_>>();
let source = source_from_data(&erasure, &data, &[]);
let engine = LegacyEcDecodeEngine::new(erasure.clone());
let mut reader = ErasureDecodeReader::new_with_fill_policy(
source,
engine,
data.len(),
GET_OBJECT_PATH_CODEC_STREAMING,
FillPolicy::SingleInFlight,
)
.expect("reader should be constructed");
let mut decoded = Vec::new();
reader
.read_to_end(&mut decoded)
.await
.expect("reader should decode all stripes");
assert_eq!(decoded, data);
assert!(reader.reusable_output_bufs.len() <= reader.max_reusable_output_bufs());
assert!(
reader
.reusable_output_bufs
.iter()
.any(|buf| buf.capacity() >= erasure.block_size),
"drained stripe output buffers should be available for reuse"
);
}
#[tokio::test]
async fn erasure_decode_reader_reuses_single_fill_worker_across_fills() {
let erasure = Erasure::new(4, 2, 32);
let data = (0..128u16)
.map(|value| value.wrapping_mul(7).to_le_bytes()[0])
.collect::<Vec<_>>();
let source = source_from_data(&erasure, &data, &[]);
let engine = LegacyEcDecodeEngine::new(erasure);
let mut reader = ErasureDecodeReader::new_with_fill_policy(
source,
engine,
data.len(),
GET_OBJECT_PATH_CODEC_STREAMING,
FillPolicy::SingleInFlight,
)
.expect("reader should be constructed");
let mut decoded = Vec::new();
let mut first_read = [0u8; 1];
let read = reader.read(&mut first_read).await.expect("first read should succeed");
decoded.extend_from_slice(&first_read[..read]);
assert!(reader.worker.is_some());
assert!(reader.source.is_none());
assert!(reader.engine.is_none());
assert!(reader.workspace.is_none());
reader
.read_to_end(&mut decoded)
.await
.expect("reader should continue using the fill worker");
assert_eq!(decoded, data);
assert!(reader.worker.is_some());
}
}
crates/ecstore/src/set_disk/mod.rs
+100 -40
View File
@@ -171,10 +171,10 @@ const EVENT_SET_DISK_PUT_OBJECT_STAGE_SUMMARY: &str = "set_disk_put_object_stage
const SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS: u128 = 5_000;
const ENV_RUSTFS_PUT_LARGE_BATCH_MIN_SIZE_BYTES: &str = "RUSTFS_PUT_LARGE_BATCH_MIN_SIZE_BYTES";
const DEFAULT_RUSTFS_PUT_LARGE_BATCH_MIN_SIZE_BYTES: usize = 64 * 1024 * 1024;
static CACHED_PUT_LARGE_BATCH_MIN_SIZE_BYTES: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
static CACHED_PUT_LARGE_BATCH_MIN_SIZE_BYTES: OnceLock<usize> = OnceLock::new();
const ENV_RUSTFS_MULTIPART_PUT_LARGE_BATCH_MIN_SIZE_BYTES: &str = "RUSTFS_MULTIPART_PUT_LARGE_BATCH_MIN_SIZE_BYTES";
const DEFAULT_RUSTFS_MULTIPART_PUT_LARGE_BATCH_MIN_SIZE_BYTES: usize = 128 * 1024 * 1024;
static CACHED_MULTIPART_PUT_LARGE_BATCH_MIN_SIZE_BYTES: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
static CACHED_MULTIPART_PUT_LARGE_BATCH_MIN_SIZE_BYTES: OnceLock<usize> = OnceLock::new();
use crate::io_support::rio::{EtagResolvable, HashReader, HashReaderMut, TryGetIndex as _};
@@ -355,6 +355,9 @@ const DEFAULT_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT: u32 = 100;
const ENV_RUSTFS_GET_CODEC_STREAMING_MULTIPART_ENABLE: &str = "RUSTFS_GET_CODEC_STREAMING_MULTIPART_ENABLE";
const DEFAULT_RUSTFS_GET_CODEC_STREAMING_MULTIPART_ENABLE: bool = false;
const ENV_RUSTFS_GET_CODEC_STREAMING_MULTIPART_MAX_PARTS: &str = "RUSTFS_GET_CODEC_STREAMING_MULTIPART_MAX_PARTS";
const DEFAULT_RUSTFS_GET_CODEC_STREAMING_MULTIPART_MAX_PARTS: usize = 256;
// --- Metadata Early-Stop Configuration ---
const ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_EARLY_STOP_ENABLE";
@@ -472,13 +475,43 @@ fn is_get_codec_streaming_enabled() -> bool {
/// When enabled, multipart objects use per-part codec streaming
/// instead of falling back to the legacy duplex path.
fn is_codec_streaming_multipart_enabled() -> bool {
static CACHED: OnceLock<bool> = OnceLock::new();
*CACHED.get_or_init(|| {
#[cfg(test)]
{
rustfs_utils::get_env_bool(
ENV_RUSTFS_GET_CODEC_STREAMING_MULTIPART_ENABLE,
DEFAULT_RUSTFS_GET_CODEC_STREAMING_MULTIPART_ENABLE,
)
})
}
#[cfg(not(test))]
{
static CACHED: OnceLock<bool> = OnceLock::new();
*CACHED.get_or_init(|| {
rustfs_utils::get_env_bool(
ENV_RUSTFS_GET_CODEC_STREAMING_MULTIPART_ENABLE,
DEFAULT_RUSTFS_GET_CODEC_STREAMING_MULTIPART_ENABLE,
)
})
}
}
fn get_codec_streaming_multipart_max_parts() -> usize {
#[cfg(test)]
{
rustfs_utils::get_env_usize(
ENV_RUSTFS_GET_CODEC_STREAMING_MULTIPART_MAX_PARTS,
DEFAULT_RUSTFS_GET_CODEC_STREAMING_MULTIPART_MAX_PARTS,
)
}
#[cfg(not(test))]
{
static CACHED: OnceLock<usize> = OnceLock::new();
*CACHED.get_or_init(|| {
rustfs_utils::get_env_usize(
ENV_RUSTFS_GET_CODEC_STREAMING_MULTIPART_MAX_PARTS,
DEFAULT_RUSTFS_GET_CODEC_STREAMING_MULTIPART_MAX_PARTS,
)
})
}
}
/// Check if metadata early-stop is enabled (base flag).
@@ -506,10 +539,17 @@ fn is_version_early_stop_enabled() -> bool {
// --- Rollout Percentage Functions ---
fn get_codec_streaming_rollout_pct() -> u32 {
static CACHED: OnceLock<u32> = OnceLock::new();
*CACHED.get_or_init(|| {
#[cfg(test)]
{
rustfs_utils::get_env_u32(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT)
})
}
#[cfg(not(test))]
{
static CACHED: OnceLock<u32> = OnceLock::new();
*CACHED.get_or_init(|| {
rustfs_utils::get_env_u32(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT)
})
}
}
fn get_metadata_early_stop_rollout_pct() -> u32 {
@@ -545,7 +585,6 @@ fn is_optimization_enabled_for_request(base_enabled: bool, rollout_pct: u32, buc
(hash as u32) < rollout_pct
}
/// Should this specific request use codec streaming?
pub fn should_use_codec_streaming(bucket: &str, object: &str) -> bool {
let base = is_get_codec_streaming_enabled();
@@ -633,6 +672,7 @@ impl GetCodecStreamingRollout {
enum GetCodecStreamingFallbackReason {
Disabled,
RolloutNotOptedIn,
RolloutPctNotSelected,
BodyCompatibilityUnconfirmed,
HeaderCompatibilityUnconfirmed,
LockOptimizationDisabled,
@@ -644,6 +684,7 @@ enum GetCodecStreamingFallbackReason {
Multipart,
InvalidMinSize,
ReadQuorumNotSafe,
MultipartPartLimit,
}
impl GetCodecStreamingFallbackReason {
@@ -651,6 +692,7 @@ impl GetCodecStreamingFallbackReason {
match self {
Self::Disabled => "disabled",
Self::RolloutNotOptedIn => "rollout_not_opted_in",
Self::RolloutPctNotSelected => "rollout_pct_not_selected",
Self::BodyCompatibilityUnconfirmed => "body_compatibility_unconfirmed",
Self::HeaderCompatibilityUnconfirmed => "header_compatibility_unconfirmed",
Self::LockOptimizationDisabled => "lock_optimization_disabled",
@@ -662,6 +704,7 @@ impl GetCodecStreamingFallbackReason {
Self::Multipart => "multipart",
Self::InvalidMinSize => "invalid_min_size",
Self::ReadQuorumNotSafe => "read_quorum_not_safe",
Self::MultipartPartLimit => "multipart_part_limit",
}
}
}
@@ -732,6 +775,8 @@ fn classify_get_codec_streaming_object_class(
}
fn get_codec_streaming_reader_gate(
bucket: &str,
object: &str,
range: &Option<HTTPRangeSpec>,
object_info: &ObjectInfo,
fi: &FileInfo,
@@ -751,6 +796,12 @@ fn get_codec_streaming_reader_gate(
decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::RolloutNotOptedIn),
};
}
if !should_use_codec_streaming(bucket, object) {
return GetCodecStreamingGate {
object_class,
decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::RolloutPctNotSelected),
};
}
if !is_get_codec_streaming_body_compat_confirmed() {
return GetCodecStreamingGate {
object_class,
@@ -807,10 +858,18 @@ fn get_codec_streaming_reader_gate(
};
}
if object_class == GetCodecStreamingObjectClass::Multipart {
return GetCodecStreamingGate {
object_class,
decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Multipart),
};
if !is_codec_streaming_multipart_enabled() {
return GetCodecStreamingGate {
object_class,
decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Multipart),
};
}
if fi.parts.len() > get_codec_streaming_multipart_max_parts() {
return GetCodecStreamingGate {
object_class,
decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::MultipartPartLimit),
};
}
}
GetCodecStreamingGate {
@@ -1615,7 +1674,8 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
}
}
let codec_streaming_gate = get_codec_streaming_reader_gate(&range, &object_info, &fi, lock_optimization_enabled);
let codec_streaming_gate =
get_codec_streaming_reader_gate(bucket, object, &range, &object_info, &fi, lock_optimization_enabled);
if object_info.is_remote() {
if let GetCodecStreamingDecision::Fallback(reason) = codec_streaming_gate.decision {
@@ -1826,7 +1886,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
let result: Result<ObjectInfo> = async {
let erasure =
crate::erasure::coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
let is_inline_buffer =
runtime_sources::storage_class_should_inline(erasure.shard_file_size(data.size()), opts.versioned);
@@ -2411,10 +2471,10 @@ impl SetDisks {
}
}
Ok((_client_idx, Err(err))) => {
tracing::warn!("late distributed delete lock batch request failed: {}", err);
warn!("late distributed delete lock batch request failed: {}", err);
}
Err(err) => {
tracing::warn!("late distributed delete lock batch task join failed: {}", err);
warn!("late distributed delete lock batch task join failed: {}", err);
}
}
}
@@ -2426,7 +2486,7 @@ impl SetDisks {
} else {
Some(async move {
if let Err(err) = client.release_locks_batch(&lock_ids).await {
tracing::warn!(
warn!(
client_idx,
lock_count = lock_ids.len(),
"failed to cleanup late distributed delete locks in batch: {}",
@@ -2488,7 +2548,7 @@ impl SetDisks {
} else {
Some(async move {
if let Err(err) = client.release_locks_batch(&lock_ids).await {
tracing::warn!(
warn!(
client_idx,
lock_count = lock_ids.len(),
"failed to release distributed delete locks in batch: {}",
@@ -3588,7 +3648,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
if let Err(err) = dest_obj {
return Err(to_object_err(err, vec![]));
}
let dest_obj = dest_obj.unwrap();
let dest_obj = dest_obj?;
let oi = ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended);
let mut transition_meta = (*oi.user_defined).clone();
@@ -3721,7 +3781,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
if let Err(err) = fi {
return set_restore_header_fn(&mut oi, Some(to_object_err(err, vec![bucket, object]))).await;
}
let (actual_fi, _, _) = fi.unwrap();
let (actual_fi, _, _) = fi?;
oi = ObjectInfo::from_file_info(&actual_fi, bucket, object, opts.versioned || opts.version_suspended);
let ropts = put_restore_opts(bucket, object, &opts.transition.restore_request, &oi).await?;
@@ -3733,7 +3793,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
if let Err(err) = gr {
return set_restore_header_fn(&mut oi, Some(to_object_err(err.into(), vec![bucket, object]))).await;
}
let gr = gr.unwrap();
let gr = gr?;
let reader = BufReader::new(gr.stream);
let hash_reader = HashReader::from_stream(reader, gr.object_info.size, gr.object_info.size, None, None, false)?;
let mut p_reader = PutObjReader::new(hash_reader);
@@ -4159,7 +4219,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
let tmp_part_path = Arc::new(format!("{tmp_part}/{part_suffix}"));
let erasure =
crate::erasure::coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
let writer_setup_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now);
let mut writers = Vec::with_capacity(shuffle_disks.len());
@@ -5853,7 +5913,7 @@ async fn get_disks_info(disks: &[Option<DiskStore>], eps: &[Endpoint]) -> Vec<ru
let offline_duration_seconds = disk.offline_duration_secs();
let capacity_snapshot = disk.last_capacity_snapshot();
if runtime_state.should_probe_for_admin()
|| runtime_state == crate::disk::health_state::RuntimeDriveHealthState::Suspect
|| runtime_state == disk::health_state::RuntimeDriveHealthState::Suspect
{
match disk.disk_info(&DiskInfoOptions::default()).await {
Ok(res) => {
@@ -5946,7 +6006,7 @@ async fn get_disks_info(disks: &[Option<DiskStore>], eps: &[Endpoint]) -> Vec<ru
fn build_runtime_snapshot_disk(
endpoint: &Endpoint,
runtime_state: crate::disk::health_state::RuntimeDriveHealthState,
runtime_state: disk::health_state::RuntimeDriveHealthState,
offline_duration_seconds: Option<u64>,
capacity_snapshot: Option<(u64, u64, u64, u64)>,
) -> rustfs_madmin::Disk {
@@ -6809,7 +6869,7 @@ mod tests {
..Default::default()
};
let result = tokio::time::timeout(
let result = timeout(
Duration::from_secs(1),
set_disks.copy_object(
"bucket",
@@ -6881,7 +6941,7 @@ mod tests {
.await
.expect("outer write lock should be acquired");
let result = tokio::time::timeout(
let result = timeout(
Duration::from_secs(1),
set_disks.delete_object(
"bucket",
@@ -6919,7 +6979,7 @@ mod tests {
.await
.expect("outer write lock should be acquired");
tokio::time::timeout(
timeout(
Duration::from_secs(1),
set_disks.delete_object(
"bucket",
@@ -6952,7 +7012,7 @@ mod tests {
.await
.expect("outer write lock should be acquired");
tokio::time::timeout(
timeout(
Duration::from_secs(1),
set_disks.delete_object(
"bucket",
@@ -6987,7 +7047,7 @@ mod tests {
.await
.expect("outer write lock should be acquired");
let result = tokio::time::timeout(
let result = timeout(
Duration::from_millis(50),
set_disks.delete_object(
"bucket",
@@ -7617,11 +7677,11 @@ mod tests {
disk.force_runtime_state_for_test(RuntimeDriveHealthState::Offline);
}
let (tx, _rx) = tokio::sync::mpsc::channel(1);
let (tx, _rx) = mpsc::channel(1);
let err = set_disks
.list_path(
CancellationToken::new(),
crate::store::list_objects::ListPathOptions {
ListPathOptions {
bucket: "bucket".to_string(),
recursive: true,
..Default::default()
@@ -7675,7 +7735,7 @@ mod tests {
disk.make_volume(bucket).await.expect("bucket should be created");
let metadata_path = format!("{object}/{STORAGE_FORMAT_FILE}");
disk.write_all(bucket, &metadata_path, bytes::Bytes::from_static(b"not-xl-meta"))
disk.write_all(bucket, &metadata_path, Bytes::from_static(b"not-xl-meta"))
.await
.expect("corrupt metadata file should be written");
@@ -7730,7 +7790,7 @@ mod tests {
disk.make_volume(bucket).await.expect("bucket should be created");
let metadata_path = format!("{object}/{STORAGE_FORMAT_FILE}");
disk.write_all(bucket, &metadata_path, bytes::Bytes::from_static(b"not-an-xl-meta"))
disk.write_all(bucket, &metadata_path, Bytes::from_static(b"not-an-xl-meta"))
.await
.expect("metadata file should be created");
@@ -7768,7 +7828,7 @@ mod tests {
assert_eq!(walk_err, DiskError::Timeout);
assert_eq!(disk.runtime_state(), RuntimeDriveHealthState::Online);
let (tx, mut rx) = tokio::sync::mpsc::channel::<MetaCacheEntry>(4);
let (tx, mut rx) = mpsc::channel::<MetaCacheEntry>(4);
set_disks
.list_path(
CancellationToken::new(),
@@ -7819,7 +7879,7 @@ mod tests {
let object = "config/iam/sts/test/identity.json";
let metadata_path = format!("{object}/{STORAGE_FORMAT_FILE}");
disk.write_all(RUSTFS_META_BUCKET, &metadata_path, bytes::Bytes::from_static(b"not-an-xl-meta"))
disk.write_all(RUSTFS_META_BUCKET, &metadata_path, Bytes::from_static(b"not-an-xl-meta"))
.await
.expect("system path metadata file should be created");
@@ -7858,7 +7918,7 @@ mod tests {
assert_eq!(walk_err, DiskError::Timeout);
assert_eq!(disk.runtime_state(), RuntimeDriveHealthState::Online);
let (tx, mut rx) = tokio::sync::mpsc::channel::<MetaCacheEntry>(4);
let (tx, mut rx) = mpsc::channel::<MetaCacheEntry>(4);
set_disks
.list_path(
CancellationToken::new(),
@@ -8302,7 +8362,7 @@ mod tests {
fi.size = payload.len() as i64;
fi.add_object_part(1, String::new(), payload.len(), None, payload.len() as i64, None, None);
let erasure = crate::erasure::coding::Erasure::new_with_options(
let erasure = coding::Erasure::new_with_options(
fi.erasure.data_blocks,
fi.erasure.parity_blocks,
fi.erasure.block_size,
@@ -8587,7 +8647,7 @@ mod tests {
.await
.expect("format should be saved");
std::mem::forget(dir);
mem::forget(dir);
endpoints.push(endpoint);
disks.push(Some(disk));
}
@@ -8637,7 +8697,7 @@ mod tests {
.expect("format should be saved");
}
std::mem::forget(dir);
mem::forget(dir);
endpoints.push(endpoint);
disks.push(Some(disk));
}
crates/ecstore/src/set_disk/read.rs
+421 -40
View File
@@ -22,7 +22,8 @@ use crate::diagnostics::get::{
GET_METADATA_RESPONSE_ERROR, GET_METADATA_RESPONSE_IGNORED, GET_METADATA_RESPONSE_NOT_FOUND, GET_METADATA_RESPONSE_TIMEOUT,
GET_METADATA_RESPONSE_VALID, GET_METADATA_RESPONSE_VERSION_NOT_FOUND, GET_OBJECT_PATH_CODEC_STREAMING,
GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_DECODE, GET_STAGE_RANGE, GET_STAGE_READER_SETUP, GetObjectFailureReason,
classify_disk_error, record_get_object_pipeline_failure, record_get_object_pipeline_failure_for_path,
classify_disk_error, get_stage_timer_if_enabled, record_get_object_pipeline_failure,
record_get_object_pipeline_failure_for_path, record_get_stage_duration_if_enabled,
};
use crate::erasure::coding::BitrotReader;
use crate::io_support::bitrot::create_deferred_bitrot_reader;
@@ -31,13 +32,14 @@ use futures::stream::{FuturesUnordered, StreamExt};
use metrics::counter;
use rustfs_config::{DEFAULT_OBJECT_ZERO_COPY_ENABLE, ENV_OBJECT_ZERO_COPY_ENABLE};
use std::{
collections::HashMap,
collections::{HashMap, VecDeque},
future::Future,
pin::Pin,
sync::OnceLock,
task::{Context, Poll},
time::{Duration, Instant},
};
use tokio::io::AsyncRead;
use tokio::io::{AsyncRead, ReadBuf};
use tokio::sync::RwLock;
use tokio::task::JoinSet;
@@ -53,6 +55,39 @@ pub(super) enum GetCodecStreamingReaderBuildOutcome {
Fallback(GetCodecStreamingFallbackReason),
}
struct MultipartCodecStreamingReader {
readers: VecDeque<Box<dyn AsyncRead + Unpin + Send + Sync>>,
}
impl MultipartCodecStreamingReader {
fn new(readers: Vec<Box<dyn AsyncRead + Unpin + Send + Sync>>) -> Self {
Self {
readers: VecDeque::from(readers),
}
}
}
impl AsyncRead for MultipartCodecStreamingReader {
fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
if buf.remaining() == 0 {
return Poll::Ready(Ok(()));
}
loop {
let Some(reader) = self.readers.front_mut() else {
return Poll::Ready(Ok(()));
};
let filled_before = buf.filled().len();
match Pin::new(reader).poll_read(cx, buf) {
Poll::Ready(Ok(())) if buf.filled().len() == filled_before => {
self.readers.pop_front();
}
result => return result,
}
}
}
}
pub(super) fn codec_streaming_reader_setup_fallback_reason(missing_shards: usize) -> Option<GetCodecStreamingFallbackReason> {
(missing_shards > 0).then_some(GetCodecStreamingFallbackReason::ReadQuorumNotSafe)
}
@@ -2242,9 +2277,6 @@ impl SetDisks {
skip_verify_bitrot: bool,
) -> Result<GetCodecStreamingReaderBuildOutcome> {
let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi);
if fi.parts.len() != 1 {
return Err(Error::other("codec streaming reader only supports single-part plain objects"));
}
let erasure = crate::erasure::coding::Erasure::new_with_options(
fi.erasure.data_blocks,
@@ -2252,14 +2284,94 @@ impl SetDisks {
fi.erasure.block_size,
fi.uses_legacy_checksum,
);
let part = &fi.parts[0];
let part_number = part.number;
let part_size = part.size;
let part_length = usize::try_from(fi.size).map_err(|_| Error::other("codec streaming reader object size is invalid"))?;
if fi.parts.len() == 1 {
let part = &fi.parts[0];
let part_length =
usize::try_from(fi.size).map_err(|_| Error::other("codec streaming reader object size is invalid"))?;
return Self::build_codec_streaming_part_reader(
bucket,
object,
fi,
&files,
&disks,
&erasure,
part.number,
0,
part_length,
part.size,
skip_verify_bitrot,
)
.await;
}
if !is_codec_streaming_multipart_enabled() {
return Ok(GetCodecStreamingReaderBuildOutcome::Fallback(GetCodecStreamingFallbackReason::Multipart));
}
if fi.parts.len() > get_codec_streaming_multipart_max_parts() {
return Ok(GetCodecStreamingReaderBuildOutcome::Fallback(
GetCodecStreamingFallbackReason::MultipartPartLimit,
));
}
let object_length =
usize::try_from(fi.size).map_err(|_| Error::other("codec streaming reader object size is invalid"))?;
let mut total_part_size = 0usize;
for part in &fi.parts {
total_part_size = total_part_size
.checked_add(part.size)
.ok_or_else(|| Error::other("codec streaming multipart part sizes overflow"))?;
}
if total_part_size != object_length {
return Err(Error::other("codec streaming multipart part sizes do not match object size"));
}
let mut readers = Vec::with_capacity(fi.parts.len());
for part in &fi.parts {
match Self::build_codec_streaming_part_reader(
bucket,
object,
fi,
&files,
&disks,
&erasure,
part.number,
0,
part.size,
part.size,
skip_verify_bitrot,
)
.await?
{
GetCodecStreamingReaderBuildOutcome::Reader(reader) => readers.push(reader),
GetCodecStreamingReaderBuildOutcome::Fallback(reason) => {
return Ok(GetCodecStreamingReaderBuildOutcome::Fallback(reason));
}
}
}
Ok(GetCodecStreamingReaderBuildOutcome::Reader(Box::new(MultipartCodecStreamingReader::new(
readers,
))))
}
#[allow(clippy::too_many_arguments)]
async fn build_codec_streaming_part_reader(
bucket: &str,
object: &str,
fi: &FileInfo,
files: &[FileInfo],
disks: &[Option<DiskStore>],
erasure: &crate::erasure::coding::Erasure,
part_number: usize,
part_offset: usize,
part_length: usize,
part_size: usize,
skip_verify_bitrot: bool,
) -> Result<GetCodecStreamingReaderBuildOutcome> {
if part_length > part_size {
return Err(Error::other("codec streaming reader part length exceeds part size"));
}
let checksum_info = fi.erasure.get_checksum_info(part_number);
let checksum_algo = if fi.uses_legacy_checksum && checksum_info.algorithm == rustfs_utils::HashAlgorithm::HighwayHash256S
{
@@ -2268,21 +2380,24 @@ impl SetDisks {
checksum_info.algorithm
};
let use_mmap_read = rustfs_utils::get_env_bool(ENV_OBJECT_ZERO_COPY_ENABLE, DEFAULT_OBJECT_ZERO_COPY_ENABLE);
let till_offset = erasure.shard_file_offset(0, part_length, part_size);
let till_offset = erasure.shard_file_offset(part_offset, part_length, part_size);
let read_offset = (part_offset / erasure.block_size) * erasure.shard_size();
let read_length = till_offset.saturating_sub(read_offset);
let reader_setup_stage_start = Instant::now();
let stage_metrics_enabled = rustfs_io_metrics::get_stage_metrics_enabled();
let reader_setup_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let read_costs = disks
.iter()
.map(|disk| shard_read_cost_for_disk(disk.as_ref()))
.collect::<Vec<_>>();
let reader_setup = create_bitrot_readers_until_quorum(
&files,
&disks,
files,
disks,
bucket,
object,
part_number,
0,
till_offset,
read_offset,
read_length,
erasure.shard_size(),
checksum_algo,
skip_verify_bitrot,
@@ -2293,11 +2408,7 @@ impl SetDisks {
)
.await;
let metrics_path = get_codec_streaming_metrics_path();
rustfs_io_metrics::record_get_object_stage_duration(
metrics_path,
GET_STAGE_READER_SETUP,
reader_setup_stage_start.elapsed().as_secs_f64(),
);
record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READER_SETUP, reader_setup_stage_start);
let available_shards = reader_setup.available_shards();
if available_shards < erasure.data_shards {
@@ -2320,12 +2431,12 @@ impl SetDisks {
crate::erasure::coding::decode::ParallelReader::new_with_metrics_path_read_costs_and_reconstruction_verification(
readers,
erasure.clone(),
0,
part_offset,
part_size,
Some(metrics_path),
read_costs,
);
let engine = build_get_codec_streaming_decode_engine(erasure)?;
let engine = build_get_codec_streaming_decode_engine(erasure.clone())?;
let reader = crate::erasure::coding::decode_reader::ErasureDecodeReader::new_with_metrics_path(
source,
engine,
@@ -2696,6 +2807,15 @@ mod metadata_cache_tests {
#[cfg(test)]
mod tests {
use super::*;
use std::io::Cursor;
use std::sync::{
Arc,
atomic::{AtomicUsize, Ordering},
};
use tokio::io::AsyncReadExt;
const CODEC_STREAMING_TEST_BUCKET: &str = "bucket";
const CODEC_STREAMING_TEST_OBJECT: &str = "object";
fn metadata_fanout_test_fileinfo(object: &str) -> FileInfo {
let mut fi = FileInfo::new(object, 2, 2);
@@ -3247,7 +3367,147 @@ mod tests {
}
fn codec_streaming_test_object_info(fi: &FileInfo) -> ObjectInfo {
ObjectInfo::from_file_info(fi, "bucket", "object", false)
ObjectInfo::from_file_info(fi, CODEC_STREAMING_TEST_BUCKET, CODEC_STREAMING_TEST_OBJECT, false)
}
fn codec_streaming_reader_gate_for_test(
range: &Option<HTTPRangeSpec>,
object_info: &ObjectInfo,
fi: &FileInfo,
lock_optimization_enabled: bool,
) -> GetCodecStreamingGate {
get_codec_streaming_reader_gate(
CODEC_STREAMING_TEST_BUCKET,
CODEC_STREAMING_TEST_OBJECT,
range,
object_info,
fi,
lock_optimization_enabled,
)
}
#[tokio::test]
async fn multipart_codec_streaming_reader_reads_parts_in_order() {
let readers: Vec<Box<dyn AsyncRead + Unpin + Send + Sync>> = vec![
Box::new(Cursor::new(b"hello ".to_vec())),
Box::new(Cursor::new(b"multipart".to_vec())),
];
let mut reader = MultipartCodecStreamingReader::new(readers);
let mut output = Vec::new();
reader
.read_to_end(&mut output)
.await
.expect("multipart codec reader should read all parts");
assert_eq!(output, b"hello multipart");
}
struct OneByteAsyncReader {
data: Vec<u8>,
position: usize,
}
impl OneByteAsyncReader {
fn new(data: &'static [u8]) -> Self {
Self {
data: data.to_vec(),
position: 0,
}
}
}
impl AsyncRead for OneByteAsyncReader {
fn poll_read(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
if self.position >= self.data.len() || buf.remaining() == 0 {
return Poll::Ready(Ok(()));
}
buf.put_slice(&self.data[self.position..self.position + 1]);
self.position += 1;
Poll::Ready(Ok(()))
}
}
#[tokio::test]
async fn multipart_codec_streaming_reader_crosses_part_boundaries_with_short_reads() {
let readers: Vec<Box<dyn AsyncRead + Unpin + Send + Sync>> = vec![
Box::new(OneByteAsyncReader::new(b"abc")),
Box::new(OneByteAsyncReader::new(b"def")),
];
let mut reader = MultipartCodecStreamingReader::new(readers);
let mut first = [0u8; 5];
let mut second = Vec::new();
reader
.read_exact(&mut first)
.await
.expect("multipart codec reader should cross part boundaries");
reader
.read_to_end(&mut second)
.await
.expect("multipart codec reader should drain the final part");
assert_eq!(&first, b"abcde");
assert_eq!(second, b"f");
}
struct DropCountingReader {
data: Vec<u8>,
position: usize,
drops: Arc<AtomicUsize>,
}
impl DropCountingReader {
fn new(data: &'static [u8], drops: Arc<AtomicUsize>) -> Self {
Self {
data: data.to_vec(),
position: 0,
drops,
}
}
}
impl AsyncRead for DropCountingReader {
fn poll_read(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
if self.position >= self.data.len() || buf.remaining() == 0 {
return Poll::Ready(Ok(()));
}
let available = self.data.len() - self.position;
let count = available.min(buf.remaining());
let end = self.position + count;
buf.put_slice(&self.data[self.position..end]);
self.position = end;
Poll::Ready(Ok(()))
}
}
impl Drop for DropCountingReader {
fn drop(&mut self) {
self.drops.fetch_add(1, Ordering::SeqCst);
}
}
#[tokio::test]
async fn multipart_codec_streaming_reader_drops_remaining_parts_on_abort() {
let drops = Arc::new(AtomicUsize::new(0));
{
let readers: Vec<Box<dyn AsyncRead + Unpin + Send + Sync>> = vec![
Box::new(DropCountingReader::new(b"abc", Arc::clone(&drops))),
Box::new(DropCountingReader::new(b"def", Arc::clone(&drops))),
];
let mut reader = MultipartCodecStreamingReader::new(readers);
let mut first = [0u8; 1];
reader
.read_exact(&mut first)
.await
.expect("multipart codec reader should support partial reads");
assert_eq!(&first, b"a");
}
assert_eq!(drops.load(Ordering::SeqCst), 2);
}
fn inline_reader_setup_fileinfo(data: Option<&'static [u8]>) -> FileInfo {
@@ -3395,12 +3655,12 @@ mod tests {
let fi = codec_streaming_test_fileinfo(1024, 1);
let object_info = codec_streaming_test_object_info(&fi);
assert_eq!(
get_codec_streaming_reader_gate(&None, &object_info, &fi, true).decision,
codec_streaming_reader_gate_for_test(&None, &object_info, &fi, true).decision,
GetCodecStreamingDecision::Use
);
assert_eq!(
get_codec_streaming_reader_gate(&None, &object_info, &fi, false).decision,
codec_streaming_reader_gate_for_test(&None, &object_info, &fi, false).decision,
GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::LockOptimizationDisabled)
);
@@ -3410,14 +3670,14 @@ mod tests {
end: 1,
});
assert_eq!(
get_codec_streaming_reader_gate(&range, &object_info, &fi, true).decision,
codec_streaming_reader_gate_for_test(&range, &object_info, &fi, true).decision,
GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Range)
);
let multipart_fi = codec_streaming_test_fileinfo(1024, 2);
let multipart_object_info = codec_streaming_test_object_info(&multipart_fi);
assert_eq!(
get_codec_streaming_reader_gate(&None, &multipart_object_info, &multipart_fi, true).decision,
codec_streaming_reader_gate_for_test(&None, &multipart_object_info, &multipart_fi, true).decision,
GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Multipart)
);
@@ -3427,7 +3687,7 @@ mod tests {
.insert("x-amz-server-side-encryption".to_string(), "AES256".to_string());
let encrypted = codec_streaming_test_object_info(&encrypted_fi);
assert_eq!(
get_codec_streaming_reader_gate(&None, &encrypted, &encrypted_fi, true).decision,
codec_streaming_reader_gate_for_test(&None, &encrypted, &encrypted_fi, true).decision,
GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Encrypted)
);
@@ -3435,14 +3695,14 @@ mod tests {
insert_str(&mut compressed_fi.metadata, SUFFIX_COMPRESSION, "lz4".to_string());
let compressed = codec_streaming_test_object_info(&compressed_fi);
assert_eq!(
get_codec_streaming_reader_gate(&None, &compressed, &compressed_fi, true).decision,
codec_streaming_reader_gate_for_test(&None, &compressed, &compressed_fi, true).decision,
GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Compressed)
);
let small_fi = codec_streaming_test_fileinfo(0, 1);
let small_object_info = codec_streaming_test_object_info(&small_fi);
assert_eq!(
get_codec_streaming_reader_gate(&None, &small_object_info, &small_fi, true).decision,
codec_streaming_reader_gate_for_test(&None, &small_object_info, &small_fi, true).decision,
GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::BelowMinSize)
);
@@ -3450,7 +3710,7 @@ mod tests {
remote_fi.transition_status = crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE.to_string();
let remote = codec_streaming_test_object_info(&remote_fi);
assert_eq!(
get_codec_streaming_reader_gate(&None, &remote, &remote_fi, true).decision,
codec_streaming_reader_gate_for_test(&None, &remote, &remote_fi, true).decision,
GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Remote)
);
},
@@ -3488,13 +3748,86 @@ mod tests {
let object_info = codec_streaming_test_object_info(&fi);
assert_eq!(
get_codec_streaming_reader_gate(&None, &object_info, &fi, true).decision,
codec_streaming_reader_gate_for_test(&None, &object_info, &fi, true).decision,
GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Disabled)
);
},
);
}
#[test]
fn codec_streaming_reader_gate_allows_multipart_when_explicitly_enabled() {
temp_env::with_vars(
[
(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("benchmark")),
(ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_MULTIPART_ENABLE, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")),
],
|| {
let fi = codec_streaming_test_fileinfo(1024, 2);
let object_info = codec_streaming_test_object_info(&fi);
let gate = codec_streaming_reader_gate_for_test(&None, &object_info, &fi, true);
assert_eq!(gate.object_class, GetCodecStreamingObjectClass::Multipart);
assert_eq!(gate.decision, GetCodecStreamingDecision::Use);
},
);
}
#[test]
fn codec_streaming_reader_gate_keeps_multipart_default_off() {
temp_env::with_vars(
[
(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("benchmark")),
(ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_MULTIPART_ENABLE, None),
(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")),
],
|| {
let fi = codec_streaming_test_fileinfo(1024, 2);
let object_info = codec_streaming_test_object_info(&fi);
let gate = codec_streaming_reader_gate_for_test(&None, &object_info, &fi, true);
assert_eq!(gate.object_class, GetCodecStreamingObjectClass::Multipart);
assert_eq!(
gate.decision,
GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Multipart)
);
},
);
}
#[test]
fn codec_streaming_reader_gate_limits_multipart_part_count() {
temp_env::with_vars(
[
(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("benchmark")),
(ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_MULTIPART_ENABLE, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_MULTIPART_MAX_PARTS, Some("1")),
(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")),
],
|| {
let fi = codec_streaming_test_fileinfo(1024, 2);
let object_info = codec_streaming_test_object_info(&fi);
let gate = codec_streaming_reader_gate_for_test(&None, &object_info, &fi, true);
assert_eq!(gate.object_class, GetCodecStreamingObjectClass::Multipart);
assert_eq!(
gate.decision,
GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::MultipartPartLimit)
);
},
);
}
#[test]
fn codec_streaming_decode_engine_builder_selects_rustfs() {
temp_env::with_var(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS), || {
@@ -3526,6 +3859,10 @@ mod tests {
fn codec_streaming_fallback_metric_labels_are_stable() {
assert_eq!(GetCodecStreamingFallbackReason::Disabled.as_str(), "disabled");
assert_eq!(GetCodecStreamingFallbackReason::RolloutNotOptedIn.as_str(), "rollout_not_opted_in");
assert_eq!(
GetCodecStreamingFallbackReason::RolloutPctNotSelected.as_str(),
"rollout_pct_not_selected"
);
assert_eq!(
GetCodecStreamingFallbackReason::BodyCompatibilityUnconfirmed.as_str(),
"body_compatibility_unconfirmed"
@@ -3546,6 +3883,7 @@ mod tests {
assert_eq!(GetCodecStreamingFallbackReason::Multipart.as_str(), "multipart");
assert_eq!(GetCodecStreamingFallbackReason::InvalidMinSize.as_str(), "invalid_min_size");
assert_eq!(GetCodecStreamingFallbackReason::ReadQuorumNotSafe.as_str(), "read_quorum_not_safe");
assert_eq!(GetCodecStreamingFallbackReason::MultipartPartLimit.as_str(), "multipart_part_limit");
assert_eq!(GetCodecStreamingObjectClass::PlainSinglePart.as_str(), "plain_single_part");
assert_eq!(GetCodecStreamingObjectClass::Range.as_str(), "range");
assert_eq!(GetCodecStreamingObjectClass::Encrypted.as_str(), "encrypted");
@@ -3569,7 +3907,7 @@ mod tests {
let object_info = codec_streaming_test_object_info(&fi);
assert_eq!(
get_codec_streaming_reader_gate(&None, &object_info, &fi, true).decision,
codec_streaming_reader_gate_for_test(&None, &object_info, &fi, true).decision,
GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Disabled)
);
},
@@ -3591,7 +3929,7 @@ mod tests {
let object_info = codec_streaming_test_object_info(&fi);
assert_eq!(
get_codec_streaming_reader_gate(&None, &object_info, &fi, true).decision,
codec_streaming_reader_gate_for_test(&None, &object_info, &fi, true).decision,
GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::RolloutNotOptedIn)
);
},
@@ -3610,7 +3948,7 @@ mod tests {
let object_info = codec_streaming_test_object_info(&fi);
assert_eq!(
get_codec_streaming_reader_gate(&None, &object_info, &fi, true).decision,
codec_streaming_reader_gate_for_test(&None, &object_info, &fi, true).decision,
GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::BodyCompatibilityUnconfirmed)
);
},
@@ -3629,13 +3967,56 @@ mod tests {
let object_info = codec_streaming_test_object_info(&fi);
assert_eq!(
get_codec_streaming_reader_gate(&None, &object_info, &fi, true).decision,
codec_streaming_reader_gate_for_test(&None, &object_info, &fi, true).decision,
GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::HeaderCompatibilityUnconfirmed)
);
},
);
}
#[test]
fn codec_streaming_reader_gate_honors_rollout_percentage() {
temp_env::with_vars(
[
(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("benchmark")),
(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT, Some("0")),
(ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")),
],
|| {
let fi = codec_streaming_test_fileinfo(1024, 1);
let object_info = codec_streaming_test_object_info(&fi);
assert_eq!(
codec_streaming_reader_gate_for_test(&None, &object_info, &fi, true).decision,
GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::RolloutPctNotSelected)
);
},
);
temp_env::with_vars(
[
(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("benchmark")),
(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT, Some("100")),
(ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")),
(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1")),
],
|| {
let fi = codec_streaming_test_fileinfo(1024, 1);
let object_info = codec_streaming_test_object_info(&fi);
assert_eq!(
codec_streaming_reader_gate_for_test(&None, &object_info, &fi, true).decision,
GetCodecStreamingDecision::Use
);
},
);
}
#[test]
fn codec_streaming_reader_gate_records_object_classes() {
temp_env::with_vars(
@@ -3650,7 +4031,7 @@ mod tests {
let fi = codec_streaming_test_fileinfo(1024, 1);
let object_info = codec_streaming_test_object_info(&fi);
assert_eq!(
get_codec_streaming_reader_gate(&None, &object_info, &fi, true).object_class,
codec_streaming_reader_gate_for_test(&None, &object_info, &fi, true).object_class,
GetCodecStreamingObjectClass::PlainSinglePart
);
@@ -3660,7 +4041,7 @@ mod tests {
end: 1,
});
assert_eq!(
get_codec_streaming_reader_gate(&range, &object_info, &fi, true).object_class,
codec_streaming_reader_gate_for_test(&range, &object_info, &fi, true).object_class,
GetCodecStreamingObjectClass::Range
);
},
crates/io-metrics/src/lib.rs
+38
View File
@@ -555,6 +555,21 @@ pub fn record_get_object_reconstruct_duration(path: &'static str, duration_secs:
record_get_object_stage_duration(path, "reconstruct", duration_secs);
}
/// Record the reconstruction outcome for a GetObject reader path.
#[inline(always)]
pub fn record_get_object_reconstruct_outcome(path: &'static str, engine: &'static str, outcome: &'static str) {
if !get_stage_metrics_enabled() {
return;
}
counter!(
"rustfs_io_get_object_reconstruct_outcome_total",
"path" => path,
"engine" => engine,
"outcome" => outcome
)
.increment(1);
}
/// Record GetObject emit duration.
#[inline(always)]
pub fn record_get_object_emit_duration(path: &'static str, duration_secs: f64) {
@@ -715,6 +730,24 @@ pub fn record_get_object_fill_queued(path: &'static str, policy: &'static str, q
histogram!("rustfs_io_get_object_fill_queued", "path" => path, "policy" => policy).record(usize_to_f64(queued));
}
/// Record that a background fill task was started for a GetObject reader path.
#[inline(always)]
pub fn record_get_object_fill_started(path: &'static str, policy: &'static str) {
if !get_stage_metrics_enabled() {
return;
}
counter!("rustfs_io_get_object_fill_started_total", "path" => path, "policy" => policy).increment(1);
}
/// Record that a persistent fill worker was started for a GetObject reader path.
#[inline(always)]
pub fn record_get_object_fill_worker_started(path: &'static str, policy: &'static str) {
if !get_stage_metrics_enabled() {
return;
}
counter!("rustfs_io_get_object_fill_worker_started_total", "path" => path, "policy" => policy).increment(1);
}
/// Record that a fill completed while the current output buffer still had unread bytes.
#[inline(always)]
pub fn record_get_object_fill_completed_before_output_drained(path: &'static str, policy: &'static str) {
@@ -1675,6 +1708,7 @@ mod tests {
record_get_object_first_shard_read_duration("codec_streaming", 0.004);
record_get_object_bitrot_verify_duration("codec_streaming", 0.005);
record_get_object_reconstruct_duration("codec_streaming", 0.006);
record_get_object_reconstruct_outcome("codec_streaming", "legacy", "skip_data_complete");
record_get_object_emit_duration("codec_streaming", 0.007);
record_get_object_first_byte_latency("s3_handler", 0.008);
record_get_object_full_body_latency("s3_handler", 0.009);
@@ -1696,6 +1730,8 @@ mod tests {
#[test]
fn test_record_get_object_fill_metrics() {
record_get_object_fill_queued("codec_streaming", "single_inflight", 1);
record_get_object_fill_started("codec_streaming", "single_inflight");
record_get_object_fill_worker_started("codec_streaming", "single_inflight");
record_get_object_fill_completed_before_output_drained("codec_streaming", "single_inflight");
record_get_object_fill_waited_by_output("codec_streaming", "single_inflight", 0.0003);
record_get_object_fill_cancelled_on_drop("codec_streaming", "single_inflight");
@@ -1770,6 +1806,7 @@ mod tests {
record_get_object_first_shard_read_duration("codec_streaming", 0.004);
record_get_object_bitrot_verify_duration("codec_streaming", 0.005);
record_get_object_reconstruct_duration("codec_streaming", 0.006);
record_get_object_reconstruct_outcome("codec_streaming", "legacy", "legacy_called");
record_get_object_emit_duration("codec_streaming", 0.007);
record_get_object_first_byte_latency("s3_handler", 0.008);
record_get_object_full_body_latency("s3_handler", 0.009);
@@ -1813,6 +1850,7 @@ mod tests {
record_get_object_first_shard_read_duration("codec_streaming", 0.004);
record_get_object_bitrot_verify_duration("codec_streaming", 0.005);
record_get_object_reconstruct_duration("codec_streaming", 0.006);
record_get_object_reconstruct_outcome("codec_streaming", "rustfs", "rustfs_called");
record_get_object_emit_duration("codec_streaming", 0.007);
record_get_object_first_byte_latency("s3_handler", 0.008);
record_get_object_full_body_latency("s3_handler", 0.009);
rustfs/src/admin/handlers/bucket_meta.rs
+15 -15
View File
@@ -379,9 +379,9 @@ impl Operation for ExportBucketMetadata {
.finish()
.map_err(|e| s3_error!(InternalError, "failed to finalize export archive: {e}"))?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/zip".parse().unwrap());
header.insert(CONTENT_DISPOSITION, "attachment; filename=bucket-meta.zip".parse().unwrap());
header.insert(CONTENT_LENGTH, zip_bytes.get_ref().len().to_string().parse().unwrap());
header.insert(CONTENT_TYPE, "application/zip".parse().expect("valid header value"));
header.insert(CONTENT_DISPOSITION, "attachment; filename=bucket-meta.zip".parse().expect("valid header value"));
header.insert(CONTENT_LENGTH, zip_bytes.get_ref().len().to_string().parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(zip_bytes.into_inner())), header))
}
}
@@ -597,7 +597,7 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = bucket_metadatas.get_mut(bucket_name).unwrap();
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
metadata.policy_config_json = content;
metadata.policy_config_updated_at = update_at;
}
@@ -617,7 +617,7 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = bucket_metadatas.get_mut(bucket_name).unwrap();
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
metadata.notification_config_xml = content;
metadata.notification_config_updated_at = update_at;
}
@@ -638,7 +638,7 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = bucket_metadatas.get_mut(bucket_name).unwrap();
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
metadata.lifecycle_config_xml = content;
metadata.lifecycle_config_updated_at = update_at;
}
@@ -659,7 +659,7 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = bucket_metadatas.get_mut(bucket_name).unwrap();
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
metadata.encryption_config_xml = content;
metadata.encryption_config_updated_at = update_at;
}
@@ -680,7 +680,7 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = bucket_metadatas.get_mut(bucket_name).unwrap();
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
metadata.tagging_config_xml = content;
metadata.tagging_config_updated_at = update_at;
}
@@ -701,7 +701,7 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = bucket_metadatas.get_mut(bucket_name).unwrap();
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
metadata.quota_config_json = content;
metadata.quota_config_updated_at = update_at;
}
@@ -722,7 +722,7 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = bucket_metadatas.get_mut(bucket_name).unwrap();
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
metadata.object_lock_config_xml = content;
metadata.object_lock_config_updated_at = update_at;
}
@@ -743,7 +743,7 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = bucket_metadatas.get_mut(bucket_name).unwrap();
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
metadata.versioning_config_xml = content;
metadata.versioning_config_updated_at = update_at;
}
@@ -764,7 +764,7 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = bucket_metadatas.get_mut(bucket_name).unwrap();
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
metadata.replication_config_xml = content;
metadata.replication_config_updated_at = update_at;
}
@@ -785,7 +785,7 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = bucket_metadatas.get_mut(bucket_name).unwrap();
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
metadata.bucket_targets_config_json = content;
metadata.bucket_targets_config_updated_at = update_at;
}
@@ -797,8 +797,8 @@ impl Operation for ImportBucketMetadata {
// TODO: site replication notify
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_LENGTH, "0".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
header.insert(CONTENT_LENGTH, "0".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
}
}
rustfs/src/admin/handlers/group.rs
+8 -8
View File
@@ -135,7 +135,7 @@ impl Operation for ListGroups {
let body = serde_json::to_vec(&groups).map_err(|e| s3_error!(InternalError, "failed to serialize response: {:?}", e))?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(body)), header))
}
@@ -201,7 +201,7 @@ impl Operation for GetGroup {
let body = serde_json::to_vec(&g).map_err(|e| s3_error!(InternalError, "failed to serialize response: {:?}", e))?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(body)), header))
}
@@ -317,8 +317,8 @@ impl Operation for DeleteGroup {
}
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_LENGTH, "0".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
header.insert(CONTENT_LENGTH, "0".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
}
}
@@ -469,8 +469,8 @@ impl Operation for SetGroupStatus {
}
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_LENGTH, "0".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
header.insert(CONTENT_LENGTH, "0".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
}
}
@@ -664,8 +664,8 @@ impl Operation for UpdateGroupMembers {
}
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_LENGTH, "0".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
header.insert(CONTENT_LENGTH, "0".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
}
}
rustfs/src/admin/handlers/policies.rs
+12 -12
View File
@@ -169,7 +169,7 @@ impl Operation for ListCannedPolicies {
let body = serde_json::to_vec(&kvs).map_err(|e| s3_error!(InternalError, "failed to serialize response: {:?}", e))?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(body)), header))
}
@@ -294,8 +294,8 @@ impl Operation for AddCannedPolicy {
}
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_LENGTH, "0".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
header.insert(CONTENT_LENGTH, "0".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
}
}
@@ -360,7 +360,7 @@ impl Operation for InfoCannedPolicy {
let body = serde_json::to_vec(&pd).map_err(|e| s3_error!(InternalError, "failed to serialize response: {:?}", e))?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(body)), header))
}
@@ -440,8 +440,8 @@ impl Operation for RemoveCannedPolicy {
}
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_LENGTH, "0".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
header.insert(CONTENT_LENGTH, "0".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
}
}
@@ -597,8 +597,8 @@ impl Operation for SetPolicyForUserOrGroup {
}
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_LENGTH, "0".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
header.insert(CONTENT_LENGTH, "0".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
}
}
@@ -932,8 +932,8 @@ async fn handle_builtin_policy_entities(req: S3Request<Body>) -> S3Result<S3Resp
let (body, content_type) = encode_compatible_admin_payload(&req_path, &cred.secret_key, body)?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, content_type.parse().unwrap());
header.insert(CONTENT_LENGTH, body.len().to_string().parse().unwrap());
header.insert(CONTENT_TYPE, content_type.parse().expect("valid header value"));
header.insert(CONTENT_LENGTH, body.len().to_string().parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(body)), header))
}
@@ -1095,8 +1095,8 @@ async fn handle_builtin_policy_association(req: S3Request<Body>, is_attach: bool
let (body, content_type) = encode_compatible_admin_payload(&req_path, &cred.secret_key, body)?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, content_type.parse().unwrap());
header.insert(CONTENT_LENGTH, body.len().to_string().parse().unwrap());
header.insert(CONTENT_TYPE, content_type.parse().expect("valid header value"));
header.insert(CONTENT_LENGTH, body.len().to_string().parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(body)), header))
}
rustfs/src/admin/handlers/service_account.rs
+10 -10
View File
@@ -441,7 +441,7 @@ impl Operation for AddServiceAccount {
let (body, content_type) = encode_compatible_admin_payload(req.uri.path(), &cred.secret_key, body)?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, content_type.parse().unwrap());
header.insert(CONTENT_TYPE, content_type.parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(body)), header))
}
@@ -616,8 +616,8 @@ impl Operation for UpdateServiceAccount {
}
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_LENGTH, "0".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
header.insert(CONTENT_LENGTH, "0".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::NO_CONTENT, Body::empty()), header))
}
}
@@ -688,7 +688,7 @@ impl Operation for InfoServiceAccount {
let (body, content_type) = encode_compatible_admin_payload(req.uri.path(), &cred.secret_key, body)?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, content_type.parse().unwrap());
header.insert(CONTENT_TYPE, content_type.parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(body)), header))
}
@@ -756,7 +756,7 @@ impl Operation for TemporaryAccountInfo {
let (body, content_type) = encode_compatible_admin_payload(req.uri.path(), &cred.secret_key, body)?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, content_type.parse().unwrap());
header.insert(CONTENT_TYPE, content_type.parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(body)), header))
}
@@ -854,7 +854,7 @@ impl Operation for InfoAccessKey {
let (body, content_type) = encode_compatible_admin_payload(req.uri.path(), &cred.secret_key, body)?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, content_type.parse().unwrap());
header.insert(CONTENT_TYPE, content_type.parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(body)), header))
}
@@ -980,7 +980,7 @@ impl Operation for ListServiceAccount {
let (data, content_type) = encode_compatible_admin_payload(req.uri.path(), &cred.secret_key, data)?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, content_type.parse().unwrap());
header.insert(CONTENT_TYPE, content_type.parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header))
}
@@ -1209,7 +1209,7 @@ impl Operation for ListAccessKeysBulk {
let (data, content_type) = encode_compatible_admin_payload(req.uri.path(), &cred.secret_key, data)?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, content_type.parse().unwrap());
header.insert(CONTENT_TYPE, content_type.parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header))
}
@@ -1339,8 +1339,8 @@ impl Operation for DeleteServiceAccount {
}
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_LENGTH, "0".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
header.insert(CONTENT_LENGTH, "0".parse().expect("valid header value"));
Ok(S3Response::with_headers(
(delete_service_account_success_status(req.uri.path()), Body::empty()),
header,
rustfs/src/admin/handlers/tier.rs
+21 -21
View File
@@ -220,31 +220,31 @@ impl Operation for AddTier {
match args.tier_type {
TierType::S3 => {
args.name = args.s3.clone().unwrap().name;
args.name = args.s3.clone().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing S3 configuration"))?.name;
}
TierType::RustFS => {
args.name = args.rustfs.clone().unwrap().name;
args.name = args.rustfs.clone().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing RustFS configuration"))?.name;
}
TierType::MinIO => {
args.name = args.minio.clone().unwrap().name;
args.name = args.minio.clone().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing MinIO configuration"))?.name;
}
TierType::Aliyun => {
args.name = args.aliyun.clone().unwrap().name;
args.name = args.aliyun.clone().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Aliyun configuration"))?.name;
}
TierType::Tencent => {
args.name = args.tencent.clone().unwrap().name;
args.name = args.tencent.clone().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Tencent configuration"))?.name;
}
TierType::Huaweicloud => {
args.name = args.huaweicloud.clone().unwrap().name;
args.name = args.huaweicloud.clone().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Huawei Cloud configuration"))?.name;
}
TierType::Azure => {
args.name = args.azure.clone().unwrap().name;
args.name = args.azure.clone().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Azure configuration"))?.name;
}
TierType::GCS => {
args.name = args.gcs.clone().unwrap().name;
args.name = args.gcs.clone().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing GCS configuration"))?.name;
}
TierType::R2 => {
args.name = args.r2.clone().unwrap().name;
args.name = args.r2.clone().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing R2 configuration"))?.name;
}
_ => (),
}
@@ -370,8 +370,8 @@ impl Operation for AddTier {
spawn_transition_tier_config_propagation("add");
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_LENGTH, "0".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
header.insert(CONTENT_LENGTH, "0".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
}
}
@@ -502,8 +502,8 @@ impl Operation for EditTier {
spawn_transition_tier_config_propagation("edit");
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_LENGTH, "0".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
header.insert(CONTENT_LENGTH, "0".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
}
}
@@ -553,7 +553,7 @@ impl Operation for ListTiers {
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("marshal tiers err {e}")))?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header))
}
@@ -673,8 +673,8 @@ impl Operation for RemoveTier {
spawn_transition_tier_config_propagation("remove");
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_LENGTH, "0".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
header.insert(CONTENT_LENGTH, "0".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
}
}
@@ -707,8 +707,8 @@ impl Operation for VerifyTier {
tier_config_mgr.verify(&tier).await.map_err(map_tier_verify_error)?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_LENGTH, "0".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
header.insert(CONTENT_LENGTH, "0".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
}
}
@@ -755,7 +755,7 @@ impl Operation for GetTierInfo {
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("marshal tier err {e}")))?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header))
}
@@ -923,8 +923,8 @@ impl Operation for ClearTier {
}
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_LENGTH, "0".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
header.insert(CONTENT_LENGTH, "0".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
}
}
rustfs/src/admin/handlers/user.rs
+12 -12
View File
@@ -306,8 +306,8 @@ impl Operation for AddUser {
}
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_LENGTH, "0".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
header.insert(CONTENT_LENGTH, "0".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
}
}
@@ -366,8 +366,8 @@ impl Operation for SetUserStatus {
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("failed to set user status: {e}")))?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_LENGTH, "0".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
header.insert(CONTENT_LENGTH, "0".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
}
}
@@ -431,7 +431,7 @@ impl Operation for ListUsers {
let (data, content_type) = encode_compatible_admin_payload(req.uri.path(), &cred.secret_key, data)?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, content_type.parse().unwrap());
header.insert(CONTENT_TYPE, content_type.parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header))
}
@@ -532,8 +532,8 @@ impl Operation for RemoveUser {
}
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_LENGTH, "0".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
header.insert(CONTENT_LENGTH, "0".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
}
}
@@ -606,7 +606,7 @@ impl Operation for GetUserInfo {
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("failed to serialize response: {e}")))?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header))
}
@@ -840,9 +840,9 @@ impl Operation for ExportIam {
.finish()
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/zip".parse().unwrap());
header.insert(CONTENT_DISPOSITION, "attachment; filename=iam-assets.zip".parse().unwrap());
header.insert(CONTENT_LENGTH, zip_bytes.get_ref().len().to_string().parse().unwrap());
header.insert(CONTENT_TYPE, "application/zip".parse().expect("valid header value"));
header.insert(CONTENT_DISPOSITION, "attachment; filename=iam-assets.zip".parse().expect("valid header value"));
header.insert(CONTENT_LENGTH, zip_bytes.get_ref().len().to_string().parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(zip_bytes.into_inner())), header))
}
}
@@ -1291,7 +1291,7 @@ impl Operation for ImportIam {
let body = serde_json::to_vec(&ret).map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(body)), header))
}
rustfs/src/app/lifecycle_transition_api_test.rs
+90
View File
@@ -61,6 +61,11 @@ use uuid::Uuid;
static GLOBAL_ENV: OnceLock<(Vec<PathBuf>, Arc<ECStore>)> = OnceLock::new();
static INIT: Once = Once::new();
const TRANSITION_WAIT_TIMEOUT: Duration = Duration::from_secs(15);
const ENV_GET_CODEC_STREAMING_ENABLE: &str = "RUSTFS_GET_CODEC_STREAMING_ENABLE";
const ENV_GET_CODEC_STREAMING_ROLLOUT: &str = "RUSTFS_GET_CODEC_STREAMING_ROLLOUT";
const ENV_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED: &str = "RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED";
const ENV_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED: &str = "RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED";
const ENV_GET_CODEC_STREAMING_MIN_SIZE: &str = "RUSTFS_GET_CODEC_STREAMING_MIN_SIZE";
fn init_tracing() {
INIT.call_once(|| {});
@@ -405,6 +410,31 @@ where
}
}
async fn with_get_codec_streaming_remote_probe_env<F, Fut>(test_fn: F)
where
F: FnOnce() -> Fut,
Fut: std::future::Future<Output = ()>,
{
let metrics_was_enabled = rustfs_io_metrics::get_stage_metrics_enabled();
rustfs_io_metrics::set_get_stage_metrics_enabled(true);
let result = std::panic::AssertUnwindSafe(temp_env::async_with_vars(
[
(ENV_GET_CODEC_STREAMING_ENABLE, Some("true")),
(ENV_GET_CODEC_STREAMING_ROLLOUT, Some("benchmark")),
(ENV_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")),
(ENV_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")),
(ENV_GET_CODEC_STREAMING_MIN_SIZE, Some("1")),
],
test_fn(),
))
.catch_unwind()
.await;
rustfs_io_metrics::set_get_stage_metrics_enabled(metrics_was_enabled);
if let Err(err) = result {
std::panic::resume_unwind(err);
}
}
async fn wait_for_remote_absence(backend: &MockWarmBackend, object: &str, timeout: Duration) -> bool {
let deadline = tokio::time::Instant::now() + timeout;
@@ -911,6 +941,66 @@ async fn complete_multipart_upload_transitions_immediately_via_usecase() {
assert!(backend.objects.lock().await.contains_key(&info.transitioned_object.name));
}
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "requires isolated global object layer state"]
async fn get_transitioned_object_uses_remote_codec_fallback_path() {
with_get_codec_streaming_remote_probe_env(|| async {
let (_disk_paths, ecstore) = setup_test_env().await;
let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
let backend = register_mock_tier(&tier_name).await;
let bucket = format!("test-api-get-remote-{}", &Uuid::new_v4().simple().to_string()[..8]);
let object = "test/remote-codec-fallback.txt";
let payload: Vec<u8> = (0..(1024 * 1024))
.map(|index| u8::try_from(index % 251).expect("payload byte fits in u8"))
.collect();
create_test_bucket(&ecstore, bucket.as_str()).await;
set_bucket_lifecycle_transition_with_tier(bucket.as_str(), &tier_name)
.await
.expect("Failed to set lifecycle configuration");
let uploaded = upload_test_object(&ecstore, bucket.as_str(), object, &payload).await;
let transition_opts = ObjectOptions {
transition: lifecycle::lifecycle_contract::TransitionOptions {
status: lifecycle::lifecycle_contract::TRANSITION_PENDING.to_string(),
tier: tier_name.clone(),
etag: uploaded.etag.clone().unwrap_or_default(),
..Default::default()
},
version_id: uploaded.version_id.map(|version| version.to_string()),
versioned: true,
mod_time: uploaded.mod_time,
..Default::default()
};
ecstore
.transition_object(bucket.as_str(), object, &transition_opts)
.await
.expect("Failed to transition object directly");
let transitioned = wait_for_transition(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT)
.await
.expect("object should transition before remote fallback GET");
assert_eq!(transitioned.transitioned_object.status, "complete");
assert_eq!(transitioned.transitioned_object.tier, tier_name);
assert!(!transitioned.transitioned_object.name.is_empty());
assert!(
backend
.objects
.lock()
.await
.contains_key(&transitioned.transitioned_object.name)
);
let actual = read_object_bytes(&ecstore, bucket.as_str(), object).await;
assert_eq!(actual, payload);
})
.await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "requires isolated global object layer state"]
rustfs/src/app/object_usecase.rs
+32 -3
View File
@@ -251,7 +251,8 @@ struct GetObjectBootstrap {
}
struct GetObjectIoPlanning<'a> {
_disk_permit: tokio::sync::SemaphorePermit<'a>,
/// `None` when inline fast path skips disk I/O semaphore.
_disk_permit: Option<tokio::sync::SemaphorePermit<'a>>,
permit_wait_duration: Duration,
queue_status: concurrency::IoQueueStatus,
queue_utilization: f64,
@@ -280,6 +281,8 @@ struct GetObjectReadSetup {
sse_customer_key_md5: Option<SSECustomerKeyMD5>,
ssekms_key_id: Option<SSEKMSKeyId>,
encryption_applied: bool,
/// `true` when the object was read via the inline data fast path (no disk I/O).
is_inline_fast_path: bool,
}
struct GetObjectPreparedRead<'a> {
@@ -2078,7 +2081,7 @@ impl DefaultObjectUsecase {
Self::ensure_get_object_not_timed_out(wrapper, timeout_config, bucket, key, GetObjectTimeoutStage::BeforeRead)?;
Ok(GetObjectIoPlanning {
_disk_permit: disk_permit,
_disk_permit: Some(disk_permit),
permit_wait_duration,
queue_status,
queue_utilization,
@@ -2148,7 +2151,8 @@ impl DefaultObjectUsecase {
part_number: Option<usize>,
) -> S3Result<GetObjectPreparedRead<'a>> {
let h = req.headers.clone();
let io_planning = Self::acquire_get_object_io_planning(manager, wrapper, timeout_config, bucket, key).await?;
// SF05: Store lookup first (cached via SF01 moka cache).
let store_lookup_start = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now);
let store = get_validated_store(bucket).await?;
if let Some(store_lookup_start) = store_lookup_start {
@@ -2159,6 +2163,8 @@ impl DefaultObjectUsecase {
);
}
// SF05: Read object metadata/data BEFORE acquiring disk I/O semaphore.
// ECStore's get_object_reader acquires its own RwLock — safe without the semaphore.
let read_start = std::time::Instant::now();
let read_stage_start = rustfs_io_metrics::get_stage_metrics_enabled().then_some(read_start);
let read_setup = Self::prepare_get_object_read(
@@ -2176,6 +2182,18 @@ impl DefaultObjectUsecase {
)
.await?;
// SF05: Skip disk I/O semaphore for inline fast path — data is already in memory.
let io_planning = if read_setup.is_inline_fast_path {
GetObjectIoPlanning {
_disk_permit: None,
permit_wait_duration: Duration::ZERO,
queue_status: concurrency::IoQueueStatus::default(),
queue_utilization: 0.0,
}
} else {
Self::acquire_get_object_io_planning(manager, wrapper, timeout_config, bucket, key).await?
};
Ok(GetObjectPreparedRead { io_planning, read_setup })
}
@@ -2310,6 +2328,16 @@ impl DefaultObjectUsecase {
None => (None, None, None, None, false, wrap_reader(reader.stream)),
};
// Detect inline fast path: data is in memory, no disk I/O semaphore needed.
// Conditions match the inline path in set_disk/mod.rs get_object_reader.
let is_inline_fast_path = info.inlined
&& info.size <= 128 * 1024
&& info.parts.len() <= 1
&& !info.is_encrypted()
&& !info.is_compressed()
&& info.transitioned_object.tier.is_empty()
&& rs.is_none();
Ok(GetObjectReadSetup {
info,
event_info,
@@ -2324,6 +2352,7 @@ impl DefaultObjectUsecase {
sse_customer_key_md5,
ssekms_key_id,
encryption_applied,
is_inline_fast_path,
})
}
#[allow(clippy::too_many_arguments)]
rustfs/src/app/storage_api.rs
+3 -1
View File
@@ -382,7 +382,9 @@ pub(crate) mod bucket {
pub(crate) const TRANSITION_COMPLETE: &str =
crate::storage::storage_api::ecstore_bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
#[cfg(test)]
pub(crate) const TRANSITION_PENDING: &str =
crate::storage::storage_api::ecstore_bucket::lifecycle::lifecycle::TRANSITION_PENDING;
pub(crate) fn expected_expiry_time(mod_time: time::OffsetDateTime, days: i32) -> time::OffsetDateTime {
crate::storage::storage_api::ecstore_bucket::lifecycle::lifecycle::expected_expiry_time(mod_time, days)
}
rustfs/src/server/layer.rs
+4 -4
View File
@@ -1435,7 +1435,7 @@ where
.unwrap());
}
let mut response = Response::builder().status(StatusCode::OK).body(ResBody::default()).unwrap();
let mut response = Response::builder().status(StatusCode::OK).body(ResBody::default()).expect("valid response body");
let cors_layer = ConditionalCorsLayer {
cors_origins: (*cors_origins).clone(),
};
@@ -1464,7 +1464,7 @@ where
let cors_allowed = cors_headers.contains_key(cors::response::ACCESS_CONTROL_ALLOW_ORIGIN);
let status = if cors_allowed { StatusCode::OK } else { StatusCode::FORBIDDEN };
let mut response = Response::builder().status(status).body(ResBody::default()).unwrap();
let mut response = Response::builder().status(status).body(ResBody::default()).expect("valid response body");
if cors_allowed {
for (key, value) in cors_headers.iter() {
response.headers_mut().insert(key, value.clone());
@@ -1474,7 +1474,7 @@ where
}
// No bucket-level CORS config: fall back to global/default CORS behavior.
let mut response = Response::builder().status(StatusCode::OK).body(ResBody::default()).unwrap();
let mut response = Response::builder().status(StatusCode::OK).body(ResBody::default()).expect("valid response body");
cors_layer.apply_cors_headers(&request_headers, response.headers_mut());
Ok(response)
});
@@ -1482,7 +1482,7 @@ where
let request_headers_clone = request_headers.clone();
return Box::pin(async move {
let mut response = Response::builder().status(StatusCode::OK).body(ResBody::default()).unwrap();
let mut response = Response::builder().status(StatusCode::OK).body(ResBody::default()).expect("valid response body");
let cors_layer = ConditionalCorsLayer {
cors_origins: (*cors_origins).clone(),
};
scripts/run_get_codec_streaming_smoke.sh
+1389 -9
View File
File diff suppressed because it is too large Load Diff
scripts/run_object_batch_bench_enhanced.sh
+45 -14
View File
@@ -220,8 +220,8 @@ setup_output() {
MEDIAN_CSV="$OUT_DIR/median_summary.csv"
COMPARE_CSV="$OUT_DIR/baseline_compare.csv"
echo "size,tool,round,attempt,concurrency,status,exit_code,round_started_at_utc,round_finished_at_utc,throughput_human,throughput_bps,reqps,latency_human,latency_ms,log_file" > "$ROUND_CSV"
echo "size,tool,concurrency,successful_rounds,failed_rounds,median_throughput_bps,median_reqps,median_latency_ms" > "$MEDIAN_CSV"
echo "size,tool,round,attempt,concurrency,status,exit_code,round_started_at_utc,round_finished_at_utc,throughput_human,throughput_bps,reqps,latency_human,latency_ms,log_file,req_p90_human,req_p90_ms,req_p99_human,req_p99_ms" > "$ROUND_CSV"
echo "size,tool,concurrency,successful_rounds,failed_rounds,median_throughput_bps,median_reqps,median_latency_ms,median_req_p90_ms,median_req_p99_ms" > "$MEDIAN_CSV"
}
trim() {
@@ -288,13 +288,32 @@ extract_first() {
rg -o "$regex" "$file" | head -n1 || true
}
normalize_duration_metric() {
local value="$1"
value="$(trim "${value:-N/A}")"
if [[ -z "$value" || "$value" == "N/A" ]]; then
echo "N/A"
return
fi
if [[ "$value" =~ ^([0-9]+(\.[0-9]+)?)(ms|us|µs|s)$ ]]; then
echo "${BASH_REMATCH[1]} ${BASH_REMATCH[3]}"
return
fi
echo "$value" | awk '{ if ($1 == "" || $1 == "N/A") print "N/A"; else print $1" "$2 }'
}
extract_metrics() {
local log_file="$1"
local throughput reqps latency
local throughput reqps latency req_p90 req_p99
throughput="$(extract_first '[0-9]+(\.[0-9]+)?[[:space:]]*(GiB/s|MiB/s|KiB/s|GB/s|MB/s|KB/s|B/s)' "$log_file")"
reqps="$(extract_first '[0-9]+(\.[0-9]+)?[[:space:]]*(obj/s|req/s|ops/s|requests/s)' "$log_file")"
latency="$(rg -o 'Reqs:[[:space:]]+Avg:[[:space:]]+[0-9]+(\.[0-9]+)?(ms|us|µs|s)' "$log_file" | head -n1 | sed -E 's/^Reqs:[[:space:]]+Avg:[[:space:]]+//')"
req_p90="$(rg -o '90%:[[:space:]]+[0-9]+(\.[0-9]+)?(ms|us|µs|s)' "$log_file" | head -n1 | sed -E 's/^90%:[[:space:]]+//')"
req_p99="$(rg -o '99%:[[:space:]]+[0-9]+(\.[0-9]+)?(ms|us|µs|s)' "$log_file" | head -n1 | sed -E 's/^99%:[[:space:]]+//')"
if [[ -z "$latency" ]]; then
latency="$(extract_first '[0-9]+(\.[0-9]+)?[[:space:]]*(ms|us|µs|s)' "$log_file")"
@@ -303,16 +322,16 @@ extract_metrics() {
throughput="$(trim "${throughput:-N/A}")"
reqps="$(trim "${reqps:-N/A}")"
latency="$(trim "${latency:-N/A}")"
req_p90="$(trim "${req_p90:-N/A}")"
req_p99="$(trim "${req_p99:-N/A}")"
# Normalize to "<num> <unit>" shape for downstream conversion helpers.
if [[ "$latency" =~ ^([0-9]+(\.[0-9]+)?)(ms|us|µs|s)$ ]]; then
latency="${BASH_REMATCH[1]} ${BASH_REMATCH[3]}"
else
latency="$(echo "$latency" | awk '{print $1" "$2}')"
fi
latency="$(normalize_duration_metric "$latency")"
req_p90="$(normalize_duration_metric "$req_p90")"
req_p99="$(normalize_duration_metric "$req_p99")"
reqps_num="$(echo "$reqps" | awk '{print $1}')"
echo "$throughput,${reqps_num:-N/A},$latency"
echo "$throughput,${reqps_num:-N/A},$latency,$req_p90,$req_p99"
}
median_from_numbers() {
@@ -414,20 +433,28 @@ run_one_attempt() {
fi
finished_at_utc="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
local metrics throughput_human reqps latency_human throughput_bps latency_ms
local metrics throughput_human reqps latency_human throughput_bps latency_ms req_p90_human req_p90_ms req_p99_human req_p99_ms
if [[ "$DRY_RUN" == "true" ]]; then
throughput_human="N/A"
reqps="N/A"
latency_human="N/A"
throughput_bps="N/A"
latency_ms="N/A"
req_p90_human="N/A"
req_p90_ms="N/A"
req_p99_human="N/A"
req_p99_ms="N/A"
else
metrics="$(extract_metrics "$log_file")"
throughput_human="$(echo "$metrics" | cut -d',' -f1)"
reqps="$(echo "$metrics" | cut -d',' -f2)"
latency_human="$(echo "$metrics" | cut -d',' -f3)"
req_p90_human="$(echo "$metrics" | cut -d',' -f4)"
req_p99_human="$(echo "$metrics" | cut -d',' -f5)"
throughput_bps="$(to_bps "$throughput_human")"
latency_ms="$(to_ms "$latency_human")"
req_p90_ms="$(to_ms "$req_p90_human")"
req_p99_ms="$(to_ms "$req_p99_human")"
fi
if [[ "$DRY_RUN" != "true" && "$status" == "ok" ]]; then
@@ -437,7 +464,7 @@ run_one_attempt() {
fi
fi
echo "$size,$TOOL,$round,$attempt,$CONCURRENCY,$status,$exit_code,$started_at_utc,$finished_at_utc,$throughput_human,$throughput_bps,$reqps,$latency_human,$latency_ms,$log_file" >> "$ROUND_CSV"
echo "$size,$TOOL,$round,$attempt,$CONCURRENCY,$status,$exit_code,$started_at_utc,$finished_at_utc,$throughput_human,$throughput_bps,$reqps,$latency_human,$latency_ms,$log_file,$req_p90_human,$req_p90_ms,$req_p99_human,$req_p99_ms" >> "$ROUND_CSV"
echo "$status"
}
@@ -481,20 +508,24 @@ build_median_summary() {
size="$(trim "$raw")"
[[ -z "$size" ]] && continue
local ok_rounds fail_rounds t_vals r_vals l_vals
local ok_rounds fail_rounds t_vals r_vals l_vals p90_vals p99_vals
ok_rounds="$(awk -F',' -v s="$size" 'NR>1 && $1==s && $6=="ok" {c++} END{print c+0}' "$ROUND_CSV")"
fail_rounds="$(awk -F',' -v s="$size" 'NR>1 && $1==s && $6!="ok" {c++} END{print c+0}' "$ROUND_CSV")"
t_vals="$(awk -F',' -v s="$size" 'NR>1 && $1==s && $6=="ok" && $11!="N/A" {print $11}' "$ROUND_CSV")"
r_vals="$(awk -F',' -v s="$size" 'NR>1 && $1==s && $6=="ok" && $12!="N/A" {print $12}' "$ROUND_CSV")"
l_vals="$(awk -F',' -v s="$size" 'NR>1 && $1==s && $6=="ok" && $14!="N/A" {print $14}' "$ROUND_CSV")"
p90_vals="$(awk -F',' -v s="$size" 'NR>1 && $1==s && $6=="ok" && $17!="N/A" {print $17}' "$ROUND_CSV")"
p99_vals="$(awk -F',' -v s="$size" 'NR>1 && $1==s && $6=="ok" && $19!="N/A" {print $19}' "$ROUND_CSV")"
local m_t m_r m_l
local m_t m_r m_l m_p90 m_p99
m_t="$(median_from_numbers "$t_vals")"
m_r="$(median_from_numbers "$r_vals")"
m_l="$(median_from_numbers "$l_vals")"
m_p90="$(median_from_numbers "$p90_vals")"
m_p99="$(median_from_numbers "$p99_vals")"
echo "$size,$TOOL,$CONCURRENCY,$ok_rounds,$fail_rounds,$m_t,$m_r,$m_l" >> "$MEDIAN_CSV"
echo "$size,$TOOL,$CONCURRENCY,$ok_rounds,$fail_rounds,$m_t,$m_r,$m_l,$m_p90,$m_p99" >> "$MEDIAN_CSV"
done
}