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

fix: make ci and local use the same toolchain (#72)

Browse Source 
Signed-off-by: yihong0618 <zouzou0208@gmail.com>
This commit is contained in:
yihong
2025-07-07 10:40:53 +08:00
committed by GitHub
parent abd5dff9b5
commit 72aead5466
41 changed files with 1248 additions and 894 deletions
Download Patch File Download Diff File Expand all files Collapse all files
DEVELOPMENT.md
+5
View File
@@ -11,21 +11,25 @@
Before every commit, you **MUST**: Before every commit, you **MUST**:
1. **Format your code**: 1. **Format your code**:
```bash ```bash
cargo fmt --all cargo fmt --all
``` ```
2. **Verify formatting**: 2. **Verify formatting**:
```bash ```bash
cargo fmt --all --check cargo fmt --all --check
``` ```
3. **Pass clippy checks**: 3. **Pass clippy checks**:
```bash ```bash
cargo clippy --all-targets --all-features -- -D warnings cargo clippy --all-targets --all-features -- -D warnings
``` ```
4. **Ensure compilation**: 4. **Ensure compilation**:
```bash ```bash
cargo check --all-targets cargo check --all-targets
``` ```
@@ -136,6 +140,7 @@ Install the `rust-analyzer` extension and add to your `settings.json`:
#### Other IDEs #### Other IDEs
Configure your IDE to: Configure your IDE to:
- Use the project's `rustfmt.toml` configuration - Use the project's `rustfmt.toml` configuration
- Format on save - Format on save
- Run clippy checks - Run clippy checks
crates/common/src/last_minute.rs
+8 -5
View File
@@ -568,11 +568,14 @@ mod tests {
let mut latency = LastMinuteLatency::default(); let mut latency = LastMinuteLatency::default();
// Add data at time 1000 // Add data at time 1000
latency.add_all(1000, &AccElem { latency.add_all(
total: 10, 1000,
size: 0, &AccElem {
n: 1, total: 10,
}); size: 0,
n: 1,
},
);
// Forward to time 1030 (30 seconds later) // Forward to time 1030 (30 seconds later)
latency.forward_to(1030); latency.forward_to(1030);
crates/ecstore/src/bucket/metadata_sys.rs
+7 -4
View File
@@ -236,10 +236,13 @@ impl BucketMetadataSys {
futures.push(async move { futures.push(async move {
sleep(Duration::from_millis(30)).await; sleep(Duration::from_millis(30)).await;
let _ = api let _ = api
.heal_bucket(&bucket, &HealOpts { .heal_bucket(
recreate: true, &bucket,
..Default::default() &HealOpts {
}) recreate: true,
..Default::default()
},
)
.await; .await;
load_bucket_metadata(self.api.clone(), bucket.as_str()).await load_bucket_metadata(self.api.clone(), bucket.as_str()).await
}); });
crates/ecstore/src/client/api_bucket_policy.rs
+40 -34
View File
@@ -74,23 +74,26 @@ impl TransitionClient {
url_values.insert("policy".to_string(), "".to_string()); url_values.insert("policy".to_string(), "".to_string());
let resp = self let resp = self
.execute_method(http::Method::DELETE, &mut RequestMetadata { .execute_method(
bucket_name: bucket_name.to_string(), http::Method::DELETE,
query_values: url_values, &mut RequestMetadata {
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(), bucket_name: bucket_name.to_string(),
object_name: "".to_string(), query_values: url_values,
custom_header: HeaderMap::new(), content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
content_body: ReaderImpl::Body(Bytes::new()), object_name: "".to_string(),
content_length: 0, custom_header: HeaderMap::new(),
content_md5_base64: "".to_string(), content_body: ReaderImpl::Body(Bytes::new()),
stream_sha256: false, content_length: 0,
trailer: HeaderMap::new(), content_md5_base64: "".to_string(),
pre_sign_url: Default::default(), stream_sha256: false,
add_crc: Default::default(), trailer: HeaderMap::new(),
extra_pre_sign_header: Default::default(), pre_sign_url: Default::default(),
bucket_location: Default::default(), add_crc: Default::default(),
expires: Default::default(), extra_pre_sign_header: Default::default(),
}) bucket_location: Default::default(),
expires: Default::default(),
},
)
.await?; .await?;
//defer closeResponse(resp) //defer closeResponse(resp)
@@ -111,23 +114,26 @@ impl TransitionClient {
url_values.insert("policy".to_string(), "".to_string()); url_values.insert("policy".to_string(), "".to_string());
let resp = self let resp = self
.execute_method(http::Method::GET, &mut RequestMetadata { .execute_method(
bucket_name: bucket_name.to_string(), http::Method::GET,
query_values: url_values, &mut RequestMetadata {
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(), bucket_name: bucket_name.to_string(),
object_name: "".to_string(), query_values: url_values,
custom_header: HeaderMap::new(), content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
content_body: ReaderImpl::Body(Bytes::new()), object_name: "".to_string(),
content_length: 0, custom_header: HeaderMap::new(),
content_md5_base64: "".to_string(), content_body: ReaderImpl::Body(Bytes::new()),
stream_sha256: false, content_length: 0,
trailer: HeaderMap::new(), content_md5_base64: "".to_string(),
pre_sign_url: Default::default(), stream_sha256: false,
add_crc: Default::default(), trailer: HeaderMap::new(),
extra_pre_sign_header: Default::default(), pre_sign_url: Default::default(),
bucket_location: Default::default(), add_crc: Default::default(),
expires: Default::default(), extra_pre_sign_header: Default::default(),
}) bucket_location: Default::default(),
expires: Default::default(),
},
)
.await?; .await?;
let policy = String::from_utf8_lossy(&resp.body().bytes().expect("err").to_vec()).to_string(); let policy = String::from_utf8_lossy(&resp.body().bytes().expect("err").to_vec()).to_string();
crates/ecstore/src/client/api_get_object.rs
+20 -17
View File
@@ -43,23 +43,26 @@ impl TransitionClient {
opts: &GetObjectOptions, opts: &GetObjectOptions,
) -> Result<(ObjectInfo, HeaderMap, ReadCloser), std::io::Error> { ) -> Result<(ObjectInfo, HeaderMap, ReadCloser), std::io::Error> {
let resp = self let resp = self
.execute_method(http::Method::GET, &mut RequestMetadata { .execute_method(
bucket_name: bucket_name.to_string(), http::Method::GET,
object_name: object_name.to_string(), &mut RequestMetadata {
query_values: opts.to_query_values(), bucket_name: bucket_name.to_string(),
custom_header: opts.header(), object_name: object_name.to_string(),
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(), query_values: opts.to_query_values(),
content_body: ReaderImpl::Body(Bytes::new()), custom_header: opts.header(),
content_length: 0, content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
content_md5_base64: "".to_string(), content_body: ReaderImpl::Body(Bytes::new()),
stream_sha256: false, content_length: 0,
trailer: HeaderMap::new(), content_md5_base64: "".to_string(),
pre_sign_url: Default::default(), stream_sha256: false,
add_crc: Default::default(), trailer: HeaderMap::new(),
extra_pre_sign_header: Default::default(), pre_sign_url: Default::default(),
bucket_location: Default::default(), add_crc: Default::default(),
expires: Default::default(), extra_pre_sign_header: Default::default(),
}) bucket_location: Default::default(),
expires: Default::default(),
},
)
.await?; .await?;
let resp = &resp; let resp = &resp;
crates/ecstore/src/client/api_get_object_acl.rs
+20 -17
View File
@@ -63,23 +63,26 @@ impl TransitionClient {
let mut url_values = HashMap::new(); let mut url_values = HashMap::new();
url_values.insert("acl".to_string(), "".to_string()); url_values.insert("acl".to_string(), "".to_string());
let mut resp = self let mut resp = self
.execute_method(http::Method::GET, &mut RequestMetadata { .execute_method(
bucket_name: bucket_name.to_string(), http::Method::GET,
object_name: object_name.to_string(), &mut RequestMetadata {
query_values: url_values, bucket_name: bucket_name.to_string(),
custom_header: HeaderMap::new(), object_name: object_name.to_string(),
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(), query_values: url_values,
content_body: ReaderImpl::Body(Bytes::new()), custom_header: HeaderMap::new(),
content_length: 0, content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
content_md5_base64: "".to_string(), content_body: ReaderImpl::Body(Bytes::new()),
stream_sha256: false, content_length: 0,
trailer: HeaderMap::new(), content_md5_base64: "".to_string(),
pre_sign_url: Default::default(), stream_sha256: false,
add_crc: Default::default(), trailer: HeaderMap::new(),
extra_pre_sign_header: Default::default(), pre_sign_url: Default::default(),
bucket_location: Default::default(), add_crc: Default::default(),
expires: Default::default(), extra_pre_sign_header: Default::default(),
}) bucket_location: Default::default(),
expires: Default::default(),
},
)
.await?; .await?;
if resp.status() != http::StatusCode::OK { if resp.status() != http::StatusCode::OK {
crates/ecstore/src/client/api_get_object_attributes.rs
+20 -17
View File
@@ -193,23 +193,26 @@ impl TransitionClient {
}*/ }*/
let mut resp = self let mut resp = self
.execute_method(http::Method::HEAD, &mut RequestMetadata { .execute_method(
bucket_name: bucket_name.to_string(), http::Method::HEAD,
object_name: object_name.to_string(), &mut RequestMetadata {
query_values: url_values, bucket_name: bucket_name.to_string(),
custom_header: headers, object_name: object_name.to_string(),
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(), query_values: url_values,
content_md5_base64: "".to_string(), custom_header: headers,
content_body: ReaderImpl::Body(Bytes::new()), content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
content_length: 0, content_md5_base64: "".to_string(),
stream_sha256: false, content_body: ReaderImpl::Body(Bytes::new()),
trailer: HeaderMap::new(), content_length: 0,
pre_sign_url: Default::default(), stream_sha256: false,
add_crc: Default::default(), trailer: HeaderMap::new(),
extra_pre_sign_header: Default::default(), pre_sign_url: Default::default(),
bucket_location: Default::default(), add_crc: Default::default(),
expires: Default::default(), extra_pre_sign_header: Default::default(),
}) bucket_location: Default::default(),
expires: Default::default(),
},
)
.await?; .await?;
let h = resp.headers(); let h = resp.headers();
crates/ecstore/src/client/api_list.rs
+20 -17
View File
@@ -76,23 +76,26 @@ impl TransitionClient {
} }
let mut resp = self let mut resp = self
.execute_method(http::Method::GET, &mut RequestMetadata { .execute_method(
bucket_name: bucket_name.to_string(), http::Method::GET,
object_name: "".to_string(), &mut RequestMetadata {
query_values: url_values, bucket_name: bucket_name.to_string(),
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(), object_name: "".to_string(),
custom_header: headers, query_values: url_values,
content_body: ReaderImpl::Body(Bytes::new()), content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
content_length: 0, custom_header: headers,
content_md5_base64: "".to_string(), content_body: ReaderImpl::Body(Bytes::new()),
stream_sha256: false, content_length: 0,
trailer: HeaderMap::new(), content_md5_base64: "".to_string(),
pre_sign_url: Default::default(), stream_sha256: false,
add_crc: Default::default(), trailer: HeaderMap::new(),
extra_pre_sign_header: Default::default(), pre_sign_url: Default::default(),
bucket_location: Default::default(), add_crc: Default::default(),
expires: Default::default(), extra_pre_sign_header: Default::default(),
}) bucket_location: Default::default(),
expires: Default::default(),
},
)
.await?; .await?;
if resp.status() != StatusCode::OK { if resp.status() != StatusCode::OK {
return Err(std::io::Error::other(http_resp_to_error_response(resp, vec![], bucket_name, ""))); return Err(std::io::Error::other(http_resp_to_error_response(resp, vec![], bucket_name, "")));
crates/ecstore/src/client/api_remove.rs
+109 -90
View File
@@ -78,23 +78,26 @@ impl TransitionClient {
let headers = HeaderMap::new(); let headers = HeaderMap::new();
let resp = self let resp = self
.execute_method(Method::DELETE, &mut RequestMetadata { .execute_method(
bucket_name: bucket_name.to_string(), Method::DELETE,
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(), &mut RequestMetadata {
custom_header: headers, bucket_name: bucket_name.to_string(),
object_name: "".to_string(), content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
query_values: Default::default(), custom_header: headers,
content_body: ReaderImpl::Body(Bytes::new()), object_name: "".to_string(),
content_length: 0, query_values: Default::default(),
content_md5_base64: "".to_string(), content_body: ReaderImpl::Body(Bytes::new()),
stream_sha256: false, content_length: 0,
trailer: HeaderMap::new(), content_md5_base64: "".to_string(),
pre_sign_url: Default::default(), stream_sha256: false,
add_crc: Default::default(), trailer: HeaderMap::new(),
extra_pre_sign_header: Default::default(), pre_sign_url: Default::default(),
bucket_location: Default::default(), add_crc: Default::default(),
expires: Default::default(), extra_pre_sign_header: Default::default(),
}) bucket_location: Default::default(),
expires: Default::default(),
},
)
.await?; .await?;
{ {
@@ -106,23 +109,26 @@ impl TransitionClient {
pub async fn remove_bucket(&self, bucket_name: &str) -> Result<(), std::io::Error> { pub async fn remove_bucket(&self, bucket_name: &str) -> Result<(), std::io::Error> {
let resp = self let resp = self
.execute_method(http::Method::DELETE, &mut RequestMetadata { .execute_method(
bucket_name: bucket_name.to_string(), http::Method::DELETE,
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(), &mut RequestMetadata {
custom_header: Default::default(), bucket_name: bucket_name.to_string(),
object_name: "".to_string(), content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
query_values: Default::default(), custom_header: Default::default(),
content_body: ReaderImpl::Body(Bytes::new()), object_name: "".to_string(),
content_length: 0, query_values: Default::default(),
content_md5_base64: "".to_string(), content_body: ReaderImpl::Body(Bytes::new()),
stream_sha256: false, content_length: 0,
trailer: HeaderMap::new(), content_md5_base64: "".to_string(),
pre_sign_url: Default::default(), stream_sha256: false,
add_crc: Default::default(), trailer: HeaderMap::new(),
extra_pre_sign_header: Default::default(), pre_sign_url: Default::default(),
bucket_location: Default::default(), add_crc: Default::default(),
expires: Default::default(), extra_pre_sign_header: Default::default(),
}) bucket_location: Default::default(),
expires: Default::default(),
},
)
.await?; .await?;
{ {
@@ -157,23 +163,26 @@ impl TransitionClient {
} }
let resp = self let resp = self
.execute_method(http::Method::DELETE, &mut RequestMetadata { .execute_method(
bucket_name: bucket_name.to_string(), http::Method::DELETE,
object_name: object_name.to_string(), &mut RequestMetadata {
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(), bucket_name: bucket_name.to_string(),
query_values: url_values, object_name: object_name.to_string(),
custom_header: headers, content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
content_body: ReaderImpl::Body(Bytes::new()), query_values: url_values,
content_length: 0, custom_header: headers,
content_md5_base64: "".to_string(), content_body: ReaderImpl::Body(Bytes::new()),
stream_sha256: false, content_length: 0,
trailer: HeaderMap::new(), content_md5_base64: "".to_string(),
pre_sign_url: Default::default(), stream_sha256: false,
add_crc: Default::default(), trailer: HeaderMap::new(),
extra_pre_sign_header: Default::default(), pre_sign_url: Default::default(),
bucket_location: Default::default(), add_crc: Default::default(),
expires: Default::default(), extra_pre_sign_header: Default::default(),
}) bucket_location: Default::default(),
expires: Default::default(),
},
)
.await?; .await?;
Ok(RemoveObjectResult { Ok(RemoveObjectResult {
@@ -268,11 +277,15 @@ impl TransitionClient {
while let Some(object) = objects_rx.recv().await { while let Some(object) = objects_rx.recv().await {
if has_invalid_xml_char(&object.name) { if has_invalid_xml_char(&object.name) {
let remove_result = self let remove_result = self
.remove_object_inner(bucket_name, &object.name, RemoveObjectOptions { .remove_object_inner(
version_id: object.version_id.expect("err").to_string(), bucket_name,
governance_bypass: opts.governance_bypass, &object.name,
..Default::default() RemoveObjectOptions {
}) version_id: object.version_id.expect("err").to_string(),
governance_bypass: opts.governance_bypass,
..Default::default()
},
)
.await?; .await?;
let remove_result_clone = remove_result.clone(); let remove_result_clone = remove_result.clone();
if !remove_result.err.is_none() { if !remove_result.err.is_none() {
@@ -309,23 +322,26 @@ impl TransitionClient {
let remove_bytes = generate_remove_multi_objects_request(&batch); let remove_bytes = generate_remove_multi_objects_request(&batch);
let resp = self let resp = self
.execute_method(http::Method::POST, &mut RequestMetadata { .execute_method(
bucket_name: bucket_name.to_string(), http::Method::POST,
query_values: url_values.clone(), &mut RequestMetadata {
content_body: ReaderImpl::Body(Bytes::from(remove_bytes.clone())), bucket_name: bucket_name.to_string(),
content_length: remove_bytes.len() as i64, query_values: url_values.clone(),
content_md5_base64: base64_encode(&HashAlgorithm::Md5.hash_encode(&remove_bytes).as_ref()), content_body: ReaderImpl::Body(Bytes::from(remove_bytes.clone())),
content_sha256_hex: base64_encode(&HashAlgorithm::SHA256.hash_encode(&remove_bytes).as_ref()), content_length: remove_bytes.len() as i64,
custom_header: headers, content_md5_base64: base64_encode(&HashAlgorithm::Md5.hash_encode(&remove_bytes).as_ref()),
object_name: "".to_string(), content_sha256_hex: base64_encode(&HashAlgorithm::SHA256.hash_encode(&remove_bytes).as_ref()),
stream_sha256: false, custom_header: headers,
trailer: HeaderMap::new(), object_name: "".to_string(),
pre_sign_url: Default::default(), stream_sha256: false,
add_crc: Default::default(), trailer: HeaderMap::new(),
extra_pre_sign_header: Default::default(), pre_sign_url: Default::default(),
bucket_location: Default::default(), add_crc: Default::default(),
expires: Default::default(), extra_pre_sign_header: Default::default(),
}) bucket_location: Default::default(),
expires: Default::default(),
},
)
.await?; .await?;
let body_bytes: Vec<u8> = resp.body().bytes().expect("err").to_vec(); let body_bytes: Vec<u8> = resp.body().bytes().expect("err").to_vec();
@@ -353,23 +369,26 @@ impl TransitionClient {
url_values.insert("uploadId".to_string(), upload_id.to_string()); url_values.insert("uploadId".to_string(), upload_id.to_string());
let resp = self let resp = self
.execute_method(http::Method::DELETE, &mut RequestMetadata { .execute_method(
bucket_name: bucket_name.to_string(), http::Method::DELETE,
object_name: object_name.to_string(), &mut RequestMetadata {
query_values: url_values, bucket_name: bucket_name.to_string(),
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(), object_name: object_name.to_string(),
custom_header: HeaderMap::new(), query_values: url_values,
content_body: ReaderImpl::Body(Bytes::new()), content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
content_length: 0, custom_header: HeaderMap::new(),
content_md5_base64: "".to_string(), content_body: ReaderImpl::Body(Bytes::new()),
stream_sha256: false, content_length: 0,
trailer: HeaderMap::new(), content_md5_base64: "".to_string(),
pre_sign_url: Default::default(), stream_sha256: false,
add_crc: Default::default(), trailer: HeaderMap::new(),
extra_pre_sign_header: Default::default(), pre_sign_url: Default::default(),
bucket_location: Default::default(), add_crc: Default::default(),
expires: Default::default(), extra_pre_sign_header: Default::default(),
}) bucket_location: Default::default(),
expires: Default::default(),
},
)
.await?; .await?;
//if resp.is_some() { //if resp.is_some() {
if resp.status() != StatusCode::NO_CONTENT { if resp.status() != StatusCode::NO_CONTENT {
crates/ecstore/src/client/api_restore.rs
+20 -17
View File
@@ -141,23 +141,26 @@ impl TransitionClient {
let restore_request_buffer = Bytes::from(restore_request_bytes.clone()); let restore_request_buffer = Bytes::from(restore_request_bytes.clone());
let resp = self let resp = self
.execute_method(http::Method::HEAD, &mut RequestMetadata { .execute_method(
bucket_name: bucket_name.to_string(), http::Method::HEAD,
object_name: object_name.to_string(), &mut RequestMetadata {
query_values: url_values, bucket_name: bucket_name.to_string(),
custom_header: HeaderMap::new(), object_name: object_name.to_string(),
content_sha256_hex: "".to_string(), //sum_sha256_hex(&restore_request_bytes), query_values: url_values,
content_md5_base64: "".to_string(), //sum_md5_base64(&restore_request_bytes), custom_header: HeaderMap::new(),
content_body: ReaderImpl::Body(restore_request_buffer), content_sha256_hex: "".to_string(), //sum_sha256_hex(&restore_request_bytes),
content_length: restore_request_bytes.len() as i64, content_md5_base64: "".to_string(), //sum_md5_base64(&restore_request_bytes),
stream_sha256: false, content_body: ReaderImpl::Body(restore_request_buffer),
trailer: HeaderMap::new(), content_length: restore_request_bytes.len() as i64,
pre_sign_url: Default::default(), stream_sha256: false,
add_crc: Default::default(), trailer: HeaderMap::new(),
extra_pre_sign_header: Default::default(), pre_sign_url: Default::default(),
bucket_location: Default::default(), add_crc: Default::default(),
expires: Default::default(), extra_pre_sign_header: Default::default(),
}) bucket_location: Default::default(),
expires: Default::default(),
},
)
.await?; .await?;
let b = resp.body().bytes().expect("err").to_vec(); let b = resp.body().bytes().expect("err").to_vec();
crates/ecstore/src/client/api_stat.rs
+40 -34
View File
@@ -35,23 +35,26 @@ use s3s::header::{X_AMZ_DELETE_MARKER, X_AMZ_VERSION_ID};
impl TransitionClient { impl TransitionClient {
pub async fn bucket_exists(&self, bucket_name: &str) -> Result<bool, std::io::Error> { pub async fn bucket_exists(&self, bucket_name: &str) -> Result<bool, std::io::Error> {
let resp = self let resp = self
.execute_method(http::Method::HEAD, &mut RequestMetadata { .execute_method(
bucket_name: bucket_name.to_string(), http::Method::HEAD,
object_name: "".to_string(), &mut RequestMetadata {
query_values: HashMap::new(), bucket_name: bucket_name.to_string(),
custom_header: HeaderMap::new(), object_name: "".to_string(),
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(), query_values: HashMap::new(),
content_md5_base64: "".to_string(), custom_header: HeaderMap::new(),
content_body: ReaderImpl::Body(Bytes::new()), content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
content_length: 0, content_md5_base64: "".to_string(),
stream_sha256: false, content_body: ReaderImpl::Body(Bytes::new()),
trailer: HeaderMap::new(), content_length: 0,
pre_sign_url: Default::default(), stream_sha256: false,
add_crc: Default::default(), trailer: HeaderMap::new(),
extra_pre_sign_header: Default::default(), pre_sign_url: Default::default(),
bucket_location: Default::default(), add_crc: Default::default(),
expires: Default::default(), extra_pre_sign_header: Default::default(),
}) bucket_location: Default::default(),
expires: Default::default(),
},
)
.await; .await;
if let Ok(resp) = resp { if let Ok(resp) = resp {
@@ -82,23 +85,26 @@ impl TransitionClient {
} }
let resp = self let resp = self
.execute_method(http::Method::HEAD, &mut RequestMetadata { .execute_method(
bucket_name: bucket_name.to_string(), http::Method::HEAD,
object_name: object_name.to_string(), &mut RequestMetadata {
query_values: opts.to_query_values(), bucket_name: bucket_name.to_string(),
custom_header: headers, object_name: object_name.to_string(),
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(), query_values: opts.to_query_values(),
content_md5_base64: "".to_string(), custom_header: headers,
content_body: ReaderImpl::Body(Bytes::new()), content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
content_length: 0, content_md5_base64: "".to_string(),
stream_sha256: false, content_body: ReaderImpl::Body(Bytes::new()),
trailer: HeaderMap::new(), content_length: 0,
pre_sign_url: Default::default(), stream_sha256: false,
add_crc: Default::default(), trailer: HeaderMap::new(),
extra_pre_sign_header: Default::default(), pre_sign_url: Default::default(),
bucket_location: Default::default(), add_crc: Default::default(),
expires: Default::default(), extra_pre_sign_header: Default::default(),
}) bucket_location: Default::default(),
expires: Default::default(),
},
)
.await; .await;
match resp { match resp {
crates/ecstore/src/client/object_handlers_common.rs
+9 -5
View File
@@ -31,10 +31,14 @@ pub async fn delete_object_versions(api: ECStore, bucket: &str, to_del: &[Object
remaining = &[]; remaining = &[];
} }
let vc = BucketVersioningSys::get(bucket).await.expect("err!"); let vc = BucketVersioningSys::get(bucket).await.expect("err!");
let _deleted_objs = api.delete_objects(bucket, to_del.to_vec(), ObjectOptions { let _deleted_objs = api.delete_objects(
//prefix_enabled_fn: vc.prefix_enabled(""), bucket,
version_suspended: vc.suspended(), to_del.to_vec(),
..Default::default() ObjectOptions {
}); //prefix_enabled_fn: vc.prefix_enabled(""),
version_suspended: vc.suspended(),
..Default::default()
},
);
} }
} }
crates/ecstore/src/config/com.rs
+18 -9
View File
@@ -70,20 +70,29 @@ pub async fn read_config_with_metadata<S: StorageAPI>(
} }
pub async fn save_config<S: StorageAPI>(api: Arc<S>, file: &str, data: Vec<u8>) -> Result<()> { pub async fn save_config<S: StorageAPI>(api: Arc<S>, file: &str, data: Vec<u8>) -> Result<()> {
save_config_with_opts(api, file, data, &ObjectOptions { save_config_with_opts(
max_parity: true, api,
..Default::default() file,
}) data,
&ObjectOptions {
max_parity: true,
..Default::default()
},
)
.await .await
} }
pub async fn delete_config<S: StorageAPI>(api: Arc<S>, file: &str) -> Result<()> { pub async fn delete_config<S: StorageAPI>(api: Arc<S>, file: &str) -> Result<()> {
match api match api
.delete_object(RUSTFS_META_BUCKET, file, ObjectOptions { .delete_object(
delete_prefix: true, RUSTFS_META_BUCKET,
delete_prefix_object: true, file,
..Default::default() ObjectOptions {
}) delete_prefix: true,
delete_prefix_object: true,
..Default::default()
},
)
.await .await
{ {
Ok(_) => Ok(()), Ok(_) => Ok(()),
crates/ecstore/src/disk/local.rs
+9 -5
View File
@@ -2026,11 +2026,15 @@ impl DiskAPI for LocalDisk {
) -> Result<()> { ) -> Result<()> {
if path.starts_with(SLASH_SEPARATOR) { if path.starts_with(SLASH_SEPARATOR) {
return self return self
.delete(volume, path, DeleteOptions { .delete(
recursive: false, volume,
immediate: false, path,
..Default::default() DeleteOptions {
}) recursive: false,
immediate: false,
..Default::default()
},
)
.await; .await;
} }
crates/ecstore/src/erasure_coding/bitrot.rs
+7 -4
View File
@@ -317,10 +317,13 @@ enum WriterType {
impl std::fmt::Debug for BitrotWriterWrapper { impl std::fmt::Debug for BitrotWriterWrapper {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("BitrotWriterWrapper") f.debug_struct("BitrotWriterWrapper")
.field("writer_type", &match self.writer_type { .field(
WriterType::InlineBuffer => "InlineBuffer", "writer_type",
WriterType::Other => "Other", &match self.writer_type {
}) WriterType::InlineBuffer => "InlineBuffer",
WriterType::Other => "Other",
},
)
.finish() .finish()
} }
} }
crates/ecstore/src/heal/data_usage.rs
+12 -8
View File
@@ -174,10 +174,13 @@ pub async fn load_data_usage_from_backend(store: Arc<ECStore>) -> Result<DataUsa
.bucket_sizes .bucket_sizes
.iter() .iter()
.map(|(bucket, &size)| { .map(|(bucket, &size)| {
(bucket.clone(), BucketUsageInfo { (
size, bucket.clone(),
..Default::default() BucketUsageInfo {
}) size,
..Default::default()
},
)
}) })
.collect(); .collect();
} }
@@ -198,16 +201,17 @@ pub async fn load_data_usage_from_backend(store: Arc<ECStore>) -> Result<DataUsa
{ {
if let Ok((cfg, _)) = get_replication_config(bucket).await { if let Ok((cfg, _)) = get_replication_config(bucket).await {
if !cfg.role.is_empty() { if !cfg.role.is_empty() {
data_usage_info data_usage_info.replication_info.insert(
.replication_info cfg.role.clone(),
.insert(cfg.role.clone(), BucketTargetUsageInfo { BucketTargetUsageInfo {
replication_failed_size: bui.replication_failed_size_v1, replication_failed_size: bui.replication_failed_size_v1,
replication_failed_count: bui.replication_failed_count_v1, replication_failed_count: bui.replication_failed_count_v1,
replicated_size: bui.replicated_size_v1, replicated_size: bui.replicated_size_v1,
replication_pending_count: bui.replication_pending_count_v1, replication_pending_count: bui.replication_pending_count_v1,
replication_pending_size: bui.replication_pending_size_v1, replication_pending_size: bui.replication_pending_size_v1,
..Default::default() ..Default::default()
}); },
);
} }
} }
} }
crates/ecstore/src/heal/data_usage_cache.rs
+40 -22
View File
@@ -183,11 +183,14 @@ impl AllTierStats {
fn populate_stats(&self, stats: &mut HashMap<String, TierStats>) { fn populate_stats(&self, stats: &mut HashMap<String, TierStats>) {
for (tier, st) in &self.tiers { for (tier, st) in &self.tiers {
stats.insert(tier.clone(), TierStats { stats.insert(
total_size: st.total_size, tier.clone(),
num_versions: st.num_versions, TierStats {
num_objects: st.num_objects, total_size: st.total_size,
}); num_versions: st.num_versions,
num_objects: st.num_objects,
},
);
} }
} }
} }
@@ -443,10 +446,16 @@ impl DataUsageCache {
let path = Path::new(BUCKET_META_PREFIX).join(name); let path = Path::new(BUCKET_META_PREFIX).join(name);
// warn!("Loading data usage cache from backend: {}", path.display()); // warn!("Loading data usage cache from backend: {}", path.display());
match store match store
.get_object_reader(RUSTFS_META_BUCKET, path.to_str().unwrap(), None, HeaderMap::new(), &ObjectOptions { .get_object_reader(
no_lock: true, RUSTFS_META_BUCKET,
..Default::default() path.to_str().unwrap(),
}) None,
HeaderMap::new(),
&ObjectOptions {
no_lock: true,
..Default::default()
},
)
.await .await
{ {
Ok(mut reader) => { Ok(mut reader) => {
@@ -460,10 +469,16 @@ impl DataUsageCache {
match err { match err {
Error::FileNotFound | Error::VolumeNotFound => { Error::FileNotFound | Error::VolumeNotFound => {
match store match store
.get_object_reader(RUSTFS_META_BUCKET, name, None, HeaderMap::new(), &ObjectOptions { .get_object_reader(
no_lock: true, RUSTFS_META_BUCKET,
..Default::default() name,
}) None,
HeaderMap::new(),
&ObjectOptions {
no_lock: true,
..Default::default()
},
)
.await .await
{ {
Ok(mut reader) => { Ok(mut reader) => {
@@ -804,15 +819,18 @@ impl DataUsageCache {
bui.replica_count = rs.replica_count; bui.replica_count = rs.replica_count;
for (arn, stat) in rs.targets.iter() { for (arn, stat) in rs.targets.iter() {
bui.replication_info.insert(arn.clone(), BucketTargetUsageInfo { bui.replication_info.insert(
replication_pending_size: stat.pending_size, arn.clone(),
replicated_size: stat.replicated_size, BucketTargetUsageInfo {
replication_failed_size: stat.failed_size, replication_pending_size: stat.pending_size,
replication_pending_count: stat.pending_count, replicated_size: stat.replicated_size,
replication_failed_count: stat.failed_count, replication_failed_size: stat.failed_size,
replicated_count: stat.replicated_count, replication_pending_count: stat.pending_count,
..Default::default() replication_failed_count: stat.failed_count,
}); replicated_count: stat.replicated_count,
..Default::default()
},
);
} }
} }
dst.insert(bucket.name.clone(), bui); dst.insert(bucket.name.clone(), bui);
crates/ecstore/src/heal/heal_commands.rs
+9 -5
View File
@@ -255,11 +255,15 @@ impl HealingTracker {
pub async fn delete(&self) -> Result<()> { pub async fn delete(&self) -> Result<()> {
if let Some(disk) = &self.disk { if let Some(disk) = &self.disk {
let file_path = Path::new(BUCKET_META_PREFIX).join(HEALING_TRACKER_FILENAME); let file_path = Path::new(BUCKET_META_PREFIX).join(HEALING_TRACKER_FILENAME);
disk.delete(RUSTFS_META_BUCKET, file_path.to_str().unwrap(), DeleteOptions { disk.delete(
recursive: false, RUSTFS_META_BUCKET,
immediate: false, file_path.to_str().unwrap(),
..Default::default() DeleteOptions {
}) recursive: false,
immediate: false,
..Default::default()
},
)
.await?; .await?;
} }
crates/ecstore/src/metrics_realtime.rs
+8 -5
View File
@@ -148,11 +148,14 @@ async fn collect_local_disks_metrics(disks: &HashSet<String>) -> HashMap<String,
} }
if d.state != *DRIVE_STATE_OK && d.state != *DRIVE_STATE_UNFORMATTED { if d.state != *DRIVE_STATE_OK && d.state != *DRIVE_STATE_UNFORMATTED {
metrics.insert(d.endpoint.clone(), DiskMetric { metrics.insert(
n_disks: 1, d.endpoint.clone(),
offline: 1, DiskMetric {
..Default::default() n_disks: 1,
}); offline: 1,
..Default::default()
},
);
continue; continue;
} }
crates/ecstore/src/pools.rs
+105 -69
View File
@@ -367,11 +367,14 @@ impl PoolMeta {
} }
decom_started = true; decom_started = true;
} }
remembered_pools.insert(pool.cmd_line.clone(), PoolInfo { remembered_pools.insert(
position: idx, pool.cmd_line.clone(),
completed: complete, PoolInfo {
decom_started, position: idx,
}); completed: complete,
decom_started,
},
);
} }
let mut specified_pools = HashMap::new(); let mut specified_pools = HashMap::new();
@@ -722,16 +725,20 @@ impl ECStore {
if version.deleted { if version.deleted {
// TODO: other params // TODO: other params
if let Err(err) = self if let Err(err) = self
.delete_object(bucket.as_str(), &version.name, ObjectOptions { .delete_object(
versioned: true, bucket.as_str(),
version_id: version_id.clone(), &version.name,
mod_time: version.mod_time, ObjectOptions {
src_pool_idx: idx, versioned: true,
data_movement: true, version_id: version_id.clone(),
delete_marker: true, mod_time: version.mod_time,
skip_decommissioned: true, src_pool_idx: idx,
..Default::default() data_movement: true,
}) delete_marker: true,
skip_decommissioned: true,
..Default::default()
},
)
.await .await
{ {
if is_err_object_not_found(&err) || is_err_version_not_found(&err) || is_err_data_movement_overwrite(&err) { if is_err_object_not_found(&err) || is_err_version_not_found(&err) || is_err_data_movement_overwrite(&err) {
@@ -844,12 +851,16 @@ impl ECStore {
if decommissioned == fivs.versions.len() { if decommissioned == fivs.versions.len() {
if let Err(err) = set if let Err(err) = set
.delete_object(bucket.as_str(), &encode_dir_object(&entry.name), ObjectOptions { .delete_object(
delete_prefix: true, bucket.as_str(),
delete_prefix_object: true, &encode_dir_object(&entry.name),
ObjectOptions {
delete_prefix: true,
delete_prefix_object: true,
..Default::default() ..Default::default()
}) },
)
.await .await
{ {
error!("decommission_pool: delete_object err {:?}", &err); error!("decommission_pool: delete_object err {:?}", &err);
@@ -1187,13 +1198,17 @@ impl ECStore {
if object_info.is_multipart() { if object_info.is_multipart() {
let res = match self let res = match self
.new_multipart_upload(&bucket, &object_info.name, &ObjectOptions { .new_multipart_upload(
version_id: object_info.version_id.as_ref().map(|v| v.to_string()), &bucket,
user_defined: object_info.user_defined.clone(), &object_info.name,
src_pool_idx: pool_idx, &ObjectOptions {
data_movement: true, version_id: object_info.version_id.as_ref().map(|v| v.to_string()),
..Default::default() user_defined: object_info.user_defined.clone(),
}) src_pool_idx: pool_idx,
data_movement: true,
..Default::default()
},
)
.await .await
{ {
Ok(res) => res, Ok(res) => res,
@@ -1224,10 +1239,17 @@ impl ECStore {
let mut data = PutObjReader::from_vec(chunk); let mut data = PutObjReader::from_vec(chunk);
let pi = match self let pi = match self
.put_object_part(&bucket, &object_info.name, &res.upload_id, part.number, &mut data, &ObjectOptions { .put_object_part(
preserve_etag: Some(part.etag.clone()), &bucket,
..Default::default() &object_info.name,
}) &res.upload_id,
part.number,
&mut data,
&ObjectOptions {
preserve_etag: Some(part.etag.clone()),
..Default::default()
},
)
.await .await
{ {
Ok(pi) => pi, Ok(pi) => pi,
@@ -1247,11 +1269,17 @@ impl ECStore {
if let Err(err) = self if let Err(err) = self
.clone() .clone()
.complete_multipart_upload(&bucket, &object_info.name, &res.upload_id, parts, &ObjectOptions { .complete_multipart_upload(
data_movement: true, &bucket,
mod_time: object_info.mod_time, &object_info.name,
..Default::default() &res.upload_id,
}) parts,
&ObjectOptions {
data_movement: true,
mod_time: object_info.mod_time,
..Default::default()
},
)
.await .await
{ {
error!("decommission_object: complete_multipart_upload err {:?}", &err); error!("decommission_object: complete_multipart_upload err {:?}", &err);
@@ -1267,16 +1295,21 @@ impl ECStore {
let mut data = PutObjReader::new(hrd); let mut data = PutObjReader::new(hrd);
if let Err(err) = self if let Err(err) = self
.put_object(&bucket, &object_info.name, &mut data, &ObjectOptions { .put_object(
src_pool_idx: pool_idx, &bucket,
data_movement: true, &object_info.name,
version_id: object_info.version_id.as_ref().map(|v| v.to_string()), &mut data,
mod_time: object_info.mod_time, &ObjectOptions {
user_defined: object_info.user_defined.clone(), src_pool_idx: pool_idx,
preserve_etag: object_info.etag.clone(), data_movement: true,
version_id: object_info.version_id.as_ref().map(|v| v.to_string()),
mod_time: object_info.mod_time,
user_defined: object_info.user_defined.clone(),
preserve_etag: object_info.etag.clone(),
..Default::default() ..Default::default()
}) },
)
.await .await
{ {
error!("decommission_object: put_object err {:?}", &err); error!("decommission_object: put_object err {:?}", &err);
@@ -1316,31 +1349,34 @@ impl SetDisks {
let cb1 = cb_func.clone(); let cb1 = cb_func.clone();
list_path_raw(rx, ListPathRawOptions { list_path_raw(
disks: disks.iter().cloned().map(Some).collect(), rx,
bucket: bucket_info.name.clone(), ListPathRawOptions {
path: bucket_info.prefix.clone(), disks: disks.iter().cloned().map(Some).collect(),
recursice: true, bucket: bucket_info.name.clone(),
min_disks: listing_quorum, path: bucket_info.prefix.clone(),
agreed: Some(Box::new(move |entry: MetaCacheEntry| Box::pin(cb1(entry)))), recursice: true,
partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| { min_disks: listing_quorum,
let resolver = resolver.clone(); agreed: Some(Box::new(move |entry: MetaCacheEntry| Box::pin(cb1(entry)))),
let cb_func = cb_func.clone(); partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| {
match entries.resolve(resolver) { let resolver = resolver.clone();
Some(entry) => { let cb_func = cb_func.clone();
warn!("decommission_pool: list_objects_to_decommission get {}", &entry.name); match entries.resolve(resolver) {
Box::pin(async move { Some(entry) => {
cb_func(entry).await; warn!("decommission_pool: list_objects_to_decommission get {}", &entry.name);
}) Box::pin(async move {
cb_func(entry).await;
})
}
None => {
warn!("decommission_pool: list_objects_to_decommission get none");
Box::pin(async {})
}
} }
None => { })),
warn!("decommission_pool: list_objects_to_decommission get none"); ..Default::default()
Box::pin(async {}) },
} )
}
})),
..Default::default()
})
.await?; .await?;
Ok(()) Ok(())
crates/ecstore/src/rebalance.rs
+98 -65
View File
@@ -774,16 +774,20 @@ impl ECStore {
let mut error = None; let mut error = None;
if version.deleted { if version.deleted {
if let Err(err) = set if let Err(err) = set
.delete_object(&bucket, &version.name, ObjectOptions { .delete_object(
versioned: true, &bucket,
version_id: version_id.clone(), &version.name,
mod_time: version.mod_time, ObjectOptions {
src_pool_idx: pool_index, versioned: true,
data_movement: true, version_id: version_id.clone(),
delete_marker: true, mod_time: version.mod_time,
skip_decommissioned: true, src_pool_idx: pool_index,
..Default::default() data_movement: true,
}) delete_marker: true,
skip_decommissioned: true,
..Default::default()
},
)
.await .await
{ {
if is_err_object_not_found(&err) || is_err_version_not_found(&err) || is_err_data_movement_overwrite(&err) { if is_err_object_not_found(&err) || is_err_version_not_found(&err) || is_err_data_movement_overwrite(&err) {
@@ -879,12 +883,16 @@ impl ECStore {
if rebalanced == fivs.versions.len() { if rebalanced == fivs.versions.len() {
if let Err(err) = set if let Err(err) = set
.delete_object(bucket.as_str(), &encode_dir_object(&entry.name), ObjectOptions { .delete_object(
delete_prefix: true, bucket.as_str(),
delete_prefix_object: true, &encode_dir_object(&entry.name),
ObjectOptions {
delete_prefix: true,
delete_prefix_object: true,
..Default::default() ..Default::default()
}) },
)
.await .await
{ {
error!("rebalance_entry: delete_object err {:?}", &err); error!("rebalance_entry: delete_object err {:?}", &err);
@@ -903,13 +911,17 @@ impl ECStore {
if object_info.is_multipart() { if object_info.is_multipart() {
let res = match self let res = match self
.new_multipart_upload(&bucket, &object_info.name, &ObjectOptions { .new_multipart_upload(
version_id: object_info.version_id.as_ref().map(|v| v.to_string()), &bucket,
user_defined: object_info.user_defined.clone(), &object_info.name,
src_pool_idx: pool_idx, &ObjectOptions {
data_movement: true, version_id: object_info.version_id.as_ref().map(|v| v.to_string()),
..Default::default() user_defined: object_info.user_defined.clone(),
}) src_pool_idx: pool_idx,
data_movement: true,
..Default::default()
},
)
.await .await
{ {
Ok(res) => res, Ok(res) => res,
@@ -943,10 +955,17 @@ impl ECStore {
let mut data = PutObjReader::from_vec(chunk); let mut data = PutObjReader::from_vec(chunk);
let pi = match self let pi = match self
.put_object_part(&bucket, &object_info.name, &res.upload_id, part.number, &mut data, &ObjectOptions { .put_object_part(
preserve_etag: Some(part.etag.clone()), &bucket,
..Default::default() &object_info.name,
}) &res.upload_id,
part.number,
&mut data,
&ObjectOptions {
preserve_etag: Some(part.etag.clone()),
..Default::default()
},
)
.await .await
{ {
Ok(pi) => pi, Ok(pi) => pi,
@@ -964,11 +983,17 @@ impl ECStore {
if let Err(err) = self if let Err(err) = self
.clone() .clone()
.complete_multipart_upload(&bucket, &object_info.name, &res.upload_id, parts, &ObjectOptions { .complete_multipart_upload(
data_movement: true, &bucket,
mod_time: object_info.mod_time, &object_info.name,
..Default::default() &res.upload_id,
}) parts,
&ObjectOptions {
data_movement: true,
mod_time: object_info.mod_time,
..Default::default()
},
)
.await .await
{ {
error!("rebalance_object: complete_multipart_upload err {:?}", &err); error!("rebalance_object: complete_multipart_upload err {:?}", &err);
@@ -983,16 +1008,21 @@ impl ECStore {
let mut data = PutObjReader::new(hrd); let mut data = PutObjReader::new(hrd);
if let Err(err) = self if let Err(err) = self
.put_object(&bucket, &object_info.name, &mut data, &ObjectOptions { .put_object(
src_pool_idx: pool_idx, &bucket,
data_movement: true, &object_info.name,
version_id: object_info.version_id.as_ref().map(|v| v.to_string()), &mut data,
mod_time: object_info.mod_time, &ObjectOptions {
user_defined: object_info.user_defined.clone(), src_pool_idx: pool_idx,
preserve_etag: object_info.etag.clone(), data_movement: true,
version_id: object_info.version_id.as_ref().map(|v| v.to_string()),
mod_time: object_info.mod_time,
user_defined: object_info.user_defined.clone(),
preserve_etag: object_info.etag.clone(),
..Default::default() ..Default::default()
}) },
)
.await .await
{ {
error!("rebalance_object: put_object err {:?}", &err); error!("rebalance_object: put_object err {:?}", &err);
@@ -1137,33 +1167,36 @@ impl SetDisks {
}; };
let cb1 = cb.clone(); let cb1 = cb.clone();
list_path_raw(rx, ListPathRawOptions { list_path_raw(
disks: disks.iter().cloned().map(Some).collect(), rx,
bucket: bucket.clone(), ListPathRawOptions {
recursice: true, disks: disks.iter().cloned().map(Some).collect(),
min_disks: listing_quorum, bucket: bucket.clone(),
agreed: Some(Box::new(move |entry: MetaCacheEntry| { recursice: true,
info!("list_objects_to_rebalance: agreed: {:?}", &entry.name); min_disks: listing_quorum,
Box::pin(cb1(entry)) agreed: Some(Box::new(move |entry: MetaCacheEntry| {
})), info!("list_objects_to_rebalance: agreed: {:?}", &entry.name);
partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| { Box::pin(cb1(entry))
// let cb = cb.clone(); })),
let resolver = resolver.clone(); partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| {
let cb = cb.clone(); // let cb = cb.clone();
let resolver = resolver.clone();
let cb = cb.clone();
match entries.resolve(resolver) { match entries.resolve(resolver) {
Some(entry) => { Some(entry) => {
info!("list_objects_to_rebalance: list_objects_to_decommission get {}", &entry.name); info!("list_objects_to_rebalance: list_objects_to_decommission get {}", &entry.name);
Box::pin(async move { cb(entry).await }) Box::pin(async move { cb(entry).await })
}
None => {
info!("list_objects_to_rebalance: list_objects_to_decommission get none");
Box::pin(async {})
}
} }
None => { })),
info!("list_objects_to_rebalance: list_objects_to_decommission get none"); ..Default::default()
Box::pin(async {}) },
} )
}
})),
..Default::default()
})
.await?; .await?;
info!("list_objects_to_rebalance: list_objects_to_rebalance done"); info!("list_objects_to_rebalance: list_objects_to_rebalance done");
crates/ecstore/src/rpc/tonic_service.rs
+7 -4
View File
@@ -262,10 +262,13 @@ impl Node for NodeService {
let request = request.into_inner(); let request = request.into_inner();
match self match self
.local_peer .local_peer
.delete_bucket(&request.bucket, &DeleteBucketOptions { .delete_bucket(
force: false, &request.bucket,
..Default::default() &DeleteBucketOptions {
}) force: false,
..Default::default()
},
)
.await .await
{ {
Ok(_) => Ok(tonic::Response::new(DeleteBucketResponse { Ok(_) => Ok(tonic::Response::new(DeleteBucketResponse {
crates/ecstore/src/set_disk.rs
+139 -87
View File
@@ -406,11 +406,17 @@ impl SetDisks {
let src_object = src_object.clone(); let src_object = src_object.clone();
futures.push(tokio::spawn(async move { futures.push(tokio::spawn(async move {
let _ = disk let _ = disk
.delete_version(&src_bucket, &src_object, fi, false, DeleteOptions { .delete_version(
undo_write: true, &src_bucket,
old_data_dir, &src_object,
..Default::default() fi,
}) false,
DeleteOptions {
undo_write: true,
old_data_dir,
..Default::default()
},
)
.await .await
.map_err(|e| { .map_err(|e| {
debug!("rename_data delete_version err {:?}", e); debug!("rename_data delete_version err {:?}", e);
@@ -483,10 +489,14 @@ impl SetDisks {
tokio::spawn(async move { tokio::spawn(async move {
if let Some(disk) = disk { if let Some(disk) = disk {
(disk (disk
.delete(&bucket, &file_path, DeleteOptions { .delete(
recursive: true, &bucket,
..Default::default() &file_path,
}) DeleteOptions {
recursive: true,
..Default::default()
},
)
.await) .await)
.err() .err()
} else { } else {
@@ -685,10 +695,14 @@ impl SetDisks {
if let Some(disk) = disks[i].as_ref() { if let Some(disk) = disks[i].as_ref() {
let _ = disk let _ = disk
.delete(bucket, &path_join_buf(&[prefix, STORAGE_FORMAT_FILE]), DeleteOptions { .delete(
recursive: true, bucket,
..Default::default() &path_join_buf(&[prefix, STORAGE_FORMAT_FILE]),
}) DeleteOptions {
recursive: true,
..Default::default()
},
)
.await .await
.map_err(|e| { .map_err(|e| {
warn!("write meta revert err {:?}", e); warn!("write meta revert err {:?}", e);
@@ -1667,10 +1681,14 @@ impl SetDisks {
for disk in disks.iter() { for disk in disks.iter() {
futures.push(async move { futures.push(async move {
if let Some(disk) = disk { if let Some(disk) = disk {
disk.delete(bucket, prefix, DeleteOptions { disk.delete(
recursive: true, bucket,
..Default::default() prefix,
}) DeleteOptions {
recursive: true,
..Default::default()
},
)
.await .await
} else { } else {
Err(DiskError::DiskNotFound) Err(DiskError::DiskNotFound)
@@ -2391,10 +2409,17 @@ impl SetDisks {
// Allow for dangling deletes, on versions that have DataDir missing etc. // Allow for dangling deletes, on versions that have DataDir missing etc.
// this would end up restoring the correct readable versions. // this would end up restoring the correct readable versions.
return match self return match self
.delete_if_dang_ling(bucket, object, &parts_metadata, &errs, &data_errs_by_part, ObjectOptions { .delete_if_dang_ling(
version_id: version_id_op.clone(), bucket,
..Default::default() object,
}) &parts_metadata,
&errs,
&data_errs_by_part,
ObjectOptions {
version_id: version_id_op.clone(),
..Default::default()
},
)
.await .await
{ {
Ok(m) => { Ok(m) => {
@@ -2700,11 +2725,15 @@ impl SetDisks {
if parts_metadata[index].is_remote() { if parts_metadata[index].is_remote() {
let rm_data_dir = parts_metadata[index].data_dir.unwrap().to_string(); let rm_data_dir = parts_metadata[index].data_dir.unwrap().to_string();
let d_path = Path::new(&encode_dir_object(object)).join(rm_data_dir); let d_path = Path::new(&encode_dir_object(object)).join(rm_data_dir);
disk.delete(bucket, d_path.to_str().unwrap(), DeleteOptions { disk.delete(
immediate: true, bucket,
recursive: true, d_path.to_str().unwrap(),
..Default::default() DeleteOptions {
}) immediate: true,
recursive: true,
..Default::default()
},
)
.await?; .await?;
} }
@@ -2727,10 +2756,17 @@ impl SetDisks {
Err(err) => { Err(err) => {
let data_errs_by_part = HashMap::new(); let data_errs_by_part = HashMap::new();
match self match self
.delete_if_dang_ling(bucket, object, &parts_metadata, &errs, &data_errs_by_part, ObjectOptions { .delete_if_dang_ling(
version_id: version_id_op.clone(), bucket,
..Default::default() object,
}) &parts_metadata,
&errs,
&data_errs_by_part,
ObjectOptions {
version_id: version_id_op.clone(),
..Default::default()
},
)
.await .await
{ {
Ok(m) => { Ok(m) => {
@@ -2786,11 +2822,15 @@ impl SetDisks {
let object = object.to_string(); let object = object.to_string();
futures.push(tokio::spawn(async move { futures.push(tokio::spawn(async move {
let _ = disk let _ = disk
.delete(&bucket, &object, DeleteOptions { .delete(
recursive: false, &bucket,
immediate: false, &object,
..Default::default() DeleteOptions {
}) recursive: false,
immediate: false,
..Default::default()
},
)
.await; .await;
})); }));
} }
@@ -3485,11 +3525,16 @@ impl SetDisks {
Ok(fivs) => fivs, Ok(fivs) => fivs,
Err(err) => { Err(err) => {
match self_clone match self_clone
.heal_object(&bucket, &encoded_entry_name, "", &HealOpts { .heal_object(
scan_mode, &bucket,
remove: HEAL_DELETE_DANGLING, &encoded_entry_name,
..Default::default() "",
}) &HealOpts {
scan_mode,
remove: HEAL_DELETE_DANGLING,
..Default::default()
},
)
.await .await
{ {
Ok((res, None)) => { Ok((res, None)) => {
@@ -3606,53 +3651,56 @@ impl SetDisks {
let bucket_partial = bucket.clone(); let bucket_partial = bucket.clone();
let heal_entry_agree = heal_entry.clone(); let heal_entry_agree = heal_entry.clone();
let heal_entry_partial = heal_entry.clone(); let heal_entry_partial = heal_entry.clone();
if let Err(err) = list_path_raw(rx, ListPathRawOptions { if let Err(err) = list_path_raw(
disks, rx,
fallback_disks, ListPathRawOptions {
bucket: bucket.clone(), disks,
recursice: true, fallback_disks,
forward_to, bucket: bucket.clone(),
min_disks: 1, recursice: true,
report_not_found: false, forward_to,
agreed: Some(Box::new(move |entry: MetaCacheEntry| { min_disks: 1,
let jt = jt_agree.clone(); report_not_found: false,
let bucket = bucket_agree.clone(); agreed: Some(Box::new(move |entry: MetaCacheEntry| {
let heal_entry = heal_entry_agree.clone(); let jt = jt_agree.clone();
Box::pin(async move { let bucket = bucket_agree.clone();
jt.take().await; let heal_entry = heal_entry_agree.clone();
let bucket = bucket.clone(); Box::pin(async move {
tokio::spawn(async move {
heal_entry(bucket, entry).await;
});
})
})),
partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| {
let jt = jt_partial.clone();
let bucket = bucket_partial.clone();
let heal_entry = heal_entry_partial.clone();
Box::pin({
let heal_entry = heal_entry.clone();
let resolver = resolver.clone();
async move {
let entry = if let Some(entry) = entries.resolve(resolver) {
entry
} else if let (Some(entry), _) = entries.first_found() {
entry
} else {
return;
};
jt.take().await; jt.take().await;
let bucket = bucket.clone(); let bucket = bucket.clone();
let heal_entry = heal_entry.clone();
tokio::spawn(async move { tokio::spawn(async move {
heal_entry(bucket, entry).await; heal_entry(bucket, entry).await;
}); });
} })
}) })),
})), partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| {
finished: None, let jt = jt_partial.clone();
..Default::default() let bucket = bucket_partial.clone();
}) let heal_entry = heal_entry_partial.clone();
Box::pin({
let heal_entry = heal_entry.clone();
let resolver = resolver.clone();
async move {
let entry = if let Some(entry) = entries.resolve(resolver) {
entry
} else if let (Some(entry), _) = entries.first_found() {
entry
} else {
return;
};
jt.take().await;
let bucket = bucket.clone();
let heal_entry = heal_entry.clone();
tokio::spawn(async move {
heal_entry(bucket, entry).await;
});
}
})
})),
finished: None,
..Default::default()
},
)
.await .await
{ {
ret_err = Some(err.into()); ret_err = Some(err.into());
@@ -3695,11 +3743,15 @@ impl SetDisks {
let prefix = prefix.to_string(); let prefix = prefix.to_string();
futures.push(async move { futures.push(async move {
if let Some(disk) = disk_op { if let Some(disk) = disk_op {
disk.delete(&bucket, &prefix, DeleteOptions { disk.delete(
recursive: true, &bucket,
immediate: true, &prefix,
..Default::default() DeleteOptions {
}) recursive: true,
immediate: true,
..Default::default()
},
)
.await .await
} else { } else {
Ok(()) Ok(())
crates/ecstore/src/sets.rs
+15 -9
View File
@@ -557,11 +557,14 @@ impl StorageAPI for Sets {
let idx = self.get_hashed_set_index(obj.object_name.as_str()); let idx = self.get_hashed_set_index(obj.object_name.as_str());
if !set_obj_map.contains_key(&idx) { if !set_obj_map.contains_key(&idx) {
set_obj_map.insert(idx, vec![DelObj { set_obj_map.insert(
// set_idx: idx, idx,
orig_idx: i, vec![DelObj {
obj: obj.clone(), // set_idx: idx,
}]); orig_idx: i,
obj: obj.clone(),
}],
);
} else if let Some(val) = set_obj_map.get_mut(&idx) { } else if let Some(val) = set_obj_map.get_mut(&idx) {
val.push(DelObj { val.push(DelObj {
// set_idx: idx, // set_idx: idx,
@@ -753,10 +756,13 @@ impl StorageAPI for Sets {
#[tracing::instrument(skip(self))] #[tracing::instrument(skip(self))]
async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option<Error>)> { async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
let (disks, _) = init_storage_disks_with_errors(&self.endpoints.endpoints, &DiskOption { let (disks, _) = init_storage_disks_with_errors(
cleanup: false, &self.endpoints.endpoints,
health_check: false, &DiskOption {
}) cleanup: false,
health_check: false,
},
)
.await; .await;
let (formats, errs) = load_format_erasure_all(&disks, true).await; let (formats, errs) = load_format_erasure_all(&disks, true).await;
if let Err(err) = check_format_erasure_values(&formats, self.set_drive_count) { if let Err(err) = check_format_erasure_values(&formats, self.set_drive_count) {
crates/ecstore/src/store.rs
+50 -28
View File
@@ -154,10 +154,13 @@ impl ECStore {
// validate_parity(partiy_count, pool_eps.drives_per_set)?; // validate_parity(partiy_count, pool_eps.drives_per_set)?;
let (disks, errs) = store_init::init_disks(&pool_eps.endpoints, &DiskOption { let (disks, errs) = store_init::init_disks(
cleanup: true, &pool_eps.endpoints,
health_check: true, &DiskOption {
}) cleanup: true,
health_check: true,
},
)
.await; .await;
check_disk_fatal_errs(&errs)?; check_disk_fatal_errs(&errs)?;
@@ -501,10 +504,14 @@ impl ECStore {
} }
async fn delete_prefix(&self, bucket: &str, object: &str) -> Result<()> { async fn delete_prefix(&self, bucket: &str, object: &str) -> Result<()> {
for pool in self.pools.iter() { for pool in self.pools.iter() {
pool.delete_object(bucket, object, ObjectOptions { pool.delete_object(
delete_prefix: true, bucket,
..Default::default() object,
}) ObjectOptions {
delete_prefix: true,
..Default::default()
},
)
.await?; .await?;
} }
@@ -614,11 +621,15 @@ impl ECStore {
async fn get_pool_idx(&self, bucket: &str, object: &str, size: i64) -> Result<usize> { async fn get_pool_idx(&self, bucket: &str, object: &str, size: i64) -> Result<usize> {
let idx = match self let idx = match self
.get_pool_idx_existing_with_opts(bucket, object, &ObjectOptions { .get_pool_idx_existing_with_opts(
skip_decommissioned: true, bucket,
skip_rebalancing: true, object,
..Default::default() &ObjectOptions {
}) skip_decommissioned: true,
skip_rebalancing: true,
..Default::default()
},
)
.await .await
{ {
Ok(res) => res, Ok(res) => res,
@@ -659,12 +670,16 @@ impl ECStore {
} }
async fn get_pool_idx_existing_no_lock(&self, bucket: &str, object: &str) -> Result<usize> { async fn get_pool_idx_existing_no_lock(&self, bucket: &str, object: &str) -> Result<usize> {
self.get_pool_idx_existing_with_opts(bucket, object, &ObjectOptions { self.get_pool_idx_existing_with_opts(
no_lock: true, bucket,
skip_decommissioned: true, object,
skip_rebalancing: true, &ObjectOptions {
..Default::default() no_lock: true,
}) skip_decommissioned: true,
skip_rebalancing: true,
..Default::default()
},
)
.await .await
} }
@@ -1359,11 +1374,14 @@ impl StorageAPI for ECStore {
if let Err(err) = self.peer_sys.make_bucket(bucket, opts).await { if let Err(err) = self.peer_sys.make_bucket(bucket, opts).await {
if !is_err_bucket_exists(&err.into()) { if !is_err_bucket_exists(&err.into()) {
let _ = self let _ = self
.delete_bucket(bucket, &DeleteBucketOptions { .delete_bucket(
no_lock: true, bucket,
no_recreate: true, &DeleteBucketOptions {
..Default::default() no_lock: true,
}) no_recreate: true,
..Default::default()
},
)
.await; .await;
} }
}; };
@@ -1666,10 +1684,14 @@ impl StorageAPI for ECStore {
for obj in objects.iter() { for obj in objects.iter() {
futures.push(async move { futures.push(async move {
self.internal_get_pool_info_existing_with_opts(bucket, &obj.object_name, &ObjectOptions { self.internal_get_pool_info_existing_with_opts(
no_lock: true, bucket,
..Default::default() &obj.object_name,
}) &ObjectOptions {
no_lock: true,
..Default::default()
},
)
.await .await
}); });
} }
crates/ecstore/src/store_list_objects.rs
+81 -71
View File
@@ -263,10 +263,14 @@ impl ECStore {
// use get // use get
if !opts.prefix.is_empty() && opts.limit == 1 && opts.marker.is_none() { if !opts.prefix.is_empty() && opts.limit == 1 && opts.marker.is_none() {
match self match self
.get_object_info(&opts.bucket, &opts.prefix, &ObjectOptions { .get_object_info(
no_lock: true, &opts.bucket,
..Default::default() &opts.prefix,
}) &ObjectOptions {
no_lock: true,
..Default::default()
},
)
.await .await
{ {
Ok(res) => { Ok(res) => {
@@ -765,45 +769,48 @@ impl ECStore {
let tx1 = sender.clone(); let tx1 = sender.clone();
let tx2 = sender.clone(); let tx2 = sender.clone();
list_path_raw(rx.resubscribe(), ListPathRawOptions { list_path_raw(
disks: disks.iter().cloned().map(Some).collect(), rx.resubscribe(),
fallback_disks: fallback_disks.iter().cloned().map(Some).collect(), ListPathRawOptions {
bucket: bucket.to_owned(), disks: disks.iter().cloned().map(Some).collect(),
path, fallback_disks: fallback_disks.iter().cloned().map(Some).collect(),
recursice: true, bucket: bucket.to_owned(),
filter_prefix: Some(filter_prefix), path,
forward_to: opts.marker.clone(), recursice: true,
min_disks: listing_quorum, filter_prefix: Some(filter_prefix),
per_disk_limit: opts.limit as i32, forward_to: opts.marker.clone(),
agreed: Some(Box::new(move |entry: MetaCacheEntry| { min_disks: listing_quorum,
Box::pin({ per_disk_limit: opts.limit as i32,
let value = tx1.clone(); agreed: Some(Box::new(move |entry: MetaCacheEntry| {
async move { Box::pin({
if entry.is_dir() { let value = tx1.clone();
return; async move {
} if entry.is_dir() {
if let Err(err) = value.send(entry).await { return;
error!("list_path send fail {:?}", err); }
}
}
})
})),
partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| {
Box::pin({
let value = tx2.clone();
let resolver = resolver.clone();
async move {
if let Some(entry) = entries.resolve(resolver) {
if let Err(err) = value.send(entry).await { if let Err(err) = value.send(entry).await {
error!("list_path send fail {:?}", err); error!("list_path send fail {:?}", err);
} }
} }
} })
}) })),
})), partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| {
finished: None, Box::pin({
..Default::default() let value = tx2.clone();
}) let resolver = resolver.clone();
async move {
if let Some(entry) = entries.resolve(resolver) {
if let Err(err) = value.send(entry).await {
error!("list_path send fail {:?}", err);
}
}
}
})
})),
finished: None,
..Default::default()
},
)
.await .await
}); });
} }
@@ -1272,42 +1279,45 @@ impl SetDisks {
let tx1 = sender.clone(); let tx1 = sender.clone();
let tx2 = sender.clone(); let tx2 = sender.clone();
list_path_raw(rx, ListPathRawOptions { list_path_raw(
disks: disks.iter().cloned().map(Some).collect(), rx,
fallback_disks: fallback_disks.iter().cloned().map(Some).collect(), ListPathRawOptions {
bucket: opts.bucket, disks: disks.iter().cloned().map(Some).collect(),
path: opts.base_dir, fallback_disks: fallback_disks.iter().cloned().map(Some).collect(),
recursice: opts.recursive, bucket: opts.bucket,
filter_prefix: opts.filter_prefix, path: opts.base_dir,
forward_to: opts.marker, recursice: opts.recursive,
min_disks: listing_quorum, filter_prefix: opts.filter_prefix,
per_disk_limit: limit, forward_to: opts.marker,
agreed: Some(Box::new(move |entry: MetaCacheEntry| { min_disks: listing_quorum,
Box::pin({ per_disk_limit: limit,
let value = tx1.clone(); agreed: Some(Box::new(move |entry: MetaCacheEntry| {
async move { Box::pin({
if let Err(err) = value.send(entry).await { let value = tx1.clone();
error!("list_path send fail {:?}", err); async move {
}
}
})
})),
partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| {
Box::pin({
let value = tx2.clone();
let resolver = resolver.clone();
async move {
if let Some(entry) = entries.resolve(resolver) {
if let Err(err) = value.send(entry).await { if let Err(err) = value.send(entry).await {
error!("list_path send fail {:?}", err); error!("list_path send fail {:?}", err);
} }
} }
} })
}) })),
})), partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| {
finished: None, Box::pin({
..Default::default() let value = tx2.clone();
}) let resolver = resolver.clone();
async move {
if let Some(entry) = entries.resolve(resolver) {
if let Err(err) = value.send(entry).await {
error!("list_path send fail {:?}", err);
}
}
}
})
})),
finished: None,
..Default::default()
},
)
.await .await
.map_err(Error::other) .map_err(Error::other)
} }
crates/ecstore/src/tier/tier.rs
+9 -4
View File
@@ -365,10 +365,15 @@ impl TierConfigMgr {
file: &str, file: &str,
data: Bytes, data: Bytes,
) -> std::result::Result<(), std::io::Error> { ) -> std::result::Result<(), std::io::Error> {
self.save_config_with_opts(api, file, data, &ObjectOptions { self.save_config_with_opts(
max_parity: true, api,
..Default::default() file,
}) data,
&ObjectOptions {
max_parity: true,
..Default::default()
},
)
.await .await
} }
crates/ecstore/src/tier/warm_backend_minio.rs
+13 -7
View File
@@ -101,13 +101,19 @@ impl WarmBackend for WarmBackendMinIO {
let part_size = optimal_part_size(length)?; let part_size = optimal_part_size(length)?;
let client = self.0.client.clone(); let client = self.0.client.clone();
let res = client let res = client
.put_object(&self.0.bucket, &self.0.get_dest(object), r, length, &PutObjectOptions { .put_object(
storage_class: self.0.storage_class.clone(), &self.0.bucket,
part_size: part_size as u64, &self.0.get_dest(object),
disable_content_sha256: true, r,
user_metadata: meta, length,
..Default::default() &PutObjectOptions {
}) storage_class: self.0.storage_class.clone(),
part_size: part_size as u64,
disable_content_sha256: true,
user_metadata: meta,
..Default::default()
},
)
.await?; .await?;
//self.ToObjectError(err, object) //self.ToObjectError(err, object)
Ok(res.version_id) Ok(res.version_id)
crates/ecstore/src/tier/warm_backend_rustfs.rs
+13 -7
View File
@@ -98,13 +98,19 @@ impl WarmBackend for WarmBackendRustFS {
let part_size = optimal_part_size(length)?; let part_size = optimal_part_size(length)?;
let client = self.0.client.clone(); let client = self.0.client.clone();
let res = client let res = client
.put_object(&self.0.bucket, &self.0.get_dest(object), r, length, &PutObjectOptions { .put_object(
storage_class: self.0.storage_class.clone(), &self.0.bucket,
part_size: part_size as u64, &self.0.get_dest(object),
disable_content_sha256: true, r,
user_metadata: meta, length,
..Default::default() &PutObjectOptions {
}) storage_class: self.0.storage_class.clone(),
part_size: part_size as u64,
disable_content_sha256: true,
user_metadata: meta,
..Default::default()
},
)
.await?; .await?;
//self.ToObjectError(err, object) //self.ToObjectError(err, object)
Ok(res.version_id) Ok(res.version_id)
crates/ecstore/src/tier/warm_backend_s3.rs
+12 -6
View File
@@ -127,12 +127,18 @@ impl WarmBackend for WarmBackendS3 {
) -> Result<String, std::io::Error> { ) -> Result<String, std::io::Error> {
let client = self.client.clone(); let client = self.client.clone();
let res = client let res = client
.put_object(&self.bucket, &self.get_dest(object), r, length, &PutObjectOptions { .put_object(
send_content_md5: true, &self.bucket,
storage_class: self.storage_class.clone(), &self.get_dest(object),
user_metadata: meta, r,
..Default::default() length,
}) &PutObjectOptions {
send_content_md5: true,
storage_class: self.storage_class.clone(),
user_metadata: meta,
..Default::default()
},
)
.await?; .await?;
Ok(res.version_id) Ok(res.version_id)
} }
crates/iam/src/manager.rs
+8 -5
View File
@@ -1575,11 +1575,14 @@ pub fn get_default_policyes() -> HashMap<String, PolicyDoc> {
default_policies default_policies
.iter() .iter()
.map(|(n, p)| { .map(|(n, p)| {
(n.to_string(), PolicyDoc { (
version: 1, n.to_string(),
policy: p.clone(), PolicyDoc {
..Default::default() version: 1,
}) policy: p.clone(),
..Default::default()
},
)
}) })
.collect() .collect()
} }
crates/lock/src/local_locker.rs
+26 -20
View File
@@ -140,17 +140,20 @@ impl Locker for LocalLocker {
} }
args.resources.iter().enumerate().for_each(|(idx, resource)| { args.resources.iter().enumerate().for_each(|(idx, resource)| {
self.lock_map.insert(resource.to_string(), vec![LockRequesterInfo { self.lock_map.insert(
name: resource.to_string(), resource.to_string(),
writer: true, vec![LockRequesterInfo {
source: args.source.to_string(), name: resource.to_string(),
owner: args.owner.to_string(), writer: true,
uid: args.uid.to_string(), source: args.source.to_string(),
group: args.resources.len() > 1, owner: args.owner.to_string(),
quorum: args.quorum, uid: args.uid.to_string(),
idx, group: args.resources.len() > 1,
..Default::default() quorum: args.quorum,
}]); idx,
..Default::default()
}],
);
let mut uuid = args.uid.to_string(); let mut uuid = args.uid.to_string();
format_uuid(&mut uuid, &idx); format_uuid(&mut uuid, &idx);
@@ -227,15 +230,18 @@ impl Locker for LocalLocker {
} }
} }
None => { None => {
self.lock_map.insert(resource.to_string(), vec![LockRequesterInfo { self.lock_map.insert(
name: resource.to_string(), resource.to_string(),
writer: false, vec![LockRequesterInfo {
source: args.source.to_string(), name: resource.to_string(),
owner: args.owner.to_string(), writer: false,
uid: args.uid.to_string(), source: args.source.to_string(),
quorum: args.quorum, owner: args.owner.to_string(),
..Default::default() uid: args.uid.to_string(),
}]); quorum: args.quorum,
..Default::default()
}],
);
} }
} }
let mut uuid = args.uid.to_string(); let mut uuid = args.uid.to_string();
crates/obs/examples/server.rs
+8 -8
View File
@@ -46,10 +46,10 @@ async fn run(bucket: String, object: String, user: String, service_name: String)
// Record Metrics // Record Metrics
let meter = global::meter("rustfs"); let meter = global::meter("rustfs");
let request_duration = meter.f64_histogram("s3_request_duration_seconds").build(); let request_duration = meter.f64_histogram("s3_request_duration_seconds").build();
request_duration.record(start_time.elapsed().unwrap().as_secs_f64(), &[opentelemetry::KeyValue::new( request_duration.record(
"operation", start_time.elapsed().unwrap().as_secs_f64(),
"run", &[opentelemetry::KeyValue::new("operation", "run")],
)]); );
match SystemObserver::init_process_observer(meter).await { match SystemObserver::init_process_observer(meter).await {
Ok(_) => info!("Process observer initialized successfully"), Ok(_) => info!("Process observer initialized successfully"),
@@ -84,10 +84,10 @@ async fn put_object(bucket: String, object: String, user: String) {
let meter = global::meter("rustfs"); let meter = global::meter("rustfs");
let request_duration = meter.f64_histogram("s3_request_duration_seconds").build(); let request_duration = meter.f64_histogram("s3_request_duration_seconds").build();
request_duration.record(start_time.elapsed().unwrap().as_secs_f64(), &[opentelemetry::KeyValue::new( request_duration.record(
"operation", start_time.elapsed().unwrap().as_secs_f64(),
"put_object", &[opentelemetry::KeyValue::new("operation", "put_object")],
)]); );
info!( info!(
"Starting PUT operation content: bucket = {}, object = {}, user = {},start_time = {}", "Starting PUT operation content: bucket = {}, object = {}, user = {},start_time = {}",
crates/obs/src/system/collector.rs
+18 -12
View File
@@ -115,18 +115,24 @@ impl Collector {
let transmitted = data.transmitted() as i64; let transmitted = data.transmitted() as i64;
self.metrics.network_io_per_interface.record( self.metrics.network_io_per_interface.record(
received, received,
&[&self.attributes.attributes[..], &[ &[
KeyValue::new(INTERFACE, interface_name.to_string()), &self.attributes.attributes[..],
KeyValue::new(DIRECTION, "received"), &[
]] KeyValue::new(INTERFACE, interface_name.to_string()),
KeyValue::new(DIRECTION, "received"),
],
]
.concat(), .concat(),
); );
self.metrics.network_io_per_interface.record( self.metrics.network_io_per_interface.record(
transmitted, transmitted,
&[&self.attributes.attributes[..], &[ &[
KeyValue::new(INTERFACE, interface_name.to_string()), &self.attributes.attributes[..],
KeyValue::new(DIRECTION, "transmitted"), &[
]] KeyValue::new(INTERFACE, interface_name.to_string()),
KeyValue::new(DIRECTION, "transmitted"),
],
]
.concat(), .concat(),
); );
} }
@@ -149,10 +155,10 @@ impl Collector {
}; };
self.metrics.process_status.record( self.metrics.process_status.record(
status_value, status_value,
&[&self.attributes.attributes[..], &[KeyValue::new( &[
STATUS, &self.attributes.attributes[..],
format!("{:?}", process.status()), &[KeyValue::new(STATUS, format!("{:?}", process.status()))],
)]] ]
.concat(), .concat(),
); );
crates/policy/src/policy/policy.rs
+165 -147
View File
@@ -276,150 +276,12 @@ pub mod default {
#[allow(clippy::incompatible_msrv)] #[allow(clippy::incompatible_msrv)]
pub static DEFAULT_POLICIES: LazyLock<[(&'static str, Policy); 6]> = LazyLock::new(|| { pub static DEFAULT_POLICIES: LazyLock<[(&'static str, Policy); 6]> = LazyLock::new(|| {
[ [
("readwrite", Policy { (
id: "".into(), "readwrite",
version: DEFAULT_VERSION.into(), Policy {
statements: vec![Statement { id: "".into(),
sid: "".into(), version: DEFAULT_VERSION.into(),
effect: Effect::Allow, statements: vec![Statement {
actions: ActionSet({
let mut hash_set = HashSet::new();
hash_set.insert(Action::S3Action(S3Action::AllActions));
hash_set
}),
not_actions: ActionSet(Default::default()),
resources: ResourceSet({
let mut hash_set = HashSet::new();
hash_set.insert(Resource::S3("*".into()));
hash_set
}),
conditions: Functions::default(),
..Default::default()
}],
}),
("readonly", Policy {
id: "".into(),
version: DEFAULT_VERSION.into(),
statements: vec![Statement {
sid: "".into(),
effect: Effect::Allow,
actions: ActionSet({
let mut hash_set = HashSet::new();
hash_set.insert(Action::S3Action(S3Action::GetBucketLocationAction));
hash_set.insert(Action::S3Action(S3Action::GetObjectAction));
hash_set
}),
not_actions: ActionSet(Default::default()),
resources: ResourceSet({
let mut hash_set = HashSet::new();
hash_set.insert(Resource::S3("*".into()));
hash_set
}),
conditions: Functions::default(),
..Default::default()
}],
}),
("writeonly", Policy {
id: "".into(),
version: DEFAULT_VERSION.into(),
statements: vec![Statement {
sid: "".into(),
effect: Effect::Allow,
actions: ActionSet({
let mut hash_set = HashSet::new();
hash_set.insert(Action::S3Action(S3Action::PutObjectAction));
hash_set
}),
not_actions: ActionSet(Default::default()),
resources: ResourceSet({
let mut hash_set = HashSet::new();
hash_set.insert(Resource::S3("*".into()));
hash_set
}),
conditions: Functions::default(),
..Default::default()
}],
}),
("writeonly", Policy {
id: "".into(),
version: DEFAULT_VERSION.into(),
statements: vec![Statement {
sid: "".into(),
effect: Effect::Allow,
actions: ActionSet({
let mut hash_set = HashSet::new();
hash_set.insert(Action::S3Action(S3Action::PutObjectAction));
hash_set
}),
not_actions: ActionSet(Default::default()),
resources: ResourceSet({
let mut hash_set = HashSet::new();
hash_set.insert(Resource::S3("*".into()));
hash_set
}),
conditions: Functions::default(),
..Default::default()
}],
}),
("diagnostics", Policy {
id: "".into(),
version: DEFAULT_VERSION.into(),
statements: vec![Statement {
sid: "".into(),
effect: Effect::Allow,
actions: ActionSet({
let mut hash_set = HashSet::new();
hash_set.insert(Action::AdminAction(AdminAction::ProfilingAdminAction));
hash_set.insert(Action::AdminAction(AdminAction::TraceAdminAction));
hash_set.insert(Action::AdminAction(AdminAction::ConsoleLogAdminAction));
hash_set.insert(Action::AdminAction(AdminAction::ServerInfoAdminAction));
hash_set.insert(Action::AdminAction(AdminAction::TopLocksAdminAction));
hash_set.insert(Action::AdminAction(AdminAction::HealthInfoAdminAction));
hash_set.insert(Action::AdminAction(AdminAction::PrometheusAdminAction));
hash_set.insert(Action::AdminAction(AdminAction::BandwidthMonitorAction));
hash_set
}),
not_actions: ActionSet(Default::default()),
resources: ResourceSet({
let mut hash_set = HashSet::new();
hash_set.insert(Resource::S3("*".into()));
hash_set
}),
conditions: Functions::default(),
..Default::default()
}],
}),
("consoleAdmin", Policy {
id: "".into(),
version: DEFAULT_VERSION.into(),
statements: vec![
Statement {
sid: "".into(),
effect: Effect::Allow,
actions: ActionSet({
let mut hash_set = HashSet::new();
hash_set.insert(Action::AdminAction(AdminAction::AllAdminActions));
hash_set
}),
not_actions: ActionSet(Default::default()),
resources: ResourceSet(HashSet::new()),
conditions: Functions::default(),
..Default::default()
},
Statement {
sid: "".into(),
effect: Effect::Allow,
actions: ActionSet({
let mut hash_set = HashSet::new();
hash_set.insert(Action::KmsAction(KmsAction::AllActions));
hash_set
}),
not_actions: ActionSet(Default::default()),
resources: ResourceSet(HashSet::new()),
conditions: Functions::default(),
..Default::default()
},
Statement {
sid: "".into(), sid: "".into(),
effect: Effect::Allow, effect: Effect::Allow,
actions: ActionSet({ actions: ActionSet({
@@ -435,9 +297,165 @@ pub mod default {
}), }),
conditions: Functions::default(), conditions: Functions::default(),
..Default::default() ..Default::default()
}, }],
], },
}), ),
(
"readonly",
Policy {
id: "".into(),
version: DEFAULT_VERSION.into(),
statements: vec![Statement {
sid: "".into(),
effect: Effect::Allow,
actions: ActionSet({
let mut hash_set = HashSet::new();
hash_set.insert(Action::S3Action(S3Action::GetBucketLocationAction));
hash_set.insert(Action::S3Action(S3Action::GetObjectAction));
hash_set
}),
not_actions: ActionSet(Default::default()),
resources: ResourceSet({
let mut hash_set = HashSet::new();
hash_set.insert(Resource::S3("*".into()));
hash_set
}),
conditions: Functions::default(),
..Default::default()
}],
},
),
(
"writeonly",
Policy {
id: "".into(),
version: DEFAULT_VERSION.into(),
statements: vec![Statement {
sid: "".into(),
effect: Effect::Allow,
actions: ActionSet({
let mut hash_set = HashSet::new();
hash_set.insert(Action::S3Action(S3Action::PutObjectAction));
hash_set
}),
not_actions: ActionSet(Default::default()),
resources: ResourceSet({
let mut hash_set = HashSet::new();
hash_set.insert(Resource::S3("*".into()));
hash_set
}),
conditions: Functions::default(),
..Default::default()
}],
},
),
(
"writeonly",
Policy {
id: "".into(),
version: DEFAULT_VERSION.into(),
statements: vec![Statement {
sid: "".into(),
effect: Effect::Allow,
actions: ActionSet({
let mut hash_set = HashSet::new();
hash_set.insert(Action::S3Action(S3Action::PutObjectAction));
hash_set
}),
not_actions: ActionSet(Default::default()),
resources: ResourceSet({
let mut hash_set = HashSet::new();
hash_set.insert(Resource::S3("*".into()));
hash_set
}),
conditions: Functions::default(),
..Default::default()
}],
},
),
(
"diagnostics",
Policy {
id: "".into(),
version: DEFAULT_VERSION.into(),
statements: vec![Statement {
sid: "".into(),
effect: Effect::Allow,
actions: ActionSet({
let mut hash_set = HashSet::new();
hash_set.insert(Action::AdminAction(AdminAction::ProfilingAdminAction));
hash_set.insert(Action::AdminAction(AdminAction::TraceAdminAction));
hash_set.insert(Action::AdminAction(AdminAction::ConsoleLogAdminAction));
hash_set.insert(Action::AdminAction(AdminAction::ServerInfoAdminAction));
hash_set.insert(Action::AdminAction(AdminAction::TopLocksAdminAction));
hash_set.insert(Action::AdminAction(AdminAction::HealthInfoAdminAction));
hash_set.insert(Action::AdminAction(AdminAction::PrometheusAdminAction));
hash_set.insert(Action::AdminAction(AdminAction::BandwidthMonitorAction));
hash_set
}),
not_actions: ActionSet(Default::default()),
resources: ResourceSet({
let mut hash_set = HashSet::new();
hash_set.insert(Resource::S3("*".into()));
hash_set
}),
conditions: Functions::default(),
..Default::default()
}],
},
),
(
"consoleAdmin",
Policy {
id: "".into(),
version: DEFAULT_VERSION.into(),
statements: vec![
Statement {
sid: "".into(),
effect: Effect::Allow,
actions: ActionSet({
let mut hash_set = HashSet::new();
hash_set.insert(Action::AdminAction(AdminAction::AllAdminActions));
hash_set
}),
not_actions: ActionSet(Default::default()),
resources: ResourceSet(HashSet::new()),
conditions: Functions::default(),
..Default::default()
},
Statement {
sid: "".into(),
effect: Effect::Allow,
actions: ActionSet({
let mut hash_set = HashSet::new();
hash_set.insert(Action::KmsAction(KmsAction::AllActions));
hash_set
}),
not_actions: ActionSet(Default::default()),
resources: ResourceSet(HashSet::new()),
conditions: Functions::default(),
..Default::default()
},
Statement {
sid: "".into(),
effect: Effect::Allow,
actions: ActionSet({
let mut hash_set = HashSet::new();
hash_set.insert(Action::S3Action(S3Action::AllActions));
hash_set
}),
not_actions: ActionSet(Default::default()),
resources: ResourceSet({
let mut hash_set = HashSet::new();
hash_set.insert(Resource::S3("*".into()));
hash_set
}),
conditions: Functions::default(),
..Default::default()
},
],
},
),
] ]
}); });
} }
crates/protos/src/main.rs
+7 -3
View File
@@ -85,9 +85,13 @@ fn main() -> Result<(), AnyError> {
Err(_) => "flatc".to_string(), Err(_) => "flatc".to_string(),
}; };
compile_flatbuffers_models(&mut generated_mod_rs, &flatc_path, proto_dir.clone(), flatbuffer_out_dir.clone(), vec![ compile_flatbuffers_models(
"models", &mut generated_mod_rs,
])?; &flatc_path,
proto_dir.clone(),
flatbuffer_out_dir.clone(),
vec!["models"],
)?;
fmt(); fmt();
Ok(()) Ok(())
crates/signer/src/request_signature_v4.rs
+1 -4
View File
@@ -340,10 +340,7 @@ fn sign_v4_inner(
let headers = req.headers_mut(); let headers = req.headers_mut();
let auth = format!( let auth = format!("{SIGN_V4_ALGORITHM} Credential={credential}, SignedHeaders={signed_headers}, Signature={signature}");
"{} Credential={}, SignedHeaders={}, Signature={}",
SIGN_V4_ALGORITHM, credential, signed_headers, signature
);
headers.insert("Authorization", auth.parse().unwrap()); headers.insert("Authorization", auth.parse().unwrap());
if !trailer.is_empty() { if !trailer.is_empty() {
rust-toolchain.toml
+17
View File
@@ -0,0 +1,17 @@
# 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.
[toolchain]
channel = "stable"
components = ["rustfmt", "clippy", "rust-src", "rust-analyzer"]
rustfs/src/admin/handlers/bucket_meta.rs
+7 -4
View File
@@ -453,10 +453,13 @@ impl Operation for ImportBucketMetadata {
// create bucket if not exists // create bucket if not exists
if !bucket_metadatas.contains_key(bucket_name) { if !bucket_metadatas.contains_key(bucket_name) {
if let Err(e) = store if let Err(e) = store
.make_bucket(bucket_name, &MakeBucketOptions { .make_bucket(
force_create: true, bucket_name,
..Default::default() &MakeBucketOptions {
}) force_create: true,
..Default::default()
},
)
.await .await
{ {
warn!("create bucket failed: {e}"); warn!("create bucket failed: {e}");
rustfs/src/admin/handlers/user.rs
+13 -10
View File
@@ -445,17 +445,20 @@ impl Operation for ExportIam {
let users: HashMap<String, AddOrUpdateUserReq> = users let users: HashMap<String, AddOrUpdateUserReq> = users
.into_iter() .into_iter()
.map(|(k, v)| { .map(|(k, v)| {
(k, AddOrUpdateUserReq { (
secret_key: v.credentials.secret_key, k,
status: { AddOrUpdateUserReq {
if v.credentials.status == "off" { secret_key: v.credentials.secret_key,
AccountStatus::Disabled status: {
} else { if v.credentials.status == "off" {
AccountStatus::Enabled AccountStatus::Disabled
} } else {
AccountStatus::Enabled
}
},
policy: None,
}, },
policy: None, )
})
}) })
.collect::<HashMap<String, AddOrUpdateUserReq>>(); .collect::<HashMap<String, AddOrUpdateUserReq>>();
rustfs/src/storage/ecfs.rs
+15 -9
View File
@@ -310,11 +310,14 @@ impl S3 for FS {
}; };
store store
.make_bucket(&bucket, &MakeBucketOptions { .make_bucket(
force_create: true, &bucket,
lock_enabled: object_lock_enabled_for_bucket.is_some_and(|v| v), &MakeBucketOptions {
..Default::default() force_create: true,
}) lock_enabled: object_lock_enabled_for_bucket.is_some_and(|v| v),
..Default::default()
},
)
.await .await
.map_err(ApiError::from)?; .map_err(ApiError::from)?;
@@ -495,10 +498,13 @@ impl S3 for FS {
}; };
store store
.delete_bucket(&input.bucket, &DeleteBucketOptions { .delete_bucket(
force: false, &input.bucket,
..Default::default() &DeleteBucketOptions {
}) force: false,
..Default::default()
},
)
.await .await
.map_err(ApiError::from)?; .map_err(ApiError::from)?;