diff --git a/crates/obs/src/telemetry.rs b/crates/obs/src/telemetry.rs index 9ffc265dd..654ae4145 100644 --- a/crates/obs/src/telemetry.rs +++ b/crates/obs/src/telemetry.rs @@ -3,7 +3,7 @@ use crate::utils::get_local_ip_with_default; use crate::OtelConfig; use opentelemetry::trace::TracerProvider; use opentelemetry::{global, KeyValue}; -use opentelemetry_appender_tracing::layer; +use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge; use opentelemetry_otlp::WithExportConfig; use opentelemetry_sdk::logs::SdkLoggerProvider; use opentelemetry_sdk::{ @@ -202,55 +202,33 @@ pub fn init_telemetry(config: &OtelConfig) -> OtelGuard { // configuring tracing { - // optimize filter configuration - let otel_layer = { - let filter_otel = match logger_level { - "trace" | "debug" => { - info!("OpenTelemetry tracing initialized with level: {}", logger_level); - EnvFilter::new(logger_level) - } - _ => { - let mut filter = EnvFilter::new(logger_level); - - // use smallvec to avoid heap allocation - let directives: SmallVec<[&str; 5]> = smallvec::smallvec!["hyper", "tonic", "h2", "reqwest", "tower"]; - - for directive in directives { - filter = filter.add_directive(format!("{}=off", directive).parse().unwrap()); - } - filter - } - }; - layer::OpenTelemetryTracingBridge::new(&logger_provider).with_filter(filter_otel) - }; - - let tracer = tracer_provider.tracer(Cow::Borrowed(service_name).to_string()); - - // Configure registry to avoid repeated calls to filter methods - let level_filter = switch_level(logger_level); - let registry = tracing_subscriber::registry() - .with(level_filter) - .with(OpenTelemetryLayer::new(tracer)) - .with(MetricsLayer::new(meter_provider.clone())) - .with(otel_layer) - .with(EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(logger_level))); - // configure the formatting layer let enable_color = std::io::stdout().is_terminal(); let fmt_layer = tracing_subscriber::fmt::layer() + .with_target(true) .with_ansi(enable_color) .with_thread_names(true) + .with_thread_ids(true) .with_file(true) - .with_line_number(true) - .with_filter( - EnvFilter::new(logger_level).add_directive( - format!("opentelemetry={}", if endpoint.is_empty() { logger_level } else { "off" }) - .parse() - .unwrap(), - ), - ); + .with_line_number(true); - registry.with(ErrorLayer::default()).with(fmt_layer).init(); + let filter = build_env_filter(logger_level, None); + + let tracer = tracer_provider.tracer(Cow::Borrowed(service_name).to_string()); + + let otel_filter = build_env_filter(logger_level, None); + let otel_layer = OpenTelemetryTracingBridge::new(&logger_provider).with_filter(otel_filter); + + // Configure registry to avoid repeated calls to filter methods + tracing_subscriber::registry() + .with(filter) + .with(fmt_layer) + .with(ErrorLayer::default()) + .with(otel_layer) + .with(MetricsLayer::new(meter_provider.clone())) + .with(OpenTelemetryLayer::new(tracer)) + .with(ErrorLayer::default()) + .init(); if !endpoint.is_empty() { info!( @@ -267,15 +245,16 @@ pub fn init_telemetry(config: &OtelConfig) -> OtelGuard { } } -/// Switch log level -fn switch_level(logger_level: &str) -> tracing_subscriber::filter::LevelFilter { - use tracing_subscriber::filter::LevelFilter; - match logger_level { - "error" => LevelFilter::ERROR, - "warn" => LevelFilter::WARN, - "info" => LevelFilter::INFO, - "debug" => LevelFilter::DEBUG, - "trace" => LevelFilter::TRACE, - _ => LevelFilter::OFF, +fn build_env_filter(logger_level: &str, default_level: Option<&str>) -> EnvFilter { + let level = default_level.unwrap_or(logger_level); + let mut filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(level)); + + if !matches!(logger_level, "trace" | "debug") { + let directives: SmallVec<[&str; 5]> = smallvec::smallvec!["hyper", "tonic", "h2", "reqwest", "tower"]; + for directive in directives { + filter = filter.add_directive(format!("{}=off", directive).parse().unwrap()); + } } + + filter } diff --git a/ecstore/src/bitrot.rs b/ecstore/src/bitrot.rs index 757401557..68ccff71e 100644 --- a/ecstore/src/bitrot.rs +++ b/ecstore/src/bitrot.rs @@ -6,6 +6,7 @@ use crate::{ }; use blake2::Blake2b512; use blake2::Digest as _; +use bytes::Bytes; use common::error::{Error, Result}; use highway::{HighwayHash, HighwayHasher, Key}; use lazy_static::lazy_static; @@ -533,20 +534,26 @@ impl Writer for BitrotFileWriter { self } - async fn write(&mut self, buf: &[u8]) -> Result<()> { + async fn write(&mut self, buf: Bytes) -> Result<()> { if buf.is_empty() { return Ok(()); } - self.hasher.reset(); - self.hasher.update(buf); - let hash_bytes = self.hasher.clone().finalize(); + let mut hasher = self.hasher.clone(); + let h_buf = buf.clone(); + let hash_bytes = tokio::spawn(async move { + hasher.reset(); + hasher.update(h_buf); + hasher.finalize() + }) + .await + .unwrap(); if let Some(f) = self.inner.as_mut() { f.write_all(&hash_bytes).await?; - f.write_all(buf).await?; + f.write_all(&buf).await?; } else { self.inline_data.extend_from_slice(&hash_bytes); - self.inline_data.extend_from_slice(buf); + self.inline_data.extend_from_slice(&buf); } Ok(()) diff --git a/ecstore/src/disk/error.rs b/ecstore/src/disk/error.rs index d7ea2854a..0d422b106 100644 --- a/ecstore/src/disk/error.rs +++ b/ecstore/src/disk/error.rs @@ -364,6 +364,10 @@ pub fn is_unformatted_disk(err: &Error) -> bool { } pub fn is_err_file_not_found(err: &Error) -> bool { + if let Some(ioerr) = err.downcast_ref::() { + return ioerr.kind() == ErrorKind::NotFound; + } + matches!(err.downcast_ref::(), Some(DiskError::FileNotFound)) } diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index ddefb55fd..74cdaa690 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -629,7 +629,11 @@ impl LocalDisk { let _ = fm.data.remove(vec![vid, dir]); let dir_path = self.get_object_path(volume, format!("{}/{}", path, dir).as_str())?; - self.move_to_trash(&dir_path, true, false).await?; + if let Err(err) = self.move_to_trash(&dir_path, true, false).await { + if !(is_err_file_not_found(&err) || is_err_os_not_exist(&err)) { + return Err(err); + } + }; } } @@ -1075,30 +1079,39 @@ fn skip_access_checks(p: impl AsRef) -> bool { #[async_trait::async_trait] impl DiskAPI for LocalDisk { + #[tracing::instrument(skip(self))] fn to_string(&self) -> String { self.root.to_string_lossy().to_string() } + #[tracing::instrument(skip(self))] fn is_local(&self) -> bool { true } + #[tracing::instrument(skip(self))] fn host_name(&self) -> String { self.endpoint.host_port() } + #[tracing::instrument(skip(self))] async fn is_online(&self) -> bool { self.check_format_json().await.is_ok() } + #[tracing::instrument(skip(self))] fn endpoint(&self) -> Endpoint { self.endpoint.clone() } + #[tracing::instrument(skip(self))] async fn close(&self) -> Result<()> { Ok(()) } + + #[tracing::instrument(skip(self))] fn path(&self) -> PathBuf { self.root.clone() } + #[tracing::instrument(skip(self))] fn get_disk_location(&self) -> DiskLocation { DiskLocation { pool_idx: { @@ -1175,6 +1188,7 @@ impl DiskAPI for LocalDisk { Ok(Some(disk_id)) } + #[tracing::instrument(skip(self))] async fn set_disk_id(&self, id: Option) -> Result<()> { // 本地不需要设置 // TODO: add check_id_store @@ -1184,6 +1198,7 @@ impl DiskAPI for LocalDisk { } #[must_use] + #[tracing::instrument(skip(self))] async fn read_all(&self, volume: &str, path: &str) -> Result> { if volume == super::RUSTFS_META_BUCKET && path == super::FORMAT_CONFIG_FILE { let format_info = self.format_info.read().await; @@ -1198,10 +1213,12 @@ impl DiskAPI for LocalDisk { Ok(data) } + #[tracing::instrument(skip(self))] async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()> { self.write_all_public(volume, path, data).await } + #[tracing::instrument(skip(self))] async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> { let volume_dir = self.get_bucket_path(volume)?; if !skip_access_checks(volume) { @@ -1219,6 +1236,7 @@ impl DiskAPI for LocalDisk { Ok(()) } + #[tracing::instrument(skip(self))] async fn verify_file(&self, volume: &str, path: &str, fi: &FileInfo) -> Result { let volume_dir = self.get_bucket_path(volume)?; if !skip_access_checks(volume) { @@ -1228,7 +1246,7 @@ impl DiskAPI for LocalDisk { } let mut resp = CheckPartsResp { - results: Vec::with_capacity(fi.parts.len()), + results: vec![0; fi.parts.len()], }; let erasure = &fi.erasure; @@ -1264,6 +1282,7 @@ impl DiskAPI for LocalDisk { Ok(resp) } + #[tracing::instrument(skip(self))] async fn check_parts(&self, volume: &str, path: &str, fi: &FileInfo) -> Result { let volume_dir = self.get_bucket_path(volume)?; check_path_length(volume_dir.join(path).to_string_lossy().as_ref())?; @@ -1401,6 +1420,8 @@ impl DiskAPI for LocalDisk { Ok(()) } + + #[tracing::instrument(skip(self))] async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> { let src_volume_dir = self.get_bucket_path(src_volume)?; let dst_volume_dir = self.get_bucket_path(dst_volume)?; @@ -1700,7 +1721,7 @@ impl DiskAPI for LocalDisk { Ok(()) } - // #[tracing::instrument(skip(self))] + #[tracing::instrument(skip(self))] async fn rename_data( &self, src_volume: &str, @@ -1906,6 +1927,7 @@ impl DiskAPI for LocalDisk { }) } + #[tracing::instrument(skip(self))] async fn make_volumes(&self, volumes: Vec<&str>) -> Result<()> { for vol in volumes { if let Err(e) = self.make_volume(vol).await { @@ -1917,6 +1939,8 @@ impl DiskAPI for LocalDisk { } Ok(()) } + + #[tracing::instrument(skip(self))] async fn make_volume(&self, volume: &str) -> Result<()> { if !Self::is_valid_volname(volume) { return Err(Error::msg("Invalid arguments specified")); @@ -1940,6 +1964,8 @@ impl DiskAPI for LocalDisk { Err(Error::from(DiskError::VolumeExists)) } + + #[tracing::instrument(skip(self))] async fn list_volumes(&self) -> Result> { let mut volumes = Vec::new(); @@ -1964,6 +1990,8 @@ impl DiskAPI for LocalDisk { Ok(volumes) } + + #[tracing::instrument(skip(self))] async fn stat_volume(&self, volume: &str) -> Result { let volume_dir = self.get_bucket_path(volume)?; let meta = match utils::fs::lstat(&volume_dir).await { @@ -1991,6 +2019,8 @@ impl DiskAPI for LocalDisk { created: modtime, }) } + + #[tracing::instrument(skip(self))] async fn delete_paths(&self, volume: &str, paths: &[String]) -> Result<()> { let volume_dir = self.get_bucket_path(volume)?; if !skip_access_checks(volume) { @@ -2009,6 +2039,8 @@ impl DiskAPI for LocalDisk { Ok(()) } + + #[tracing::instrument(skip(self))] async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: &UpdateMetadataOpts) -> Result<()> { if fi.metadata.is_some() { let volume_dir = self.get_bucket_path(volume)?; @@ -2049,6 +2081,8 @@ impl DiskAPI for LocalDisk { Err(Error::msg("Invalid Argument")) } + + #[tracing::instrument(skip(self))] async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> { let p = self.get_object_path(volume, format!("{}/{}", path, super::STORAGE_FORMAT_FILE).as_str())?; @@ -2102,6 +2136,8 @@ impl DiskAPI for LocalDisk { Ok(RawFileInfo { buf }) } + + #[tracing::instrument(skip(self))] async fn delete_version( &self, volume: &str, @@ -2210,6 +2246,8 @@ impl DiskAPI for LocalDisk { Ok(errs) } + + #[tracing::instrument(skip(self))] async fn read_multiple(&self, req: ReadMultipleReq) -> Result> { let mut results = Vec::new(); let mut found = 0; @@ -2269,6 +2307,7 @@ impl DiskAPI for LocalDisk { Ok(results) } + #[tracing::instrument(skip(self))] async fn delete_volume(&self, volume: &str) -> Result<()> { let p = self.get_bucket_path(volume)?; @@ -2291,6 +2330,7 @@ impl DiskAPI for LocalDisk { Ok(()) } + #[tracing::instrument(skip(self))] async fn disk_info(&self, _: &DiskInfoOptions) -> Result { let mut info = Cache::get(self.disk_info_cache.clone()).await?; // TODO: nr_requests, rotational @@ -2450,6 +2490,7 @@ impl DiskAPI for LocalDisk { Ok(data_usage_info) } + #[tracing::instrument(skip(self))] async fn healing(&self) -> Option { let healing_file = path_join(&[ self.path(), diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index e26eebb8b..ddefd1bed 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -49,6 +49,7 @@ pub enum Disk { #[async_trait::async_trait] impl DiskAPI for Disk { + #[tracing::instrument(skip(self))] fn to_string(&self) -> String { match self { Disk::Local(local_disk) => local_disk.to_string(), @@ -56,6 +57,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] fn is_local(&self) -> bool { match self { Disk::Local(local_disk) => local_disk.is_local(), @@ -63,30 +65,39 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] fn host_name(&self) -> String { match self { Disk::Local(local_disk) => local_disk.host_name(), Disk::Remote(remote_disk) => remote_disk.host_name(), } } + + #[tracing::instrument(skip(self))] async fn is_online(&self) -> bool { match self { Disk::Local(local_disk) => local_disk.is_online().await, Disk::Remote(remote_disk) => remote_disk.is_online().await, } } + + #[tracing::instrument(skip(self))] fn endpoint(&self) -> Endpoint { match self { Disk::Local(local_disk) => local_disk.endpoint(), Disk::Remote(remote_disk) => remote_disk.endpoint(), } } + + #[tracing::instrument(skip(self))] async fn close(&self) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.close().await, Disk::Remote(remote_disk) => remote_disk.close().await, } } + + #[tracing::instrument(skip(self))] fn path(&self) -> PathBuf { match self { Disk::Local(local_disk) => local_disk.path(), @@ -94,6 +105,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] fn get_disk_location(&self) -> DiskLocation { match self { Disk::Local(local_disk) => local_disk.get_disk_location(), @@ -101,12 +113,15 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn get_disk_id(&self) -> Result> { match self { Disk::Local(local_disk) => local_disk.get_disk_id().await, Disk::Remote(remote_disk) => remote_disk.get_disk_id().await, } } + + #[tracing::instrument(skip(self))] async fn set_disk_id(&self, id: Option) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.set_disk_id(id).await, @@ -114,6 +129,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn read_all(&self, volume: &str, path: &str) -> Result> { match self { Disk::Local(local_disk) => local_disk.read_all(volume, path).await, @@ -121,6 +137,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.write_all(volume, path, data).await, @@ -128,6 +145,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.delete(volume, path, opt).await, @@ -135,6 +153,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn verify_file(&self, volume: &str, path: &str, fi: &FileInfo) -> Result { match self { Disk::Local(local_disk) => local_disk.verify_file(volume, path, fi).await, @@ -142,6 +161,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn check_parts(&self, volume: &str, path: &str, fi: &FileInfo) -> Result { match self { Disk::Local(local_disk) => local_disk.check_parts(volume, path, fi).await, @@ -149,6 +169,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Vec) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.rename_part(src_volume, src_path, dst_volume, dst_path, meta).await, @@ -159,6 +180,8 @@ impl DiskAPI for Disk { } } } + + #[tracing::instrument(skip(self))] async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.rename_file(src_volume, src_path, dst_volume, dst_path).await, @@ -166,6 +189,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn create_file(&self, _origvolume: &str, volume: &str, path: &str, _file_size: usize) -> Result { match self { Disk::Local(local_disk) => local_disk.create_file(_origvolume, volume, path, _file_size).await, @@ -173,6 +197,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn append_file(&self, volume: &str, path: &str) -> Result { match self { Disk::Local(local_disk) => local_disk.append_file(volume, path).await, @@ -180,6 +205,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn read_file(&self, volume: &str, path: &str) -> Result { match self { Disk::Local(local_disk) => local_disk.read_file(volume, path).await, @@ -187,6 +213,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result { match self { Disk::Local(local_disk) => local_disk.read_file_stream(volume, path, offset, length).await, @@ -194,6 +221,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn list_dir(&self, _origvolume: &str, volume: &str, _dir_path: &str, _count: i32) -> Result> { match self { Disk::Local(local_disk) => local_disk.list_dir(_origvolume, volume, _dir_path, _count).await, @@ -201,6 +229,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self, wr))] async fn walk_dir(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.walk_dir(opts, wr).await, @@ -208,6 +237,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn rename_data( &self, src_volume: &str, @@ -222,6 +252,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn make_volumes(&self, volumes: Vec<&str>) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.make_volumes(volumes).await, @@ -229,6 +260,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn make_volume(&self, volume: &str) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.make_volume(volume).await, @@ -236,6 +268,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn list_volumes(&self) -> Result> { match self { Disk::Local(local_disk) => local_disk.list_volumes().await, @@ -243,6 +276,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn stat_volume(&self, volume: &str) -> Result { match self { Disk::Local(local_disk) => local_disk.stat_volume(volume).await, @@ -250,12 +284,15 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn delete_paths(&self, volume: &str, paths: &[String]) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.delete_paths(volume, paths).await, Disk::Remote(remote_disk) => remote_disk.delete_paths(volume, paths).await, } } + + #[tracing::instrument(skip(self))] async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: &UpdateMetadataOpts) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.update_metadata(volume, path, fi, opts).await, @@ -263,6 +300,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.write_metadata(_org_volume, volume, path, fi).await, @@ -285,13 +323,15 @@ impl DiskAPI for Disk { } } - #[tracing::instrument] + #[tracing::instrument(skip(self))] async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result { match self { Disk::Local(local_disk) => local_disk.read_xl(volume, path, read_data).await, Disk::Remote(remote_disk) => remote_disk.read_xl(volume, path, read_data).await, } } + + #[tracing::instrument(skip(self))] async fn delete_version( &self, volume: &str, @@ -305,6 +345,8 @@ impl DiskAPI for Disk { Disk::Remote(remote_disk) => remote_disk.delete_version(volume, path, fi, force_del_marker, opts).await, } } + + #[tracing::instrument(skip(self))] async fn delete_versions( &self, volume: &str, @@ -317,6 +359,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn read_multiple(&self, req: ReadMultipleReq) -> Result> { match self { Disk::Local(local_disk) => local_disk.read_multiple(req).await, @@ -324,6 +367,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn delete_volume(&self, volume: &str) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.delete_volume(volume).await, @@ -331,6 +375,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn disk_info(&self, opts: &DiskInfoOptions) -> Result { match self { Disk::Local(local_disk) => local_disk.disk_info(opts).await, @@ -338,6 +383,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self, cache, we_sleep, scan_mode))] async fn ns_scanner( &self, cache: &DataUsageCache, @@ -351,6 +397,7 @@ impl DiskAPI for Disk { } } + #[tracing::instrument(skip(self))] async fn healing(&self) -> Option { match self { Disk::Local(local_disk) => local_disk.healing().await, diff --git a/ecstore/src/disk/remote.rs b/ecstore/src/disk/remote.rs index c635d744a..2b4e64dc2 100644 --- a/ecstore/src/disk/remote.rs +++ b/ecstore/src/disk/remote.rs @@ -70,17 +70,21 @@ impl RemoteDisk { // TODO: all api need to handle errors #[async_trait::async_trait] impl DiskAPI for RemoteDisk { + #[tracing::instrument(skip(self))] fn to_string(&self) -> String { self.endpoint.to_string() } + #[tracing::instrument(skip(self))] fn is_local(&self) -> bool { false } + #[tracing::instrument(skip(self))] fn host_name(&self) -> String { self.endpoint.host_port() } + #[tracing::instrument(skip(self))] async fn is_online(&self) -> bool { // TODO: 连接状态 if (node_service_time_out_client(&self.addr).await).is_ok() { @@ -88,16 +92,20 @@ impl DiskAPI for RemoteDisk { } false } + #[tracing::instrument(skip(self))] fn endpoint(&self) -> Endpoint { self.endpoint.clone() } + #[tracing::instrument(skip(self))] async fn close(&self) -> Result<()> { Ok(()) } + #[tracing::instrument(skip(self))] fn path(&self) -> PathBuf { self.root.clone() } + #[tracing::instrument(skip(self))] fn get_disk_location(&self) -> DiskLocation { DiskLocation { pool_idx: { @@ -124,9 +132,12 @@ impl DiskAPI for RemoteDisk { } } + #[tracing::instrument(skip(self))] async fn get_disk_id(&self) -> Result> { Ok(*self.id.lock().await) } + + #[tracing::instrument(skip(self))] async fn set_disk_id(&self, id: Option) -> Result<()> { let mut lock = self.id.lock().await; *lock = id; @@ -134,6 +145,7 @@ impl DiskAPI for RemoteDisk { Ok(()) } + #[tracing::instrument(skip(self))] async fn read_all(&self, volume: &str, path: &str) -> Result> { info!("read_all {}/{}", volume, path); let mut client = node_service_time_out_client(&self.addr) @@ -154,6 +166,7 @@ impl DiskAPI for RemoteDisk { Ok(response.data) } + #[tracing::instrument(skip(self))] async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()> { info!("write_all"); let mut client = node_service_time_out_client(&self.addr) @@ -179,6 +192,7 @@ impl DiskAPI for RemoteDisk { Ok(()) } + #[tracing::instrument(skip(self))] async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> { info!("delete {}/{}/{}", self.endpoint.to_string(), volume, path); let options = serde_json::to_string(&opt)?; @@ -205,6 +219,7 @@ impl DiskAPI for RemoteDisk { Ok(()) } + #[tracing::instrument(skip(self))] async fn verify_file(&self, volume: &str, path: &str, fi: &FileInfo) -> Result { info!("verify_file"); let file_info = serde_json::to_string(&fi)?; @@ -233,6 +248,7 @@ impl DiskAPI for RemoteDisk { Ok(check_parts_resp) } + #[tracing::instrument(skip(self))] async fn check_parts(&self, volume: &str, path: &str, fi: &FileInfo) -> Result { info!("check_parts"); let file_info = serde_json::to_string(&fi)?; @@ -261,6 +277,7 @@ impl DiskAPI for RemoteDisk { Ok(check_parts_resp) } + #[tracing::instrument(skip(self))] async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Vec) -> Result<()> { info!("rename_part {}/{}", src_volume, src_path); let mut client = node_service_time_out_client(&self.addr) @@ -287,6 +304,7 @@ impl DiskAPI for RemoteDisk { Ok(()) } + #[tracing::instrument(level = "debug", skip(self))] async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> { info!("rename_file"); @@ -365,6 +383,7 @@ impl DiskAPI for RemoteDisk { )) } + #[tracing::instrument(skip(self))] async fn list_dir(&self, _origvolume: &str, volume: &str, _dir_path: &str, _count: i32) -> Result> { info!("list_dir {}/{}", volume, _dir_path); let mut client = node_service_time_out_client(&self.addr) @@ -389,6 +408,7 @@ impl DiskAPI for RemoteDisk { } // FIXME: TODO: use writer + #[tracing::instrument(skip(self, wr))] async fn walk_dir(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> { let now = std::time::SystemTime::now(); info!("walk_dir {}/{}/{:?}", self.endpoint.to_string(), opts.bucket, opts.filter_prefix); @@ -429,6 +449,7 @@ impl DiskAPI for RemoteDisk { Ok(()) } + #[tracing::instrument(skip(self))] async fn rename_data( &self, src_volume: &str, @@ -466,6 +487,7 @@ impl DiskAPI for RemoteDisk { Ok(rename_data_resp) } + #[tracing::instrument(skip(self))] async fn make_volumes(&self, volumes: Vec<&str>) -> Result<()> { info!("make_volumes"); let mut client = node_service_time_out_client(&self.addr) @@ -489,6 +511,7 @@ impl DiskAPI for RemoteDisk { Ok(()) } + #[tracing::instrument(skip(self))] async fn make_volume(&self, volume: &str) -> Result<()> { info!("make_volume"); let mut client = node_service_time_out_client(&self.addr) @@ -512,6 +535,7 @@ impl DiskAPI for RemoteDisk { Ok(()) } + #[tracing::instrument(skip(self))] async fn list_volumes(&self) -> Result> { info!("list_volumes"); let mut client = node_service_time_out_client(&self.addr) @@ -540,6 +564,7 @@ impl DiskAPI for RemoteDisk { Ok(infos) } + #[tracing::instrument(skip(self))] async fn stat_volume(&self, volume: &str) -> Result { info!("stat_volume"); let mut client = node_service_time_out_client(&self.addr) @@ -565,6 +590,7 @@ impl DiskAPI for RemoteDisk { Ok(volume_info) } + #[tracing::instrument(skip(self))] async fn delete_paths(&self, volume: &str, paths: &[String]) -> Result<()> { info!("delete_paths"); let paths = paths.to_owned(); @@ -589,6 +615,8 @@ impl DiskAPI for RemoteDisk { Ok(()) } + + #[tracing::instrument(skip(self))] async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: &UpdateMetadataOpts) -> Result<()> { info!("update_metadata"); let file_info = serde_json::to_string(&fi)?; @@ -618,6 +646,7 @@ impl DiskAPI for RemoteDisk { Ok(()) } + #[tracing::instrument(skip(self))] async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> { info!("write_metadata {}/{}", volume, path); let file_info = serde_json::to_string(&fi)?; @@ -644,6 +673,7 @@ impl DiskAPI for RemoteDisk { Ok(()) } + #[tracing::instrument(skip(self))] async fn read_version( &self, _org_volume: &str, @@ -707,6 +737,8 @@ impl DiskAPI for RemoteDisk { Ok(raw_file_info) } + + #[tracing::instrument(skip(self))] async fn delete_version( &self, volume: &str, @@ -745,6 +777,8 @@ impl DiskAPI for RemoteDisk { Ok(()) } + + #[tracing::instrument(skip(self))] async fn delete_versions( &self, volume: &str, @@ -790,6 +824,7 @@ impl DiskAPI for RemoteDisk { Ok(errors) } + #[tracing::instrument(skip(self))] async fn read_multiple(&self, req: ReadMultipleReq) -> Result> { info!("read_multiple {}/{}/{}", self.endpoint.to_string(), req.bucket, req.prefix); let read_multiple_req = serde_json::to_string(&req)?; @@ -820,6 +855,7 @@ impl DiskAPI for RemoteDisk { Ok(read_multiple_resps) } + #[tracing::instrument(skip(self))] async fn delete_volume(&self, volume: &str) -> Result<()> { info!("delete_volume {}/{}", self.endpoint.to_string(), volume); let mut client = node_service_time_out_client(&self.addr) @@ -843,6 +879,7 @@ impl DiskAPI for RemoteDisk { Ok(()) } + #[tracing::instrument(skip(self))] async fn disk_info(&self, opts: &DiskInfoOptions) -> Result { let opts = serde_json::to_string(&opts)?; let mut client = node_service_time_out_client(&self.addr) @@ -868,6 +905,7 @@ impl DiskAPI for RemoteDisk { Ok(disk_info) } + #[tracing::instrument(skip(self, cache, scan_mode, _we_sleep))] async fn ns_scanner( &self, cache: &DataUsageCache, @@ -909,6 +947,7 @@ impl DiskAPI for RemoteDisk { } } + #[tracing::instrument(skip(self))] async fn healing(&self) -> Option { None } diff --git a/ecstore/src/erasure.rs b/ecstore/src/erasure.rs index d99c7eea5..021ae7bed 100644 --- a/ecstore/src/erasure.rs +++ b/ecstore/src/erasure.rs @@ -86,7 +86,6 @@ impl Erasure { } self.buf.resize(new_len, 0u8); - match reader.read_exact(&mut self.buf).await { Ok(res) => res, Err(e) => { @@ -101,20 +100,19 @@ impl Erasure { } self.encode_data(&self.buf, &mut blocks)?; - let mut errs = Vec::new(); - // TODO: 并发写入 - for (i, w_op) in writers.iter_mut().enumerate() { - if let Some(w) = w_op { - match w.write(blocks[i].as_ref()).await { - Ok(_) => errs.push(None), - Err(e) => errs.push(Some(e)), + let write_futures = writers.iter_mut().enumerate().map(|(i, w_op)| { + let i_inner = i; + let blocks_inner = blocks.clone(); + async move { + if let Some(w) = w_op { + (w.write(blocks_inner[i_inner].clone()).await).err() + } else { + Some(Error::new(DiskError::DiskNotFound)) } - } else { - errs.push(Some(Error::new(DiskError::DiskNotFound))); } - } - + }); + let errs = join_all(write_futures).await; let none_count = errs.iter().filter(|&x| x.is_none()).count(); if none_count >= write_quorum { if total_size == 0 { @@ -472,7 +470,7 @@ impl Erasure { self.encoder.as_ref().unwrap().reconstruct(&mut bufs)?; } - let shards: Vec> = bufs.into_iter().flatten().collect::>(); + let shards = bufs.into_iter().flatten().collect::>(); if shards.len() != self.parity_shards + self.data_shards { return Err(Error::from_string("can not reconstruct data")); } @@ -481,7 +479,7 @@ impl Erasure { if w.is_none() { continue; } - match w.as_mut().unwrap().write(shards[i].as_ref()).await { + match w.as_mut().unwrap().write(shards[i].clone().into()).await { Ok(_) => {} Err(e) => { info!("write failed, err: {:?}", e); @@ -501,7 +499,7 @@ impl Erasure { #[async_trait::async_trait] pub trait Writer { fn as_any(&self) -> &dyn Any; - async fn write(&mut self, buf: &[u8]) -> Result<()>; + async fn write(&mut self, buf: Bytes) -> Result<()>; async fn close(&mut self) -> Result<()> { Ok(()) } diff --git a/ecstore/src/heal/data_scanner.rs b/ecstore/src/heal/data_scanner.rs index 71a8d8a25..b1b110a54 100644 --- a/ecstore/src/heal/data_scanner.rs +++ b/ecstore/src/heal/data_scanner.rs @@ -696,8 +696,8 @@ impl FolderScanner { } } } - if found_objects && *GLOBAL_IsErasure.read().await { - // If we found an object in erasure mode, we skip subdirs (only datadirs)... + // if found_objects && *GLOBAL_IsErasure.read().await { + if found_objects { break 'outer; } @@ -1029,11 +1029,13 @@ impl FolderScanner { #[tracing::instrument(level = "info", skip(into, folder_scanner))] async fn scan(folder: &CachedFolder, into: &mut DataUsageEntry, folder_scanner: &mut FolderScanner) { - if !into.compacted { - *into = DataUsageEntry::default(); - } + let mut dst = if !into.compacted { + DataUsageEntry::default() + } else { + into.clone() + }; - if Box::pin(folder_scanner.scan_folder(folder, into)).await.is_err() { + if Box::pin(folder_scanner.scan_folder(folder, &mut dst)).await.is_err() { return; } if !into.compacted { diff --git a/ecstore/src/heal/data_usage_cache.rs b/ecstore/src/heal/data_usage_cache.rs index 11814c3ef..2dad9f55b 100644 --- a/ecstore/src/heal/data_usage_cache.rs +++ b/ecstore/src/heal/data_usage_cache.rs @@ -21,7 +21,7 @@ use tokio::time::sleep; use tracing::warn; use super::data_scanner::{SizeSummary, DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS}; -use super::data_usage::{BucketTargetUsageInfo, BucketUsageInfo, DataUsageInfo, DATA_USAGE_ROOT}; +use super::data_usage::{BucketTargetUsageInfo, BucketUsageInfo, DataUsageInfo}; // DATA_USAGE_BUCKET_LEN must be length of ObjectsHistogramIntervals pub const DATA_USAGE_BUCKET_LEN: usize = 11; @@ -858,9 +858,5 @@ impl DataUsageHash { } pub fn hash_path(data: &str) -> DataUsageHash { - let mut data = data; - if data != DATA_USAGE_ROOT { - data = data.trim_matches('/'); - } DataUsageHash(Path::new(&data).clean().to_string_lossy().to_string()) } diff --git a/ecstore/src/io.rs b/ecstore/src/io.rs index 3ca27fe0d..f2affe8cb 100644 --- a/ecstore/src/io.rs +++ b/ecstore/src/io.rs @@ -1,3 +1,4 @@ +use bytes::Bytes; use futures::TryStreamExt; use md5::Digest; use md5::Md5; @@ -7,6 +8,7 @@ use std::task::Poll; use tokio::io::AsyncRead; use tokio::io::AsyncWrite; use tokio::io::ReadBuf; +use tokio::sync::mpsc; use tokio::sync::oneshot; use tokio_util::io::ReaderStream; use tokio_util::io::StreamReader; @@ -125,33 +127,51 @@ impl AsyncRead for HttpFileReader { pub struct EtagReader { inner: R, - md5: Md5, + bytes_tx: mpsc::Sender, + md5_rx: oneshot::Receiver, } impl EtagReader { pub fn new(inner: R) -> Self { - EtagReader { inner, md5: Md5::new() } + let (bytes_tx, mut bytes_rx) = mpsc::channel::(8); + let (md5_tx, md5_rx) = oneshot::channel::(); + + tokio::task::spawn_blocking(move || { + let mut md5 = Md5::new(); + while let Some(bytes) = bytes_rx.blocking_recv() { + md5.update(&bytes); + } + let digest = md5.finalize(); + let etag = hex_simd::encode_to_string(digest, hex_simd::AsciiCase::Lower); + let _ = md5_tx.send(etag); + }); + + EtagReader { inner, bytes_tx, md5_rx } } - pub fn etag(self) -> String { - hex_simd::encode_to_string(self.md5.finalize(), hex_simd::AsciiCase::Lower) + pub async fn etag(self) -> String { + drop(self.inner); + drop(self.bytes_tx); + self.md5_rx.await.unwrap() } } impl AsyncRead for EtagReader { + #[tracing::instrument(level = "debug", skip_all)] fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { - let befor_size = buf.filled().len(); - - match Pin::new(&mut self.inner).poll_read(cx, buf) { - Poll::Ready(Ok(())) => { - if buf.filled().len() > befor_size { - let bytes = &buf.filled()[befor_size..]; - self.md5.update(bytes); - } - - Poll::Ready(Ok(())) + let poll = Pin::new(&mut self.inner).poll_read(cx, buf); + if let Poll::Ready(Ok(())) = &poll { + if buf.remaining() == 0 { + let bytes = buf.filled(); + let bytes = Bytes::copy_from_slice(bytes); + let tx = self.bytes_tx.clone(); + tokio::spawn(async move { + if let Err(e) = tx.send(bytes).await { + warn!("EtagReader send error: {:?}", e); + } + }); } - other => other, } + poll } } diff --git a/ecstore/src/set_disk.rs b/ecstore/src/set_disk.rs index c64bc189b..19c244190 100644 --- a/ecstore/src/set_disk.rs +++ b/ecstore/src/set_disk.rs @@ -33,7 +33,7 @@ use crate::{ ListMultipartsInfo, ListObjectsV2Info, MakeBucketOptions, MultipartInfo, MultipartUploadResult, ObjectIO, ObjectInfo, ObjectOptions, ObjectPartInfo, ObjectToDelete, PartInfo, PutObjReader, RawFileInfo, StorageAPI, DEFAULT_BITROT_ALGO, }, - store_err::{to_object_err, StorageError}, + store_err::{is_err_object_not_found, to_object_err, StorageError}, store_init::{load_format_erasure, ErasureError}, utils::{ self, @@ -325,6 +325,7 @@ impl SetDisks { } } + let mut futures = Vec::with_capacity(disks.len()); if let Some(err) = reduce_write_quorum_errs(&errs, object_op_ignored_errs().as_ref(), write_quorum) { // TODO: 并发 for (i, err) in errs.iter().enumerate() { @@ -334,26 +335,30 @@ impl SetDisks { if let Some(disk) = disks[i].as_ref() { let fi = file_infos[i].clone(); - let _ = disk - .delete_version( - src_bucket, - src_object, - fi, - false, - DeleteOptions { - undo_write: true, - old_data_dir: data_dirs[i], - ..Default::default() - }, - ) - .await - .map_err(|e| { - debug!("rename_data delete_version err {:?}", e); - e - }); + let old_data_dir = data_dirs[i]; + futures.push(async move { + let _ = disk + .delete_version( + src_bucket, + src_object, + fi, + false, + DeleteOptions { + undo_write: true, + old_data_dir, + ..Default::default() + }, + ) + .await + .map_err(|e| { + debug!("rename_data delete_version err {:?}", e); + e + }); + }); } } + let _ = join_all(futures).await; return Err(err); } @@ -454,6 +459,7 @@ impl SetDisks { Ok(()) } + #[tracing::instrument(skip(disks))] async fn cleanup_multipart_path(disks: &[Option], paths: &[String]) { let mut futures = Vec::with_capacity(disks.len()); @@ -485,6 +491,8 @@ impl SetDisks { warn!("cleanup_multipart_path errs {:?}", &errs); } } + + #[tracing::instrument(skip(disks, meta))] async fn rename_part( disks: &[Option], src_bucket: &str, @@ -575,6 +583,7 @@ impl SetDisks { // errors // } + #[tracing::instrument(skip(disks, files))] async fn write_unique_file_info( disks: &[Option], org_bucket: &str, @@ -861,7 +870,6 @@ impl SetDisks { }; if let Some(err) = reduce_read_quorum_errs(errs, object_op_ignored_errs().as_ref(), expected_rquorum) { - warn!("object_quorum_from_meta err {:?}", &err); return Err(err); } @@ -874,7 +882,6 @@ impl SetDisks { let parity_blocks = Self::common_parity(&parities, default_parity_count as i32); if parity_blocks < 0 { - warn!("QuorumError::Read, common_parity < 0 "); return Err(Error::new(QuorumError::Read)); } @@ -928,7 +935,15 @@ impl SetDisks { let (parts_metadata, errs) = Self::read_all_fileinfo(&disks, bucket, RUSTFS_META_MULTIPART_BUCKET, &upload_id_path, "", false, false).await; - let (read_quorum, write_quorum) = Self::object_quorum_from_meta(&parts_metadata, &errs, self.default_parity_count)?; + let map_err_notfound = |err: Error| { + if is_err_object_not_found(&err) { + return Error::new(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned())); + } + err + }; + + let (read_quorum, write_quorum) = + Self::object_quorum_from_meta(&parts_metadata, &errs, self.default_parity_count).map_err(map_err_notfound)?; if read_quorum < 0 { return Err(Error::new(QuorumError::Read)); @@ -943,10 +958,10 @@ impl SetDisks { quorum = write_quorum as usize; if let Some(err) = reduce_write_quorum_errs(&errs, object_op_ignored_errs().as_ref(), quorum) { - return Err(err); + return Err(map_err_notfound(err)); } } else if let Some(err) = reduce_read_quorum_errs(&errs, object_op_ignored_errs().as_ref(), quorum) { - return Err(err); + return Err(map_err_notfound(err)); } let (_, mod_time, etag) = Self::list_online_disks(&disks, &parts_metadata, &errs, quorum); @@ -1746,7 +1761,7 @@ impl SetDisks { // TODO: 优化并发 可用数量中断 let (parts_metadata, errs) = Self::read_all_fileinfo(&disks, "", bucket, object, vid.as_str(), read_data, false).await; // warn!("get_object_fileinfo parts_metadata {:?}", &parts_metadata); - warn!("get_object_fileinfo {}/{} errs {:?}", bucket, object, &errs); + // warn!("get_object_fileinfo {}/{} errs {:?}", bucket, object, &errs); let _min_disks = self.set_drive_count - self.default_parity_count; @@ -1754,12 +1769,6 @@ impl SetDisks { .map_err(|err| to_object_err(err, vec![bucket, object]))?; if let Some(err) = reduce_read_quorum_errs(&errs, object_op_ignored_errs().as_ref(), read_quorum as usize) { - error!( - "reduce_read_quorum_errs disks: {} read_quorum {} \n {:?}", - disks.len(), - read_quorum, - &errs - ); return Err(to_object_err(err, vec![bucket, object])); } @@ -2587,7 +2596,6 @@ impl SetDisks { } } Err(err) => { - warn!("object_quorum_from_meta failed, err: {}", err.to_string()); let data_errs_by_part = HashMap::new(); match self .delete_if_dang_ling( @@ -3802,7 +3810,7 @@ impl ObjectIO for SetDisks { error!("close_bitrot_writers err {:?}", err); } - let etag = etag_stream.etag(); + let etag = etag_stream.etag().await; //TODO: userDefined user_defined.insert("etag".to_owned(), etag.clone()); @@ -3879,14 +3887,17 @@ impl ObjectIO for SetDisks { #[async_trait::async_trait] impl StorageAPI for SetDisks { + #[tracing::instrument(skip(self))] async fn backend_info(&self) -> madmin::BackendInfo { unimplemented!() } + #[tracing::instrument(skip(self))] async fn storage_info(&self) -> madmin::StorageInfo { let disks = self.get_disks_internal().await; get_storage_info(&disks, &self.set_endpoints).await } + #[tracing::instrument(skip(self))] async fn local_storage_info(&self) -> madmin::StorageInfo { let disks = self.get_disks_internal().await; @@ -3902,17 +3913,21 @@ impl StorageAPI for SetDisks { get_storage_info(&local_disks, &local_endpoints).await } + #[tracing::instrument(skip(self))] async fn list_bucket(&self, _opts: &BucketOptions) -> Result> { unimplemented!() } + #[tracing::instrument(skip(self))] async fn make_bucket(&self, _bucket: &str, _opts: &MakeBucketOptions) -> Result<()> { unimplemented!() } + #[tracing::instrument(skip(self))] async fn get_bucket_info(&self, _bucket: &str, _opts: &BucketOptions) -> Result { unimplemented!() } + #[tracing::instrument(skip(self))] async fn copy_object( &self, src_bucket: &str, @@ -4017,6 +4032,7 @@ impl StorageAPI for SetDisks { Ok(fi.to_object_info(src_bucket, src_object, src_opts.versioned || src_opts.version_suspended)) } + #[tracing::instrument(skip(self))] async fn delete_objects( &self, bucket: &str, @@ -4127,6 +4143,8 @@ impl StorageAPI for SetDisks { Ok((del_objects, del_errs)) } + + #[tracing::instrument(skip(self))] async fn delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result { if opts.delete_prefix { self.delete_prefix(bucket, object) @@ -4138,6 +4156,7 @@ impl StorageAPI for SetDisks { unimplemented!() } + #[tracing::instrument(skip(self))] async fn list_objects_v2( self: Arc, _bucket: &str, @@ -4150,6 +4169,8 @@ impl StorageAPI for SetDisks { ) -> Result { unimplemented!() } + + #[tracing::instrument(skip(self))] async fn list_object_versions( self: Arc, _bucket: &str, @@ -4161,6 +4182,8 @@ impl StorageAPI for SetDisks { ) -> Result { unimplemented!() } + + #[tracing::instrument(skip(self))] async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { // let mut _ns = None; // if !opts.no_lock { @@ -4202,6 +4225,7 @@ impl StorageAPI for SetDisks { Ok(oi) } + #[tracing::instrument(skip(self))] async fn put_object_metadata(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { // TODO: nslock @@ -4284,6 +4308,7 @@ impl StorageAPI for SetDisks { Ok(fi.to_object_info(bucket, object, opts.versioned || opts.version_suspended)) } + #[tracing::instrument(skip(self))] async fn get_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { let oi = self.get_object_info(bucket, object, opts).await?; Ok(oi.user_tags) @@ -4310,10 +4335,13 @@ impl StorageAPI for SetDisks { // TODO: versioned Ok(fi.to_object_info(bucket, object, opts.versioned || opts.version_suspended)) } + + #[tracing::instrument(skip(self))] async fn delete_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { self.put_object_tags(bucket, object, "", opts).await } + #[tracing::instrument(skip(self))] async fn copy_object_part( &self, _src_bucket: &str, @@ -4393,7 +4421,7 @@ impl StorageAPI for SetDisks { error!("close_bitrot_writers err {:?}", err); } - let mut etag = etag_stream.etag(); + let mut etag = etag_stream.etag().await; if let Some(ref tag) = opts.preserve_etag { etag = tag.clone(); @@ -4436,6 +4464,8 @@ impl StorageAPI for SetDisks { Ok(ret) } + + #[tracing::instrument(skip(self))] async fn list_multipart_uploads( &self, bucket: &str, @@ -4475,7 +4505,7 @@ impl StorageAPI for SetDisks { Err(err) => { if DiskError::DiskNotFound.is(&err) { None - } else if DiskError::FileNotFound.is(&err) { + } else if is_err_object_not_found(&err) { return Ok(ListMultipartsInfo { key_marker: key_marker.to_owned(), max_uploads, @@ -4579,9 +4609,9 @@ impl StorageAPI for SetDisks { ..Default::default() }) } - async fn new_multipart_upload(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { - warn!("new_multipart_upload opt {:?}", opts); + #[tracing::instrument(skip(self))] + async fn new_multipart_upload(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { let disks = self.disks.read().await; let disks = disks.clone(); @@ -4684,6 +4714,8 @@ impl StorageAPI for SetDisks { Ok(MultipartUploadResult { upload_id }) } + + #[tracing::instrument(skip(self))] async fn get_multipart_info( &self, bucket: &str, @@ -4705,6 +4737,8 @@ impl StorageAPI for SetDisks { ..Default::default() }) } + + #[tracing::instrument(skip(self))] async fn abort_multipart_upload(&self, bucket: &str, object: &str, upload_id: &str, _opts: &ObjectOptions) -> Result<()> { self.check_upload_id_exists(bucket, object, upload_id, false).await?; let upload_id_path = Self::get_upload_id_dir(bucket, object, upload_id); @@ -4712,7 +4746,7 @@ impl StorageAPI for SetDisks { self.delete_all(RUSTFS_META_MULTIPART_BUCKET, &upload_id_path).await } // complete_multipart_upload 完成 - // #[tracing::instrument(skip(self))] + #[tracing::instrument(skip(self))] async fn complete_multipart_upload( &self, bucket: &str, @@ -4961,24 +4995,32 @@ impl StorageAPI for SetDisks { Ok(fi.to_object_info(bucket, object, opts.versioned || opts.version_suspended)) } + #[tracing::instrument(skip(self))] async fn get_disks(&self, _pool_idx: usize, _set_idx: usize) -> Result>> { Ok(self.get_disks_internal().await) } + #[tracing::instrument(skip(self))] fn set_drive_counts(&self) -> Vec { unimplemented!() } + #[tracing::instrument(skip(self))] async fn delete_bucket(&self, _bucket: &str, _opts: &DeleteBucketOptions) -> Result<()> { unimplemented!() } + #[tracing::instrument(skip(self))] async fn heal_format(&self, _dry_run: bool) -> Result<(HealResultItem, Option)> { unimplemented!() } + + #[tracing::instrument(skip(self))] async fn heal_bucket(&self, _bucket: &str, _opts: &HealOpts) -> Result { unimplemented!() } + + #[tracing::instrument(skip(self))] async fn heal_object( &self, bucket: &str, @@ -5029,6 +5071,8 @@ impl StorageAPI for SetDisks { } return Ok((result, err)); } + + #[tracing::instrument(skip(self))] async fn heal_objects( &self, _bucket: &str, @@ -5039,9 +5083,13 @@ impl StorageAPI for SetDisks { ) -> Result<()> { unimplemented!() } + + #[tracing::instrument(skip(self))] async fn get_pool_and_set(&self, _id: &str) -> Result<(Option, Option, Option)> { unimplemented!() } + + #[tracing::instrument(skip(self))] async fn check_abandoned_parts(&self, _bucket: &str, _object: &str, _opts: &HealOpts) -> Result<()> { unimplemented!() } diff --git a/ecstore/src/sets.rs b/ecstore/src/sets.rs index 0a2e716fa..7a4cf93a3 100644 --- a/ecstore/src/sets.rs +++ b/ecstore/src/sets.rs @@ -308,9 +308,11 @@ impl ObjectIO for Sets { #[async_trait::async_trait] impl StorageAPI for Sets { + #[tracing::instrument(skip(self))] async fn backend_info(&self) -> madmin::BackendInfo { unimplemented!() } + #[tracing::instrument(skip(self))] async fn storage_info(&self) -> madmin::StorageInfo { let mut futures = Vec::with_capacity(self.disk_set.len()); @@ -331,6 +333,7 @@ impl StorageAPI for Sets { ..Default::default() } } + #[tracing::instrument(skip(self))] async fn local_storage_info(&self) -> madmin::StorageInfo { let mut futures = Vec::with_capacity(self.disk_set.len()); @@ -350,16 +353,21 @@ impl StorageAPI for Sets { ..Default::default() } } + #[tracing::instrument(skip(self))] async fn list_bucket(&self, _opts: &BucketOptions) -> Result> { unimplemented!() } + #[tracing::instrument(skip(self))] async fn make_bucket(&self, _bucket: &str, _opts: &MakeBucketOptions) -> Result<()> { unimplemented!() } + #[tracing::instrument(skip(self))] async fn get_bucket_info(&self, _bucket: &str, _opts: &BucketOptions) -> Result { unimplemented!() } + + #[tracing::instrument(skip(self))] async fn copy_object( &self, src_bucket: &str, @@ -416,6 +424,8 @@ impl StorageAPI for Sets { "put_object_reader2 is none".to_owned(), ))) } + + #[tracing::instrument(skip(self))] async fn delete_objects( &self, bucket: &str, @@ -498,6 +508,8 @@ impl StorageAPI for Sets { Ok((del_objects, del_errs)) } + + #[tracing::instrument(skip(self))] async fn delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result { if opts.delete_prefix && !opts.delete_prefix_object { self.delete_prefix(bucket, object).await?; @@ -506,6 +518,8 @@ impl StorageAPI for Sets { self.get_disks_by_key(object).delete_object(bucket, object, opts).await } + + #[tracing::instrument(skip(self))] async fn list_objects_v2( self: Arc, _bucket: &str, @@ -518,6 +532,8 @@ impl StorageAPI for Sets { ) -> Result { unimplemented!() } + + #[tracing::instrument(skip(self))] async fn list_object_versions( self: Arc, _bucket: &str, @@ -529,14 +545,18 @@ impl StorageAPI for Sets { ) -> Result { unimplemented!() } + + #[tracing::instrument(skip(self))] async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { self.get_disks_by_key(object).get_object_info(bucket, object, opts).await } + #[tracing::instrument(skip(self))] async fn put_object_metadata(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { self.get_disks_by_key(object).put_object_metadata(bucket, object, opts).await } + #[tracing::instrument(skip(self))] async fn get_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { self.get_disks_by_key(object).get_object_tags(bucket, object, opts).await } @@ -546,10 +566,13 @@ impl StorageAPI for Sets { .put_object_tags(bucket, object, tags, opts) .await } + + #[tracing::instrument(skip(self))] async fn delete_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { self.get_disks_by_key(object).delete_object_tags(bucket, object, opts).await } + #[tracing::instrument(skip(self))] async fn copy_object_part( &self, _src_bucket: &str, @@ -566,6 +589,8 @@ impl StorageAPI for Sets { ) -> Result<()> { unimplemented!() } + + #[tracing::instrument(skip(self))] async fn put_object_part( &self, bucket: &str, @@ -579,6 +604,8 @@ impl StorageAPI for Sets { .put_object_part(bucket, object, upload_id, part_id, data, opts) .await } + + #[tracing::instrument(skip(self))] async fn list_multipart_uploads( &self, bucket: &str, @@ -592,9 +619,13 @@ impl StorageAPI for Sets { .list_multipart_uploads(bucket, prefix, key_marker, upload_id_marker, delimiter, max_uploads) .await } + + #[tracing::instrument(skip(self))] async fn new_multipart_upload(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { self.get_disks_by_key(object).new_multipart_upload(bucket, object, opts).await } + + #[tracing::instrument(skip(self))] async fn get_multipart_info( &self, bucket: &str, @@ -606,11 +637,15 @@ impl StorageAPI for Sets { .get_multipart_info(bucket, object, upload_id, opts) .await } + + #[tracing::instrument(skip(self))] async fn abort_multipart_upload(&self, bucket: &str, object: &str, upload_id: &str, opts: &ObjectOptions) -> Result<()> { self.get_disks_by_key(object) .abort_multipart_upload(bucket, object, upload_id, opts) .await } + + #[tracing::instrument(skip(self))] async fn complete_multipart_upload( &self, bucket: &str, @@ -623,17 +658,23 @@ impl StorageAPI for Sets { .complete_multipart_upload(bucket, object, upload_id, uploaded_parts, opts) .await } + + #[tracing::instrument(skip(self))] async fn get_disks(&self, _pool_idx: usize, _set_idx: usize) -> Result>> { unimplemented!() } + #[tracing::instrument(skip(self))] fn set_drive_counts(&self) -> Vec { unimplemented!() } + #[tracing::instrument(skip(self))] async fn delete_bucket(&self, _bucket: &str, _opts: &DeleteBucketOptions) -> Result<()> { unimplemented!() } + + #[tracing::instrument(skip(self))] async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option)> { let (disks, _) = init_storage_disks_with_errors( &self.endpoints.endpoints, @@ -718,9 +759,11 @@ impl StorageAPI for Sets { } Ok((res, None)) } + #[tracing::instrument(skip(self))] async fn heal_bucket(&self, _bucket: &str, _opts: &HealOpts) -> Result { unimplemented!() } + #[tracing::instrument(skip(self))] async fn heal_object( &self, bucket: &str, @@ -732,6 +775,7 @@ impl StorageAPI for Sets { .heal_object(bucket, object, version_id, opts) .await } + #[tracing::instrument(skip(self))] async fn heal_objects( &self, _bucket: &str, @@ -742,9 +786,11 @@ impl StorageAPI for Sets { ) -> Result<()> { unimplemented!() } + #[tracing::instrument(skip(self))] async fn get_pool_and_set(&self, _id: &str) -> Result<(Option, Option, Option)> { unimplemented!() } + #[tracing::instrument(skip(self))] async fn check_abandoned_parts(&self, _bucket: &str, _object: &str, _opts: &HealOpts) -> Result<()> { unimplemented!() } diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs index 4a7e06ec1..5a2075775 100644 --- a/ecstore/src/store.rs +++ b/ecstore/src/store.rs @@ -223,7 +223,7 @@ impl ECStore { disk_map, pools, peer_sys, - pool_meta: pool_meta.into(), + pool_meta: RwLock::new(pool_meta), rebalance_meta: RwLock::new(None), decommission_cancelers, }); @@ -268,7 +268,7 @@ impl ECStore { meta.load(self.pools[0].clone(), self.pools.clone()).await?; let update = meta.validate(self.pools.clone())?; - if update { + if !update { { let mut pool_meta = self.pool_meta.write().await; *pool_meta = meta.clone(); @@ -1246,6 +1246,7 @@ lazy_static! { #[async_trait::async_trait] impl StorageAPI for ECStore { + #[tracing::instrument(skip(self))] async fn backend_info(&self) -> madmin::BackendInfo { let (standard_sc_parity, rr_sc_parity) = { if let Some(sc) = GLOBAL_StorageClass.get() { @@ -1290,6 +1291,7 @@ impl StorageAPI for ECStore { ..Default::default() } } + #[tracing::instrument(skip(self))] async fn storage_info(&self) -> madmin::StorageInfo { let Some(notification_sy) = get_global_notification_sys() else { return madmin::StorageInfo::default(); @@ -1297,6 +1299,7 @@ impl StorageAPI for ECStore { notification_sy.storage_info(self).await } + #[tracing::instrument(skip(self))] async fn local_storage_info(&self) -> madmin::StorageInfo { let mut futures = Vec::with_capacity(self.pools.len()); @@ -1316,6 +1319,7 @@ impl StorageAPI for ECStore { madmin::StorageInfo { backend, disks } } + #[tracing::instrument(skip(self))] async fn list_bucket(&self, opts: &BucketOptions) -> Result> { // TODO: opts.cached @@ -1331,6 +1335,7 @@ impl StorageAPI for ECStore { Ok(buckets) } + #[tracing::instrument(skip(self))] async fn delete_bucket(&self, bucket: &str, opts: &DeleteBucketOptions) -> Result<()> { if is_meta_bucketname(bucket) { return Err(StorageError::BucketNameInvalid(bucket.to_string()).into()); @@ -1360,6 +1365,7 @@ impl StorageAPI for ECStore { .await?; Ok(()) } + #[tracing::instrument(skip(self))] async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> { if !is_meta_bucketname(bucket) { if let Err(err) = check_valid_bucket_name_strict(bucket) { @@ -1403,6 +1409,7 @@ impl StorageAPI for ECStore { Ok(()) } + #[tracing::instrument(skip(self))] async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result { let mut info = self .peer_sys @@ -1420,6 +1427,7 @@ impl StorageAPI for ECStore { } // TODO: review + #[tracing::instrument(skip(self))] async fn copy_object( &self, src_bucket: &str, @@ -1490,6 +1498,7 @@ impl StorageAPI for ECStore { } // TODO: review + #[tracing::instrument(skip(self))] async fn delete_objects( &self, bucket: &str, @@ -1643,6 +1652,7 @@ impl StorageAPI for ECStore { Ok((del_objects, del_errs)) } + #[tracing::instrument(skip(self))] async fn delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result { check_del_obj_args(bucket, object)?; @@ -1717,6 +1727,7 @@ impl StorageAPI for ECStore { // @start_after as marker when continuation_token empty // @delimiter default="/", empty when recursive // @max_keys limit + #[tracing::instrument(skip(self))] async fn list_objects_v2( self: Arc, bucket: &str, @@ -1730,6 +1741,7 @@ impl StorageAPI for ECStore { self.inner_list_objects_v2(bucket, prefix, continuation_token, delimiter, max_keys, fetch_owner, start_after) .await } + #[tracing::instrument(skip(self))] async fn list_object_versions( self: Arc, bucket: &str, @@ -1742,6 +1754,7 @@ impl StorageAPI for ECStore { self.inner_list_object_versions(bucket, prefix, marker, version_marker, delimiter, max_keys) .await } + #[tracing::instrument(skip(self))] async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { check_object_args(bucket, object)?; @@ -1758,6 +1771,7 @@ impl StorageAPI for ECStore { Ok(info) } + #[tracing::instrument(skip(self))] async fn get_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { let object = encode_dir_object(object); @@ -1770,6 +1784,7 @@ impl StorageAPI for ECStore { Ok(oi.user_tags) } + #[tracing::instrument(skip(self))] async fn put_object_metadata(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { let object = encode_dir_object(object); if self.single_pool() { @@ -1796,6 +1811,7 @@ impl StorageAPI for ECStore { self.pools[idx].put_object_tags(bucket, object.as_str(), tags, opts).await } + #[tracing::instrument(skip(self))] async fn delete_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { let object = encode_dir_object(object); @@ -1807,6 +1823,8 @@ impl StorageAPI for ECStore { self.pools[idx].delete_object_tags(bucket, object.as_str(), opts).await } + + #[tracing::instrument(skip(self))] async fn copy_object_part( &self, src_bucket: &str, @@ -1828,6 +1846,7 @@ impl StorageAPI for ECStore { unimplemented!() } + #[tracing::instrument(skip(self, data))] async fn put_object_part( &self, bucket: &str, @@ -1859,6 +1878,7 @@ impl StorageAPI for ECStore { }; if let Some(err) = err { + error!("put_object_part err: {:?}", err); return Err(err); } } @@ -1870,6 +1890,7 @@ impl StorageAPI for ECStore { ))) } + #[tracing::instrument(skip(self))] async fn list_multipart_uploads( &self, bucket: &str, @@ -1917,6 +1938,8 @@ impl StorageAPI for ECStore { ..Default::default() }) } + + #[tracing::instrument(skip(self))] async fn new_multipart_upload(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { check_new_multipart_args(bucket, object)?; @@ -1946,6 +1969,8 @@ impl StorageAPI for ECStore { self.pools[idx].new_multipart_upload(bucket, object, opts).await } + + #[tracing::instrument(skip(self))] async fn get_multipart_info( &self, bucket: &str, @@ -1981,6 +2006,7 @@ impl StorageAPI for ECStore { upload_id.to_owned(), ))) } + #[tracing::instrument(skip(self))] async fn abort_multipart_upload(&self, bucket: &str, object: &str, upload_id: &str, opts: &ObjectOptions) -> Result<()> { check_abort_multipart_args(bucket, object, upload_id)?; @@ -2016,6 +2042,7 @@ impl StorageAPI for ECStore { upload_id.to_owned(), ))) } + #[tracing::instrument(skip(self))] async fn complete_multipart_upload( &self, bucket: &str, @@ -2062,6 +2089,7 @@ impl StorageAPI for ECStore { ))) } + #[tracing::instrument(skip(self))] async fn get_disks(&self, pool_idx: usize, set_idx: usize) -> Result>> { if pool_idx < self.pools.len() && set_idx < self.pools[pool_idx].disk_set.len() { self.pools[pool_idx].disk_set[set_idx].get_disks(0, 0).await @@ -2070,6 +2098,7 @@ impl StorageAPI for ECStore { } } + #[tracing::instrument(skip(self))] fn set_drive_counts(&self) -> Vec { let mut counts = vec![0; self.pools.len()]; @@ -2078,6 +2107,8 @@ impl StorageAPI for ECStore { } counts } + + #[tracing::instrument(skip(self))] async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option)> { info!("heal_format"); let mut r = HealResultItem { @@ -2112,9 +2143,11 @@ impl StorageAPI for ECStore { Ok((r, None)) } + #[tracing::instrument(skip(self))] async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result { self.peer_sys.heal_bucket(bucket, opts).await } + #[tracing::instrument(skip(self))] async fn heal_object( &self, bucket: &str, @@ -2188,6 +2221,8 @@ impl StorageAPI for ECStore { Ok((HealResultItem::default(), Some(Error::new(DiskError::FileNotFound)))) } + + #[tracing::instrument(skip(self))] async fn heal_objects( &self, bucket: &str, @@ -2299,6 +2334,7 @@ impl StorageAPI for ECStore { Ok(()) } + #[tracing::instrument(skip(self))] async fn get_pool_and_set(&self, id: &str) -> Result<(Option, Option, Option)> { for (pool_idx, pool) in self.pools.iter().enumerate() { for (set_idx, set) in pool.format.erasure.sets.iter().enumerate() { @@ -2313,6 +2349,7 @@ impl StorageAPI for ECStore { Err(Error::new(DiskError::DiskNotFound)) } + #[tracing::instrument(skip(self))] async fn check_abandoned_parts(&self, bucket: &str, object: &str, opts: &HealOpts) -> Result<()> { let object = utils::path::encode_dir_object(object); if self.single_pool() { diff --git a/madmin/src/info_commands.rs b/madmin/src/info_commands.rs index 72c5972f2..53afc8c13 100644 --- a/madmin/src/info_commands.rs +++ b/madmin/src/info_commands.rs @@ -45,7 +45,9 @@ pub struct DiskMetrics { #[derive(Serialize, Deserialize, Debug, Default, Clone)] pub struct Disk { pub endpoint: String, + #[serde(rename = "rootDisk")] pub root_disk: bool, + #[serde(rename = "path")] pub drive_path: String, pub healing: bool, pub scanning: bool, @@ -54,12 +56,19 @@ pub struct Disk { pub major: u32, pub minor: u32, pub model: Option, + #[serde(rename = "totalspace")] pub total_space: u64, + #[serde(rename = "usedspace")] pub used_space: u64, + #[serde(rename = "availspace")] pub available_space: u64, + #[serde(rename = "readthroughput")] pub read_throughput: f64, + #[serde(rename = "writethroughput")] pub write_throughput: f64, + #[serde(rename = "readlatency")] pub read_latency: f64, + #[serde(rename = "writelatency")] pub write_latency: f64, pub utilization: f64, pub metrics: Option, diff --git a/rustfs/src/admin/handlers/rebalance.rs b/rustfs/src/admin/handlers/rebalance.rs index 65b5277fc..fefd7df42 100644 --- a/rustfs/src/admin/handlers/rebalance.rs +++ b/rustfs/src/admin/handlers/rebalance.rs @@ -1,4 +1,5 @@ use ecstore::{ + config::error::is_err_config_not_found, new_object_layer_fn, notification_sys::get_global_notification_sys, rebalance::{DiskStat, RebalSaveOpt}, @@ -128,9 +129,13 @@ impl Operation for RebalanceStatus { }; let mut meta = RebalanceMeta::new(); - meta.load(store.pools[0].clone()) - .await - .map_err(|e| s3_error!(InternalError, "Failed to load rebalance meta: {}", e))?; + if let Err(err) = meta.load(store.pools[0].clone()).await { + if is_err_config_not_found(&err) { + return Err(s3_error!(NoSuchResource, "Pool rebalance is not started")); + } + + return Err(s3_error!(InternalError, "Failed to load rebalance meta: {}", err)); + } // Compute disk usage percentage let si = store.storage_info().await; diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index ba1579abd..ca73521f8 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -335,20 +335,27 @@ async fn run(opt: config::Opt) -> Result<()> { span }) .on_request(|request: &HttpRequest<_>, _span: &Span| { - debug!("started method: {}, url path: {}", request.method(), request.uri().path()) + info!( + counter.rustfs_api_requests_total = 1_u64, + key_request_method = %request.method().to_string(), + key_request_uri_path = %request.uri().path().to_owned(), + "handle request api total", + ); + debug!("http started method: {}, url path: {}", request.method(), request.uri().path()) }) .on_response(|response: &Response<_>, latency: Duration, _span: &Span| { - _span.record("status_code", tracing::field::display(response.status())); - debug!("response generated in {:?}", latency) + _span.record("http response status_code", tracing::field::display(response.status())); + debug!("http response generated in {:?}", latency) }) .on_body_chunk(|chunk: &Bytes, latency: Duration, _span: &Span| { - debug!("sending {} bytes in {:?}", chunk.len(), latency) + info!(histogram.request.body.len = chunk.len(), "histogram request body lenght",); + debug!("http body sending {} bytes in {:?}", chunk.len(), latency) }) .on_eos(|_trailers: Option<&HeaderMap>, stream_duration: Duration, _span: &Span| { - debug!("stream closed after {:?}", stream_duration) + debug!("http stream closed after {:?}", stream_duration) }) .on_failure(|_error, latency: Duration, _span: &Span| { - debug!("request error: {:?} in {:?}", _error, latency) + debug!("http request failure error: {:?} in {:?}", _error, latency) }), ) .layer(CorsLayer::permissive()) diff --git a/scripts/dev.sh b/scripts/dev.sh index 8dc19c88b..0df907a84 100755 --- a/scripts/dev.sh +++ b/scripts/dev.sh @@ -15,13 +15,6 @@ REMOTE_PATH="~" # 格式:服务器IP 用户名 目标路径 SERVER_LIST=( "root@121.89.80.13" - "root@121.89.80.198" - "root@8.130.78.237" - "root@8.130.189.236" - "root@121.89.80.230" - "root@121.89.80.45" - "root@8.130.191.95" - "root@121.89.80.91" ) # 遍历服务器列表