From cf70e12e14e60cfa8a2757d02667b8687175ca96 Mon Sep 17 00:00:00 2001 From: weisd Date: Fri, 18 Apr 2025 16:21:14 +0800 Subject: [PATCH] add tracing --- ecstore/src/notification_sys.rs | 1 + ecstore/src/pools.rs | 16 +++++++++++++--- ecstore/src/rebalance.rs | 15 +++++++++++++++ rustfs/src/admin/handlers/pools.rs | 4 ++++ rustfs/src/admin/handlers/rebalance.rs | 3 +++ rustfs/src/console.rs | 2 +- rustfs/src/grpc.rs | 1 + 7 files changed, 38 insertions(+), 4 deletions(-) diff --git a/ecstore/src/notification_sys.rs b/ecstore/src/notification_sys.rs index 9bb73bc43..8bc810fdc 100644 --- a/ecstore/src/notification_sys.rs +++ b/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() { diff --git a/ecstore/src/pools.rs b/ecstore/src/pools.rs index c3e2eed70..e2249dcc2 100644 --- a/ecstore/src/pools.rs +++ b/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, indices: Vec) -> 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, idx: usize, @@ -844,6 +847,7 @@ impl ECStore { } } + #[tracing::instrument(skip(self, rx))] async fn decommission_pool( self: &Arc, rx: B_Receiver, @@ -908,6 +912,7 @@ impl ECStore { Ok(()) } + #[tracing::instrument(skip(self, rx))] pub async fn do_decommission_in_routine(self: &Arc, rx: B_Receiver, 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, rx: B_Receiver, 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) -> 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, 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 BoxFuture<'static, ()> + Send + Sync + 'static>; impl SetDisks { - // + #[tracing::instrument(skip(self, rx, cb_func))] async fn list_objects_to_decommission( self: &Arc, rx: B_Receiver, 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]| { let resolver = resolver.clone(); if let Ok(Some(entry)) = entries.resolve(resolver) { - func(entry) + cb_func(entry) } else { Box::pin(async {}) } diff --git a/ecstore/src/rebalance.rs b/ecstore/src/rebalance.rs index 74cf21c48..56f10f0d0 100644 --- a/ecstore/src/rebalance.rs +++ b/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) -> Result { 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> { 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) -> 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) { // 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, rx: B_Receiver, pool_index: usize) -> Result<()> { let (done_tx, mut done_rx) = tokio::sync::mpsc::channel::>(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, rx: B_Receiver, 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, rx: B_Receiver, diff --git a/rustfs/src/admin/handlers/pools.rs b/rustfs/src/admin/handlers/pools.rs index bd3152790..c98432f0b 100644 --- a/rustfs/src/admin/handlers/pools.rs +++ b/rustfs/src/admin/handlers/pools.rs @@ -14,6 +14,7 @@ pub struct ListPools {} #[async_trait::async_trait] impl Operation for ListPools { // GET //pools/list + #[tracing::instrument(skip_all)] async fn call(&self, _req: S3Request, _params: Params<'_, '_>) -> S3Result> { warn!("handle ListPools"); @@ -60,6 +61,7 @@ pub struct StatusPool {} #[async_trait::async_trait] impl Operation for StatusPool { // GET //pools/status?pool=http://server{1...4}/disk{1...4} + #[tracing::instrument(skip_all)] async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { warn!("handle StatusPool"); @@ -122,6 +124,7 @@ pub struct StartDecommission {} #[async_trait::async_trait] impl Operation for StartDecommission { // POST //pools/decommission?pool=http://server{1...4}/disk{1...4} + #[tracing::instrument(skip_all)] async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { warn!("handle StartDecommission"); @@ -204,6 +207,7 @@ pub struct CancelDecommission {} #[async_trait::async_trait] impl Operation for CancelDecommission { // POST //pools/cancel?pool=http://server{1...4}/disk{1...4} + #[tracing::instrument(skip_all)] async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { warn!("handle CancelDecommission"); diff --git a/rustfs/src/admin/handlers/rebalance.rs b/rustfs/src/admin/handlers/rebalance.rs index 06b71e676..65b5277fc 100644 --- a/rustfs/src/admin/handlers/rebalance.rs +++ b/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, _params: Params<'_, '_>) -> S3Result> { 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, _params: Params<'_, '_>) -> S3Result> { 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, _params: Params<'_, '_>) -> S3Result> { warn!("handle RebalanceStop"); diff --git a/rustfs/src/console.rs b/rustfs/src/console.rs index f2eda0995..127d2144e 100644 --- a/rustfs/src/console.rs +++ b/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; diff --git a/rustfs/src/grpc.rs b/rustfs/src/grpc.rs index f2fde2c2c..6e702617b 100644 --- a/rustfs/src/grpc.rs +++ b/rustfs/src/grpc.rs @@ -2374,6 +2374,7 @@ impl Node for NodeService { })) } + #[tracing::instrument(skip_all)] async fn load_rebalance_meta( &self, request: Request,