From 25c6bdf490d4b26245530722a8debd0ff250ebc9 Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 19 May 2026 18:20:24 +0800 Subject: [PATCH] perf(filemeta): phase-1~3 rename_data metadata optimization (#3011) * chore(perf): harden amd64 profiling benchmark flow * fix(profiling): isolate bench buckets and map protobuf conflict * perf: avoid blocking owned local writes * style: format profile admin handler * docs: clarify observability trace validation * perf: reduce mkdir overhead on local writes * perf: add rename_data meta microbenchmark * perf(filemeta): fast-path data_dir decode in version meta * perf(filemeta): collapse data-dir lookup into one scan * perf(filemeta): reduce scan allocs and refresh meta bench * perf(ecstore): skip mkdir path on read-only open * perf(filemeta): single-pass unshared data-dir scan * perf(filemeta): add two-key inline remove fast path * perf(filemeta): compare remove-two keys by bytes first * bench(ecstore): add remove_two-only micro benchmark * bench(ecstore): stabilize rename_data meta benchmark timing * bench(ecstore): align rename_data path with remove_two * perf(filemeta): avoid uuid string alloc in remove_two * perf(filemeta): add fast-path for empty inline data * perf(filemeta): streamline add_version match branch * perf(filemeta): fast-return remove_key on miss * perf(filemeta): speed up add_version insertion lookup * style(ecstore): normalize formatting in perf-tuning files * refactor(filemeta): unify inline data removal paths --- .docker/compose/README.md | 40 +++ ...se.cluster.local-build.profiling-amd64.yml | 236 ++++++++++++++ .docker/observability/README.md | 48 +++ .docker/observability/README_ZH.md | 44 +++ crates/ecstore/Cargo.toml | 4 + .../benches/rename_data_meta_benchmark.rs | 118 +++++++ crates/ecstore/src/disk/local.rs | 56 ++-- crates/ecstore/src/disk/os.rs | 31 +- crates/filemeta/src/filemeta.rs | 55 ++-- crates/filemeta/src/filemeta/inline_data.rs | 107 +++++++ crates/filemeta/src/filemeta/version.rs | 101 ++++++ crates/filemeta/src/filemeta_inline.rs | 292 ++++++++++++------ rustfs/src/admin/handlers/profile_admin.rs | 40 ++- rustfs/src/config/cli.rs | 5 +- .../run_four_node_cluster_failover_bench.sh | 33 +- scripts/run_object_batch_bench.sh | 26 ++ 16 files changed, 1059 insertions(+), 177 deletions(-) create mode 100644 .docker/compose/docker-compose.cluster.local-build.profiling-amd64.yml create mode 100644 crates/ecstore/benches/rename_data_meta_benchmark.rs diff --git a/.docker/compose/README.md b/.docker/compose/README.md index 1065c46e4..8ca772500 100644 --- a/.docker/compose/README.md +++ b/.docker/compose/README.md @@ -54,6 +54,46 @@ Strict mode is available when you explicitly want `/health/ready == 200` as the WAIT_PROBE_MODE=ready ./scripts/run_four_node_cluster_failover_bench.sh ``` +### Profiling + Trace Validation + +The profiling-focused 4-node compose keeps profiling enabled and points RustFS +to an OTLP/HTTP collector endpoint: + +```bash +docker compose -f .docker/compose/docker-compose.cluster.local-build.profiling-amd64.yml up -d +``` + +Important behavior notes: + +- `RUSTFS_OBS_ENDPOINT` is the OTLP/HTTP base URL. RustFS automatically sends + traces to `/v1/traces`, metrics to `/v1/metrics`, and logs to `/v1/logs`. +- Startup usually produces logs and metrics first. That does not guarantee + visible traces yet. +- Trace data becomes obvious only after real HTTP/S3/gRPC requests hit RustFS. +- `RUSTFS_OBS_LOGGER_LEVEL=info` keeps the top-level request span but filters + many nested `debug` spans. If Tempo/Jaeger looks sparse, retry with + `RUSTFS_OBS_LOGGER_LEVEL=debug` before suspecting the collector. + +Minimal trace verification flow: + +```bash +# 1. Start the profiling compose with richer span visibility. +RUSTFS_OBS_LOGGER_LEVEL=debug \ +docker compose -f .docker/compose/docker-compose.cluster.local-build.profiling-amd64.yml up -d + +# 2. Generate real request traffic after startup. +curl -I http://127.0.0.1:9000/health +curl -I http://127.0.0.1:9000/health/ready + +# 3. Then inspect Tempo or Jaeger. +# Grafana: http://localhost:3000 +# Jaeger: http://localhost:16686 +``` + +If logs and metrics are present but traces are sparse, the most common cause is +"no real request traffic yet" or "`info` level filtered nested spans", not an +OTLP routing failure. + ### (Deprecated) Minimal Observability ```bash diff --git a/.docker/compose/docker-compose.cluster.local-build.profiling-amd64.yml b/.docker/compose/docker-compose.cluster.local-build.profiling-amd64.yml new file mode 100644 index 000000000..ecaf98874 --- /dev/null +++ b/.docker/compose/docker-compose.cluster.local-build.profiling-amd64.yml @@ -0,0 +1,236 @@ +# Copyright 2024 RustFS Team +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# Profiling-first 4-node local-build compose. +# +# Goals: +# - force linux/amd64 runtime/build on Apple Silicon hosts; +# - enable RustFS built-in CPU profiling; +# - keep all tuning knobs host-overridable via env. +# +# Observability notes: +# - `RUSTFS_OBS_ENDPOINT` is the OTLP/HTTP base URL. RustFS appends +# `/v1/traces`, `/v1/metrics`, and `/v1/logs` automatically. +# - Logs and metrics usually appear during startup. Traces mainly appear after +# real HTTP/S3/gRPC requests create spans. +# - `RUSTFS_OBS_LOGGER_LEVEL=info` keeps the top-level request trace span but +# filters many `debug`-level nested spans. Use `debug` when validating trace +# richness rather than collector reachability. + +services: + node1: + platform: ${RUSTFS_DOCKER_PLATFORM:-linux/amd64} + image: ${RUSTFS_IMAGE:-rustfs/rustfs:local-4node} + build: + context: ../.. + dockerfile: Dockerfile.source + hostname: node1 + environment: + - RUSTFS_VOLUMES=http://node{1...4}:9000/data/rustfs{0...3} + - RUSTFS_ADDRESS=:9000 + - RUSTFS_CONSOLE_ENABLE=true + - RUSTFS_ACCESS_KEY=${RUSTFS_ACCESS_KEY:-rustfs-cluster-admin} + - RUSTFS_SECRET_KEY=${RUSTFS_SECRET_KEY:-rustfs-cluster-secret} + - RUSTFS_OBS_ENDPOINT=${RUSTFS_OBS_ENDPOINT:-http://host.docker.internal:4318} + # `info` is enough for startup logs/metrics. Use `debug` if Tempo/Jaeger + # should show richer nested spans during request-path verification. + - RUSTFS_OBS_LOGGER_LEVEL=${RUSTFS_OBS_LOGGER_LEVEL:-info} + - RUSTFS_OBS_PROFILING_ENDPOINT=${RUSTFS_OBS_PROFILING_ENDPOINT:-http://host.docker.internal:4040} + - RUSTFS_OBS_PROFILING_EXPORT_ENABLED=${RUSTFS_OBS_PROFILING_EXPORT_ENABLED:-true} + - RUSTFS_UNSAFE_BYPASS_DISK_CHECK=${RUSTFS_UNSAFE_BYPASS_DISK_CHECK:-true} + - RUSTFS_ENABLE_PROFILING=${RUSTFS_ENABLE_PROFILING:-true} + - RUSTFS_PROF_CPU_MODE=${RUSTFS_PROF_CPU_MODE:-continuous} + - RUSTFS_PROF_CPU_FREQ=${RUSTFS_PROF_CPU_FREQ:-99} + - RUSTFS_PROF_OUTPUT_DIR=${RUSTFS_PROF_OUTPUT_DIR:-/tmp/rustfs-profiles} + - RUSTFS_OBJECT_MAX_CONCURRENT_DISK_READS=${RUSTFS_OBJECT_MAX_CONCURRENT_DISK_READS:-48} + - RUSTFS_OBJECT_IO_BUFFER_SIZE=${RUSTFS_OBJECT_IO_BUFFER_SIZE:-262144} + - RUSTFS_OBJECT_MEDIUM_CONCURRENCY_THRESHOLD=${RUSTFS_OBJECT_MEDIUM_CONCURRENCY_THRESHOLD:-6} + - RUSTFS_OBJECT_HIGH_CONCURRENCY_THRESHOLD=${RUSTFS_OBJECT_HIGH_CONCURRENCY_THRESHOLD:-12} + - RUSTFS_OBJECT_IO_RANDOM_READAHEAD_DISABLE_CONCURRENCY=${RUSTFS_OBJECT_IO_RANDOM_READAHEAD_DISABLE_CONCURRENCY:-8} + - RUSTFS_OBJECT_DUPLEX_BUFFER_SIZE=${RUSTFS_OBJECT_DUPLEX_BUFFER_SIZE:-8388608} + - RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES=${RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES:-25165824} + - RUSTFS_RUNTIME_WORKER_THREADS=${RUSTFS_RUNTIME_WORKER_THREADS:-12} + - RUSTFS_RUNTIME_MAX_BLOCKING_THREADS=${RUSTFS_RUNTIME_MAX_BLOCKING_THREADS:-512} + - RUSTFS_ALLOCATOR_RECLAIM_ENABLED=${RUSTFS_ALLOCATOR_RECLAIM_ENABLED:-false} + extra_hosts: + - "host.docker.internal:host-gateway" + volumes: + - node1_data_0:/data/rustfs0 + - node1_data_1:/data/rustfs1 + - node1_data_2:/data/rustfs2 + - node1_data_3:/data/rustfs3 + ports: + - "9000:9000" + networks: + - rustfs-cluster-net + + node2: + platform: ${RUSTFS_DOCKER_PLATFORM:-linux/amd64} + image: ${RUSTFS_IMAGE:-rustfs/rustfs:local-4node} + build: + context: ../.. + dockerfile: Dockerfile.source + hostname: node2 + environment: + - RUSTFS_VOLUMES=http://node{1...4}:9000/data/rustfs{0...3} + - RUSTFS_ADDRESS=:9000 + - RUSTFS_CONSOLE_ENABLE=true + - RUSTFS_ACCESS_KEY=${RUSTFS_ACCESS_KEY:-rustfs-cluster-admin} + - RUSTFS_SECRET_KEY=${RUSTFS_SECRET_KEY:-rustfs-cluster-secret} + - RUSTFS_OBS_ENDPOINT=${RUSTFS_OBS_ENDPOINT:-http://host.docker.internal:4318} + # `info` is enough for startup logs/metrics. Use `debug` if Tempo/Jaeger + # should show richer nested spans during request-path verification. + - RUSTFS_OBS_LOGGER_LEVEL=${RUSTFS_OBS_LOGGER_LEVEL:-info} + - RUSTFS_OBS_PROFILING_ENDPOINT=${RUSTFS_OBS_PROFILING_ENDPOINT:-http://host.docker.internal:4040} + - RUSTFS_OBS_PROFILING_EXPORT_ENABLED=${RUSTFS_OBS_PROFILING_EXPORT_ENABLED:-true} + - RUSTFS_UNSAFE_BYPASS_DISK_CHECK=${RUSTFS_UNSAFE_BYPASS_DISK_CHECK:-true} + - RUSTFS_ENABLE_PROFILING=${RUSTFS_ENABLE_PROFILING:-true} + - RUSTFS_PROF_CPU_MODE=${RUSTFS_PROF_CPU_MODE:-continuous} + - RUSTFS_PROF_CPU_FREQ=${RUSTFS_PROF_CPU_FREQ:-99} + - RUSTFS_PROF_OUTPUT_DIR=${RUSTFS_PROF_OUTPUT_DIR:-/tmp/rustfs-profiles} + - RUSTFS_OBJECT_MAX_CONCURRENT_DISK_READS=${RUSTFS_OBJECT_MAX_CONCURRENT_DISK_READS:-48} + - RUSTFS_OBJECT_IO_BUFFER_SIZE=${RUSTFS_OBJECT_IO_BUFFER_SIZE:-262144} + - RUSTFS_OBJECT_MEDIUM_CONCURRENCY_THRESHOLD=${RUSTFS_OBJECT_MEDIUM_CONCURRENCY_THRESHOLD:-6} + - RUSTFS_OBJECT_HIGH_CONCURRENCY_THRESHOLD=${RUSTFS_OBJECT_HIGH_CONCURRENCY_THRESHOLD:-12} + - RUSTFS_OBJECT_IO_RANDOM_READAHEAD_DISABLE_CONCURRENCY=${RUSTFS_OBJECT_IO_RANDOM_READAHEAD_DISABLE_CONCURRENCY:-8} + - RUSTFS_OBJECT_DUPLEX_BUFFER_SIZE=${RUSTFS_OBJECT_DUPLEX_BUFFER_SIZE:-8388608} + - RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES=${RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES:-25165824} + - RUSTFS_RUNTIME_WORKER_THREADS=${RUSTFS_RUNTIME_WORKER_THREADS:-12} + - RUSTFS_RUNTIME_MAX_BLOCKING_THREADS=${RUSTFS_RUNTIME_MAX_BLOCKING_THREADS:-512} + - RUSTFS_ALLOCATOR_RECLAIM_ENABLED=${RUSTFS_ALLOCATOR_RECLAIM_ENABLED:-false} + extra_hosts: + - "host.docker.internal:host-gateway" + volumes: + - node2_data_0:/data/rustfs0 + - node2_data_1:/data/rustfs1 + - node2_data_2:/data/rustfs2 + - node2_data_3:/data/rustfs3 + ports: + - "9001:9000" + networks: + - rustfs-cluster-net + + node3: + platform: ${RUSTFS_DOCKER_PLATFORM:-linux/amd64} + image: ${RUSTFS_IMAGE:-rustfs/rustfs:local-4node} + build: + context: ../.. + dockerfile: Dockerfile.source + hostname: node3 + environment: + - RUSTFS_VOLUMES=http://node{1...4}:9000/data/rustfs{0...3} + - RUSTFS_ADDRESS=:9000 + - RUSTFS_CONSOLE_ENABLE=true + - RUSTFS_ACCESS_KEY=${RUSTFS_ACCESS_KEY:-rustfs-cluster-admin} + - RUSTFS_SECRET_KEY=${RUSTFS_SECRET_KEY:-rustfs-cluster-secret} + - RUSTFS_OBS_ENDPOINT=${RUSTFS_OBS_ENDPOINT:-http://host.docker.internal:4318} + # `info` is enough for startup logs/metrics. Use `debug` if Tempo/Jaeger + # should show richer nested spans during request-path verification. + - RUSTFS_OBS_LOGGER_LEVEL=${RUSTFS_OBS_LOGGER_LEVEL:-info} + - RUSTFS_OBS_PROFILING_ENDPOINT=${RUSTFS_OBS_PROFILING_ENDPOINT:-http://host.docker.internal:4040} + - RUSTFS_OBS_PROFILING_EXPORT_ENABLED=${RUSTFS_OBS_PROFILING_EXPORT_ENABLED:-true} + - RUSTFS_UNSAFE_BYPASS_DISK_CHECK=${RUSTFS_UNSAFE_BYPASS_DISK_CHECK:-true} + - RUSTFS_ENABLE_PROFILING=${RUSTFS_ENABLE_PROFILING:-true} + - RUSTFS_PROF_CPU_MODE=${RUSTFS_PROF_CPU_MODE:-continuous} + - RUSTFS_PROF_CPU_FREQ=${RUSTFS_PROF_CPU_FREQ:-99} + - RUSTFS_PROF_OUTPUT_DIR=${RUSTFS_PROF_OUTPUT_DIR:-/tmp/rustfs-profiles} + - RUSTFS_OBJECT_MAX_CONCURRENT_DISK_READS=${RUSTFS_OBJECT_MAX_CONCURRENT_DISK_READS:-48} + - RUSTFS_OBJECT_IO_BUFFER_SIZE=${RUSTFS_OBJECT_IO_BUFFER_SIZE:-262144} + - RUSTFS_OBJECT_MEDIUM_CONCURRENCY_THRESHOLD=${RUSTFS_OBJECT_MEDIUM_CONCURRENCY_THRESHOLD:-6} + - RUSTFS_OBJECT_HIGH_CONCURRENCY_THRESHOLD=${RUSTFS_OBJECT_HIGH_CONCURRENCY_THRESHOLD:-12} + - RUSTFS_OBJECT_IO_RANDOM_READAHEAD_DISABLE_CONCURRENCY=${RUSTFS_OBJECT_IO_RANDOM_READAHEAD_DISABLE_CONCURRENCY:-8} + - RUSTFS_OBJECT_DUPLEX_BUFFER_SIZE=${RUSTFS_OBJECT_DUPLEX_BUFFER_SIZE:-8388608} + - RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES=${RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES:-25165824} + - RUSTFS_RUNTIME_WORKER_THREADS=${RUSTFS_RUNTIME_WORKER_THREADS:-12} + - RUSTFS_RUNTIME_MAX_BLOCKING_THREADS=${RUSTFS_RUNTIME_MAX_BLOCKING_THREADS:-512} + - RUSTFS_ALLOCATOR_RECLAIM_ENABLED=${RUSTFS_ALLOCATOR_RECLAIM_ENABLED:-false} + extra_hosts: + - "host.docker.internal:host-gateway" + volumes: + - node3_data_0:/data/rustfs0 + - node3_data_1:/data/rustfs1 + - node3_data_2:/data/rustfs2 + - node3_data_3:/data/rustfs3 + ports: + - "9002:9000" + networks: + - rustfs-cluster-net + + node4: + platform: ${RUSTFS_DOCKER_PLATFORM:-linux/amd64} + image: ${RUSTFS_IMAGE:-rustfs/rustfs:local-4node} + build: + context: ../.. + dockerfile: Dockerfile.source + hostname: node4 + environment: + - RUSTFS_VOLUMES=http://node{1...4}:9000/data/rustfs{0...3} + - RUSTFS_ADDRESS=:9000 + - RUSTFS_CONSOLE_ENABLE=true + - RUSTFS_ACCESS_KEY=${RUSTFS_ACCESS_KEY:-rustfs-cluster-admin} + - RUSTFS_SECRET_KEY=${RUSTFS_SECRET_KEY:-rustfs-cluster-secret} + - RUSTFS_OBS_ENDPOINT=${RUSTFS_OBS_ENDPOINT:-http://host.docker.internal:4318} + # `info` is enough for startup logs/metrics. Use `debug` if Tempo/Jaeger + # should show richer nested spans during request-path verification. + - RUSTFS_OBS_LOGGER_LEVEL=${RUSTFS_OBS_LOGGER_LEVEL:-info} + - RUSTFS_OBS_PROFILING_ENDPOINT=${RUSTFS_OBS_PROFILING_ENDPOINT:-http://host.docker.internal:4040} + - RUSTFS_OBS_PROFILING_EXPORT_ENABLED=${RUSTFS_OBS_PROFILING_EXPORT_ENABLED:-true} + - RUSTFS_UNSAFE_BYPASS_DISK_CHECK=${RUSTFS_UNSAFE_BYPASS_DISK_CHECK:-true} + - RUSTFS_ENABLE_PROFILING=${RUSTFS_ENABLE_PROFILING:-true} + - RUSTFS_PROF_CPU_MODE=${RUSTFS_PROF_CPU_MODE:-continuous} + - RUSTFS_PROF_CPU_FREQ=${RUSTFS_PROF_CPU_FREQ:-99} + - RUSTFS_PROF_OUTPUT_DIR=${RUSTFS_PROF_OUTPUT_DIR:-/tmp/rustfs-profiles} + - RUSTFS_OBJECT_MAX_CONCURRENT_DISK_READS=${RUSTFS_OBJECT_MAX_CONCURRENT_DISK_READS:-48} + - RUSTFS_OBJECT_IO_BUFFER_SIZE=${RUSTFS_OBJECT_IO_BUFFER_SIZE:-262144} + - RUSTFS_OBJECT_MEDIUM_CONCURRENCY_THRESHOLD=${RUSTFS_OBJECT_MEDIUM_CONCURRENCY_THRESHOLD:-6} + - RUSTFS_OBJECT_HIGH_CONCURRENCY_THRESHOLD=${RUSTFS_OBJECT_HIGH_CONCURRENCY_THRESHOLD:-12} + - RUSTFS_OBJECT_IO_RANDOM_READAHEAD_DISABLE_CONCURRENCY=${RUSTFS_OBJECT_IO_RANDOM_READAHEAD_DISABLE_CONCURRENCY:-8} + - RUSTFS_OBJECT_DUPLEX_BUFFER_SIZE=${RUSTFS_OBJECT_DUPLEX_BUFFER_SIZE:-8388608} + - RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES=${RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES:-25165824} + - RUSTFS_RUNTIME_WORKER_THREADS=${RUSTFS_RUNTIME_WORKER_THREADS:-12} + - RUSTFS_RUNTIME_MAX_BLOCKING_THREADS=${RUSTFS_RUNTIME_MAX_BLOCKING_THREADS:-512} + - RUSTFS_ALLOCATOR_RECLAIM_ENABLED=${RUSTFS_ALLOCATOR_RECLAIM_ENABLED:-false} + extra_hosts: + - "host.docker.internal:host-gateway" + volumes: + - node4_data_0:/data/rustfs0 + - node4_data_1:/data/rustfs1 + - node4_data_2:/data/rustfs2 + - node4_data_3:/data/rustfs3 + ports: + - "9003:9000" + networks: + - rustfs-cluster-net + +volumes: + node1_data_0: + node1_data_1: + node1_data_2: + node1_data_3: + node2_data_0: + node2_data_1: + node2_data_2: + node2_data_3: + node3_data_0: + node3_data_1: + node3_data_2: + node3_data_3: + node4_data_0: + node4_data_1: + node4_data_2: + node4_data_3: + +networks: + rustfs-cluster-net: + driver: bridge diff --git a/.docker/observability/README.md b/.docker/observability/README.md index daa5f6888..81639e418 100644 --- a/.docker/observability/README.md +++ b/.docker/observability/README.md @@ -90,6 +90,54 @@ docker compose down -v - **Grafana**: Dashboards and datasources are provisioned from the `grafana/` directory. - **Collector**: Edit `otel-collector-config.yaml` to modify pipelines, processors, or exporters. +### Verifying RustFS Traces + +When RustFS points `RUSTFS_OBS_ENDPOINT` at this stack, treat the value as the +OTLP/HTTP base URL, for example: + +```bash +export RUSTFS_OBS_ENDPOINT=http://host.docker.internal:4318 +``` + +RustFS automatically expands that base URL to: + +- `/v1/traces` +- `/v1/metrics` +- `/v1/logs` + +Important behavior notes: + +- Logs and metrics usually appear during startup, so seeing those two signals + first is expected. +- Visible trace data usually requires real HTTP/S3/gRPC request traffic after + startup, because request-path spans are created on demand. +- `RUSTFS_OBS_LOGGER_LEVEL=info` keeps the top-level request span but filters + many nested `debug` spans. If Tempo or Jaeger looks sparse, retry with + `RUSTFS_OBS_LOGGER_LEVEL=debug` before suspecting collector or Tempo issues. + +Minimal validation flow: + +```bash +# 1. Start this observability stack. +docker compose up -d + +# 2. Start RustFS with OTLP/HTTP export and richer span visibility. +export RUSTFS_OBS_ENDPOINT=http://host.docker.internal:4318 +export RUSTFS_OBS_LOGGER_LEVEL=debug + +# 3. Generate real request traffic. +curl -I http://127.0.0.1:9000/health +curl -I http://127.0.0.1:9000/health/ready + +# 4. Inspect Grafana or Jaeger. +# Grafana: http://localhost:3000 +# Jaeger: http://localhost:16686 +``` + +If logs and metrics are present but traces are sparse, the most common cause is +"no real request traffic yet" or "`info` level filtered nested spans", not an +OTLP routing failure. + ## Troubleshooting - **Service Health**: Check the health of services using `docker compose ps`. diff --git a/.docker/observability/README_ZH.md b/.docker/observability/README_ZH.md index ff3e41403..f38dcf9cf 100644 --- a/.docker/observability/README_ZH.md +++ b/.docker/observability/README_ZH.md @@ -90,6 +90,50 @@ docker compose down -v - **Grafana**: 仪表盘和数据源从 `grafana/` 目录预置。 - **Collector**: 编辑 `otel-collector-config.yaml` 以修改管道、处理器或导出器。 +### 验证 RustFS Trace + +当 RustFS 将 `RUSTFS_OBS_ENDPOINT` 指向这套技术栈时，应将该值视为 +OTLP/HTTP 的基础 URL，例如： + +```bash +export RUSTFS_OBS_ENDPOINT=http://host.docker.internal:4318 +``` + +RustFS 会自动在该基础 URL 后补全： + +- `/v1/traces` +- `/v1/metrics` +- `/v1/logs` + +需要注意： + +- 启动阶段通常会先看到日志和指标，因此“先有日志/指标、后有 trace”是正常现象。 +- 可见的 trace 数据通常依赖启动后的真实 HTTP/S3/gRPC 请求流量，因为请求路径上的 span 是按需创建的。 +- `RUSTFS_OBS_LOGGER_LEVEL=info` 会保留顶层请求 span，但会过滤掉很多 `debug` 级别的嵌套 span。 + 如果 Tempo 或 Jaeger 中的 trace 看起来很稀疏，建议先改成 `RUSTFS_OBS_LOGGER_LEVEL=debug`，再判断是否是 collector 或 Tempo 问题。 + +最小验证流程： + +```bash +# 1. 启动本目录下的可观测性技术栈。 +docker compose up -d + +# 2. 以 OTLP/HTTP 导出方式启动 RustFS，并提高 span 可见性。 +export RUSTFS_OBS_ENDPOINT=http://host.docker.internal:4318 +export RUSTFS_OBS_LOGGER_LEVEL=debug + +# 3. 产生真实请求流量。 +curl -I http://127.0.0.1:9000/health +curl -I http://127.0.0.1:9000/health/ready + +# 4. 到 Grafana 或 Jaeger 中检查。 +# Grafana: http://localhost:3000 +# Jaeger: http://localhost:16686 +``` + +如果日志和指标已经正常，但 trace 仍然稀疏，最常见的原因通常是 +“还没有真实请求流量”或“`info` 级别过滤了嵌套 span”，而不是 OTLP 路由失败。 + ## 故障排除 - **服务健康**: 使用 `docker compose ps` 检查服务健康状况。 diff --git a/crates/ecstore/Cargo.toml b/crates/ecstore/Cargo.toml index d438982cf..c91fa821d 100644 --- a/crates/ecstore/Cargo.toml +++ b/crates/ecstore/Cargo.toml @@ -145,5 +145,9 @@ harness = false name = "comparison_benchmark" harness = false +[[bench]] +name = "rename_data_meta_benchmark" +harness = false + [lib] doctest = false diff --git a/crates/ecstore/benches/rename_data_meta_benchmark.rs b/crates/ecstore/benches/rename_data_meta_benchmark.rs new file mode 100644 index 000000000..0be93c6a3 --- /dev/null +++ b/crates/ecstore/benches/rename_data_meta_benchmark.rs @@ -0,0 +1,118 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use criterion::{BatchSize, BenchmarkId, Criterion, criterion_group, criterion_main}; +use rustfs_filemeta::{ErasureAlgo, FileInfo, FileMeta}; +use std::hint::black_box; +use std::time::Duration; +use time::OffsetDateTime; +use uuid::Uuid; + +const VERSION_COUNT_CASES: &[usize] = &[1, 8, 32, 64]; +const BENCH_BASE_TIME_UNIX_SECS: i64 = 1_700_000_000; + +fn make_file_info(version_id: Uuid, data_dir: Uuid, size: i64, mod_time: OffsetDateTime) -> FileInfo { + FileInfo { + version_id: Some(version_id), + data_dir: Some(data_dir), + size, + mod_time: Some(mod_time), + metadata: [("etag".to_string(), format!("etag-{version_id}"))].into_iter().collect(), + erasure: rustfs_filemeta::ErasureInfo { + algorithm: ErasureAlgo::ReedSolomon.to_string(), + data_blocks: 4, + parity_blocks: 2, + block_size: 1024 * 1024, + index: 1, + distribution: vec![1, 2, 3, 4, 5, 6], + ..Default::default() + }, + ..Default::default() + } +} + +fn build_meta_with_versions(version_count: usize) -> FileMeta { + let mut meta = FileMeta::new(); + let base_time = OffsetDateTime::from_unix_timestamp(BENCH_BASE_TIME_UNIX_SECS).expect("valid bench base timestamp"); + for i in 0..version_count { + let fi = make_file_info(Uuid::new_v4(), Uuid::new_v4(), 64 * 1024, base_time - Duration::from_secs(i as u64)); + meta.add_version(fi).expect("seed add_version should succeed"); + } + meta +} + +fn bench_rename_data_meta_path(c: &mut Criterion) { + let mut group = c.benchmark_group("rename_data_meta"); + group.sample_size(20); + group.measurement_time(Duration::from_secs(10)); + + for &version_count in VERSION_COUNT_CASES { + let seeded = build_meta_with_versions(version_count); + let dst_buf = seeded.marshal_msg().expect("marshal seeded meta"); + let base_time = OffsetDateTime::from_unix_timestamp(BENCH_BASE_TIME_UNIX_SECS).expect("valid bench base timestamp"); + let replace_version_id = seeded + .versions + .first() + .and_then(|v| v.header.version_id) + .unwrap_or(Uuid::nil()); + + group.bench_with_input(BenchmarkId::new("read_modify_write", version_count), &version_count, |b, _| { + b.iter(|| { + let mut xlmeta = FileMeta::load(black_box(&dst_buf)).expect("load dst meta"); + let search_version_id = Some(replace_version_id); + let has_old_data_dir = xlmeta.find_unshared_data_dir_for_version(search_version_id); + if let Some(old_data_dir) = has_old_data_dir { + let _ = xlmeta.data.remove_two(replace_version_id, old_data_dir); + } + let fi = make_file_info(replace_version_id, Uuid::new_v4(), 64 * 1024, base_time + Duration::from_millis(1)); + xlmeta.add_version(fi).expect("add new version"); + let out = xlmeta.marshal_msg().expect("marshal updated meta"); + black_box(out); + }); + }); + + let mut prepared = FileMeta::load(&dst_buf).expect("load prepared meta"); + if let Some(old_data_dir) = prepared.find_unshared_data_dir_for_version(Some(replace_version_id)) { + let _ = prepared.data.remove_two(replace_version_id, old_data_dir); + } + group.bench_with_input(BenchmarkId::new("add_version_marshal_only", version_count), &version_count, |b, _| { + b.iter_batched( + || prepared.clone(), + |mut xlmeta| { + let fi = make_file_info(replace_version_id, Uuid::new_v4(), 64 * 1024, base_time + Duration::from_millis(1)); + xlmeta.add_version(fi).expect("add new version"); + let out = xlmeta.marshal_msg().expect("marshal updated meta"); + black_box(out); + }, + BatchSize::SmallInput, + ); + }); + + group.bench_with_input(BenchmarkId::new("remove_two_only", version_count), &version_count, |b, _| { + b.iter(|| { + let mut xlmeta = FileMeta::load(black_box(&dst_buf)).expect("load dst meta"); + let removed = if let Some(old_data_dir) = xlmeta.find_unshared_data_dir_for_version(Some(replace_version_id)) { + xlmeta.data.remove_two(replace_version_id, old_data_dir).expect("remove two") + } else { + false + }; + black_box(removed); + black_box(xlmeta); + }); + }); + } +} + +criterion_group!(benches, bench_rename_data_meta_path); +criterion_main!(benches); diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index a7baa18b3..a7567c6a4 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -1194,7 +1194,9 @@ impl LocalDisk { skip_parent = self.root.as_path(); } - if let Some(parent) = path.as_ref().parent() { + if let Some(parent) = path.as_ref().parent() + && parent != skip_parent + { os::make_dir_all(parent, skip_parent).await?; } @@ -1203,6 +1205,11 @@ impl LocalDisk { Ok(f) } + async fn open_file_read_only(&self, path: impl AsRef) -> Result { + let f = super::fs::open_file(path.as_ref(), O_RDONLY).await.map_err(to_file_error)?; + Ok(f) + } + #[allow(dead_code)] fn get_metrics(&self) -> DiskMetrics { DiskMetrics::default() @@ -2137,7 +2144,7 @@ impl DiskAPI for LocalDisk { let file_path = self.get_object_path(volume, path)?; check_path_length(file_path.to_string_lossy().as_ref())?; - let f = self.open_file(file_path, O_RDONLY, volume_dir).await?; + let f = self.open_file_read_only(file_path).await?; Ok(Box::new(f)) } @@ -2154,7 +2161,7 @@ impl DiskAPI for LocalDisk { let file_path = self.get_object_path(volume, path)?; check_path_length(file_path.to_string_lossy().as_ref())?; - let mut f = self.open_file(file_path, O_RDONLY, volume_dir).await?; + let mut f = self.open_file_read_only(file_path).await?; let meta = f.metadata().await?; let end_offset = offset.checked_add(length).ok_or(DiskError::FileCorrupt)?; @@ -2513,32 +2520,41 @@ impl DiskAPI for LocalDisk { // TODO: Healing - let search_version_id = fi.version_id.or(Some(Uuid::nil())); + let version_id = fi.version_id.unwrap_or_default(); + let search_version_id = Some(version_id); + let no_inline = fi.data.is_none() && fi.size > 0; // Check if there's an existing version with the same version_id that has a data_dir to clean up - let has_old_data_dir = { - xlmeta.find_version(search_version_id).ok().and_then(|(_, ver)| { - // shard_count == 0 means no other version shares this data_dir - ver.get_data_dir() - .filter(|&data_dir| xlmeta.shard_data_dir_count(&search_version_id, &Some(data_dir)) == 0) - }) - }; + // Reuse one metadata scan to find the version data_dir and determine whether it is shared. + let has_old_data_dir = xlmeta.find_unshared_data_dir_for_version(search_version_id); if let Some(old_data_dir) = has_old_data_dir.as_ref() { - let _ = xlmeta.data.remove(vec![search_version_id.unwrap_or_default(), *old_data_dir]); + let _ = xlmeta.data.remove_two(version_id, *old_data_dir); } - xlmeta.add_version(fi.clone())?; + xlmeta.add_version(fi)?; if xlmeta.versions.len() <= 10 { // TODO: Sign } - let new_dst_buf = xlmeta.marshal_msg()?; - - self.write_all(src_volume, format!("{}/{}", &src_path, STORAGE_FORMAT_FILE).as_str(), new_dst_buf.into()) - .await?; if let Some((src_data_path, dst_data_path)) = has_data_dir_path.as_ref() { - let no_inline = fi.data.is_none() && fi.size > 0; + let src_file_parent = src_file_path.parent().unwrap_or(src_volume_dir.as_path()); + let meta_skip_parent = if no_inline { + src_file_parent + } else { + src_volume_dir.as_path() + }; + let new_dst_buf = xlmeta.marshal_msg()?; + + self.write_all_private( + src_volume, + format!("{}/{}", &src_path, STORAGE_FORMAT_FILE).as_str(), + new_dst_buf.into(), + true, + meta_skip_parent, + ) + .await?; + if no_inline && let Err(err) = rename_all(&src_data_path, &dst_data_path, &skip_parent).await { let _ = self.delete_file(&dst_volume_dir, dst_data_path, false, false).await; info!( @@ -2547,6 +2563,10 @@ impl DiskAPI for LocalDisk { ); return Err(err); } + } else { + let new_dst_buf = xlmeta.marshal_msg()?; + self.write_all(src_volume, format!("{}/{}", &src_path, STORAGE_FORMAT_FILE).as_str(), new_dst_buf.into()) + .await?; } if let Some(old_data_dir) = has_old_data_dir { diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index 730f90e4d..ac6751429 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -215,24 +215,29 @@ pub async fn os_mkdir_all(dir_path: impl AsRef, base_dir: impl AsRef return Ok(()); } - if let Some(parent) = dir_path.as_ref().parent() { - // Without recursion support, fall back to create_dir_all - if let Err(e) = super::fs::make_dir_all(&parent).await { - if e.kind() == io::ErrorKind::AlreadyExists { - return Ok(()); - } - - return Err(e); - } - // Box::pin(os_mkdir_all(&parent, &base_dir)).await?; - } - if let Err(e) = super::fs::mkdir(dir_path.as_ref()).await { if e.kind() == io::ErrorKind::AlreadyExists { return Ok(()); } - return Err(e); + if e.kind() != io::ErrorKind::NotFound { + return Err(e); + } + + if let Some(parent) = dir_path.as_ref().parent() { + // Fall back to creating the missing parent chain only when the direct mkdir proves it is required. + if let Err(parent_err) = super::fs::make_dir_all(parent).await + && parent_err.kind() != io::ErrorKind::AlreadyExists + { + return Err(parent_err); + } + } + + if let Err(retry_err) = super::fs::mkdir(dir_path.as_ref()).await + && retry_err.kind() != io::ErrorKind::AlreadyExists + { + return Err(retry_err); + } } Ok(()) diff --git a/crates/filemeta/src/filemeta.rs b/crates/filemeta/src/filemeta.rs index f8be9abc4..e86aad93b 100644 --- a/crates/filemeta/src/filemeta.rs +++ b/crates/filemeta/src/filemeta.rs @@ -282,6 +282,11 @@ impl FileMeta { fi.version_id = Some(Uuid::nil()); } + if fi.data.is_none() && self.data.after_version().is_empty() { + let version = FileMetaVersion::from(fi); + return self.add_version_filemata(version); + } + let version_key = data_key_for_version(fi.version_id); let mut next_data = self.data.clone(); @@ -317,46 +322,30 @@ impl FileMeta { } let vid = version.get_version_id(); - - // Match existing version for replace; null version: None and Some(nil) are equivalent - let matches = |h: &Option| { - let v_null = vid.is_none() || vid == Some(Uuid::nil()); - let h_null = h.is_none() || *h == Some(Uuid::nil()); - (v_null && h_null) || (vid == *h) + let vid_is_null = vid.is_none() || vid == Some(Uuid::nil()); + let existing_idx = if vid_is_null { + self.versions + .iter() + .position(|v| v.header.version_id.is_none() || v.header.version_id == Some(Uuid::nil())) + } else { + self.versions.iter().position(|v| v.header.version_id == vid) }; - if let Some(fidx) = self.versions.iter().position(|v| matches(&v.header.version_id)) { + if let Some(fidx) = existing_idx { return self.set_idx(fidx, version); } - // append placeholder to find insert position - let placeholder = FileMetaShallowVersion { - header: FileMetaVersionHeader { - mod_time: None, // None sorts before any real mod_time - ..Default::default() - }, - meta: Vec::new(), - }; - self.versions.push(placeholder); - let mod_time = version.get_mod_time(); let new_shallow = FileMetaShallowVersion::try_from(version)?; - - for (idx, exist) in self.versions.iter().enumerate() { - let ex_mt = exist.header.mod_time; - let insert_here = match (ex_mt, mod_time) { - (None, _) => true, // placeholder: always insert before - (Some(em), Some(nm)) => em <= nm, - (Some(_), None) => false, - }; - if insert_here { - self.versions.insert(idx, new_shallow); - self.versions.pop(); // remove placeholder - return Ok(()); - } - } - self.versions.pop(); // remove placeholder on fallback - Err(Error::other("add_version failed")) + let insert_pos = match mod_time { + Some(nm) => self.versions.partition_point(|exist| match exist.header.mod_time { + Some(em) => em > nm, + None => false, + }), + None => self.versions.partition_point(|exist| exist.header.mod_time.is_some()), + }; + self.versions.insert(insert_pos, new_shallow); + Ok(()) // if !ver.valid() { // return Err(Error::other("attempted to add invalid version")); diff --git a/crates/filemeta/src/filemeta/inline_data.rs b/crates/filemeta/src/filemeta/inline_data.rs index 9f57e1fb0..92e774fc2 100644 --- a/crates/filemeta/src/filemeta/inline_data.rs +++ b/crates/filemeta/src/filemeta/inline_data.rs @@ -13,8 +13,48 @@ // limitations under the License. use super::*; +use std::collections::HashSet; impl FileMeta { + pub fn find_unshared_data_dir_for_version(&self, version_id: Option) -> Option { + let vid = version_id.unwrap_or_default(); + let mut target_data_dir = None; + let mut target_selected = false; + let mut other_data_dirs = HashSet::new(); + + for version in self + .versions + .iter() + .filter(|v| v.header.version_type == VersionType::Object && v.header.uses_data_dir()) + { + let is_target_version = version.header.version_id.unwrap_or_default() == vid; + if is_target_version { + if target_selected { + continue; + } + + target_selected = true; + target_data_dir = FileMetaVersion::decode_data_dir_from_meta(&version.meta).unwrap_or_default(); + if let Some(dir) = target_data_dir + && other_data_dirs.contains(&dir) + { + return None; + } + continue; + } + + let dir = FileMetaVersion::decode_data_dir_from_meta(&version.meta).unwrap_or_default(); + if let Some(dir) = dir { + if target_data_dir == Some(dir) { + return None; + } + other_data_dirs.insert(dir); + } + } + + target_data_dir + } + pub fn shard_data_dir_count(&self, vid: &Option, data_dir: &Option) -> usize { let vid = vid.unwrap_or_default(); self.versions @@ -58,3 +98,70 @@ impl FileMeta { .count() } } + +#[cfg(test)] +mod tests { + use super::*; + use s3s::header::X_AMZ_RESTORE; + use time::format_description::well_known::Rfc3339; + use time::{Duration, OffsetDateTime}; + + fn make_file_info(version_id: Uuid, data_dir: Uuid) -> FileInfo { + let restore_header = format!( + "ongoing-request=\"false\", expiry-date=\"{}\"", + (OffsetDateTime::now_utc() + Duration::days(1)) + .format(&Rfc3339) + .expect("format restore expiry"), + ); + FileInfo { + version_id: Some(version_id), + data_dir: Some(data_dir), + size: 64 * 1024, + mod_time: Some(OffsetDateTime::now_utc()), + metadata: [ + ("etag".to_string(), format!("etag-{version_id}")), + (X_AMZ_RESTORE.as_str().to_string(), restore_header), + ] + .into_iter() + .collect(), + erasure: ErasureInfo { + algorithm: ErasureAlgo::ReedSolomon.to_string(), + data_blocks: 4, + parity_blocks: 2, + block_size: 1024 * 1024, + index: 1, + distribution: vec![1, 2, 3, 4, 5, 6], + ..Default::default() + }, + ..Default::default() + } + } + + #[test] + fn find_unshared_data_dir_for_version_returns_data_dir_when_unique() { + let target_version = Uuid::new_v4(); + let target_data_dir = Uuid::new_v4(); + let mut meta = FileMeta::new(); + meta.add_version(make_file_info(target_version, target_data_dir)) + .expect("seed target version"); + meta.add_version(make_file_info(Uuid::new_v4(), Uuid::new_v4())) + .expect("seed non-shared version"); + + let got = meta.find_unshared_data_dir_for_version(Some(target_version)); + assert_eq!(got, Some(target_data_dir)); + } + + #[test] + fn find_unshared_data_dir_for_version_returns_none_when_shared() { + let target_version = Uuid::new_v4(); + let shared_data_dir = Uuid::new_v4(); + let mut meta = FileMeta::new(); + meta.add_version(make_file_info(target_version, shared_data_dir)) + .expect("seed target version"); + meta.add_version(make_file_info(Uuid::new_v4(), shared_data_dir)) + .expect("seed shared version"); + + let got = meta.find_unshared_data_dir_for_version(Some(target_version)); + assert_eq!(got, None); + } +} diff --git a/crates/filemeta/src/filemeta/version.rs b/crates/filemeta/src/filemeta/version.rs index aa3f7d151..525607bd7 100644 --- a/crates/filemeta/src/filemeta/version.rs +++ b/crates/filemeta/src/filemeta/version.rs @@ -312,6 +312,77 @@ pub struct FileMetaVersion { } impl FileMetaVersion { + fn decode_data_dir_from_v2_object(buf: &[u8]) -> Result> { + let mut cur = std::io::Cursor::new(buf); + let mut fields = rmp::decode::read_map_len(&mut cur)?; + let mut version_type = VersionType::Invalid; + + while fields > 0 { + fields -= 1; + + let key_len = rmp::decode::read_str_len(&mut cur)? as usize; + let mut key_buf = vec![0u8; key_len]; + cur.read_exact(&mut key_buf)?; + let key = String::from_utf8(key_buf)?; + + match key.as_str() { + "Type" => { + let v: i64 = rmp::decode::read_int(&mut cur)?; + version_type = VersionType::from_u8(v as u8); + } + "V2Obj" => { + if version_type != VersionType::Object { + skip_msgp_value(&mut cur)?; + continue; + } + + let mut first = [0u8; 1]; + cur.read_exact(&mut first)?; + if first[0] == 0xc0 { + return Ok(None); + } + + let mut prepend = PrependByteReader { + byte: Some(first[0]), + inner: &mut cur, + }; + let mut obj_fields = rmp::decode::read_map_len(&mut prepend)?; + let mut data_dir: Option = None; + + while obj_fields > 0 { + obj_fields -= 1; + + let obj_key_len = rmp::decode::read_str_len(&mut prepend)? as usize; + let mut obj_key_buf = vec![0u8; obj_key_len]; + prepend.read_exact(&mut obj_key_buf)?; + let obj_key = String::from_utf8(obj_key_buf)?; + + if obj_key == "DDir" { + let bin_len = rmp::decode::read_bin_len(&mut prepend)? as usize; + if bin_len != 16 { + return Err(Error::other(format!("DDir must be 16 bytes, got {bin_len}"))); + } + let mut raw = [0u8; 16]; + prepend.read_exact(&mut raw)?; + let id = Uuid::from_bytes(raw); + data_dir = if id.is_nil() { None } else { Some(id) }; + break; + } + + skip_msgp_value(&mut prepend)?; + } + + return Ok(data_dir); + } + _ => { + skip_msgp_value(&mut cur)?; + } + } + } + + Ok(None) + } + pub fn valid(&self) -> bool { if !self.version_type.valid() { return false; @@ -367,6 +438,9 @@ impl FileMetaVersion { // decode_data_dir_from_meta reads data_dir from meta TODO: directly parse only data_dir from meta buf, msg.skip pub fn decode_data_dir_from_meta(buf: &[u8]) -> Result> { + if let Ok(data_dir) = Self::decode_data_dir_from_v2_object(buf) { + return Ok(data_dir); + } Ok(Self::try_from(buf)?.get_data_dir()) } @@ -3213,4 +3287,31 @@ mod tests { assert_eq!(fi.mod_time, Some(sample_mod_time())); assert_eq!(fi.metadata.get("x-rustfs-test").map(String::as_str), Some("gone")); } + + #[test] + fn decode_data_dir_from_meta_extracts_v2_object_fast_path() { + let data_dir = Uuid::new_v4(); + let version = FileMetaVersion { + version_type: VersionType::Object, + object: Some(MetaObject { + version_id: Some(Uuid::new_v4()), + data_dir: Some(data_dir), + erasure_algorithm: ErasureAlgo::ReedSolomon, + erasure_m: 2, + erasure_n: 4, + erasure_block_size: 1024 * 1024, + erasure_index: 1, + erasure_dist: vec![1, 2, 3, 4, 5, 6], + bitrot_checksum_algo: ChecksumAlgo::HighwayHash, + size: 64 * 1024, + mod_time: Some(OffsetDateTime::now_utc()), + ..Default::default() + }), + ..Default::default() + }; + + let encoded = version.marshal_msg().expect("marshal"); + let decoded = FileMetaVersion::decode_data_dir_from_meta(&encoded).expect("decode data_dir"); + assert_eq!(decoded, Some(data_dir)); + } } diff --git a/crates/filemeta/src/filemeta_inline.rs b/crates/filemeta/src/filemeta_inline.rs index fe52055cc..d66910cb0 100644 --- a/crates/filemeta/src/filemeta_inline.rs +++ b/crates/filemeta/src/filemeta_inline.rs @@ -23,6 +23,141 @@ pub struct InlineData(Vec); const INLINE_DATA_VER: u8 = 1; impl InlineData { + fn contains_key_by(&self, mut should_remove: F) -> Result + where + F: FnMut(&[u8]) -> bool, + { + let buf = self.after_version(); + if buf.is_empty() { + return Ok(false); + } + + let mut scan_cur = Cursor::new(buf); + let mut scan_fields_len = rmp::decode::read_map_len(&mut scan_cur)? as usize; + + while scan_fields_len > 0 { + scan_fields_len -= 1; + + let str_len = rmp::decode::read_str_len(&mut scan_cur)? as usize; + let key_start = scan_cur.position() as usize; + let key_end = key_start + str_len; + scan_cur.set_position(key_end as u64); + + let bin_len = rmp::decode::read_bin_len(&mut scan_cur)? as usize; + let value_start = scan_cur.position() as usize; + let value_end = value_start + bin_len; + scan_cur.set_position(value_end as u64); + + if should_remove(&buf[key_start..key_end]) { + return Ok(true); + } + } + + Ok(false) + } + + fn remove_keys_by(&mut self, mut should_remove: F) -> Result + where + F: FnMut(&[u8]) -> bool, + { + let buf = self.after_version(); + if buf.is_empty() { + return Ok(false); + } + + let mut cur = Cursor::new(buf); + let mut fields_len = rmp::decode::read_map_len(&mut cur)? as usize; + let mut keys = Vec::with_capacity(fields_len); + let mut values = Vec::with_capacity(fields_len); + let mut found = false; + + while fields_len > 0 { + fields_len -= 1; + + let str_len = rmp::decode::read_str_len(&mut cur)? as usize; + let mut field_buf = vec![0u8; str_len]; + cur.read_exact(&mut field_buf)?; + + let bin_len = rmp::decode::read_bin_len(&mut cur)? as usize; + let start = cur.position() as usize; + let end = start + bin_len; + cur.set_position(end as u64); + + if should_remove(field_buf.as_slice()) { + found = true; + continue; + } + + keys.push(String::from_utf8(field_buf)?); + values.push(buf[start..end].to_vec()); + } + + if !found { + return Ok(false); + } + + if keys.is_empty() { + self.0 = Vec::new(); + return Ok(true); + } + + self.serialize(keys, values)?; + Ok(true) + } + + fn remove_two_keys_by_bytes(&mut self, first_key: &[u8], second_key: &[u8]) -> Result { + let buf = self.after_version(); + if buf.is_empty() { + return Ok(false); + } + + let same = first_key == second_key; + let mut cur = Cursor::new(buf); + let mut fields_len = rmp::decode::read_map_len(&mut cur)? as usize; + let mut keys = Vec::with_capacity(fields_len + 1); + let mut values = Vec::with_capacity(fields_len + 1); + let mut found = false; + + while fields_len > 0 { + fields_len -= 1; + + let str_len = rmp::decode::read_str_len(&mut cur)? as usize; + let mut field_buf = vec![0u8; str_len]; + cur.read_exact(&mut field_buf)?; + + let bin_len = rmp::decode::read_bin_len(&mut cur)? as usize; + let start = cur.position() as usize; + let end = start + bin_len; + cur.set_position(end as u64); + + let should_remove = if same { + field_buf.as_slice() == first_key + } else { + field_buf.as_slice() == first_key || field_buf.as_slice() == second_key + }; + + if should_remove { + found = true; + continue; + } + + keys.push(String::from_utf8(field_buf)?); + values.push(buf[start..end].to_vec()); + } + + if !found { + return Ok(false); + } + + if keys.is_empty() { + self.0 = Vec::new(); + return Ok(true); + } + + self.serialize(keys, values)?; + Ok(true) + } + pub fn new() -> Self { Self(Vec::new()) } @@ -183,115 +318,29 @@ impl InlineData { } pub fn remove_key(&mut self, key: &str) -> Result { - let buf = self.after_version(); - if buf.is_empty() { + let key_bytes = key.as_bytes(); + if !self.contains_key_by(|candidate| candidate == key_bytes)? { return Ok(false); } - - let mut cur = Cursor::new(buf); - - let mut fields_len = rmp::decode::read_map_len(&mut cur)? as usize; - let mut keys = Vec::with_capacity(fields_len); - let mut values = Vec::with_capacity(fields_len); - let mut found = false; - - while fields_len > 0 { - fields_len -= 1; - - let str_len = rmp::decode::read_str_len(&mut cur)?; - - let mut field_buff = vec![0u8; str_len as usize]; - - cur.read_exact(&mut field_buff)?; - - let find_key = String::from_utf8(field_buff)?; - - let bin_len = rmp::decode::read_bin_len(&mut cur)? as usize; - let start = cur.position() as usize; - let end = start + bin_len; - cur.set_position(end as u64); - - if find_key == key { - found = true; - continue; - } - - keys.push(find_key); - values.push(buf[start..end].to_vec()); - } - - if !found { - return Ok(false); - } - - if keys.is_empty() { - self.0 = Vec::new(); - return Ok(true); - } - - self.serialize(keys, values)?; - Ok(true) + self.remove_keys_by(|candidate| candidate == key_bytes) } pub fn remove(&mut self, remove_keys: Vec) -> Result { - let buf = self.after_version(); - if buf.is_empty() { - return Ok(false); - } - let mut cur = Cursor::new(buf); - - let mut fields_len = rmp::decode::read_map_len(&mut cur)? as usize; - let mut keys = Vec::with_capacity(fields_len + 1); - let mut values = Vec::with_capacity(fields_len + 1); - - let remove_key = |found_key: &str| { - for key in remove_keys.iter() { - if key.to_string().as_str() == found_key { - return true; - } - } - false - }; - - let mut found = false; - - while fields_len > 0 { - fields_len -= 1; - - let str_len = rmp::decode::read_str_len(&mut cur)?; - - let mut field_buff = vec![0u8; str_len as usize]; - - cur.read_exact(&mut field_buff)?; - - let find_key = String::from_utf8(field_buff)?; - - let bin_len = rmp::decode::read_bin_len(&mut cur)? as usize; - let start = cur.position() as usize; - let end = start + bin_len; - cur.set_position(end as u64); - - let find_value = &buf[start..end]; - - if !remove_key(&find_key) { - values.push(find_value.to_vec()); - keys.push(find_key); - } else { - found = true; - } + let mut encoded_keys = Vec::with_capacity(remove_keys.len()); + for key in remove_keys { + let mut buf = Uuid::encode_buffer(); + encoded_keys.push(key.hyphenated().encode_lower(&mut buf).to_string().into_bytes()); } - if !found { - return Ok(false); - } + self.remove_keys_by(|candidate| encoded_keys.iter().any(|key| candidate == key.as_slice())) + } - if keys.is_empty() { - self.0 = Vec::new(); - return Ok(true); - } - - self.serialize(keys, values)?; - Ok(true) + pub fn remove_two(&mut self, first: Uuid, second: Uuid) -> Result { + let mut first_buf = Uuid::encode_buffer(); + let mut second_buf = Uuid::encode_buffer(); + let first_key = first.hyphenated().encode_lower(&mut first_buf).as_bytes(); + let second_key = second.hyphenated().encode_lower(&mut second_buf).as_bytes(); + self.remove_two_keys_by_bytes(first_key, second_key) } fn serialize(&mut self, keys: Vec, values: Vec>) -> Result<()> { assert_eq!(keys.len(), values.len(), "InlineData serialize: keys/values not match"); @@ -319,3 +368,44 @@ impl InlineData { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn remove_key_miss_keeps_inline_data_unchanged() { + let mut data = InlineData::new(); + data.replace("keep", b"value".to_vec()).expect("seed inline data"); + let before = data.as_slice().to_vec(); + + let removed = data.remove_key("missing").expect("remove_key should succeed"); + + assert!(!removed); + assert_eq!(data.as_slice(), before.as_slice()); + } + + #[test] + fn remove_two_removes_only_matching_keys() { + let first = Uuid::new_v4(); + let second = Uuid::new_v4(); + let keep = Uuid::new_v4(); + let mut data = InlineData::new(); + data.replace(first.hyphenated().to_string().as_str(), b"first".to_vec()) + .expect("seed first key"); + data.replace(second.hyphenated().to_string().as_str(), b"second".to_vec()) + .expect("seed second key"); + data.replace(keep.hyphenated().to_string().as_str(), b"keep".to_vec()) + .expect("seed keep key"); + + let removed = data.remove_two(first, second).expect("remove_two should succeed"); + + assert!(removed); + assert_eq!(data.find(first.hyphenated().to_string().as_str()).expect("find first"), None); + assert_eq!(data.find(second.hyphenated().to_string().as_str()).expect("find second"), None); + assert_eq!( + data.find(keep.hyphenated().to_string().as_str()).expect("find keep"), + Some(b"keep".to_vec()) + ); + } +} diff --git a/rustfs/src/admin/handlers/profile_admin.rs b/rustfs/src/admin/handlers/profile_admin.rs index 4e2d830d6..9b129b5c5 100644 --- a/rustfs/src/admin/handlers/profile_admin.rs +++ b/rustfs/src/admin/handlers/profile_admin.rs @@ -54,6 +54,17 @@ pub fn register_profiling_route(r: &mut S3Router) -> std::io::Re pub struct ProfileHandler {} +#[allow(dead_code)] +fn map_cpu_profile_collect_error_message(err: &str) -> (StatusCode, String) { + if err.contains("start running cpu profiler error") { + return ( + StatusCode::CONFLICT, + "CPU profiler is already running. Disable RUSTFS_OBS_PROFILING_EXPORT_ENABLED or retry later.".to_string(), + ); + } + (StatusCode::INTERNAL_SERVER_ERROR, format!("Failed to collect CPU profile: {err}")) +} + #[async_trait::async_trait] impl Operation for ProfileHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { @@ -95,15 +106,19 @@ impl Operation for ProfileHandler { headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/octet-stream")); Ok(S3Response::with_headers((StatusCode::OK, Body::from(bytes)), headers)) } - Err(e) => Ok(S3Response::new(( - StatusCode::INTERNAL_SERVER_ERROR, - Body::from(format!("Failed to read profile file: {e}")), - ))), + Err(e) => { + error!("Failed to read profile file {}: {}", path.display(), e); + Ok(S3Response::new(( + StatusCode::INTERNAL_SERVER_ERROR, + Body::from(format!("Failed to read profile file: {e}")), + ))) + } }, - Err(e) => Ok(S3Response::new(( - StatusCode::INTERNAL_SERVER_ERROR, - Body::from(format!("Failed to collect CPU profile: {e}")), - ))), + Err(e) => { + let (status, message) = map_cpu_profile_collect_error_message(&e); + error!("CPU protobuf profile collection failed: {}", e); + Ok(S3Response::new((status, Body::from(message)))) + } }, "flamegraph" | "svg" => { let freq = get_env_usize(ENV_CPU_FREQ, DEFAULT_CPU_FREQ) as i32; @@ -213,6 +228,7 @@ mod tests { use crate::admin::router::Operation; use http::{Extensions, HeaderMap, Uri}; use hyper::Method; + use hyper::StatusCode; use matchit::Params; use s3s::{Body, S3ErrorCode, S3Request}; @@ -268,4 +284,12 @@ mod tests { assert_eq!(err.code(), &S3ErrorCode::AccessDenied); assert_eq!(err.message(), Some("Signature is required")); } + + #[test] + fn cpu_profile_collect_error_maps_profiler_conflict_to_409() { + let (status, message) = + super::map_cpu_profile_collect_error_message("create profiler failed: start running cpu profiler error"); + assert_eq!(status, StatusCode::CONFLICT); + assert!(message.contains("CPU profiler is already running")); + } } diff --git a/rustfs/src/config/cli.rs b/rustfs/src/config/cli.rs index 559402e10..7720a91a3 100644 --- a/rustfs/src/config/cli.rs +++ b/rustfs/src/config/cli.rs @@ -196,7 +196,10 @@ pub struct ServerOpts { )] pub console_address: String, - /// Observability endpoint for trace, metrics and logs,only support grpc mode. + /// Root OTLP endpoint for traces, metrics, and logs. + /// For the current observability pipeline this should be an OTLP/HTTP base + /// URL such as `http://otel-collector:4318` or + /// `http://host.docker.internal:4318`. #[arg( long, default_value_t = rustfs_config::DEFAULT_OBS_ENDPOINT.to_string(), diff --git a/scripts/run_four_node_cluster_failover_bench.sh b/scripts/run_four_node_cluster_failover_bench.sh index 2c14b3ca7..a87c7a45e 100755 --- a/scripts/run_four_node_cluster_failover_bench.sh +++ b/scripts/run_four_node_cluster_failover_bench.sh @@ -13,9 +13,11 @@ RUN_BENCHMARK="${RUN_BENCHMARK:-true}" KEEP_UP="${KEEP_UP:-false}" PRECHECK_AUTO_CLEANUP="${PRECHECK_AUTO_CLEANUP:-true}" WAIT_PROBE_MODE="${WAIT_PROBE_MODE:-service}" +COMPOSE_UP_NO_BUILD="${COMPOSE_UP_NO_BUILD:-false}" RUSTFS_ACCESS_KEY="${RUSTFS_ACCESS_KEY:-rustfs-cluster-admin}" RUSTFS_SECRET_KEY="${RUSTFS_SECRET_KEY:-rustfs-cluster-secret}" +RUSTFS_DOCKER_PLATFORM="${RUSTFS_DOCKER_PLATFORM:-}" RUSTFS_OBS_ENDPOINT="${RUSTFS_OBS_ENDPOINT:-}" RUSTFS_UNSAFE_BYPASS_DISK_CHECK="${RUSTFS_UNSAFE_BYPASS_DISK_CHECK:-true}" @@ -29,6 +31,8 @@ BENCH_WAIT_MODE="${BENCH_WAIT_MODE:-ready}" BENCH_ENDPOINT="${BENCH_ENDPOINT:-http://127.0.0.1:9000}" BENCH_BUCKET="${BENCH_BUCKET:-rustfs-four-node-bench}" +BENCH_AUTO_NEW_BUCKET="${BENCH_AUTO_NEW_BUCKET:-true}" +BENCH_BUCKET_PREFIX="${BENCH_BUCKET_PREFIX:-rustfs-four-node-bench}" BENCH_CONCURRENCY="${BENCH_CONCURRENCY:-}" BENCH_CONCURRENCIES="${BENCH_CONCURRENCIES:-}" BENCH_DURATION="${BENCH_DURATION:-60s}" @@ -65,6 +69,7 @@ Options: Environment: CLUSTER_COMPOSE OBS_COMPOSE PROJECT_NAME IMAGE_TAG WITH_OBSERVABILITY BUILD_LOCAL_IMAGE RUN_FAILOVER RUN_BENCHMARK KEEP_UP + COMPOSE_UP_NO_BUILD (true|false, default: false) RUSTFS_ACCESS_KEY RUSTFS_SECRET_KEY RUSTFS_OBS_ENDPOINT PRECHECK_AUTO_CLEANUP (true|false, default: true) WAIT_PROBE_MODE (service|ready, default: service) @@ -73,6 +78,8 @@ Environment: BENCH_CONCURRENCIES BENCH_DURATION BENCH_SIZES OUT_DIR BENCH_WAIT_MODE (ready|service, default: ready) BENCH_READY_TIMEOUT_SECS (default: 180) + BENCH_AUTO_NEW_BUCKET (true|false, default: true) + BENCH_BUCKET_PREFIX (default: rustfs-four-node-bench) USAGE } @@ -543,6 +550,8 @@ run_benchmark() { local bench_out_dir local conc local conc_dir + local bench_bucket + local -a bench_extra_args=() bench_out_dir="${OUT_DIR}/benchmark" mkdir -p "${bench_out_dir}" @@ -551,6 +560,11 @@ run_benchmark() { exit 1 fi + bench_bucket="${BENCH_BUCKET}" + if [[ "${BENCH_AUTO_NEW_BUCKET}" == "true" ]]; then + bench_extra_args+=(--auto-new-bucket) + fi + log_info "Waiting for benchmark endpoint readiness (mode=${BENCH_WAIT_MODE})" wait_bench_endpoint_ready @@ -574,7 +588,9 @@ run_benchmark() { --endpoint "${BENCH_ENDPOINT}" \ --access-key "${RUSTFS_ACCESS_KEY}" \ --secret-key "${RUSTFS_SECRET_KEY}" \ - --bucket "${BENCH_BUCKET}" \ + --bucket "${bench_bucket}" \ + --bucket-prefix "${BENCH_BUCKET_PREFIX}" \ + "${bench_extra_args[@]}" \ --concurrency "${conc}" \ --duration "${BENCH_DURATION}" \ --sizes "${BENCH_SIZES}" \ @@ -690,6 +706,8 @@ main() { resolve_bool "RUN_FAILOVER" "${RUN_FAILOVER}" resolve_bool "RUN_BENCHMARK" "${RUN_BENCHMARK}" resolve_bool "KEEP_UP" "${KEEP_UP}" + resolve_bool "COMPOSE_UP_NO_BUILD" "${COMPOSE_UP_NO_BUILD}" + resolve_bool "BENCH_AUTO_NEW_BUCKET" "${BENCH_AUTO_NEW_BUCKET}" resolve_bool "PRECHECK_AUTO_CLEANUP" "${PRECHECK_AUTO_CLEANUP}" resolve_probe_mode resolve_bench_wait_mode @@ -731,7 +749,12 @@ main() { if [[ "${BUILD_LOCAL_IMAGE}" == "true" ]]; then log_info "Building local image from Dockerfile.source: ${IMAGE_TAG}" - docker build -f "${PROJECT_ROOT}/Dockerfile.source" -t "${IMAGE_TAG}" "${PROJECT_ROOT}" + if [[ -n "${RUSTFS_DOCKER_PLATFORM}" ]]; then + log_info "Using docker build platform: ${RUSTFS_DOCKER_PLATFORM}" + docker build --platform "${RUSTFS_DOCKER_PLATFORM}" -f "${PROJECT_ROOT}/Dockerfile.source" -t "${IMAGE_TAG}" "${PROJECT_ROOT}" + else + docker build -f "${PROJECT_ROOT}/Dockerfile.source" -t "${IMAGE_TAG}" "${PROJECT_ROOT}" + fi else log_info "Skipping image build" fi @@ -739,7 +762,11 @@ main() { run_precheck_after_build log_info "Starting compose stack" - compose up -d + if [[ "${COMPOSE_UP_NO_BUILD}" == "true" ]]; then + compose up -d --no-build + else + compose up -d + fi log_info "Waiting for 4-node cluster readiness (mode=${WAIT_PROBE_MODE})" wait_cluster_ready diff --git a/scripts/run_object_batch_bench.sh b/scripts/run_object_batch_bench.sh index 120231a72..d266fb86f 100755 --- a/scripts/run_object_batch_bench.sh +++ b/scripts/run_object_batch_bench.sh @@ -11,6 +11,8 @@ ENDPOINT="" ACCESS_KEY="" SECRET_KEY="" BUCKET="rustfs-bench" +AUTO_NEW_BUCKET=false +BUCKET_PREFIX="rustfs-bench" REGION="us-east-1" CONCURRENCY=128 DURATION="60s" @@ -38,6 +40,8 @@ Required: Optional: --bucket Bucket name (default: rustfs-bench) + --auto-new-bucket Auto-generate a unique bucket for this run + --bucket-prefix Prefix used with --auto-new-bucket (default: rustfs-bench) --region Region (default: us-east-1) --concurrency Concurrency for all sizes (default: 128) --duration warp duration, e.g. 60s/2m (default: 60s) @@ -92,6 +96,8 @@ parse_args() { --access-key) ACCESS_KEY="$2"; shift 2 ;; --secret-key) SECRET_KEY="$2"; shift 2 ;; --bucket) BUCKET="$2"; shift 2 ;; + --auto-new-bucket) AUTO_NEW_BUCKET=true; shift ;; + --bucket-prefix) BUCKET_PREFIX="$2"; shift 2 ;; --region) REGION="$2"; shift 2 ;; --concurrency) CONCURRENCY="$2"; shift 2 ;; --duration) DURATION="$2"; shift 2 ;; @@ -156,6 +162,15 @@ setup_output() { echo "size,tool,concurrency,status,throughput,requests_per_sec,avg_latency,log_file" > "$SUMMARY_CSV" } +resolve_bucket() { + if [[ "$AUTO_NEW_BUCKET" != "true" ]]; then + return + fi + local suffix + suffix="$(date +%Y%m%d%H%M%S)-$RANDOM" + BUCKET="${BUCKET_PREFIX}-${suffix}" +} + extract_value() { local pattern="$1" local file="$2" @@ -238,6 +253,14 @@ run_one() { fi fi + if [[ "$TOOL" == "warp" ]]; then + # Warp may still exit with code 0 even when it prints runtime failures. + # Treat explicit error lines as failed runs to keep summary.csv reliable. + if rg -q 'warp: ' "$log_file"; then + status="failed" + fi + fi + local metrics throughput reqps latency metrics="$(collect_metrics "$log_file")" throughput="$(echo "$metrics" | cut -d',' -f1)" @@ -250,6 +273,7 @@ run_one() { main() { parse_args "$@" validate_args + resolve_bucket require_cmd rg if [[ "$TOOL" == "warp" ]]; then require_cmd "$WARP_BIN" @@ -261,8 +285,10 @@ main() { echo "Output dir: $OUT_DIR" echo "Tool: $TOOL" + echo "Bucket: $BUCKET" echo "Sizes: $SIZES" echo "Concurrency: $CONCURRENCY" + echo "$BUCKET" > "$OUT_DIR/bucket.txt" IFS=',' read -r -a size_arr <<< "$SIZES" for raw_size in "${size_arr[@]}"; do