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/.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/.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/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/README.md b/README.md index 511f6736e..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 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..95bedfc4a 100644 --- a/crates/obs/src/telemetry.rs +++ b/crates/obs/src/telemetry.rs @@ -213,22 +213,29 @@ 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(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 !endpoint.is_empty() { info!( 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 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 接收器配置 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/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/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); 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/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/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/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/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( 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 1d1470d93..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" @@ -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 @@ -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"