add tracing

ecstore/src/notification_sys.rs

@@ -133,6 +133,7 @@ impl NotificationSys {
         }
     }
 
+    #[tracing::instrument(skip(self))]
     pub async fn load_rebalance_meta(&self, start: bool) {
         let mut futures = Vec::with_capacity(self.peer_clients.len());
         for client in self.peer_clients.iter().flatten() {

ecstore/src/pools.rs

@@ -592,6 +592,7 @@ impl ECStore {
         }
     }
 
+    #[tracing::instrument(skip(self))]
     pub async fn decommission_cancel(&self, idx: usize) -> Result<()> {
         if self.single_pool() {
             return Err(Error::msg("InvalidArgument"));
@@ -625,6 +626,7 @@ impl ECStore {
         false
     }
 
+    #[tracing::instrument(skip(self, rx))]
     pub async fn decommission(&self, rx: B_Receiver<bool>, indices: Vec<usize>) -> Result<()> {
         if indices.is_empty() {
             return Err(Error::msg("errInvalidArgument"));
@@ -650,6 +652,7 @@ impl ECStore {
     }
 
     #[allow(unused_assignments)]
+    #[tracing::instrument(skip(self, set, wk, rcfg))]
     async fn decommission_entry(
         self: &Arc<Self>,
         idx: usize,
@@ -844,6 +847,7 @@ impl ECStore {
         }
     }
 
+    #[tracing::instrument(skip(self, rx))]
     async fn decommission_pool(
         self: &Arc<Self>,
         rx: B_Receiver<bool>,
@@ -908,6 +912,7 @@ impl ECStore {
         Ok(())
     }
 
+    #[tracing::instrument(skip(self, rx))]
     pub async fn do_decommission_in_routine(self: &Arc<Self>, rx: B_Receiver<bool>, idx: usize) {
         if let Err(err) = self.decommission_in_background(rx, idx).await {
             error!("decom err {:?}", &err);
@@ -941,6 +946,7 @@ impl ECStore {
         }
     }
 
+    #[tracing::instrument(skip(self))]
     pub async fn decommission_failed(&self, idx: usize) -> Result<()> {
         if self.single_pool() {
             return Err(Error::msg("errInvalidArgument"));
@@ -957,6 +963,7 @@ impl ECStore {
         Ok(())
     }
 
+    #[tracing::instrument(skip(self))]
     pub async fn complete_decommission(&self, idx: usize) -> Result<()> {
         if self.single_pool() {
             return Err(Error::msg("errInvalidArgument"));
@@ -973,6 +980,7 @@ impl ECStore {
         Ok(())
     }
 
+    #[tracing::instrument(skip(self, rx))]
     async fn decommission_in_background(self: &Arc<Self>, rx: B_Receiver<bool>, idx: usize) -> Result<()> {
         let pool = self.pools[idx].clone();
 
@@ -1024,6 +1032,7 @@ impl ECStore {
         Ok(())
     }
 
+    #[tracing::instrument(skip(self))]
     pub async fn start_decommission(&self, indices: Vec<usize>) -> Result<()> {
         if indices.is_empty() {
             return Err(Error::msg("errInvalidArgument"));
@@ -1096,6 +1105,7 @@ impl ECStore {
         Ok(ret)
     }
 
+    #[tracing::instrument(skip(self, rd))]
     async fn decommission_object(self: Arc<Self>, pool_idx: usize, bucket: String, rd: GetObjectReader) -> Result<()> {
         let object_info = rd.object_info.clone();
 
@@ -1228,12 +1238,12 @@ impl ECStore {
 pub type ListCallback = Arc<dyn Fn(MetaCacheEntry) -> BoxFuture<'static, ()> + Send + Sync + 'static>;
 
 impl SetDisks {
-    //
+    #[tracing::instrument(skip(self, rx, cb_func))]
     async fn list_objects_to_decommission(
         self: &Arc<Self>,
         rx: B_Receiver<bool>,
         bucket_info: DecomBucketInfo,
-        func: ListCallback,
+        cb_func: ListCallback,
     ) -> Result<()> {
         let (disks, _) = self.get_online_disks_with_healing(false).await;
         if disks.is_empty() {
@@ -1260,7 +1270,7 @@ impl SetDisks {
                 partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<Error>]| {
                     let resolver = resolver.clone();
                     if let Ok(Some(entry)) = entries.resolve(resolver) {
-                        func(entry)
+                        cb_func(entry)
                     } else {
                         Box::pin(async {})
                     }

ecstore/src/rebalance.rs

@@ -211,6 +211,7 @@ impl RebalanceMeta {
 }
 
 impl ECStore {
+    #[tracing::instrument(skip_all)]
     pub async fn load_rebalance_meta(&self) -> Result<()> {
         let mut meta = RebalanceMeta::new();
         match meta.load(self.pools[0].clone()).await {
@@ -233,6 +234,7 @@ impl ECStore {
         Ok(())
     }
 
+    #[tracing::instrument(skip_all)]
     pub async fn update_rebalance_stats(&self) -> Result<()> {
         let mut ok = false;
         let mut rebalance_meta = self.rebalance_meta.write().await;
@@ -263,6 +265,7 @@ impl ECStore {
         None
     }
 
+    #[tracing::instrument(skip(self))]
     pub async fn init_rebalance_meta(&self, bucktes: Vec<String>) -> Result<String> {
         let si = self.storage_info().await;
 
@@ -326,6 +329,7 @@ impl ECStore {
         Ok(id)
     }
 
+    #[tracing::instrument(skip(self, fi))]
     pub async fn update_pool_stats(&self, pool_index: usize, bucket: String, fi: &FileInfo) -> Result<()> {
         let mut rebalance_meta = self.rebalance_meta.write().await;
         if let Some(meta) = rebalance_meta.as_mut() {
@@ -337,6 +341,7 @@ impl ECStore {
         Ok(())
     }
 
+    #[tracing::instrument(skip(self))]
     pub async fn next_rebal_bucket(&self, pool_index: usize) -> Result<Option<String>> {
         let rebalance_meta = self.rebalance_meta.read().await;
         if let Some(meta) = rebalance_meta.as_ref() {
@@ -355,6 +360,7 @@ impl ECStore {
         Ok(None)
     }
 
+    #[tracing::instrument(skip(self))]
     pub async fn bucket_rebalance_done(&self, pool_index: usize, bucket: String) -> Result<()> {
         let mut rebalance_meta = self.rebalance_meta.write().await;
         if let Some(meta) = rebalance_meta.as_mut() {
@@ -404,6 +410,7 @@ impl ECStore {
         false
     }
 
+    #[tracing::instrument(skip(self))]
     pub async fn stop_rebalance(self: &Arc<Self>) -> Result<()> {
         let rebalance_meta = self.rebalance_meta.read().await;
         if let Some(meta) = rebalance_meta.as_ref() {
@@ -415,6 +422,7 @@ impl ECStore {
         Ok(())
     }
 
+    #[tracing::instrument(skip_all)]
     pub async fn start_rebalance(self: &Arc<Self>) {
         // let rebalance_meta = self.rebalance_meta.read().await;
 
@@ -473,6 +481,7 @@ impl ECStore {
         }
     }
 
+    #[tracing::instrument(skip(self, rx))]
     async fn rebalance_buckets(self: &Arc<Self>, rx: B_Receiver<bool>, pool_index: usize) -> Result<()> {
         let (done_tx, mut done_rx) = tokio::sync::mpsc::channel::<Result<()>>(1);
 
@@ -589,6 +598,7 @@ impl ECStore {
     }
 
     #[allow(unused_assignments)]
+    #[tracing::instrument(skip(self, wk, set))]
     async fn rebalance_entry(
         &self,
         bucket: String,
@@ -763,6 +773,8 @@ impl ECStore {
             }
         }
     }
+
+    #[tracing::instrument(skip(self, rd))]
     async fn rebalance_object(&self, pool_idx: usize, bucket: String, rd: GetObjectReader) -> Result<()> {
         let object_info = rd.object_info.clone();
 
@@ -889,6 +901,7 @@ impl ECStore {
         Ok(())
     }
 
+    #[tracing::instrument(skip(self, rx))]
     async fn rebalance_bucket(self: &Arc<Self>, rx: B_Receiver<bool>, bucket: String, pool_index: usize) -> Result<()> {
         // Placeholder for actual bucket rebalance logic
         tracing::info!("Rebalancing bucket {} in pool {}", bucket, pool_index);
@@ -944,6 +957,7 @@ impl ECStore {
         Ok(())
     }
 
+    #[tracing::instrument(skip(self))]
     pub async fn save_rebalance_stats(&self, pool_idx: usize, opt: RebalSaveOpt) -> Result<()> {
         // TODO: NSLOOK
 
@@ -989,6 +1003,7 @@ impl ECStore {
 }
 
 impl SetDisks {
+    #[tracing::instrument(skip(self, rx, cb))]
     pub async fn list_objects_to_rebalance(
         self: &Arc<Self>,
         rx: B_Receiver<bool>,

rustfs/src/admin/handlers/pools.rs

@@ -14,6 +14,7 @@ pub struct ListPools {}
 #[async_trait::async_trait]
 impl Operation for ListPools {
     // GET <endpoint>/<admin-API>/pools/list
+    #[tracing::instrument(skip_all)]
     async fn call(&self, _req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         warn!("handle ListPools");
 
@@ -60,6 +61,7 @@ pub struct StatusPool {}
 #[async_trait::async_trait]
 impl Operation for StatusPool {
     // GET <endpoint>/<admin-API>/pools/status?pool=http://server{1...4}/disk{1...4}
+    #[tracing::instrument(skip_all)]
     async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         warn!("handle StatusPool");
 
@@ -122,6 +124,7 @@ pub struct StartDecommission {}
 #[async_trait::async_trait]
 impl Operation for StartDecommission {
     // POST <endpoint>/<admin-API>/pools/decommission?pool=http://server{1...4}/disk{1...4}
+    #[tracing::instrument(skip_all)]
     async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         warn!("handle StartDecommission");
 
@@ -204,6 +207,7 @@ pub struct CancelDecommission {}
 #[async_trait::async_trait]
 impl Operation for CancelDecommission {
     // POST <endpoint>/<admin-API>/pools/cancel?pool=http://server{1...4}/disk{1...4}
+    #[tracing::instrument(skip_all)]
     async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         warn!("handle CancelDecommission");
 

rustfs/src/admin/handlers/rebalance.rs

@@ -63,6 +63,7 @@ pub struct RebalanceStart {}
 
 #[async_trait::async_trait]
 impl Operation for RebalanceStart {
+    #[tracing::instrument(skip_all)]
     async fn call(&self, _req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         warn!("handle RebalanceStart");
 
@@ -118,6 +119,7 @@ pub struct RebalanceStatus {}
 
 #[async_trait::async_trait]
 impl Operation for RebalanceStatus {
+    #[tracing::instrument(skip_all)]
     async fn call(&self, _req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         warn!("handle RebalanceStatus");
 
@@ -213,6 +215,7 @@ pub struct RebalanceStop {}
 
 #[async_trait::async_trait]
 impl Operation for RebalanceStop {
+    #[tracing::instrument(skip_all)]
     async fn call(&self, _req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         warn!("handle RebalanceStop");
 

rustfs/src/console.rs

@@ -17,7 +17,7 @@ use mime_guess::from_path;
 use rust_embed::RustEmbed;
 use serde::Serialize;
 use shadow_rs::shadow;
-use std::net::{IpAddr, Ipv4Addr, SocketAddr};
+use std::net::{Ipv4Addr, SocketAddr};
 use std::sync::OnceLock;
 use std::time::Duration;
 use tokio::signal;

rustfs/src/grpc.rs

@@ -2374,6 +2374,7 @@ impl Node for NodeService {
         }))
     }
 
+    #[tracing::instrument(skip_all)]
     async fn load_rebalance_meta(
         &self,
         request: Request<LoadRebalanceMetaRequest>,