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

fix(site-replication): align IAM and bucket metadata replication (#4318)

Browse Source
This commit is contained in:
唐小鸭
2026-07-06 22:15:02 +08:00
committed by GitHub
parent 5c67c508cb
commit 849ea9a122
6 changed files with 137 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/quota/checker.rs
+4 -3
View File
@@ -19,6 +19,7 @@ use rustfs_common::metrics::Metric;
use rustfs_config::QUOTA_CONFIG_FILE;
use std::sync::Arc;
use std::time::Instant;
use time::OffsetDateTime;
use tokio::sync::RwLock;
use tracing::{debug, warn};
@@ -137,18 +138,18 @@ impl QuotaChecker {
Ok(quota)
}
pub async fn set_quota_config(&mut self, bucket: &str, quota: BucketQuota) -> Result<(), QuotaError> {
pub async fn set_quota_config(&mut self, bucket: &str, quota: BucketQuota) -> Result<OffsetDateTime, QuotaError> {
let json_data = serde_json::to_vec(&quota).map_err(|e| QuotaError::InvalidConfig {
reason: format!("Failed to serialize quota config: {}", e),
})?;
let start_time = Instant::now();
update(bucket, QUOTA_CONFIG_FILE, json_data)
let updated_at = update(bucket, QUOTA_CONFIG_FILE, json_data)
.await
.map_err(QuotaError::StorageError)?;
rustfs_common::metrics::Metrics::inc_time(Metric::QuotaSync, start_time.elapsed());
Ok(())
Ok(updated_at)
}
pub async fn get_quota_stats(&self, bucket: &str) -> Result<(BucketQuota, Option<u64>), QuotaError> {
rustfs/src/admin/handlers/quota.rs
+52 -2
View File
@@ -15,6 +15,7 @@
//! Quota admin handlers for HTTP API
use crate::admin::auth::{validate_admin_request, validate_admin_request_with_bucket};
use crate::admin::handlers::site_replication::site_replication_bucket_meta_hook;
use crate::admin::router::{AdminOperation, Operation, S3Router};
use crate::admin::runtime_sources::{current_bucket_metadata_handle, current_object_store_handle};
use crate::admin::storage_api::bucket::metadata_sys::BucketMetadataSys;
@@ -24,6 +25,7 @@ use crate::auth::{check_key_valid, get_session_token};
use crate::server::ADMIN_PREFIX;
use hyper::{Method, StatusCode};
use matchit::Params;
use rustfs_madmin::{SITE_REPL_API_VERSION, SRBucketMeta};
use rustfs_policy::policy::action::{Action, AdminAction, S3Action};
use s3s::{Body, S3Request, S3Response, S3Result, s3_error};
use serde::{Deserialize, Serialize};
@@ -284,11 +286,35 @@ impl Operation for SetBucketQuotaHandler {
.ok_or_else(|| s3_error!(InternalError, "{}", rustfs_config::QUOTA_METADATA_SYSTEM_ERROR_MSG))?;
let mut quota_checker = QuotaChecker::new(metadata_sys_lock.clone());
quota_checker
let updated_at = quota_checker
.set_quota_config(&bucket, quota.clone())
.await
.map_err(|e| s3_error!(InternalError, "failed to set quota: {}", e))?;
if let Err(err) = site_replication_bucket_meta_hook(SRBucketMeta {
bucket: bucket.clone(),
r#type: "quota-config".to_string(),
quota: Some(
serde_json::to_value(&quota).map_err(|e| s3_error!(InternalError, "failed to encode quota payload: {}", e))?,
),
updated_at: Some(updated_at),
api_version: Some(SITE_REPL_API_VERSION.to_string()),
..Default::default()
})
.await
{
warn!(
event = EVENT_ADMIN_QUOTA_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_QUOTA,
action = "set_bucket_quota",
bucket = %bucket,
state = "site_replication_hook_failed",
error = ?err,
"admin quota state"
);
}
// Get real-time usage from data usage system
let current_usage = current_usage_from_context(&bucket).await;
@@ -427,11 +453,35 @@ impl Operation for ClearBucketQuotaHandler {
// Clear quota (set to None)
let quota = BucketQuota::new(None);
quota_checker
let updated_at = quota_checker
.set_quota_config(&bucket, quota.clone())
.await
.map_err(|e| s3_error!(InternalError, "failed to clear quota: {}", e))?;
if let Err(err) = site_replication_bucket_meta_hook(SRBucketMeta {
bucket: bucket.clone(),
r#type: "quota-config".to_string(),
quota: Some(
serde_json::to_value(&quota).map_err(|e| s3_error!(InternalError, "failed to encode quota payload: {}", e))?,
),
updated_at: Some(updated_at),
api_version: Some(SITE_REPL_API_VERSION.to_string()),
..Default::default()
})
.await
{
warn!(
event = EVENT_ADMIN_QUOTA_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_QUOTA,
action = "clear_bucket_quota",
bucket = %bucket,
state = "site_replication_hook_failed",
error = ?err,
"admin quota state"
);
}
info!(
event = EVENT_ADMIN_QUOTA_STATE,
component = LOG_COMPONENT_ADMIN,
rustfs/src/admin/handlers/site_replication.rs
+17 -4
View File
@@ -4648,16 +4648,29 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> {
let Some(user) = item.iam_user else {
return Err(s3_error!(InvalidRequest, "iamUser is required"));
};
if let Some(local) = iam_sys.get_user(&user.access_key).await
&& is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
{
return Ok(());
}
if user.is_delete_req {
iam_sys.delete_user(&user.access_key, true).await.map_err(ApiError::from)?;
} else {
let Some(user_req) = user.user_req else {
return Err(s3_error!(InvalidRequest, "userReq is required"));
};
iam_sys
.create_user(&user.access_key, &user_req)
.await
.map_err(ApiError::from)?;
let is_status_only_update = user_req.secret_key.is_empty() && user_req.policy.is_none();
if is_status_only_update {
iam_sys
.set_user_status(&user.access_key, user_req.status)
.await
.map_err(ApiError::from)?;
} else {
iam_sys
.create_user(&user.access_key, &user_req)
.await
.map_err(ApiError::from)?;
}
}
Ok(())
}
rustfs/src/admin/handlers/user.rs
+32 -2
View File
@@ -360,11 +360,41 @@ impl Operation for SetUserStatus {
return Err(s3_error!(InternalError, "iam is not initialized"));
};
iam_store
.set_user_status(ak, status)
let updated_at = iam_store
.set_user_status(ak, status.clone())
.await
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("failed to set user status: {e}")))?;
if let Err(err) = site_replication_iam_change_hook(SRIAMItem {
r#type: "iam-user".to_string(),
iam_user: Some(SRIAMUser {
access_key: ak.to_string(),
is_delete_req: false,
user_req: Some(AddOrUpdateUserReq {
secret_key: String::new(),
policy: None,
status: status.clone(),
}),
api_version: Some(SITE_REPL_API_VERSION.to_string()),
}),
updated_at: Some(updated_at),
api_version: Some(SITE_REPL_API_VERSION.to_string()),
..Default::default()
})
.await
{
warn!(
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_USER,
event = EVENT_ADMIN_USER_STATE,
access_key = %ak,
action = "set_user_status",
result = "site_replication_hook_failed",
error = ?err,
"admin user state"
);
}
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
header.insert(CONTENT_LENGTH, "0".parse().expect("valid header value"));
rustfs/src/app/object_usecase.rs
+6
View File
@@ -5581,6 +5581,12 @@ impl DefaultObjectUsecase {
{
response.headers.insert(header_name, header_value);
}
if info.replication_status != ReplicationStatusType::Empty
&& let Ok(header_name) = http::HeaderName::from_bytes(AMZ_BUCKET_REPLICATION_STATUS.to_ascii_lowercase().as_bytes())
&& let Ok(header_value) = HeaderValue::from_str(info.replication_status.as_str())
{
response.headers.insert(header_name, header_value);
}
let result = Ok(response);
let _ = helper.complete(&result);
rustfs/src/storage/ecfs.rs
+26 -1
View File
@@ -21,6 +21,7 @@ use super::{
update_bucket_metadata_config,
};
use super::{StorageReplicationConfigExt as _, StorageVersioningConfigExt as _};
use crate::admin::handlers::site_replication::site_replication_bucket_meta_hook;
use crate::error::ApiError;
use crate::storage::access::has_bypass_governance_header;
use crate::storage::helper::OperationHelper;
@@ -38,6 +39,7 @@ use crate::table_catalog;
use http::StatusCode;
use metrics::{counter, histogram};
use rustfs_io_metrics::record_s3_op;
use rustfs_madmin::{SITE_REPL_API_VERSION, SRBucketMeta};
use rustfs_s3_ops::S3Operation;
use rustfs_targets::EventName;
use rustfs_utils::http::headers::{
@@ -1325,8 +1327,10 @@ impl S3 for FS {
};
let data = serialize(&input_cfg).map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("{err}")))?;
let object_lock_config =
String::from_utf8(data.clone()).map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("{err}")))?;
update_bucket_metadata_config(&bucket, OBJECT_LOCK_CONFIG, data)
let updated_at = update_bucket_metadata_config(&bucket, OBJECT_LOCK_CONFIG, data)
.await
.map_err(ApiError::from)?;
@@ -1345,6 +1349,27 @@ impl S3 for FS {
.map_err(ApiError::from)?;
}
if let Err(err) = site_replication_bucket_meta_hook(SRBucketMeta {
bucket: bucket.clone(),
r#type: "object-lock-config".to_string(),
object_lock_config: Some(object_lock_config),
updated_at: Some(updated_at),
api_version: Some(SITE_REPL_API_VERSION.to_string()),
..Default::default()
})
.await
{
warn!(
component = LOG_COMPONENT_STORAGE,
subsystem = LOG_SUBSYSTEM_OBJECT_LOCK,
event = "put_object_lock_configuration",
bucket = %bucket,
result = "site_replication_hook_failed",
error = ?err,
"storage object lock state"
);
}
rustfs_scanner::record_dirty_usage_bucket(&bucket);
Ok(S3Response::new(PutObjectLockConfigurationOutput::default()))
}