mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-28 16:07:05 +00:00
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
This commit is contained in:
@@ -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
|
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
|
### (Deprecated) Minimal Observability
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
|
|||||||
@@ -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
|
||||||
@@ -90,6 +90,54 @@ docker compose down -v
|
|||||||
- **Grafana**: Dashboards and datasources are provisioned from the `grafana/` directory.
|
- **Grafana**: Dashboards and datasources are provisioned from the `grafana/` directory.
|
||||||
- **Collector**: Edit `otel-collector-config.yaml` to modify pipelines, processors, or exporters.
|
- **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
|
## Troubleshooting
|
||||||
|
|
||||||
- **Service Health**: Check the health of services using `docker compose ps`.
|
- **Service Health**: Check the health of services using `docker compose ps`.
|
||||||
|
|||||||
@@ -90,6 +90,50 @@ docker compose down -v
|
|||||||
- **Grafana**: 仪表盘和数据源从 `grafana/` 目录预置。
|
- **Grafana**: 仪表盘和数据源从 `grafana/` 目录预置。
|
||||||
- **Collector**: 编辑 `otel-collector-config.yaml` 以修改管道、处理器或导出器。
|
- **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` 检查服务健康状况。
|
- **服务健康**: 使用 `docker compose ps` 检查服务健康状况。
|
||||||
|
|||||||
@@ -145,5 +145,9 @@ harness = false
|
|||||||
name = "comparison_benchmark"
|
name = "comparison_benchmark"
|
||||||
harness = false
|
harness = false
|
||||||
|
|
||||||
|
[[bench]]
|
||||||
|
name = "rename_data_meta_benchmark"
|
||||||
|
harness = false
|
||||||
|
|
||||||
[lib]
|
[lib]
|
||||||
doctest = false
|
doctest = false
|
||||||
|
|||||||
@@ -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);
|
||||||
@@ -1194,7 +1194,9 @@ impl LocalDisk {
|
|||||||
skip_parent = self.root.as_path();
|
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?;
|
os::make_dir_all(parent, skip_parent).await?;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1203,6 +1205,11 @@ impl LocalDisk {
|
|||||||
Ok(f)
|
Ok(f)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn open_file_read_only(&self, path: impl AsRef<Path>) -> Result<File> {
|
||||||
|
let f = super::fs::open_file(path.as_ref(), O_RDONLY).await.map_err(to_file_error)?;
|
||||||
|
Ok(f)
|
||||||
|
}
|
||||||
|
|
||||||
#[allow(dead_code)]
|
#[allow(dead_code)]
|
||||||
fn get_metrics(&self) -> DiskMetrics {
|
fn get_metrics(&self) -> DiskMetrics {
|
||||||
DiskMetrics::default()
|
DiskMetrics::default()
|
||||||
@@ -2137,7 +2144,7 @@ impl DiskAPI for LocalDisk {
|
|||||||
let file_path = self.get_object_path(volume, path)?;
|
let file_path = self.get_object_path(volume, path)?;
|
||||||
check_path_length(file_path.to_string_lossy().as_ref())?;
|
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))
|
Ok(Box::new(f))
|
||||||
}
|
}
|
||||||
@@ -2154,7 +2161,7 @@ impl DiskAPI for LocalDisk {
|
|||||||
let file_path = self.get_object_path(volume, path)?;
|
let file_path = self.get_object_path(volume, path)?;
|
||||||
check_path_length(file_path.to_string_lossy().as_ref())?;
|
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 meta = f.metadata().await?;
|
||||||
let end_offset = offset.checked_add(length).ok_or(DiskError::FileCorrupt)?;
|
let end_offset = offset.checked_add(length).ok_or(DiskError::FileCorrupt)?;
|
||||||
@@ -2513,32 +2520,41 @@ impl DiskAPI for LocalDisk {
|
|||||||
|
|
||||||
// TODO: Healing
|
// 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
|
// 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 = {
|
// Reuse one metadata scan to find the version data_dir and determine whether it is shared.
|
||||||
xlmeta.find_version(search_version_id).ok().and_then(|(_, ver)| {
|
let has_old_data_dir = xlmeta.find_unshared_data_dir_for_version(search_version_id);
|
||||||
// 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)
|
|
||||||
})
|
|
||||||
};
|
|
||||||
if let Some(old_data_dir) = has_old_data_dir.as_ref() {
|
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 {
|
if xlmeta.versions.len() <= 10 {
|
||||||
// TODO: Sign
|
// 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() {
|
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 {
|
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;
|
let _ = self.delete_file(&dst_volume_dir, dst_data_path, false, false).await;
|
||||||
info!(
|
info!(
|
||||||
@@ -2547,6 +2563,10 @@ impl DiskAPI for LocalDisk {
|
|||||||
);
|
);
|
||||||
return Err(err);
|
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 {
|
if let Some(old_data_dir) = has_old_data_dir {
|
||||||
|
|||||||
@@ -215,24 +215,29 @@ pub async fn os_mkdir_all(dir_path: impl AsRef<Path>, base_dir: impl AsRef<Path>
|
|||||||
return Ok(());
|
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 let Err(e) = super::fs::mkdir(dir_path.as_ref()).await {
|
||||||
if e.kind() == io::ErrorKind::AlreadyExists {
|
if e.kind() == io::ErrorKind::AlreadyExists {
|
||||||
return Ok(());
|
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(())
|
Ok(())
|
||||||
|
|||||||
@@ -282,6 +282,11 @@ impl FileMeta {
|
|||||||
fi.version_id = Some(Uuid::nil());
|
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 version_key = data_key_for_version(fi.version_id);
|
||||||
let mut next_data = self.data.clone();
|
let mut next_data = self.data.clone();
|
||||||
|
|
||||||
@@ -317,46 +322,30 @@ impl FileMeta {
|
|||||||
}
|
}
|
||||||
|
|
||||||
let vid = version.get_version_id();
|
let vid = version.get_version_id();
|
||||||
|
let vid_is_null = vid.is_none() || vid == Some(Uuid::nil());
|
||||||
// Match existing version for replace; null version: None and Some(nil) are equivalent
|
let existing_idx = if vid_is_null {
|
||||||
let matches = |h: &Option<Uuid>| {
|
self.versions
|
||||||
let v_null = vid.is_none() || vid == Some(Uuid::nil());
|
.iter()
|
||||||
let h_null = h.is_none() || *h == Some(Uuid::nil());
|
.position(|v| v.header.version_id.is_none() || v.header.version_id == Some(Uuid::nil()))
|
||||||
(v_null && h_null) || (vid == *h)
|
} 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);
|
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 mod_time = version.get_mod_time();
|
||||||
let new_shallow = FileMetaShallowVersion::try_from(version)?;
|
let new_shallow = FileMetaShallowVersion::try_from(version)?;
|
||||||
|
let insert_pos = match mod_time {
|
||||||
for (idx, exist) in self.versions.iter().enumerate() {
|
Some(nm) => self.versions.partition_point(|exist| match exist.header.mod_time {
|
||||||
let ex_mt = exist.header.mod_time;
|
Some(em) => em > nm,
|
||||||
let insert_here = match (ex_mt, mod_time) {
|
None => false,
|
||||||
(None, _) => true, // placeholder: always insert before
|
}),
|
||||||
(Some(em), Some(nm)) => em <= nm,
|
None => self.versions.partition_point(|exist| exist.header.mod_time.is_some()),
|
||||||
(Some(_), None) => false,
|
};
|
||||||
};
|
self.versions.insert(insert_pos, new_shallow);
|
||||||
if insert_here {
|
Ok(())
|
||||||
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"))
|
|
||||||
|
|
||||||
// if !ver.valid() {
|
// if !ver.valid() {
|
||||||
// return Err(Error::other("attempted to add invalid version"));
|
// return Err(Error::other("attempted to add invalid version"));
|
||||||
|
|||||||
@@ -13,8 +13,48 @@
|
|||||||
// limitations under the License.
|
// limitations under the License.
|
||||||
|
|
||||||
use super::*;
|
use super::*;
|
||||||
|
use std::collections::HashSet;
|
||||||
|
|
||||||
impl FileMeta {
|
impl FileMeta {
|
||||||
|
pub fn find_unshared_data_dir_for_version(&self, version_id: Option<Uuid>) -> Option<Uuid> {
|
||||||
|
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<Uuid>, data_dir: &Option<Uuid>) -> usize {
|
pub fn shard_data_dir_count(&self, vid: &Option<Uuid>, data_dir: &Option<Uuid>) -> usize {
|
||||||
let vid = vid.unwrap_or_default();
|
let vid = vid.unwrap_or_default();
|
||||||
self.versions
|
self.versions
|
||||||
@@ -58,3 +98,70 @@ impl FileMeta {
|
|||||||
.count()
|
.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);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -312,6 +312,77 @@ pub struct FileMetaVersion {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl FileMetaVersion {
|
impl FileMetaVersion {
|
||||||
|
fn decode_data_dir_from_v2_object(buf: &[u8]) -> Result<Option<Uuid>> {
|
||||||
|
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<Uuid> = 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 {
|
pub fn valid(&self) -> bool {
|
||||||
if !self.version_type.valid() {
|
if !self.version_type.valid() {
|
||||||
return false;
|
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
|
// 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<Option<Uuid>> {
|
pub fn decode_data_dir_from_meta(buf: &[u8]) -> Result<Option<Uuid>> {
|
||||||
|
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())
|
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.mod_time, Some(sample_mod_time()));
|
||||||
assert_eq!(fi.metadata.get("x-rustfs-test").map(String::as_str), Some("gone"));
|
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));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -23,6 +23,141 @@ pub struct InlineData(Vec<u8>);
|
|||||||
const INLINE_DATA_VER: u8 = 1;
|
const INLINE_DATA_VER: u8 = 1;
|
||||||
|
|
||||||
impl InlineData {
|
impl InlineData {
|
||||||
|
fn contains_key_by<F>(&self, mut should_remove: F) -> Result<bool>
|
||||||
|
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<F>(&mut self, mut should_remove: F) -> Result<bool>
|
||||||
|
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<bool> {
|
||||||
|
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 {
|
pub fn new() -> Self {
|
||||||
Self(Vec::new())
|
Self(Vec::new())
|
||||||
}
|
}
|
||||||
@@ -183,115 +318,29 @@ impl InlineData {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub fn remove_key(&mut self, key: &str) -> Result<bool> {
|
pub fn remove_key(&mut self, key: &str) -> Result<bool> {
|
||||||
let buf = self.after_version();
|
let key_bytes = key.as_bytes();
|
||||||
if buf.is_empty() {
|
if !self.contains_key_by(|candidate| candidate == key_bytes)? {
|
||||||
return Ok(false);
|
return Ok(false);
|
||||||
}
|
}
|
||||||
|
self.remove_keys_by(|candidate| candidate == key_bytes)
|
||||||
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)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn remove(&mut self, remove_keys: Vec<Uuid>) -> Result<bool> {
|
pub fn remove(&mut self, remove_keys: Vec<Uuid>) -> Result<bool> {
|
||||||
let buf = self.after_version();
|
let mut encoded_keys = Vec::with_capacity(remove_keys.len());
|
||||||
if buf.is_empty() {
|
for key in remove_keys {
|
||||||
return Ok(false);
|
let mut buf = Uuid::encode_buffer();
|
||||||
}
|
encoded_keys.push(key.hyphenated().encode_lower(&mut buf).to_string().into_bytes());
|
||||||
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;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if !found {
|
self.remove_keys_by(|candidate| encoded_keys.iter().any(|key| candidate == key.as_slice()))
|
||||||
return Ok(false);
|
}
|
||||||
}
|
|
||||||
|
|
||||||
if keys.is_empty() {
|
pub fn remove_two(&mut self, first: Uuid, second: Uuid) -> Result<bool> {
|
||||||
self.0 = Vec::new();
|
let mut first_buf = Uuid::encode_buffer();
|
||||||
return Ok(true);
|
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.serialize(keys, values)?;
|
self.remove_two_keys_by_bytes(first_key, second_key)
|
||||||
Ok(true)
|
|
||||||
}
|
}
|
||||||
fn serialize(&mut self, keys: Vec<String>, values: Vec<Vec<u8>>) -> Result<()> {
|
fn serialize(&mut self, keys: Vec<String>, values: Vec<Vec<u8>>) -> Result<()> {
|
||||||
assert_eq!(keys.len(), values.len(), "InlineData serialize: keys/values not match");
|
assert_eq!(keys.len(), values.len(), "InlineData serialize: keys/values not match");
|
||||||
@@ -319,3 +368,44 @@ impl InlineData {
|
|||||||
Ok(())
|
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())
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -54,6 +54,17 @@ pub fn register_profiling_route(r: &mut S3Router<AdminOperation>) -> std::io::Re
|
|||||||
|
|
||||||
pub struct ProfileHandler {}
|
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]
|
#[async_trait::async_trait]
|
||||||
impl Operation for ProfileHandler {
|
impl Operation for ProfileHandler {
|
||||||
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
||||||
@@ -95,15 +106,19 @@ impl Operation for ProfileHandler {
|
|||||||
headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/octet-stream"));
|
headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/octet-stream"));
|
||||||
Ok(S3Response::with_headers((StatusCode::OK, Body::from(bytes)), headers))
|
Ok(S3Response::with_headers((StatusCode::OK, Body::from(bytes)), headers))
|
||||||
}
|
}
|
||||||
Err(e) => Ok(S3Response::new((
|
Err(e) => {
|
||||||
StatusCode::INTERNAL_SERVER_ERROR,
|
error!("Failed to read profile file {}: {}", path.display(), e);
|
||||||
Body::from(format!("Failed to read profile file: {e}")),
|
Ok(S3Response::new((
|
||||||
))),
|
StatusCode::INTERNAL_SERVER_ERROR,
|
||||||
|
Body::from(format!("Failed to read profile file: {e}")),
|
||||||
|
)))
|
||||||
|
}
|
||||||
},
|
},
|
||||||
Err(e) => Ok(S3Response::new((
|
Err(e) => {
|
||||||
StatusCode::INTERNAL_SERVER_ERROR,
|
let (status, message) = map_cpu_profile_collect_error_message(&e);
|
||||||
Body::from(format!("Failed to collect CPU profile: {e}")),
|
error!("CPU protobuf profile collection failed: {}", e);
|
||||||
))),
|
Ok(S3Response::new((status, Body::from(message))))
|
||||||
|
}
|
||||||
},
|
},
|
||||||
"flamegraph" | "svg" => {
|
"flamegraph" | "svg" => {
|
||||||
let freq = get_env_usize(ENV_CPU_FREQ, DEFAULT_CPU_FREQ) as i32;
|
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 crate::admin::router::Operation;
|
||||||
use http::{Extensions, HeaderMap, Uri};
|
use http::{Extensions, HeaderMap, Uri};
|
||||||
use hyper::Method;
|
use hyper::Method;
|
||||||
|
use hyper::StatusCode;
|
||||||
use matchit::Params;
|
use matchit::Params;
|
||||||
use s3s::{Body, S3ErrorCode, S3Request};
|
use s3s::{Body, S3ErrorCode, S3Request};
|
||||||
|
|
||||||
@@ -268,4 +284,12 @@ mod tests {
|
|||||||
assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
|
assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
|
||||||
assert_eq!(err.message(), Some("Signature is required"));
|
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"));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -196,7 +196,10 @@ pub struct ServerOpts {
|
|||||||
)]
|
)]
|
||||||
pub console_address: String,
|
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(
|
#[arg(
|
||||||
long,
|
long,
|
||||||
default_value_t = rustfs_config::DEFAULT_OBS_ENDPOINT.to_string(),
|
default_value_t = rustfs_config::DEFAULT_OBS_ENDPOINT.to_string(),
|
||||||
|
|||||||
@@ -13,9 +13,11 @@ RUN_BENCHMARK="${RUN_BENCHMARK:-true}"
|
|||||||
KEEP_UP="${KEEP_UP:-false}"
|
KEEP_UP="${KEEP_UP:-false}"
|
||||||
PRECHECK_AUTO_CLEANUP="${PRECHECK_AUTO_CLEANUP:-true}"
|
PRECHECK_AUTO_CLEANUP="${PRECHECK_AUTO_CLEANUP:-true}"
|
||||||
WAIT_PROBE_MODE="${WAIT_PROBE_MODE:-service}"
|
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_ACCESS_KEY="${RUSTFS_ACCESS_KEY:-rustfs-cluster-admin}"
|
||||||
RUSTFS_SECRET_KEY="${RUSTFS_SECRET_KEY:-rustfs-cluster-secret}"
|
RUSTFS_SECRET_KEY="${RUSTFS_SECRET_KEY:-rustfs-cluster-secret}"
|
||||||
|
RUSTFS_DOCKER_PLATFORM="${RUSTFS_DOCKER_PLATFORM:-}"
|
||||||
RUSTFS_OBS_ENDPOINT="${RUSTFS_OBS_ENDPOINT:-}"
|
RUSTFS_OBS_ENDPOINT="${RUSTFS_OBS_ENDPOINT:-}"
|
||||||
RUSTFS_UNSAFE_BYPASS_DISK_CHECK="${RUSTFS_UNSAFE_BYPASS_DISK_CHECK:-true}"
|
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_ENDPOINT="${BENCH_ENDPOINT:-http://127.0.0.1:9000}"
|
||||||
BENCH_BUCKET="${BENCH_BUCKET:-rustfs-four-node-bench}"
|
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_CONCURRENCY="${BENCH_CONCURRENCY:-}"
|
||||||
BENCH_CONCURRENCIES="${BENCH_CONCURRENCIES:-}"
|
BENCH_CONCURRENCIES="${BENCH_CONCURRENCIES:-}"
|
||||||
BENCH_DURATION="${BENCH_DURATION:-60s}"
|
BENCH_DURATION="${BENCH_DURATION:-60s}"
|
||||||
@@ -65,6 +69,7 @@ Options:
|
|||||||
Environment:
|
Environment:
|
||||||
CLUSTER_COMPOSE OBS_COMPOSE PROJECT_NAME IMAGE_TAG
|
CLUSTER_COMPOSE OBS_COMPOSE PROJECT_NAME IMAGE_TAG
|
||||||
WITH_OBSERVABILITY BUILD_LOCAL_IMAGE RUN_FAILOVER RUN_BENCHMARK KEEP_UP
|
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
|
RUSTFS_ACCESS_KEY RUSTFS_SECRET_KEY RUSTFS_OBS_ENDPOINT
|
||||||
PRECHECK_AUTO_CLEANUP (true|false, default: true)
|
PRECHECK_AUTO_CLEANUP (true|false, default: true)
|
||||||
WAIT_PROBE_MODE (service|ready, default: service)
|
WAIT_PROBE_MODE (service|ready, default: service)
|
||||||
@@ -73,6 +78,8 @@ Environment:
|
|||||||
BENCH_CONCURRENCIES BENCH_DURATION BENCH_SIZES OUT_DIR
|
BENCH_CONCURRENCIES BENCH_DURATION BENCH_SIZES OUT_DIR
|
||||||
BENCH_WAIT_MODE (ready|service, default: ready)
|
BENCH_WAIT_MODE (ready|service, default: ready)
|
||||||
BENCH_READY_TIMEOUT_SECS (default: 180)
|
BENCH_READY_TIMEOUT_SECS (default: 180)
|
||||||
|
BENCH_AUTO_NEW_BUCKET (true|false, default: true)
|
||||||
|
BENCH_BUCKET_PREFIX (default: rustfs-four-node-bench)
|
||||||
USAGE
|
USAGE
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -543,6 +550,8 @@ run_benchmark() {
|
|||||||
local bench_out_dir
|
local bench_out_dir
|
||||||
local conc
|
local conc
|
||||||
local conc_dir
|
local conc_dir
|
||||||
|
local bench_bucket
|
||||||
|
local -a bench_extra_args=()
|
||||||
bench_out_dir="${OUT_DIR}/benchmark"
|
bench_out_dir="${OUT_DIR}/benchmark"
|
||||||
mkdir -p "${bench_out_dir}"
|
mkdir -p "${bench_out_dir}"
|
||||||
|
|
||||||
@@ -551,6 +560,11 @@ run_benchmark() {
|
|||||||
exit 1
|
exit 1
|
||||||
fi
|
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})"
|
log_info "Waiting for benchmark endpoint readiness (mode=${BENCH_WAIT_MODE})"
|
||||||
wait_bench_endpoint_ready
|
wait_bench_endpoint_ready
|
||||||
|
|
||||||
@@ -574,7 +588,9 @@ run_benchmark() {
|
|||||||
--endpoint "${BENCH_ENDPOINT}" \
|
--endpoint "${BENCH_ENDPOINT}" \
|
||||||
--access-key "${RUSTFS_ACCESS_KEY}" \
|
--access-key "${RUSTFS_ACCESS_KEY}" \
|
||||||
--secret-key "${RUSTFS_SECRET_KEY}" \
|
--secret-key "${RUSTFS_SECRET_KEY}" \
|
||||||
--bucket "${BENCH_BUCKET}" \
|
--bucket "${bench_bucket}" \
|
||||||
|
--bucket-prefix "${BENCH_BUCKET_PREFIX}" \
|
||||||
|
"${bench_extra_args[@]}" \
|
||||||
--concurrency "${conc}" \
|
--concurrency "${conc}" \
|
||||||
--duration "${BENCH_DURATION}" \
|
--duration "${BENCH_DURATION}" \
|
||||||
--sizes "${BENCH_SIZES}" \
|
--sizes "${BENCH_SIZES}" \
|
||||||
@@ -690,6 +706,8 @@ main() {
|
|||||||
resolve_bool "RUN_FAILOVER" "${RUN_FAILOVER}"
|
resolve_bool "RUN_FAILOVER" "${RUN_FAILOVER}"
|
||||||
resolve_bool "RUN_BENCHMARK" "${RUN_BENCHMARK}"
|
resolve_bool "RUN_BENCHMARK" "${RUN_BENCHMARK}"
|
||||||
resolve_bool "KEEP_UP" "${KEEP_UP}"
|
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_bool "PRECHECK_AUTO_CLEANUP" "${PRECHECK_AUTO_CLEANUP}"
|
||||||
resolve_probe_mode
|
resolve_probe_mode
|
||||||
resolve_bench_wait_mode
|
resolve_bench_wait_mode
|
||||||
@@ -731,7 +749,12 @@ main() {
|
|||||||
|
|
||||||
if [[ "${BUILD_LOCAL_IMAGE}" == "true" ]]; then
|
if [[ "${BUILD_LOCAL_IMAGE}" == "true" ]]; then
|
||||||
log_info "Building local image from Dockerfile.source: ${IMAGE_TAG}"
|
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
|
else
|
||||||
log_info "Skipping image build"
|
log_info "Skipping image build"
|
||||||
fi
|
fi
|
||||||
@@ -739,7 +762,11 @@ main() {
|
|||||||
run_precheck_after_build
|
run_precheck_after_build
|
||||||
|
|
||||||
log_info "Starting compose stack"
|
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})"
|
log_info "Waiting for 4-node cluster readiness (mode=${WAIT_PROBE_MODE})"
|
||||||
wait_cluster_ready
|
wait_cluster_ready
|
||||||
|
|||||||
@@ -11,6 +11,8 @@ ENDPOINT=""
|
|||||||
ACCESS_KEY=""
|
ACCESS_KEY=""
|
||||||
SECRET_KEY=""
|
SECRET_KEY=""
|
||||||
BUCKET="rustfs-bench"
|
BUCKET="rustfs-bench"
|
||||||
|
AUTO_NEW_BUCKET=false
|
||||||
|
BUCKET_PREFIX="rustfs-bench"
|
||||||
REGION="us-east-1"
|
REGION="us-east-1"
|
||||||
CONCURRENCY=128
|
CONCURRENCY=128
|
||||||
DURATION="60s"
|
DURATION="60s"
|
||||||
@@ -38,6 +40,8 @@ Required:
|
|||||||
|
|
||||||
Optional:
|
Optional:
|
||||||
--bucket Bucket name (default: rustfs-bench)
|
--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)
|
--region Region (default: us-east-1)
|
||||||
--concurrency Concurrency for all sizes (default: 128)
|
--concurrency Concurrency for all sizes (default: 128)
|
||||||
--duration warp duration, e.g. 60s/2m (default: 60s)
|
--duration warp duration, e.g. 60s/2m (default: 60s)
|
||||||
@@ -92,6 +96,8 @@ parse_args() {
|
|||||||
--access-key) ACCESS_KEY="$2"; shift 2 ;;
|
--access-key) ACCESS_KEY="$2"; shift 2 ;;
|
||||||
--secret-key) SECRET_KEY="$2"; shift 2 ;;
|
--secret-key) SECRET_KEY="$2"; shift 2 ;;
|
||||||
--bucket) BUCKET="$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 ;;
|
--region) REGION="$2"; shift 2 ;;
|
||||||
--concurrency) CONCURRENCY="$2"; shift 2 ;;
|
--concurrency) CONCURRENCY="$2"; shift 2 ;;
|
||||||
--duration) DURATION="$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"
|
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() {
|
extract_value() {
|
||||||
local pattern="$1"
|
local pattern="$1"
|
||||||
local file="$2"
|
local file="$2"
|
||||||
@@ -238,6 +253,14 @@ run_one() {
|
|||||||
fi
|
fi
|
||||||
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: <ERROR>' "$log_file"; then
|
||||||
|
status="failed"
|
||||||
|
fi
|
||||||
|
fi
|
||||||
|
|
||||||
local metrics throughput reqps latency
|
local metrics throughput reqps latency
|
||||||
metrics="$(collect_metrics "$log_file")"
|
metrics="$(collect_metrics "$log_file")"
|
||||||
throughput="$(echo "$metrics" | cut -d',' -f1)"
|
throughput="$(echo "$metrics" | cut -d',' -f1)"
|
||||||
@@ -250,6 +273,7 @@ run_one() {
|
|||||||
main() {
|
main() {
|
||||||
parse_args "$@"
|
parse_args "$@"
|
||||||
validate_args
|
validate_args
|
||||||
|
resolve_bucket
|
||||||
require_cmd rg
|
require_cmd rg
|
||||||
if [[ "$TOOL" == "warp" ]]; then
|
if [[ "$TOOL" == "warp" ]]; then
|
||||||
require_cmd "$WARP_BIN"
|
require_cmd "$WARP_BIN"
|
||||||
@@ -261,8 +285,10 @@ main() {
|
|||||||
|
|
||||||
echo "Output dir: $OUT_DIR"
|
echo "Output dir: $OUT_DIR"
|
||||||
echo "Tool: $TOOL"
|
echo "Tool: $TOOL"
|
||||||
|
echo "Bucket: $BUCKET"
|
||||||
echo "Sizes: $SIZES"
|
echo "Sizes: $SIZES"
|
||||||
echo "Concurrency: $CONCURRENCY"
|
echo "Concurrency: $CONCURRENCY"
|
||||||
|
echo "$BUCKET" > "$OUT_DIR/bucket.txt"
|
||||||
|
|
||||||
IFS=',' read -r -a size_arr <<< "$SIZES"
|
IFS=',' read -r -a size_arr <<< "$SIZES"
|
||||||
for raw_size in "${size_arr[@]}"; do
|
for raw_size in "${size_arr[@]}"; do
|
||||||
|
|||||||
Reference in New Issue
Block a user