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

Fix/fix issues #1564 (#1708)

Browse Source 
Signed-off-by: houseme <housemecn@gmail.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
houseme
2026-02-05 13:45:14 +08:00
committed by GitHub
parent e30781654d
commit 6bba41f11f
13 changed files with 956 additions and 720 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+45 -57
View File
@@ -645,9 +645,9 @@ checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8"
[[package]]
name = "aws-config"
version = "1.8.12"
version = "1.8.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "96571e6996817bf3d58f6b569e4b9fd2e9d2fcf9f7424eed07b2ce9bb87535e5"
checksum = "c456581cb3c77fafcc8c67204a70680d40b61112d6da78c77bd31d945b65f1b5"
dependencies = [
"aws-credential-types",
"aws-runtime",
@@ -655,7 +655,7 @@ dependencies = [
"aws-sdk-ssooidc",
"aws-sdk-sts",
"aws-smithy-async",
"aws-smithy-http 0.62.6",
"aws-smithy-http",
"aws-smithy-json",
"aws-smithy-runtime",
"aws-smithy-runtime-api",
@@ -710,15 +710,15 @@ dependencies = [
[[package]]
name = "aws-runtime"
version = "1.5.18"
version = "1.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "959dab27ce613e6c9658eb3621064d0e2027e5f2acb65bc526a43577facea557"
checksum = "c635c2dc792cb4a11ce1a4f392a925340d1bdf499289b5ec1ec6810954eb43f5"
dependencies = [
"aws-credential-types",
"aws-sigv4",
"aws-smithy-async",
"aws-smithy-eventstream",
"aws-smithy-http 0.62.6",
"aws-smithy-http",
"aws-smithy-runtime",
"aws-smithy-runtime-api",
"aws-smithy-types",
@@ -726,7 +726,9 @@ dependencies = [
"bytes",
"fastrand",
"http 0.2.12",
"http 1.4.0",
"http-body 0.4.6",
"http-body 1.0.1",
"percent-encoding",
"pin-project-lite",
"tracing",
@@ -735,9 +737,9 @@ dependencies = [
[[package]]
name = "aws-sdk-s3"
version = "1.121.0"
version = "1.122.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "61948728b681f88a1e49b9500469cf9e36575a424e745e2c5a651a42386e7d9c"
checksum = "94c2ca0cba97e8e279eb6c0b2d0aa10db5959000e602ab2b7c02de6b85d4c19b"
dependencies = [
"aws-credential-types",
"aws-runtime",
@@ -745,7 +747,7 @@ dependencies = [
"aws-smithy-async",
"aws-smithy-checksums",
"aws-smithy-eventstream",
"aws-smithy-http 0.62.6",
"aws-smithy-http",
"aws-smithy-json",
"aws-smithy-observability",
"aws-smithy-runtime",
@@ -759,7 +761,7 @@ dependencies = [
"hmac 0.12.1",
"http 0.2.12",
"http 1.4.0",
"http-body 0.4.6",
"http-body 1.0.1",
"lru",
"percent-encoding",
"regex-lite",
@@ -770,14 +772,14 @@ dependencies = [
[[package]]
name = "aws-sdk-sso"
version = "1.92.0"
version = "1.93.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b7d63bd2bdeeb49aa3f9b00c15e18583503b778b2e792fc06284d54e7d5b6566"
checksum = "9dcb38bb33fc0a11f1ffc3e3e85669e0a11a37690b86f77e75306d8f369146a0"
dependencies = [
"aws-credential-types",
"aws-runtime",
"aws-smithy-async",
"aws-smithy-http 0.62.6",
"aws-smithy-http",
"aws-smithy-json",
"aws-smithy-observability",
"aws-smithy-runtime",
@@ -787,20 +789,21 @@ dependencies = [
"bytes",
"fastrand",
"http 0.2.12",
"http 1.4.0",
"regex-lite",
"tracing",
]
[[package]]
name = "aws-sdk-ssooidc"
version = "1.94.0"
version = "1.95.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "532d93574bf731f311bafb761366f9ece345a0416dbcc273d81d6d1a1205239b"
checksum = "2ada8ffbea7bd1be1f53df1dadb0f8fdb04badb13185b3321b929d1ee3caad09"
dependencies = [
"aws-credential-types",
"aws-runtime",
"aws-smithy-async",
"aws-smithy-http 0.62.6",
"aws-smithy-http",
"aws-smithy-json",
"aws-smithy-observability",
"aws-smithy-runtime",
@@ -810,20 +813,21 @@ dependencies = [
"bytes",
"fastrand",
"http 0.2.12",
"http 1.4.0",
"regex-lite",
"tracing",
]
[[package]]
name = "aws-sdk-sts"
version = "1.96.0"
version = "1.97.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "357e9a029c7524db6a0099cd77fbd5da165540339e7296cca603531bc783b56c"
checksum = "e6443ccadc777095d5ed13e21f5c364878c9f5bad4e35187a6cdbd863b0afcad"
dependencies = [
"aws-credential-types",
"aws-runtime",
"aws-smithy-async",
"aws-smithy-http 0.62.6",
"aws-smithy-http",
"aws-smithy-json",
"aws-smithy-observability",
"aws-smithy-query",
@@ -834,19 +838,20 @@ dependencies = [
"aws-types",
"fastrand",
"http 0.2.12",
"http 1.4.0",
"regex-lite",
"tracing",
]
[[package]]
name = "aws-sigv4"
version = "1.3.7"
version = "1.3.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "69e523e1c4e8e7e8ff219d732988e22bfeae8a1cafdbe6d9eca1546fa080be7c"
checksum = "efa49f3c607b92daae0c078d48a4571f599f966dce3caee5f1ea55c4d9073f99"
dependencies = [
"aws-credential-types",
"aws-smithy-eventstream",
"aws-smithy-http 0.62.6",
"aws-smithy-http",
"aws-smithy-runtime-api",
"aws-smithy-types",
"bytes",
@@ -879,17 +884,18 @@ dependencies = [
[[package]]
name = "aws-smithy-checksums"
version = "0.63.13"
version = "0.64.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "23374b9170cbbcc6f5df8dc5ebb9b6c5c28a3c8f599f0e8b8b10eb6f4a5c6e74"
checksum = "ddcf418858f9f3edd228acb8759d77394fed7531cce78d02bdda499025368439"
dependencies = [
"aws-smithy-http 0.62.6",
"aws-smithy-http",
"aws-smithy-types",
"bytes",
"crc-fast",
"hex",
"http 0.2.12",
"http-body 0.4.6",
"http 1.4.0",
"http-body 1.0.1",
"http-body-util",
"md-5 0.10.6",
"pin-project-lite",
"sha1 0.10.6",
@@ -908,34 +914,13 @@ dependencies = [
"crc32fast",
]
[[package]]
name = "aws-smithy-http"
version = "0.62.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "826141069295752372f8203c17f28e30c464d22899a43a0c9fd9c458d469c88b"
dependencies = [
"aws-smithy-eventstream",
"aws-smithy-runtime-api",
"aws-smithy-types",
"bytes",
"bytes-utils",
"futures-core",
"futures-util",
"http 0.2.12",
"http 1.4.0",
"http-body 0.4.6",
"percent-encoding",
"pin-project-lite",
"pin-utils",
"tracing",
]
[[package]]
name = "aws-smithy-http"
version = "0.63.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "630e67f2a31094ffa51b210ae030855cb8f3b7ee1329bdd8d085aaf61e8b97fc"
dependencies = [
"aws-smithy-eventstream",
"aws-smithy-runtime-api",
"aws-smithy-types",
"bytes",
@@ -977,9 +962,9 @@ dependencies = [
[[package]]
name = "aws-smithy-json"
version = "0.61.9"
version = "0.62.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "49fa1213db31ac95288d981476f78d05d9cbb0353d22cdf3472cc05bb02f6551"
checksum = "3cb96aa208d62ee94104645f7b2ecaf77bf27edf161590b6224bfbac2832f979"
dependencies = [
"aws-smithy-types",
]
@@ -1010,7 +995,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f3df87c14f0127a0d77eb261c3bc45d5b4833e2a1f63583ebfb728e4852134ee"
dependencies = [
"aws-smithy-async",
"aws-smithy-http 0.63.3",
"aws-smithy-http",
"aws-smithy-http-client",
"aws-smithy-observability",
"aws-smithy-runtime-api",
@@ -1605,9 +1590,9 @@ dependencies = [
[[package]]
name = "clap"
version = "4.5.56"
version = "4.5.57"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a75ca66430e33a14957acc24c5077b503e7d374151b2b4b3a10c83b4ceb4be0e"
checksum = "6899ea499e3fb9305a65d5ebf6e3d2248c5fab291f300ad0a704fbe142eae31a"
dependencies = [
"clap_builder",
"clap_derive",
@@ -1615,9 +1600,9 @@ dependencies = [
[[package]]
name = "clap_builder"
version = "4.5.56"
version = "4.5.57"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "793207c7fa6300a0608d1080b858e5fdbe713cdc1c8db9fb17777d8a13e63df0"
checksum = "7b12c8b680195a62a8364d16b8447b01b6c2c8f9aaf68bee653be34d4245e238"
dependencies = [
"anstream",
"anstyle",
@@ -7629,10 +7614,13 @@ name = "rustfs"
version = "0.0.5"
dependencies = [
"aes-gcm 0.11.0-rc.2",
"anyhow",
"astral-tokio-tar",
"async-trait",
"atoi",
"atomic_enum",
"aws-config",
"aws-sdk-s3",
"axum",
"axum-server",
"backtrace",
@@ -8566,7 +8554,7 @@ checksum = "a50f4cf475b65d88e057964e0e9bb1f0aa9bbb2036dc65c64596b42932536984"
[[package]]
name = "s3s"
version = "0.13.0-alpha.2"
source = "git+https://github.com/rustfs/s3s.git?branch=dependabot%2Fcargo%2Fdep-0202#13f968fc89e3bd286d308ca855456b23734d4f74"
source = "git+https://github.com/rustfs/s3s.git?rev=13f968fc89e3bd286d308ca855456b23734d4f74#13f968fc89e3bd286d308ca855456b23734d4f74"
dependencies = [
"arc-swap",
"arrayvec",
Cargo.toml
+4 -4
View File
@@ -174,16 +174,16 @@ arc-swap = "1.8.0"
astral-tokio-tar = "0.5.6"
atoi = "2.0.0"
atomic_enum = "0.3.0"
aws-config = { version = "1.8.12" }
aws-config = { version = "1.8.13" }
aws-credential-types = { version = "1.2.11" }
aws-sdk-s3 = { version = "1.121.0", default-features = false, features = ["sigv4a", "default-https-client", "rt-tokio"] }
aws-sdk-s3 = { version = "1.122.0", default-features = false, features = ["sigv4a", "default-https-client", "rt-tokio"] }
aws-smithy-types = { version = "1.4.3" }
backtrace = "0.3.76"
base64 = "0.22.1"
base64-simd = "0.8.0"
brotli = "8.0.2"
cfg-if = "1.0.4"
clap = { version = "4.5.56", features = ["derive", "env"] }
clap = { version = "4.5.57", features = ["derive", "env"] }
const-str = { version = "1.1.0", features = ["std", "proc"] }
convert_case = "0.11.0"
criterion = { version = "0.8", features = ["html_reports"] }
@@ -229,7 +229,7 @@ rumqttc = { version = "0.25.1" }
rustix = { version = "1.1.3", features = ["fs"] }
rust-embed = { version = "8.11.0" }
rustc-hash = { version = "2.1.1" }
s3s = { version = "0.13.0-alpha.2", features = ["minio"], git = "https://github.com/rustfs/s3s.git", branch = "dependabot/cargo/dep-0202" }
s3s = { version = "0.13.0-alpha.2", features = ["minio"], git = "https://github.com/rustfs/s3s.git", rev = "13f968fc89e3bd286d308ca855456b23734d4f74" }
serial_test = "3.3.1"
shadow-rs = { version = "1.7.0", default-features = false }
siphasher = "1.0.2"
crates/ecstore/src/bucket/quota/checker.rs
+2 -2
View File
@@ -58,12 +58,12 @@ impl QuotaChecker {
let current_usage = self.get_real_time_usage(bucket).await?;
let expected_usage = match operation {
QuotaOperation::PutObject | QuotaOperation::CopyObject => current_usage + operation_size,
QuotaOperation::PutObject | QuotaOperation::PostObject | QuotaOperation::CopyObject => current_usage + operation_size,
QuotaOperation::DeleteObject => current_usage.saturating_sub(operation_size),
};
let allowed = match operation {
QuotaOperation::PutObject | QuotaOperation::CopyObject => {
QuotaOperation::PutObject | QuotaOperation::PostObject | QuotaOperation::CopyObject => {
quota_config.check_operation_allowed(current_usage, operation_size)
}
QuotaOperation::DeleteObject => true,
crates/ecstore/src/bucket/quota/mod.rs
+1
View File
@@ -90,6 +90,7 @@ pub struct QuotaCheckResult {
#[derive(Debug)]
pub enum QuotaOperation {
PutObject,
PostObject,
CopyObject,
DeleteObject,
}
rustfs/Cargo.toml
+4
View File
@@ -161,6 +161,10 @@ pprof = { workspace = true }
uuid = { workspace = true, features = ["v4"] }
serial_test = { workspace = true }
tempfile = { workspace = true }
aws-sdk-s3 = { workspace = true }
aws-config = { workspace = true }
anyhow = { workspace = true }
tokio = { workspace = true, features = ["test-util"] }
[build-dependencies]
http.workspace = true
rustfs/src/admin/handlers/quota.rs
+5
View File
@@ -415,6 +415,7 @@ impl Operation for CheckBucketQuotaHandler {
let operation: QuotaOperation = match request.operation_type.to_uppercase().as_str() {
"PUT" | "PUTOBJECT" => QuotaOperation::PutObject,
"POST" | "POSTOBJECT" => QuotaOperation::PostObject,
"COPY" | "COPYOBJECT" => QuotaOperation::CopyObject,
"DELETE" | "DELETEOBJECT" => QuotaOperation::DeleteObject,
_ => QuotaOperation::PutObject, // Default to PUT operation
@@ -455,6 +456,7 @@ mod tests {
fn test_quota_operation_parsing() {
let parse_operation = |operation: &str| match operation.to_uppercase().as_str() {
"PUT" | "PUTOBJECT" => QuotaOperation::PutObject,
"POST" | "POSTOBJECT" => QuotaOperation::PostObject,
"COPY" | "COPYOBJECT" => QuotaOperation::CopyObject,
"DELETE" | "DELETEOBJECT" => QuotaOperation::DeleteObject,
_ => QuotaOperation::PutObject,
@@ -463,6 +465,9 @@ mod tests {
assert!(matches!(parse_operation("put"), QuotaOperation::PutObject));
assert!(matches!(parse_operation("PUT"), QuotaOperation::PutObject));
assert!(matches!(parse_operation("PutObject"), QuotaOperation::PutObject));
assert!(matches!(parse_operation("post"), QuotaOperation::PostObject));
assert!(matches!(parse_operation("POST"), QuotaOperation::PostObject));
assert!(matches!(parse_operation("PostObject"), QuotaOperation::PostObject));
assert!(matches!(parse_operation("copy"), QuotaOperation::CopyObject));
assert!(matches!(parse_operation("DELETE"), QuotaOperation::DeleteObject));
assert!(matches!(parse_operation("unknown"), QuotaOperation::PutObject));
rustfs/src/storage/ecfs.rs
+9 -396
View File
@@ -30,15 +30,15 @@ use crate::storage::{
access::{ReqInfo, authorize_request, has_bypass_governance_header},
ecfs_extend::RFC1123,
options::{
copy_dst_opts, copy_src_opts, del_opts, extract_metadata, extract_metadata_from_mime_with_object_name,
get_complete_multipart_upload_opts, get_opts, parse_copy_source_range, put_opts,
copy_dst_opts, copy_src_opts, del_opts, extract_metadata, get_complete_multipart_upload_opts, get_opts,
parse_copy_source_range, put_opts,
},
};
use crate::storage::{
apply_lock_retention, check_preconditions, get_buffer_size_opt_in, get_validated_store, has_replication_rules,
parse_object_lock_legal_hold, parse_object_lock_retention, process_lambda_configurations, process_queue_configurations,
process_topic_configurations, validate_bucket_object_lock_enabled, validate_list_object_unordered_with_delimiter,
validate_object_key, wrap_response_with_cors,
check_preconditions, get_buffer_size_opt_in, get_validated_store, has_replication_rules, parse_object_lock_legal_hold,
parse_object_lock_retention, process_lambda_configurations, process_queue_configurations, process_topic_configurations,
validate_bucket_object_lock_enabled, validate_list_object_unordered_with_delimiter, validate_object_key,
wrap_response_with_cors,
};
use crate::storage::{entity, parse_part_number_i32_to_usize};
// base64 imports moved to sse module
@@ -226,6 +226,8 @@ impl FS {
Self {}
}
#[instrument(level = "debug", skip(self, req))]
#[allow(dead_code)]
async fn put_object_extract(&self, req: S3Request<PutObjectInput>) -> S3Result<S3Response<PutObjectOutput>> {
let helper = OperationHelper::new(&req, EventName::ObjectCreatedPut, "s3:PutObject").suppress_event();
let input = req.input;
@@ -4312,396 +4314,7 @@ impl S3 for FS {
#[instrument(level = "debug", skip(self, req))]
async fn put_object(&self, req: S3Request<PutObjectInput>) -> S3Result<S3Response<PutObjectOutput>> {
let mut helper = OperationHelper::new(&req, EventName::ObjectCreatedPut, "s3:PutObject");
if req
.headers
.get("X-Amz-Meta-Snowball-Auto-Extract")
.is_some_and(|v| v.to_str().unwrap_or_default() == "true")
{
return self.put_object_extract(req).await;
}
let input = req.input;
// Save SSE-C parameters before moving input
if let Some(ref storage_class) = input.storage_class
&& !is_valid_storage_class(storage_class.as_str())
{
return Err(s3_error!(InvalidStorageClass));
}
let PutObjectInput {
body,
bucket,
key,
content_length,
content_type,
tagging,
metadata,
version_id,
server_side_encryption,
sse_customer_algorithm,
sse_customer_key,
sse_customer_key_md5,
ssekms_key_id,
content_md5,
if_match,
if_none_match,
..
} = input;
// Validate object key
validate_object_key(&key, "PUT")?;
if if_match.is_some() || if_none_match.is_some() {
let Some(store) = new_object_layer_fn() else {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
};
match store.get_object_info(&bucket, &key, &ObjectOptions::default()).await {
Ok(info) => {
if !info.delete_marker {
if let Some(ifmatch) = if_match
&& let Some(strong_etag) = ifmatch.into_etag()
&& info
.etag
.as_ref()
.is_some_and(|etag| ETag::Strong(etag.clone()) != strong_etag)
{
return Err(s3_error!(PreconditionFailed));
}
if let Some(ifnonematch) = if_none_match
&& let Some(strong_etag) = ifnonematch.into_etag()
&& info
.etag
.as_ref()
.is_some_and(|etag| ETag::Strong(etag.clone()) == strong_etag)
{
return Err(s3_error!(PreconditionFailed));
}
}
}
Err(err) => {
if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
return Err(ApiError::from(err).into());
}
if if_match.is_some() && (is_err_object_not_found(&err) || is_err_version_not_found(&err)) {
return Err(ApiError::from(err).into());
}
}
}
}
// check quota for put operation
if let Some(size) = content_length
&& let Some(metadata_sys) = rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys.get()
{
let quota_checker = QuotaChecker::new(metadata_sys.clone());
match quota_checker
.check_quota(&bucket, QuotaOperation::PutObject, size as u64)
.await
{
Ok(check_result) => {
if !check_result.allowed {
return Err(S3Error::with_message(
S3ErrorCode::InvalidRequest,
format!(
"Bucket quota exceeded. Current usage: {} bytes, limit: {} bytes",
check_result.current_usage,
check_result.quota_limit.unwrap_or(0)
),
));
}
}
Err(e) => {
warn!("Quota check failed for bucket {}: {}, allowing operation", bucket, e);
}
}
}
let Some(body) = body else { return Err(s3_error!(IncompleteBody)) };
let mut size = match content_length {
Some(c) => c,
None => {
if let Some(val) = req.headers.get(AMZ_DECODED_CONTENT_LENGTH) {
match atoi::atoi::<i64>(val.as_bytes()) {
Some(x) => x,
None => return Err(s3_error!(UnexpectedContent)),
}
} else {
return Err(s3_error!(UnexpectedContent));
}
}
};
if size == -1 {
return Err(s3_error!(UnexpectedContent));
}
// Apply adaptive buffer sizing based on file size for optimal streaming performance.
// Uses workload profile configuration (enabled by default) to select appropriate buffer size.
// Buffer sizes range from 32KB to 4MB depending on file size and configured workload profile.
let buffer_size = get_buffer_size_opt_in(size);
let body = tokio::io::BufReader::with_capacity(
buffer_size,
StreamReader::new(body.map(|f| f.map_err(|e| std::io::Error::other(e.to_string())))),
);
// let body = Box::new(StreamReader::new(body.map(|f| f.map_err(|e| std::io::Error::other(e.to_string())))));
// let mut reader = PutObjReader::new(body, content_length as usize);
let store = get_validated_store(&bucket).await?;
let mut metadata = metadata.unwrap_or_default();
let object_lock_configuration = match metadata_sys::get_object_lock_config(&bucket).await {
Ok((cfg, _created)) => Some(cfg),
Err(err) => {
if err == StorageError::ConfigNotFound {
None
} else {
warn!("get_object_lock_config err {:?}", err);
return Err(S3Error::with_message(
S3ErrorCode::InternalError,
"Failed to load Object Lock configuration".to_string(),
));
}
}
};
apply_lock_retention(object_lock_configuration, &mut metadata);
if let Some(content_type) = content_type {
metadata.insert("content-type".to_string(), content_type.to_string());
}
extract_metadata_from_mime_with_object_name(&req.headers, &mut metadata, true, Some(&key));
if let Some(tags) = tagging {
metadata.insert(AMZ_OBJECT_TAGGING.to_owned(), tags.to_string());
}
let mut opts: ObjectOptions = put_opts(&bucket, &key, version_id.clone(), &req.headers, metadata.clone())
.await
.map_err(ApiError::from)?;
let mut reader: Box<dyn Reader> = Box::new(WarpReader::new(body));
let actual_size = size;
let mut md5hex = if let Some(base64_md5) = content_md5 {
let md5 = base64_simd::STANDARD
.decode_to_vec(base64_md5.as_bytes())
.map_err(|e| ApiError::from(StorageError::other(format!("Invalid content MD5: {e}"))))?;
Some(hex_simd::encode_to_string(&md5, hex_simd::AsciiCase::Lower))
} else {
None
};
let mut sha256hex = get_content_sha256(&req.headers);
if is_compressible(&req.headers, &key) && size > MIN_COMPRESSIBLE_SIZE as i64 {
let algorithm = CompressionAlgorithm::default();
metadata.insert(format!("{RESERVED_METADATA_PREFIX_LOWER}compression"), algorithm.to_string());
metadata.insert(format!("{RESERVED_METADATA_PREFIX_LOWER}actual-size",), size.to_string());
let mut hrd = HashReader::new(reader, size as i64, size as i64, md5hex, sha256hex, false).map_err(ApiError::from)?;
if let Err(err) = hrd.add_checksum_from_s3s(&req.headers, req.trailing_headers.clone(), false) {
return Err(ApiError::from(StorageError::other(format!("add_checksum error={err:?}"))).into());
}
opts.want_checksum = hrd.checksum();
opts.user_defined
.insert(format!("{RESERVED_METADATA_PREFIX_LOWER}compression"), algorithm.to_string());
opts.user_defined
.insert(format!("{RESERVED_METADATA_PREFIX_LOWER}actual-size",), size.to_string());
reader = Box::new(CompressReader::new(hrd, algorithm));
size = HashReader::SIZE_PRESERVE_LAYER;
md5hex = None;
sha256hex = None;
}
let mut reader = HashReader::new(reader, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)?;
if size >= 0 {
if let Err(err) = reader.add_checksum_from_s3s(&req.headers, req.trailing_headers.clone(), false) {
return Err(ApiError::from(StorageError::other(format!("add_checksum error={err:?}"))).into());
}
opts.want_checksum = reader.checksum();
}
// Apply encryption using unified SSE API
let encryption_request = EncryptionRequest {
bucket: &bucket,
key: &key,
server_side_encryption,
ssekms_key_id,
sse_customer_algorithm: sse_customer_algorithm.clone(),
sse_customer_key,
sse_customer_key_md5: sse_customer_key_md5.clone(),
content_size: actual_size,
part_number: None,
part_key: None,
part_nonce: None,
};
let (effective_sse, effective_kms_key_id) = match sse_encryption(encryption_request).await? {
Some(material) => {
let server_side_encryption = Some(material.server_side_encryption.clone());
let ssekms_key_id = material.kms_key_id.clone();
// Apply encryption wrapper
let encrypted_reader = material.wrap_reader(reader);
reader = HashReader::new(encrypted_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false)
.map_err(ApiError::from)?;
// Merge encryption metadata
metadata.extend(material.metadata);
(server_side_encryption, ssekms_key_id)
}
None => (None, None),
};
let mut reader = PutObjReader::new(reader);
let mt2 = metadata.clone();
opts.user_defined.extend(metadata);
let repoptions =
get_must_replicate_options(&mt2, "".to_string(), ReplicationStatusType::Empty, ReplicationType::Object, opts.clone());
let dsc = must_replicate(&bucket, &key, repoptions).await;
if dsc.replicate_any() {
let k = format!("{}{}", RESERVED_METADATA_PREFIX_LOWER, "replication-timestamp");
opts.user_defined.insert(k, jiff::Zoned::now().to_string());
let k = format!("{}{}", RESERVED_METADATA_PREFIX_LOWER, "replication-status");
opts.user_defined.insert(k, dsc.pending_status().unwrap_or_default());
}
let obj_info = store
.put_object(&bucket, &key, &mut reader, &opts)
.await
.map_err(ApiError::from)?;
// Fast in-memory update for immediate quota consistency
rustfs_ecstore::data_usage::increment_bucket_usage_memory(&bucket, obj_info.size as u64).await;
// Invalidate cache for the written object to prevent stale data
let manager = get_concurrency_manager();
let put_bucket = bucket.clone();
let put_key = key.clone();
let mut put_version = obj_info.version_id.map(|v| v.to_string());
if opts.version_suspended && obj_info.version_id.is_none_or(|v| v.is_nil()) {
put_version = Some("null".to_string());
}
helper = helper.object(obj_info.clone());
if let Some(version_id) = &put_version {
helper = helper.version_id(version_id.clone());
}
let put_version_clone = put_version.clone();
tokio::spawn(async move {
manager
.invalidate_cache_versioned(&put_bucket, &put_key, put_version_clone.as_deref())
.await;
});
let e_tag = obj_info.etag.clone().map(|etag| to_s3s_etag(&etag));
let repoptions =
get_must_replicate_options(&mt2, "".to_string(), ReplicationStatusType::Empty, ReplicationType::Object, opts);
let dsc = must_replicate(&bucket, &key, repoptions).await;
if dsc.replicate_any() {
schedule_replication(obj_info, store, dsc, ReplicationType::Object).await;
}
let mut checksum_crc32 = input.checksum_crc32;
let mut checksum_crc32c = input.checksum_crc32c;
let mut checksum_sha1 = input.checksum_sha1;
let mut checksum_sha256 = input.checksum_sha256;
let mut checksum_crc64nvme = input.checksum_crc64nvme;
if let Some(alg) = &input.checksum_algorithm
&& let Some(Some(checksum_str)) = req.trailing_headers.as_ref().map(|trailer| {
let key = match alg.as_str() {
ChecksumAlgorithm::CRC32 => rustfs_rio::ChecksumType::CRC32.key(),
ChecksumAlgorithm::CRC32C => rustfs_rio::ChecksumType::CRC32C.key(),
ChecksumAlgorithm::SHA1 => rustfs_rio::ChecksumType::SHA1.key(),
ChecksumAlgorithm::SHA256 => rustfs_rio::ChecksumType::SHA256.key(),
ChecksumAlgorithm::CRC64NVME => rustfs_rio::ChecksumType::CRC64_NVME.key(),
_ => return None,
};
trailer.read(|headers| {
headers
.get(key.unwrap_or_default())
.and_then(|value| value.to_str().ok().map(|s| s.to_string()))
})
})
{
match alg.as_str() {
ChecksumAlgorithm::CRC32 => checksum_crc32 = checksum_str,
ChecksumAlgorithm::CRC32C => checksum_crc32c = checksum_str,
ChecksumAlgorithm::SHA1 => checksum_sha1 = checksum_str,
ChecksumAlgorithm::SHA256 => checksum_sha256 = checksum_str,
ChecksumAlgorithm::CRC64NVME => checksum_crc64nvme = checksum_str,
_ => (),
}
}
let output = PutObjectOutput {
e_tag,
server_side_encryption: effective_sse, // TDD: Return effective encryption config
sse_customer_algorithm,
sse_customer_key_md5,
ssekms_key_id: effective_kms_key_id, // TDD: Return effective KMS key ID
checksum_crc32,
checksum_crc32c,
checksum_sha1,
checksum_sha256,
checksum_crc64nvme,
version_id: put_version,
..Default::default()
};
// TODO fix response for POST Policy (multipart/form-data) ，wait s3s crate update,fix issue #1564
// // If it is a POST Policy(multipart/form-data) path, the PutObjectInput carries the success_action_* field
// // Here, the response is uniformly rewritten, with the default being 204, redirect prioritizing 303, and status supporting 200/201/204
// if input.success_action_status.is_some() || input.success_action_redirect.is_some() {
// let mut form_fields = HashMap::<String, String>::new();
// if let Some(v) = &input.success_action_status {
// form_fields.insert("success_action_status".to_string(), v.to_string());
// }
// if let Some(v) = &input.success_action_redirect {
// form_fields.insert("success_action_redirect".to_string(), v.to_string());
// }
//
// // obj_info.etag has been converted to e_tag (s3s etag) above, so try to pass the original string here
// let etag_str = e_tag.as_ref().map(|v| v.as_str());
//
// // Returns using POST semantics: 204/303/201/200
// let resp = build_post_object_success_response(&form_fields, &bucket, &key, etag_str, None)?;
//
// // Keep helper event complete (note: (StatusCode, Body) is returned here instead of PutObjectOutput)
// let result = Ok(resp);
// let _ = helper.complete(&result);
// return result;
// }
let result = Ok(S3Response::new(output));
let _ = helper.complete(&result);
result
crate::storage::objects::GLOBAL_OBJECTS.put_object(req).await
}
async fn put_object_acl(&self, req: S3Request<PutObjectAclInput>) -> S3Result<S3Response<PutObjectAclOutput>> {
rustfs/src/storage/ecfs_extend.rs
-135
View File
@@ -18,7 +18,6 @@ use crate::config::workload_profiles::{
use crate::error::ApiError;
use crate::server::cors;
use crate::storage::ecfs::ListObjectUnorderedQuery;
use axum::body::Body;
use http::{HeaderMap, HeaderValue, StatusCode};
use metrics::counter;
use rustfs_ecstore::bucket::metadata_sys;
@@ -91,140 +90,6 @@ pub(crate) fn apply_lock_retention(object_lock_config: Option<ObjectLockConfigur
}
}
/// =======================
/// Presigned POST helpers
/// =======================
///
/// AWS S3 RESTObjectPOST (HTML form upload) semantics:
/// - Default success response is 204 (No Content)
/// - If `success_action_status` is specified, it may be 200, 201, or 204
/// - If `success_action_redirect` is specified, respond with 303 and Location header.
///
/// Reference:
/// https://docs.aws.amazon.com/AmazonS3/latest/API/RESTObjectPOST.html
#[allow(dead_code)]
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub(crate) enum PostObjectSuccessAction {
#[default]
NoContent204,
Ok200,
Created201,
Redirect303 {
location: String,
},
}
/// Parse success action from Presigned POST form fields.
///
/// Integration point (manual):
/// - In the PostPolicy handler, after parsing form fields, call this function to determine the desired success action.
///
/// # Arguments
/// * `fields` - HashMap of form fields from the POST request
///
/// # Returns
/// * `S3Result<PostObjectSuccessAction>` - Parsed success action or error
///
/// Notes:
/// - Follows AWS S3 behavior: `success_action_redirect` takes precedence over `success_action_status`.
/// - Validates `success_action_status` values; invalid values result in MalformedPOSTRequest error.
///
#[allow(dead_code)]
pub(crate) fn parse_success_action_from_form_fields(fields: &HashMap<String, String>) -> S3Result<PostObjectSuccessAction> {
// 1) success_action_redirect wins over success_action_status (AWS compatible behavior).
if let Some(loc) = fields
.get("success_action_redirect")
.map(|s| s.trim())
.filter(|s| !s.is_empty())
{
return Ok(PostObjectSuccessAction::Redirect303 {
location: loc.to_string(),
});
}
// 2) success_action_status is optional; default is 204.
let Some(status_str) = fields
.get("success_action_status")
.map(|s| s.trim())
.filter(|s| !s.is_empty())
else {
return Ok(PostObjectSuccessAction::NoContent204);
};
// AWS allows only 200/201/204 for POST form success_action_status.
// Treat invalid values as MalformedPOSTRequest to match S3 strictness.
match status_str {
"200" => Ok(PostObjectSuccessAction::Ok200),
"201" => Ok(PostObjectSuccessAction::Created201),
"204" => Ok(PostObjectSuccessAction::NoContent204),
_ => Err(S3Error::with_message(
S3ErrorCode::MalformedPOSTRequest,
format!("Invalid success_action_status: {status_str}. Allowed values are 200, 201, 204."),
)),
}
}
/// Build the final S3Response for a successful Presigned POST upload.
///
/// Integration point (manual):
/// - After `put_object` succeeds in the PostPolicy handler, call this function
/// with parsed form fields + object info to produce the correct HTTP status.
///
/// Notes:
/// - For 204: empty body
/// - For 303: empty body + Location header
/// - For 200: empty body (some clients accept this); you may optionally return XML/HTML body if you already implement it.
/// - For 201: prefer returning PostResponse XML; if not available, empty body still satisfies most clients, but strict tests may require XML. If you have a PostResponse serializer already, plug it in here.
#[allow(dead_code)]
pub(crate) fn build_post_object_success_response(
form_fields: &HashMap<String, String>,
// These are optional; used if you want to return richer responses for 200/201.
bucket: &str,
key: &str,
etag: Option<&str>,
location: Option<&str>,
) -> S3Result<S3Response<(StatusCode, Body)>> {
let action = parse_success_action_from_form_fields(form_fields)?;
match action {
PostObjectSuccessAction::NoContent204 => Ok(S3Response::new((StatusCode::NO_CONTENT, Body::empty()))),
PostObjectSuccessAction::Redirect303 { location } => {
let mut headers = HeaderMap::new();
headers.insert(
http::header::LOCATION,
HeaderValue::from_str(&location)
.map_err(|_| S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid success_action_redirect URL"))?,
);
Ok(S3Response::with_headers((StatusCode::SEE_OTHER, Body::empty()), headers))
}
PostObjectSuccessAction::Ok200 => {
// AWS may return 200 with an HTML/redirect response for browser workflows.
// For compatibility, returning empty body is acceptable unless strict clients require content.
// Keep Content-Length implicit; Body::empty() -> 0.
Ok(S3Response::new((StatusCode::OK, Body::empty())))
}
PostObjectSuccessAction::Created201 => {
// AWS 201 response is XML:
// <PostResponse>
// <Location>...</Location>
// <Bucket>...</Bucket>
// <Key>...</Key>
// <ETag>...</ETag>
// </PostResponse>
//
// If RustFS already has a DTO for this, switch to it here.
// To keep this patch minimal and safe, return empty body with 201 by default.
//
// IMPORTANT: If you run strict s3-tests for POST Object, you may need to implement XML body.
let _ = (bucket, key, etag, location);
Ok(S3Response::new((StatusCode::CREATED, Body::empty())))
}
}
}
/// Calculate adaptive buffer size with workload profile support.
///
/// This enhanced version supports different workload profiles for optimal performance
rustfs/src/storage/mod.rs
+5 -6
View File
@@ -14,20 +14,19 @@
pub mod access;
pub mod concurrency;
#[cfg(test)]
mod concurrent_get_object_test;
pub mod ecfs;
mod ecfs_extend;
pub(crate) mod entity;
pub(crate) mod helper;
pub mod options;
pub mod tonic_service;
#[cfg(test)]
mod concurrent_get_object_test;
mod ecfs_extend;
pub(crate) use ecfs_extend::*;
#[cfg(test)]
mod ecfs_test;
pub(crate) mod head_prefix;
mod objects;
mod sse;
#[cfg(test)]
mod sse_test;
pub(crate) use ecfs_extend::*;
rustfs/src/storage/objects/mod.rs
+27
View File
@@ -0,0 +1,27 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use std::sync::LazyLock;
mod put_object;
pub(crate) struct Objects;
pub(crate) static GLOBAL_OBJECTS: LazyLock<Objects> = LazyLock::new(Objects::new);
impl Objects {
pub fn new() -> Self {
Objects
}
}
rustfs/src/storage/objects/put_object.rs
+666
View File
@@ -0,0 +1,666 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::error::ApiError;
use crate::storage::concurrency::get_concurrency_manager;
use crate::storage::helper::OperationHelper;
use crate::storage::objects::Objects;
use crate::storage::options::{extract_metadata_from_mime_with_object_name, get_content_sha256, put_opts};
use crate::storage::sse::{EncryptionRequest, sse_encryption};
use crate::storage::{apply_lock_retention, get_buffer_size_opt_in, get_validated_store, validate_object_key};
use futures_util::StreamExt;
use http::HeaderMap;
use rustfs_ecstore::bucket::metadata_sys;
use rustfs_ecstore::bucket::quota::QuotaOperation;
use rustfs_ecstore::bucket::quota::checker::QuotaChecker;
use rustfs_ecstore::bucket::replication::{get_must_replicate_options, must_replicate, schedule_replication};
use rustfs_ecstore::client::object_api_utils::to_s3s_etag;
use rustfs_ecstore::compress::{MIN_COMPRESSIBLE_SIZE, is_compressible};
use rustfs_ecstore::error::{StorageError, is_err_object_not_found, is_err_version_not_found};
use rustfs_ecstore::set_disk::is_valid_storage_class;
use rustfs_ecstore::store_api::{ObjectIO, ObjectOptions, PutObjReader};
use rustfs_ecstore::{StorageAPI, new_object_layer_fn};
use rustfs_filemeta::{ReplicationStatusType, ReplicationType};
use rustfs_notify::notifier_global;
use rustfs_rio::{CompressReader, HashReader, Reader, WarpReader};
use rustfs_targets::EventName;
use rustfs_utils::http::{AMZ_DECODED_CONTENT_LENGTH, AMZ_OBJECT_TAGGING, RESERVED_METADATA_PREFIX_LOWER};
use rustfs_utils::{
CompressionAlgorithm, extract_params_header, extract_resp_elements, get_request_host, get_request_port,
get_request_user_agent,
};
use rustfs_zip::CompressionFormat;
use s3s::dto::{ChecksumAlgorithm, ETag, PutObjectInput, PutObjectOutput};
use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
use std::collections::HashMap;
use std::path::Path;
use tokio_tar::Archive;
use tokio_util::io::StreamReader;
use tracing::{debug, error, instrument, warn};
impl Objects {
#[instrument(level = "debug", skip(self, req))]
pub async fn put_object(&self, req: S3Request<PutObjectInput>) -> S3Result<S3Response<PutObjectOutput>> {
let mut helper = OperationHelper::new(&req, EventName::ObjectCreatedPut, "s3:PutObject");
if req
.headers
.get("X-Amz-Meta-Snowball-Auto-Extract")
.is_some_and(|v| v.to_str().unwrap_or_default() == "true")
{
return self.put_object_extract(req).await;
}
let input = req.input;
// Save SSE-C parameters before moving input
if let Some(ref storage_class) = input.storage_class
&& !is_valid_storage_class(storage_class.as_str())
{
return Err(s3_error!(InvalidStorageClass));
}
let PutObjectInput {
body,
bucket,
key,
content_length,
content_type,
tagging,
metadata,
version_id,
server_side_encryption,
sse_customer_algorithm,
sse_customer_key,
sse_customer_key_md5,
ssekms_key_id,
content_md5,
if_match,
if_none_match,
..
} = input;
// Validate object key
validate_object_key(&key, "PUT")?;
if if_match.is_some() || if_none_match.is_some() {
let Some(store) = new_object_layer_fn() else {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
};
match store.get_object_info(&bucket, &key, &ObjectOptions::default()).await {
Ok(info) => {
if !info.delete_marker {
if let Some(ifmatch) = if_match
&& let Some(strong_etag) = ifmatch.into_etag()
&& info
.etag
.as_ref()
.is_some_and(|etag| ETag::Strong(etag.clone()) != strong_etag)
{
return Err(s3_error!(PreconditionFailed));
}
if let Some(rematching) = if_none_match
&& let Some(strong_etag) = rematching.into_etag()
&& info
.etag
.as_ref()
.is_some_and(|etag| ETag::Strong(etag.clone()) == strong_etag)
{
return Err(s3_error!(PreconditionFailed));
}
}
}
Err(err) => {
if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
return Err(ApiError::from(err).into());
}
if if_match.is_some() && (is_err_object_not_found(&err) || is_err_version_not_found(&err)) {
return Err(ApiError::from(err).into());
}
}
}
}
// check quota for put operation
if let Some(size) = content_length
&& let Some(metadata_sys) = rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys.get()
{
let quota_checker = QuotaChecker::new(metadata_sys.clone());
match quota_checker
.check_quota(&bucket, QuotaOperation::PutObject, size as u64)
.await
{
Ok(check_result) => {
if !check_result.allowed {
return Err(S3Error::with_message(
S3ErrorCode::InvalidRequest,
format!(
"Bucket quota exceeded. Current usage: {} bytes, limit: {} bytes",
check_result.current_usage,
check_result.quota_limit.unwrap_or(0)
),
));
}
}
Err(e) => {
warn!("Quota check failed for bucket {}: {}, allowing operation", bucket, e);
}
}
}
let Some(body) = body else { return Err(s3_error!(IncompleteBody)) };
let mut size = match content_length {
Some(c) => c,
None => {
if let Some(val) = req.headers.get(AMZ_DECODED_CONTENT_LENGTH) {
match atoi::atoi::<i64>(val.as_bytes()) {
Some(x) => x,
None => return Err(s3_error!(UnexpectedContent)),
}
} else {
return Err(s3_error!(UnexpectedContent));
}
}
};
if size == -1 {
return Err(s3_error!(UnexpectedContent));
}
// Apply adaptive buffer sizing based on file size for optimal streaming performance.
// Uses workload profile configuration (enabled by default) to select appropriate buffer size.
// Buffer sizes range from 32KB to 4MB depending on file size and configured workload profile.
let buffer_size = get_buffer_size_opt_in(size);
let body = tokio::io::BufReader::with_capacity(
buffer_size,
StreamReader::new(body.map(|f| f.map_err(|e| std::io::Error::other(e.to_string())))),
);
// let body = Box::new(StreamReader::new(body.map(|f| f.map_err(|e| std::io::Error::other(e.to_string())))));
// let mut reader = PutObjReader::new(body, content_length as usize);
let store = get_validated_store(&bucket).await?;
let mut metadata = metadata.unwrap_or_default();
let object_lock_configuration = match metadata_sys::get_object_lock_config(&bucket).await {
Ok((cfg, _created)) => Some(cfg),
Err(err) => {
if err == StorageError::ConfigNotFound {
None
} else {
warn!("get_object_lock_config err {:?}", err);
return Err(S3Error::with_message(
S3ErrorCode::InternalError,
"Failed to load Object Lock configuration".to_string(),
));
}
}
};
apply_lock_retention(object_lock_configuration, &mut metadata);
if let Some(content_type) = content_type {
metadata.insert("content-type".to_string(), content_type.to_string());
}
extract_metadata_from_mime_with_object_name(&req.headers, &mut metadata, true, Some(&key));
if let Some(tags) = tagging {
metadata.insert(AMZ_OBJECT_TAGGING.to_owned(), tags.to_string());
}
let mut opts: ObjectOptions = put_opts(&bucket, &key, version_id.clone(), &req.headers, metadata.clone())
.await
.map_err(ApiError::from)?;
let mut reader: Box<dyn Reader> = Box::new(WarpReader::new(body));
let actual_size = size;
let mut md5hex = if let Some(base64_md5) = content_md5 {
let md5 = base64_simd::STANDARD
.decode_to_vec(base64_md5.as_bytes())
.map_err(|e| ApiError::from(StorageError::other(format!("Invalid content MD5: {e}"))))?;
Some(hex_simd::encode_to_string(&md5, hex_simd::AsciiCase::Lower))
} else {
None
};
let mut sha256hex = get_content_sha256(&req.headers);
if is_compressible(&req.headers, &key) && size > MIN_COMPRESSIBLE_SIZE as i64 {
let algorithm = CompressionAlgorithm::default();
metadata.insert(format!("{RESERVED_METADATA_PREFIX_LOWER}compression"), algorithm.to_string());
metadata.insert(format!("{RESERVED_METADATA_PREFIX_LOWER}actual-size",), size.to_string());
let mut hrd = HashReader::new(reader, size as i64, size as i64, md5hex, sha256hex, false).map_err(ApiError::from)?;
if let Err(err) = hrd.add_checksum_from_s3s(&req.headers, req.trailing_headers.clone(), false) {
return Err(ApiError::from(StorageError::other(format!("add_checksum error={err:?}"))).into());
}
opts.want_checksum = hrd.checksum();
opts.user_defined
.insert(format!("{RESERVED_METADATA_PREFIX_LOWER}compression"), algorithm.to_string());
opts.user_defined
.insert(format!("{RESERVED_METADATA_PREFIX_LOWER}actual-size",), size.to_string());
reader = Box::new(CompressReader::new(hrd, algorithm));
size = HashReader::SIZE_PRESERVE_LAYER;
md5hex = None;
sha256hex = None;
}
let mut reader = HashReader::new(reader, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)?;
if size >= 0 {
if let Err(err) = reader.add_checksum_from_s3s(&req.headers, req.trailing_headers.clone(), false) {
return Err(ApiError::from(StorageError::other(format!("add_checksum error={err:?}"))).into());
}
opts.want_checksum = reader.checksum();
}
// Apply encryption using unified SSE API
let encryption_request = EncryptionRequest {
bucket: &bucket,
key: &key,
server_side_encryption,
ssekms_key_id,
sse_customer_algorithm: sse_customer_algorithm.clone(),
sse_customer_key,
sse_customer_key_md5: sse_customer_key_md5.clone(),
content_size: actual_size,
part_number: None,
part_key: None,
part_nonce: None,
};
let (effective_sse, effective_kms_key_id) = match sse_encryption(encryption_request).await? {
Some(material) => {
let server_side_encryption = Some(material.server_side_encryption.clone());
let ssekms_key_id = material.kms_key_id.clone();
// Apply encryption wrapper
let encrypted_reader = material.wrap_reader(reader);
reader = HashReader::new(encrypted_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false)
.map_err(ApiError::from)?;
// Merge encryption metadata
metadata.extend(material.metadata);
(server_side_encryption, ssekms_key_id)
}
None => (None, None),
};
let mut reader = PutObjReader::new(reader);
let mt2 = metadata.clone();
opts.user_defined.extend(metadata);
let repoptions =
get_must_replicate_options(&mt2, "".to_string(), ReplicationStatusType::Empty, ReplicationType::Object, opts.clone());
let dsc = must_replicate(&bucket, &key, repoptions).await;
if dsc.replicate_any() {
let k = format!("{}{}", RESERVED_METADATA_PREFIX_LOWER, "replication-timestamp");
opts.user_defined.insert(k, jiff::Zoned::now().to_string());
let k = format!("{}{}", RESERVED_METADATA_PREFIX_LOWER, "replication-status");
opts.user_defined.insert(k, dsc.pending_status().unwrap_or_default());
}
let obj_info = store
.put_object(&bucket, &key, &mut reader, &opts)
.await
.map_err(ApiError::from)?;
// Fast in-memory update for immediate quota consistency
rustfs_ecstore::data_usage::increment_bucket_usage_memory(&bucket, obj_info.size as u64).await;
// Invalidate cache for the written object to prevent stale data
let manager = get_concurrency_manager();
let put_bucket = bucket.clone();
let put_key = key.clone();
let mut put_version = obj_info.version_id.map(|v| v.to_string());
if opts.version_suspended && obj_info.version_id.is_none_or(|v| v.is_nil()) {
put_version = Some("null".to_string());
}
helper = helper.object(obj_info.clone());
if let Some(version_id) = &put_version {
helper = helper.version_id(version_id.clone());
}
let put_version_clone = put_version.clone();
tokio::spawn(async move {
manager
.invalidate_cache_versioned(&put_bucket, &put_key, put_version_clone.as_deref())
.await;
});
let e_tag = obj_info.etag.clone().map(|etag| to_s3s_etag(&etag));
let repoptions =
get_must_replicate_options(&mt2, "".to_string(), ReplicationStatusType::Empty, ReplicationType::Object, opts);
let dsc = must_replicate(&bucket, &key, repoptions).await;
if dsc.replicate_any() {
schedule_replication(obj_info, store, dsc, ReplicationType::Object).await;
}
let mut checksum_crc32 = input.checksum_crc32;
let mut checksum_crc32c = input.checksum_crc32c;
let mut checksum_sha1 = input.checksum_sha1;
let mut checksum_sha256 = input.checksum_sha256;
let mut checksum_crc64nvme = input.checksum_crc64nvme;
if let Some(alg) = &input.checksum_algorithm
&& let Some(Some(checksum_str)) = req.trailing_headers.as_ref().map(|trailer| {
let key = match alg.as_str() {
ChecksumAlgorithm::CRC32 => rustfs_rio::ChecksumType::CRC32.key(),
ChecksumAlgorithm::CRC32C => rustfs_rio::ChecksumType::CRC32C.key(),
ChecksumAlgorithm::SHA1 => rustfs_rio::ChecksumType::SHA1.key(),
ChecksumAlgorithm::SHA256 => rustfs_rio::ChecksumType::SHA256.key(),
ChecksumAlgorithm::CRC64NVME => rustfs_rio::ChecksumType::CRC64_NVME.key(),
_ => return None,
};
trailer.read(|headers| {
headers
.get(key.unwrap_or_default())
.and_then(|value| value.to_str().ok().map(|s| s.to_string()))
})
})
{
match alg.as_str() {
ChecksumAlgorithm::CRC32 => checksum_crc32 = checksum_str,
ChecksumAlgorithm::CRC32C => checksum_crc32c = checksum_str,
ChecksumAlgorithm::SHA1 => checksum_sha1 = checksum_str,
ChecksumAlgorithm::SHA256 => checksum_sha256 = checksum_str,
ChecksumAlgorithm::CRC64NVME => checksum_crc64nvme = checksum_str,
_ => (),
}
}
let output = PutObjectOutput {
e_tag,
server_side_encryption: effective_sse, // TDD: Return effective encryption config
sse_customer_algorithm,
sse_customer_key_md5,
ssekms_key_id: effective_kms_key_id, // TDD: Return effective KMS key ID
checksum_crc32,
checksum_crc32c,
checksum_sha1,
checksum_sha256,
checksum_crc64nvme,
version_id: put_version,
..Default::default()
};
let result = Ok(S3Response::new(output));
let _ = helper.complete(&result);
result
}
async fn put_object_extract(&self, req: S3Request<PutObjectInput>) -> S3Result<S3Response<PutObjectOutput>> {
let helper = OperationHelper::new(&req, EventName::ObjectCreatedPut, "s3:PutObject").suppress_event();
let input = req.input;
let PutObjectInput {
body,
bucket,
key,
version_id,
content_length,
content_md5,
..
} = input;
let event_version_id = version_id;
let Some(body) = body else { return Err(s3_error!(IncompleteBody)) };
let size = match content_length {
Some(c) => c,
None => {
if let Some(val) = req.headers.get(AMZ_DECODED_CONTENT_LENGTH) {
match atoi::atoi::<i64>(val.as_bytes()) {
Some(x) => x,
None => return Err(s3_error!(UnexpectedContent)),
}
} else {
return Err(s3_error!(UnexpectedContent));
}
}
};
// Apply adaptive buffer sizing based on file size for optimal streaming performance.
// Uses workload profile configuration (enabled by default) to select appropriate buffer size.
// Buffer sizes range from 32KB to 4MB depending on file size and configured workload profile.
let buffer_size = get_buffer_size_opt_in(size);
let body = tokio::io::BufReader::with_capacity(
buffer_size,
StreamReader::new(body.map(|f| f.map_err(|e| std::io::Error::other(e.to_string())))),
);
let Some(ext) = Path::new(&key).extension().and_then(|s| s.to_str()) else {
return Err(s3_error!(InvalidArgument, "key extension not found"));
};
let ext = ext.to_owned();
let md5hex = if let Some(base64_md5) = content_md5 {
let md5 = base64_simd::STANDARD
.decode_to_vec(base64_md5.as_bytes())
.map_err(|e| ApiError::from(StorageError::other(format!("Invalid content MD5: {e}"))))?;
Some(hex_simd::encode_to_string(&md5, hex_simd::AsciiCase::Lower))
} else {
None
};
let sha256hex = get_content_sha256(&req.headers);
let actual_size = size;
let reader: Box<dyn Reader> = Box::new(WarpReader::new(body));
let mut hreader = HashReader::new(reader, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)?;
if let Err(err) = hreader.add_checksum_from_s3s(&req.headers, req.trailing_headers.clone(), false) {
return Err(ApiError::from(StorageError::other(format!("add_checksum error={err:?}"))).into());
}
// TODO: support zip
let decoder = CompressionFormat::from_extension(&ext).get_decoder(hreader).map_err(|e| {
error!("get_decoder err {:?}", e);
s3_error!(InvalidArgument, "get_decoder err")
})?;
let mut ar = Archive::new(decoder);
let mut entries = ar.entries().map_err(|e| {
error!("get entries err {:?}", e);
s3_error!(InvalidArgument, "get entries err")
})?;
let Some(store) = new_object_layer_fn() else {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
};
let prefix = req
.headers
.get("X-Amz-Meta-Rustfs-Snowball-Prefix")
.map(|v| v.to_str().unwrap_or_default())
.unwrap_or_default();
let version_id = match event_version_id {
Some(v) => v.to_string(),
None => String::new(),
};
while let Some(entry) = entries.next().await {
let f = match entry {
Ok(f) => f,
Err(e) => {
error!("Failed to read archive entry: {}", e);
return Err(s3_error!(InvalidArgument, "Failed to read archive entry: {:?}", e));
}
};
if f.header().entry_type().is_dir() {
continue;
}
if let Ok(fpath) = f.path() {
let mut fpath = fpath.to_string_lossy().to_string();
if !prefix.is_empty() {
fpath = format!("{prefix}/{fpath}");
}
let mut size = f.header().size().unwrap_or_default() as i64;
debug!("Extracting file: {}, size: {} bytes", fpath, size);
let mut reader: Box<dyn Reader> = Box::new(WarpReader::new(f));
let mut metadata = HashMap::new();
let actual_size = size;
if is_compressible(&HeaderMap::new(), &fpath) && size > MIN_COMPRESSIBLE_SIZE as i64 {
metadata.insert(
format!("{RESERVED_METADATA_PREFIX_LOWER}compression"),
CompressionAlgorithm::default().to_string(),
);
metadata.insert(format!("{RESERVED_METADATA_PREFIX_LOWER}actual-size",), size.to_string());
let hrd = HashReader::new(reader, size, actual_size, None, None, false).map_err(ApiError::from)?;
reader = Box::new(CompressReader::new(hrd, CompressionAlgorithm::default()));
size = HashReader::SIZE_PRESERVE_LAYER;
}
let hrd = HashReader::new(reader, size, actual_size, None, None, false).map_err(ApiError::from)?;
let mut reader = PutObjReader::new(hrd);
let _obj_info = store
.put_object(&bucket, &fpath, &mut reader, &ObjectOptions::default())
.await
.map_err(ApiError::from)?;
// Invalidate cache for the written object to prevent stale data
let manager = get_concurrency_manager();
let fpath_clone = fpath.clone();
let bucket_clone = bucket.clone();
tokio::spawn(async move {
manager.invalidate_cache_versioned(&bucket_clone, &fpath_clone, None).await;
});
let e_tag = _obj_info.etag.clone().map(|etag| to_s3s_etag(&etag));
// // store.put_object(bucket, object, data, opts);
let output = PutObjectOutput {
e_tag,
..Default::default()
};
let event_args = rustfs_notify::EventArgs {
event_name: EventName::ObjectCreatedPut,
bucket_name: bucket.clone(),
object: _obj_info.clone(),
req_params: extract_params_header(&req.headers),
resp_elements: extract_resp_elements(&S3Response::new(output.clone())),
version_id: version_id.clone(),
host: get_request_host(&req.headers),
port: get_request_port(&req.headers),
user_agent: get_request_user_agent(&req.headers),
};
// Asynchronous call will not block the response of the current request
tokio::spawn(async move {
notifier_global::notify(event_args).await;
});
}
}
// match decompress(
// body,
// CompressionFormat::from_extension(&ext),
// |entry: tokio_tar::Entry<tokio_tar::Archive<Box<dyn AsyncRead + Send + Unpin + 'static>>>| async move {
// let path = entry.path().unwrap();
// debug!("Extracted: {}", path.display());
// Ok(())
// },
// )
// .await
// {
// Ok(_) => info!("Decompression completed successfully"),
// Err(e) => error!("Decompression failed: {}", e),
// }
let mut checksum_crc32 = input.checksum_crc32;
let mut checksum_crc32c = input.checksum_crc32c;
let mut checksum_sha1 = input.checksum_sha1;
let mut checksum_sha256 = input.checksum_sha256;
let mut checksum_crc64nvme = input.checksum_crc64nvme;
if let Some(alg) = &input.checksum_algorithm
&& let Some(Some(checksum_str)) = req.trailing_headers.as_ref().map(|trailer| {
let key = match alg.as_str() {
ChecksumAlgorithm::CRC32 => rustfs_rio::ChecksumType::CRC32.key(),
ChecksumAlgorithm::CRC32C => rustfs_rio::ChecksumType::CRC32C.key(),
ChecksumAlgorithm::SHA1 => rustfs_rio::ChecksumType::SHA1.key(),
ChecksumAlgorithm::SHA256 => rustfs_rio::ChecksumType::SHA256.key(),
ChecksumAlgorithm::CRC64NVME => rustfs_rio::ChecksumType::CRC64_NVME.key(),
_ => return None,
};
trailer.read(|headers| {
headers
.get(key.unwrap_or_default())
.and_then(|value| value.to_str().ok().map(|s| s.to_string()))
})
})
{
match alg.as_str() {
ChecksumAlgorithm::CRC32 => checksum_crc32 = checksum_str,
ChecksumAlgorithm::CRC32C => checksum_crc32c = checksum_str,
ChecksumAlgorithm::SHA1 => checksum_sha1 = checksum_str,
ChecksumAlgorithm::SHA256 => checksum_sha256 = checksum_str,
ChecksumAlgorithm::CRC64NVME => checksum_crc64nvme = checksum_str,
_ => (),
}
}
warn!(
"put object extract checksum_crc32={checksum_crc32:?}, checksum_crc32c={checksum_crc32c:?}, checksum_sha1={checksum_sha1:?}, checksum_sha256={checksum_sha256:?}, checksum_crc64nvme={checksum_crc64nvme:?}",
);
// TODO: etag
let output = PutObjectOutput {
checksum_crc32,
checksum_crc32c,
checksum_sha1,
checksum_sha256,
checksum_crc64nvme,
..Default::default()
};
let result = Ok(S3Response::new(output));
let _ = helper.complete(&result);
result
}
}
rustfs/tests/lifecycle_minio_sdk_test.rs
+188
View File
@@ -0,0 +1,188 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use anyhow::{Context, Result};
use aws_config::BehaviorVersion;
use aws_sdk_s3::Client;
use aws_sdk_s3::config::{Credentials, Region};
use aws_sdk_s3::presigning::PresigningConfig;
use aws_sdk_s3::types::{
BucketLifecycleConfiguration, ExpirationStatus, LifecycleExpiration, LifecycleRule, LifecycleRuleFilter,
};
use serial_test::serial;
use std::env;
use std::time::Duration;
use uuid::Uuid;
struct Settings {
oss_endpoint: String,
oss_access_key: String,
oss_secret_key: String,
oss_bucket_name: String,
oss_lifecycle_days: i32,
#[allow(dead_code)]
oss_secure: bool,
oss_region: String,
}
impl Settings {
fn new() -> Self {
Self {
oss_endpoint: env::var("TEST_RUSTFS_SERVER").unwrap_or_else(|_| "http://localhost:9000".to_string()),
oss_access_key: "rustfsadmin".to_string(),
oss_secret_key: "rustfsadmin".to_string(),
oss_bucket_name: "mblock99".to_string(),
oss_lifecycle_days: 1,
oss_secure: false,
oss_region: "us-east-1".to_string(),
}
}
}
struct Oss {
client: Client,
bucket_name: String,
lifecycle_days: i32,
}
impl Oss {
async fn new(settings: &Settings) -> Result<Self> {
let credentials = Credentials::new(&settings.oss_access_key, &settings.oss_secret_key, None, None, "test");
let config = aws_config::defaults(BehaviorVersion::latest())
.credentials_provider(credentials)
.region(Region::new(settings.oss_region.clone()))
.endpoint_url(&settings.oss_endpoint)
.load()
.await;
let client = Client::new(&config);
Ok(Self {
client,
bucket_name: settings.oss_bucket_name.clone(),
lifecycle_days: settings.oss_lifecycle_days,
})
}
fn new_uuid(&self) -> String {
Uuid::new_v4().to_string()
}
async fn create_bucket(&self) -> Result<()> {
match self.client.head_bucket().bucket(&self.bucket_name).send().await {
Ok(_) => {
println!("Bucket {} already exists", self.bucket_name);
Ok(())
}
Err(_) => match self.client.create_bucket().bucket(&self.bucket_name).send().await {
Ok(_) => {
self.set_lifecycle_expiration(self.lifecycle_days, None).await?;
println!("Bucket {} created successfully", self.bucket_name);
Ok(())
}
Err(err) => {
let err_msg = err.to_string();
if err_msg.contains("BucketAlreadyOwnedByYou") {
println!("Bucket {} already owned by you; continuing", self.bucket_name);
Ok(())
} else {
Err(anyhow::anyhow!("Failed to create bucket: {}", err))
}
}
},
}
}
async fn set_lifecycle_expiration(&self, days: i32, prefix: Option<&str>) -> Result<()> {
let prefix_str = prefix.unwrap_or("");
let rule_id = format!("expire-{}-{}d", if prefix_str.is_empty() { "all" } else { prefix_str }, days);
let filter = LifecycleRuleFilter::builder().prefix(prefix_str.to_string()).build();
let expiration = LifecycleExpiration::builder().days(days).build();
let rule = LifecycleRule::builder()
.id(rule_id)
.status(ExpirationStatus::Enabled)
.filter(filter)
.expiration(expiration)
.build()
.context("Failed to build lifecycle rule")?;
let lifecycle_config = BucketLifecycleConfiguration::builder()
.rules(rule)
.build()
.context("Failed to build bucket lifecycle configuration")?;
self.client
.put_bucket_lifecycle_configuration()
.bucket(&self.bucket_name)
.lifecycle_configuration(lifecycle_config)
.send()
.await
.context("Failed to set bucket lifecycle")?;
Ok(())
}
#[allow(dead_code)]
async fn upload_file(&self, filename: &str, content: &[u8]) -> Result<String> {
let ext = std::path::Path::new(filename)
.extension()
.and_then(|s| s.to_str())
.unwrap_or("");
let uuid = self.new_uuid();
let object_name = if ext.is_empty() { uuid } else { format!("{}.{}", uuid, ext) };
self.client
.put_object()
.bucket(&self.bucket_name)
.key(&object_name)
.body(content.to_vec().into())
.send()
.await
.context("Failed to upload file")?;
Ok(object_name)
}
#[allow(dead_code)]
async fn get_presigned_url(&self, filename: &str) -> Result<String> {
let expires_in = Duration::from_secs((self.lifecycle_days * 24 * 60 * 60) as u64);
let presigning_config = PresigningConfig::expires_in(expires_in)?;
let presigned_req = self
.client
.get_object()
.bucket(&self.bucket_name)
.key(filename)
.presigned(presigning_config)
.await
.context("Failed to get presigned URL")?;
Ok(presigned_req.uri().to_string())
}
}
#[tokio::test]
#[serial]
#[ignore]
async fn test_lifecycle_minio_sdk() -> Result<()> {
let settings = Settings::new();
let oss = Oss::new(&settings).await?;
oss.create_bucket().await?;
Ok(())
}
rustfs/tests/lifecycle_miniosdk_test.py
-120
View File
@@ -1,120 +0,0 @@
# 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.
from datetime import timedelta
import uuid
from fastapi import UploadFile
from minio import Minio
from minio.commonconfig import Filter
from minio.error import S3Error
from minio.lifecycleconfig import Expiration, LifecycleConfig, Rule
import os
from collections import namedtuple
Settings = namedtuple("Settings", ['oss_endpoint',
'oss_access_key',
'oss_secret_key',
'oss_bucket_name',
'oss_lifecycle_days',
'oss_secure',
'oss_region'])
settings = Settings(
oss_endpoint=os.getenv("TEST_RUSTFS_SERVER", "localhost:9000"),
oss_access_key = "rustfsadmin",
oss_secret_key = "rustfsadmin",
oss_bucket_name = "mblock99",
oss_lifecycle_days = 1,
oss_secure = False,
oss_region = ""
)
class OSS:
def __init__(self):
self.bucket_name = settings.oss_bucket_name
self.lifecycle_days = settings.oss_lifecycle_days
self.client = Minio(
endpoint=settings.oss_endpoint,
access_key=settings.oss_access_key,
secret_key=settings.oss_secret_key,
secure=settings.oss_secure,
region=settings.oss_region,
)
def new_uuid(self):
return str(uuid.uuid4())
def create_bucket(self):
found = self.client.bucket_exists(self.bucket_name)
if not found:
try:
self.client.make_bucket(self.bucket_name)
self.set_lifecycle_expiration(days=self.lifecycle_days)
print(f"Bucket {self.bucket_name} created successfully")
except S3Error as exc:
if exc.code == "BucketAlreadyOwnedByYou":
print(f"Bucket {self.bucket_name} already owned by you; continuing")
else:
raise
else:
print(f"Bucket {self.bucket_name} already exists")
def set_lifecycle_expiration(self,days: int = 1, prefix: str = "") -> None:
"""
设置按天自动过期删除（MinIO 按天、每天巡检一次；<1天不生效）
"""
rule_filter = Filter(prefix=prefix or "")
rule = Rule(
rule_id=f"expire-{prefix or 'all'}-{days}d",
status="Enabled",
rule_filter=rule_filter,
expiration=Expiration(days=int(days)),
)
cfg = LifecycleConfig([rule])
self.client.set_bucket_lifecycle(self.bucket_name, cfg)
def upload_file(self, file: UploadFile):
"""
上传文件到OSS，返回文件的UUID
"""
ext = os.path.splitext(file.filename)[1]
uuid = self.new_uuid()
filename = f'{uuid}{ext}'
file.file.seek(0)
self.client.put_object(
self.bucket_name,
filename,
file.file,
length=-1,
part_size=10*1024*1024,
)
return filename
def get_presigned_url(self, filename: str):
"""
获取文件的预签名URL，用于下载文件
"""
return self.client.presigned_get_object(
self.bucket_name, filename, expires=timedelta(days=self.lifecycle_days)
)
def get_oss():
"""
获取OSS实例
"""
return OSS()
if __name__ == "__main__":
oss = get_oss()
oss.create_bucket()