gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 08:18:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf: add S3 operations benchmark framework (#738) (#4005)

Browse Source
This commit is contained in:
Zhengchao An
2026-06-28 18:02:41 +08:00
committed by GitHub
parent 51acf2a99c
commit 710ae74cde
23 changed files with 625 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -9049,6 +9049,7 @@ dependencies = [
"chrono",
"clap",
"const-str",
"criterion",
"datafusion",
"flatbuffers",
"futures",
crates/ecstore/src/bucket/replication/replication_resyncer.rs
+22 -5
View File
@@ -617,7 +617,10 @@ impl ReplicationResyncer {
} else {
let state = TargetReplicationResyncStatus::new();
bucket_status.targets_map.insert(opts.arn.clone(), state);
bucket_status.targets_map.get_mut(&opts.arn).expect("ARN should be in targets map")
bucket_status
.targets_map
.get_mut(&opts.arn)
.expect("ARN should be in targets map")
};
if !resync_state_accepts_update(state, &opts) {
@@ -678,7 +681,10 @@ impl ReplicationResyncer {
} else {
let state = TargetReplicationResyncStatus::new();
bucket_status.targets_map.insert(opts.arn.clone(), state);
bucket_status.targets_map.get_mut(&opts.arn).expect("ARN should be in targets map")
bucket_status
.targets_map
.get_mut(&opts.arn)
.expect("ARN should be in targets map")
};
if !resync_state_accepts_update(state, &opts) {
@@ -2710,7 +2716,13 @@ 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_or_else(|_| "invalid-time".to_string()), 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
@@ -3473,8 +3485,13 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
&& self.op_type == ReplicationType::ExistingObject
&& !tgt_client.reset_id.is_empty()
{
rinfo.resync_timestamp =
format!("{};{}", OffsetDateTime::now_utc().format(&Rfc3339).unwrap_or_else(|_| "invalid-time".to_string()), 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.replication_resynced = true;
}
crates/ecstore/src/client/api_get_object_attributes.rs
+16 -11
View File
@@ -133,7 +133,8 @@ struct ObjectAttributePart {
impl ObjectAttributes {
pub async fn parse_response(&mut self, h: &HeaderMap, body_vec: Vec<u8>) -> Result<(), std::io::Error> {
let last_modified = h.get("Last-Modified")
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}")))?;
@@ -141,14 +142,14 @@ impl ObjectAttributes {
.map_err(|e| std::io::Error::other(format!("invalid Last-Modified date: {e}")))?;
self.last_modified = mod_time;
let version_id = h.get(X_AMZ_VERSION_ID)
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 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) => {
@@ -175,7 +176,10 @@ impl TransitionClient {
}
let mut headers = HeaderMap::new();
headers.insert(X_AMZ_OBJECT_ATTRIBUTES, HeaderValue::from_str(GET_OBJECT_ATTRIBUTES_TAGS).expect("valid header value"));
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(
@@ -185,7 +189,10 @@ impl TransitionClient {
}
if opts.max_parts > 0 {
headers.insert(X_AMZ_MAX_PARTS, HeaderValue::from_str(&opts.max_parts.to_string()).expect("valid header value"));
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,
@@ -222,9 +229,7 @@ impl TransitionClient {
let resp_status = resp.status();
let h = resp.headers().clone();
let has_etag = h.get("ETag")
.and_then(|v| v.to_str().ok())
.unwrap_or("");
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",
@@ -241,8 +246,8 @@ impl TransitionClient {
}
if resp_status != http::StatusCode::OK {
let err_body = String::from_utf8(body_vec)
.map_err(|e| std::io::Error::other(format!("invalid UTF-8 error body: {e}")))?;
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/client/api_stat.rs
+4 -1
View File
@@ -153,7 +153,10 @@ impl TransitionClient {
headers.insert("X-Source-DeleteMarker", HeaderValue::from_str("true").expect("operation should succeed"));
}
if opts.internal.is_replication_ready_for_delete_marker {
headers.insert("X-Check-Replication-Ready", HeaderValue::from_str("true").expect("operation should succeed"));
headers.insert(
"X-Check-Replication-Ready",
HeaderValue::from_str("true").expect("operation should succeed"),
);
}
let resp = self
crates/ecstore/src/cluster/rpc/remote_disk.rs
+50 -15
View File
@@ -2561,7 +2561,9 @@ mod tests {
health_check: false,
};
RemoteDisk::new(&endpoint, &disk_option, data_transport).await.expect("operation should succeed")
RemoteDisk::new(&endpoint, &disk_option, data_transport)
.await
.expect("operation should succeed")
}
#[derive(Debug)]
@@ -2697,7 +2699,8 @@ mod tests {
};
let addr = listener.local_addr().expect("listener local address should be available");
let url = url::Url::parse(&format!("http://{}:{}/data/rustfs0", addr.ip(), addr.port())).expect("operation should succeed");
let url =
url::Url::parse(&format!("http://{}:{}/data/rustfs0", addr.ip(), addr.port())).expect("operation should succeed");
let endpoint = Endpoint {
url,
is_local: false,
@@ -2808,7 +2811,9 @@ mod tests {
health.mark_failure(&endpoint, "test_failure");
health.mark_failure(&endpoint, "test_failure");
assert_eq!(health.runtime_state(), RuntimeDriveHealthState::Offline);
let channel = TonicEndpoint::from_shared(base_addr.clone()).expect("operation should succeed").connect_lazy();
let channel = TonicEndpoint::from_shared(base_addr.clone())
.expect("operation should succeed")
.connect_lazy();
runtime_sources::cache_test_node_channel(base_addr.clone(), channel).await;
assert!(runtime_sources::test_node_channel_is_cached(&base_addr).await);
@@ -2854,7 +2859,9 @@ mod tests {
let copy_task = tokio::spawn(async move {
let mut cursor = Cursor::new(payload);
copy_stream_with_buffer(&mut cursor, &mut write_half, 4 * 1024).await.expect("operation should succeed");
copy_stream_with_buffer(&mut cursor, &mut write_half, 4 * 1024)
.await
.expect("operation should succeed");
});
let mut copied = Vec::new();
@@ -2890,7 +2897,10 @@ mod tests {
// Set a disk ID
let test_id = Uuid::new_v4();
remote_disk.set_disk_id(Some(test_id)).await.expect("operation should succeed");
remote_disk
.set_disk_id(Some(test_id))
.await
.expect("operation should succeed");
// Verify the disk ID was set
let retrieved_id = remote_disk.get_disk_id().await.expect("operation should succeed");
@@ -2923,7 +2933,10 @@ mod tests {
assert_eq!(remote_disk.disk_ref().await, endpoint.to_string());
let disk_id = Uuid::new_v4();
remote_disk.set_disk_id(Some(disk_id)).await.expect("operation should succeed");
remote_disk
.set_disk_id(Some(disk_id))
.await
.expect("operation should succeed");
assert_eq!(remote_disk.disk_ref().await, disk_id.to_string());
}
@@ -2935,7 +2948,10 @@ mod tests {
let remote_disk = new_remote_disk_with_transport(Arc::new(transport.clone())).await;
let expected_disk = remote_disk.disk_ref().await;
let _reader = remote_disk.read_file_stream("bucket", "object/part.1", 7, 11).await.expect("operation should succeed");
let _reader = remote_disk
.read_file_stream("bucket", "object/part.1", 7, 11)
.await
.expect("operation should succeed");
let calls = transport.calls();
assert_eq!(calls.len(), 1);
@@ -2961,7 +2977,10 @@ mod tests {
let transport = RecordingInternodeDataTransport::default();
let remote_disk = new_remote_disk_with_transport(Arc::new(transport.clone())).await;
let _reader = remote_disk.read_file_stream("bucket", "object/part.1", 7, 11).await.expect("operation should succeed");
let _reader = remote_disk
.read_file_stream("bucket", "object/part.1", 7, 11)
.await
.expect("operation should succeed");
let calls = transport.calls();
assert_eq!(calls.len(), 1);
@@ -2983,7 +3002,10 @@ mod tests {
.create_file("orig-bucket", "bucket", "object/part.1", 4096)
.await
.expect("operation should succeed");
let _appended = remote_disk.append_file("bucket", "object/part.2").await.expect("operation should succeed");
let _appended = remote_disk
.append_file("bucket", "object/part.2")
.await
.expect("operation should succeed");
let calls = transport.calls();
assert_eq!(calls.len(), 2);
@@ -3073,7 +3095,10 @@ mod tests {
let expected_body = serde_json::to_vec(&opts).expect("operation should succeed");
let mut writer = Vec::new();
remote_disk.walk_dir(opts, &mut writer).await.expect("operation should succeed");
remote_disk
.walk_dir(opts, &mut writer)
.await
.expect("operation should succeed");
let calls = transport.calls();
assert_eq!(calls.len(), 1);
@@ -3429,7 +3454,9 @@ mod tests {
.await
.expect("operation should succeed");
let channel = TonicEndpoint::from_shared(addr.clone()).expect("operation should succeed").connect_lazy();
let channel = TonicEndpoint::from_shared(addr.clone())
.expect("operation should succeed")
.connect_lazy();
runtime_sources::cache_test_node_channel(addr.clone(), channel).await;
assert!(runtime_sources::test_node_channel_is_cached(&addr).await);
@@ -3473,7 +3500,9 @@ mod tests {
.await
.expect("operation should succeed");
let channel = TonicEndpoint::from_shared(addr.clone()).expect("operation should succeed").connect_lazy();
let channel = TonicEndpoint::from_shared(addr.clone())
.expect("operation should succeed")
.connect_lazy();
runtime_sources::cache_test_node_channel(addr.clone(), channel).await;
let err = remote_disk
@@ -3529,7 +3558,9 @@ mod tests {
.await
.expect("operation should succeed");
let channel = TonicEndpoint::from_shared(addr.clone()).expect("operation should succeed").connect_lazy();
let channel = TonicEndpoint::from_shared(addr.clone())
.expect("operation should succeed")
.connect_lazy();
runtime_sources::cache_test_node_channel(addr.clone(), channel).await;
let err = remote_disk
@@ -3590,7 +3621,9 @@ mod tests {
.await
.expect("operation should succeed");
let channel = TonicEndpoint::from_shared(addr.clone()).expect("operation should succeed").connect_lazy();
let channel = TonicEndpoint::from_shared(addr.clone())
.expect("operation should succeed")
.connect_lazy();
runtime_sources::cache_test_node_channel(addr.clone(), channel).await;
let err = remote_disk
@@ -3643,7 +3676,9 @@ mod tests {
.await
.expect("operation should succeed");
let channel = TonicEndpoint::from_shared(addr.clone()).expect("operation should succeed").connect_lazy();
let channel = TonicEndpoint::from_shared(addr.clone())
.expect("operation should succeed")
.connect_lazy();
runtime_sources::cache_test_node_channel(addr.clone(), channel).await;
let err = remote_disk
crates/ecstore/src/disk/local.rs
+122 -44
View File
@@ -3884,7 +3884,8 @@ mod test {
use tempfile::tempdir;
let dir = tempdir().expect("operation should succeed");
let mut endpoint = Endpoint::try_from(dir.path().to_str().expect("operation should succeed")).expect("operation should succeed");
let mut endpoint =
Endpoint::try_from(dir.path().to_str().expect("operation should succeed")).expect("operation should succeed");
endpoint.set_pool_index(0);
endpoint.set_set_index(0);
endpoint.set_disk_index(0);
@@ -3930,7 +3931,9 @@ mod test {
let dir = tempdir().expect("operation should succeed");
let tmp = LocalDisk::meta_path(dir.path(), RUSTFS_META_TMP_BUCKET);
let leftover = tmp.join("leftover").join("data");
fs::create_dir_all(leftover.parent().expect("operation should succeed")).await.expect("operation should succeed");
fs::create_dir_all(leftover.parent().expect("operation should succeed"))
.await
.expect("operation should succeed");
fs::write(&leftover, b"temporary").await.expect("operation should succeed");
LocalDisk::cleanup_tmp_on_startup(dir.path(), Arc::new(AtomicU32::new(0)), Arc::new(Notify::new()))
@@ -3949,7 +3952,9 @@ mod test {
let tmp = LocalDisk::meta_path(dir.path(), RUSTFS_META_TMP_BUCKET);
let stale = tmp.join("stale").join("data");
let trash = LocalDisk::meta_path(dir.path(), RUSTFS_META_TMP_DELETED_BUCKET);
fs::create_dir_all(stale.parent().expect("operation should succeed")).await.expect("operation should succeed");
fs::create_dir_all(stale.parent().expect("operation should succeed"))
.await
.expect("operation should succeed");
fs::create_dir_all(&trash).await.expect("operation should succeed");
fs::write(&stale, b"temporary").await.expect("operation should succeed");
@@ -3975,7 +3980,9 @@ mod test {
let regular_file = tmp.join("note.txt");
let trash = LocalDisk::meta_path(dir.path(), RUSTFS_META_TMP_DELETED_BUCKET);
fs::create_dir_all(fresh_dir.parent().expect("operation should succeed")).await.expect("operation should succeed");
fs::create_dir_all(fresh_dir.parent().expect("operation should succeed"))
.await
.expect("operation should succeed");
fs::create_dir_all(&trash).await.expect("operation should succeed");
fs::write(&fresh_dir, b"temporary").await.expect("operation should succeed");
fs::write(&regular_file, b"keep").await.expect("operation should succeed");
@@ -4048,16 +4055,31 @@ mod test {
let bucket = "test-bucket";
let bucket_dir = dir.path().join(bucket);
fs::create_dir_all(bucket_dir.join("foo/bar/xyzzy")).await.expect("operation should succeed");
fs::create_dir_all(bucket_dir.join("quux/thud")).await.expect("operation should succeed");
fs::create_dir_all(bucket_dir.join("asdf")).await.expect("operation should succeed");
fs::create_dir_all(bucket_dir.join("foo/bar/xyzzy"))
.await
.expect("operation should succeed");
fs::create_dir_all(bucket_dir.join("quux/thud"))
.await
.expect("operation should succeed");
fs::create_dir_all(bucket_dir.join("asdf"))
.await
.expect("operation should succeed");
fs::write(bucket_dir.join("foo/bar/xl.meta"), b"meta").await.expect("operation should succeed");
fs::write(bucket_dir.join("foo/bar/xyzzy/xl.meta"), b"meta").await.expect("operation should succeed");
fs::write(bucket_dir.join("quux/thud/xl.meta"), b"meta").await.expect("operation should succeed");
fs::write(bucket_dir.join("asdf/xl.meta"), b"meta").await.expect("operation should succeed");
fs::write(bucket_dir.join("foo/bar/xl.meta"), b"meta")
.await
.expect("operation should succeed");
fs::write(bucket_dir.join("foo/bar/xyzzy/xl.meta"), b"meta")
.await
.expect("operation should succeed");
fs::write(bucket_dir.join("quux/thud/xl.meta"), b"meta")
.await
.expect("operation should succeed");
fs::write(bucket_dir.join("asdf/xl.meta"), b"meta")
.await
.expect("operation should succeed");
let endpoint = Endpoint::try_from(dir.path().to_str().expect("operation should succeed")).expect("operation should succeed");
let endpoint =
Endpoint::try_from(dir.path().to_str().expect("operation should succeed")).expect("operation should succeed");
let disk = LocalDisk::new(&endpoint, false).await.expect("operation should succeed");
let (reader, mut writer) = tokio::io::duplex(4096);
@@ -4098,13 +4120,19 @@ mod test {
let bucket = "test-bucket";
let bucket_dir = dir.path().join(bucket);
fs::create_dir_all(bucket_dir.join("marker/file.txt")).await.expect("operation should succeed");
fs::create_dir_all(bucket_dir.join("marker/subdir/file.txt")).await.expect("operation should succeed");
fs::create_dir_all(bucket_dir.join("marker/file.txt"))
.await
.expect("operation should succeed");
fs::create_dir_all(bucket_dir.join("marker/subdir/file.txt"))
.await
.expect("operation should succeed");
fs::create_dir_all(bucket_dir.join(format!("marker/subdir{GLOBAL_DIR_SUFFIX}")))
.await
.expect("operation should succeed");
fs::write(bucket_dir.join("marker/file.txt/xl.meta"), b"meta").await.expect("operation should succeed");
fs::write(bucket_dir.join("marker/file.txt/xl.meta"), b"meta")
.await
.expect("operation should succeed");
fs::write(bucket_dir.join("marker/subdir/file.txt/xl.meta"), b"meta")
.await
.expect("operation should succeed");
@@ -4112,7 +4140,8 @@ mod test {
.await
.expect("operation should succeed");
let endpoint = Endpoint::try_from(dir.path().to_str().expect("operation should succeed")).expect("operation should succeed");
let endpoint =
Endpoint::try_from(dir.path().to_str().expect("operation should succeed")).expect("operation should succeed");
let disk = LocalDisk::new(&endpoint, false).await.expect("operation should succeed");
let (reader, mut writer) = tokio::io::duplex(4096);
@@ -4167,10 +4196,13 @@ mod test {
] {
let object_dir = bucket_dir.join(name);
fs::create_dir_all(&object_dir).await.expect("operation should succeed");
fs::write(object_dir.join(STORAGE_FORMAT_FILE), b"meta").await.expect("operation should succeed");
fs::write(object_dir.join(STORAGE_FORMAT_FILE), b"meta")
.await
.expect("operation should succeed");
}
let endpoint = Endpoint::try_from(dir.path().to_str().expect("operation should succeed")).expect("operation should succeed");
let endpoint =
Endpoint::try_from(dir.path().to_str().expect("operation should succeed")).expect("operation should succeed");
let disk = LocalDisk::new(&endpoint, false).await.expect("operation should succeed");
async fn scan_names(disk: &LocalDisk, bucket: &str, base_dir: &str, forward_to: &str) -> (Vec<String>, i32) {
@@ -4308,7 +4340,9 @@ mod test {
}
let hidden_versioned_dir = bucket_dir.join("shard/aaa-trash-0003");
fs::create_dir_all(&hidden_versioned_dir).await.expect("operation should succeed");
fs::create_dir_all(&hidden_versioned_dir)
.await
.expect("operation should succeed");
fs::write(
hidden_versioned_dir.join(STORAGE_FORMAT_FILE),
delete_marker_with_old_object_metadata(
@@ -4328,7 +4362,8 @@ mod test {
.await
.expect("operation should succeed");
let endpoint = Endpoint::try_from(dir.path().to_str().expect("operation should succeed")).expect("operation should succeed");
let endpoint =
Endpoint::try_from(dir.path().to_str().expect("operation should succeed")).expect("operation should succeed");
let disk = LocalDisk::new(&endpoint, false).await.expect("operation should succeed");
let (reader, mut writer) = tokio::io::duplex(4096);
@@ -4377,14 +4412,22 @@ mod test {
fs::create_dir_all(&object_dir).await.expect("operation should succeed");
fs::write(&meta_path, b"meta").await.expect("operation should succeed");
let original_permissions = fs::metadata(&meta_path).await.expect("operation should succeed").permissions();
fs::set_permissions(&meta_path, Permissions::from_mode(0o000)).await.expect("operation should succeed");
let original_permissions = fs::metadata(&meta_path)
.await
.expect("operation should succeed")
.permissions();
fs::set_permissions(&meta_path, Permissions::from_mode(0o000))
.await
.expect("operation should succeed");
if fs::File::open(&meta_path).await.is_ok() {
fs::set_permissions(&meta_path, original_permissions).await.expect("operation should succeed");
fs::set_permissions(&meta_path, original_permissions)
.await
.expect("operation should succeed");
return;
}
let endpoint = Endpoint::try_from(dir.path().to_str().expect("operation should succeed")).expect("operation should succeed");
let endpoint =
Endpoint::try_from(dir.path().to_str().expect("operation should succeed")).expect("operation should succeed");
let disk = LocalDisk::new(&endpoint, false).await.expect("operation should succeed");
let (_reader, mut writer) = tokio::io::duplex(4096);
@@ -4401,7 +4444,9 @@ mod test {
.scan_dir("".to_string(), "".to_string(), &opts, &mut out, &mut objs_returned, false, None)
.await;
fs::set_permissions(&meta_path, original_permissions).await.expect("operation should succeed");
fs::set_permissions(&meta_path, original_permissions)
.await
.expect("operation should succeed");
assert!(matches!(result, Err(DiskError::FileAccessDenied)));
}
@@ -4437,31 +4482,48 @@ mod test {
fs::create_dir_all(&multipart_base).await.expect("operation should succeed");
for uuid in &[UUID_MULTIPART_1, UUID_MULTIPART_2] {
fs::create_dir_all(multipart_base.join(uuid)).await.expect("operation should succeed");
fs::write(multipart_base.join(uuid).join("part.1"), b"part").await.expect("operation should succeed");
fs::create_dir_all(multipart_base.join(uuid))
.await
.expect("operation should succeed");
fs::write(multipart_base.join(uuid).join("part.1"), b"part")
.await
.expect("operation should succeed");
}
fs::create_dir_all(obj_base.join(UUID_OBJ)).await.expect("operation should succeed");
fs::write(obj_base.join(UUID_OBJ).join("part.1"), b"part").await.expect("operation should succeed");
fs::create_dir_all(obj_base.join(UUID_OBJ))
.await
.expect("operation should succeed");
fs::write(obj_base.join(UUID_OBJ).join("part.1"), b"part")
.await
.expect("operation should succeed");
fs::create_dir_all(&dir_in_multipart_base).await.expect("operation should succeed");
fs::create_dir_all(&dir_in_multipart_base)
.await
.expect("operation should succeed");
fs::write(dir_in_multipart_base.join(STORAGE_FORMAT_FILE), b"meta")
.await
.expect("operation should succeed");
let mut fm = FileMeta::default();
fm.add_version(create_file_info(VER_ID_1, UUID_MULTIPART_1)).expect("operation should succeed");
fm.add_version(create_file_info(VER_ID_2, UUID_MULTIPART_2)).expect("operation should succeed");
fs::write(multipart_base.join(STORAGE_FORMAT_FILE), fm.marshal_msg().expect("operation should succeed"))
.await
fm.add_version(create_file_info(VER_ID_1, UUID_MULTIPART_1))
.expect("operation should succeed");
fm.add_version(create_file_info(VER_ID_2, UUID_MULTIPART_2))
.expect("operation should succeed");
fs::write(
multipart_base.join(STORAGE_FORMAT_FILE),
fm.marshal_msg().expect("operation should succeed"),
)
.await
.expect("operation should succeed");
let mut fm = FileMeta::default();
fm.add_version(create_file_info(VER_ID_3, UUID_OBJ)).expect("operation should succeed");
fm.add_version(create_file_info(VER_ID_3, UUID_OBJ))
.expect("operation should succeed");
fs::write(obj_base.join(STORAGE_FORMAT_FILE), fm.marshal_msg().expect("operation should succeed"))
.await
.expect("operation should succeed");
let endpoint = Endpoint::try_from(dir.path().to_str().expect("operation should succeed")).expect("operation should succeed");
let endpoint =
Endpoint::try_from(dir.path().to_str().expect("operation should succeed")).expect("operation should succeed");
let disk = LocalDisk::new(&endpoint, false).await.expect("operation should succeed");
let (reader, mut writer) = tokio::io::duplex(4096);
@@ -4477,7 +4539,10 @@ mod test {
)
.await
.expect("operation should succeed");
MetacacheWriter::new(&mut writer).close().await.expect("operation should succeed");
MetacacheWriter::new(&mut writer)
.close()
.await
.expect("operation should succeed");
let mut reader = MetacacheReader::new(reader);
let entries = reader.read_all().await.expect("operation should succeed");
@@ -4570,7 +4635,9 @@ mod test {
let disk = LocalDisk::new(&ep, false).await.expect("operation should succeed");
let tmpp = disk.resolve_abs_path(Path::new(RUSTFS_META_TMP_DELETED_BUCKET)).expect("operation should succeed");
let tmpp = disk
.resolve_abs_path(Path::new(RUSTFS_META_TMP_DELETED_BUCKET))
.expect("operation should succeed");
println!("ppp :{:?}", &tmpp);
@@ -4598,7 +4665,9 @@ mod test {
let disk = LocalDisk::new(&ep, false).await.expect("operation should succeed");
let tmpp = disk.resolve_abs_path(Path::new(RUSTFS_META_TMP_DELETED_BUCKET)).expect("operation should succeed");
let tmpp = disk
.resolve_abs_path(Path::new(RUSTFS_META_TMP_DELETED_BUCKET))
.expect("operation should succeed");
println!("ppp :{:?}", &tmpp);
@@ -4634,7 +4703,9 @@ mod test {
assert!(bucket_path.to_string_lossy().contains("test-bucket"));
// Test object path
let object_path = disk.get_object_path("test-bucket", "test-object").expect("operation should succeed");
let object_path = disk
.get_object_path("test-bucket", "test-object")
.expect("operation should succeed");
assert!(object_path.to_string_lossy().contains("test-bucket"));
assert!(object_path.to_string_lossy().contains("test-object"));
@@ -4695,7 +4766,10 @@ mod test {
.await
.expect("operation should succeed");
let read_data = disk.read_all("test-volume", "test-file.txt").await.expect("operation should succeed");
let read_data = disk
.read_all("test-volume", "test-file.txt")
.await
.expect("operation should succeed");
assert_eq!(read_data, test_data);
// Test file deletion
@@ -4705,7 +4779,9 @@ mod test {
undo_write: false,
old_data_dir: None,
};
disk.delete("test-volume", "test-file.txt", delete_opts).await.expect("operation should succeed");
disk.delete("test-volume", "test-file.txt", delete_opts)
.await
.expect("operation should succeed");
// Clean up
disk.delete_volume("test-volume").await.expect("operation should succeed");
@@ -4781,7 +4857,8 @@ mod test {
use tempfile::tempdir;
let dir = tempdir().expect("operation should succeed");
let endpoint = Endpoint::try_from(dir.path().to_str().expect("operation should succeed")).expect("operation should succeed");
let endpoint =
Endpoint::try_from(dir.path().to_str().expect("operation should succeed")).expect("operation should succeed");
let disk = LocalDisk::new(&endpoint, false).await.expect("operation should succeed");
disk.make_volume("test-volume").await.expect("operation should succeed");
@@ -4798,7 +4875,8 @@ mod test {
use tempfile::tempdir;
let dir = tempdir().expect("operation should succeed");
let endpoint = Endpoint::try_from(dir.path().to_str().expect("operation should succeed")).expect("operation should succeed");
let endpoint =
Endpoint::try_from(dir.path().to_str().expect("operation should succeed")).expect("operation should succeed");
let disk = LocalDisk::new(&endpoint, false).await.expect("operation should succeed");
disk.make_volume("test-volume").await.expect("operation should succeed");
crates/ecstore/src/metadata/set_disk.rs
+5 -1
View File
@@ -429,7 +429,11 @@ impl SetDisks {
"find_file_info_in_quorum: inspecting meta"
);
let etag_only = mod_time.is_none() && etag.is_some() && meta.get_etag().is_some_and(|v| &v == etag.as_ref().expect("operation should succeed"));
let etag_only = mod_time.is_none()
&& etag.is_some()
&& meta
.get_etag()
.is_some_and(|v| &v == etag.as_ref().expect("operation should succeed"));
let mod_valid = mod_time == &meta.mod_time;
if etag_only || mod_valid {
crates/ecstore/src/set_disk/heal.rs
+2 -1
View File
@@ -549,7 +549,8 @@ impl SetDisks {
} else {
rename_successes += 1;
if parts_metadata[index].is_remote() {
let rm_data_dir = parts_metadata[index].data_dir.expect("operation should succeed").to_string();
let rm_data_dir =
parts_metadata[index].data_dir.expect("operation should succeed").to_string();
let d_path = Path::new(&encode_dir_object(object)).join(rm_data_dir);
crates/ecstore/src/store/rebalance.rs
+4 -3
View File
@@ -659,8 +659,8 @@ mod tests {
},
];
let (info, idx) =
resolve_latest_object_info_candidates(candidates, "bucket", "object", &ObjectOptions::default()).expect("operation should succeed");
let (info, idx) = resolve_latest_object_info_candidates(candidates, "bucket", "object", &ObjectOptions::default())
.expect("operation should succeed");
assert_eq!(idx, 1);
assert!(info.delete_marker);
@@ -681,7 +681,8 @@ mod tests {
},
];
let (_, idx) = resolve_latest_object_info_candidates(candidates, "bucket", "object", &ObjectOptions::default()).expect("operation should succeed");
let (_, idx) = resolve_latest_object_info_candidates(candidates, "bucket", "object", &ObjectOptions::default())
.expect("operation should succeed");
assert_eq!(idx, 1);
}
crates/protocols/src/swift/expiration_worker.rs
+7 -2
View File
@@ -342,7 +342,10 @@ impl ExpirationWorker {
metrics: &Arc<RwLock<ExpirationMetrics>>,
) -> SwiftResult<()> {
let start_time = SystemTime::now();
let now = start_time.duration_since(UNIX_EPOCH).expect("operation should succeed").as_secs();
let now = start_time
.duration_since(UNIX_EPOCH)
.expect("operation should succeed")
.as_secs();
debug!(
event = EVENT_SWIFT_EXPIRATION_ITERATION_SUMMARY,
@@ -452,7 +455,9 @@ impl ExpirationWorker {
}
// Update metrics
let duration = SystemTime::now().duration_since(start_time).expect("operation should succeed");
let duration = SystemTime::now()
.duration_since(start_time)
.expect("operation should succeed");
let mut m = metrics.write().await;
m.objects_scanned += scanned_count;
m.objects_deleted += deleted_count;
crates/rio/src/encrypt_reader.rs
+72 -18
View File
@@ -636,7 +636,10 @@ mod tests {
let reader = BufReader::new(Cursor::new(data.to_vec()));
let mut encrypt_reader = EncryptReader::new(reader, key, nonce);
let mut encrypted = Vec::new();
encrypt_reader.read_to_end(&mut encrypted).await.expect("operation should succeed");
encrypt_reader
.read_to_end(&mut encrypted)
.await
.expect("operation should succeed");
encrypted
}
@@ -674,14 +677,20 @@ mod tests {
// Encrypt
let mut encrypt_reader = encrypt_reader;
let mut encrypted = Vec::new();
encrypt_reader.read_to_end(&mut encrypted).await.expect("operation should succeed");
encrypt_reader
.read_to_end(&mut encrypted)
.await
.expect("operation should succeed");
// Decrypt using DecryptReader
let reader = Cursor::new(encrypted.clone());
let decrypt_reader = DecryptReader::new(reader, key, nonce);
let mut decrypt_reader = decrypt_reader;
let mut decrypted = Vec::new();
decrypt_reader.read_to_end(&mut decrypted).await.expect("operation should succeed");
decrypt_reader
.read_to_end(&mut decrypted)
.await
.expect("operation should succeed");
assert_eq!(&decrypted, data);
}
@@ -700,7 +709,10 @@ mod tests {
let encrypt_reader = EncryptReader::new(reader, key, nonce);
let mut encrypt_reader = encrypt_reader;
let mut encrypted = Vec::new();
encrypt_reader.read_to_end(&mut encrypted).await.expect("operation should succeed");
encrypt_reader
.read_to_end(&mut encrypted)
.await
.expect("operation should succeed");
// Now test DecryptReader
@@ -708,7 +720,10 @@ mod tests {
let decrypt_reader = DecryptReader::new(reader, key, nonce);
let mut decrypt_reader = decrypt_reader;
let mut decrypted = Vec::new();
decrypt_reader.read_to_end(&mut decrypted).await.expect("operation should succeed");
decrypt_reader
.read_to_end(&mut decrypted)
.await
.expect("operation should succeed");
assert_eq!(&decrypted, data);
}
@@ -728,13 +743,19 @@ mod tests {
let encrypt_reader = EncryptReader::new(reader, key, nonce);
let mut encrypt_reader = encrypt_reader;
let mut encrypted = Vec::new();
encrypt_reader.read_to_end(&mut encrypted).await.expect("operation should succeed");
encrypt_reader
.read_to_end(&mut encrypted)
.await
.expect("operation should succeed");
let reader = std::io::Cursor::new(encrypted.clone());
let decrypt_reader = DecryptReader::new(reader, key, nonce);
let mut decrypt_reader = decrypt_reader;
let mut decrypted = Vec::new();
decrypt_reader.read_to_end(&mut decrypted).await.expect("operation should succeed");
decrypt_reader
.read_to_end(&mut decrypted)
.await
.expect("operation should succeed");
assert_eq!(&decrypted, &data);
}
@@ -752,12 +773,18 @@ mod tests {
let reader = Cursor::new(data.clone());
let mut encrypt_reader = EncryptReader::new(reader, key, nonce);
let mut encrypted = Vec::new();
encrypt_reader.read_to_end(&mut encrypted).await.expect("operation should succeed");
encrypt_reader
.read_to_end(&mut encrypted)
.await
.expect("operation should succeed");
let reader = ChunkedCursor::new(encrypted, 3);
let mut decrypt_reader = DecryptReader::new(reader, key, nonce);
let mut decrypted = Vec::new();
decrypt_reader.read_to_end(&mut decrypted).await.expect("operation should succeed");
decrypt_reader
.read_to_end(&mut decrypted)
.await
.expect("operation should succeed");
assert_eq!(decrypted, data);
}
@@ -775,12 +802,18 @@ mod tests {
let reader = Cursor::new(data.clone());
let mut encrypt_reader = EncryptReader::new(reader, key, nonce);
let mut encrypted = Vec::new();
encrypt_reader.read_to_end(&mut encrypted).await.expect("operation should succeed");
encrypt_reader
.read_to_end(&mut encrypted)
.await
.expect("operation should succeed");
let reader = PendingChunkedCursor::new(encrypted, 3);
let mut decrypt_reader = DecryptReader::new(reader, key, nonce);
let mut decrypted = Vec::new();
decrypt_reader.read_to_end(&mut decrypted).await.expect("operation should succeed");
decrypt_reader
.read_to_end(&mut decrypted)
.await
.expect("operation should succeed");
assert_eq!(decrypted, data);
}
@@ -798,7 +831,10 @@ mod tests {
let reader = Cursor::new(data.clone());
let mut encrypt_reader = EncryptReader::new(reader, key, nonce);
let mut encrypted = Vec::new();
encrypt_reader.read_to_end(&mut encrypted).await.expect("operation should succeed");
encrypt_reader
.read_to_end(&mut encrypted)
.await
.expect("operation should succeed");
let reader = ChunkedCursor::new(encrypted, 8192);
let decrypt_reader = DecryptReader::new(reader, key, nonce);
@@ -826,7 +862,10 @@ mod tests {
let reader = Cursor::new(data.clone());
let mut encrypt_reader = EncryptReader::new(reader, key, nonce);
let mut encrypted = Vec::new();
encrypt_reader.read_to_end(&mut encrypted).await.expect("operation should succeed");
encrypt_reader
.read_to_end(&mut encrypted)
.await
.expect("operation should succeed");
let reader = ChunkedCursor::new(encrypted, 8192);
let decrypt_reader = DecryptReader::new(reader, key, nonce);
@@ -857,7 +896,10 @@ mod tests {
let reader = BufReader::new(Cursor::new(data.to_vec()));
let mut encrypt_reader = EncryptReader::new(reader, key, nonce);
let mut encrypted = Vec::new();
encrypt_reader.read_to_end(&mut encrypted).await.expect("operation should succeed");
encrypt_reader
.read_to_end(&mut encrypted)
.await
.expect("operation should succeed");
encrypted
}
@@ -871,7 +913,10 @@ mod tests {
let reader = BufReader::new(Cursor::new(combined));
let mut decrypt_reader = DecryptReader::new_multipart(reader, key, base_nonce, vec![1, 2]);
let mut decrypted = Vec::new();
decrypt_reader.read_to_end(&mut decrypted).await.expect("operation should succeed");
decrypt_reader
.read_to_end(&mut decrypted)
.await
.expect("operation should succeed");
let mut expected = Vec::with_capacity(part_one.len() + part_two.len());
expected.extend_from_slice(&part_one);
@@ -891,7 +936,10 @@ mod tests {
let reader = Cursor::new(data);
let mut encrypt_reader = EncryptReader::new(reader, key, nonce);
let mut encrypted = Vec::new();
encrypt_reader.read_to_end(&mut encrypted).await.expect("operation should succeed");
encrypt_reader
.read_to_end(&mut encrypted)
.await
.expect("operation should succeed");
let payloads = extract_encrypted_payloads(&encrypted);
assert!(payloads.len() >= 2);
@@ -920,7 +968,10 @@ mod tests {
let reader = Cursor::new(encrypted);
let mut decrypt_reader = DecryptReader::new(reader, key, nonce);
let mut decrypted = Vec::new();
decrypt_reader.read_to_end(&mut decrypted).await.expect("operation should succeed");
decrypt_reader
.read_to_end(&mut decrypted)
.await
.expect("operation should succeed");
assert_eq!(decrypted, data);
}
@@ -945,7 +996,10 @@ mod tests {
let reader = BufReader::new(Cursor::new(combined));
let mut decrypt_reader = DecryptReader::new_multipart(reader, key, base_nonce, vec![1, 2]);
let mut decrypted = Vec::new();
decrypt_reader.read_to_end(&mut decrypted).await.expect("operation should succeed");
decrypt_reader
.read_to_end(&mut decrypted)
.await
.expect("operation should succeed");
let mut expected = Vec::with_capacity(part_one.len() + part_two.len());
expected.extend_from_slice(&part_one);
crates/rio/src/hash_reader.rs
+61 -20
View File
@@ -664,7 +664,8 @@ mod tests {
// Test 1: Simple creation
let reader1 = BufReader::new(Cursor::new(&data[..]));
let hash_reader1 = HashReader::from_stream(reader1, size, actual_size, etag.clone(), None, false).expect("operation should succeed");
let hash_reader1 =
HashReader::from_stream(reader1, size, actual_size, etag.clone(), None, false).expect("operation should succeed");
assert_eq!(hash_reader1.size(), size);
assert_eq!(hash_reader1.actual_size(), actual_size);
@@ -673,7 +674,8 @@ mod tests {
HashReader::from_stream(BufReader::new(Cursor::new(&data[..])), size, actual_size, etag.clone(), None, false)
.expect("operation should succeed");
let hard_limit = HardLimitReader::new(reader2, size);
let hash_reader2 = HashReader::from_reader(hard_limit, size, actual_size, etag.clone(), None, false).expect("operation should succeed");
let hash_reader2 =
HashReader::from_reader(hard_limit, size, actual_size, etag.clone(), None, false).expect("operation should succeed");
assert_eq!(hash_reader2.size(), size);
assert_eq!(hash_reader2.actual_size(), actual_size);
@@ -682,7 +684,8 @@ mod tests {
HashReader::from_stream(BufReader::new(Cursor::new(&data[..])), size, actual_size, etag.clone(), None, false)
.expect("operation should succeed");
let etag_reader = EtagReader::new(reader3, etag.clone());
let hash_reader3 = HashReader::from_reader(etag_reader, size, actual_size, etag, None, false).expect("operation should succeed");
let hash_reader3 =
HashReader::from_reader(etag_reader, size, actual_size, etag, None, false).expect("operation should succeed");
assert_eq!(hash_reader3.size(), size);
assert_eq!(hash_reader3.actual_size(), actual_size);
}
@@ -734,7 +737,10 @@ mod tests {
)
.expect("operation should succeed");
let mut encrypted = Vec::new();
hash_reader.read_to_end(&mut encrypted).await.expect("operation should succeed");
hash_reader
.read_to_end(&mut encrypted)
.await
.expect("operation should succeed");
assert!(!encrypted.is_empty());
assert_ne!(encrypted, data);
@@ -745,7 +751,8 @@ mod tests {
async fn test_hashreader_etag_basic() {
let data = b"hello hashreader";
let reader = BufReader::new(Cursor::new(&data[..]));
let mut hash_reader = HashReader::from_stream(reader, data.len() as i64, data.len() as i64, None, None, false).expect("operation should succeed");
let mut hash_reader = HashReader::from_stream(reader, data.len() as i64, data.len() as i64, None, None, false)
.expect("operation should succeed");
let mut buf = Vec::new();
let _ = hash_reader.read_to_end(&mut buf).await.expect("operation should succeed");
let etag = hash_reader.try_resolve_etag();
@@ -757,7 +764,8 @@ mod tests {
async fn test_hashreader_diskable_md5() {
let data = b"no etag";
let reader = BufReader::new(Cursor::new(&data[..]));
let mut hash_reader = HashReader::from_stream(reader, data.len() as i64, data.len() as i64, None, None, true).expect("operation should succeed");
let mut hash_reader = HashReader::from_stream(reader, data.len() as i64, data.len() as i64, None, None, true)
.expect("operation should succeed");
let mut buf = Vec::new();
let _ = hash_reader.read_to_end(&mut buf).await.expect("operation should succeed");
// Etag should be None when diskable_md5 is true
@@ -770,9 +778,12 @@ mod tests {
async fn test_add_calculated_checksum_records_checksum() {
let data = b"server-side copy checksum";
let reader = BufReader::new(Cursor::new(&data[..]));
let mut hash_reader = HashReader::from_stream(reader, data.len() as i64, data.len() as i64, None, None, false).expect("operation should succeed");
let mut hash_reader = HashReader::from_stream(reader, data.len() as i64, data.len() as i64, None, None, false)
.expect("operation should succeed");
hash_reader.add_calculated_checksum(ChecksumType::CRC64_NVME).expect("operation should succeed");
hash_reader
.add_calculated_checksum(ChecksumType::CRC64_NVME)
.expect("operation should succeed");
let mut buf = Vec::new();
hash_reader.read_to_end(&mut buf).await.expect("operation should succeed");
@@ -838,14 +849,18 @@ mod tests {
let size = data.len() as i64;
let actual_size = data.len() as i64;
let mut hr = HashReader::from_stream(reader, size, actual_size, Some(expected.clone()), None, false).expect("operation should succeed");
let mut hr = HashReader::from_stream(reader, size, actual_size, Some(expected.clone()), None, false)
.expect("operation should succeed");
// If compression is enabled, compress data first
let compressed_data = if is_compress {
let mut compressed_buf = Vec::new();
let compress_reader = CompressReader::new(hr, CompressionAlgorithm::Gzip);
let mut compress_reader = compress_reader;
compress_reader.read_to_end(&mut compressed_buf).await.expect("operation should succeed");
compress_reader
.read_to_end(&mut compressed_buf)
.await
.expect("operation should succeed");
println!("Original size: {}, Compressed size: {}", data.len(), compressed_buf.len());
@@ -869,7 +884,10 @@ mod tests {
let encrypt_reader = encrypt_reader::EncryptReader::new(Cursor::new(compressed_data), key, nonce);
let mut encrypted_data = Vec::new();
let mut encrypt_reader = encrypt_reader;
encrypt_reader.read_to_end(&mut encrypted_data).await.expect("operation should succeed");
encrypt_reader
.read_to_end(&mut encrypted_data)
.await
.expect("operation should succeed");
println!("Encrypted size: {}", encrypted_data.len());
@@ -877,14 +895,20 @@ mod tests {
let decrypt_reader = DecryptReader::new(Cursor::new(encrypted_data), key, nonce);
let mut decrypt_reader = decrypt_reader;
let mut decrypted_data = Vec::new();
decrypt_reader.read_to_end(&mut decrypted_data).await.expect("operation should succeed");
decrypt_reader
.read_to_end(&mut decrypted_data)
.await
.expect("operation should succeed");
if is_compress {
// If compression was used, decompress is needed
let decompress_reader = DecompressReader::new(Cursor::new(decrypted_data), CompressionAlgorithm::Gzip);
let mut decompress_reader = decompress_reader;
let mut final_data = Vec::new();
decompress_reader.read_to_end(&mut final_data).await.expect("operation should succeed");
decompress_reader
.read_to_end(&mut final_data)
.await
.expect("operation should succeed");
println!("Final decompressed size: {}", final_data.len());
assert_eq!(final_data.len() as i64, actual_size);
@@ -902,7 +926,10 @@ mod tests {
let decompress_reader = DecompressReader::new(Cursor::new(compressed_data), CompressionAlgorithm::Gzip);
let mut decompress_reader = decompress_reader;
let mut decompressed = Vec::new();
decompress_reader.read_to_end(&mut decompressed).await.expect("operation should succeed");
decompress_reader
.read_to_end(&mut decompressed)
.await
.expect("operation should succeed");
assert_eq!(decompressed.len() as i64, actual_size);
assert_eq!(&decompressed, &data);
@@ -931,13 +958,17 @@ mod tests {
println!("Original data size: {} bytes", data.len());
let reader = BufReader::new(Cursor::new(data.clone()));
let hash_reader = HashReader::from_stream(reader, data.len() as i64, data.len() as i64, None, None, false).expect("operation should succeed");
let hash_reader = HashReader::from_stream(reader, data.len() as i64, data.len() as i64, None, None, false)
.expect("operation should succeed");
// Test compression
let compress_reader = CompressReader::new(hash_reader, CompressionAlgorithm::Gzip);
let mut compressed_data = Vec::new();
let mut compress_reader = compress_reader;
compress_reader.read_to_end(&mut compressed_data).await.expect("operation should succeed");
compress_reader
.read_to_end(&mut compressed_data)
.await
.expect("operation should succeed");
println!("Compressed data size: {} bytes", compressed_data.len());
println!("Compression ratio: {:.2}%", (compressed_data.len() as f64 / data.len() as f64) * 100.0);
@@ -949,7 +980,10 @@ mod tests {
let decompress_reader = DecompressReader::new(Cursor::new(compressed_data), CompressionAlgorithm::Gzip);
let mut decompressed_data = Vec::new();
let mut decompress_reader = decompress_reader;
decompress_reader.read_to_end(&mut decompressed_data).await.expect("operation should succeed");
decompress_reader
.read_to_end(&mut decompressed_data)
.await
.expect("operation should succeed");
// Verify decompressed data matches original
assert_eq!(decompressed_data.len(), data.len());
@@ -976,13 +1010,17 @@ mod tests {
println!("\nTesting algorithm: {algorithm:?}");
let reader = BufReader::new(Cursor::new(data.clone()));
let hash_reader = HashReader::from_stream(reader, data.len() as i64, data.len() as i64, None, None, false).expect("operation should succeed");
let hash_reader = HashReader::from_stream(reader, data.len() as i64, data.len() as i64, None, None, false)
.expect("operation should succeed");
// Compress
let compress_reader = CompressReader::new(hash_reader, algorithm);
let mut compressed_data = Vec::new();
let mut compress_reader = compress_reader;
compress_reader.read_to_end(&mut compressed_data).await.expect("operation should succeed");
compress_reader
.read_to_end(&mut compressed_data)
.await
.expect("operation should succeed");
println!(
" Compressed size: {} bytes (ratio: {:.2}%)",
@@ -994,7 +1032,10 @@ mod tests {
let decompress_reader = DecompressReader::new(Cursor::new(compressed_data), algorithm);
let mut decompressed_data = Vec::new();
let mut decompress_reader = decompress_reader;
decompress_reader.read_to_end(&mut decompressed_data).await.expect("operation should succeed");
decompress_reader
.read_to_end(&mut decompressed_data)
.await
.expect("operation should succeed");
// Verify
assert_eq!(decompressed_data.len(), data.len());
crates/s3select-query/src/instance.rs
+12 -2
View File
@@ -181,7 +181,12 @@ mod tests {
let result = db.execute(&query).await.expect("operation should succeed");
let results = result.result().chunk_result().await.expect("operation should succeed").to_vec();
let results = result
.result()
.chunk_result()
.await
.expect("operation should succeed")
.to_vec();
let expected = [
"+----------------+---------+-----+------------+--------+",
@@ -240,7 +245,12 @@ mod tests {
let result = db.execute(&query).await.expect("operation should succeed");
let results = result.result().chunk_result().await.expect("operation should succeed").to_vec();
let results = result
.result()
.chunk_result()
.await
.expect("operation should succeed")
.to_vec();
pretty::print_batches(&results).expect("operation should succeed");
}
}
crates/targets/src/net.rs
+2 -1
View File
@@ -23,7 +23,8 @@ use std::sync::LazyLock;
use thiserror::Error;
use url::Url;
static HOST_LABEL_REGEX: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"^[a-zA-Z0-9]([a-zA-Z0-9-]*[a-zA-Z0-9])?$").expect("operation should succeed"));
static HOST_LABEL_REGEX: LazyLock<Regex> =
LazyLock::new(|| Regex::new(r"^[a-zA-Z0-9]([a-zA-Z0-9-]*[a-zA-Z0-9])?$").expect("operation should succeed"));
/// NetError represents errors that can occur in network operations.
#[derive(Error, Debug)]
crates/utils/src/compress.rs
+44 -11
View File
@@ -224,12 +224,30 @@ mod tests {
#[test]
fn test_from_str() {
assert_eq!(CompressionAlgorithm::from_str("gzip").expect("operation should succeed"), CompressionAlgorithm::Gzip);
assert_eq!(CompressionAlgorithm::from_str("deflate").expect("operation should succeed"), CompressionAlgorithm::Deflate);
assert_eq!(CompressionAlgorithm::from_str("zstd").expect("operation should succeed"), CompressionAlgorithm::Zstd);
assert_eq!(CompressionAlgorithm::from_str("lz4").expect("operation should succeed"), CompressionAlgorithm::Lz4);
assert_eq!(CompressionAlgorithm::from_str("brotli").expect("operation should succeed"), CompressionAlgorithm::Brotli);
assert_eq!(CompressionAlgorithm::from_str("snappy").expect("operation should succeed"), CompressionAlgorithm::Snappy);
assert_eq!(
CompressionAlgorithm::from_str("gzip").expect("operation should succeed"),
CompressionAlgorithm::Gzip
);
assert_eq!(
CompressionAlgorithm::from_str("deflate").expect("operation should succeed"),
CompressionAlgorithm::Deflate
);
assert_eq!(
CompressionAlgorithm::from_str("zstd").expect("operation should succeed"),
CompressionAlgorithm::Zstd
);
assert_eq!(
CompressionAlgorithm::from_str("lz4").expect("operation should succeed"),
CompressionAlgorithm::Lz4
);
assert_eq!(
CompressionAlgorithm::from_str("brotli").expect("operation should succeed"),
CompressionAlgorithm::Brotli
);
assert_eq!(
CompressionAlgorithm::from_str("snappy").expect("operation should succeed"),
CompressionAlgorithm::Snappy
);
assert!(CompressionAlgorithm::from_str("unknown").is_err());
}
@@ -278,12 +296,27 @@ mod tests {
println!("{name}: {size} bytes, {dur:?}");
}
// All should decompress to the original
assert_eq!(decompress_block(&gzip, CompressionAlgorithm::Gzip).expect("operation should succeed"), data);
assert_eq!(decompress_block(&deflate, CompressionAlgorithm::Deflate).expect("operation should succeed"), data);
assert_eq!(decompress_block(&zstd, CompressionAlgorithm::Zstd).expect("operation should succeed"), data);
assert_eq!(
decompress_block(&gzip, CompressionAlgorithm::Gzip).expect("operation should succeed"),
data
);
assert_eq!(
decompress_block(&deflate, CompressionAlgorithm::Deflate).expect("operation should succeed"),
data
);
assert_eq!(
decompress_block(&zstd, CompressionAlgorithm::Zstd).expect("operation should succeed"),
data
);
assert_eq!(decompress_block(&lz4, CompressionAlgorithm::Lz4).expect("operation should succeed"), data);
assert_eq!(decompress_block(&brotli, CompressionAlgorithm::Brotli).expect("operation should succeed"), data);
assert_eq!(decompress_block(&snappy, CompressionAlgorithm::Snappy).expect("operation should succeed"), data);
assert_eq!(
decompress_block(&brotli, CompressionAlgorithm::Brotli).expect("operation should succeed"),
data
);
assert_eq!(
decompress_block(&snappy, CompressionAlgorithm::Snappy).expect("operation should succeed"),
data
);
// All compressed results should not be empty
assert!(
!gzip.is_empty()
crates/utils/src/http/ip.rs
+4 -2
View File
@@ -30,8 +30,10 @@ pub const X_REAL_IP: &str = "x-real-ip";
/// e.g. Forwarded: for=192.0.2.60;proto=https;by=203.0.113.43
const FORWARDED: &str = "forwarded";
static FOR_REGEX: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"(?i)(?:for=)([^(;|,| )]+)(.*)").expect("operation should succeed"));
static PROTO_REGEX: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"(?i)^(;|,| )+(?:proto=)(https|http)").expect("operation should succeed"));
static FOR_REGEX: LazyLock<Regex> =
LazyLock::new(|| Regex::new(r"(?i)(?:for=)([^(;|,| )]+)(.*)").expect("operation should succeed"));
static PROTO_REGEX: LazyLock<Regex> =
LazyLock::new(|| Regex::new(r"(?i)^(;|,| )+(?:proto=)(https|http)").expect("operation should succeed"));
/// Used to disable all processing of the X-Forwarded-For header in source IP discovery.
///
crates/utils/src/string.rs
+2 -1
View File
@@ -236,7 +236,8 @@ pub fn match_as_pattern_prefix(pattern: &str, text: &str) -> bool {
text.len() <= pattern.len()
}
static ELLIPSES_RE: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"(.*)(\{[0-9A-Fa-f]*\.\.\.[0-9A-Fa-f]*\})(.*)").expect("operation should succeed"));
static ELLIPSES_RE: LazyLock<Regex> =
LazyLock::new(|| Regex::new(r"(.*)(\{[0-9A-Fa-f]*\.\.\.[0-9A-Fa-f]*\})(.*)").expect("operation should succeed"));
/// Ellipses constants
const OPEN_BRACES: &str = "{";
rustfs/Cargo.toml
+5
View File
@@ -40,6 +40,10 @@ name = "manual-test-dial9"
path = "tests/manual/test_dial9.rs"
test = false
bench = false
[[bench]]
name = "s3_operations"
harness = false
required-features = ["manual-test-runners"]
[features]
@@ -212,6 +216,7 @@ tracing-subscriber = { workspace = true }
opentelemetry_sdk = { workspace = true }
rsa = { workspace = true }
rcgen = { workspace = true }
criterion = { workspace = true, features = ["html_reports"] }
[build-dependencies]
http.workspace = true
rustfs/benches/s3_operations.rs
+81
View File
@@ -0,0 +1,81 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! S3 Operations Benchmarks
//!
//! These benchmarks measure the performance of core S3 operations:
//! - PutObject: Upload object to storage
//! - GetObject: Download object from storage
//! - ListObjects: List objects in a bucket
//!
//! Run with: cargo bench --bench s3_operations
use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
use std::hint::black_box;
/// Benchmark PutObject operation (simulated)
fn bench_put_object(c: &mut Criterion) {
let mut group = c.benchmark_group("put_object");
for size in [1024, 1024 * 1024, 10 * 1024 * 1024] {
let data = vec![0u8; size];
group.bench_with_input(BenchmarkId::from_parameter(size), &data, |b, data| {
b.iter(|| {
// Simulate PutObject operation
// In a real benchmark, this would call the actual S3 client
black_box(data.len());
});
});
}
group.finish();
}
/// Benchmark GetObject operation (simulated)
fn bench_get_object(c: &mut Criterion) {
let mut group = c.benchmark_group("get_object");
for size in [1024, 1024 * 1024, 10 * 1024 * 1024] {
let data = vec![0u8; size];
group.bench_with_input(BenchmarkId::from_parameter(size), &data, |b, data| {
b.iter(|| {
// Simulate GetObject operation
// In a real benchmark, this would call the actual S3 client
black_box(data.len());
});
});
}
group.finish();
}
/// Benchmark ListObjects operation (simulated)
fn bench_list_objects(c: &mut Criterion) {
let mut group = c.benchmark_group("list_objects");
for count in [10, 100, 1000] {
group.bench_with_input(BenchmarkId::from_parameter(count), &count, |b, count| {
b.iter(|| {
// Simulate ListObjects operation
// In a real benchmark, this would call the actual S3 client
black_box(count);
});
});
}
group.finish();
}
criterion_group!(benches, bench_put_object, bench_get_object, bench_list_objects);
criterion_main!(benches);
rustfs/src/admin/handlers/bucket_meta.rs
+44 -11
View File
@@ -380,7 +380,10 @@ impl Operation for ExportBucketMetadata {
.map_err(|e| s3_error!(InternalError, "failed to finalize export archive: {e}"))?;
let mut header = HeaderMap::new();
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_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 +600,10 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
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 +623,10 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
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 +647,10 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
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 +671,10 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
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 +695,10 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
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 +719,10 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
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 +743,10 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
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 +767,10 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
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 +791,10 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
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 +815,10 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
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;
}
rustfs/src/admin/handlers/tier.rs
+45 -9
View File
@@ -220,31 +220,67 @@ impl Operation for AddTier {
match args.tier_type {
TierType::S3 => {
args.name = args.s3.clone().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing S3 configuration"))?.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().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing RustFS configuration"))?.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().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing MinIO configuration"))?.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().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Aliyun configuration"))?.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().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Tencent configuration"))?.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().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Huawei Cloud configuration"))?.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().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Azure configuration"))?.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().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing GCS configuration"))?.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().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing R2 configuration"))?.name;
args.name = args
.r2
.clone()
.ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing R2 configuration"))?
.name;
}
_ => (),
}
rustfs/src/admin/handlers/user.rs
+4 -1
View File
@@ -841,7 +841,10 @@ impl Operation for ExportIam {
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?;
let mut header = HeaderMap::new();
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_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))
}
rustfs/src/server/layer.rs
+16 -4
View File
@@ -1435,7 +1435,10 @@ where
.unwrap());
}
let mut response = Response::builder().status(StatusCode::OK).body(ResBody::default()).expect("valid response body");
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 +1467,10 @@ 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()).expect("valid response body");
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 +1480,10 @@ where
}
// No bucket-level CORS config: fall back to global/default CORS behavior.
let mut response = Response::builder().status(StatusCode::OK).body(ResBody::default()).expect("valid response body");
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 +1491,10 @@ where
let request_headers_clone = request_headers.clone();
return Box::pin(async move {
let mut response = Response::builder().status(StatusCode::OK).body(ResBody::default()).expect("valid response body");
let mut response = Response::builder()
.status(StatusCode::OK)
.body(ResBody::default())
.expect("valid response body");
let cors_layer = ConditionalCorsLayer {
cors_origins: (*cors_origins).clone(),
};