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

Merge branch 'main' of github.com:rustfs/s3-rustfs

Browse Source 
* 'main' of github.com:rustfs/s3-rustfs:
  change rustfs-rsc new version not use openssl

# Conflicts:
#	ecstore/Cargo.toml
This commit is contained in:
houseme
2025-06-06 15:17:58 +08:00
parent 2fec6d161c 3f8d56501c
commit 3eaa3d68c7
1 changed files with 18 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ecstore/src/cmd/bucket_replication.rs
+18 -19
View File
@@ -540,7 +540,7 @@ pub async fn get_heal_replicate_object_info(
existing_obj_resync: Default::default(), existing_obj_resync: Default::default(),
target_statuses: tgt_statuses, target_statuses: tgt_statuses,
target_purge_statuses: purge_statuses, target_purge_statuses: purge_statuses,
replication_timestamp: tm.unwrap_or_else(|| Utc::now), replication_timestamp: tm.unwrap_or_else(|| Utc::now()),
//ssec: crypto::is_encrypted(&oi.user_defined), //ssec: crypto::is_encrypted(&oi.user_defined),
ssec: false, ssec: false,
user_tags: oi.user_tags.clone(), user_tags: oi.user_tags.clone(),
@@ -751,7 +751,7 @@ impl ReplicationPool {
tokio::spawn(async move { tokio::spawn(async move {
while let Some(operation) = receiver.recv().await { while let Some(operation) = receiver.recv().await {
println!("resize workers 1"); println!("resize workers 1");
active_workers_clone.fetch_add(1, std::sync::atomic::Ordering::SeqCst); active_workers_clone.fetch_add(1, Ordering::SeqCst);
if let Some(info) = operation.as_any().downcast_ref::<ReplicateObjectInfo>() { if let Some(info) = operation.as_any().downcast_ref::<ReplicateObjectInfo>() {
replicate_object(info.clone(), obj_layer_clone.clone()).await; replicate_object(info.clone(), obj_layer_clone.clone()).await;
@@ -761,7 +761,7 @@ impl ReplicationPool {
eprintln!("Unknown replication type"); eprintln!("Unknown replication type");
} }
active_workers_clone.fetch_sub(1, std::sync::atomic::Ordering::SeqCst); active_workers_clone.fetch_sub(1, Ordering::SeqCst);
} }
}); });
} }
@@ -799,7 +799,7 @@ impl ReplicationPool {
tokio::spawn(async move { tokio::spawn(async move {
while let Some(operation) = receiver.recv().await { while let Some(operation) = receiver.recv().await {
// Simulate work being processed // Simulate work being processed
active_workers_clone.fetch_add(1, std::sync::atomic::Ordering::SeqCst); active_workers_clone.fetch_add(1, Ordering::SeqCst);
if let Some(info) = operation.as_any().downcast_ref::<ReplicateObjectInfo>() { if let Some(info) = operation.as_any().downcast_ref::<ReplicateObjectInfo>() {
//self.stats.inc_q(&info.bucket, info.size, info.delete_marker, &info.op_type); //self.stats.inc_q(&info.bucket, info.size, info.delete_marker, &info.op_type);
@@ -813,7 +813,7 @@ impl ReplicationPool {
eprintln!("Unknown replication type"); eprintln!("Unknown replication type");
} }
active_workers_clone.fetch_sub(1, std::sync::atomic::Ordering::SeqCst); active_workers_clone.fetch_sub(1, Ordering::SeqCst);
} }
}); });
} }
@@ -1614,7 +1614,7 @@ impl ConfigProcess for s3s::dto::ReplicationConfiguration {
debug!("rule size is {}", &self.rules.len()); debug!("rule size is {}", &self.rules.len());
for rule in &self.rules { for rule in &self.rules {
if rule.status.as_str() == s3s::dto::ReplicationRuleStatus::DISABLED { if rule.status.as_str() == ReplicationRuleStatus::DISABLED {
debug!("rule size is"); debug!("rule size is");
continue; continue;
} }
@@ -1724,10 +1724,10 @@ impl TraitForObjectInfo for ObjectInfo {
targets: HashMap::new(), targets: HashMap::new(),
purge_targets: HashMap::new(), purge_targets: HashMap::new(),
reset_statuses_map: HashMap::new(), reset_statuses_map: HashMap::new(),
replica_timestamp: chrono::Utc::now(), replica_timestamp: Utc::now(),
replica_status: ReplicationStatusType::Pending, replica_status: ReplicationStatusType::Pending,
delete_marker: false, delete_marker: false,
replication_timestamp: chrono::Utc::now(), replication_timestamp: Utc::now(),
}; };
// Set targets and purge_targets using respective functions // Set targets and purge_targets using respective functions
@@ -1760,7 +1760,7 @@ fn convert_offsetdatetime_to_chrono(offset_dt: Option<OffsetDateTime>) -> Option
pub async fn schedule_replication(oi: ObjectInfo, o: Arc<store::ECStore>, dsc: ReplicateDecision, op_type: i32) { pub async fn schedule_replication(oi: ObjectInfo, o: Arc<store::ECStore>, dsc: ReplicateDecision, op_type: i32) {
let tgt_statuses = replication_statuses_map(&oi.replication_status_internal); let tgt_statuses = replication_statuses_map(&oi.replication_status_internal);
// //let purge_statuses = version_purge_statuses_map(&oi.); // //let purge_statuses = version_purge_statuses_map(&oi.);
let replication_timestamp = chrono::Utc::now(); // Placeholder for timestamp parsing let replication_timestamp = Utc::now(); // Placeholder for timestamp parsing
let replication_state = oi.replication_state(); let replication_state = oi.replication_state();
let actual_size = oi.actual_size.unwrap_or(0); let actual_size = oi.actual_size.unwrap_or(0);
@@ -1960,7 +1960,7 @@ pub trait ObjectInfoExt {
fn is_multipart(&self) -> bool; fn is_multipart(&self) -> bool;
} }
impl ObjectInfoExt for store_api::ObjectInfo { impl ObjectInfoExt for ObjectInfo {
fn target_replication_status(&self, arn: String) -> ReplicationStatusType { fn target_replication_status(&self, arn: String) -> ReplicationStatusType {
let rep_stat_matches = REPL_STATUS_REGEX.captures_iter(&self.replication_status_internal); let rep_stat_matches = REPL_STATUS_REGEX.captures_iter(&self.replication_status_internal);
for matched in rep_stat_matches { for matched in rep_stat_matches {
@@ -2144,8 +2144,7 @@ async fn replicate_object_with_multipart(
.endpoint(target_info.endpoint.clone()) .endpoint(target_info.endpoint.clone())
.provider(provider) .provider(provider)
.secure(false) .secure(false)
.build() .build()?;
.unwrap();
let ret = minio_cli let ret = minio_cli
.create_multipart_upload_with_versionid(tgt_cli.bucket.clone(), local_obj_info.name.clone(), rep_obj.version_id.clone()) .create_multipart_upload_with_versionid(tgt_cli.bucket.clone(), local_obj_info.name.clone(), rep_obj.version_id.clone())
@@ -2264,7 +2263,7 @@ impl ReplicateObjectInfo {
} }
async fn replicate_object(&self, target: &TargetClient, _arn: String) -> ReplicatedTargetInfo { async fn replicate_object(&self, target: &TargetClient, _arn: String) -> ReplicatedTargetInfo {
let _start_time = chrono::Utc::now(); let _start_time = Utc::now();
// 初始化 ReplicatedTargetInfo // 初始化 ReplicatedTargetInfo
warn!("replicate is {}", _arn.clone()); warn!("replicate is {}", _arn.clone());
@@ -2277,7 +2276,7 @@ impl ReplicateObjectInfo {
replication_action: ReplicationAction::ReplicateAll, replication_action: ReplicationAction::ReplicateAll,
endpoint: target.endpoint.clone(), endpoint: target.endpoint.clone(),
secure: target.endpoint.clone().contains("https://"), secure: target.endpoint.clone().contains("https://"),
resync_timestamp: chrono::Utc::now().to_string(), resync_timestamp: Utc::now().to_string(),
replication_resynced: false, replication_resynced: false,
duration: Duration::default(), duration: Duration::default(),
err: None, err: None,
@@ -2309,7 +2308,7 @@ impl ReplicateObjectInfo {
// versionSuspended := globalBucketVersioningSys.PrefixSuspended(bucket, object) // versionSuspended := globalBucketVersioningSys.PrefixSuspended(bucket, object)
// 模拟对象获取和元数据检查 // 模拟对象获取和元数据检查
let mut opt = store_api::ObjectOptions::default(); let mut opt = ObjectOptions::default();
opt.version_id = Some(self.version_id.clone()); opt.version_id = Some(self.version_id.clone());
opt.versioned = true; opt.versioned = true;
opt.version_suspended = false; opt.version_suspended = false;
@@ -2410,7 +2409,7 @@ impl ReplicateObjectInfo {
} }
} }
} }
return rinfo; rinfo
} }
fn is_target_offline(&self, endpoint: &str) -> bool { fn is_target_offline(&self, endpoint: &str) -> bool {
@@ -2419,11 +2418,11 @@ impl ReplicateObjectInfo {
false false
} }
async fn get_object_info(&self, opts: store_api::ObjectOptions) -> Result<ObjectInfo, Error> { async fn get_object_info(&self, opts: ObjectOptions) -> Result<ObjectInfo, Error> {
let objectlayer = new_object_layer_fn(); let objectlayer = new_object_layer_fn();
//let opts = ecstore::store_api::ObjectOptions { max_parity: (), mod_time: (), part_number: (), delete_prefix: (), version_id: (), no_lock: (), versioned: (), version_suspended: (), skip_decommissioned: (), skip_rebalancing: (), data_movement: (), src_pool_idx: (), user_defined: (), preserve_etag: (), metadata_chg: (), replication_request: (), delete_marker: () } //let opts = ecstore::store_api::ObjectOptions { max_parity: (), mod_time: (), part_number: (), delete_prefix: (), version_id: (), no_lock: (), versioned: (), version_suspended: (), skip_decommissioned: (), skip_rebalancing: (), data_movement: (), src_pool_idx: (), user_defined: (), preserve_etag: (), metadata_chg: (), replication_request: (), delete_marker: () }
let res = objectlayer.unwrap().get_object_info(&self.bucket, &self.name, &opts).await; let res = objectlayer.unwrap().get_object_info(&self.bucket, &self.name, &opts).await;
return res; res
} }
fn perform_replication(&self, target: &RemotePeerS3Client, object_info: &ObjectInfo) -> Result<(), String> { fn perform_replication(&self, target: &RemotePeerS3Client, object_info: &ObjectInfo) -> Result<(), String> {
@@ -2618,7 +2617,7 @@ pub async fn replicate_object(ri: ReplicateObjectInfo, object_api: Arc<store::EC
let rinfos = Arc::new(Mutex::new(ReplicatedInfos::default())); let rinfos = Arc::new(Mutex::new(ReplicatedInfos::default()));
let cri = Arc::new(ri.clone()); let cri = Arc::new(ri.clone());
let mut tasks: Vec<tokio::task::JoinHandle<()>> = vec![]; let mut tasks: Vec<task::JoinHandle<()>> = vec![];
for tgt_arn in tgt_arns { for tgt_arn in tgt_arns {
let tgt = bucket_targets::get_bucket_target_client(&ri.bucket, &tgt_arn).await; let tgt = bucket_targets::get_bucket_target_client(&ri.bucket, &tgt_arn).await;