feat(grpc): walk_dir http

fix(ecstore): rebalance loop

rustfs/src/admin/handlers/rebalance.rs

@@ -10,7 +10,8 @@ use http::{HeaderMap, StatusCode};
 use matchit::Params;
 use s3s::{Body, S3Request, S3Response, S3Result, header::CONTENT_TYPE, s3_error};
 use serde::{Deserialize, Serialize};
-use std::time::{Duration, SystemTime};
+use std::time::Duration;
+use time::OffsetDateTime;
 use tracing::warn;
 
 use crate::admin::router::Operation;
@@ -56,8 +57,8 @@ pub struct RebalanceAdminStatus {
     pub id: String, // Identifies the ongoing rebalance operation by a UUID
     #[serde(rename = "pools")]
     pub pools: Vec<RebalancePoolStatus>, // Contains all pools, including inactive
-    #[serde(rename = "stoppedAt")]
-    pub stopped_at: Option<SystemTime>, // Optional timestamp when rebalance was stopped
+    #[serde(rename = "stoppedAt", with = "offsetdatetime_rfc3339")]
+    pub stopped_at: Option<OffsetDateTime>, // Optional timestamp when rebalance was stopped
 }
 
 pub struct RebalanceStart {}
@@ -101,11 +102,13 @@ impl Operation for RebalanceStart {
             }
         };
 
+        store.start_rebalance().await;
+
         warn!("Rebalance started with id: {}", id);
         if let Some(notification_sys) = get_global_notification_sys() {
-            warn!("Loading rebalance meta");
+            warn!("RebalanceStart Loading rebalance meta start");
             notification_sys.load_rebalance_meta(true).await;
-            warn!("Rebalance meta loaded");
+            warn!("RebalanceStart Loading rebalance meta done");
         }
 
         let resp = RebalanceResp { id };
@@ -175,15 +178,14 @@ impl Operation for RebalanceStatus {
             let total_bytes_to_rebal = ps.init_capacity as f64 * meta.percent_free_goal - ps.init_free_space as f64;
 
             let mut elapsed = if let Some(start_time) = ps.info.start_time {
-                SystemTime::now()
-                    .duration_since(start_time)
-                    .map_err(|e| s3_error!(InternalError, "Failed to calculate elapsed time: {}", e))?
+                let now = OffsetDateTime::now_utc();
+                now - start_time
             } else {
                 return Err(s3_error!(InternalError, "Start time is not available"));
             };
 
             let mut eta = if ps.bytes > 0 {
-                Duration::from_secs_f64(total_bytes_to_rebal * elapsed.as_secs_f64() / ps.bytes as f64)
+                Duration::from_secs_f64(total_bytes_to_rebal * elapsed.as_seconds_f64() / ps.bytes as f64)
             } else {
                 Duration::ZERO
             };
@@ -193,10 +195,8 @@ impl Operation for RebalanceStatus {
             }
 
             if let Some(stopped_at) = stop_time {
-                if let Ok(du) = stopped_at.duration_since(ps.info.start_time.unwrap_or(stopped_at)) {
-                    elapsed = du;
-                } else {
-                    return Err(s3_error!(InternalError, "Failed to calculate elapsed time"));
+                if let Some(start_time) = ps.info.start_time {
+                    elapsed = stopped_at - start_time;
                 }
 
                 eta = Duration::ZERO;
@@ -208,7 +208,7 @@ impl Operation for RebalanceStatus {
                 bytes: ps.bytes,
                 bucket: ps.bucket.clone(),
                 object: ps.object.clone(),
-                elapsed: elapsed.as_secs(),
+                elapsed: elapsed.whole_seconds() as u64,
                 eta: eta.as_secs(),
             });
         }
@@ -244,10 +244,45 @@ impl Operation for RebalanceStop {
             .await
             .map_err(|e| s3_error!(InternalError, "Failed to stop rebalance: {}", e))?;
 
+        warn!("handle RebalanceStop save_rebalance_stats done ");
         if let Some(notification_sys) = get_global_notification_sys() {
-            notification_sys.load_rebalance_meta(true).await;
+            warn!("handle RebalanceStop notification_sys load_rebalance_meta");
+            notification_sys.load_rebalance_meta(false).await;
+            warn!("handle RebalanceStop notification_sys load_rebalance_meta done");
         }
 
-        return Err(s3_error!(NotImplemented));
+        Ok(S3Response::new((StatusCode::OK, Body::empty())))
+    }
+}
+
+mod offsetdatetime_rfc3339 {
+    use serde::{self, Deserialize, Deserializer, Serializer};
+    use time::{OffsetDateTime, format_description::well_known::Rfc3339};
+
+    pub fn serialize<S>(dt: &Option<OffsetDateTime>, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        match dt {
+            Some(dt) => {
+                let s = dt.format(&Rfc3339).map_err(serde::ser::Error::custom)?;
+                serializer.serialize_some(&s)
+            }
+            None => serializer.serialize_none(),
+        }
+    }
+
+    pub fn deserialize<'de, D>(deserializer: D) -> Result<Option<OffsetDateTime>, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let opt = Option::<String>::deserialize(deserializer)?;
+        match opt {
+            Some(s) => {
+                let dt = OffsetDateTime::parse(&s, &Rfc3339).map_err(serde::de::Error::custom)?;
+                Ok(Some(dt))
+            }
+            None => Ok(None),
+        }
     }
 }

rustfs/src/admin/rpc.rs

@@ -3,6 +3,7 @@ use super::router::Operation;
 use super::router::S3Router;
 use crate::storage::ecfs::bytes_stream;
 use ecstore::disk::DiskAPI;
+use ecstore::disk::WalkDirOptions;
 use ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE;
 use ecstore::store::find_local_disk;
 use futures::TryStreamExt;
@@ -18,6 +19,7 @@ use s3s::s3_error;
 use serde_urlencoded::from_bytes;
 use tokio_util::io::ReaderStream;
 use tokio_util::io::StreamReader;
+use tracing::warn;
 
 pub const RPC_PREFIX: &str = "/rustfs/rpc";
 
@@ -28,12 +30,30 @@ pub fn regist_rpc_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()>
         AdminOperation(&ReadFile {}),
     )?;
 
+    r.insert(
+        Method::HEAD,
+        format!("{}{}", RPC_PREFIX, "/read_file_stream").as_str(),
+        AdminOperation(&ReadFile {}),
+    )?;
+
     r.insert(
         Method::PUT,
         format!("{}{}", RPC_PREFIX, "/put_file_stream").as_str(),
         AdminOperation(&PutFile {}),
     )?;
 
+    r.insert(
+        Method::GET,
+        format!("{}{}", RPC_PREFIX, "/walk_dir").as_str(),
+        AdminOperation(&WalkDir {}),
+    )?;
+
+    r.insert(
+        Method::HEAD,
+        format!("{}{}", RPC_PREFIX, "/walk_dir").as_str(),
+        AdminOperation(&WalkDir {}),
+    )?;
+
     Ok(())
 }
 
@@ -50,6 +70,9 @@ pub struct ReadFile {}
 #[async_trait::async_trait]
 impl Operation for ReadFile {
     async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        if req.method == Method::HEAD {
+            return Ok(S3Response::new((StatusCode::OK, Body::empty())));
+        }
         let query = {
             if let Some(query) = req.uri.query() {
                 let input: ReadFileQuery =
@@ -79,6 +102,61 @@ impl Operation for ReadFile {
     }
 }
 
+#[derive(Debug, Default, serde::Deserialize)]
+pub struct WalkDirQuery {
+    disk: String,
+}
+
+pub struct WalkDir {}
+
+#[async_trait::async_trait]
+impl Operation for WalkDir {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        if req.method == Method::HEAD {
+            return Ok(S3Response::new((StatusCode::OK, Body::empty())));
+        }
+
+        let query = {
+            if let Some(query) = req.uri.query() {
+                let input: WalkDirQuery =
+                    from_bytes(query.as_bytes()).map_err(|e| s3_error!(InvalidArgument, "get query failed1 {:?}", e))?;
+                input
+            } else {
+                WalkDirQuery::default()
+            }
+        };
+
+        let mut input = req.input;
+        let body = match input.store_all_unlimited().await {
+            Ok(b) => b,
+            Err(e) => {
+                warn!("get body failed, e: {:?}", e);
+                return Err(s3_error!(InvalidRequest, "get body failed"));
+            }
+        };
+
+        // let body_bytes = decrypt_data(input_cred.secret_key.expose().as_bytes(), &body)
+        //     .map_err(|e| S3Error::with_message(S3ErrorCode::InvalidArgument, format!("decrypt_data err {}", e)))?;
+
+        let args: WalkDirOptions =
+            serde_json::from_slice(&body).map_err(|e| s3_error!(InternalError, "unmarshal body err {}", e))?;
+        let Some(disk) = find_local_disk(&query.disk).await else {
+            return Err(s3_error!(InvalidArgument, "disk not found"));
+        };
+
+        let (rd, mut wd) = tokio::io::duplex(DEFAULT_READ_BUFFER_SIZE);
+
+        tokio::spawn(async move {
+            if let Err(e) = disk.walk_dir(args, &mut wd).await {
+                warn!("walk dir err {}", e);
+            }
+        });
+
+        let body = Body::from(StreamingBlob::wrap(ReaderStream::with_capacity(rd, DEFAULT_READ_BUFFER_SIZE)));
+        Ok(S3Response::new((StatusCode::OK, body)))
+    }
+}
+
 // /rustfs/rpc/read_file_stream?disk={}&volume={}&path={}&offset={}&length={}"
 #[derive(Debug, Default, serde::Deserialize)]
 pub struct PutFileQuery {

rustfs/src/grpc.rs

@@ -807,11 +807,38 @@ impl Node for NodeService {
                                 }
                             }
                             Err(err) => {
+                                if err == rustfs_filemeta::Error::Unexpected {
+                                    let _ = tx
+                                        .send(Ok(WalkDirResponse {
+                                            success: false,
+                                            meta_cache_entry: "".to_string(),
+                                            error_info: Some(err.to_string()),
+                                        }))
+                                        .await;
+
+                                    break;
+                                }
+
                                 if rustfs_filemeta::is_io_eof(&err) {
+                                    let _ = tx
+                                        .send(Ok(WalkDirResponse {
+                                            success: false,
+                                            meta_cache_entry: "".to_string(),
+                                            error_info: Some(err.to_string()),
+                                        }))
+                                        .await;
                                     break;
                                 }
 
                                 println!("get err {:?}", err);
+
+                                let _ = tx
+                                    .send(Ok(WalkDirResponse {
+                                        success: false,
+                                        meta_cache_entry: "".to_string(),
+                                        error_info: Some(err.to_string()),
+                                    }))
+                                    .await;
                                 break;
                             }
                         }