From 9590d99e7c19f5ac584299c2f103fab9ce6d5497 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 26 Apr 2025 22:36:38 +0800 Subject: [PATCH 1/9] Feature/upgrade obs docker (#364) * upgrade docker config * upgrade obs.toml * modify dockerfile image from alpine to ubuntu --- .docker/observability/config/obs.toml | 4 ++-- Dockerfile.obs | 8 +++++-- crates/obs/src/config.rs | 29 ++++++++++++++++++-------- docker-compose-obs.yaml | 30 +++++++++++++-------------- scripts/run.sh | 1 + 5 files changed, 44 insertions(+), 28 deletions(-) diff --git a/.docker/observability/config/obs.toml b/.docker/observability/config/obs.toml index ecb340061..3b26644cf 100644 --- a/.docker/observability/config/obs.toml +++ b/.docker/observability/config/obs.toml @@ -1,12 +1,12 @@ [observability] -endpoint = "http://localhost:4317" # Default is "http://localhost:4317" if not specified +endpoint = "http://otel-collector:4317" # Default is "http://localhost:4317" if not specified use_stdout = false # Output with stdout, true output, false no output sample_ratio = 2.0 meter_interval = 30 service_name = "rustfs" service_version = "0.1.0" environments = "production" -logger_level = "info" +logger_level = "debug" [sinks] [sinks.kafka] # Kafka sink is disabled by default diff --git a/Dockerfile.obs b/Dockerfile.obs index fdcfcc3fe..f16ea3dcc 100644 --- a/Dockerfile.obs +++ b/Dockerfile.obs @@ -1,4 +1,4 @@ -FROM alpine:latest +FROM ubuntu:latest # RUN apk add --no-cache # 如果 rustfs 有依赖，可以在这里添加，例如： @@ -7,7 +7,11 @@ FROM alpine:latest WORKDIR /app -COPY ./target/x86_64-unknown-linux-musl/release/rustfs /app/rustfs +# 创建与 RUSTFS_VOLUMES 一致的目录 +RUN mkdir -p /root/data/target/volume/test1 /root/data/target/volume/test2 /root/data/target/volume/test3 /root/data/target/volume/test4 + +# COPY ./target/x86_64-unknown-linux-musl/release/rustfs /app/rustfs +COPY ./target/x86_64-unknown-linux-gnu/release/rustfs /app/rustfs RUN chmod +x /app/rustfs diff --git a/crates/obs/src/config.rs b/crates/obs/src/config.rs index 33f974f97..737d2c0d2 100644 --- a/crates/obs/src/config.rs +++ b/crates/obs/src/config.rs @@ -10,19 +10,21 @@ use std::env; /// Add endpoint for metric collection /// Add use_stdout for output to stdout /// Add logger level for log level +/// Add local_logging_enabled for local logging enabled #[derive(Debug, Deserialize, Clone)] pub struct OtelConfig { - pub endpoint: String, - pub use_stdout: Option, - pub sample_ratio: Option, - pub meter_interval: Option, - pub service_name: Option, - pub service_version: Option, - pub environment: Option, - pub logger_level: Option, + pub endpoint: String, // Endpoint for metric collection + pub use_stdout: Option, // Output to stdout + pub sample_ratio: Option, // Trace sampling ratio + pub meter_interval: Option, // Metric collection interval + pub service_name: Option, // Service name + pub service_version: Option, // Service version + pub environment: Option, // Environment + pub logger_level: Option, // Logger level + pub local_logging_enabled: Option, // Local logging enabled } -// 辅助函数：从环境变量中提取可观测性配置 +// Helper function: Extract observable configuration from environment variables fn extract_otel_config_from_env() -> OtelConfig { OtelConfig { endpoint: env::var("RUSTFS_OBSERVABILITY_ENDPOINT").unwrap_or_else(|_| "".to_string()), @@ -54,6 +56,10 @@ fn extract_otel_config_from_env() -> OtelConfig { .ok() .and_then(|v| v.parse().ok()) .or(Some(LOGGER_LEVEL.to_string())), + local_logging_enabled: env::var("RUSTFS_OBSERVABILITY_LOCAL_LOGGING_ENABLED") + .ok() + .and_then(|v| v.parse().ok()) + .or(Some(false)), } } @@ -179,6 +185,10 @@ pub struct AppConfig { } impl AppConfig { + /// Create a new instance of AppConfig with default values + /// + /// # Returns + /// A new instance of AppConfig pub fn new() -> Self { Self { observability: OtelConfig::default(), @@ -195,6 +205,7 @@ impl Default for AppConfig { } } +/// Default configuration file name const DEFAULT_CONFIG_FILE: &str = "obs"; /// Loading the configuration file diff --git a/docker-compose-obs.yaml b/docker-compose-obs.yaml index 9a01db4cb..505f287b5 100644 --- a/docker-compose-obs.yaml +++ b/docker-compose-obs.yaml @@ -62,18 +62,18 @@ services: dockerfile: Dockerfile.obs container_name: node1 environment: - - RUSTFS_VOLUMES=/root/data/target/volume/test{1...4} + - RUSTFS_VOLUMES=http://node{1...4}:9000/root/data/target/volume/test{1...4} - RUSTFS_ADDRESS=0.0.0.0:9000 - RUSTFS_CONSOLE_ENABLE=true - RUSTFS_CONSOLE_ADDRESS=0.0.0.0:9002 - - RUSTFS_OBS_CONFIG=/etc/observability/config.obs.toml + - RUSTFS_OBS_CONFIG=/etc/observability/config/obs.toml platform: linux/amd64 ports: - "9001:9000" # 映射宿主机的 9001 端口到容器的 9000 端口 - "9101:9002" volumes: - - ./data:/root/data # 将当前路径挂载到容器内的 /root/data - - ./.docker/observability/config/obs.toml:/etc/observability/config.obs.toml + # - ./data:/root/data # 将当前路径挂载到容器内的 /root/data + - ./.docker/observability/config:/etc/observability/config networks: - rustfs-network @@ -87,14 +87,14 @@ services: - RUSTFS_ADDRESS=0.0.0.0:9000 - RUSTFS_CONSOLE_ENABLE=true - RUSTFS_CONSOLE_ADDRESS=0.0.0.0:9002 - - RUSTFS_OBS_CONFIG=/etc/observability/config.obs.toml + - RUSTFS_OBS_CONFIG=/etc/observability/config/obs.toml platform: linux/amd64 ports: - "9002:9000" # 映射宿主机的 9002 端口到容器的 9000 端口 - "9102:9002" volumes: - - ./data:/root/data - - ./.docker/observability/config/obs.toml:/etc/observability/config.obs.toml + # - ./data:/root/data + - ./.docker/observability/config:/etc/observability/config networks: - rustfs-network @@ -104,18 +104,18 @@ services: dockerfile: Dockerfile.obs container_name: node3 environment: - - RUSTFS_VOLUMES=/root/data/target/volume/test{1...4} + - RUSTFS_VOLUMES=http://node{1...4}:9000/root/data/target/volume/test{1...4} - RUSTFS_ADDRESS=0.0.0.0:9000 - RUSTFS_CONSOLE_ENABLE=true - RUSTFS_CONSOLE_ADDRESS=0.0.0.0:9002 - - RUSTFS_OBS_CONFIG=/etc/observability/config.obs.toml + - RUSTFS_OBS_CONFIG=/etc/observability/config/obs.toml platform: linux/amd64 ports: - "9003:9000" # 映射宿主机的 9003 端口到容器的 9000 端口 - "9103:9002" volumes: - - ./data:/root/data - - ./.docker/observability/config/obs.toml:/etc/observability/config.obs.toml + # - ./data:/root/data + - ./.docker/observability/config:/etc/observability/config networks: - rustfs-network @@ -125,18 +125,18 @@ services: dockerfile: Dockerfile.obs container_name: node4 environment: - - RUSTFS_VOLUMES=/root/data/target/volume/test{1...4} + - RUSTFS_VOLUMES=http://node{1...4}:9000/root/data/target/volume/test{1...4} - RUSTFS_ADDRESS=0.0.0.0:9000 - RUSTFS_CONSOLE_ENABLE=true - RUSTFS_CONSOLE_ADDRESS=0.0.0.0:9002 - - RUSTFS_OBS_CONFIG=/etc/observability/config.obs.toml + - RUSTFS_OBS_CONFIG=/etc/observability/config/obs.toml platform: linux/amd64 ports: - "9004:9000" # 映射宿主机的 9004 端口到容器的 9000 端口 - "9104:9002" volumes: - - ./data:/root/data - - ./.docker/observability/config/obs.toml:/etc/observability/config.obs.toml + # - ./data:/root/data + - ./.docker/observability/config:/etc/observability/config networks: - rustfs-network diff --git a/scripts/run.sh b/scripts/run.sh index c31ce7ed7..1d1470d93 100755 --- a/scripts/run.sh +++ b/scripts/run.sh @@ -47,6 +47,7 @@ export RUSTFS__OBSERVABILITY__SERVICE_NAME=rustfs export RUSTFS__OBSERVABILITY__SERVICE_VERSION=0.1.0 export RUSTFS__OBSERVABILITY__ENVIRONMENT=develop export RUSTFS__OBSERVABILITY__LOGGER_LEVEL=debug +export RUSTFS__OBSERVABILITY__LOCAL_LOGGER_ENABLED=true export RUSTFS__SINKS__FILE__ENABLED=true export RUSTFS__SINKS__FILE__PATH="./deploy/logs/rustfs.log" export RUSTFS__SINKS__WEBHOOK__ENABLED=false From 87bec2e6a6b53e2bf0b48baec1d4f48194a17b64 Mon Sep 17 00:00:00 2001 From: weisd Date: Sat, 26 Apr 2025 17:55:30 +0800 Subject: [PATCH 2/9] fix rebalance/decom --- ecstore/src/bitrot.rs | 5 +- ecstore/src/cache_value/metacache_set.rs | 7 +- ecstore/src/disk/local.rs | 2 +- ecstore/src/heal/data_usage_cache.rs | 3 +- ecstore/src/notification_sys.rs | 4 +- ecstore/src/peer_rest_client.rs | 2 + ecstore/src/pools.rs | 91 +++++++++++++++++++++--- ecstore/src/rebalance.rs | 70 ++++++++++++++---- ecstore/src/set_disk.rs | 2 +- rustfs/src/admin/handlers/rebalance.rs | 41 ++++++----- rustfs/src/grpc.rs | 12 +++- 11 files changed, 192 insertions(+), 47 deletions(-) diff --git a/ecstore/src/bitrot.rs b/ecstore/src/bitrot.rs index 68ccff71e..b64c84b3b 100644 --- a/ecstore/src/bitrot.rs +++ b/ecstore/src/bitrot.rs @@ -661,7 +661,10 @@ impl ReadAt for BitrotFileReader { } if offset != self.curr_offset { - error!("BitrotFileReader read_at offset != self.curr_offset, {} != {}", offset, self.curr_offset); + error!( + "BitrotFileReader read_at {}/{} offset != self.curr_offset, {} != {}", + &self.volume, &self.file_path, offset, self.curr_offset + ); return Err(Error::new(DiskError::Unexpected)); } diff --git a/ecstore/src/cache_value/metacache_set.rs b/ecstore/src/cache_value/metacache_set.rs index 48174d64f..e716a3051 100644 --- a/ecstore/src/cache_value/metacache_set.rs +++ b/ecstore/src/cache_value/metacache_set.rs @@ -294,7 +294,12 @@ pub async fn list_path_raw(mut rx: B_Receiver, opts: ListPathRawOptions) - jobs.push(revjob); - let _ = join_all(jobs).await; + let results = join_all(jobs).await; + for result in results { + if let Err(err) = result { + error!("list_path_raw err {:?}", err); + } + } Ok(()) } diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 74cdaa690..45abdbf5f 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -455,7 +455,7 @@ impl LocalDisk { { if let Err(aerr) = access(volume_dir.as_ref()).await { if os_is_not_exist(&aerr) { - warn!("read_metadata_with_dmtime os err {:?}", &aerr); + // warn!("read_metadata_with_dmtime os err {:?}", &aerr); return Err(Error::new(DiskError::VolumeNotFound)); } } diff --git a/ecstore/src/heal/data_usage_cache.rs b/ecstore/src/heal/data_usage_cache.rs index 2dad9f55b..50a570f5e 100644 --- a/ecstore/src/heal/data_usage_cache.rs +++ b/ecstore/src/heal/data_usage_cache.rs @@ -18,7 +18,6 @@ use std::path::Path; use std::time::{Duration, SystemTime}; use tokio::sync::mpsc::Sender; 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}; @@ -402,7 +401,7 @@ impl DataUsageCache { break; } Err(err) => { - warn!("Failed to load data usage cache from backend: {}", &err); + // warn!("Failed to load data usage cache from backend: {}", &err); match err.downcast_ref::() { Some(DiskError::FileNotFound) | Some(DiskError::VolumeNotFound) => { match store diff --git a/ecstore/src/notification_sys.rs b/ecstore/src/notification_sys.rs index d364cc281..ed4649069 100644 --- a/ecstore/src/notification_sys.rs +++ b/ecstore/src/notification_sys.rs @@ -9,7 +9,7 @@ use lazy_static::lazy_static; use madmin::{ItemState, ServerProperties}; use std::sync::OnceLock; use std::time::SystemTime; -use tracing::error; +use tracing::{error, warn}; lazy_static! { pub static ref GLOBAL_NotificationSys: OnceLock = OnceLock::new(); @@ -151,6 +151,8 @@ impl NotificationSys { for result in results { if let Err(err) = result { error!("notification load_rebalance_meta err {:?}", err); + } else { + warn!("notification load_rebalance_meta success"); } } } diff --git a/ecstore/src/peer_rest_client.rs b/ecstore/src/peer_rest_client.rs index 71f2e3f27..9a121c9b3 100644 --- a/ecstore/src/peer_rest_client.rs +++ b/ecstore/src/peer_rest_client.rs @@ -663,6 +663,8 @@ impl PeerRestClient { let request = Request::new(LoadRebalanceMetaRequest { start_rebalance }); let response = client.load_rebalance_meta(request).await?.into_inner(); + + warn!("load_rebalance_meta response {:?}", response); if !response.success { if let Some(msg) = response.error_info { return Err(Error::msg(msg)); diff --git a/ecstore/src/pools.rs b/ecstore/src/pools.rs index 3e868f19b..f6f996941 100644 --- a/ecstore/src/pools.rs +++ b/ecstore/src/pools.rs @@ -2,6 +2,7 @@ use crate::bucket::versioning_sys::BucketVersioningSys; use crate::cache_value::metacache_set::{list_path_raw, ListPathRawOptions}; use crate::config::com::{read_config, save_config, CONFIG_PREFIX}; use crate::config::error::ConfigError; +use crate::disk::error::is_err_volume_not_found; use crate::disk::{MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, BUCKET_META_PREFIX, RUSTFS_META_BUCKET}; use crate::heal::data_usage::DATA_USAGE_CACHE_NAME; use crate::heal::heal_commands::HealOpts; @@ -609,6 +610,12 @@ impl ECStore { let mut lock = self.pool_meta.write().await; if lock.decommission_cancel(idx) { lock.save(self.pools.clone()).await?; + + drop(lock); + + if let Some(notification_sys) = get_global_notification_sys() { + notification_sys.reload_pool_meta().await; + } } Ok(()) @@ -628,6 +635,7 @@ impl ECStore { #[tracing::instrument(skip(self, rx))] pub async fn decommission(&self, rx: B_Receiver, indices: Vec) -> Result<()> { + warn!("decommission: {:?}", indices); if indices.is_empty() { return Err(Error::msg("errInvalidArgument")); } @@ -662,8 +670,10 @@ impl ECStore { wk: Arc, rcfg: Option, ) { + warn!("decommission_entry: {} {}", &bucket, &entry.name); wk.give().await; if entry.is_dir() { + warn!("decommission_entry: skip dir {}", &entry.name); return; } @@ -782,6 +792,9 @@ impl ECStore { } }; + let bucket_name = bucket.clone(); + let object_name = rd.object_info.name.clone(); + if let Err(err) = self.clone().decommission_object(idx, bucket, rd).await { if is_err_object_not_found(&err) || is_err_version_not_found(&err) || is_err_data_movement_overwrite(&err) { ignore = true; @@ -794,6 +807,11 @@ impl ECStore { continue; } + warn!( + "decommission_pool: decommission_object done {}/{} {}", + &bucket_name, &object_name, &version.name + ); + failure = false; break; } @@ -841,10 +859,16 @@ impl ECStore { .update_after(idx, self.pools.clone(), Duration::seconds(30)) .await .unwrap_or_default(); + + drop(pool_meta); if ok { - // TODO: ReloadPoolMeta + if let Some(notification_sys) = get_global_notification_sys() { + notification_sys.reload_pool_meta().await; + } } } + + warn!("decommission_pool: decommission_entry done {} {}", &bucket, &entry.name); } #[tracing::instrument(skip(self, rx))] @@ -872,6 +896,8 @@ impl ECStore { for (set_idx, set) in pool.disk_set.iter().enumerate() { wk.clone().take().await; + warn!("decommission_pool: decommission_pool {} {}", set_idx, &bi.name); + let decommission_entry: ListCallback = Arc::new({ let this = Arc::clone(self); let bucket = bi.name.clone(); @@ -895,20 +921,39 @@ impl ECStore { tokio::spawn(async move { loop { if rx.try_recv().is_ok() { + warn!("decommission_pool: cancel {}", set_id); break; } - if let Err(err) = set + warn!("decommission_pool: list_objects_to_decommission {} {}", set_id, &bi.name); + + match set .list_objects_to_decommission(rx.resubscribe(), bi.clone(), decommission_entry.clone()) .await { - error!("decommission_pool: list_objects_to_decommission {} err {:?}", set_id, &err); + Ok(_) => { + warn!("decommission_pool: list_objects_to_decommission {} done", set_id); + break; + } + Err(err) => { + error!("decommission_pool: list_objects_to_decommission {} err {:?}", set_id, &err); + if is_err_volume_not_found(&err) { + warn!("decommission_pool: list_objects_to_decommission {} volume not found", set_id); + break; + } + + tokio::time::sleep(tokio::time::Duration::from_secs(5)).await; + } } } }); } + warn!("decommission_pool: decommission_pool wait {} {}", idx, &bi.name); + wk.wait().await; + warn!("decommission_pool: decommission_pool done {} {}", idx, &bi.name); + Ok(()) } @@ -918,11 +963,15 @@ impl ECStore { error!("decom err {:?}", &err); if let Err(er) = self.decommission_failed(idx).await { error!("decom failed err {:?}", &er); + } else { + warn!("decommission: decommission_failed {}", idx); } return; } + warn!("decommission: decommission_in_background complete {}", idx); + let (failed, cmd_line) = { let pool_meta = self.pool_meta.read().await; let failed = { @@ -944,6 +993,8 @@ impl ECStore { } else if let Err(er) = self.complete_decommission(idx).await { error!("decom complete err {:?}", &er); } + + warn!("Decommissioning complete for pool {}", cmd_line); } #[tracing::instrument(skip(self))] @@ -955,6 +1006,9 @@ impl ECStore { let mut pool_meta = self.pool_meta.write().await; if pool_meta.decommission_failed(idx) { pool_meta.save(self.pools.clone()).await?; + + drop(pool_meta); + if let Some(notification_sys) = get_global_notification_sys() { notification_sys.reload_pool_meta().await; } @@ -972,6 +1026,7 @@ impl ECStore { let mut pool_meta = self.pool_meta.write().await; if pool_meta.decommission_complete(idx) { pool_meta.save(self.pools.clone()).await?; + drop(pool_meta); if let Some(notification_sys) = get_global_notification_sys() { notification_sys.reload_pool_meta().await; } @@ -996,7 +1051,7 @@ impl ECStore { }; if is_decommissioned { - info!("decommission: already done, moving on {}", bucket.to_string()); + warn!("decommission: already done, moving on {}", bucket.to_string()); { let mut pool_meta = self.pool_meta.write().await; @@ -1009,7 +1064,7 @@ impl ECStore { continue; } - info!("decommission: currently on bucket {}", &bucket.name); + warn!("decommission: currently on bucket {}", &bucket.name); if let Err(err) = self .decommission_pool(rx.resubscribe(), idx, pool.clone(), bucket.clone()) @@ -1017,6 +1072,8 @@ impl ECStore { { error!("decommission: decommission_pool err {:?}", &err); return Err(err); + } else { + warn!("decommission: decommission_pool done {}", &bucket.name); } { @@ -1026,6 +1083,8 @@ impl ECStore { error!("decom pool_meta.save err {:?}", err); } } + + warn!("decommission: decommission_pool bucket_done {}", &bucket.name); } } @@ -1107,6 +1166,7 @@ impl ECStore { #[tracing::instrument(skip(self, rd))] async fn decommission_object(self: Arc, pool_idx: usize, bucket: String, rd: GetObjectReader) -> Result<()> { + warn!("decommission_object: {} {}", &bucket, &rd.object_info.name); let object_info = rd.object_info.clone(); // TODO: check : use size or actual_size ? @@ -1269,10 +1329,23 @@ impl SetDisks { min_disks: listing_quorum, partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option]| { let resolver = resolver.clone(); - if let Ok(Some(entry)) = entries.resolve(resolver) { - cb_func(entry) - } else { - Box::pin(async {}) + let cb_func = cb_func.clone(); + + match entries.resolve(resolver) { + Ok(Some(entry)) => { + warn!("decommission_pool: list_objects_to_decommission get {}", &entry.name); + Box::pin(async move { + cb_func(entry).await; + }) + } + Ok(None) => { + warn!("decommission_pool: list_objects_to_decommission get none"); + Box::pin(async {}) + } + Err(err) => { + error!("decommission_pool: list_objects_to_decommission get err {:?}", &err); + Box::pin(async {}) + } } })), ..Default::default() diff --git a/ecstore/src/rebalance.rs b/ecstore/src/rebalance.rs index 56f10f0d0..62c5d696f 100644 --- a/ecstore/src/rebalance.rs +++ b/ecstore/src/rebalance.rs @@ -20,7 +20,7 @@ use http::HeaderMap; use serde::{Deserialize, Serialize}; use tokio::sync::broadcast::{self, Receiver as B_Receiver}; use tokio::time::{Duration, Instant}; -use tracing::{error, info}; +use tracing::{error, info, warn}; use uuid::Uuid; use workers::workers::Workers; @@ -162,6 +162,7 @@ impl RebalanceMeta { pub async fn load_with_opts(&mut self, store: Arc, opts: ObjectOptions) -> Result<()> { let (data, _) = read_config_with_metadata(store, REBAL_META_NAME, &opts).await?; if data.is_empty() { + warn!("rebalanceMeta: no data"); return Ok(()); } if data.len() <= 4 { @@ -183,6 +184,7 @@ impl RebalanceMeta { self.last_refreshed_at = Some(SystemTime::now()); + warn!("rebalanceMeta: loaded meta done"); Ok(()) } @@ -214,20 +216,33 @@ impl ECStore { #[tracing::instrument(skip_all)] pub async fn load_rebalance_meta(&self) -> Result<()> { let mut meta = RebalanceMeta::new(); + warn!("rebalanceMeta: load rebalance meta"); match meta.load(self.pools[0].clone()).await { Ok(_) => { - let mut rebalance_meta = self.rebalance_meta.write().await; + warn!("rebalanceMeta: rebalance meta loaded0"); + { + let mut rebalance_meta = self.rebalance_meta.write().await; - *rebalance_meta = Some(meta); + *rebalance_meta = Some(meta); + + drop(rebalance_meta); + } + + warn!("rebalanceMeta: rebalance meta loaded1"); if let Err(err) = self.update_rebalance_stats().await { error!("Failed to update rebalance stats: {}", err); + } else { + warn!("rebalanceMeta: rebalance meta loaded2"); } } Err(err) => { if !is_err_config_not_found(&err) { + error!("rebalanceMeta: load rebalance meta err {:?}", &err); return Err(err); } + + error!("rebalanceMeta: not found, rebalance not started"); } } @@ -237,21 +252,24 @@ impl ECStore { #[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; for i in 0..self.pools.len() { if self.find_index(i).await.is_none() { + let mut rebalance_meta = self.rebalance_meta.write().await; if let Some(meta) = rebalance_meta.as_mut() { meta.pool_stats.push(RebalanceStats::default()); } ok = true; + drop(rebalance_meta); } } if ok { + let mut rebalance_meta = self.rebalance_meta.write().await; if let Some(meta) = rebalance_meta.as_mut() { meta.save(self.pools[0].clone()).await?; } + drop(rebalance_meta); } Ok(()) @@ -267,6 +285,7 @@ impl ECStore { #[tracing::instrument(skip(self))] pub async fn init_rebalance_meta(&self, bucktes: Vec) -> Result { + warn!("init_rebalance_meta: start rebalance"); let si = self.storage_info().await; let mut disk_stats = vec![DiskStat::default(); self.pools.len()]; @@ -321,10 +340,15 @@ impl ECStore { meta.save(self.pools[0].clone()).await?; + warn!("init_rebalance_meta: rebalance meta saved"); + let id = meta.id.clone(); - let mut rebalance_meta = self.rebalance_meta.write().await; - *rebalance_meta = Some(meta); + { + let mut rebalance_meta = self.rebalance_meta.write().await; + *rebalance_meta = Some(meta); + drop(rebalance_meta); + } Ok(id) } @@ -424,6 +448,7 @@ impl ECStore { #[tracing::instrument(skip_all)] pub async fn start_rebalance(self: &Arc) { + warn!("start_rebalance: start rebalance"); // let rebalance_meta = self.rebalance_meta.read().await; let (tx, rx) = broadcast::channel::(1); @@ -433,13 +458,17 @@ impl ECStore { if let Some(meta) = rebalance_meta.as_mut() { meta.cancel = Some(tx) } else { + error!("start_rebalance: rebalance_meta is None exit"); return; } + + drop(rebalance_meta); } let participants = { if let Some(ref meta) = *self.rebalance_meta.read().await { if meta.stopped_at.is_some() { + warn!("start_rebalance: rebalance already stopped exit"); return; } @@ -457,6 +486,7 @@ impl ECStore { for (idx, participating) in participants.iter().enumerate() { if !*participating { + warn!("start_rebalance: pool {} is not participating, skipping", idx); continue; } @@ -465,6 +495,7 @@ impl ECStore { .get(idx) .map_or(true, |v| v.endpoints.as_ref().first().map_or(true, |e| e.is_local)) { + warn!("start_rebalance: pool {} is not local, skipping", idx); continue; } @@ -479,6 +510,8 @@ impl ECStore { } }); } + + warn!("start_rebalance: rebalance started done"); } #[tracing::instrument(skip(self, rx))] @@ -540,6 +573,7 @@ impl ECStore { } if quit { + warn!("{}: exiting save_task", msg); return; } @@ -547,13 +581,14 @@ impl ECStore { } }); - tracing::info!("Pool {} rebalancing is started", pool_index + 1); + tracing::warn!("Pool {} rebalancing is started", pool_index + 1); while let Some(bucket) = self.next_rebal_bucket(pool_index).await? { tracing::info!("Rebalancing bucket: {}", bucket); if let Err(err) = self.rebalance_bucket(rx.resubscribe(), bucket.clone(), pool_index).await { if err.to_string().contains("not initialized") { + warn!("rebalance_bucket: rebalance not initialized, continue"); continue; } tracing::error!("Error rebalancing bucket {}: {:?}", bucket, err); @@ -564,7 +599,7 @@ impl ECStore { self.bucket_rebalance_done(pool_index, bucket).await?; } - tracing::info!("Pool {} rebalancing is done", pool_index + 1); + tracing::warn!("Pool {} rebalancing is done", pool_index + 1); done_tx.send(Ok(())).await.ok(); save_task.await.ok(); @@ -1037,10 +1072,21 @@ impl SetDisks { partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option]| { // let cb = cb.clone(); let resolver = resolver.clone(); - if let Ok(Some(entry)) = entries.resolve(resolver) { - cb(entry) - } else { - Box::pin(async {}) + let cb = cb.clone(); + + match entries.resolve(resolver) { + Ok(Some(entry)) => { + warn!("rebalance: list_objects_to_decommission get {}", &entry.name); + Box::pin(async move { cb(entry).await }) + } + Ok(None) => { + warn!("rebalance: list_objects_to_decommission get none"); + Box::pin(async {}) + } + Err(err) => { + error!("rebalance: list_objects_to_decommission get err {:?}", &err); + Box::pin(async {}) + } } })), ..Default::default() diff --git a/ecstore/src/set_disk.rs b/ecstore/src/set_disk.rs index 19c244190..bd55ab39f 100644 --- a/ecstore/src/set_disk.rs +++ b/ecstore/src/set_disk.rs @@ -3667,7 +3667,7 @@ impl ObjectIO for SetDisks { error!("get_object_with_fileinfo err {:?}", e); }; - // error!("get_object_with_fileinfo end"); + // error!("get_object_with_fileinfo end {}/{}", bucket, object); }); Ok(reader) diff --git a/rustfs/src/admin/handlers/rebalance.rs b/rustfs/src/admin/handlers/rebalance.rs index fefd7df42..c33767781 100644 --- a/rustfs/src/admin/handlers/rebalance.rs +++ b/rustfs/src/admin/handlers/rebalance.rs @@ -34,9 +34,9 @@ pub struct RebalPoolProgress { #[serde(rename = "object")] pub object: String, #[serde(rename = "elapsed")] - pub elapsed: Duration, + pub elapsed: u64, #[serde(rename = "eta")] - pub eta: Duration, + pub eta: u64, } #[derive(Debug, Clone, Serialize, Deserialize, Default)] @@ -78,13 +78,13 @@ impl Operation for RebalanceStart { if store.is_decommission_running().await { return Err(s3_error!( - InternalError, + InvalidRequest, "Rebalance cannot be started, decommission is already in progress" )); } if store.is_rebalance_started().await { - return Err(s3_error!(InternalError, "Rebalance already in progress")); + return Err(s3_error!(OperationAborted, "Rebalance already in progress")); } let bucket_infos = store @@ -101,8 +101,11 @@ impl Operation for RebalanceStart { } }; + warn!("Rebalance started with id: {}", id); if let Some(notification_sys) = get_global_notification_sys() { + warn!("Loading rebalance meta"); notification_sys.load_rebalance_meta(true).await; + warn!("Rebalance meta loaded"); } let resp = RebalanceResp { id }; @@ -149,7 +152,7 @@ impl Operation for RebalanceStatus { disk_stats[disk.pool_index as usize].total_space += disk.total_space; } - let stop_time = meta.stopped_at; + let mut stop_time = meta.stopped_at; let mut admin_status = RebalanceAdminStatus { id: meta.id.clone(), stopped_at: meta.stopped_at, @@ -171,7 +174,7 @@ impl Operation for RebalanceStatus { // Calculate total bytes to be rebalanced let total_bytes_to_rebal = ps.init_capacity as f64 * meta.percent_free_goal - ps.init_free_space as f64; - let elapsed = if let Some(start_time) = ps.info.start_time { + 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))? @@ -179,21 +182,25 @@ impl Operation for RebalanceStatus { return Err(s3_error!(InternalError, "Start time is not available")); }; - let eta = if ps.bytes > 0 { + let mut eta = if ps.bytes > 0 { Duration::from_secs_f64(total_bytes_to_rebal * elapsed.as_secs_f64() / ps.bytes as f64) } else { Duration::ZERO }; - let stop_time = ps.info.end_time.unwrap_or(stop_time.unwrap_or(SystemTime::now())); + if ps.info.end_time.is_some() { + stop_time = ps.info.end_time; + } - let elapsed = if ps.info.end_time.is_some() || meta.stopped_at.is_some() { - stop_time - .duration_since(ps.info.start_time.unwrap_or(stop_time)) - .unwrap_or_default() - } else { - elapsed - }; + 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")); + } + + eta = Duration::ZERO; + } admin_status.pools[i].progress = Some(RebalPoolProgress { num_objects: ps.num_objects, @@ -201,8 +208,8 @@ impl Operation for RebalanceStatus { bytes: ps.bytes, bucket: ps.bucket.clone(), object: ps.object.clone(), - elapsed, - eta, + elapsed: elapsed.as_secs(), + eta: eta.as_secs(), }); } diff --git a/rustfs/src/grpc.rs b/rustfs/src/grpc.rs index 6e702617b..ce8092dfb 100644 --- a/rustfs/src/grpc.rs +++ b/rustfs/src/grpc.rs @@ -40,7 +40,7 @@ use tokio::spawn; use tokio::sync::mpsc; use tokio_stream::wrappers::ReceiverStream; use tonic::{Request, Response, Status, Streaming}; -use tracing::{debug, error, info}; +use tracing::{debug, error, info, warn}; type ResponseStream = Pin> + Send>>; @@ -2388,19 +2388,27 @@ impl Node for NodeService { let LoadRebalanceMetaRequest { start_rebalance } = request.into_inner(); + warn!("handle LoadRebalanceMetaRequest"); + store.load_rebalance_meta().await.map_err(|err| { error!("load_rebalance_meta err {:?}", err); Status::internal(err.to_string()) })?; + warn!("load_rebalance_meta success"); + if start_rebalance { + warn!("start rebalance"); let store = store.clone(); tokio::spawn(async move { store.start_rebalance().await; }); } - unimplemented!() + Ok(tonic::Response::new(LoadRebalanceMetaResponse { + success: true, + error_info: None, + })) } async fn load_transition_tier_config( From e8f8a0872d4368c80397654ad8ce356041709c7e Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 27 Apr 2025 09:50:32 +0800 Subject: [PATCH 3/9] modify Telemetry filter order --- crates/obs/Cargo.toml | 3 ++- crates/obs/src/telemetry.rs | 18 +++++++++--------- deploy/config/obs.example.toml | 1 + scripts/run.sh | 2 +- 4 files changed, 13 insertions(+), 11 deletions(-) diff --git a/crates/obs/Cargo.toml b/crates/obs/Cargo.toml index f6d9c4787..26c264335 100644 --- a/crates/obs/Cargo.toml +++ b/crates/obs/Cargo.toml @@ -21,6 +21,7 @@ full = ["file", "gpu", "kafka", "webhook"] async-trait = { workspace = true } chrono = { workspace = true } config = { workspace = true } +local-ip-address = { workspace = true } nvml-wrapper = { workspace = true, optional = true } opentelemetry = { workspace = true } opentelemetry-appender-tracing = { workspace = true, features = ["experimental_use_tracing_span_context", "experimental_metadata_attributes"] } @@ -41,7 +42,7 @@ reqwest = { workspace = true, optional = true, default-features = false } serde_json = { workspace = true } sysinfo = { workspace = true } thiserror = { workspace = true } -local-ip-address = { workspace = true } + [dev-dependencies] diff --git a/crates/obs/src/telemetry.rs b/crates/obs/src/telemetry.rs index 654ae4145..931e24496 100644 --- a/crates/obs/src/telemetry.rs +++ b/crates/obs/src/telemetry.rs @@ -213,22 +213,22 @@ pub fn init_telemetry(config: &OtelConfig) -> OtelGuard { .with_line_number(true); 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); - + let tracer = tracer_provider.tracer(Cow::Borrowed(service_name).to_string()); // Configure registry to avoid repeated calls to filter methods - tracing_subscriber::registry() + let registry = 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(); + .with(ErrorLayer::default()); + info!("Telemetry logging enabled: {:?}", config.local_logging_enabled); + if config.local_logging_enabled.unwrap_or(false) { + registry.with(fmt_layer).init(); + } else { + registry.init(); + } if !endpoint.is_empty() { info!( diff --git a/deploy/config/obs.example.toml b/deploy/config/obs.example.toml index 97192fb4f..e38e06f81 100644 --- a/deploy/config/obs.example.toml +++ b/deploy/config/obs.example.toml @@ -7,6 +7,7 @@ service_name = "rustfs" service_version = "0.1.0" environment = "develop" logger_level = "error" +local_logging_enabled = true [sinks] [sinks.kafka] # Kafka sink is disabled by default diff --git a/scripts/run.sh b/scripts/run.sh index 1d1470d93..7717a371d 100755 --- a/scripts/run.sh +++ b/scripts/run.sh @@ -47,7 +47,7 @@ export RUSTFS__OBSERVABILITY__SERVICE_NAME=rustfs export RUSTFS__OBSERVABILITY__SERVICE_VERSION=0.1.0 export RUSTFS__OBSERVABILITY__ENVIRONMENT=develop export RUSTFS__OBSERVABILITY__LOGGER_LEVEL=debug -export RUSTFS__OBSERVABILITY__LOCAL_LOGGER_ENABLED=true +export RUSTFS__OBSERVABILITY__LOCAL_LOGGING_ENABLED=true export RUSTFS__SINKS__FILE__ENABLED=true export RUSTFS__SINKS__FILE__PATH="./deploy/logs/rustfs.log" export RUSTFS__SINKS__WEBHOOK__ENABLED=false From e9d6e2ca95e085181692c89e5847adf8dcc8e122 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 27 Apr 2025 23:44:26 +0800 Subject: [PATCH 4/9] feat: improve address binding and port handling mechanism (#366) * feat: improve address binding and port handling mechanism 1. Add support for ":port" format to enable dual-stack binding (IPv4/IPv6) 2. Implement automatic port allocation when port 0 is specified 3. Optimize server startup process with unified address resolution 4. Enhance error handling and logging for address resolution 5. Improve graceful shutdown with signal listening 6. Clean up commented code in console.rs Files: - ecstore/src/utils/net.rs - rustfs/src/console.rs - rustfs/src/main.rs Branch: feature/server-and-console-port * improve code for console * improve code * improve code for console and net.rs * Update rustfs/src/main.rs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update rustfs/src/utils/mod.rs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- Cargo.toml | 1 - ecstore/src/utils/net.rs | 29 ++++++- rustfs/src/config/mod.rs | 4 +- rustfs/src/console.rs | 65 ++++------------ rustfs/src/main.rs | 74 +++++++++++------- rustfs/src/utils/mod.rs | 161 ++++++++++++++++++++++++++++++++++++++- scripts/run.sh | 11 ++- 7 files changed, 259 insertions(+), 86 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index b9eb8dfd1..59c48cc14 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -106,7 +106,6 @@ opentelemetry-otlp = { version = "0.29.0" } opentelemetry-semantic-conventions = { version = "0.29.0", features = ["semconv_experimental"] } parking_lot = "0.12.3" pin-project-lite = "0.2.16" -prometheus = "0.14.0" # pin-utils = "0.1.0" prost = "0.13.5" prost-build = "0.13.5" diff --git a/ecstore/src/utils/net.rs b/ecstore/src/utils/net.rs index e892dce37..bcd2c80d1 100644 --- a/ecstore/src/utils/net.rs +++ b/ecstore/src/utils/net.rs @@ -3,7 +3,7 @@ use lazy_static::lazy_static; use std::{ collections::HashSet, fmt::Display, - net::{IpAddr, SocketAddr, TcpListener, ToSocketAddrs}, + net::{IpAddr, Ipv6Addr, SocketAddr, TcpListener, ToSocketAddrs}, }; use url::Host; @@ -141,6 +141,33 @@ impl TryFrom for XHost { } } +/// parses the address string, process the ":port" format for double-stack binding, +/// and resolve the host name or IP address. If the port is 0, an available port is assigned. +pub fn parse_and_resolve_address(addr_str: &str) -> Result { + let resolved_addr: SocketAddr = if let Some(port) = addr_str.strip_prefix(":") { + // Process the ":port" format for double stack binding + let port_str = port; + let port: u16 = port_str + .parse() + .map_err(|e| Error::from_string(format!("Invalid port format: {}, err:{:?}", addr_str, e)))?; + let final_port = if port == 0 { + get_available_port() // assume get_available_port is available here + } else { + port + }; + // Using IPv6 without address specified [::], it should handle both IPv4 and IPv6 + SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), final_port) + } else { + // Use existing logic to handle regular address formats + let mut addr = check_local_server_addr(addr_str)?; // assume check_local_server_addr is available here + if addr.port() == 0 { + addr.set_port(get_available_port()); + } + addr + }; + Ok(resolved_addr) +} + #[cfg(test)] mod test { use std::net::Ipv4Addr; diff --git a/rustfs/src/config/mod.rs b/rustfs/src/config/mod.rs index e877ebada..7a7bdc4c4 100644 --- a/rustfs/src/config/mod.rs +++ b/rustfs/src/config/mod.rs @@ -29,11 +29,11 @@ pub const DEFAULT_OBS_CONFIG: &str = "config/obs.toml"; /// Default TLS key for rustfs /// This is the default key for TLS. -pub(crate) const RUSTFS_TLS_KEY: &str = "rustfs_private.key"; +pub(crate) const RUSTFS_TLS_KEY: &str = "rustfs_key.pem"; /// Default TLS cert for rustfs /// This is the default cert for TLS. -pub(crate) const RUSTFS_TLS_CERT: &str = "rustfs_public.crt"; +pub(crate) const RUSTFS_TLS_CERT: &str = "rustfs_cert.pem"; #[allow(clippy::const_is_empty)] const SHORT_VERSION: &str = { diff --git a/rustfs/src/console.rs b/rustfs/src/console.rs index 1125b24d0..c3edbaaa1 100644 --- a/rustfs/src/console.rs +++ b/rustfs/src/console.rs @@ -207,42 +207,6 @@ async fn config_handler(uri: Uri, Host(host): Host) -> impl IntoResponse { let mut cfg = CONSOLE_CONFIG.get().unwrap().clone(); let url = format!("{}://{}:{}", scheme, host, cfg.port); - - // // 如果指定入口，直接使用 - // let url = if let Some(endpoint) = &config::get_config().console_fs_endpoint { - // debug!("axum Using rustfs endpoint address: {}", endpoint); - // endpoint.clone() - // } else { - // let host_with_port = if host.contains(':') { - // host.clone() - // } else { - // format!("{}:80", host) - // }; - // // 尝试解析为 socket address，但不强制要求一定要是 IP 地址 - // let socket_addr = host_with_port.to_socket_addrs().ok().and_then(|mut addrs| addrs.next()); - // debug!("axum Using host with port: {}, Socket address: {:?}", host_with_port, socket_addr); - // match socket_addr { - // Some(addr) if addr.ip().is_ipv4() => { - // let ipv4 = addr.ip().to_string(); - // // 如果是私有 IP、环回地址或未指定地址，保留原始域名 - // if is_private_ip(addr.ip()) || addr.ip().is_loopback() || addr.ip().is_unspecified() { - // let (host, _) = host_with_port.split_once(':').unwrap_or((&host, "80")); - // debug!("axum Using private IPv4 address: {}", host); - // format!("http://{}:{}", host, cfg.port) - // } else { - // debug!("axum Using public IPv4 address"); - // format!("http://{}:{}", ipv4, cfg.port) - // } - // } - // _ => { - // // 如果不是有效的 IPv4 地址，保留原始域名 - // let (host, _) = host_with_port.split_once(':').unwrap_or((&host, "80")); - // debug!("axum Using domain address: {}", host); - // format!("http://{}:{}", host, cfg.port) - // } - // } - // }; - cfg.api.base_url = format!("{}{}", url, RUSTFS_ADMIN_PREFIX); cfg.s3.endpoint = url; @@ -273,26 +237,29 @@ pub async fn start_static_file_server( .layer(cors) .layer(tower_http::compression::CompressionLayer::new().gzip(true).deflate(true)) .layer(TraceLayer::new_for_http()); - let local_addr: SocketAddr = addrs.parse().expect("Failed to parse socket address"); - info!("WebUI: http://{}:{} http://127.0.0.1:{}", local_ip, local_addr.port(), local_addr.port()); + + use ecstore::utils::net; + let server_addr = net::parse_and_resolve_address(addrs).expect("Failed to parse socket address"); + let server_port = server_addr.port(); + let server_address = server_addr.to_string(); + + info!( + "WebUI: http://{}:{} http://127.0.0.1:{} http://{}", + local_ip, server_port, server_port, server_address + ); info!(" RootUser: {}", access_key); info!(" RootPass: {}", secret_key); // Check and start the HTTPS/HTTP server - match start_server(addrs, local_addr, tls_path, app.clone()).await { + match start_server(server_addr, tls_path, app.clone()).await { Ok(_) => info!("Server shutdown gracefully"), Err(e) => error!("Server error: {}", e), } } -async fn start_server(addrs: &str, local_addr: SocketAddr, tls_path: Option, app: Router) -> io::Result<()> { +async fn start_server(server_addr: SocketAddr, tls_path: Option, app: Router) -> io::Result<()> { let tls_path = tls_path.unwrap_or_default(); let key_path = format!("{}/{}", tls_path, RUSTFS_TLS_KEY); let cert_path = format!("{}/{}", tls_path, RUSTFS_TLS_CERT); - - let addr = addrs - .parse::() - .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("Invalid address: {}", e)))?; - let handle = axum_server::Handle::new(); // create a signal off listening task let handle_clone = handle.clone(); @@ -309,24 +276,24 @@ async fn start_server(addrs: &str, local_addr: SocketAddr, tls_path: Option { info!("Starting HTTPS server..."); - axum_server::bind_rustls(local_addr, config) + axum_server::bind_rustls(server_addr, config) .handle(handle.clone()) .serve(app.into_make_service()) .await .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?; - info!("HTTPS server running on https://{}", addr); + info!("HTTPS server running on https://{}", server_addr); Ok(()) } Err(e) => { error!("Failed to create TLS config: {}", e); - start_http_server(addr, app, handle).await + start_http_server(server_addr, app, handle).await } } } else { info!("TLS certificates not found at {} and {}", key_path, cert_path); - start_http_server(addr, app, handle).await + start_http_server(server_addr, app, handle).await } } diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index ca73521f8..7871fba58 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -27,7 +27,7 @@ use ecstore::config as ecconfig; use ecstore::config::GLOBAL_ConfigSys; use ecstore::heal::background_heal_ops::init_auto_heal; use ecstore::store_api::BucketOptions; -use ecstore::utils::net::{self, get_available_port}; +use ecstore::utils::net; use ecstore::StorageAPI; use ecstore::{ endpoints::EndpointServerPools, @@ -136,14 +136,8 @@ async fn run(opt: config::Opt) -> Result<()> { debug!("opt: {:?}", &opt); - let mut server_addr = net::check_local_server_addr(opt.address.as_str())?; - - if server_addr.port() == 0 { - server_addr.set_port(get_available_port()); - } - + let server_addr = net::parse_and_resolve_address(opt.address.as_str())?; let server_port = server_addr.port(); - let server_address = server_addr.to_string(); debug!("server_address {}", &server_address); @@ -263,32 +257,58 @@ async fn run(opt: config::Opt) -> Result<()> { } }); - let rpc_service = NodeServiceServer::with_interceptor(make_server(), check_auth); - let tls_path = opt.tls_path.clone().unwrap_or_default(); - let key_path = format!("{}/{}", tls_path, RUSTFS_TLS_KEY); - let cert_path = format!("{}/{}", tls_path, RUSTFS_TLS_CERT); - let has_tls_certs = tokio::try_join!(tokio::fs::metadata(key_path.clone()), tokio::fs::metadata(cert_path.clone())).is_ok(); - debug!("Main TLS certs: {:?}", has_tls_certs); + let has_tls_certs = tokio::fs::metadata(&tls_path).await.is_ok(); let tls_acceptor = if has_tls_certs { - debug!("Found TLS certificates, starting with HTTPS"); - let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); - let certs = utils::load_certs(cert_path.as_str()).map_err(|e| error(e.to_string()))?; - let key = utils::load_private_key(key_path.as_str()).map_err(|e| error(e.to_string()))?; - let mut server_config = ServerConfig::builder() - .with_no_client_auth() - .with_single_cert(certs, key) - .map_err(|e| error(e.to_string()))?; - server_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec(), b"http/1.0".to_vec()]; - Some(TlsAcceptor::from(Arc::new(server_config))) + debug!("Found TLS directory, checking for certificates"); + + // 1. Try to load all certificates directly (including root and subdirectories) + match utils::load_all_certs_from_directory(&tls_path) { + Ok(cert_key_pairs) if !cert_key_pairs.is_empty() => { + debug!("Found {} certificates, starting with HTTPS", cert_key_pairs.len()); + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + + // create a multi certificate configuration + let mut server_config = ServerConfig::builder() + .with_no_client_auth() + .with_cert_resolver(Arc::new(utils::create_multi_cert_resolver(cert_key_pairs)?)); + + server_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec(), b"http/1.0".to_vec()]; + Some(TlsAcceptor::from(Arc::new(server_config))) + } + _ => { + // 2. If the synthesis fails, fall back to the traditional document certificate mode (backward compatible) + let key_path = format!("{}/{}", tls_path, RUSTFS_TLS_KEY); + let cert_path = format!("{}/{}", tls_path, RUSTFS_TLS_CERT); + let has_single_cert = + tokio::try_join!(tokio::fs::metadata(key_path.clone()), tokio::fs::metadata(cert_path.clone())).is_ok(); + + if has_single_cert { + debug!("Found legacy single TLS certificate, starting with HTTPS"); + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + let certs = utils::load_certs(cert_path.as_str()).map_err(|e| error(e.to_string()))?; + let key = utils::load_private_key(key_path.as_str()).map_err(|e| error(e.to_string()))?; + let mut server_config = ServerConfig::builder() + .with_no_client_auth() + .with_single_cert(certs, key) + .map_err(|e| error(e.to_string()))?; + server_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec(), b"http/1.0".to_vec()]; + Some(TlsAcceptor::from(Arc::new(server_config))) + } else { + debug!("No valid TLS certificates found, starting with HTTP"); + None + } + } + } } else { debug!("TLS certificates not found, starting with HTTP"); None }; + let rpc_service = NodeServiceServer::with_interceptor(make_server(), check_auth); let state_manager = ServiceStateManager::new(); let worker_state_manager = state_manager.clone(); - // 更新服务状态为启动中 + // Update service status to Starting state_manager.update(ServiceState::Starting); // Create shutdown channel @@ -296,7 +316,7 @@ async fn run(opt: config::Opt) -> Result<()> { let shutdown_tx_clone = shutdown_tx.clone(); tokio::spawn(async move { - // 错误处理改进 + // error handling improvements let sigterm_inner = match signal(SignalKind::terminate()) { Ok(signal) => signal, Err(e) => { @@ -367,7 +387,7 @@ async fn run(opt: config::Opt) -> Result<()> { let graceful = Arc::new(GracefulShutdown::new()); debug!("graceful initiated"); - // 服务准备就绪 + // service ready worker_state_manager.update(ServiceState::Ready); let value = hybrid_service.clone(); loop { diff --git a/rustfs/src/utils/mod.rs b/rustfs/src/utils/mod.rs index 2d12f64d2..fad2718c8 100644 --- a/rustfs/src/utils/mod.rs +++ b/rustfs/src/utils/mod.rs @@ -1,8 +1,19 @@ +use crate::config::{RUSTFS_TLS_CERT, RUSTFS_TLS_KEY}; +use rustls::server::{ClientHello, ResolvesServerCert, ResolvesServerCertUsingSni}; +use rustls::sign::CertifiedKey; use rustls_pemfile::{certs, private_key}; use rustls_pki_types::{CertificateDer, PrivateKeyDer}; +use std::collections::HashMap; +use std::fmt::Debug; +use std::io::Error; use std::net::IpAddr; +use std::path::Path; +use std::sync::Arc; use std::{fs, io}; +use tracing::{debug, warn}; +/// Get the local IP address. +/// This function retrieves the local IP address of the machine. pub(crate) fn get_local_ip() -> Option { match local_ip_address::local_ip() { Ok(IpAddr::V4(ip)) => Some(ip), @@ -12,17 +23,24 @@ pub(crate) fn get_local_ip() -> Option { } /// Load public certificate from file. +/// This function loads a public certificate from the specified file. pub(crate) fn load_certs(filename: &str) -> io::Result>> { // Open certificate file. let cert_file = fs::File::open(filename).map_err(|e| error(format!("failed to open {}: {}", filename, e)))?; let mut reader = io::BufReader::new(cert_file); // Load and return certificate. - let certs = certs(&mut reader).collect::, _>>()?; + let certs = certs(&mut reader) + .collect::, _>>() + .map_err(|_| error(format!("certificate file {} format error", filename)))?; + if certs.is_empty() { + return Err(error(format!("No valid certificate was found in the certificate file {}", filename))); + } Ok(certs) } /// Load private key from file. +/// This function loads a private key from the specified file. pub(crate) fn load_private_key(filename: &str) -> io::Result> { // Open keyfile. let keyfile = fs::File::open(filename).map_err(|e| error(format!("failed to open {}: {}", filename, e)))?; @@ -32,6 +50,143 @@ pub(crate) fn load_private_key(filename: &str) -> io::Result io::Error { - io::Error::new(io::ErrorKind::Other, err) +/// error function +pub(crate) fn error(err: String) -> Error { + Error::new(io::ErrorKind::Other, err) +} + +/// Load all certificates and private keys in the directory +/// This function loads all certificate and private key pairs from the specified directory. +/// It looks for files named `rustfs_cert.pem` and `rustfs_key.pem` in each subdirectory. +/// The root directory can also contain a default certificate/private key pair. +pub(crate) fn load_all_certs_from_directory( + dir_path: &str, +) -> io::Result>, PrivateKeyDer<'static>)>> { + let mut cert_key_pairs = HashMap::new(); + let dir = Path::new(dir_path); + + if !dir.exists() || !dir.is_dir() { + return Err(error(format!( + "The certificate directory does not exist or is not a directory: {}", + dir_path + ))); + } + + // 1. First check whether there is a certificate/private key pair in the root directory + let root_cert_path = dir.join(RUSTFS_TLS_CERT); + let root_key_path = dir.join(RUSTFS_TLS_KEY); + + if root_cert_path.exists() && root_key_path.exists() { + debug!("find the root directory certificate: {:?}", root_cert_path); + let root_cert_str = root_cert_path + .to_str() + .ok_or_else(|| error(format!("Invalid UTF-8 in root certificate path: {:?}", root_cert_path)))?; + let root_key_str = root_key_path + .to_str() + .ok_or_else(|| error(format!("Invalid UTF-8 in root key path: {:?}", root_key_path)))?; + match load_cert_key_pair(root_cert_str, root_key_str) { + Ok((certs, key)) => { + // The root directory certificate is used as the default certificate and is stored using special keys. + cert_key_pairs.insert("default".to_string(), (certs, key)); + } + Err(e) => { + warn!("unable to load root directory certificate: {}", e); + } + } + } + + // 2.iterate through all folders in the directory + for entry in fs::read_dir(dir)? { + let entry = entry?; + let path = entry.path(); + + if path.is_dir() { + let domain_name = path + .file_name() + .and_then(|name| name.to_str()) + .ok_or_else(|| error(format!("invalid domain name directory:{:?}", path)))?; + + // find certificate and private key files + let cert_path = path.join(RUSTFS_TLS_CERT); // e.g., rustfs_cert.pem + let key_path = path.join(RUSTFS_TLS_KEY); // e.g., rustfs_key.pem + + if cert_path.exists() && key_path.exists() { + debug!("find the domain name certificate: {} in {:?}", domain_name, cert_path); + match load_cert_key_pair(cert_path.to_str().unwrap(), key_path.to_str().unwrap()) { + Ok((certs, key)) => { + cert_key_pairs.insert(domain_name.to_string(), (certs, key)); + } + Err(e) => { + warn!("unable to load the certificate for {} domain name: {}", domain_name, e); + } + } + } + } + } + + if cert_key_pairs.is_empty() { + return Err(error(format!("No valid certificate/private key pair found in directory {}", dir_path))); + } + + Ok(cert_key_pairs) +} + +/// loading a single certificate private key pair +/// This function loads a certificate and private key from the specified paths. +/// It returns a tuple containing the certificate and private key. +fn load_cert_key_pair(cert_path: &str, key_path: &str) -> io::Result<(Vec>, PrivateKeyDer<'static>)> { + let certs = load_certs(cert_path)?; + let key = load_private_key(key_path)?; + Ok((certs, key)) +} + +/// Create a multi-cert resolver +/// This function loads all certificates and private keys from the specified directory. +/// It uses the first certificate/private key pair found in the root directory as the default certificate. +/// The rest of the certificates/private keys are used for SNI resolution. +/// +pub fn create_multi_cert_resolver( + cert_key_pairs: HashMap>, PrivateKeyDer<'static>)>, +) -> io::Result { + #[derive(Debug)] + struct MultiCertResolver { + cert_resolver: ResolvesServerCertUsingSni, + default_cert: Option>, + } + impl ResolvesServerCert for MultiCertResolver { + fn resolve(&self, client_hello: ClientHello) -> Option> { + // try matching certificates with sni + if let Some(cert) = self.cert_resolver.resolve(client_hello) { + return Some(cert); + } + + // If there is no matching SNI certificate, use the default certificate + self.default_cert.clone() + } + } + + let mut resolver = ResolvesServerCertUsingSni::new(); + let mut default_cert = None; + + for (domain, (certs, key)) in cert_key_pairs { + // create a signature + let signing_key = rustls::crypto::aws_lc_rs::sign::any_supported_type(&key) + .map_err(|_| error(format!("unsupported private key types:{}", domain)))?; + + // create a CertifiedKey + let certified_key = CertifiedKey::new(certs, signing_key); + if domain == "default" { + default_cert = Some(Arc::new(certified_key.clone())); + } else { + // add certificate to resolver + resolver + .add(&domain, certified_key) + .map_err(|e| error(format!("failed to add a domain name certificate:{},err: {:?}", domain, e)))?; + } + } + + Ok(MultiCertResolver { + cert_resolver: resolver, + default_cert, + }) } diff --git a/scripts/run.sh b/scripts/run.sh index 7717a371d..690ac8511 100755 --- a/scripts/run.sh +++ b/scripts/run.sh @@ -28,12 +28,12 @@ fi export RUSTFS_VOLUMES="./target/volume/test{0...4}" # export RUSTFS_VOLUMES="./target/volume/test" -export RUSTFS_ADDRESS="[::]:9000" +export RUSTFS_ADDRESS=":9000" export RUSTFS_CONSOLE_ENABLE=true -export RUSTFS_CONSOLE_ADDRESS="[::]:9002" +export RUSTFS_CONSOLE_ADDRESS=":9002" # export RUSTFS_SERVER_DOMAINS="localhost:9000" # HTTPS 证书目录 -# export RUSTFS_TLS_PATH="./deploy/certs" + export RUSTFS_TLS_PATH="./deploy/certs" # 具体路径修改为配置文件真实路径，obs.example.toml 仅供参考 其中`RUSTFS_OBS_CONFIG` 和下面变量二选一 export RUSTFS_OBS_CONFIG="./deploy/config/obs.example.toml" @@ -58,6 +58,11 @@ export RUSTFS__SINKS__KAFKA__BOOTSTRAP_SERVERS="" export RUSTFS__SINKS__KAFKA__TOPIC="" export RUSTFS__LOGGER__QUEUE_CAPACITY=10 +export OTEL_INSTRUMENTATION_NAME="rustfs" +export OTEL_INSTRUMENTATION_VERSION="0.1.1" +export OTEL_INSTRUMENTATION_SCHEMA_URL="https://opentelemetry.io/schemas/1.31.0" +export OTEL_INSTRUMENTATION_ATTRIBUTES="env=production" + # 事件消息配置 export RUSTFS_EVENT_CONFIG="./deploy/config/event.example.toml" From c25c94ec304a53874590dff142a1e767c8336e72 Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 28 Apr 2025 11:34:52 +0800 Subject: [PATCH 5/9] upgrade config file --- .docker/observability/config/obs.toml | 1 + .docker/observability/jaeger-config.yaml | 11 +++++++++++ deploy/config/obs-zh.example.toml | 3 ++- 3 files changed, 14 insertions(+), 1 deletion(-) diff --git a/.docker/observability/config/obs.toml b/.docker/observability/config/obs.toml index 3b26644cf..e4ea037b1 100644 --- a/.docker/observability/config/obs.toml +++ b/.docker/observability/config/obs.toml @@ -7,6 +7,7 @@ service_name = "rustfs" service_version = "0.1.0" environments = "production" logger_level = "debug" +local_logging_enabled = true [sinks] [sinks.kafka] # Kafka sink is disabled by default diff --git a/.docker/observability/jaeger-config.yaml b/.docker/observability/jaeger-config.yaml index 0df35f836..205d804df 100644 --- a/.docker/observability/jaeger-config.yaml +++ b/.docker/observability/jaeger-config.yaml @@ -37,9 +37,14 @@ extensions: traces_archive: another_store ui: config_file: ./cmd/jaeger/config-ui.json + log_access: true # The maximum duration that is considered for clock skew adjustments. # Defaults to 0 seconds, which means it's disabled. max_clock_skew_adjust: 0s + grpc: + endpoint: 0.0.0.0:16685 + http: + endpoint: 0.0.0.0:16686 jaeger_storage: backends: @@ -49,6 +54,12 @@ extensions: another_store: memory: max_traces: 100000 + metric_backends: + some_metrics_storage: + prometheus: + endpoint: http://prometheus:9090 + normalize_calls: true + normalize_duration: true remote_sampling: # You can either use file or adaptive sampling strategy in remote_sampling diff --git a/deploy/config/obs-zh.example.toml b/deploy/config/obs-zh.example.toml index 2bf4260f8..712ae6cef 100644 --- a/deploy/config/obs-zh.example.toml +++ b/deploy/config/obs-zh.example.toml @@ -6,7 +6,8 @@ meter_interval = 30 # 指标收集间隔，单位为秒 service_name = "rustfs" # 服务名称，用于标识当前服务 service_version = "0.1.0" # 服务版本号 environments = "develop" # 运行环境，如开发环境 (develop) -logger_level = "debug" # 日志级别，可选 debug/info/warn/error 等 +logger_level = "debug" # 日志级别，可选 debug/info/warn/error 等 +local_logging_enabled = true # 是否启用本地 stdout 日志记录，true 表示启用，false 表示禁用 [sinks] [sinks.kafka] # Kafka 接收器配置 From 68a89d59e58b81627db11fc9fcca36a91c9e444f Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 28 Apr 2025 12:50:56 +0800 Subject: [PATCH 6/9] modify --- crates/obs/src/telemetry.rs | 23 +++++++++++++++-------- 1 file changed, 15 insertions(+), 8 deletions(-) diff --git a/crates/obs/src/telemetry.rs b/crates/obs/src/telemetry.rs index 931e24496..95bedfc4a 100644 --- a/crates/obs/src/telemetry.rs +++ b/crates/obs/src/telemetry.rs @@ -216,19 +216,26 @@ pub fn init_telemetry(config: &OtelConfig) -> OtelGuard { let otel_filter = build_env_filter(logger_level, None); let otel_layer = OpenTelemetryTracingBridge::new(&logger_provider).with_filter(otel_filter); let tracer = tracer_provider.tracer(Cow::Borrowed(service_name).to_string()); + // Configure registry to avoid repeated calls to filter methods - let registry = tracing_subscriber::registry() + let _registry = tracing_subscriber::registry() .with(filter) + .with(ErrorLayer::default()) + .with(if config.local_logging_enabled.unwrap_or(false) { + Some(fmt_layer) + } else { + None + }) + .with(OpenTelemetryLayer::new(tracer)) .with(otel_layer) .with(MetricsLayer::new(meter_provider.clone())) - .with(OpenTelemetryLayer::new(tracer)) - .with(ErrorLayer::default()); + .init(); info!("Telemetry logging enabled: {:?}", config.local_logging_enabled); - if config.local_logging_enabled.unwrap_or(false) { - registry.with(fmt_layer).init(); - } else { - registry.init(); - } + // if config.local_logging_enabled.unwrap_or(false) { + // registry.with(fmt_layer).init(); + // } else { + // registry.init(); + // } if !endpoint.is_empty() { info!( From 569099af9e1961c6daa9fdeba530fcb77eb0b748 Mon Sep 17 00:00:00 2001 From: junxiang Mu <1948535941@qq.com> Date: Mon, 28 Apr 2025 06:25:20 +0000 Subject: [PATCH 7/9] fix readme Signed-off-by: junxiang Mu <1948535941@qq.com> --- README.md | 2 +- ecstore/src/erasure.rs | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 511f6736e..812f13a3c 100644 --- a/README.md +++ b/README.md @@ -84,7 +84,7 @@ observability data formats (e.g. Jaeger, Prometheus, etc.) sending to one or mor 2. Run the following command: ```bash -docker-compose -f docker-compose.yml up -d +docker compose -f docker-compose.yml up -d ``` 3. Access the Grafana dashboard by navigating to `http://localhost:3000` in your browser. The default username and diff --git a/ecstore/src/erasure.rs b/ecstore/src/erasure.rs index 021ae7bed..590889e78 100644 --- a/ecstore/src/erasure.rs +++ b/ecstore/src/erasure.rs @@ -470,7 +470,7 @@ impl Erasure { self.encoder.as_ref().unwrap().reconstruct(&mut bufs)?; } - let shards = bufs.into_iter().flatten().collect::>(); + let shards = bufs.into_iter().flatten().map(Bytes::from).collect::>(); if shards.len() != self.parity_shards + self.data_shards { return Err(Error::from_string("can not reconstruct data")); } @@ -479,7 +479,7 @@ impl Erasure { if w.is_none() { continue; } - match w.as_mut().unwrap().write(shards[i].clone().into()).await { + match w.as_mut().unwrap().write(shards[i].clone()).await { Ok(_) => {} Err(e) => { info!("write failed, err: {:?}", e); From 43df8b6927ea04a12017321aff8d17533207bce0 Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 28 Apr 2025 14:37:28 +0800 Subject: [PATCH 8/9] improve readme.md --- .docker/observability/README.md | 16 ++++++++++++---- deploy/README.md | 18 +++++++++++++----- deploy/build/rustfs-zh.service | 4 ++++ deploy/build/rustfs.service | 4 ++++ deploy/certs/README.md | 16 +++++++++++++--- .../config/{.example.env => .example.obs.env} | 0 6 files changed, 46 insertions(+), 12 deletions(-) rename deploy/config/{.example.env => .example.obs.env} (100%) diff --git a/.docker/observability/README.md b/.docker/observability/README.md index a84f7592d..3d40319b1 100644 --- a/.docker/observability/README.md +++ b/.docker/observability/README.md @@ -6,7 +6,7 @@ This directory contains the observability stack for the application. The stack i - Grafana 11.6.0 - Loki 3.4.2 - Jaeger 2.4.0 -- Otel Collector 0.120.0 #0.121.0 remove loki +- Otel Collector 0.120.0 # 0.121.0 remove loki ## Prometheus @@ -47,8 +47,16 @@ observability data formats (e.g. Jaeger, Prometheus, etc.) sending to one or mor To deploy the observability stack, run the following command: +- docker latest version + ```bash -docker-compose -f docker-compose.yml -f docker-compose.override.yml up -d +docker compose -f docker-compose.yml -f docker-compose.override.yml up -d +``` + +- docker compose v2.0.0 or before + +```bash +docke-compose -f docker-compose.yml -f docker-compose.override.yml up -d ``` To access the Grafana dashboard, navigate to `http://localhost:3000` in your browser. The default username and password @@ -63,7 +71,7 @@ To access the Prometheus dashboard, navigate to `http://localhost:9090` in your To stop the observability stack, run the following command: ```bash -docker-compose -f docker-compose.yml -f docker-compose.override.yml down +docker compose -f docker-compose.yml -f docker-compose.override.yml down ``` ## How to remove data @@ -71,7 +79,7 @@ docker-compose -f docker-compose.yml -f docker-compose.override.yml down To remove the data generated by the observability stack, run the following command: ```bash -docker-compose -f docker-compose.yml -f docker-compose.override.yml down -v +docker compose -f docker-compose.yml -f docker-compose.override.yml down -v ``` ## How to configure diff --git a/deploy/README.md b/deploy/README.md index 4d3d4476a..2efdd85ec 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -23,13 +23,21 @@ managing and monitoring the system. | |--rustfs.service // systemd service file | |--rustfs-zh.service.md // systemd service file in Chinese |--certs -| |--README.md // certs readme -| |--rustfs_tls_cert.pem // API cert.pem -| |--rustfs_tls_key.pem // API key.pem -| |--rustfs_console_tls_cert.pem // console cert.pem -| |--rustfs_console_tls_key.pem // console key.pem +| ├── rustfs_cert.pem // Default｜fallback certificate +| ├── rustfs_key.pem // Default｜fallback private key +| ├── example.com/ // certificate directory of specific domain names +| │ ├── rustfs_cert.pem +| │ └── rustfs_key.pem +| ├── api.example.com/ +| │ ├── rustfs_cert.pem +| │ └── rustfs_key.pem +| └── cdn.example.com/ +| ├── rustfs_cert.pem +| └── rustfs_key.pem |--config | |--obs.example.yaml // example config | |--rustfs.env // env config | |--rustfs-zh.env // env config in Chinese +| |--.example.obs.env // example env config +| |--event.example.toml // event config ``` \ No newline at end of file diff --git a/deploy/build/rustfs-zh.service b/deploy/build/rustfs-zh.service index a1cd0df05..17351e675 100644 --- a/deploy/build/rustfs-zh.service +++ b/deploy/build/rustfs-zh.service @@ -51,6 +51,10 @@ ExecStart=/usr/local/bin/rustfs \ EnvironmentFile=-/etc/default/rustfs ExecStart=/usr/local/bin/rustfs $RUSTFS_VOLUMES $RUSTFS_OPTS +# standard output and error log configuration +StandardOutput=append:/data/deploy/rust/logs/rustfs.log +StandardError=append:/data/deploy/rust/logs/rustfs-err.log + # resource constraints LimitNOFILE=1048576 # 设置文件描述符上限为 1048576，支持高并发连接。 diff --git a/deploy/build/rustfs.service b/deploy/build/rustfs.service index df6e4067b..9c72e4276 100644 --- a/deploy/build/rustfs.service +++ b/deploy/build/rustfs.service @@ -31,6 +31,10 @@ ExecStart=/usr/local/bin/rustfs \ EnvironmentFile=-/etc/default/rustfs ExecStart=/usr/local/bin/rustfs $RUSTFS_VOLUMES $RUSTFS_OPTS +# service log configuration +StandardOutput=append:/data/deploy/rust/logs/rustfs.log +StandardError=append:/data/deploy/rust/logs/rustfs-err.log + # resource constraints LimitNOFILE=1048576 LimitNPROC=32768 diff --git a/deploy/certs/README.md b/deploy/certs/README.md index 84b733e79..e36d188b4 100644 --- a/deploy/certs/README.md +++ b/deploy/certs/README.md @@ -32,7 +32,17 @@ openssl req -x509 -newkey rsa:2048 -keyout key.pem -out cert.pem -days 365 -node ### TLS File ```text - rustfs_public.crt #api cert.pem - - rustfs_private.key #api key.pem +cd deploy/certs/ +ls -la + ├── rustfs_cert.pem // Default｜fallback certificate + ├── rustfs_key.pem // Default｜fallback private key + ├── example.com/ // certificate directory of specific domain names + │ ├── rustfs_cert.pem + │ └── rustfs_key.pem + ├── api.example.com/ + │ ├── rustfs_cert.pem + │ └── rustfs_key.pem + └── cdn.example.com/ + ├── rustfs_cert.pem + └── rustfs_key.pem ``` \ No newline at end of file diff --git a/deploy/config/.example.env b/deploy/config/.example.obs.env similarity index 100% rename from deploy/config/.example.env rename to deploy/config/.example.obs.env From a6e3561f836f29c586b5b4d8866979d645b6c5d8 Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 28 Apr 2025 14:51:51 +0800 Subject: [PATCH 9/9] improve code for readme.md add chinese readme.md --- .docker/observability/README_ZH.md | 42 +++++++++ README.md | 141 ++++++++++++++++------------- README_ZH.md | 136 ++++++++++++++++++++++++++++ 3 files changed, 255 insertions(+), 64 deletions(-) create mode 100644 .docker/observability/README_ZH.md create mode 100644 README_ZH.md diff --git a/.docker/observability/README_ZH.md b/.docker/observability/README_ZH.md new file mode 100644 index 000000000..7ba5342bf --- /dev/null +++ b/.docker/observability/README_ZH.md @@ -0,0 +1,42 @@ +## 部署可观测性系统 + +OpenTelemetry Collector 提供了一个厂商中立的遥测数据处理方案，用于接收、处理和导出遥测数据。它消除了为支持多种开源可观测性数据格式（如 +Jaeger、Prometheus 等）而需要运行和维护多个代理/收集器的必要性。 + +### 快速部署 + +1. 进入 `.docker/observability` 目录 +2. 执行以下命令启动服务： + +```bash +docker compose up -d -f docker-compose.yml +``` + +### 访问监控面板 + +服务启动后，可通过以下地址访问各个监控面板： + +- Grafana: `http://localhost:3000` (默认账号/密码：`admin`/`admin`) +- Jaeger: `http://localhost:16686` +- Prometheus: `http://localhost:9090` + +## 配置可观测性 + +### 创建配置文件 + +1. 进入 `deploy/config` 目录 +2. 复制示例配置：`cp obs.toml.example obs.toml` +3. 编辑 `obs.toml` 配置文件，修改以下关键参数： + +| 配置项 | 说明 | 示例值 | +|-----------------|----------------------------|-----------------------| +| endpoint | OpenTelemetry Collector 地址 | http://localhost:4317 | +| service_name | 服务名称 | rustfs | +| service_version | 服务版本 | 1.0.0 | +| environment | 运行环境 | production | +| meter_interval | 指标导出间隔 (秒) | 30 | +| sample_ratio | 采样率 | 1.0 | +| use_stdout | 是否输出到控制台 | true/false | +| logger_level | 日志级别 | info | + +``` \ No newline at end of file diff --git a/README.md b/README.md index 812f13a3c..79d9a342c 100644 --- a/README.md +++ b/README.md @@ -1,54 +1,68 @@ -# How to compile RustFS +# RustFS -| Must package | Version | download link | -|--------------|---------|----------------------------------------------------------------------------------------------------------------------------------| -| Rust | 1.8.5 | https://www.rust-lang.org/tools/install | -| protoc | 30.2 | [protoc-30.2-linux-x86_64.zip](https://github.com/protocolbuffers/protobuf/releases/download/v30.2/protoc-30.2-linux-x86_64.zip) | -| flatc | 24.0+ | [Linux.flatc.binary.g++-13.zip](https://github.com/google/flatbuffers/releases/download/v25.2.10/Linux.flatc.binary.g++-13.zip) | +## English Documentation |[中文文档](README_ZH.md) -Download Links: +### Prerequisites -https://github.com/google/flatbuffers/releases/download/v25.2.10/Linux.flatc.binary.g++-13.zip +| Package | Version | Download Link | +|---------|---------|----------------------------------------------------------------------------------------------------------------------------------| +| Rust | 1.8.5+ | [rust-lang.org/tools/install](https://www.rust-lang.org/tools/install) | +| protoc | 30.2+ | [protoc-30.2-linux-x86_64.zip](https://github.com/protocolbuffers/protobuf/releases/download/v30.2/protoc-30.2-linux-x86_64.zip) | +| flatc | 24.0+ | [Linux.flatc.binary.g++-13.zip](https://github.com/google/flatbuffers/releases/download/v25.2.10/Linux.flatc.binary.g++-13.zip) | -https://github.com/protocolbuffers/protobuf/releases/download/v30.2/protoc-30.2-linux-x86_64.zip +### Building RustFS -generate protobuf code: +#### Generate Protobuf Code -```cargo run --bin gproto``` +```bash +cargo run --bin gproto +``` -Or use Docker: +#### Using Docker for Prerequisites -```yml +```yaml - uses: arduino/setup-protoc@v3 with: - version: "30.2" + version: "30.2" - uses: Nugine/setup-flatc@v1 with: - version: "25.2.10" + version: "25.2.10" ``` -# How to add Console web +#### Adding Console Web UI -1. `wget https://dl.rustfs.com/artifacts/console/rustfs-console-latest.zip` +1. Download the latest console UI: + ```bash + wget https://dl.rustfs.com/artifacts/console/rustfs-console-latest.zip + ``` +2. Create the static directory: + ```bash + mkdir -p ./rustfs/static + ``` +3. Extract and compile RustFS: + ```bash + unzip rustfs-console-latest.zip -d ./rustfs/static + cargo build + ``` -2. mkdir in this repos folder `./rustfs/static` +### Running RustFS -3. Compile RustFS +#### Configuration -# Star RustFS +Set the required environment variables: -Add Env Information: - -``` +```bash +# Basic config export RUSTFS_VOLUMES="./target/volume/test" export RUSTFS_ADDRESS="0.0.0.0:9000" export RUSTFS_CONSOLE_ENABLE=true export RUSTFS_CONSOLE_ADDRESS="0.0.0.0:9001" -# 具体路径修改为配置文件真实路径，obs.example.toml 仅供参考 其中`RUSTFS_OBS_CONFIG` 和下面变量二选一 -export RUSTFS_OBS_CONFIG="./deploy/config/obs.example.toml" -# 如下变量需要必须参数都有值才可以，以及会覆盖配置文件`obs.example.toml`中的值 +# Observability config (option 1: config file) +export RUSTFS_OBS_CONFIG="./deploy/config/obs.toml" + +# Observability config (option 2: environment variables) export RUSTFS__OBSERVABILITY__ENDPOINT=http://localhost:4317 export RUSTFS__OBSERVABILITY__USE_STDOUT=true export RUSTFS__OBSERVABILITY__SAMPLE_RATIO=2.0 @@ -57,6 +71,9 @@ export RUSTFS__OBSERVABILITY__SERVICE_NAME=rustfs export RUSTFS__OBSERVABILITY__SERVICE_VERSION=0.1.0 export RUSTFS__OBSERVABILITY__ENVIRONMENT=develop export RUSTFS__OBSERVABILITY__LOGGER_LEVEL=info +export RUSTFS__OBSERVABILITY__LOCAL_LOGGING_ENABLED=true + +# Logging sinks export RUSTFS__SINKS__FILE__ENABLED=true export RUSTFS__SINKS__FILE__PATH="./deploy/logs/rustfs.log" export RUSTFS__SINKS__WEBHOOK__ENABLED=false @@ -68,54 +85,50 @@ export RUSTFS__SINKS__KAFKA__TOPIC="" export RUSTFS__LOGGER__QUEUE_CAPACITY=10 ``` -You need replace your real data folder: +#### Start the service -``` +```bash ./rustfs /data/rustfs ``` -## How to deploy the observability stack +### Observability Stack -The OpenTelemetry Collector offers a vendor-agnostic implementation on how to receive, process, and export telemetry -data. It removes the need to run, operate, and maintain multiple agents/collectors in order to support open-source -observability data formats (e.g. Jaeger, Prometheus, etc.) sending to one or more open-source or commercial back-ends. +#### Deployment -1. Enter the `.docker/observability` directory, -2. Run the following command: +1. Navigate to the observability directory: + ```bash + cd .docker/observability + ``` -```bash -docker compose -f docker-compose.yml up -d -``` +2. Start the observability stack: + ```bash + docker compose up -d -f docker-compose.yml + ``` -3. Access the Grafana dashboard by navigating to `http://localhost:3000` in your browser. The default username and - password are `admin` and `admin`, respectively. +#### Access Monitoring Dashboards -4. Access the Jaeger dashboard by navigating to `http://localhost:16686` in your browser. +- Grafana: `http://localhost:3000` (credentials: `admin`/`admin`) +- Jaeger: `http://localhost:16686` +- Prometheus: `http://localhost:9090` -5. Access the Prometheus dashboard by navigating to `http://localhost:9090` in your browser. +#### Configuring Observability -## Create a new Observability configuration file - -#### 1. Enter the `deploy/config` directory, - -#### 2. Copy `obs.toml.example` to `obs.toml` - -#### 3. Modify the `obs.toml` configuration file - -##### 3.1. Modify the `endpoint` value to the address of the OpenTelemetry Collector - -##### 3.2. Modify the `service_name` value to the name of the service - -##### 3.3. Modify the `service_version` value to the version of the service - -##### 3.4. Modify the `environment` value to the environment of the service - -##### 3.5. Modify the `meter_interval` value to export interval - -##### 3.6. Modify the `sample_ratio` value to the sample ratio - -##### 3.7. Modify the `use_stdout` value to export to stdout - -##### 3.8. Modify the `logger_level` value to the logger level +1. Copy the example configuration: + ```bash + cd deploy/config + cp obs.toml.example obs.toml + ``` +2. Edit `obs.toml` with the following parameters: +| Parameter | Description | Example | +|----------------------|-----------------------------------|-----------------------| +| endpoint | OpenTelemetry Collector address | http://localhost:4317 | +| service_name | Service name | rustfs | +| service_version | Service version | 1.0.0 | +| environment | Runtime environment | production | +| meter_interval | Metrics export interval (seconds) | 30 | +| sample_ratio | Sampling ratio | 1.0 | +| use_stdout | Output to console | true/false | +| logger_level | Log level | info | +| local_logging_enable | stdout | true/false | diff --git a/README_ZH.md b/README_ZH.md new file mode 100644 index 000000000..c2016fb7c --- /dev/null +++ b/README_ZH.md @@ -0,0 +1,136 @@ +# RustFS + +## [English Documentation](README.md) ｜中文文档 + +### 前置要求 + +| 软件包 | 版本 | 下载链接 | +|--------|--------|----------------------------------------------------------------------------------------------------------------------------------| +| Rust | 1.8.5+ | [rust-lang.org/tools/install](https://www.rust-lang.org/tools/install) | +| protoc | 30.2+ | [protoc-30.2-linux-x86_64.zip](https://github.com/protocolbuffers/protobuf/releases/download/v30.2/protoc-30.2-linux-x86_64.zip) | +| flatc | 24.0+ | [Linux.flatc.binary.g++-13.zip](https://github.com/google/flatbuffers/releases/download/v25.2.10/Linux.flatc.binary.g++-13.zip) | + +### 构建 RustFS + +#### 生成 Protobuf 代码 + +```bash +cargo run --bin gproto +``` + +#### 使用 Docker 安装依赖 + +```yaml +- uses: arduino/setup-protoc@v3 + with: + version: "30.2" + +- uses: Nugine/setup-flatc@v1 + with: + version: "25.2.10" +``` + +#### 添加控制台 Web UI + +1. 下载最新的控制台 UI： + ```bash + wget https://dl.rustfs.com/artifacts/console/rustfs-console-latest.zip + ``` +2. 创建静态资源目录： + ```bash + mkdir -p ./rustfs/static + ``` +3. 解压并编译 RustFS： + ```bash + unzip rustfs-console-latest.zip -d ./rustfs/static + cargo build + ``` + +### 运行 RustFS + +#### 配置 + +设置必要的环境变量： + +```bash +# 基础配置 +export RUSTFS_VOLUMES="./target/volume/test" +export RUSTFS_ADDRESS="0.0.0.0:9000" +export RUSTFS_CONSOLE_ENABLE=true +export RUSTFS_CONSOLE_ADDRESS="0.0.0.0:9001" + +# 可观测性配置（方式一：配置文件） +export RUSTFS_OBS_CONFIG="./deploy/config/obs.toml" + +# 可观测性配置（方式二：环境变量） +export RUSTFS__OBSERVABILITY__ENDPOINT=http://localhost:4317 +export RUSTFS__OBSERVABILITY__USE_STDOUT=true +export RUSTFS__OBSERVABILITY__SAMPLE_RATIO=2.0 +export RUSTFS__OBSERVABILITY__METER_INTERVAL=30 +export RUSTFS__OBSERVABILITY__SERVICE_NAME=rustfs +export RUSTFS__OBSERVABILITY__SERVICE_VERSION=0.1.0 +export RUSTFS__OBSERVABILITY__ENVIRONMENT=develop +export RUSTFS__OBSERVABILITY__LOGGER_LEVEL=info +export RUSTFS__OBSERVABILITY__LOCAL_LOGGING_ENABLED=true + +# 日志接收器 +export RUSTFS__SINKS__FILE__ENABLED=true +export RUSTFS__SINKS__FILE__PATH="./deploy/logs/rustfs.log" +export RUSTFS__SINKS__WEBHOOK__ENABLED=false +export RUSTFS__SINKS__WEBHOOK__ENDPOINT="" +export RUSTFS__SINKS__WEBHOOK__AUTH_TOKEN="" +export RUSTFS__SINKS__KAFKA__ENABLED=false +export RUSTFS__SINKS__KAFKA__BOOTSTRAP_SERVERS="" +export RUSTFS__SINKS__KAFKA__TOPIC="" +export RUSTFS__LOGGER__QUEUE_CAPACITY=10 +``` + +#### 启动服务 + +```bash +./rustfs /data/rustfs +``` + +### 可观测性系统 + +#### 部署 + +1. 进入可观测性目录： + ```bash + cd .docker/observability + ``` + +2. 启动可观测性系统： + ```bash + docker compose up -d -f docker-compose.yml + ``` + +#### 访问监控面板 + +- Grafana: `http://localhost:3000` (默认账号/密码：`admin`/`admin`) +- Jaeger: `http://localhost:16686` +- Prometheus: `http://localhost:9090` + +#### 配置可观测性 + +1. 复制示例配置： + ```bash + cd deploy/config + cp obs.toml.example obs.toml + ``` + +2. 编辑 `obs.toml` 配置文件，参数如下： + +| 配置项 | 说明 | 示例值 | +|----------------------|----------------------------|-----------------------| +| endpoint | OpenTelemetry Collector 地址 | http://localhost:4317 | +| service_name | 服务名称 | rustfs | +| service_version | 服务版本 | 1.0.0 | +| environment | 运行环境 | production | +| meter_interval | 指标导出间隔 (秒) | 30 | +| sample_ratio | 采样率 | 1.0 | +| use_stdout | 是否输出到控制台 | true/false | +| logger_level | 日志级别 | info | +| local_logging_enable | 控制台是否答应日志 | true/false | + +``` \ No newline at end of file