gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 08:18:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(obs): improve telemetry stack, replication metrics, and Grafana alignment (#2672)

Browse Source 
Co-authored-by: Filipe Monteiro <a22407332@alunos.ulht.pt>
Co-authored-by: cxymds <Cxymds@qq.com>
Co-authored-by: weisd <im@weisd.in>
Co-authored-by: loverustfs <hello@rustfs.com>
Co-authored-by: 安正超 <anzhengchao@gmail.com>
This commit is contained in:
houseme
2026-04-24 21:50:17 +08:00
committed by GitHub
parent 2705e3f53b
commit 13b4500212
19 changed files with 4603 additions and 531 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.docker/observability/README.md
+12
View File
@@ -13,6 +13,9 @@ The stack is composed of the following best-in-class open-source components:
- **Jaeger** (v1.59.0): Distributed tracing system (configured as a secondary UI/storage).
- **OpenTelemetry Collector** (v0.104.0): A vendor-agnostic implementation for receiving, processing, and exporting telemetry data.
By default, this stack uses Tempo in single-binary mode and does not require Kafka/Redpanda.
If you want the Kafka-backed HA Tempo path, use `docker-compose-example-for-rustfs.yml` together with `docker-compose-tempo-ha-override.yml`.
## Architecture
1. **Telemetry Collection**: Applications send OTLP (OpenTelemetry Protocol) data (Metrics, Logs, Traces) to the **OpenTelemetry Collector**.
@@ -46,6 +49,15 @@ Run the following command to start the entire stack:
docker compose up -d
```
### High Availability Tempo
The default `docker-compose.yml` is the single-node stack.
If you need the Kafka-backed HA Tempo configuration, start it with:
```bash
docker compose -f docker-compose-example-for-rustfs.yml -f docker-compose-tempo-ha-override.yml up -d
```
### Access Dashboards
| Service | URL | Credentials | Description |
.docker/observability/README_ZH.md
+12
View File
@@ -13,6 +13,9 @@
- **Jaeger** (v1.59.0): 分布式追踪系统（配置为辅助 UI/存储）。
- **OpenTelemetry Collector** (v0.104.0): 接收、处理和导出遥测数据的供应商无关实现。
默认情况下，这套技术栈使用 Tempo 单二进制模式，不依赖 Kafka/Redpanda。
如果需要基于 Kafka 的 HA Tempo 路径，请使用 `docker-compose-example-for-rustfs.yml` 配合 `docker-compose-tempo-ha-override.yml`。
## 架构
1. **遥测收集**: 应用程序将 OTLP (OpenTelemetry Protocol) 数据（指标、日志、追踪）发送到 **OpenTelemetry Collector**。
@@ -46,6 +49,15 @@
docker compose up -d
```
### Tempo 高可用模式
默认的 `docker-compose.yml` 对应单机栈。
如果需要基于 Kafka 的 HA Tempo 配置，请使用：
```bash
docker compose -f docker-compose-example-for-rustfs.yml -f docker-compose-tempo-ha-override.yml up -d
```
### 访问仪表盘
| 服务 | URL | 凭据 | 描述 |
.docker/observability/docker-compose-example-for-rustfs.yml
+2 -4
View File
@@ -85,10 +85,8 @@ services:
networks:
- otel-network
restart: unless-stopped
depends_on:
- redpanda
healthcheck:
test: [ "CMD", "wget", "--spider", "-q", "http://localhost:3200/ready" ]
test: [ "CMD", "/tempo", "-version" ]
interval: 10s
timeout: 5s
retries: 3
@@ -205,7 +203,7 @@ services:
- prometheus
- loki
healthcheck:
test: [ "CMD", "wget", "--spider", "-q", "http://localhost:13133" ]
test: [ "CMD", "/otelcol-contrib", "--version" ]
interval: 10s
timeout: 5s
retries: 3
.docker/observability/docker-compose.yml
+3 -25
View File
@@ -16,10 +16,8 @@ services:
# --- Tracing ---
tempo:
image: grafana/tempo:2.10.3
image: grafana/tempo:2.10.5
container_name: tempo
depends_on:
- redpanda
command: [ "-config.file=/etc/tempo.yaml" ]
volumes:
- ./tempo.yaml:/etc/tempo.yaml:ro
@@ -33,31 +31,11 @@ services:
- otel-network
restart: unless-stopped
healthcheck:
test: [ "CMD", "wget", "--spider", "-q", "http://localhost:3200/ready" ]
test: [ "CMD", "/tempo", "-version" ]
interval: 10s
timeout: 5s
retries: 3
start_period: 15s
redpanda:
image: redpandadata/redpanda:latest
ports:
- "9092:9092" # Kafka API for clients
command: >
redpanda start --overprovisioned
--mode=dev-container
--kafka-addr=PLAINTEXT://0.0.0.0:9092
--advertise-kafka-addr=PLAINTEXT://redpanda:9092
redpanda-console:
image: docker.redpanda.com/redpandadata/console:latest
environment:
- CONFIG_FILEPATH=/etc/redpanda/redpanda-console-config.yaml
volumes:
- ./redpanda-console.yaml:/etc/redpanda/redpanda-console-config.yaml
ports:
- "8080:8080"
depends_on:
- redpanda
vulture:
image: grafana/tempo-vulture:latest
@@ -168,7 +146,7 @@ services:
- prometheus
- loki
healthcheck:
test: [ "CMD", "wget", "--spider", "-q", "http://localhost:13133" ]
test: [ "CMD", "/otelcol-contrib", "--version" ]
interval: 10s
timeout: 5s
retries: 3
.docker/observability/grafana/dashboards/rustfs.json
+3375 -142
View File
File diff suppressed because it is too large Load Diff
.docker/observability/grafana/provisioning/datasources.yaml
-97
View File
@@ -1,97 +0,0 @@
# 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.
apiVersion: 1
datasources:
- name: Prometheus
type: prometheus
uid: prometheus
access: proxy
orgId: 1
url: http://prometheus:9090
isDefault: true
version: 1
editable: false
jsonData:
httpMethod: GET
exemplarTraceIdDestinations:
- name: trace_id
datasourceUid: tempo
- name: Tempo
type: tempo
uid: tempo
access: proxy
orgId: 1
url: http://tempo:3200
isDefault: false
version: 1
editable: false
jsonData:
httpMethod: GET
serviceMap:
datasourceUid: prometheus
tracesToLogs:
datasourceUid: loki
tags: [ 'job', 'instance', 'pod', 'namespace', 'service.name' ]
mappedTags: [ { key: 'service.name', value: 'app' } ]
spanStartTimeShift: '1s'
spanEndTimeShift: '-1s'
filterByTraceID: true
filterBySpanID: false
tracesToMetrics:
datasourceUid: prometheus
tags: [ { key: 'service.name' }, { key: 'job' } ]
queries:
- name: 'Service-Level Latency'
query: 'sum(rate(traces_spanmetrics_latency_bucket{$$__tags}[5m])) by (le)'
- name: 'Service-Level Calls'
query: 'sum(rate(traces_spanmetrics_calls_total{$$__tags}[5m]))'
- name: 'Service-Level Errors'
query: 'sum(rate(traces_spanmetrics_calls_total{status_code="ERROR", $$__tags}[5m]))'
nodeGraph:
enabled: true
- name: Loki
type: loki
uid: loki
orgId: 1
url: http://loki:3100
isDefault: false
version: 1
editable: false
jsonData:
derivedFields:
- datasourceUid: tempo
matcherRegex: 'trace_id=(\w+)'
name: 'TraceID'
url: '$${__value.raw}'
- name: Jaeger
type: jaeger
uid: jaeger
url: http://jaeger:16686
access: proxy
isDefault: false
editable: false
jsonData:
tracesToLogs:
datasourceUid: loki
tags: [ 'job', 'instance', 'pod', 'namespace', 'service.name' ]
mappedTags: [ { key: 'service.name', value: 'app' } ]
spanStartTimeShift: '1s'
spanEndTimeShift: '-1s'
filterByTraceID: true
filterBySpanID: false
.docker/observability/tempo.yaml
+4 -31
View File
@@ -12,7 +12,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
partition_ring_live_store: true
stream_over_http_enabled: true
server:
@@ -33,33 +32,17 @@ distributor:
endpoint: "0.0.0.0:4317"
http:
endpoint: "0.0.0.0:4318"
#log_received_spans:
# enabled: true
# log_discarded_spans:
# enabled: true
backend_scheduler:
provider:
compaction:
compaction:
block_retention: 1h
backend_worker:
backend_scheduler_addr: localhost:3200
compaction:
block_retention: 1h
ring:
kvstore:
store: memberlist
querier:
query_live_store: true
ingester:
max_block_duration: 5m
metrics_generator:
registry:
external_labels:
source: tempo
cluster: docker-compose
traces_storage:
path: /var/tempo/generator/traces
storage:
path: /var/tempo/generator/wal
remote_write:
@@ -85,15 +68,5 @@ overrides:
processors: [ "span-metrics", "service-graphs", "local-blocks" ]
generate_native_histograms: both
ingest:
enabled: true
kafka:
address: redpanda:9092
topic: tempo-ingest
block_builder:
consume_cycle_duration: 30s
usage_report:
reporting_enabled: false
crates/ecstore/src/bucket/replication/replication_pool.rs
+15
View File
@@ -956,6 +956,9 @@ pub type DynReplicationPool = dyn ReplicationPoolTrait + Send + Sync;
/// Trait that abstracts the replication pool operations
#[async_trait::async_trait]
pub trait ReplicationPoolTrait: std::fmt::Debug {
fn active_workers(&self) -> i32;
fn active_mrf_workers(&self) -> i32;
fn active_lrg_workers(&self) -> i32;
async fn queue_replica_task(&self, ri: ReplicateObjectInfo);
async fn queue_replica_delete_task(&self, ri: DeletedObjectReplicationInfo);
async fn resize(&self, priority: ReplicationPriority, max_workers: usize, max_l_workers: usize);
@@ -972,6 +975,18 @@ pub trait ReplicationPoolTrait: std::fmt::Debug {
// Implement the trait for ReplicationPool
#[async_trait::async_trait]
impl<S: StorageAPI> ReplicationPoolTrait for ReplicationPool<S> {
fn active_workers(&self) -> i32 {
ReplicationPool::<S>::active_workers(self)
}
fn active_mrf_workers(&self) -> i32 {
ReplicationPool::<S>::active_mrf_workers(self)
}
fn active_lrg_workers(&self) -> i32 {
ReplicationPool::<S>::active_lrg_workers(self)
}
async fn queue_replica_task(&self, ri: ReplicateObjectInfo) {
self.queue_replica_task(ri).await;
}
crates/ecstore/src/bucket/replication/replication_resyncer.rs
+113 -26
View File
@@ -35,7 +35,7 @@ use crate::set_disk::get_lock_acquire_timeout;
use crate::store_api::{DeletedObject, HTTPRangeSpec, ObjectInfo, ObjectOptions, ObjectToDelete, WalkOptions};
use crate::{StorageAPI, new_object_layer_fn};
use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError};
use aws_sdk_s3::operation::head_object::HeadObjectOutput;
use aws_sdk_s3::operation::head_object::{HeadObjectError, HeadObjectOutput};
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::{CompletedPart, ObjectLockLegalHoldStatus};
use aws_smithy_types::body::SdkBody;
@@ -115,6 +115,41 @@ fn resync_state_accepts_update(state: &TargetReplicationResyncStatus, opts: &Res
state.resync_id.is_empty() || opts.resync_id.is_empty() || state.resync_id == opts.resync_id
}
fn should_count_head_proxy_failure(is_not_found: bool, code: Option<&str>, raw_status: Option<u16>) -> bool {
if is_not_found || matches!(code, Some("MethodNotAllowed" | "405")) {
return false;
}
!matches!(raw_status, Some(404 | 405))
}
fn is_head_proxy_failure(err: &SdkError<HeadObjectError>) -> bool {
let (is_not_found, code) = err
.as_service_error()
.map(|service_err| (service_err.is_not_found(), service_err.code()))
.unwrap_or((false, None));
let raw_status = err.raw_response().map(|resp| resp.status().as_u16());
should_count_head_proxy_failure(is_not_found, code, raw_status)
}
async fn record_proxy_request(bucket: &str, api: &str, is_err: bool) {
if let Some(stats) = GLOBAL_REPLICATION_STATS.get() {
stats.inc_proxy(bucket, api, is_err).await;
}
}
async fn head_object_with_proxy_stats(
source_bucket: &str,
target_client: &TargetClient,
target_bucket: &str,
object: &str,
version_id: Option<String>,
) -> std::result::Result<HeadObjectOutput, SdkError<HeadObjectError>> {
let result = target_client.head_object(target_bucket, object, version_id).await;
let is_err = result.as_ref().err().is_some_and(is_head_proxy_failure);
record_proxy_request(source_bucket, "HeadObject", is_err).await;
result
}
#[derive(Debug, Clone, Default)]
pub struct ResyncOpts {
pub bucket: String,
@@ -748,10 +783,15 @@ impl ReplicationResyncer {
let reset_id = target_client.reset_id.clone();
let (size, err) = if let Err(err) = target_client
.head_object(&target_client.bucket, &roi.name, roi.version_id.map(|v| v.to_string()))
.await
{
let head_result = head_object_with_proxy_stats(
&bucket_name,
target_client.as_ref(),
&target_client.bucket,
&roi.name,
roi.version_id.map(|v| v.to_string()),
)
.await;
let (size, err) = if let Err(err) = head_result {
if roi.delete_marker {
st.replicated_count += 1;
} else {
@@ -2120,9 +2160,14 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli
};
if dobj.delete_object.delete_marker && dobj.delete_object.delete_marker_version_id.is_some() {
match tgt_client
.head_object(&tgt_client.bucket, &dobj.delete_object.object_name, version_id.clone())
.await
match head_object_with_proxy_stats(
&dobj.bucket,
tgt_client.as_ref(),
&tgt_client.bucket,
&dobj.delete_object.object_name,
version_id.clone(),
)
.await
{
Ok(_) => {}
Err(e) => {
@@ -2471,9 +2516,14 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
}
let mut replication_action = replication_action;
match tgt_client
.head_object(&tgt_client.bucket, &object, self.version_id.map(|v| v.to_string()))
.await
match head_object_with_proxy_stats(
&bucket,
tgt_client.as_ref(),
&tgt_client.bucket,
&object,
self.version_id.map(|v| v.to_string()),
)
.await
{
Ok(oi) => {
replication_action = get_replication_action(&object_info, &oi, self.op_type);
@@ -2530,7 +2580,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
if let Some(err) = if is_multipart {
drop(gr);
replicate_object_with_multipart(MultipartReplicationContext {
let result = replicate_object_with_multipart(MultipartReplicationContext {
storage: storage.clone(),
cli: tgt_client.clone(),
src_bucket: &bucket,
@@ -2541,16 +2591,18 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
arn: &rinfo.arn,
put_opts,
})
.await
.err()
.await;
record_proxy_request(&bucket, "PutObject", result.is_err()).await;
result.err()
} else {
gr.stream = wrap_with_bandwidth_monitor(gr.stream, &put_opts, &bucket, &rinfo.arn);
let byte_stream = async_read_to_bytestream(gr.stream);
tgt_client
let result = tgt_client
.put_object(&tgt_client.bucket, &object, size, byte_stream, &put_opts)
.await
.map_err(|e| std::io::Error::other(e.to_string()))
.err()
.map_err(|e| std::io::Error::other(e.to_string()));
record_proxy_request(&bucket, "PutObject", result.is_err()).await;
result.err()
} {
rinfo.replication_status = ReplicationStatusType::Failed;
rinfo.error = Some(err.to_string());
@@ -2690,9 +2742,14 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
warn!("failed to set replication tagging directive header: {err}");
}
match tgt_client
.head_object(&tgt_client.bucket, &object, self.version_id.map(|v| v.to_string()))
.await
match head_object_with_proxy_stats(
&bucket,
tgt_client.as_ref(),
&tgt_client.bucket,
&object,
self.version_id.map(|v| v.to_string()),
)
.await
{
Ok(oi) => {
replication_action = get_replication_action(&object_info, &oi, self.op_type);
@@ -2812,7 +2869,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
if let Some(err) = if is_multipart {
drop(gr);
replicate_object_with_multipart(MultipartReplicationContext {
let result = replicate_object_with_multipart(MultipartReplicationContext {
storage: storage.clone(),
cli: tgt_client.clone(),
src_bucket: &bucket,
@@ -2823,16 +2880,18 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
arn: &rinfo.arn,
put_opts,
})
.await
.err()
.await;
record_proxy_request(&bucket, "PutObject", result.is_err()).await;
result.err()
} else {
gr.stream = wrap_with_bandwidth_monitor(gr.stream, &put_opts, &bucket, &rinfo.arn);
let byte_stream = async_read_to_bytestream(gr.stream);
tgt_client
let result = tgt_client
.put_object(&tgt_client.bucket, &object, size, byte_stream, &put_opts)
.await
.map_err(|e| std::io::Error::other(e.to_string()))
.err()
.map_err(|e| std::io::Error::other(e.to_string()));
record_proxy_request(&bucket, "PutObject", result.is_err()).await;
result.err()
} {
rinfo.replication_status = ReplicationStatusType::Failed;
rinfo.error = Some(err.to_string());
@@ -3748,6 +3807,34 @@ mod tests {
);
}
#[test]
fn test_should_count_head_proxy_failure_ignores_not_found_and_405() {
assert!(
!should_count_head_proxy_failure(true, Some("NoSuchKey"), Some(404)),
"not-found heads are expected when the object has not reached the target yet"
);
assert!(
!should_count_head_proxy_failure(false, Some("MethodNotAllowed"), Some(405)),
"405 delete-marker probing responses should not be counted as proxy failures"
);
assert!(
!should_count_head_proxy_failure(false, Some("405"), Some(405)),
"numeric 405 codes must align with MethodNotAllowed semantics"
);
}
#[test]
fn test_should_count_head_proxy_failure_counts_unexpected_errors() {
assert!(
should_count_head_proxy_failure(false, Some("AccessDenied"), Some(403)),
"non-NotFound and non-405 service errors should be counted as failures"
);
assert!(
should_count_head_proxy_failure(false, None, Some(500)),
"raw 5xx head responses should be counted as proxy failures"
);
}
#[tokio::test]
async fn test_get_heal_replicate_object_info_failed_object_returns_heal_roi() {
let oi = ObjectInfo {
crates/ecstore/src/bucket/replication/replication_state.rs
+233 -47
View File
@@ -12,18 +12,22 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::bucket::replication::get_global_replication_pool;
use crate::error::Error;
use crate::global::get_global_bucket_monitor;
use rustfs_filemeta::{ReplicatedTargetInfo, ReplicationStatusType, ReplicationType};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use std::sync::atomic::{AtomicI64, Ordering};
use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering};
use std::time::{Duration, SystemTime};
use std::time::{Duration, Instant, SystemTime};
use tokio::sync::{Mutex, RwLock};
use tokio::time::interval;
const ROLLING_WINDOW: Duration = Duration::from_secs(60);
const FAILURE_LAST_HOUR_WINDOW: Duration = Duration::from_secs(60 * 60);
/// Exponential Moving Average with thread-safe interior mutability
#[derive(Debug)]
pub struct ExponentialMovingAverage {
@@ -328,6 +332,13 @@ pub struct InQueueStats {
pub now_count: AtomicI64,
}
#[derive(Debug, Clone)]
struct QueueSample {
observed_at: Instant,
bytes: i64,
count: i64,
}
impl Clone for InQueueStats {
fn clone(&self) -> Self {
Self {
@@ -359,9 +370,60 @@ pub struct InQueueMetric {
pub curr: InQueueStats,
pub avg: InQueueStats,
pub max: InQueueStats,
pub last_minute: InQueueStats,
#[serde(skip)]
samples: VecDeque<QueueSample>,
}
impl InQueueMetric {
fn observe(&mut self, observed_at: Instant) {
let bytes = self.curr.now_bytes.load(Ordering::Relaxed);
let count = self.curr.now_count.load(Ordering::Relaxed);
self.curr.bytes = bytes;
self.curr.count = count;
self.samples.push_back(QueueSample {
observed_at,
bytes,
count,
});
while self
.samples
.front()
.is_some_and(|sample| observed_at.duration_since(sample.observed_at) > ROLLING_WINDOW)
{
self.samples.pop_front();
}
if self.samples.is_empty() {
self.avg = InQueueStats::default();
self.max = InQueueStats::default();
self.last_minute = InQueueStats::default();
return;
}
let sample_count = self.samples.len() as i64;
let total_bytes = self.samples.iter().map(|sample| sample.bytes).sum::<i64>();
let total_count = self.samples.iter().map(|sample| sample.count).sum::<i64>();
let max_bytes = self.samples.iter().map(|sample| sample.bytes).max().unwrap_or(0);
let max_count = self.samples.iter().map(|sample| sample.count).max().unwrap_or(0);
self.avg.bytes = total_bytes / sample_count;
self.avg.count = total_count / sample_count;
self.max.bytes = max_bytes;
self.max.count = max_count;
self.last_minute.bytes = self.avg.bytes;
self.last_minute.count = self.avg.count;
}
fn snapshot(&self) -> Self {
let mut snapshot = self.clone();
snapshot.curr.bytes = snapshot.curr.now_bytes.load(Ordering::Relaxed);
snapshot.curr.count = snapshot.curr.now_count.load(Ordering::Relaxed);
snapshot
}
pub fn merge(&self, other: &InQueueMetric) -> Self {
Self {
curr: InQueueStats {
@@ -384,6 +446,12 @@ impl InQueueMetric {
count: self.max.count.max(other.max.count),
..Default::default()
},
last_minute: InQueueStats {
bytes: self.last_minute.bytes + other.last_minute.bytes,
count: self.last_minute.count + other.last_minute.count,
..Default::default()
},
samples: VecDeque::new(),
}
}
}
@@ -391,8 +459,8 @@ impl InQueueMetric {
/// Queue cache
#[derive(Debug, Default)]
pub struct QueueCache {
pub bucket_stats: HashMap<String, InQueueStats>,
pub sr_queue_stats: InQueueStats,
pub bucket_stats: HashMap<String, InQueueMetric>,
pub sr_queue_stats: InQueueMetric,
}
impl QueueCache {
@@ -401,36 +469,19 @@ impl QueueCache {
}
pub fn update(&mut self) {
// Update queue statistics cache
// In actual implementation, this would get latest statistics from queue system
let observed_at = Instant::now();
self.sr_queue_stats.observe(observed_at);
for stats in self.bucket_stats.values_mut() {
stats.observe(observed_at);
}
}
pub fn get_bucket_stats(&self, bucket: &str) -> InQueueMetric {
if let Some(bucket_stat) = self.bucket_stats.get(bucket) {
InQueueMetric {
curr: InQueueStats {
bytes: bucket_stat.now_bytes.load(Ordering::Relaxed),
count: bucket_stat.now_count.load(Ordering::Relaxed),
..Default::default()
},
avg: InQueueStats::default(), // simplified implementation
max: InQueueStats::default(), // simplified implementation
}
} else {
InQueueMetric::default()
}
self.bucket_stats.get(bucket).map(InQueueMetric::snapshot).unwrap_or_default()
}
pub fn get_site_stats(&self) -> InQueueMetric {
InQueueMetric {
curr: InQueueStats {
bytes: self.sr_queue_stats.now_bytes.load(Ordering::Relaxed),
count: self.sr_queue_stats.now_count.load(Ordering::Relaxed),
..Default::default()
},
avg: InQueueStats::default(), // simplified implementation
max: InQueueStats::default(), // simplified implementation
}
self.sr_queue_stats.snapshot()
}
}
@@ -505,11 +556,19 @@ impl ProxyStatsCache {
}
}
#[derive(Debug, Clone)]
struct FailureSample {
observed_at: Instant,
size: i64,
}
/// Failure statistics
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct FailStats {
pub count: i64,
pub size: i64,
#[serde(skip)]
recent: VecDeque<FailureSample>,
}
impl FailStats {
@@ -518,14 +577,42 @@ impl FailStats {
}
pub fn add_size(&mut self, size: i64, _err: Option<&Error>) {
let observed_at = Instant::now();
self.count += 1;
self.size += size;
self.recent.push_back(FailureSample { observed_at, size });
self.prune(observed_at);
}
fn prune(&mut self, observed_at: Instant) {
while self
.recent
.front()
.is_some_and(|sample| observed_at.duration_since(sample.observed_at) > FAILURE_LAST_HOUR_WINDOW)
{
self.recent.pop_front();
}
}
pub fn recent_since(&self, window: Duration) -> FailedMetric {
let now = Instant::now();
let mut count = 0i64;
let mut size = 0i64;
for sample in self.recent.iter().rev() {
if now.duration_since(sample.observed_at) > window {
break;
}
count += 1;
size += sample.size;
}
FailedMetric { count, size }
}
pub fn merge(&self, other: &FailStats) -> Self {
Self {
count: self.count + other.count,
size: self.size + other.size,
recent: VecDeque::new(),
}
}
@@ -674,6 +761,14 @@ pub struct ActiveWorkerStat {
pub curr: i32,
pub max: i32,
pub avg: f64,
#[serde(skip)]
samples: VecDeque<WorkerSample>,
}
#[derive(Debug, Clone)]
struct WorkerSample {
observed_at: Instant,
workers: i32,
}
impl ActiveWorkerStat {
@@ -685,9 +780,31 @@ impl ActiveWorkerStat {
self.clone()
}
pub fn update(&mut self) {
// Simulate worker statistics update logic
// In actual implementation, this would get current active count from worker pool
pub fn update(&mut self, curr: i32) {
let observed_at = Instant::now();
self.curr = curr;
self.samples.push_back(WorkerSample {
observed_at,
workers: curr,
});
while self
.samples
.front()
.is_some_and(|sample| observed_at.duration_since(sample.observed_at) > ROLLING_WINDOW)
{
self.samples.pop_front();
}
if self.samples.is_empty() {
self.max = curr;
self.avg = curr as f64;
return;
}
self.max = self.samples.iter().map(|sample| sample.workers).max().unwrap_or(curr);
let total = self.samples.iter().map(|sample| sample.workers as i64).sum::<i64>();
self.avg = total as f64 / self.samples.len() as f64;
}
}
@@ -740,8 +857,11 @@ impl ReplicationStats {
let mut interval = interval(Duration::from_secs(2));
loop {
interval.tick().await;
let current = get_global_replication_pool()
.map(|pool| pool.active_workers() + pool.active_lrg_workers() + pool.active_mrf_workers())
.unwrap_or(0);
let mut workers = workers_clone.lock().await;
workers.update();
workers.update(current);
}
});
@@ -925,14 +1045,26 @@ impl ReplicationStats {
/// Get replication metrics for all buckets
pub async fn get_all(&self) -> HashMap<String, BucketReplicationStats> {
let cache = self.cache.read().await;
let mut result = HashMap::new();
let mut result = HashMap::with_capacity(cache.len());
for (bucket, stats) in cache.iter() {
let mut cloned_stats = stats.clone_stats();
// Add queue statistics
result.insert(bucket.clone(), stats.clone_stats());
}
drop(cache);
{
let q_cache = self.q_cache.lock().await;
cloned_stats.q_stat = q_cache.get_bucket_stats(bucket);
result.insert(bucket.clone(), cloned_stats);
for (bucket, queue_stats) in &q_cache.bucket_stats {
let bucket_stats = result.entry(bucket.clone()).or_insert_with(BucketReplicationStats::new);
bucket_stats.q_stat = queue_stats.snapshot();
}
}
{
let p_cache = self.p_cache.lock().await;
for bucket in p_cache.bucket_stats.keys() {
result.entry(bucket.clone()).or_insert_with(BucketReplicationStats::new);
}
}
result
@@ -1114,12 +1246,12 @@ impl ReplicationStats {
let stats = q_cache
.bucket_stats
.entry(bucket.to_string())
.or_insert_with(InQueueStats::default);
stats.now_bytes.fetch_add(size, Ordering::Relaxed);
stats.now_count.fetch_add(1, Ordering::Relaxed);
.or_insert_with(InQueueMetric::default);
stats.curr.now_bytes.fetch_add(size, Ordering::Relaxed);
stats.curr.now_count.fetch_add(1, Ordering::Relaxed);
q_cache.sr_queue_stats.now_bytes.fetch_add(size, Ordering::Relaxed);
q_cache.sr_queue_stats.now_count.fetch_add(1, Ordering::Relaxed);
q_cache.sr_queue_stats.curr.now_bytes.fetch_add(size, Ordering::Relaxed);
q_cache.sr_queue_stats.curr.now_count.fetch_add(1, Ordering::Relaxed);
}
/// Decrease queue statistics
@@ -1128,12 +1260,12 @@ impl ReplicationStats {
let stats = q_cache
.bucket_stats
.entry(bucket.to_string())
.or_insert_with(InQueueStats::default);
stats.now_bytes.fetch_sub(size, Ordering::Relaxed);
stats.now_count.fetch_sub(1, Ordering::Relaxed);
.or_insert_with(InQueueMetric::default);
stats.curr.now_bytes.fetch_sub(size, Ordering::Relaxed);
stats.curr.now_count.fetch_sub(1, Ordering::Relaxed);
q_cache.sr_queue_stats.now_bytes.fetch_sub(size, Ordering::Relaxed);
q_cache.sr_queue_stats.now_count.fetch_sub(1, Ordering::Relaxed);
q_cache.sr_queue_stats.curr.now_bytes.fetch_sub(size, Ordering::Relaxed);
q_cache.sr_queue_stats.curr.now_count.fetch_sub(1, Ordering::Relaxed);
}
/// Increase proxy metrics
@@ -1166,6 +1298,51 @@ mod tests {
assert_eq!(workers.curr, 0);
}
#[test]
fn test_in_queue_metric_observe_updates_rolling_stats() {
let mut metric = InQueueMetric::default();
metric.curr.now_bytes.store(128, Ordering::Relaxed);
metric.curr.now_count.store(4, Ordering::Relaxed);
metric.observe(Instant::now());
metric.curr.now_bytes.store(256, Ordering::Relaxed);
metric.curr.now_count.store(6, Ordering::Relaxed);
metric.observe(Instant::now());
assert_eq!(metric.curr.bytes, 256);
assert_eq!(metric.curr.count, 6);
assert_eq!(metric.max.bytes, 256);
assert_eq!(metric.max.count, 6);
assert_eq!(metric.last_minute.bytes, 192);
assert_eq!(metric.last_minute.count, 5);
}
#[test]
fn test_fail_stats_recent_since_tracks_windows() {
let mut stats = FailStats::default();
stats.add_size(64, None);
stats.add_size(32, None);
let last_minute = stats.recent_since(Duration::from_secs(60));
let last_hour = stats.recent_since(Duration::from_secs(60 * 60));
assert_eq!(last_minute.count, 2);
assert_eq!(last_minute.size, 96);
assert_eq!(last_hour.count, 2);
assert_eq!(last_hour.size, 96);
}
#[test]
fn test_active_worker_stat_update_tracks_rolling_avg_and_max() {
let mut stats = ActiveWorkerStat::default();
stats.update(2);
stats.update(6);
stats.update(4);
assert_eq!(stats.curr, 4);
assert_eq!(stats.max, 6);
assert_eq!(stats.avg, 4.0);
}
#[tokio::test]
async fn test_delete_bucket_stats() {
let stats = ReplicationStats::new();
@@ -1218,6 +1395,15 @@ mod tests {
assert_eq!(stat.replicated_count, 1);
}
#[tokio::test]
async fn test_get_all_includes_proxy_only_bucket() {
let stats = ReplicationStats::new();
stats.inc_proxy("proxy-only-bucket", "HeadObject", false).await;
let all = stats.get_all().await;
assert!(all.contains_key("proxy-only-bucket"));
}
#[test]
fn test_sr_stats() {
let sr_stats = SRStats::new();
crates/io-core/src/pool.rs
+133 -74
View File
@@ -84,6 +84,12 @@ struct PoolTier {
available_buffers: Mutex<Vec<BytesMut>>,
/// Metrics for tracking this tier
metrics: Mutex<Option<Arc<BytesPoolMetrics>>>,
/// Total acquisitions for this tier
tier_total_acquires: AtomicU64,
/// Total hits for this tier
tier_pool_hits: AtomicU64,
/// Current allocated bytes for this tier
tier_current_allocated_bytes: AtomicU64,
}
/// Pool metrics for monitoring and optimization.
@@ -291,6 +297,9 @@ impl PoolTier {
name,
available_buffers: Mutex::new(Vec::new()),
metrics: Mutex::new(None),
tier_total_acquires: AtomicU64::new(0),
tier_pool_hits: AtomicU64::new(0),
tier_current_allocated_bytes: AtomicU64::new(0),
}
}
@@ -298,6 +307,65 @@ impl PoolTier {
*self.metrics.lock().unwrap() = Some(metrics);
}
fn take_or_allocate_buffer(&self, size: usize, pool_metrics: &BytesPoolMetrics) -> (BytesMut, bool) {
let buffer_opt = {
let mut available = self.available_buffers.lock().unwrap();
available.pop()
};
let was_reused = buffer_opt.is_some();
let buffer = if let Some(mut buf) = buffer_opt {
let previous_capacity = buf.capacity();
buf.clear();
if previous_capacity < size {
buf.reserve(size - previous_capacity);
}
let current_capacity = buf.capacity();
if current_capacity > previous_capacity {
let delta = (current_capacity - previous_capacity) as u64;
pool_metrics.total_bytes_allocated.fetch_add(delta, Ordering::Relaxed);
pool_metrics.current_allocated_bytes.fetch_add(delta, Ordering::Relaxed);
self.tier_current_allocated_bytes.fetch_add(delta, Ordering::Relaxed);
}
buf
} else {
let buf = BytesMut::with_capacity(size.max(self.buffer_size));
let allocated_bytes = buf.capacity() as u64;
pool_metrics
.total_bytes_allocated
.fetch_add(allocated_bytes, Ordering::Relaxed);
pool_metrics
.current_allocated_bytes
.fetch_add(allocated_bytes, Ordering::Relaxed);
self.tier_current_allocated_bytes
.fetch_add(allocated_bytes, Ordering::Relaxed);
buf
};
(buffer, was_reused)
}
fn record_acquire_metrics(&self, pool_metrics: &BytesPoolMetrics, buffer_capacity: usize, was_reused: bool) {
rustfs_io_metrics::record_bytes_pool_acquire(self.name, buffer_capacity, was_reused);
if was_reused {
pool_metrics.pool_hits.fetch_add(1, Ordering::Relaxed);
self.tier_pool_hits.fetch_add(1, Ordering::Relaxed);
} else {
pool_metrics.pool_misses.fetch_add(1, Ordering::Relaxed);
}
let tier_total_acquires = self.tier_total_acquires.load(Ordering::Relaxed);
let tier_pool_hits = self.tier_pool_hits.load(Ordering::Relaxed);
let tier_hit_rate = if tier_total_acquires == 0 {
0.0
} else {
tier_pool_hits as f64 / tier_total_acquires as f64
};
rustfs_io_metrics::record_bytes_pool_hit_rate(self.name, tier_hit_rate);
rustfs_io_metrics::record_bytes_pool_allocated(self.name, self.tier_current_allocated_bytes.load(Ordering::Relaxed));
}
async fn acquire_buffer(&self, size: usize, pool_metrics: &BytesPoolMetrics) -> PooledBuffer {
// Acquire semaphore permit (owned for storage in PooledBuffer)
let permit = Arc::clone(&self.semaphore).acquire_owned().await.unwrap();
@@ -308,45 +376,11 @@ impl PoolTier {
// Record acquisition
pool_metrics.total_acquires.fetch_add(1, Ordering::Relaxed);
self.tier_total_acquires.fetch_add(1, Ordering::Relaxed);
// Try to get a buffer from the pool
let buffer_opt = {
let mut available = self.available_buffers.lock().unwrap();
available.pop()
};
let was_reused = buffer_opt.is_some();
let buffer = if let Some(mut buf) = buffer_opt {
// Reuse existing buffer - clear and ensure capacity
buf.clear();
if buf.capacity() < size {
buf.reserve(size - buf.capacity());
}
buf
} else {
// Allocate new buffer
let buf = BytesMut::with_capacity(size.max(self.buffer_size));
pool_metrics
.total_bytes_allocated
.fetch_add(buf.capacity() as u64, Ordering::Relaxed);
pool_metrics
.current_allocated_bytes
.fetch_add(buf.capacity() as u64, Ordering::Relaxed);
buf
};
let (buffer, was_reused) = self.take_or_allocate_buffer(size, pool_metrics);
let buffer_capacity = buffer.capacity();
// Record metrics
rustfs_io_metrics::record_bytes_pool_acquire(self.name, buffer_capacity, was_reused);
// Record hit/miss (pool_metrics and metrics point to same Arc)
if was_reused {
pool_metrics.pool_hits.fetch_add(1, Ordering::Relaxed);
} else {
pool_metrics.pool_misses.fetch_add(1, Ordering::Relaxed);
}
self.record_acquire_metrics(pool_metrics, buffer_capacity, was_reused);
PooledBuffer {
buffer: ManuallyDrop::new(buffer),
@@ -365,45 +399,11 @@ impl PoolTier {
// Record acquisition
pool_metrics.total_acquires.fetch_add(1, Ordering::Relaxed);
self.tier_total_acquires.fetch_add(1, Ordering::Relaxed);
// Try to get a buffer from the pool
let buffer_opt = {
let mut available = self.available_buffers.lock().unwrap();
available.pop()
};
let was_reused = buffer_opt.is_some();
let buffer = if let Some(mut buf) = buffer_opt {
// Reuse existing buffer
buf.clear();
if buf.capacity() < size {
buf.reserve(size - buf.capacity());
}
buf
} else {
// Allocate new buffer
let buf = BytesMut::with_capacity(size.max(self.buffer_size));
pool_metrics
.total_bytes_allocated
.fetch_add(buf.capacity() as u64, Ordering::Relaxed);
pool_metrics
.current_allocated_bytes
.fetch_add(buf.capacity() as u64, Ordering::Relaxed);
buf
};
let (buffer, was_reused) = self.take_or_allocate_buffer(size, pool_metrics);
let buffer_capacity = buffer.capacity();
// Record metrics
rustfs_io_metrics::record_bytes_pool_acquire(self.name, buffer_capacity, was_reused);
// Record hit/miss (pool_metrics and metrics point to same Arc)
if was_reused {
pool_metrics.pool_hits.fetch_add(1, Ordering::Relaxed);
} else {
pool_metrics.pool_misses.fetch_add(1, Ordering::Relaxed);
}
self.record_acquire_metrics(pool_metrics, buffer_capacity, was_reused);
Some(PooledBuffer {
buffer: ManuallyDrop::new(buffer),
@@ -421,8 +421,24 @@ impl PoolTier {
if let Some(ref metrics) = *self.metrics.lock().unwrap() {
metrics.available_buffers.fetch_add(1, Ordering::Relaxed);
}
} else {
let released_bytes = buffer.capacity() as u64;
self.tier_current_allocated_bytes
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
Some(current.saturating_sub(released_bytes))
})
.ok();
if let Some(ref metrics) = *self.metrics.lock().unwrap() {
metrics
.current_allocated_bytes
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
Some(current.saturating_sub(released_bytes))
})
.ok();
}
}
// If pool is full, buffer is dropped and memory is freed
rustfs_io_metrics::record_bytes_pool_allocated(self.name, self.tier_current_allocated_bytes.load(Ordering::Relaxed));
}
}
@@ -617,4 +633,47 @@ mod tests {
let delta_hits = pool.metrics().pool_hits.load(Ordering::Relaxed) - initial_hits;
assert_eq!(delta_hits, 1);
}
#[tokio::test]
async fn test_tier_allocated_bytes_tracks_real_allocations() {
let pool = BytesPool::with_config(BytesPoolConfig {
small_size: 1024,
small_max: 2,
..Default::default()
});
// First acquire allocates one small-tier buffer.
let buf1 = pool.acquire_buffer(512).await;
assert_eq!(pool.small_pool.tier_current_allocated_bytes.load(Ordering::Relaxed), 1024);
// Return and reuse should not increase allocated bytes.
drop(buf1);
let buf2 = pool.acquire_buffer(512).await;
assert_eq!(pool.small_pool.tier_current_allocated_bytes.load(Ordering::Relaxed), 1024);
// A second in-flight buffer forces one more allocation.
let _buf3 = pool.acquire_buffer(512).await;
assert_eq!(pool.small_pool.tier_current_allocated_bytes.load(Ordering::Relaxed), 2048);
drop(buf2);
}
#[tokio::test]
async fn test_tier_hit_rate_counters_track_reuse() {
let pool = BytesPool::with_config(BytesPoolConfig {
small_size: 1024,
small_max: 2,
..Default::default()
});
// First acquire is miss.
let buf1 = pool.acquire_buffer(512).await;
drop(buf1);
// Second acquire reuses previous buffer and counts as hit.
let _buf2 = pool.acquire_buffer(512).await;
assert_eq!(pool.small_pool.tier_total_acquires.load(Ordering::Relaxed), 2);
assert_eq!(pool.small_pool.tier_pool_hits.load(Ordering::Relaxed), 1);
}
}
crates/obs/src/global.rs
+12 -12
View File
@@ -24,16 +24,16 @@ static GLOBAL_GUARD: OnceCell<Arc<Mutex<OtelGuard>>> = OnceCell::const_new();
pub(crate) static OBSERVABILITY_METRIC_ENABLED: OnceCell<bool> = OnceCell::const_new();
/// Namespaced metrics for cleaner and rolling logging.
pub(crate) const METRIC_LOG_CLEANER_DELETED_FILES_TOTAL: &str = "rustfs.log_cleaner.deleted_files_total";
pub(crate) const METRIC_LOG_CLEANER_FREED_BYTES_TOTAL: &str = "rustfs.log_cleaner.freed_bytes_total";
pub(crate) const METRIC_LOG_CLEANER_COMPRESS_DURATION_SECONDS: &str = "rustfs.log_cleaner.compress_duration_seconds";
pub(crate) const METRIC_LOG_CLEANER_STEAL_SUCCESS_RATE: &str = "rustfs.log_cleaner.steal_success_rate";
pub(crate) const METRIC_LOG_CLEANER_RUNS_TOTAL: &str = "rustfs.log_cleaner.runs_total";
pub(crate) const METRIC_LOG_CLEANER_RUN_FAILURES_TOTAL: &str = "rustfs.log_cleaner.run_failures_total";
pub(crate) const METRIC_LOG_CLEANER_ROTATION_TOTAL: &str = "rustfs.log_cleaner.rotation_total";
pub(crate) const METRIC_LOG_CLEANER_ROTATION_FAILURES_TOTAL: &str = "rustfs.log_cleaner.rotation_failures_total";
pub(crate) const METRIC_LOG_CLEANER_ROTATION_DURATION_SECONDS: &str = "rustfs.log_cleaner.rotation_duration_seconds";
pub(crate) const METRIC_LOG_CLEANER_ACTIVE_FILE_SIZE_BYTES: &str = "rustfs.log_cleaner.active_file_size_bytes";
pub(crate) const METRIC_LOG_CLEANER_DELETED_FILES_TOTAL: &str = "rustfs_log_cleaner_deleted_files_total";
pub(crate) const METRIC_LOG_CLEANER_FREED_BYTES_TOTAL: &str = "rustfs_log_cleaner_freed_bytes_total";
pub(crate) const METRIC_LOG_CLEANER_COMPRESS_DURATION_SECONDS: &str = "rustfs_log_cleaner_compress_duration_seconds";
pub(crate) const METRIC_LOG_CLEANER_STEAL_SUCCESS_RATE: &str = "rustfs_log_cleaner_steal_success_rate";
pub(crate) const METRIC_LOG_CLEANER_RUNS_TOTAL: &str = "rustfs_log_cleaner_runs_total";
pub(crate) const METRIC_LOG_CLEANER_RUN_FAILURES_TOTAL: &str = "rustfs_log_cleaner_run_failures_total";
pub(crate) const METRIC_LOG_CLEANER_ROTATION_TOTAL: &str = "rustfs_log_cleaner_rotation_total";
pub(crate) const METRIC_LOG_CLEANER_ROTATION_FAILURES_TOTAL: &str = "rustfs_log_cleaner_rotation_failures_total";
pub(crate) const METRIC_LOG_CLEANER_ROTATION_DURATION_SECONDS: &str = "rustfs_log_cleaner_rotation_duration_seconds";
pub(crate) const METRIC_LOG_CLEANER_ACTIVE_FILE_SIZE_BYTES: &str = "rustfs_log_cleaner_active_file_size_bytes";
/// Check whether Observability metric is enabled
pub fn observability_metric_enabled() -> bool {
@@ -198,8 +198,8 @@ mod tests {
for metric in metrics {
assert!(
metric.starts_with("rustfs.log_cleaner."),
"metric '{metric}' should use rustfs.log_cleaner.* namespace"
metric.starts_with("rustfs_log_cleaner_"),
"metric '{metric}' should use rustfs_log_cleaner_* namespace"
);
}
}
crates/obs/src/metrics/collectors/bucket_replication.rs
+243 -28
View File
@@ -12,34 +12,62 @@
// See the License for the specific language governing permissions and
// limitations under the License.
//! Bucket replication bandwidth metrics collector.
//!
//! Collects bandwidth metrics for bucket replication targets.
//!
//! This collector reuses the metric descriptors defined in `metrics_type::bucket_replication`
//! to avoid duplication of metric names, types, and help text.
//! Bucket replication metrics collector.
use crate::metrics::report::PrometheusMetric;
use crate::metrics::schema::bucket_replication::{BUCKET_REPL_BANDWIDTH_CURRENT_MD, BUCKET_REPL_BANDWIDTH_LIMIT_MD};
use crate::metrics::schema::bucket_replication::{
BUCKET_L, BUCKET_REPL_BANDWIDTH_CURRENT_MD, BUCKET_REPL_BANDWIDTH_LIMIT_MD, BUCKET_REPL_LAST_HR_FAILED_BYTES_MD,
BUCKET_REPL_LAST_HR_FAILED_COUNT_MD, BUCKET_REPL_LAST_MIN_FAILED_BYTES_MD, BUCKET_REPL_LAST_MIN_FAILED_COUNT_MD,
BUCKET_REPL_LATENCY_MS_MD, BUCKET_REPL_PROXIED_DELETE_TAGGING_REQUESTS_FAILURES_MD,
BUCKET_REPL_PROXIED_DELETE_TAGGING_REQUESTS_TOTAL_MD, BUCKET_REPL_PROXIED_GET_REQUESTS_FAILURES_MD,
BUCKET_REPL_PROXIED_GET_REQUESTS_TOTAL_MD, BUCKET_REPL_PROXIED_GET_TAGGING_REQUESTS_FAILURES_MD,
BUCKET_REPL_PROXIED_GET_TAGGING_REQUESTS_TOTAL_MD, BUCKET_REPL_PROXIED_HEAD_REQUESTS_FAILURES_MD,
BUCKET_REPL_PROXIED_HEAD_REQUESTS_TOTAL_MD, BUCKET_REPL_PROXIED_PUT_TAGGING_REQUESTS_FAILURES_MD,
BUCKET_REPL_PROXIED_PUT_TAGGING_REQUESTS_TOTAL_MD, BUCKET_REPL_SENT_BYTES_MD, BUCKET_REPL_SENT_COUNT_MD,
BUCKET_REPL_TOTAL_FAILED_BYTES_MD, BUCKET_REPL_TOTAL_FAILED_COUNT_MD, OPERATION_L, RANGE_L, TARGET_ARN_L,
};
use std::borrow::Cow;
/// Bucket replication bandwidth statistics for metrics collection.
#[derive(Debug, Clone, Default)]
pub struct BucketReplicationTargetStats {
pub target_arn: String,
pub bandwidth_limit_bytes_per_sec: u64,
pub current_bandwidth_bytes_per_sec: f64,
pub latency_ms: f64,
}
#[derive(Debug, Clone, Default)]
pub struct BucketReplicationBandwidthStats {
/// Name of the bucket
pub bucket: String,
/// Target ARN for replication
pub target_arn: String,
/// Configured bandwidth limit in bytes per second
pub limit_bytes_per_sec: u64,
/// Current bandwidth in bytes per second (EWMA)
pub current_bandwidth_bytes_per_sec: f64,
}
/// Collects bucket replication bandwidth metrics from the provided statistics.
///
/// Uses the metric descriptors from `metrics_type::bucket_replication` module.
/// Returns a vector of Prometheus metrics for replication bandwidth.
#[derive(Debug, Clone, Default)]
pub struct BucketReplicationStats {
pub bucket: String,
pub total_failed_bytes: u64,
pub total_failed_count: u64,
pub last_min_failed_bytes: u64,
pub last_min_failed_count: u64,
pub last_hour_failed_bytes: u64,
pub last_hour_failed_count: u64,
pub sent_bytes: u64,
pub sent_count: u64,
pub proxied_get_requests_total: u64,
pub proxied_get_requests_failures: u64,
pub proxied_head_requests_total: u64,
pub proxied_head_requests_failures: u64,
pub proxied_put_tagging_requests_total: u64,
pub proxied_put_tagging_requests_failures: u64,
pub proxied_get_tagging_requests_total: u64,
pub proxied_get_tagging_requests_failures: u64,
pub proxied_delete_tagging_requests_total: u64,
pub proxied_delete_tagging_requests_failures: u64,
pub targets: Vec<BucketReplicationTargetStats>,
}
pub fn collect_bucket_replication_bandwidth_metrics(stats: &[BucketReplicationBandwidthStats]) -> Vec<PrometheusMetric> {
if stats.is_empty() {
return Vec::new();
@@ -52,24 +80,206 @@ pub fn collect_bucket_replication_bandwidth_metrics(stats: &[BucketReplicationBa
metrics.push(
PrometheusMetric::from_descriptor(&BUCKET_REPL_BANDWIDTH_LIMIT_MD, stat.limit_bytes_per_sec as f64)
.with_label("bucket", bucket_label.clone())
.with_label("targetArn", target_arn_label.clone()),
.with_label(BUCKET_L, bucket_label.clone())
.with_label(TARGET_ARN_L, target_arn_label.clone()),
);
metrics.push(
PrometheusMetric::from_descriptor(&BUCKET_REPL_BANDWIDTH_CURRENT_MD, stat.current_bandwidth_bytes_per_sec)
.with_label("bucket", bucket_label)
.with_label("targetArn", target_arn_label),
.with_label(BUCKET_L, bucket_label)
.with_label(TARGET_ARN_L, target_arn_label),
);
}
metrics
}
pub fn collect_bucket_replication_metrics(stats: &[BucketReplicationStats]) -> Vec<PrometheusMetric> {
if stats.is_empty() {
return Vec::new();
}
let mut metrics = Vec::new();
for stat in stats {
let bucket_label: Cow<'static, str> = Cow::Owned(stat.bucket.clone());
metrics.push(
PrometheusMetric::from_descriptor(&BUCKET_REPL_TOTAL_FAILED_BYTES_MD, stat.total_failed_bytes as f64)
.with_label(BUCKET_L, bucket_label.clone()),
);
metrics.push(
PrometheusMetric::from_descriptor(&BUCKET_REPL_TOTAL_FAILED_COUNT_MD, stat.total_failed_count as f64)
.with_label(BUCKET_L, bucket_label.clone()),
);
metrics.push(
PrometheusMetric::from_descriptor(&BUCKET_REPL_LAST_MIN_FAILED_BYTES_MD, stat.last_min_failed_bytes as f64)
.with_label(BUCKET_L, bucket_label.clone()),
);
metrics.push(
PrometheusMetric::from_descriptor(&BUCKET_REPL_LAST_MIN_FAILED_COUNT_MD, stat.last_min_failed_count as f64)
.with_label(BUCKET_L, bucket_label.clone()),
);
metrics.push(
PrometheusMetric::from_descriptor(&BUCKET_REPL_LAST_HR_FAILED_BYTES_MD, stat.last_hour_failed_bytes as f64)
.with_label(BUCKET_L, bucket_label.clone()),
);
metrics.push(
PrometheusMetric::from_descriptor(&BUCKET_REPL_LAST_HR_FAILED_COUNT_MD, stat.last_hour_failed_count as f64)
.with_label(BUCKET_L, bucket_label.clone()),
);
metrics.push(
PrometheusMetric::from_descriptor(&BUCKET_REPL_SENT_BYTES_MD, stat.sent_bytes as f64)
.with_label(BUCKET_L, bucket_label.clone()),
);
metrics.push(
PrometheusMetric::from_descriptor(&BUCKET_REPL_SENT_COUNT_MD, stat.sent_count as f64)
.with_label(BUCKET_L, bucket_label.clone()),
);
metrics.push(
PrometheusMetric::from_descriptor(&BUCKET_REPL_PROXIED_GET_REQUESTS_TOTAL_MD, stat.proxied_get_requests_total as f64)
.with_label(BUCKET_L, bucket_label.clone()),
);
metrics.push(
PrometheusMetric::from_descriptor(
&BUCKET_REPL_PROXIED_GET_REQUESTS_FAILURES_MD,
stat.proxied_get_requests_failures as f64,
)
.with_label(BUCKET_L, bucket_label.clone()),
);
metrics.push(
PrometheusMetric::from_descriptor(
&BUCKET_REPL_PROXIED_HEAD_REQUESTS_TOTAL_MD,
stat.proxied_head_requests_total as f64,
)
.with_label(BUCKET_L, bucket_label.clone()),
);
metrics.push(
PrometheusMetric::from_descriptor(
&BUCKET_REPL_PROXIED_HEAD_REQUESTS_FAILURES_MD,
stat.proxied_head_requests_failures as f64,
)
.with_label(BUCKET_L, bucket_label.clone()),
);
metrics.push(
PrometheusMetric::from_descriptor(
&BUCKET_REPL_PROXIED_PUT_TAGGING_REQUESTS_TOTAL_MD,
stat.proxied_put_tagging_requests_total as f64,
)
.with_label(BUCKET_L, bucket_label.clone()),
);
metrics.push(
PrometheusMetric::from_descriptor(
&BUCKET_REPL_PROXIED_PUT_TAGGING_REQUESTS_FAILURES_MD,
stat.proxied_put_tagging_requests_failures as f64,
)
.with_label(BUCKET_L, bucket_label.clone()),
);
metrics.push(
PrometheusMetric::from_descriptor(
&BUCKET_REPL_PROXIED_GET_TAGGING_REQUESTS_TOTAL_MD,
stat.proxied_get_tagging_requests_total as f64,
)
.with_label(BUCKET_L, bucket_label.clone()),
);
metrics.push(
PrometheusMetric::from_descriptor(
&BUCKET_REPL_PROXIED_GET_TAGGING_REQUESTS_FAILURES_MD,
stat.proxied_get_tagging_requests_failures as f64,
)
.with_label(BUCKET_L, bucket_label.clone()),
);
metrics.push(
PrometheusMetric::from_descriptor(
&BUCKET_REPL_PROXIED_DELETE_TAGGING_REQUESTS_TOTAL_MD,
stat.proxied_delete_tagging_requests_total as f64,
)
.with_label(BUCKET_L, bucket_label.clone()),
);
metrics.push(
PrometheusMetric::from_descriptor(
&BUCKET_REPL_PROXIED_DELETE_TAGGING_REQUESTS_FAILURES_MD,
stat.proxied_delete_tagging_requests_failures as f64,
)
.with_label(BUCKET_L, bucket_label.clone()),
);
for target in &stat.targets {
let target_label: Cow<'static, str> = Cow::Owned(target.target_arn.clone());
metrics.push(
PrometheusMetric::from_descriptor(&BUCKET_REPL_LATENCY_MS_MD, target.latency_ms)
.with_label(BUCKET_L, bucket_label.clone())
.with_label(OPERATION_L, Cow::Borrowed("object_replication"))
.with_label(RANGE_L, Cow::Borrowed("all"))
.with_label(TARGET_ARN_L, target_label),
);
}
}
metrics
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_collect_bucket_replication_metrics() {
let stats = vec![BucketReplicationStats {
bucket: "b1".to_string(),
total_failed_bytes: 64,
total_failed_count: 2,
last_min_failed_bytes: 32,
last_min_failed_count: 1,
last_hour_failed_bytes: 64,
last_hour_failed_count: 2,
sent_bytes: 1024,
sent_count: 8,
proxied_get_requests_total: 5,
proxied_get_requests_failures: 1,
proxied_head_requests_total: 4,
proxied_head_requests_failures: 0,
proxied_put_tagging_requests_total: 3,
proxied_put_tagging_requests_failures: 1,
proxied_get_tagging_requests_total: 2,
proxied_get_tagging_requests_failures: 0,
proxied_delete_tagging_requests_total: 1,
proxied_delete_tagging_requests_failures: 0,
targets: vec![BucketReplicationTargetStats {
target_arn: "arn:rustfs:replication:us-east-1:1:target".to_string(),
bandwidth_limit_bytes_per_sec: 2048,
current_bandwidth_bytes_per_sec: 1024.0,
latency_ms: 15.0,
}],
}];
let metrics = collect_bucket_replication_metrics(&stats);
assert_eq!(metrics.len(), 19);
let sent_name = BUCKET_REPL_SENT_COUNT_MD.get_full_metric_name();
assert!(metrics.iter().any(|metric| {
metric.name == sent_name
&& metric.value == 8.0
&& metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1")
}));
let latency_name = BUCKET_REPL_LATENCY_MS_MD.get_full_metric_name();
assert!(metrics.iter().any(|metric| {
metric.name == latency_name
&& metric.value == 15.0
&& metric
.labels
.iter()
.any(|(key, value)| *key == TARGET_ARN_L && value == "arn:rustfs:replication:us-east-1:1:target")
}));
}
#[test]
fn test_collect_bucket_replication_metrics_empty() {
let stats: Vec<BucketReplicationStats> = Vec::new();
let metrics = collect_bucket_replication_metrics(&stats);
assert!(metrics.is_empty());
}
#[test]
fn test_collect_bucket_replication_bandwidth_metrics() {
let stats = vec![BucketReplicationBandwidthStats {
@@ -83,24 +293,29 @@ mod tests {
assert_eq!(metrics.len(), 2);
let limit_metric_name = BUCKET_REPL_BANDWIDTH_LIMIT_MD.get_full_metric_name();
let limit_metric = metrics.iter().find(|m| {
m.name == limit_metric_name && m.value == 1_048_576.0 && m.labels.iter().any(|(k, v)| *k == "bucket" && v == "b1")
let limit_metric = metrics.iter().find(|metric| {
metric.name == limit_metric_name
&& metric.value == 1_048_576.0
&& metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1")
});
assert!(limit_metric.is_some());
assert!(
limit_metric
.and_then(|m| {
m.labels
.and_then(|metric| {
metric
.labels
.iter()
.find(|(k, _)| *k == "targetArn")
.map(|(_, v)| v.as_ref() == "arn:rustfs:replication:us-east-1:1:test-2")
.find(|(key, _)| *key == TARGET_ARN_L)
.map(|(_, value)| value.as_ref() == "arn:rustfs:replication:us-east-1:1:test-2")
})
.unwrap_or(false)
);
let current_metric_name = BUCKET_REPL_BANDWIDTH_CURRENT_MD.get_full_metric_name();
let current_metric = metrics.iter().find(|m| {
m.name == current_metric_name && m.value == 204_800.0 && m.labels.iter().any(|(k, v)| *k == "bucket" && v == "b1")
let current_metric = metrics.iter().find(|metric| {
metric.name == current_metric_name
&& metric.value == 204_800.0
&& metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1")
});
assert!(current_metric.is_some());
}
crates/obs/src/metrics/collectors/mod.rs
+4 -1
View File
@@ -41,7 +41,10 @@ pub mod system_process;
pub use audit::{AuditTargetStats, collect_audit_metrics};
pub use bucket::{BucketStats, collect_bucket_metrics};
pub use bucket_replication::{BucketReplicationBandwidthStats, collect_bucket_replication_bandwidth_metrics};
pub use bucket_replication::{
BucketReplicationBandwidthStats, BucketReplicationStats, BucketReplicationTargetStats,
collect_bucket_replication_bandwidth_metrics, collect_bucket_replication_metrics,
};
pub use cluster::{ClusterStats, collect_cluster_metrics};
pub use cluster_config::{ClusterConfigStats, collect_cluster_config_metrics};
pub use cluster_erasure_set::{ErasureSetStats, collect_erasure_set_metrics};
crates/obs/src/metrics/collectors/replication.rs
+22 -2
View File
@@ -25,6 +25,14 @@ use crate::metrics::schema::replication::*;
/// Replication statistics.
#[derive(Debug, Clone, Default)]
pub struct ReplicationStats {
/// Average number of active replication workers
pub average_active_workers: f64,
/// Average queued bytes since server start
pub average_queued_bytes: i64,
/// Average queued objects since server start
pub average_queued_count: i64,
/// Average data transfer rate in bytes/sec
pub average_data_transfer_rate: f64,
/// Number of active replication workers
pub active_workers: u64,
/// Current data transfer rate in bytes/sec
@@ -50,6 +58,10 @@ pub struct ReplicationStats {
/// Returns a vector of Prometheus metrics for replication statistics.
pub fn collect_replication_metrics(stats: &ReplicationStats) -> Vec<PrometheusMetric> {
vec![
PrometheusMetric::from_descriptor(&REPLICATION_AVERAGE_ACTIVE_WORKERS_MD, stats.average_active_workers),
PrometheusMetric::from_descriptor(&REPLICATION_AVERAGE_QUEUED_BYTES_MD, stats.average_queued_bytes as f64),
PrometheusMetric::from_descriptor(&REPLICATION_AVERAGE_QUEUED_COUNT_MD, stats.average_queued_count as f64),
PrometheusMetric::from_descriptor(&REPLICATION_AVERAGE_DATA_TRANSFER_RATE_MD, stats.average_data_transfer_rate),
PrometheusMetric::from_descriptor(&REPLICATION_CURRENT_ACTIVE_WORKERS_MD, stats.active_workers as f64),
PrometheusMetric::from_descriptor(&REPLICATION_CURRENT_DATA_TRANSFER_RATE_MD, stats.current_data_transfer_rate),
PrometheusMetric::from_descriptor(&REPLICATION_LAST_MINUTE_QUEUED_BYTES_MD, stats.last_minute_queued_bytes as f64),
@@ -70,6 +82,10 @@ mod tests {
#[test]
fn test_collect_replication_metrics() {
let stats = ReplicationStats {
average_active_workers: 8.5,
average_queued_bytes: 1024 * 1024 * 40,
average_queued_count: 240,
average_data_transfer_rate: 1024.0 * 1024.0 * 3.0,
active_workers: 10,
current_data_transfer_rate: 1024.0 * 1024.0 * 5.0, // 5 MB/s
last_minute_queued_bytes: 1024 * 1024 * 100, // 100 MB
@@ -84,13 +100,17 @@ mod tests {
let metrics = collect_replication_metrics(&stats);
report_metrics(&metrics);
assert_eq!(metrics.len(), 9);
assert_eq!(metrics.len(), 13);
// Verify active workers
let active_name = REPLICATION_CURRENT_ACTIVE_WORKERS_MD.get_full_metric_name();
let active = metrics.iter().find(|m| m.name == active_name);
assert!(active.is_some());
assert_eq!(active.map(|m| m.value), Some(10.0));
let avg_active_name = REPLICATION_AVERAGE_ACTIVE_WORKERS_MD.get_full_metric_name();
let avg_active = metrics.iter().find(|m| m.name == avg_active_name);
assert_eq!(avg_active.map(|m| m.value), Some(8.5));
}
#[test]
@@ -98,7 +118,7 @@ mod tests {
let stats = ReplicationStats::default();
let metrics = collect_replication_metrics(&stats);
assert_eq!(metrics.len(), 9);
assert_eq!(metrics.len(), 13);
for metric in &metrics {
assert_eq!(metric.value, 0.0);
assert!(metric.labels.is_empty());
crates/obs/src/metrics/scheduler.rs
+30 -9
View File
@@ -35,8 +35,14 @@ use crate::metrics::collectors::{
collect_audit_metrics,
collect_bucket_metrics,
collect_bucket_replication_bandwidth_metrics,
collect_bucket_replication_metrics,
collect_cluster_health_metrics,
collect_cluster_metrics,
collect_cpu_metrics,
collect_drive_count_metrics,
collect_drive_detailed_metrics,
collect_host_network_metrics,
collect_memory_metrics,
collect_node_metrics,
collect_notification_metrics,
collect_notification_target_metrics,
@@ -45,6 +51,7 @@ use crate::metrics::collectors::{
collect_process_disk_metrics,
collect_process_memory_metrics,
collect_process_metrics,
collect_replication_metrics,
collect_resource_metrics,
};
use crate::metrics::config::{
@@ -56,14 +63,16 @@ use crate::metrics::config::{
};
use crate::metrics::report::{PrometheusMetric, report_metrics};
use crate::metrics::stats_collector::{
ProcessMetricBundle, collect_bucket_replication_bandwidth_stats, collect_bucket_stats, collect_cluster_stats,
collect_disk_stats, collect_host_network_stats, collect_process_metric_bundle,
ProcessMetricBundle, collect_bucket_replication_bandwidth_stats, collect_bucket_replication_detail_stats,
collect_bucket_stats, collect_cluster_and_health_stats, collect_disk_and_system_drive_stats, collect_host_network_stats,
collect_process_metric_bundle, collect_replication_stats, collect_system_cpu_and_memory_stats_with,
};
use rustfs_audit::audit_target_metrics;
use rustfs_notify::{notification_metrics_snapshot, notification_target_metrics};
use rustfs_utils::get_env_opt_u64;
use std::borrow::Cow;
use std::time::Duration;
use sysinfo::System;
use tokio::time::Instant;
use tokio_util::sync::CancellationToken;
use tracing::warn;
@@ -154,8 +163,9 @@ pub fn init_metrics_runtime(token: CancellationToken) {
loop {
tokio::select! {
_ = interval.tick() => {
let stats = collect_cluster_stats().await;
let metrics = collect_cluster_metrics(&stats);
let (stats, cluster_health) = collect_cluster_and_health_stats().await;
let mut metrics = collect_cluster_metrics(&stats);
metrics.extend(collect_cluster_health_metrics(&cluster_health));
report_metrics(&metrics);
}
_ = token_clone.cancelled() => {
@@ -192,8 +202,10 @@ pub fn init_metrics_runtime(token: CancellationToken) {
loop {
tokio::select! {
_ = interval.tick() => {
let stats = collect_disk_stats().await;
let metrics = collect_node_metrics(&stats);
let (disk_stats, drive_stats, drive_counts) = collect_disk_and_system_drive_stats().await;
let mut metrics = collect_node_metrics(&disk_stats);
metrics.extend(collect_drive_detailed_metrics(&drive_stats));
metrics.extend(collect_drive_count_metrics(&drive_counts));
report_metrics(&metrics);
}
_ = token_clone.cancelled() => {
@@ -212,7 +224,11 @@ pub fn init_metrics_runtime(token: CancellationToken) {
tokio::select! {
_ = interval.tick() => {
let stats = collect_bucket_replication_bandwidth_stats();
let metrics = collect_bucket_replication_bandwidth_metrics(&stats);
let mut metrics = collect_bucket_replication_bandwidth_metrics(&stats);
let bucket_replication = collect_bucket_replication_detail_stats().await;
metrics.extend(collect_bucket_replication_metrics(&bucket_replication));
let replication = collect_replication_stats().await;
metrics.extend(collect_replication_metrics(&replication));
report_metrics(&metrics);
}
_ = token_clone.cancelled() => {
@@ -297,6 +313,7 @@ pub fn init_metrics_runtime(token: CancellationToken) {
let token_clone = token;
tokio::spawn(async move {
let labels = current_process_metric_labels();
let mut host_system = System::new_all();
let process_interval = resource_interval.min(system_interval);
let mut interval = tokio::time::interval(process_interval);
let now = Instant::now();
@@ -327,9 +344,9 @@ pub fn init_metrics_runtime(token: CancellationToken) {
if now >= next_system_run {
#[cfg(feature = "gpu")]
let mut metrics = collect_system_monitoring_metrics(&bundle, &labels);
let mut metrics = collect_system_monitoring_metrics(&bundle, &labels, &mut host_system);
#[cfg(not(feature = "gpu"))]
let metrics = collect_system_monitoring_metrics(&bundle, &labels);
let metrics = collect_system_monitoring_metrics(&bundle, &labels, &mut host_system);
#[cfg(feature = "gpu")]
if let Some(pid) = current_pid {
@@ -418,6 +435,7 @@ fn fallback_process_metric_labels(err: ProcessAttributeError) -> Vec<(&'static s
fn collect_system_monitoring_metrics(
bundle: &ProcessMetricBundle,
labels: &[(&'static str, Cow<'static, str>)],
host_system: &mut System,
) -> Vec<PrometheusMetric> {
let cpu_stats = ProcessCpuStats {
usage: bundle.resource.cpu_percent,
@@ -432,8 +450,11 @@ fn collect_system_monitoring_metrics(
written_bytes: bundle.disk_write_bytes,
};
let network_stats = collect_host_network_stats();
let (system_cpu_stats, system_memory_stats) = collect_system_cpu_and_memory_stats_with(host_system);
let mut metrics = Vec::new();
metrics.extend(collect_cpu_metrics(&system_cpu_stats));
metrics.extend(collect_memory_metrics(&system_memory_stats));
metrics.extend(collect_process_cpu_metrics(&cpu_stats, Some(labels)));
metrics.extend(collect_process_memory_metrics(&memory_stats, Some(labels)));
metrics.extend(collect_process_disk_metrics(&disk_stats, Some(labels)));
crates/obs/src/metrics/schema/bucket_replication.rs
+1 -1
View File
@@ -22,7 +22,7 @@ pub const BUCKET_L: &str = "bucket";
/// Replication operation
pub const OPERATION_L: &str = "operation";
/// Replication target ARN
pub const TARGET_ARN_L: &str = "targetArn";
pub const TARGET_ARN_L: &str = "target_arn";
/// Replication range
pub const RANGE_L: &str = "range";
crates/obs/src/metrics/stats_collector.rs
+338 -25
View File
@@ -21,19 +21,41 @@
//! and convert them to the Stats structs used by collectors.
use crate::metrics::collectors::{
BucketReplicationBandwidthStats, BucketStats, ClusterStats, DiskStats, HostNetworkStats, ProcessStats, ProcessStatusType,
ResourceStats,
BucketReplicationBandwidthStats, BucketReplicationStats, BucketReplicationTargetStats, BucketStats, ClusterHealthStats,
ClusterStats, CpuStats, DiskStats, DriveCountStats, DriveDetailedStats, HostNetworkStats, MemoryStats, ProcessStats,
ProcessStatusType, ReplicationStats, ResourceStats,
};
use rustfs_ecstore::bucket::metadata_sys::get_quota_config;
use rustfs_ecstore::bucket::replication::GLOBAL_REPLICATION_STATS;
use rustfs_ecstore::data_usage::load_data_usage_from_backend;
use rustfs_ecstore::global::get_global_bucket_monitor;
use rustfs_ecstore::pools::{get_total_usable_capacity, get_total_usable_capacity_free};
use rustfs_ecstore::store_api::{BucketOperations, BucketOptions};
use rustfs_ecstore::{StorageAPI, new_object_layer_fn};
use rustfs_io_metrics::{ProcessStatusSnapshot, snapshot_process_resource_and_system};
use sysinfo::Networks;
use std::time::Duration;
use sysinfo::{Networks, System};
use tracing::{instrument, warn};
const DRIVE_STATE_OK: &str = "ok";
const DRIVE_STATE_ONLINE: &str = "online";
const DRIVE_STATE_UNFORMATTED: &str = "unformatted";
const DRIVE_RUNTIME_STATE_RETURNING: &str = "returning";
fn disk_is_online_for_metrics(state: &str, runtime_state: Option<&str>) -> bool {
let state_is_acceptable = state.eq_ignore_ascii_case(DRIVE_STATE_OK)
|| state.eq_ignore_ascii_case(DRIVE_STATE_ONLINE)
|| state.eq_ignore_ascii_case(DRIVE_STATE_UNFORMATTED);
if let Some(runtime_state) = runtime_state {
let runtime_state_is_acceptable = runtime_state.eq_ignore_ascii_case(DRIVE_STATE_ONLINE)
|| runtime_state.eq_ignore_ascii_case(DRIVE_RUNTIME_STATE_RETURNING);
return runtime_state_is_acceptable && state_is_acceptable;
}
state_is_acceptable
}
#[derive(Debug, Clone, Default)]
pub struct ProcessMetricBundle {
pub resource: ResourceStats,
@@ -42,48 +64,76 @@ pub struct ProcessMetricBundle {
pub disk_write_bytes: u64,
}
/// Collect cluster statistics from the storage layer.
#[instrument]
pub async fn collect_cluster_stats() -> ClusterStats {
/// Collect cluster and cluster-health statistics from a single storage snapshot.
pub async fn collect_cluster_and_health_stats() -> (ClusterStats, ClusterHealthStats) {
let Some(store) = new_object_layer_fn() else {
return ClusterStats::default();
return (ClusterStats::default(), ClusterHealthStats::default());
};
let storage_info = store.storage_info().await;
let raw_capacity: u64 = storage_info.disks.iter().map(|d| d.total_space).sum();
let used: u64 = storage_info.disks.iter().map(|d| d.used_space).sum();
let usable_capacity = get_total_usable_capacity(&storage_info.disks, &storage_info) as u64;
let free = get_total_usable_capacity_free(&storage_info.disks, &storage_info) as u64;
// Get bucket and object counts from data usage info
// Get bucket and object counts from data usage info.
let (buckets_count, objects_count) = match load_data_usage_from_backend(store.clone()).await {
Ok(data_usage) => (data_usage.buckets_count, data_usage.objects_total_count),
Err(e) => {
warn!("Failed to load data usage from backend: {}", e);
// Fall back to bucket list for buckets_count, objects_count stays 0
// Fall back to bucket list for buckets_count, objects_count stays 0.
let buckets = store
.list_bucket(&BucketOptions {
cached: true,
..Default::default()
})
.await
.unwrap_or_else(|e| {
warn!("Failed to list buckets for cluster metrics: {}", e);
.unwrap_or_else(|err| {
warn!("Failed to list buckets for cluster metrics: {}", err);
Vec::new()
});
(buckets.len() as u64, 0)
}
};
ClusterStats {
raw_capacity_bytes: raw_capacity,
usable_capacity_bytes: usable_capacity,
used_bytes: used,
free_bytes: free,
objects_count,
buckets_count,
let mut online = 0u64;
let mut offline = 0u64;
for disk in &storage_info.disks {
if disk_is_online_for_metrics(disk.state.as_str(), disk.runtime_state.as_deref()) {
online += 1;
} else {
offline += 1;
}
}
(
ClusterStats {
raw_capacity_bytes: raw_capacity,
usable_capacity_bytes: usable_capacity,
used_bytes: used,
free_bytes: free,
objects_count,
buckets_count,
},
ClusterHealthStats {
drives_offline_count: offline,
drives_online_count: online,
drives_count: storage_info.disks.len() as u64,
},
)
}
/// Collect cluster statistics from the storage layer.
#[instrument]
pub async fn collect_cluster_stats() -> ClusterStats {
let (cluster_stats, _) = collect_cluster_and_health_stats().await;
cluster_stats
}
/// Collect cluster health statistics from the storage layer.
pub async fn collect_cluster_health_stats() -> ClusterHealthStats {
let (_, cluster_health_stats) = collect_cluster_and_health_stats().await;
cluster_health_stats
}
/// Collect bucket statistics from the storage layer.
@@ -177,15 +227,207 @@ pub fn collect_bucket_replication_bandwidth_stats() -> Vec<BucketReplicationBand
.collect()
}
/// Collect disk statistics from the storage layer.
pub async fn collect_disk_stats() -> Vec<DiskStats> {
let Some(store) = new_object_layer_fn() else {
/// Collect bucket and target level replication stats from the global replication runtime.
pub async fn collect_bucket_replication_detail_stats() -> Vec<BucketReplicationStats> {
let Some(stats) = GLOBAL_REPLICATION_STATS.get() else {
return Vec::new();
};
let storage_info = store.storage_info().await;
let all_bucket_stats = stats.get_all().await;
let mut buckets = Vec::with_capacity(all_bucket_stats.len());
storage_info
for (bucket, bucket_stats) in all_bucket_stats {
let proxy = stats.get_proxy_stats(&bucket).await;
let mut total_failed_bytes = 0u64;
let mut total_failed_count = 0u64;
let mut last_min_failed_bytes = 0u64;
let mut last_min_failed_count = 0u64;
let mut last_hour_failed_bytes = 0u64;
let mut last_hour_failed_count = 0u64;
let mut sent_bytes = 0u64;
let mut sent_count = 0u64;
let mut targets = Vec::with_capacity(bucket_stats.stats.len());
for (target_arn, target_stats) in bucket_stats.stats {
total_failed_bytes += target_stats.fail_stats.size.max(0) as u64;
total_failed_count += target_stats.fail_stats.count.max(0) as u64;
let last_min = target_stats.fail_stats.recent_since(Duration::from_secs(60));
last_min_failed_bytes += last_min.size.max(0) as u64;
last_min_failed_count += last_min.count.max(0) as u64;
let last_hour = target_stats.fail_stats.recent_since(Duration::from_secs(60 * 60));
last_hour_failed_bytes += last_hour.size.max(0) as u64;
last_hour_failed_count += last_hour.count.max(0) as u64;
sent_bytes += target_stats.replicated_size.max(0) as u64;
sent_count += target_stats.replicated_count.max(0) as u64;
targets.push(BucketReplicationTargetStats {
target_arn,
bandwidth_limit_bytes_per_sec: target_stats.bandwidth_limit_bytes_per_sec.max(0) as u64,
current_bandwidth_bytes_per_sec: target_stats.current_bandwidth_bytes_per_sec,
latency_ms: target_stats.latency.curr,
});
}
buckets.push(BucketReplicationStats {
bucket,
total_failed_bytes,
total_failed_count,
last_min_failed_bytes,
last_min_failed_count,
last_hour_failed_bytes,
last_hour_failed_count,
sent_bytes,
sent_count,
proxied_get_requests_total: proxy.get_total.max(0) as u64,
proxied_get_requests_failures: proxy.get_failed.max(0) as u64,
proxied_head_requests_total: proxy.head_total.max(0) as u64,
proxied_head_requests_failures: proxy.head_failed.max(0) as u64,
// Proxy cache currently tracks generic PutObject requests, not tagging-specific APIs.
// Keep tagging counters zero until PutObjectTagging stats are tracked separately.
proxied_put_tagging_requests_total: 0,
proxied_put_tagging_requests_failures: 0,
proxied_get_tagging_requests_total: 0,
proxied_get_tagging_requests_failures: 0,
proxied_delete_tagging_requests_total: 0,
proxied_delete_tagging_requests_failures: 0,
targets,
});
}
buckets
}
/// Collect site-level replication stats from the global replication runtime.
pub async fn collect_replication_stats() -> ReplicationStats {
let Some(stats) = GLOBAL_REPLICATION_STATS.get() else {
return ReplicationStats::default();
};
let site_metrics = stats.get_sr_metrics_for_node().await;
let current_active_workers = u64::try_from(site_metrics.active_workers.curr).unwrap_or(0);
let bandwidth_stats = collect_bucket_replication_bandwidth_stats();
let current_data_transfer_rate = bandwidth_stats
.iter()
.map(|stat| stat.current_bandwidth_bytes_per_sec)
.sum::<f64>();
let all_bucket_stats = stats.get_all().await;
let average_data_transfer_rate = all_bucket_stats
.values()
.flat_map(|bucket| bucket.stats.values())
.map(|stat| stat.xfer_rate_lrg.avg + stat.xfer_rate_sml.avg)
.sum::<f64>();
let max_data_transfer_rate = all_bucket_stats
.values()
.flat_map(|bucket| bucket.stats.values())
.map(|stat| stat.xfer_rate_lrg.peak + stat.xfer_rate_sml.peak)
.sum::<f64>();
let recent_backlog_count = stats
.mrf_stats
.values()
.copied()
.filter(|value| *value > 0)
.sum::<i64>()
.try_into()
.unwrap_or(0);
ReplicationStats {
average_active_workers: site_metrics.active_workers.avg,
average_queued_bytes: site_metrics.queued.avg.bytes,
average_queued_count: site_metrics.queued.avg.count,
average_data_transfer_rate,
active_workers: current_active_workers,
current_data_transfer_rate,
last_minute_queued_bytes: site_metrics.queued.last_minute.bytes.max(0) as u64,
last_minute_queued_count: site_metrics.queued.last_minute.count.max(0) as u64,
max_active_workers: u64::try_from(site_metrics.active_workers.max).unwrap_or(0),
max_queued_bytes: site_metrics.queued.max.bytes.max(0) as u64,
max_queued_count: site_metrics.queued.max.count.max(0) as u64,
max_data_transfer_rate,
recent_backlog_count,
}
}
/// Collect disk statistics from the storage layer.
pub async fn collect_disk_stats() -> Vec<DiskStats> {
let (disk_stats, _, _) = collect_disk_and_system_drive_stats().await;
disk_stats
}
fn build_system_cpu_stats(system: &System) -> CpuStats {
let cpu_usage = system.global_cpu_usage() as f64;
let cpu_count = system.cpus().len().max(1) as f64;
let load_avg = System::load_average().one;
CpuStats {
avg_idle: (100.0 - cpu_usage).max(0.0),
avg_iowait: 0.0,
load_avg,
load_avg_perc: (load_avg / cpu_count) * 100.0,
nice: 0.0,
steal: 0.0,
system: cpu_usage,
user: 0.0,
}
}
fn build_system_memory_stats(system: &System) -> MemoryStats {
let total = system.total_memory();
let used = system.used_memory();
MemoryStats {
total,
used,
used_perc: if total > 0 {
(used as f64 / total as f64) * 100.0
} else {
0.0
},
free: system.free_memory(),
buffers: 0,
cache: 0,
shared: 0,
available: system.available_memory(),
}
}
/// Collect system CPU and memory statistics from a shared sysinfo snapshot.
pub fn collect_system_cpu_and_memory_stats() -> (CpuStats, MemoryStats) {
let mut system = System::new_all();
collect_system_cpu_and_memory_stats_with(&mut system)
}
/// Collect system CPU and memory statistics by refreshing a reusable sysinfo instance.
pub fn collect_system_cpu_and_memory_stats_with(system: &mut System) -> (CpuStats, MemoryStats) {
system.refresh_cpu_all();
system.refresh_memory();
(build_system_cpu_stats(system), build_system_memory_stats(system))
}
/// Collect system CPU statistics from the current host.
pub fn collect_system_cpu_stats() -> CpuStats {
let (cpu_stats, _) = collect_system_cpu_and_memory_stats();
cpu_stats
}
/// Collect system memory statistics from the current host.
pub fn collect_system_memory_stats() -> MemoryStats {
let (_, memory_stats) = collect_system_cpu_and_memory_stats();
memory_stats
}
/// Collect node disk stats and drive stats from a single storage snapshot.
pub async fn collect_disk_and_system_drive_stats() -> (Vec<DiskStats>, Vec<DriveDetailedStats>, DriveCountStats) {
let Some(store) = new_object_layer_fn() else {
return (Vec::new(), Vec::new(), DriveCountStats::default());
};
let storage_info = store.storage_info().await;
let disk_stats = storage_info
.disks
.iter()
.map(|disk| DiskStats {
@@ -195,7 +437,63 @@ pub async fn collect_disk_stats() -> Vec<DiskStats> {
used_bytes: disk.used_space,
free_bytes: disk.available_space,
})
.collect()
.collect();
let mut online_count = 0u64;
let mut offline_count = 0u64;
let drive_stats = storage_info
.disks
.iter()
.map(|disk| {
let is_online = disk_is_online_for_metrics(disk.state.as_str(), disk.runtime_state.as_deref());
if is_online {
online_count += 1;
} else {
offline_count += 1;
}
DriveDetailedStats {
server: disk.endpoint.clone(),
drive: disk.drive_path.clone(),
total_bytes: disk.total_space,
used_bytes: disk.used_space,
free_bytes: disk.available_space,
used_inodes: 0,
free_inodes: 0,
total_inodes: 0,
timeout_errors_total: 0,
io_errors_total: 0,
availability_errors_total: 0,
waiting_io: 0,
api_latency_micros: 0,
health: if is_online { 1 } else { 0 },
reads_per_sec: 0.0,
reads_kb_per_sec: 0.0,
reads_await: 0.0,
writes_per_sec: 0.0,
writes_kb_per_sec: 0.0,
writes_await: 0.0,
perc_util: if disk.total_space > 0 {
(disk.used_space as f64 / disk.total_space as f64) * 100.0
} else {
0.0
},
}
})
.collect();
let drive_count_stats = DriveCountStats {
offline_count,
online_count,
total_count: online_count + offline_count,
};
(disk_stats, drive_stats, drive_count_stats)
}
/// Collect system drive statistics using the storage layer snapshot.
pub async fn collect_system_drive_stats() -> (Vec<DriveDetailedStats>, DriveCountStats) {
let (_, drive_stats, drive_count_stats) = collect_disk_and_system_drive_stats().await;
(drive_stats, drive_count_stats)
}
/// Collect resource and process statistics for the current process in one sysinfo refresh.
@@ -286,3 +584,18 @@ pub fn collect_host_network_stats() -> HostNetworkStats {
per_interface,
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn disk_is_online_for_metrics_accepts_online_state_case_insensitive() {
assert!(disk_is_online_for_metrics("OnLiNe", Some("online")));
}
#[test]
fn disk_is_online_for_metrics_rejects_offline_runtime_state() {
assert!(!disk_is_online_for_metrics(DRIVE_STATE_OK, Some("offline")));
}
}
rustfs/src/server/http.rs
+51 -7
View File
@@ -66,6 +66,29 @@ use tower_http::trace::TraceLayer;
use tracing::{Span, debug, error, info, instrument, warn};
use tracing_opentelemetry::OpenTelemetrySpanExt;
const LABEL_REQUEST_METHOD: &str = "request_method";
const METRIC_API_REQUESTS_TOTAL: &str = "rustfs_api_requests_total";
const METRIC_API_REQUESTS_FAILURE_TOTAL: &str = "rustfs_api_requests_failure_total";
const METRIC_REQUEST_BODY_BYTES_TOTAL: &str = "rustfs_request_body_bytes_total";
const METRIC_REQUEST_LATENCY_MS: &str = "rustfs_request_latency_ms";
const METRIC_REQUEST_BODY_LEN: &str = "rustfs_request_body_len";
#[inline]
fn request_method_label(method: &Method) -> &'static str {
match method.as_str() {
"GET" => "GET",
"PUT" => "PUT",
"POST" => "POST",
"DELETE" => "DELETE",
"HEAD" => "HEAD",
"OPTIONS" => "OPTIONS",
"PATCH" => "PATCH",
"CONNECT" => "CONNECT",
"TRACE" => "TRACE",
_ => "OTHER",
}
}
pub async fn start_http_server(
config: &config::Config,
readiness: Arc<GlobalReadiness>,
@@ -715,28 +738,31 @@ fn process_connection(
.on_request(|request: &HttpRequest<_>, span: &Span| {
let _enter = span.enter();
debug!("http started method: {}, url path: {}", request.method(), request.uri().path());
let labels = [("key_request_method", request.method().to_string())];
counter!("rustfs.api.requests.total", &labels).increment(1);
counter!(
METRIC_API_REQUESTS_TOTAL,
LABEL_REQUEST_METHOD => request_method_label(request.method())
)
.increment(1);
// Aggregate request body size for throughput monitoring (lightweight)
if let Some(cl) = request.headers().get("content-length")
&& let Some(len) = cl.to_str().ok().and_then(|s| s.parse::<u64>().ok())
{
counter!("rustfs.request.body.bytes_total", "direction" => "request").increment(len);
counter!(METRIC_REQUEST_BODY_BYTES_TOTAL, "direction" => "request").increment(len);
}
})
.on_response(|response: &Response<_>, latency: Duration, span: &Span| {
span.record("status_code", tracing::field::display(response.status()));
let _enter = span.enter();
histogram!("rustfs.request.latency.ms").record(latency.as_millis() as f64);
histogram!(METRIC_REQUEST_LATENCY_MS).record(latency.as_millis() as f64);
debug!("http response generated in {:?}", latency)
})
.on_body_chunk(|chunk: &Bytes, latency: Duration, span: &Span| {
// Always track aggregate body bytes (lightweight counter, no debug logging)
counter!("rustfs.request.body.bytes_total", "direction" => "response").increment(chunk.len() as u64);
counter!(METRIC_REQUEST_BODY_BYTES_TOTAL, "direction" => "response").increment(chunk.len() as u64);
histogram!(METRIC_REQUEST_BODY_LEN, "direction" => "response").record(chunk.len() as f64);
#[cfg(feature = "tracing-chunk-debug")]
{
let _enter = span.enter();
histogram!("rustfs.request.body.len").record(chunk.len() as f64);
debug!("http body sending {} bytes in {:?}", chunk.len(), latency);
}
#[cfg(not(feature = "tracing-chunk-debug"))]
@@ -757,7 +783,7 @@ fn process_connection(
})
.on_failure(|_error, latency: Duration, span: &Span| {
let _enter = span.enter();
counter!("rustfs.api.requests.failure.total").increment(1);
counter!(METRIC_API_REQUESTS_FAILURE_TOTAL).increment(1);
debug!("http request failure error: {:?} in {:?}", _error, latency)
}),
)
@@ -1081,6 +1107,24 @@ mod tests {
assert!(keys.contains(&"content-type"));
}
#[test]
fn test_http_metric_names_and_labels_use_snake_case() {
let metric_names = [
METRIC_API_REQUESTS_TOTAL,
METRIC_API_REQUESTS_FAILURE_TOTAL,
METRIC_REQUEST_BODY_BYTES_TOTAL,
METRIC_REQUEST_LATENCY_MS,
METRIC_REQUEST_BODY_LEN,
];
for metric_name in metric_names {
assert!(metric_name.starts_with("rustfs_"));
assert!(!metric_name.contains('.'));
}
assert_eq!(LABEL_REQUEST_METHOD, "request_method");
}
#[test]
fn test_headermap_carrier_get_all() {
let mut headers = HeaderMap::new();