From 96fb06f48e8618933529f79ba19aea446a2d0073 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 18 Apr 2026 02:21:08 +0800 Subject: [PATCH] refactor(metrics): separate console stream from exports (#2583) --- Cargo.lock | 1 + crates/io-metrics/Cargo.toml | 5 ++ crates/io-metrics/README.md | 3 +- crates/io-metrics/README_zh.md | 3 +- crates/io-metrics/benches/metrics_pipeline.rs | 43 ++++++++++ crates/io-metrics/examples/metrics_example.rs | 2 +- crates/io-metrics/src/lib.rs | 2 + rustfs/src/admin/handlers/metrics.rs | 80 +++++++++++-------- rustfs/src/storage/backpressure.rs | 2 +- rustfs/src/storage/concurrency/io_schedule.rs | 2 +- rustfs/src/storage/deadlock_detector.rs | 2 +- rustfs/src/storage/lock_optimizer.rs | 2 +- rustfs/src/storage/timeout_wrapper.rs | 4 +- 13 files changed, 107 insertions(+), 44 deletions(-) create mode 100644 crates/io-metrics/benches/metrics_pipeline.rs diff --git a/Cargo.lock b/Cargo.lock index a5b44fb8a..587211b45 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8082,6 +8082,7 @@ dependencies = [ name = "rustfs-io-metrics" version = "0.0.5" dependencies = [ + "criterion", "metrics", "num_cpus", "thiserror 2.0.18", diff --git a/crates/io-metrics/Cargo.toml b/crates/io-metrics/Cargo.toml index 8173fb999..cc920732e 100644 --- a/crates/io-metrics/Cargo.toml +++ b/crates/io-metrics/Cargo.toml @@ -24,6 +24,10 @@ description = "Metrics collection and reporting for RustFS (using metrics crate keywords = ["metrics", "zero-copy", "rustfs", "otel", "performance"] categories = ["development-tools", "filesystem"] +[[bench]] +name = "metrics_pipeline" +harness = false + [dependencies] metrics = { workspace = true } num_cpus = { workspace = true } @@ -32,6 +36,7 @@ tokio = { workspace = true, features = ["sync","rt"] } tracing = { workspace = true } [dev-dependencies] +criterion = { workspace = true } tokio = { workspace = true, features = ["test-util","rt","macros"] } [lints] diff --git a/crates/io-metrics/README.md b/crates/io-metrics/README.md index a8143e23e..ddd3bdcd6 100644 --- a/crates/io-metrics/README.md +++ b/crates/io-metrics/README.md @@ -31,6 +31,7 @@ - **Bandwidth Monitoring**: Real-time bandwidth observation and analysis - **Performance Metrics**: I/O performance metrics collection - **Unified Configuration**: Centralized configuration management +- **Exporter Boundary**: Emit via `metrics`, export via `rustfs-obs`, no Prometheus HTTP endpoint ## Features @@ -199,7 +200,7 @@ cargo test --package rustfs-io-metrics cargo test --package rustfs-io-metrics --lib adaptive_ttl # Run benchmarks -cargo bench --package rustfs-io-metrics +cargo bench --package rustfs-io-metrics --bench metrics_pipeline ``` ## Documentation diff --git a/crates/io-metrics/README_zh.md b/crates/io-metrics/README_zh.md index b2baa0921..249f94fde 100644 --- a/crates/io-metrics/README_zh.md +++ b/crates/io-metrics/README_zh.md @@ -31,6 +31,7 @@ - **带宽监控**：实时带宽观测和分析 - **性能指标**：I/O 性能指标收集 - **统一配置**：集中式配置管理 +- **导出边界**：通过 `metrics` 主动上报，由 `rustfs-obs` 负责 OTEL 导出，不提供 Prometheus HTTP 端点 ## ✨ 核心功能 @@ -280,7 +281,7 @@ cargo test --package rustfs-io-metrics cargo test --package rustfs-io-metrics --lib adaptive_ttl # 运行基准测试 -cargo bench --package rustfs-io-metrics +cargo bench --package rustfs-io-metrics --bench metrics_pipeline ``` ## 📚 文档 diff --git a/crates/io-metrics/benches/metrics_pipeline.rs b/crates/io-metrics/benches/metrics_pipeline.rs new file mode 100644 index 000000000..20967f6b1 --- /dev/null +++ b/crates/io-metrics/benches/metrics_pipeline.rs @@ -0,0 +1,43 @@ +use criterion::{Criterion, criterion_group, criterion_main}; +use rustfs_io_metrics::{MetricsCollector, PerformanceMetrics, record_get_object_request_started}; +use std::hint::black_box; +use std::sync::Arc; +use std::time::Duration; + +fn bench_record_get_object_request_started(c: &mut Criterion) { + c.bench_function("record_get_object_request_started", |b| b.iter(record_get_object_request_started)); +} + +fn bench_update_concurrent_requests(c: &mut Criterion) { + let metrics = PerformanceMetrics::new(); + + c.bench_function("performance_metrics_update_concurrent_requests", |b| { + b.iter(|| metrics.update_concurrent_requests(black_box(64))) + }); +} + +fn bench_metrics_collector_record_io_operation(c: &mut Criterion) { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build tokio runtime for io-metrics benchmark"); + let collector = MetricsCollector::new(Arc::new(PerformanceMetrics::new()), 256); + + c.bench_function("metrics_collector_record_io_operation", |b| { + b.iter(|| { + runtime.block_on(collector.record_io_operation( + black_box(64 * 1024), + Duration::from_micros(black_box(250)), + black_box(true), + )) + }) + }); +} + +criterion_group!( + benches, + bench_record_get_object_request_started, + bench_update_concurrent_requests, + bench_metrics_collector_record_io_operation +); +criterion_main!(benches); diff --git a/crates/io-metrics/examples/metrics_example.rs b/crates/io-metrics/examples/metrics_example.rs index 249a4351b..153fa188a 100644 --- a/crates/io-metrics/examples/metrics_example.rs +++ b/crates/io-metrics/examples/metrics_example.rs @@ -143,7 +143,7 @@ fn metrics_recording_example() { println!(" Recorded 10 cache operations (hits: 7, misses: 3)"); println!(" Metrics reported via metrics crate"); - println!(" View via Prometheus/Grafana"); + println!(" Export via rustfs-obs OTEL pipeline"); println!(); } diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index c61fca193..4a175feea 100644 --- a/crates/io-metrics/src/lib.rs +++ b/crates/io-metrics/src/lib.rs @@ -23,6 +23,8 @@ //! - **PerformanceMetrics**: Shared atomic counter struct for advanced use cases //! - **MetricsCollector**: I/O operation tracking with percentile calculation //! - **AutoTuner**: Automatic performance optimization based on metrics +//! - **No HTTP metrics endpoint**: consumers emit metrics through the `metrics` crate; +//! `rustfs-obs` owns OTEL initialization and export //! //! # Usage //! diff --git a/rustfs/src/admin/handlers/metrics.rs b/rustfs/src/admin/handlers/metrics.rs index d1e114162..b0af4211f 100644 --- a/rustfs/src/admin/handlers/metrics.rs +++ b/rustfs/src/admin/handlers/metrics.rs @@ -12,22 +12,29 @@ // See the License for the specific language governing permissions and // limitations under the License. +//! Console realtime metrics API. +//! +//! This preserves the console's fixed `/admin/v3/metrics` contract while +//! keeping the response format explicitly NDJSON. It is not a Prometheus text +//! exposition endpoint. + use crate::admin::router::Operation; use bytes::Bytes; use futures::{Stream, StreamExt}; -use http::Uri; +use http::{HeaderMap, HeaderValue, Uri}; use hyper::StatusCode; use matchit::Params; use rustfs_ecstore::metrics_realtime::{CollectMetricsOpts, MetricType, collect_local_metrics}; use rustfs_madmin::metrics::RealtimeMetrics; use rustfs_madmin::utils::parse_duration; +use s3s::header::CONTENT_TYPE; use s3s::stream::{ByteStream, DynByteStream}; use s3s::{Body, S3Request, S3Response, S3Result, StdError, s3_error}; use serde::{Deserialize, Serialize}; use std::collections::{HashMap, HashSet}; use std::pin::Pin; use std::task::{Context, Poll}; -use std::time::Duration as std_Duration; +use std::time::Duration as StdDuration; use tokio::sync::mpsc; use tokio::time::interval; use tokio::{select, spawn}; @@ -36,6 +43,7 @@ use tracing::{debug, error, warn}; const DEFAULT_METRICS_SAMPLES: u64 = 1; const MAX_METRICS_SAMPLES: u64 = 120; +const CONSOLE_METRICS_CONTENT_TYPE: &str = "application/x-ndjson"; #[derive(Debug, Serialize, Deserialize)] struct MetricsParams { @@ -173,14 +181,15 @@ pub struct MetricsHandler {} impl Operation for MetricsHandler { async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { debug!("handle MetricsHandler, uri: {:?}, params: {:?}", req.uri, params); - let Some(_cred) = req.credentials else { return Err(s3_error!(InvalidRequest, "get cred failed")) }; - debug!("validated metrics request credentials"); + let Some(_cred) = req.credentials else { + return Err(s3_error!(InvalidRequest, "get cred failed")); + }; + debug!("validated console metrics request credentials"); let mp = extract_metrics_init_params(&req.uri); debug!("mp: {:?}", mp); - let tick = parse_duration(&mp.tick).unwrap_or_else(|_| std_Duration::from_secs(3)); - + let tick = parse_duration(&mp.tick).unwrap_or_else(|_| StdDuration::from_secs(3)); let mut n = resolve_sample_count(&mp); let types = if mp.types != 0 { @@ -193,53 +202,43 @@ impl Operation for MetricsHandler { s.split(',').filter(|part| !part.is_empty()).map(String::from).collect() } - let disks = parse_comma_separated(&mp.disks); let by_disk = mp.by_disk == "true"; - let disk_map = disks; - - let job_id = mp.by_job_id; - let hosts = parse_comma_separated(&mp.hosts); let by_host = mp.by_host == "true"; - let host_map = hosts; - - let d_id = mp.by_dep_id; let mut interval = interval(tick); - let opts = CollectMetricsOpts { - hosts: host_map, - disks: disk_map, - job_id, - dep_id: d_id, + hosts: parse_comma_separated(&mp.hosts), + disks: parse_comma_separated(&mp.disks), + job_id: mp.by_job_id, + dep_id: mp.by_dep_id, }; let (tx, rx) = mpsc::channel(10); let in_stream: DynByteStream = Box::pin(MetricsStream { inner: ReceiverStream::new(rx), }); let body = Body::from(in_stream); + spawn(async move { while n > 0 { - let mut m = RealtimeMetrics::default(); - let m_local = collect_local_metrics(types, &opts).await; - m.merge(m_local); + let mut metrics = RealtimeMetrics::default(); + let local_metrics = collect_local_metrics(types, &opts).await; + metrics.merge(local_metrics); if !by_host { - m.by_host = HashMap::new(); + metrics.by_host = HashMap::new(); } if !by_disk { - m.by_disk = HashMap::new(); + metrics.by_disk = HashMap::new(); } - m.finally = n <= 1; + metrics.finally = n <= 1; - // todo write resp - match serde_json::to_vec(&m) { - Ok(mut re) => { - // NDJSON framing allows stream clients to parse incremental records. - re.push(b'\n'); - let _ = tx.send(Ok(Bytes::from(re))).await; + match serde_json::to_vec(&metrics) { + Ok(mut encoded) => { + encoded.push(b'\n'); + let _ = tx.send(Ok(Bytes::from(encoded))).await; } - Err(e) => { - error!("MetricsHandler: json encode failed, err: {:?}", e); + Err(err) => { + error!("MetricsHandler: json encode failed, err: {:?}", err); return; } } @@ -256,13 +255,19 @@ impl Operation for MetricsHandler { } }); - Ok(S3Response::new((StatusCode::OK, body))) + let mut header = HeaderMap::new(); + header.insert(CONTENT_TYPE, HeaderValue::from_static(CONSOLE_METRICS_CONTENT_TYPE)); + + Ok(S3Response::with_headers((StatusCode::OK, body), header)) } } #[cfg(test)] mod tests { - use super::{DEFAULT_METRICS_SAMPLES, MAX_METRICS_SAMPLES, extract_metrics_init_params, resolve_sample_count}; + use super::{ + CONSOLE_METRICS_CONTENT_TYPE, DEFAULT_METRICS_SAMPLES, MAX_METRICS_SAMPLES, extract_metrics_init_params, + resolve_sample_count, + }; use http::Uri; #[test] @@ -288,4 +293,9 @@ mod tests { assert_eq!(resolve_sample_count(&mp), MAX_METRICS_SAMPLES); } + + #[test] + fn metrics_handler_uses_ndjson_content_type() { + assert_eq!(CONSOLE_METRICS_CONTENT_TYPE, "application/x-ndjson"); + } } diff --git a/rustfs/src/storage/backpressure.rs b/rustfs/src/storage/backpressure.rs index b19857150..2bb92445d 100644 --- a/rustfs/src/storage/backpressure.rs +++ b/rustfs/src/storage/backpressure.rs @@ -24,7 +24,7 @@ //! //! - Configurable buffer size with high/low watermarks //! - Backpressure state monitoring and events -//! - Prometheus metrics for backpressure events +//! - Backpressure metrics emitted through the shared metrics pipeline //! - Graceful handling of slow consumers //! //! # Architecture diff --git a/rustfs/src/storage/concurrency/io_schedule.rs b/rustfs/src/storage/concurrency/io_schedule.rs index 59a2a689f..b5e58233a 100644 --- a/rustfs/src/storage/concurrency/io_schedule.rs +++ b/rustfs/src/storage/concurrency/io_schedule.rs @@ -1683,7 +1683,7 @@ impl IoPriorityQueue { /// Global metrics for I/O priority queue monitoring. /// -/// These metrics are exposed for Prometheus scraping and provide +/// These metrics are emitted through the shared metrics pipeline and provide /// visibility into the priority queue behavior. #[allow(dead_code)] pub struct IoPriorityMetrics { diff --git a/rustfs/src/storage/deadlock_detector.rs b/rustfs/src/storage/deadlock_detector.rs index 1fcb0c2f0..bf86edbe0 100644 --- a/rustfs/src/storage/deadlock_detector.rs +++ b/rustfs/src/storage/deadlock_detector.rs @@ -34,7 +34,7 @@ //! - Request resource tracking (locks, memory, file handles) //! - Lock wait graph analysis for cycle detection //! - Configurable detection interval and hang threshold -//! - Prometheus metrics for deadlock events +//! - Deadlock metrics emitted through the shared metrics pipeline //! - Detailed diagnostic logging //! //! # Usage diff --git a/rustfs/src/storage/lock_optimizer.rs b/rustfs/src/storage/lock_optimizer.rs index 5c6255ea6..e5bbfc345 100644 --- a/rustfs/src/storage/lock_optimizer.rs +++ b/rustfs/src/storage/lock_optimizer.rs @@ -37,7 +37,7 @@ //! - Early lock release after metadata read //! - Lock hold time monitoring //! - Configurable optimization (can be disabled for debugging) -//! - Prometheus metrics for lock contention analysis +//! - Lock contention metrics emitted through the shared metrics pipeline //! //! # Architecture //! diff --git a/rustfs/src/storage/timeout_wrapper.rs b/rustfs/src/storage/timeout_wrapper.rs index 017239096..5c814d3fa 100644 --- a/rustfs/src/storage/timeout_wrapper.rs +++ b/rustfs/src/storage/timeout_wrapper.rs @@ -34,7 +34,7 @@ //! - Configurable request-level timeout (default 30 seconds) //! - Automatic cancellation of sub-tasks on timeout //! - Resource cleanup on timeout (locks, memory, file handles) -//! - Prometheus metrics for timeout monitoring +//! - Timeout metrics emitted through `rustfs-io-metrics` // Allow dead_code for public API that may be used by external modules or future features #![allow(dead_code)] @@ -42,7 +42,7 @@ //! - Configurable request-level timeout (default 30 seconds) //! - Automatic cancellation of sub-tasks on timeout //! - Resource cleanup on timeout (locks, memory, file handles) -//! - Prometheus metrics for timeout monitoring +//! - Timeout metrics emitted through `rustfs-io-metrics` //! //! # Usage //!