feat: Add RustFS Scanner Module and Multiple Bug Fixes (rustfs#1579)

Author: weisd

Cargo.lock

@@ -35,6 +35,7 @@ members = [
     "crates/targets", # Target-specific configurations and utilities
     "crates/s3select-api", # S3 Select API interface
     "crates/s3select-query", # S3 Select query engine
+    "crates/scanner", # Scanner for data integrity checks and health monitoring
     "crates/signer", # client signer
     "crates/checksums", # client checksums
     "crates/utils", # Utility functions and helpers
@@ -88,6 +89,7 @@ rustfs-protos = { path = "crates/protos", version = "0.0.5" }
 rustfs-rio = { path = "crates/rio", version = "0.0.5" }
 rustfs-s3select-api = { path = "crates/s3select-api", version = "0.0.5" }
 rustfs-s3select-query = { path = "crates/s3select-query", version = "0.0.5" }
+rustfs-scanner = { path = "crates/scanner", version = "0.0.5" }
 rustfs-signer = { path = "crates/signer", version = "0.0.5" }
 rustfs-targets = { path = "crates/targets", version = "0.0.5" }
 rustfs-utils = { path = "crates/utils", version = "0.0.5" }

Cargo.toml

@@ -183,7 +183,7 @@ impl HealChannelProcessor {
                 HealType::Object {
                     bucket: request.bucket.clone(),
                     object: prefix.clone(),
-                    version_id: None,
+                    version_id: request.object_version_id.clone(),
                 }
             } else {
                 HealType::Bucket {
@@ -366,6 +366,7 @@ mod tests {
             id: "test-id".to_string(),
             bucket: "test-bucket".to_string(),
             object_prefix: None,
+            object_version_id: None,
             disk: None,
             priority: HealChannelPriority::Normal,
             scan_mode: None,
@@ -394,6 +395,7 @@ mod tests {
             id: "test-id".to_string(),
             bucket: "test-bucket".to_string(),
             object_prefix: Some("test-object".to_string()),
+            object_version_id: None,
             disk: None,
             priority: HealChannelPriority::High,
             scan_mode: Some(HealScanMode::Deep),
@@ -425,6 +427,7 @@ mod tests {
             id: "test-id".to_string(),
             bucket: "test-bucket".to_string(),
             object_prefix: None,
+            object_version_id: None,
             disk: Some("pool_0_set_1".to_string()),
             priority: HealChannelPriority::Critical,
             scan_mode: None,
@@ -453,6 +456,7 @@ mod tests {
             id: "test-id".to_string(),
             bucket: "test-bucket".to_string(),
             object_prefix: None,
+            object_version_id: None,
             disk: Some("invalid-disk-id".to_string()),
             priority: HealChannelPriority::Normal,
             scan_mode: None,
@@ -488,6 +492,7 @@ mod tests {
                 id: "test-id".to_string(),
                 bucket: "test-bucket".to_string(),
                 object_prefix: None,
+                object_version_id: None,
                 disk: None,
                 priority: channel_priority,
                 scan_mode: None,
@@ -516,6 +521,7 @@ mod tests {
             id: "test-id".to_string(),
             bucket: "test-bucket".to_string(),
             object_prefix: None,
+            object_version_id: None,
             disk: None,
             priority: HealChannelPriority::Normal,
             scan_mode: None,
@@ -545,6 +551,7 @@ mod tests {
             id: "test-id".to_string(),
             bucket: "test-bucket".to_string(),
             object_prefix: Some("".to_string()), // Empty prefix should be treated as bucket heal
+            object_version_id: None,
             disk: None,
             priority: HealChannelPriority::Normal,
             scan_mode: None,

crates/ahm/src/heal/channel.rs

@@ -50,7 +50,7 @@ impl Display for HealItemType {
     }
 }
 
-#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize)]
 pub enum DriveState {
     Ok,
     Offline,
@@ -59,7 +59,7 @@ pub enum DriveState {
     PermissionDenied,
     Faulty,
     RootMount,
-    Unknown,
+    Unknown(String),
     Unformatted, // only returned by disk
 }
 
@@ -73,12 +73,28 @@ impl DriveState {
             DriveState::PermissionDenied => "permission-denied",
             DriveState::Faulty => "faulty",
             DriveState::RootMount => "root-mount",
-            DriveState::Unknown => "unknown",
+            DriveState::Unknown(reason) => reason,
             DriveState::Unformatted => "unformatted",
         }
     }
 }
 
+impl Clone for DriveState {
+    fn clone(&self) -> Self {
+        match self {
+            DriveState::Unknown(reason) => DriveState::Unknown(reason.clone()),
+            DriveState::Ok => DriveState::Ok,
+            DriveState::Offline => DriveState::Offline,
+            DriveState::Corrupt => DriveState::Corrupt,
+            DriveState::Missing => DriveState::Missing,
+            DriveState::PermissionDenied => DriveState::PermissionDenied,
+            DriveState::Faulty => DriveState::Faulty,
+            DriveState::RootMount => DriveState::RootMount,
+            DriveState::Unformatted => DriveState::Unformatted,
+        }
+    }
+}
+
 impl Display for DriveState {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(f, "{}", self.to_str())
@@ -212,6 +228,8 @@ pub struct HealChannelRequest {
     pub bucket: String,
     /// Object prefix (optional)
     pub object_prefix: Option<String>,
+    /// Object version ID (optional)
+    pub object_version_id: Option<String>,
     /// Force start heal
     pub force_start: bool,
     /// Priority
@@ -346,6 +364,7 @@ pub fn create_heal_request(
         id: Uuid::new_v4().to_string(),
         bucket,
         object_prefix,
+        object_version_id: None,
         force_start,
         priority: priority.unwrap_or_default(),
         pool_index: None,
@@ -374,6 +393,7 @@ pub fn create_heal_request_with_options(
         id: Uuid::new_v4().to_string(),
         bucket,
         object_prefix,
+        object_version_id: None,
         force_start,
         priority: priority.unwrap_or_default(),
         pool_index,
@@ -501,6 +521,7 @@ pub async fn send_heal_disk(set_disk_id: String, priority: Option<HealChannelPri
         bucket: "".to_string(),
         object_prefix: None,
         disk: Some(set_disk_id),
+        object_version_id: None,
         force_start: false,
         priority: priority.unwrap_or_default(),
         pool_index: None,

crates/common/src/heal_channel.rs

@@ -23,5 +23,6 @@ pub(crate) mod profiler;
 pub(crate) mod protocols;
 pub(crate) mod quota;
 pub(crate) mod runtime;
+pub(crate) mod scanner;
 pub(crate) mod targets;
 pub(crate) mod tls;

crates/config/src/constants/mod.rs

@@ -0,0 +1,28 @@
+// 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.
+
+/// Environment variable name that specifies the data scanner start delay in seconds.
+/// - Purpose: Define the delay between data scanner operations.
+/// - Unit: seconds (u64).
+/// - Valid values: any positive integer.
+/// - Semantics: This delay controls how frequently the data scanner checks for and processes data; shorter delays lead to more responsive scanning but may increase system load.
+/// - Example: `export RUSTFS_DATA_SCANNER_START_DELAY_SECS=10`
+/// - Note: Choose an appropriate delay that balances scanning responsiveness with overall system performance.
+pub const ENV_DATA_SCANNER_START_DELAY_SECS: &str = "RUSTFS_DATA_SCANNER_START_DELAY_SECS";
+
+/// Default data scanner start delay in seconds if not specified in the environment variable.
+/// - Value: 10 seconds.
+/// - Rationale: This default interval provides a reasonable balance between scanning responsiveness and system load for most deployments.
+/// - Adjustments: Users may modify this value via the `RUSTFS_DATA_SCANNER_START_DELAY_SECS` environment variable based on their specific scanning requirements and system performance.
+pub const DEFAULT_DATA_SCANNER_START_DELAY_SECS: u64 = 60;

crates/config/src/constants/scanner.rs

@@ -37,6 +37,8 @@ pub use constants::quota::*;
 #[cfg(feature = "constants")]
 pub use constants::runtime::*;
 #[cfg(feature = "constants")]
+pub use constants::scanner::*;
+#[cfg(feature = "constants")]
 pub use constants::targets::*;
 #[cfg(feature = "constants")]
 pub use constants::tls::*;

crates/config/src/lib.rs

@@ -23,6 +23,7 @@ use crate::bucket::target::{self, BucketTarget, BucketTargets, Credentials};
 use crate::bucket::versioning_sys::BucketVersioningSys;
 use aws_credential_types::Credentials as SdkCredentials;
 use aws_sdk_s3::config::Region as SdkRegion;
+use aws_sdk_s3::error::ProvideErrorMetadata;
 use aws_sdk_s3::error::SdkError;
 use aws_sdk_s3::operation::complete_multipart_upload::CompleteMultipartUploadOutput;
 use aws_sdk_s3::operation::head_bucket::HeadBucketError;
@@ -1094,8 +1095,7 @@ pub struct TargetClient {
 
 impl TargetClient {
     pub fn to_url(&self) -> Url {
-        let scheme = if self.secure { "https" } else { "http" };
-        Url::parse(&format!("{scheme}://{}", self.endpoint)).unwrap()
+        Url::parse(&self.endpoint).unwrap()
     }
 
     pub async fn bucket_exists(&self, bucket: &str) -> Result<bool, S3ClientError> {
@@ -1104,9 +1104,17 @@ impl TargetClient {
             Err(e) => match e {
                 SdkError::ServiceError(oe) => match oe.into_err() {
                     HeadBucketError::NotFound(_) => Ok(false),
-                    other => Err(S3ClientError::new(format!(
-                        "failed to check bucket exists for bucket:{bucket} please check the bucket name and credentials, error:{other:?}"
-                    ))),
+                    other => {
+                        warn!(
+                            "failed to check bucket exists for bucket:{bucket} please check the bucket name and credentials, error:{:?}",
+                            other
+                        );
+                        let message = other.meta().meta();
+                        Err(S3ClientError::new(format!(
+                            "failed to check bucket exists for bucket:{bucket} please check the bucket name and credentials, error:{:?}",
+                            message
+                        )))
+                    }
                 },
                 SdkError::DispatchFailure(e) => Err(S3ClientError::new(format!(
                     "failed to dispatch bucket exists for bucket:{bucket} error:{e:?}"
@@ -1154,10 +1162,17 @@ impl TargetClient {
         body: ByteStream,
         opts: &PutObjectOptions,
     ) -> Result<(), S3ClientError> {
-        let headers = opts.header();
+        let mut headers = opts.header();
 
         let builder = self.client.put_object();
 
+        let version_id = opts.internal.source_version_id.clone();
+        if !version_id.is_empty()
+            && let Ok(header_value) = HeaderValue::from_str(&version_id)
+        {
+            headers.insert(RUSTFS_BUCKET_SOURCE_VERSION_ID, header_value);
+        }
+
         match builder
             .bucket(bucket)
             .key(object)
@@ -1185,9 +1200,33 @@ impl TargetClient {
         &self,
         bucket: &str,
         object: &str,
-        _opts: &PutObjectOptions,
+        opts: &PutObjectOptions,
     ) -> Result<String, S3ClientError> {
-        match self.client.create_multipart_upload().bucket(bucket).key(object).send().await {
+        let mut headers = HeaderMap::new();
+        let version_id = opts.internal.source_version_id.clone();
+        if !version_id.is_empty()
+            && let Ok(header_value) = HeaderValue::from_str(&version_id)
+        {
+            headers.insert(RUSTFS_BUCKET_SOURCE_VERSION_ID, header_value);
+        }
+
+        match self
+            .client
+            .create_multipart_upload()
+            .bucket(bucket)
+            .key(object)
+            .customize()
+            .map_request(move |mut req| {
+                for (k, v) in headers.clone().into_iter() {
+                    let key_str = k.unwrap().as_str().to_string();
+                    let value_str = v.to_str().unwrap_or("").to_string();
+                    req.headers_mut().insert(key_str, value_str);
+                }
+                Result::<_, aws_smithy_types::error::operation::BuildError>::Ok(req)
+            })
+            .send()
+            .await
+        {
             Ok(res) => Ok(res.upload_id.unwrap_or_default()),
             Err(e) => Err(e.into()),
         }

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -953,7 +953,7 @@ impl LifecycleOps for ObjectInfo {
         lifecycle::ObjectOpts {
             name: self.name.clone(),
             user_tags: self.user_tags.clone(),
-            version_id: self.version_id.map(|v| v.to_string()).unwrap_or_default(),
+            version_id: self.version_id.clone(),
             mod_time: self.mod_time,
             size: self.size as usize,
             is_latest: self.is_latest,
@@ -1067,7 +1067,7 @@ pub async fn eval_action_from_lifecycle(
     event
 }
 
-async fn apply_transition_rule(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool {
+pub async fn apply_transition_rule(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool {
     if oi.delete_marker || oi.is_dir {
         return false;
     }
@@ -1161,7 +1161,7 @@ pub async fn apply_expiry_on_non_transitioned_objects(
     true
 }
 
-async fn apply_expiry_rule(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool {
+pub async fn apply_expiry_rule(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool {
     let mut expiry_state = GLOBAL_ExpiryState.write().await;
     expiry_state.enqueue_by_days(oi, event, src).await;
     true