fix: remove dep crate openssl relation

2026-08-05 12:57:42 +00:00 · 2025-06-06 15:13:55 +08:00
parent 6302764512
commit 2fec6d161c
9 changed files with 140 additions and 351 deletions
@@ -1,12 +1,5 @@
 #![allow(unused_variables)]
 #![allow(dead_code)]
-use aws_sdk_s3::config::BehaviorVersion;
-use aws_sdk_s3::config::Credentials;
-use aws_sdk_s3::config::Region;
-use aws_sdk_s3::Client as S3Client;
-use aws_sdk_s3::Config;
-use bytes::Bytes;
-use uuid::Uuid;
 // use error::Error;
 use crate::bucket::metadata_sys::get_replication_config;
 use crate::bucket::versioning_sys::BucketVersioningSys;
@@ -19,6 +12,12 @@ use crate::store_api::ObjectInfo;
 use crate::store_api::ObjectOptions;
 use crate::store_api::ObjectToDelete;
 use crate::StorageAPI;
+use aws_sdk_s3::config::BehaviorVersion;
+use aws_sdk_s3::config::Credentials;
+use aws_sdk_s3::config::Region;
+use aws_sdk_s3::Client as S3Client;
+use aws_sdk_s3::Config;
+use bytes::Bytes;
 use chrono::DateTime;
 use chrono::Duration;
 use chrono::Utc;
@@ -28,6 +27,8 @@ use futures::StreamExt;
 use http::HeaderMap;
 use http::Method;
 use lazy_static::lazy_static;
+// use std::time::SystemTime;
+use once_cell::sync::Lazy;
 use regex::Regex;
 use rustfs_rsc::provider::StaticProvider;
 use rustfs_rsc::Minio;
@@ -47,14 +48,13 @@ use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::vec;
 use time::OffsetDateTime;
-use tokio::sync::Mutex;
-use tokio::task;
-use tracing::{debug, error, warn, info};
-use xxhash_rust::xxh3::xxh3_64;
-// use std::time::SystemTime;
-use once_cell::sync::Lazy;
 use tokio::sync::mpsc::{Receiver, Sender};
+use tokio::sync::Mutex;
 use tokio::sync::RwLock;
+use tokio::task;
+use tracing::{debug, error, info, warn};
+use uuid::Uuid;
+use xxhash_rust::xxh3::xxh3_64;
 // use bucket_targets::{self, GLOBAL_Bucket_Target_Sys};

 #[derive(Serialize, Deserialize, Debug)]
@@ -222,9 +222,7 @@ pub async fn queue_replication_heal(
        return None;
    }

-    if oi.replication_status == ReplicationStatusType::Completed
-        && !roi.existing_obj_resync.must_resync()
-    {
+    if oi.replication_status == ReplicationStatusType::Completed && !roi.existing_obj_resync.must_resync() {
        return None;
    }

@@ -340,10 +338,7 @@ pub async fn check_replicate_delete(
        user_tags: Some(oi.user_tags.clone()),
        delete_marker: oi.delete_marker,
        //version_id: dobj.version_id.clone().map(|v| v.to_string()),
-        version_id: oi
-            .version_id
-            .map(|uuid| uuid.to_string())
-            .unwrap_or_default(),
+        version_id: oi.version_id.map(|uuid| uuid.to_string()).unwrap_or_default(),
        op_type: ReplicationType::DeleteReplicationType,
        target_arn: None,
        replica: true,
@@ -455,9 +450,9 @@ pub async fn get_heal_replicate_object_info(
            &oi.bucket,
            &ObjectToDelete {
                object_name: oi.name.clone(),
-                version_id: oi.version_id.clone(),
+                version_id: oi.version_id,
            },
-            &oi,
+            oi,
            &ObjectOptions {
                // versioned: global_bucket_versioning_sys::prefix_enabled(&oi.bucket, &oi.name),
                // version_suspended: global_bucket_versioning_sys::prefix_suspended(&oi.bucket, &oi.name),
@@ -466,7 +461,8 @@ pub async fn get_heal_replicate_object_info(
                ..Default::default()
            },
            None,
-        ).await
+        )
+        .await
    } else {
        // let opts: ObjectOptions = put_opts(&bucket, &key, version_id, &req.headers, Some(mt))
        // .await
@@ -480,8 +476,13 @@ pub async fn get_heal_replicate_object_info(
            mod_time: oi.mod_time,
            ..Default::default()
        };
-        let repoptions =
-                    get_must_replicate_options(mt2.as_ref().unwrap_or(&HashMap::new()), "", ReplicationStatusType::Unknown, ReplicationType::ObjectReplicationType, &opts);
+        let repoptions = get_must_replicate_options(
+            mt2.as_ref().unwrap_or(&HashMap::new()),
+            "",
+            ReplicationStatusType::Unknown,
+            ReplicationType::ObjectReplicationType,
+            &opts,
+        );

        let decision = must_replicate(&oi.bucket, &oi.name, &repoptions).await;
        error!("decision:");
@@ -509,7 +510,10 @@ pub async fn get_heal_replicate_object_info(
    let asz = oi.get_actual_size().unwrap_or(0);

    let key = format!("{}{}", RESERVED_METADATA_PREFIX_LOWER, REPLICATION_TIMESTAMP);
-    let tm: Option<DateTime<Utc>> = user_defined.as_ref().unwrap().get(&key)
+    let tm: Option<DateTime<Utc>> = user_defined
+        .as_ref()
+        .unwrap()
+        .get(&key)
        .and_then(|v| DateTime::parse_from_rfc3339(v).ok())
        .map(|dt| dt.with_timezone(&Utc));

@@ -536,7 +540,7 @@ pub async fn get_heal_replicate_object_info(
        existing_obj_resync: Default::default(),
        target_statuses: tgt_statuses,
        target_purge_statuses: purge_statuses,
-        replication_timestamp: tm.unwrap_or_else(|| chrono::Utc::now()),
+        replication_timestamp: tm.unwrap_or_else(|| Utc::now),
        //ssec: crypto::is_encrypted(&oi.user_defined),
        ssec: false,
        user_tags: oi.user_tags.clone(),
@@ -551,7 +555,10 @@ pub async fn get_heal_replicate_object_info(
        result.checksum = oi.checksum.clone();
    }

-    warn!("Replication heal for object {} in bucket {} is configured {:?}", oi.name, oi.bucket, oi.version_id);
+    warn!(
+        "Replication heal for object {} in bucket {} is configured {:?}",
+        oi.name, oi.bucket, oi.version_id
+    );

    result
 }
@@ -691,7 +698,7 @@ impl ReplicationPool {
            mrf_worker_size: workers,
            priority,
            max_workers,
-            obj_layer: obj_layer,
+            obj_layer,
        };

        warn!("work size is: {}", workers);
@@ -932,7 +939,7 @@ impl ReplicationPool {
                    let max_workers = max_workers.min(WORKER_MAX_LIMIT);
                    if worker_count < max_workers as i32 {
                        //self.resize_workers((worker_count + 1 as usize).try_into().unwrap(), worker_count).await;
-                        self.resize_workers(worker_count as usize + 1 as usize, Some(worker_count as usize))
+                        self.resize_workers(worker_count as usize + 1_usize, Some(worker_count as usize))
                            .await;
                    }

@@ -949,6 +956,12 @@ impl ReplicationPool {

 pub struct ReplicationResyncer;

+impl Default for ReplicationResyncer {
+    fn default() -> Self {
+        Self
+    }
+}
+
 impl ReplicationResyncer {
    pub fn new() -> Self {
        Self
@@ -967,6 +980,7 @@ pub async fn init_bucket_replication_pool() {
        warn!("init bucket replication pool");
        ReplicationPool::init_bucket_replication_pool(store, opts, stat).await;
    } else {
+        // TODO: to be added
    }
 }

@@ -1736,10 +1750,10 @@ impl TraitForObjectInfo for ObjectInfo {

 fn convert_offsetdatetime_to_chrono(offset_dt: Option<OffsetDateTime>) -> Option<DateTime<Utc>> {
    //offset_dt.map(|odt| {
-        let tm = offset_dt.unwrap().unix_timestamp();
-        //let naive = NaiveDateTime::from_timestamp_opt(tm, 0).expect("Invalid timestamp");
-        DateTime::<Utc>::from_timestamp(tm, 0)
-        //DateTime::from_naive_utc_and_offset(naive, Utc) // Convert to Utc first
+    let tm = offset_dt.unwrap().unix_timestamp();
+    //let naive = NaiveDateTime::from_timestamp_opt(tm, 0).expect("Invalid timestamp");
+    DateTime::<Utc>::from_timestamp(tm, 0)
+    //DateTime::from_naive_utc_and_offset(naive, Utc) // Convert to Utc first
    //})
 }

@@ -1860,8 +1874,8 @@ pub async fn must_replicate(bucket: &str, object: &str, mopts: &MustReplicateOpt
        //decision.set(ReplicateTargetDecision::new(replicate,synchronous));
        info!("targe decision arn is:{}", tgt_arn.clone());
        decision.set(ReplicateTargetDecision {
-            replicate: replicate,
-            synchronous: synchronous,
+            replicate,
+            synchronous,
            arn: tgt_arn.clone(),
            id: 0.to_string(),
        });
@@ -2227,7 +2241,7 @@ async fn replicate_object_with_multipart(
            return Err(err.into());
        }
    }
-    return Ok(());
+    Ok(())
 }

 impl ReplicateObjectInfo {
@@ -2374,7 +2388,7 @@ impl ReplicateObjectInfo {
                                .await;
                            match ret {
                                Ok(_res) => {
-                                    warn!("replicate suc: {} {} {}",  self.bucket, self.name, self.version_id);
+                                    warn!("replicate suc: {} {} {}", self.bucket, self.name, self.version_id);
                                    rinfo.replication_status = ReplicationStatusType::Completed;
                                }
                                Err(err) => {
@@ -2465,7 +2479,7 @@ pub fn get_must_replicate_options(

    MustReplicateOptions {
        meta,
-        status: status,
+        status,
        op_type: op,
        replication_request: opts.replication_request,
    }
@@ -2561,7 +2575,6 @@ impl ReplicatedInfos {
    }
 }

-
 impl ReplicatedTargetInfo {
    fn empty(&self) -> bool {
        // Implement your logic to check if the target is empty
@@ -1,26 +1,27 @@
 #![allow(unused_variables)]
 #![allow(dead_code)]
+use crate::{
+    bucket::{self, target::BucketTargets},
+    new_object_layer_fn, peer, store_api,
+};
 use crate::{
    bucket::{metadata_sys, target::BucketTarget},
    endpoints::Node,
    peer::{PeerS3Client, RemotePeerS3Client},
    StorageAPI,
 };
-use crate::{
-    bucket::{self, target::BucketTargets},
-    new_object_layer_fn, peer, store_api,
-};
+//use tokio::sync::RwLock;
+use aws_sdk_s3::Client as S3Client;
 use chrono::Utc;
+use futures::future::err;
 use lazy_static::lazy_static;
+use std::sync::Arc;
 use std::{
    collections::HashMap,
    time::{Duration, SystemTime},
 };
-//use tokio::sync::RwLock;
-use aws_sdk_s3::Client as S3Client;
-use std::sync::{Arc};
-use tokio::sync::RwLock;
 use thiserror::Error;
+use tokio::sync::RwLock;

 pub struct TClient {
    pub s3cli: S3Client,
@@ -78,21 +79,21 @@ pub struct ArnTarget {
    last_refresh: chrono::DateTime<Utc>,
 }
 impl ArnTarget {
-    pub fn new(bucket: String, endpoint: String, ak:String, sk:String) -> Self {
+    pub fn new(bucket: String, endpoint: String, ak: String, sk: String) -> Self {
        Self {
            client: TargetClient {
-                bucket: bucket,
+                bucket,
                storage_class: "STANDRD".to_string(),
                disable_proxy: false,
                health_check_duration: Duration::from_secs(100),
-                endpoint: endpoint,
+                endpoint,
                reset_id: "0".to_string(),
                replicate_sync: false,
                secure: false,
                arn: "".to_string(),
                client: reqwest::Client::new(),
-                ak:ak,
-                sk:sk,
+                ak,
+                sk,
            },
            last_refresh: Utc::now(),
        }
@@ -149,9 +150,9 @@ pub struct BucketRemoteTargetNotFound {
 pub async fn init_bucket_targets(bucket: &str, meta: Arc<bucket::metadata::BucketMetadata>) {
    println!("140 {}", bucket);
    if let Some(sys) = GLOBAL_Bucket_Target_Sys.get() {
-        if let  Some(tgts) = meta.bucket_target_config.clone() {
+        if let Some(tgts) = meta.bucket_target_config.clone() {
            for tgt in tgts.targets {
-                warn!("ak and sk is:{:?}",tgt.credentials);
+                warn!("ak and sk is:{:?}", tgt.credentials);
                let _ = sys.set_target(bucket, &tgt, false, true).await;
                //sys.targets_map.
            }
@@ -159,12 +160,17 @@ pub async fn init_bucket_targets(bucket: &str, meta: Arc<bucket::metadata::Bucke
    }
 }

-pub async fn remove_bucket_target(bucket: &str, arn_str:&str) {
+pub async fn remove_bucket_target(bucket: &str, arn_str: &str) {
    if let Some(sys) = GLOBAL_Bucket_Target_Sys.get() {
        let _ = sys.remove_target(bucket, arn_str).await;
    }
 }

+impl Default for BucketTargetSys {
+    fn default() -> Self {
+        Self::new()
+    }
+}

 impl BucketTargetSys {
    pub fn new() -> Self {
@@ -238,7 +244,7 @@ impl BucketTargetSys {
        targets
    }

-    pub async fn remove_target(&self, bucket:&str, arn_str:&str) -> Result<(), SetTargetError> {
+    pub async fn remove_target(&self, bucket: &str, arn_str: &str) -> Result<(), SetTargetError> {
        //to do need lock;
        let mut targets_map = self.targets_map.write().await;
        let tgts = targets_map.get(bucket);
@@ -263,14 +269,13 @@ impl BucketTargetSys {

        // 如果没有找到匹配的 ARN，则返回错误
        if !found {
-             return Ok(());
+            return Ok(());
        }

        // 更新 targets_map
        targets_map.insert(bucket.to_string(), targets);
        arn_remotes_map.remove(arn_str);

-
        let targets = self.list_targets(Some(&bucket), None).await;
        println!("targets is {}", targets.len());
        match serde_json::to_vec(&targets) {
@@ -376,11 +381,11 @@ impl BucketTargetSys {
        match store.get_bucket_info(_bucket, &store_api::BucketOptions::default()).await {
            Ok(info) => {
                println!("Bucket Info: {:?}", info);
-                return info.versionning;
+                info.versionning
            }
            Err(err) => {
                eprintln!("Error: {:?}", err);
-                return false;
+                false
            }
        }
    }
@@ -419,13 +424,11 @@ impl BucketTargetSys {
        // }

        //let client = self.get_remote_target_client(tgt).await?;
-        if tgt.type_ == Some("replication".to_string()) {
-            if !fromdisk {
-                let versioning_config = self.local_is_bucket_versioned(bucket).await;
-                if !versioning_config {
-                    // println!("111111111");
-                    return Err(SetTargetError::TargetNotVersioned(bucket.to_string()));
-                }
+        if tgt.type_ == Some("replication".to_string()) && !fromdisk {
+            let versioning_config = self.local_is_bucket_versioned(bucket).await;
+            if !versioning_config {
+                // println!("111111111");
+                return Err(SetTargetError::TargetNotVersioned(bucket.to_string()));
            }
        }

@@ -502,7 +505,12 @@ impl BucketTargetSys {
            println!("437 exist:{}", tgt.arn.clone().unwrap());
            targets.push(tgt.clone());
        }
-        let arntgt: ArnTarget = ArnTarget::new(tgt.target_bucket.clone(), tgt.endpoint.clone(),tgt.credentials.clone().unwrap().access_key.clone(), tgt.credentials.clone().unwrap().secret_key);
+        let arntgt: ArnTarget = ArnTarget::new(
+            tgt.target_bucket.clone(),
+            tgt.endpoint.clone(),
+            tgt.credentials.clone().unwrap().access_key.clone(),
+            tgt.credentials.clone().unwrap().secret_key,
+        );

        arn_remotes_map.insert(tgt.arn.clone().unwrap().clone(), arntgt);
        //self.update_bandwidth_limit(bucket, &tgt.arn, tgt.bandwidth_limit).await;
@@ -523,8 +531,8 @@ pub struct TargetClient {
    pub reset_id: String,
    pub endpoint: String,
    pub secure: bool,
-    pub ak:String,
-    pub sk:String,
+    pub ak: String,
+    pub sk: String,
 }

 impl TargetClient {
@@ -539,8 +547,8 @@ impl TargetClient {
        reset_id: String,
        endpoint: String,
        secure: bool,
-        ak:String,
-        sk:String,
+        ak: String,
+        sk: String,
    ) -> Self {
        TargetClient {
            client,
@@ -561,7 +569,7 @@ impl TargetClient {
        Ok(true) // Mocked implementation
    }
 }
-use tracing::{error, warn, info};
+use tracing::{error, info, warn};
 use uuid::Uuid;

 #[derive(Debug, Clone)]
@@ -613,7 +621,7 @@ impl ARN {
    /// 检查 ARN 是否为空
    pub fn is_empty(&self) -> bool {
        //!self.arn_type.is_valid()
-        return false;
+        false
    }

    /// 将 ARN 转为字符串格式
@@ -625,12 +633,12 @@ impl ARN {
    pub fn parse(s: &str) -> Result<Self, String> {
        // ARN 必须是格式 arn:rustfs:<Type>:<REGION>:<ID>:<remote-bucket>
        if !s.starts_with("arn:rustfs:") {
-            return Err(format!("Invalid ARN {}", s).into());
+            return Err(format!("Invalid ARN {}", s));
        }

        let tokens: Vec<&str> = s.split(':').collect();
        if tokens.len() != 6 || tokens[4].is_empty() || tokens[5].is_empty() {
-            return Err(format!("Invalid ARN {}", s).into());
+            return Err(format!("Invalid ARN {}", s));
        }

        Ok(ARN {
@@ -661,7 +669,7 @@ fn must_get_uuid() -> String {
 }
 fn generate_arn(target: BucketTarget, depl_id: String) -> String {
    let mut uuid: String = depl_id;
-    if uuid == "" {
+    if uuid.is_empty() {
        uuid = must_get_uuid();
    }

@@ -671,7 +679,7 @@ fn generate_arn(target: BucketTarget, depl_id: String) -> String {
        region: "us-east-1".to_string(),
        bucket: (target.target_bucket),
    };
-    return arn.to_string();
+    arn.to_string()
 }

 // use std::collections::HashMap;
@@ -434,7 +434,7 @@ impl PeerS3Client for LocalPeerS3Client {
        // TODO: reduceWriteQuorumErrs
        let mut versioned = false;
        if let Ok(sys) = metadata_sys::get(bucket).await {
-            versioned = sys.versioning(); 
+            versioned = sys.versioning();
        }

        ress.iter()
@@ -505,13 +505,13 @@ impl RemotePeerS3Client {
        let addr = node.as_ref().map(|v| v.url.to_string()).unwrap_or_default().to_string();
        Self { node, pools, addr }
    }
-    pub fn get_addr(&self)->String {
-        return self.addr.clone();
+    pub fn get_addr(&self) -> String {
+        self.addr.clone()
    }
 }

 #[async_trait]
-impl PeerS3Client for RemotePeerS3Client { 
+impl PeerS3Client for RemotePeerS3Client {
    fn get_pools(&self) -> Option<Vec<usize>> {
        self.pools.clone()
    }