mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-21 11:56:38 +00:00
Zhengchao An
d22cb5d07a
* fix: resolve release-blocking integration failures * fix: satisfy stable clippy lints * fix: satisfy Rust 1.98 CI lints
2353 lines
91 KiB
Rust
2353 lines
91 KiB
Rust
// 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.
|
|
|
|
//! Statistics collection functions for metrics.
|
|
//!
|
|
//! This module contains functions that collect statistics from various
|
|
//! RustFS internal sources (storage layer, bucket monitor, system info)
|
|
//! and convert them to the Stats structs used by collectors.
|
|
|
|
use crate::metrics::collectors::scanner::{ScannerActiveBucketDriveStats, ScannerBucketDriveResultStats, ScannerSourceWorkStats};
|
|
use crate::metrics::collectors::{
|
|
ApiRequestMetricSupport, ApiRequestStats, BucketReplicationBacklogStats, BucketReplicationBandwidthStats,
|
|
BucketReplicationRuntimeStats, BucketReplicationStats, BucketReplicationTargetBacklogStats, BucketReplicationTargetFlowStats,
|
|
BucketReplicationTargetStats, BucketStats, BucketUsageStats, ClusterConfigStats, ClusterHealthStats, ClusterStats,
|
|
ClusterUsageStats, CompressionClusterStats, CpuStats, DiskStats, DriveCountStats, DriveDetailedStats,
|
|
DriveRuntimeDetailedStats, ErasureSetStats, HostNetworkStats, IamStats, IlmActionTaskStats, IlmBackpressureStats,
|
|
IlmQueueTaskStats, IlmRuntimeStats, IlmStats, IlmTaskEventStats, MemoryStats, NetworkStats, ProcessStats, ProcessStatusType,
|
|
ReplicationStats, ResourceStats, ScannerRuntimeStats, ScannerStats,
|
|
};
|
|
use crate::metrics::runtime_sources::{ObsIlmRuntimeSnapshot, bucket_monitor_handle, iam_metrics_snapshot, ilm_runtime_snapshot};
|
|
use crate::metrics::{
|
|
BucketOperations, BucketOptions, ObsBucketReplicationStatsSnapshot, ObsEcstoreResult, ObsStore, StorageAdminApi,
|
|
obs_bucket_replication_stats_snapshot, obs_get_quota_config, obs_get_total_usable_capacity,
|
|
obs_get_total_usable_capacity_free, obs_load_compression_total_from_memory, obs_load_data_usage_from_backend,
|
|
obs_replication_site_stats_snapshot, obs_resolve_object_store_handle,
|
|
};
|
|
use crate::node_identity::current_local_node_identity;
|
|
use jiff::Timestamp;
|
|
use rustfs_common::heal_channel::HealScanMode;
|
|
use rustfs_common::metrics::{
|
|
ScannerActiveBucketDriveSnapshot, ScannerBucketDriveResultSnapshot, ScannerMetricsReport, ScannerSourceWorkSnapshot,
|
|
global_metrics,
|
|
};
|
|
use rustfs_io_metrics::internode_metrics::global_internode_metrics;
|
|
use rustfs_io_metrics::{
|
|
ProcessResourceSnapshot, ProcessSampler, ProcessStatusSnapshot, ProcessSystemSnapshot, s3_op_metrics_snapshot,
|
|
snapshot_process_resource_and_system, snapshot_process_resource_and_system_with,
|
|
};
|
|
use std::{
|
|
collections::{HashMap, HashSet},
|
|
sync::Arc,
|
|
time::SystemTime,
|
|
};
|
|
use sysinfo::{Networks, System};
|
|
use tracing::{instrument, warn};
|
|
|
|
const LOG_COMPONENT_OBS: &str = "obs";
|
|
const LOG_SUBSYSTEM_METRICS_COLLECTOR: &str = "metrics_collector";
|
|
const EVENT_METRICS_COLLECTOR_STATE: &str = "metrics_collector_state";
|
|
|
|
type ObsStorageInfo = <ObsStore as StorageAdminApi>::StorageInfo;
|
|
type ObsBackendInfo = <ObsStore as StorageAdminApi>::BackendInfo;
|
|
|
|
#[derive(Default)]
|
|
struct ObsDataUsageInfo {
|
|
last_update: Option<SystemTime>,
|
|
usage_snapshot_complete: bool,
|
|
usage_snapshot_converged: bool,
|
|
buckets_count: u64,
|
|
objects_total_count: u64,
|
|
versions_total_count: u64,
|
|
delete_markers_total_count: u64,
|
|
objects_total_size: u64,
|
|
buckets_usage: HashMap<String, ObsBucketUsageInfo>,
|
|
}
|
|
|
|
#[derive(Default)]
|
|
struct ObsBucketUsageInfo {
|
|
size: u64,
|
|
objects_count: u64,
|
|
object_size_histogram: HashMap<String, u64>,
|
|
object_versions_histogram: HashMap<String, u64>,
|
|
versions_count: u64,
|
|
delete_markers_count: u64,
|
|
}
|
|
|
|
#[derive(Debug, Clone, PartialEq)]
|
|
struct ObsBucketReplicationBandwidthStats {
|
|
bucket: String,
|
|
target_arn: String,
|
|
limit_bytes_per_sec: i64,
|
|
current_bandwidth_bytes_per_sec: f64,
|
|
}
|
|
|
|
fn usize_to_u64_saturating(value: usize) -> u64 {
|
|
u64::try_from(value).unwrap_or(u64::MAX)
|
|
}
|
|
|
|
async fn load_obs_data_usage_from_backend(store: Arc<ObsStore>) -> ObsEcstoreResult<ObsDataUsageInfo> {
|
|
let data_usage = obs_load_data_usage_from_backend(store).await?;
|
|
let usage_snapshot_complete = data_usage.is_complete_bucket_usage_snapshot();
|
|
|
|
Ok(ObsDataUsageInfo {
|
|
last_update: data_usage.last_update,
|
|
usage_snapshot_complete,
|
|
usage_snapshot_converged: data_usage.usage_snapshot_converged == Some(true),
|
|
buckets_count: data_usage.buckets_count,
|
|
objects_total_count: data_usage.objects_total_count,
|
|
versions_total_count: data_usage.versions_total_count,
|
|
delete_markers_total_count: data_usage.delete_markers_total_count,
|
|
objects_total_size: data_usage.objects_total_size,
|
|
buckets_usage: data_usage
|
|
.buckets_usage
|
|
.into_iter()
|
|
.map(|(bucket, usage)| {
|
|
(
|
|
bucket,
|
|
ObsBucketUsageInfo {
|
|
size: usage.size,
|
|
objects_count: usage.objects_count,
|
|
object_size_histogram: usage.object_size_histogram,
|
|
object_versions_histogram: usage.object_versions_histogram,
|
|
versions_count: usage.versions_count,
|
|
delete_markers_count: usage.delete_markers_count,
|
|
},
|
|
)
|
|
})
|
|
.collect(),
|
|
})
|
|
}
|
|
|
|
fn bucket_usage_metric_values(data_usage: Option<&ObsDataUsageInfo>, bucket: &str) -> (Option<u64>, Option<u64>) {
|
|
data_usage
|
|
.filter(|usage| usage.usage_snapshot_complete)
|
|
.and_then(|usage| usage.buckets_usage.get(bucket))
|
|
.map(|usage| (Some(usage.size), Some(usage.objects_count)))
|
|
.unwrap_or((None, None))
|
|
}
|
|
|
|
fn data_usage_snapshot_covers_bucket_namespace(data_usage: &ObsDataUsageInfo, buckets: &HashSet<String>) -> bool {
|
|
data_usage.usage_snapshot_complete
|
|
&& u64::try_from(buckets.len()).ok() == Some(data_usage.buckets_count)
|
|
&& buckets.len() == data_usage.buckets_usage.len()
|
|
&& buckets.iter().all(|bucket| data_usage.buckets_usage.contains_key(bucket))
|
|
}
|
|
|
|
fn resolve_obs_object_store_handle() -> Option<Arc<ObsStore>> {
|
|
obs_resolve_object_store_handle()
|
|
}
|
|
|
|
fn scanner_lifecycle_checked_versions(metrics: &ScannerMetricsReport) -> u64 {
|
|
metrics
|
|
.source_work
|
|
.iter()
|
|
.find(|source| source.source == "lifecycle")
|
|
.map(|source| source.checked)
|
|
.unwrap_or_default()
|
|
}
|
|
|
|
fn obs_total_usable_capacity_bytes(storage_info: &ObsStorageInfo) -> u64 {
|
|
usize_to_u64_saturating(obs_get_total_usable_capacity(&storage_info.disks, storage_info))
|
|
}
|
|
|
|
fn obs_total_usable_capacity_free_bytes(storage_info: &ObsStorageInfo) -> u64 {
|
|
usize_to_u64_saturating(obs_get_total_usable_capacity_free(&storage_info.disks, storage_info))
|
|
}
|
|
|
|
fn usable_capacity_used_bytes(usable_capacity: u64, free_bytes: u64) -> u64 {
|
|
usable_capacity.saturating_sub(free_bytes)
|
|
}
|
|
|
|
async fn obs_bucket_quota_limit_bytes(bucket: &str) -> u64 {
|
|
obs_get_quota_config(bucket)
|
|
.await
|
|
.ok()
|
|
.and_then(|(quota, _)| quota.get_quota_limit())
|
|
.unwrap_or(0)
|
|
}
|
|
|
|
fn obs_bucket_replication_bandwidth_stats() -> Option<Vec<ObsBucketReplicationBandwidthStats>> {
|
|
let monitor = bucket_monitor_handle()?;
|
|
Some(
|
|
monitor
|
|
.get_report(|_| true)
|
|
.bucket_stats
|
|
.into_iter()
|
|
.map(|(opts, details)| ObsBucketReplicationBandwidthStats {
|
|
bucket: opts.name,
|
|
target_arn: opts.replication_arn,
|
|
limit_bytes_per_sec: details.limit_bytes_per_sec,
|
|
current_bandwidth_bytes_per_sec: details.current_bandwidth_bytes_per_sec,
|
|
})
|
|
.collect(),
|
|
)
|
|
}
|
|
|
|
async fn obs_ilm_runtime_snapshot() -> ObsIlmRuntimeSnapshot {
|
|
ilm_runtime_snapshot().await
|
|
}
|
|
|
|
async fn obs_bucket_replication_stats_bundle() -> (Vec<BucketReplicationRuntimeStats>, Vec<BucketReplicationBacklogStats>) {
|
|
let snapshots = obs_bucket_replication_stats_snapshot().await;
|
|
let mut detail_stats = Vec::with_capacity(snapshots.len());
|
|
let mut backlog_stats = Vec::with_capacity(snapshots.len());
|
|
|
|
for stats in snapshots {
|
|
backlog_stats.push(BucketReplicationBacklogStats {
|
|
bucket: stats.bucket.clone(),
|
|
current_backlog_count: stats.current_backlog_count,
|
|
current_backlog_bytes: stats.current_backlog_bytes,
|
|
durable_mrf_available: stats.durable_mrf_available,
|
|
durable_mrf_backlog_count: stats.durable_mrf_backlog_count,
|
|
durable_mrf_backlog_bytes: stats.durable_mrf_backlog_bytes,
|
|
mrf_pending_count: stats.mrf_pending_count,
|
|
mrf_pending_bytes: stats.mrf_pending_bytes,
|
|
mrf_dropped_count: stats.mrf_dropped_count,
|
|
mrf_missed_count: stats.mrf_missed_count,
|
|
mrf_flush_failures: stats.mrf_flush_failures,
|
|
mrf_last_flush_duration_millis: stats.mrf_last_flush_duration_millis,
|
|
target_backlogs: stats
|
|
.target_backlogs
|
|
.iter()
|
|
.map(|target| BucketReplicationTargetBacklogStats {
|
|
target_arn: target.target_arn.clone(),
|
|
current_backlog_count: target.current_backlog_count,
|
|
current_backlog_bytes: target.current_backlog_bytes,
|
|
durable_mrf_backlog_count: target.durable_mrf_backlog_count,
|
|
durable_mrf_backlog_bytes: target.durable_mrf_backlog_bytes,
|
|
})
|
|
.collect(),
|
|
});
|
|
detail_stats.push(bucket_replication_detail_from_snapshot(stats));
|
|
}
|
|
|
|
(detail_stats, backlog_stats)
|
|
}
|
|
|
|
fn bucket_replication_detail_from_snapshot(stats: ObsBucketReplicationStatsSnapshot) -> BucketReplicationRuntimeStats {
|
|
let bucket = stats.bucket;
|
|
let (targets, target_flows): (Vec<_>, Vec<_>) = stats
|
|
.targets
|
|
.into_iter()
|
|
.map(|target| {
|
|
(
|
|
BucketReplicationTargetStats {
|
|
target_arn: target.target_arn.clone(),
|
|
bandwidth_limit_bytes_per_sec: target.bandwidth_limit_bytes_per_sec,
|
|
current_bandwidth_bytes_per_sec: target.current_bandwidth_bytes_per_sec,
|
|
latency_ms: target.latency_ms,
|
|
},
|
|
BucketReplicationTargetFlowStats {
|
|
target_arn: target.target_arn,
|
|
sent_bytes: target.sent_bytes,
|
|
sent_count: target.sent_count,
|
|
total_failed_bytes: target.total_failed_bytes,
|
|
total_failed_count: target.total_failed_count,
|
|
last_min_failed_bytes: target.last_min_failed_bytes,
|
|
last_min_failed_count: target.last_min_failed_count,
|
|
last_hour_failed_bytes: target.last_hour_failed_bytes,
|
|
last_hour_failed_count: target.last_hour_failed_count,
|
|
},
|
|
)
|
|
})
|
|
.unzip();
|
|
|
|
BucketReplicationRuntimeStats {
|
|
target_flows,
|
|
stats: BucketReplicationStats {
|
|
bucket,
|
|
total_failed_bytes: stats.total_failed_bytes,
|
|
total_failed_count: stats.total_failed_count,
|
|
last_min_failed_bytes: stats.last_min_failed_bytes,
|
|
last_min_failed_count: stats.last_min_failed_count,
|
|
last_hour_failed_bytes: stats.last_hour_failed_bytes,
|
|
last_hour_failed_count: stats.last_hour_failed_count,
|
|
sent_bytes: stats.sent_bytes,
|
|
sent_count: stats.sent_count,
|
|
proxied_get_requests_total: stats.proxied_get_requests_total,
|
|
proxied_get_requests_failures: stats.proxied_get_requests_failures,
|
|
proxied_head_requests_total: stats.proxied_head_requests_total,
|
|
proxied_head_requests_failures: stats.proxied_head_requests_failures,
|
|
proxied_put_requests_total: stats.proxied_put_requests_total,
|
|
proxied_put_requests_failures: stats.proxied_put_requests_failures,
|
|
proxied_put_tagging_requests_total: stats.proxied_put_tagging_requests_total,
|
|
proxied_put_tagging_requests_failures: stats.proxied_put_tagging_requests_failures,
|
|
proxied_get_tagging_requests_total: stats.proxied_get_tagging_requests_total,
|
|
proxied_get_tagging_requests_failures: stats.proxied_get_tagging_requests_failures,
|
|
proxied_delete_tagging_requests_total: stats.proxied_delete_tagging_requests_total,
|
|
proxied_delete_tagging_requests_failures: stats.proxied_delete_tagging_requests_failures,
|
|
resync_started_count: stats.resync_started_count,
|
|
resync_completed_count: stats.resync_completed_count,
|
|
resync_failed_count: stats.resync_failed_count,
|
|
resync_canceled_count: stats.resync_canceled_count,
|
|
resync_duration_ms: stats.resync_duration_ms,
|
|
targets,
|
|
},
|
|
}
|
|
}
|
|
|
|
async fn obs_site_replication_stats() -> ReplicationStats {
|
|
let current_data_transfer_rate = obs_bucket_replication_bandwidth_stats()
|
|
.into_iter()
|
|
.flatten()
|
|
.map(|stat| stat.current_bandwidth_bytes_per_sec)
|
|
.sum::<f64>();
|
|
let stats = obs_replication_site_stats_snapshot(current_data_transfer_rate).await;
|
|
|
|
ReplicationStats {
|
|
average_active_workers: stats.average_active_workers,
|
|
average_queued_bytes: stats.average_queued_bytes,
|
|
average_queued_count: stats.average_queued_count,
|
|
average_data_transfer_rate: stats.average_data_transfer_rate,
|
|
active_workers: stats.active_workers,
|
|
current_data_transfer_rate: stats.current_data_transfer_rate,
|
|
last_minute_queued_bytes: stats.last_minute_queued_bytes,
|
|
last_minute_queued_count: stats.last_minute_queued_count,
|
|
max_active_workers: stats.max_active_workers,
|
|
max_queued_bytes: stats.max_queued_bytes,
|
|
max_queued_count: stats.max_queued_count,
|
|
max_data_transfer_rate: stats.max_data_transfer_rate,
|
|
recent_backlog_count: stats.recent_backlog_count,
|
|
}
|
|
}
|
|
|
|
fn current_scanner_cycle_age_seconds(current_cycle_active: bool, current_started: Timestamp, now: Timestamp) -> u64 {
|
|
if !current_cycle_active {
|
|
0
|
|
} else {
|
|
timestamp_elapsed_seconds_since(now, current_started)
|
|
}
|
|
}
|
|
|
|
fn timestamp_elapsed_seconds_since(now: Timestamp, earlier: Timestamp) -> u64 {
|
|
let duration = now.duration_since(earlier);
|
|
if duration.is_negative() {
|
|
return 0;
|
|
}
|
|
|
|
u64::try_from(duration.as_secs()).unwrap_or(u64::MAX)
|
|
}
|
|
|
|
fn scanner_scan_mode_code(scan_mode: &str) -> u64 {
|
|
match scan_mode {
|
|
mode if mode == HealScanMode::Normal.as_str() => HealScanMode::Normal as u8 as u64,
|
|
mode if mode == HealScanMode::Deep.as_str() => HealScanMode::Deep as u8 as u64,
|
|
_ => HealScanMode::Unknown as u8 as u64,
|
|
}
|
|
}
|
|
|
|
fn scanner_work_rate_per_second(count: u64, seconds: f64) -> f64 {
|
|
if seconds > 0.0 && seconds.is_finite() {
|
|
count as f64 / seconds
|
|
} else {
|
|
0.0
|
|
}
|
|
}
|
|
|
|
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";
|
|
const CAPACITY_OBSERVATION_LIVE: &str = "live";
|
|
const CAPACITY_OBSERVATION_STALE: &str = "stale";
|
|
const CAPACITY_OBSERVATION_MISSING: &str = "missing";
|
|
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
struct ErasureSetQuorumShape {
|
|
data_shards: u32,
|
|
read_quorum: u32,
|
|
write_quorum: u32,
|
|
read_tolerance: u32,
|
|
write_tolerance: u32,
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
fn disk_capacity_observation_state(source: Option<&str>, age_seconds: Option<u64>) -> (&'static str, u64) {
|
|
let age_seconds = age_seconds.unwrap_or(0);
|
|
match source {
|
|
Some("live_probe") => (CAPACITY_OBSERVATION_LIVE, age_seconds),
|
|
Some("snapshot") => (CAPACITY_OBSERVATION_STALE, age_seconds),
|
|
_ => (CAPACITY_OBSERVATION_MISSING, age_seconds),
|
|
}
|
|
}
|
|
|
|
fn disk_topology_label(index: i32) -> Option<String> {
|
|
if index >= 0 { Some(index.to_string()) } else { None }
|
|
}
|
|
|
|
fn non_empty_disk_id(uuid: &str) -> Option<String> {
|
|
let uuid = uuid.trim();
|
|
if uuid.is_empty() { None } else { Some(uuid.to_string()) }
|
|
}
|
|
|
|
fn drive_inode_stats(used_inodes: u64, free_inodes: u64) -> (Option<u64>, Option<u64>, Option<u64>) {
|
|
let total_inodes = used_inodes.saturating_add(free_inodes);
|
|
if total_inodes == 0 {
|
|
(None, None, None)
|
|
} else {
|
|
(Some(used_inodes), Some(free_inodes), Some(total_inodes))
|
|
}
|
|
}
|
|
|
|
fn drive_api_latency_micros(actions: impl Iterator<Item = (u64, u64)>) -> Option<u64> {
|
|
let mut count = 0u64;
|
|
let mut acc_time_ns = 0u64;
|
|
let mut saw_action = false;
|
|
for (action_count, action_acc_time_ns) in actions {
|
|
saw_action = true;
|
|
if action_count > 0 {
|
|
count = count.saturating_add(action_count);
|
|
acc_time_ns = acc_time_ns.saturating_add(action_acc_time_ns);
|
|
}
|
|
}
|
|
|
|
saw_action.then(|| acc_time_ns.checked_div(count).unwrap_or_default() / 1_000)
|
|
}
|
|
|
|
fn drive_api_latency_by_api_micros<'a>(actions: impl Iterator<Item = (&'a String, u64, u64)>) -> Vec<(String, u64)> {
|
|
let mut values = actions
|
|
.map(|(api, count, acc_time)| (api.clone(), acc_time.checked_div(count).unwrap_or_default() / 1_000))
|
|
.collect::<Vec<_>>();
|
|
values.sort_by(|left, right| left.0.cmp(&right.0));
|
|
values
|
|
}
|
|
|
|
fn drive_api_calls<'a>(api_calls: impl Iterator<Item = (&'a String, &'a u64)>) -> Vec<(String, u64)> {
|
|
let mut values = api_calls.map(|(api, calls)| (api.clone(), *calls)).collect::<Vec<_>>();
|
|
values.sort_by(|left, right| left.0.cmp(&right.0));
|
|
values
|
|
}
|
|
|
|
fn drive_server_label(endpoint: &str, local_server: &str) -> String {
|
|
endpoint
|
|
.strip_prefix("http://")
|
|
.or_else(|| endpoint.strip_prefix("https://"))
|
|
.and_then(|rest| rest.split('/').next())
|
|
.filter(|authority| !authority.is_empty())
|
|
.unwrap_or(local_server)
|
|
.to_string()
|
|
}
|
|
|
|
fn derive_erasure_set_quorum_shape(set_drive_count: usize, parity: usize) -> ErasureSetQuorumShape {
|
|
let data_shards = set_drive_count.saturating_sub(parity);
|
|
let read_quorum = data_shards.max(1);
|
|
let mut write_quorum = read_quorum;
|
|
if data_shards == parity {
|
|
write_quorum += 1;
|
|
}
|
|
|
|
ErasureSetQuorumShape {
|
|
data_shards: data_shards as u32,
|
|
read_quorum: read_quorum as u32,
|
|
write_quorum: write_quorum as u32,
|
|
read_tolerance: parity as u32,
|
|
write_tolerance: set_drive_count.saturating_sub(write_quorum) as u32,
|
|
}
|
|
}
|
|
|
|
fn apply_erasure_set_health(entry: &mut ErasureSetStats) {
|
|
let online = entry.online_drives_count;
|
|
entry.read_health = u8::from(online >= entry.read_quorum);
|
|
entry.write_health = u8::from(online >= entry.write_quorum);
|
|
entry.health = u8::from(entry.write_health == 1);
|
|
}
|
|
|
|
#[derive(Debug, Clone, Default)]
|
|
pub struct ProcessMetricBundle {
|
|
pub resource: ResourceStats,
|
|
pub process: ProcessStats,
|
|
pub disk_read_bytes: u64,
|
|
pub disk_write_bytes: u64,
|
|
}
|
|
|
|
/// Collect cluster and cluster-health statistics from a single storage snapshot.
|
|
pub async fn collect_cluster_and_health_stats() -> (ClusterStats, ClusterHealthStats) {
|
|
let Some(store) = resolve_obs_object_store_handle() else {
|
|
return (ClusterStats::default(), ClusterHealthStats::default());
|
|
};
|
|
|
|
let storage_info = StorageAdminApi::storage_info(store.as_ref()).await;
|
|
let raw_capacity: u64 = storage_info.disks.iter().map(|d| d.total_space).sum();
|
|
let usable_capacity = obs_total_usable_capacity_bytes(&storage_info);
|
|
let free = obs_total_usable_capacity_free_bytes(&storage_info);
|
|
let used = usable_capacity_used_bytes(usable_capacity, free);
|
|
let stale_capacity_drives = storage_info
|
|
.disks
|
|
.iter()
|
|
.filter(|disk| {
|
|
disk_capacity_observation_state(disk.capacity_observation_source.as_deref(), disk.capacity_observation_age_seconds).0
|
|
== CAPACITY_OBSERVATION_STALE
|
|
})
|
|
.count() as u64;
|
|
let missing_capacity_drives = storage_info
|
|
.disks
|
|
.iter()
|
|
.filter(|disk| {
|
|
disk_capacity_observation_state(disk.capacity_observation_source.as_deref(), disk.capacity_observation_age_seconds).0
|
|
== CAPACITY_OBSERVATION_MISSING
|
|
})
|
|
.count() as u64;
|
|
|
|
let data_usage = match load_obs_data_usage_from_backend(store).await {
|
|
Ok(data_usage) if data_usage.usage_snapshot_complete => Some(data_usage),
|
|
Ok(_) => None,
|
|
Err(error) => {
|
|
warn!(event = EVENT_METRICS_COLLECTOR_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_METRICS_COLLECTOR, collector = "cluster_stats", result = "data_usage_load_failed", error = %error, "metrics collector state changed");
|
|
None
|
|
}
|
|
};
|
|
let buckets_count = data_usage.as_ref().map(|usage| usage.buckets_count);
|
|
let objects_count = data_usage.as_ref().map(|usage| usage.objects_total_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,
|
|
stale_capacity_drives,
|
|
missing_capacity_drives,
|
|
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.
|
|
///
|
|
/// `None` means the bucket namespace could not be observed. Callers must keep
|
|
/// their prior metric-series state instead of treating that failure as an
|
|
/// authoritative empty namespace.
|
|
pub async fn collect_bucket_stats() -> Option<Vec<BucketStats>> {
|
|
let store = resolve_obs_object_store_handle()?;
|
|
|
|
// Load data usage info from backend to get bucket sizes and object counts
|
|
let data_usage = match load_obs_data_usage_from_backend(store.clone()).await {
|
|
Ok(info) => Some(info),
|
|
Err(e) => {
|
|
warn!(event = EVENT_METRICS_COLLECTOR_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_METRICS_COLLECTOR, collector = "bucket_stats", result = "data_usage_load_failed", error = %e, "metrics collector state changed");
|
|
None
|
|
}
|
|
};
|
|
|
|
// List all buckets
|
|
let buckets = match store
|
|
.list_bucket(&BucketOptions {
|
|
cached: true,
|
|
..Default::default()
|
|
})
|
|
.await
|
|
{
|
|
Ok(buckets) => buckets,
|
|
Err(e) => {
|
|
warn!(event = EVENT_METRICS_COLLECTOR_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_METRICS_COLLECTOR, collector = "bucket_stats", result = "bucket_list_failed", error = %e, "metrics collector state changed");
|
|
return None;
|
|
}
|
|
};
|
|
|
|
let mut stats = Vec::with_capacity(buckets.len());
|
|
|
|
for bucket in buckets {
|
|
if bucket.name.starts_with('.') {
|
|
continue;
|
|
}
|
|
|
|
// Get size and objects_count from data usage info
|
|
let (size_bytes, objects_count) = bucket_usage_metric_values(data_usage.as_ref(), &bucket.name);
|
|
|
|
// Get quota from bucket metadata
|
|
let quota_bytes = obs_bucket_quota_limit_bytes(&bucket.name).await;
|
|
|
|
stats.push(BucketStats {
|
|
name: bucket.name,
|
|
size_bytes,
|
|
objects_count,
|
|
quota_bytes,
|
|
});
|
|
}
|
|
|
|
Some(stats)
|
|
}
|
|
|
|
/// Collect bucket replication bandwidth stats from the global monitor.
|
|
pub fn collect_bucket_replication_bandwidth_stats() -> Vec<BucketReplicationBandwidthStats> {
|
|
let Some(bandwidth_stats) = obs_bucket_replication_bandwidth_stats() else {
|
|
return Vec::new();
|
|
};
|
|
|
|
bandwidth_stats
|
|
.into_iter()
|
|
.map(|stat| {
|
|
let target_arn = stat.target_arn;
|
|
let limit_bytes_per_sec = u64::try_from(stat.limit_bytes_per_sec).unwrap_or_else(|_| {
|
|
warn!(event = EVENT_METRICS_COLLECTOR_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_METRICS_COLLECTOR, collector = "bucket_replication_bandwidth", result = "invalid_limit_value", target_arn = ?target_arn, limit_value = stat.limit_bytes_per_sec, "metrics collector state changed");
|
|
0
|
|
});
|
|
|
|
BucketReplicationBandwidthStats {
|
|
bucket: stat.bucket,
|
|
target_arn,
|
|
limit_bytes_per_sec,
|
|
current_bandwidth_bytes_per_sec: stat.current_bandwidth_bytes_per_sec,
|
|
}
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
/// Collect bucket and target level replication stats from the global replication runtime.
|
|
pub async fn collect_bucket_replication_detail_stats() -> Vec<BucketReplicationStats> {
|
|
obs_bucket_replication_stats_snapshot()
|
|
.await
|
|
.into_iter()
|
|
.map(|snapshot| bucket_replication_detail_from_snapshot(snapshot).stats)
|
|
.collect()
|
|
}
|
|
|
|
pub(crate) async fn collect_bucket_replication_stats_bundle()
|
|
-> (Vec<BucketReplicationRuntimeStats>, Vec<BucketReplicationBacklogStats>) {
|
|
obs_bucket_replication_stats_bundle().await
|
|
}
|
|
|
|
/// Collect site-level replication stats from the global replication runtime.
|
|
pub async fn collect_replication_stats() -> ReplicationStats {
|
|
obs_site_replication_stats().await
|
|
}
|
|
|
|
/// Collect S3 API request totals from the in-process operation recorder.
|
|
pub(crate) fn collect_api_request_stats() -> Vec<ApiRequestStats> {
|
|
let server = current_local_node_identity();
|
|
s3_op_metrics_snapshot()
|
|
.into_iter()
|
|
.map(|snapshot| ApiRequestStats {
|
|
server: server.clone(),
|
|
name: snapshot.op.to_string(),
|
|
req_type: "s3".to_string(),
|
|
total: snapshot.total,
|
|
supported_metrics: ApiRequestMetricSupport::TOTALS_ONLY,
|
|
..Default::default()
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
/// 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, server: &str) -> 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 {
|
|
server: server.to_string(),
|
|
avg_idle: (100.0 - cpu_usage).max(0.0),
|
|
load_avg,
|
|
load_avg_perc: (load_avg / cpu_count) * 100.0,
|
|
usage_perc: cpu_usage,
|
|
}
|
|
}
|
|
|
|
fn build_system_memory_stats(system: &System, server: &str) -> MemoryStats {
|
|
let total = system.total_memory();
|
|
let used = system.used_memory();
|
|
|
|
MemoryStats {
|
|
server: server.to_string(),
|
|
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();
|
|
let server = current_local_node_identity();
|
|
(build_system_cpu_stats(system, &server), build_system_memory_stats(system, &server))
|
|
}
|
|
|
|
/// 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 (disk_stats, drive_stats, drive_count_stats) = collect_disk_and_system_drive_runtime_stats().await;
|
|
(disk_stats, drive_stats.into_iter().map(|stat| stat.stats).collect(), drive_count_stats)
|
|
}
|
|
|
|
pub(crate) async fn collect_disk_and_system_drive_runtime_stats()
|
|
-> (Vec<DiskStats>, Vec<DriveRuntimeDetailedStats>, DriveCountStats) {
|
|
let Some(store) = resolve_obs_object_store_handle() else {
|
|
return (Vec::new(), Vec::new(), DriveCountStats::default());
|
|
};
|
|
|
|
let storage_info = StorageAdminApi::storage_info(store.as_ref()).await;
|
|
let local_server = current_local_node_identity();
|
|
let disk_stats = storage_info
|
|
.disks
|
|
.iter()
|
|
.map(|disk| DiskStats {
|
|
server: drive_server_label(&disk.endpoint, &local_server),
|
|
drive: disk.drive_path.clone(),
|
|
total_bytes: disk.total_space,
|
|
used_bytes: disk.used_space,
|
|
free_bytes: disk.available_space,
|
|
})
|
|
.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());
|
|
let (capacity_observation_state, capacity_observation_age_seconds) = disk_capacity_observation_state(
|
|
disk.capacity_observation_source.as_deref(),
|
|
disk.capacity_observation_age_seconds,
|
|
);
|
|
if is_online {
|
|
online_count += 1;
|
|
} else {
|
|
offline_count += 1;
|
|
}
|
|
let (used_inodes, free_inodes, total_inodes) = drive_inode_stats(disk.used_inodes, disk.free_inodes);
|
|
|
|
DriveRuntimeDetailedStats {
|
|
pool_index: disk_topology_label(disk.pool_index),
|
|
set_index: disk_topology_label(disk.set_index),
|
|
drive_index: disk_topology_label(disk.disk_index),
|
|
disk_id: non_empty_disk_id(&disk.uuid),
|
|
runtime_state: Some(disk.runtime_state.as_deref().unwrap_or("unknown").to_ascii_lowercase()),
|
|
healing: disk.healing,
|
|
scanning: disk.scanning,
|
|
offline_duration_seconds: disk.offline_duration_seconds,
|
|
api_calls: disk
|
|
.metrics
|
|
.as_ref()
|
|
.map(|metrics| drive_api_calls(metrics.api_calls.iter()))
|
|
.unwrap_or_default(),
|
|
api_latency_by_api_micros: disk
|
|
.metrics
|
|
.as_ref()
|
|
.map(|metrics| {
|
|
drive_api_latency_by_api_micros(
|
|
metrics
|
|
.last_minute
|
|
.iter()
|
|
.map(|(api, action)| (api, action.count, action.acc_time)),
|
|
)
|
|
})
|
|
.unwrap_or_default(),
|
|
stats: DriveDetailedStats {
|
|
server: drive_server_label(&disk.endpoint, &local_server),
|
|
drive: disk.drive_path.clone(),
|
|
total_bytes: disk.total_space,
|
|
used_bytes: disk.used_space,
|
|
free_bytes: disk.available_space,
|
|
capacity_observation_state,
|
|
capacity_observation_age_seconds,
|
|
used_inodes,
|
|
free_inodes,
|
|
total_inodes,
|
|
timeout_errors_total: disk.metrics.as_ref().map(|metrics| metrics.total_errors_timeout),
|
|
io_errors_total: None,
|
|
availability_errors_total: disk.metrics.as_ref().map(|metrics| metrics.total_errors_availability),
|
|
waiting_io: disk.metrics.as_ref().map(|metrics| u64::from(metrics.total_waiting)),
|
|
api_latency_micros: disk.metrics.as_ref().and_then(|metrics| {
|
|
drive_api_latency_micros(metrics.last_minute.values().map(|action| (action.count, action.acc_time)))
|
|
}),
|
|
health: if is_online { 1 } else { 0 },
|
|
writes_total: disk.metrics.as_ref().map(|metrics| metrics.total_writes),
|
|
deletes_total: disk.metrics.as_ref().map(|metrics| metrics.total_deletes),
|
|
reads_per_sec: None,
|
|
reads_kb_per_sec: None,
|
|
reads_await: None,
|
|
writes_per_sec: None,
|
|
writes_kb_per_sec: None,
|
|
writes_await: None,
|
|
perc_util: None,
|
|
},
|
|
}
|
|
})
|
|
.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.
|
|
#[inline]
|
|
pub fn collect_process_metric_bundle() -> ProcessMetricBundle {
|
|
let (resource_snapshot, process_snapshot) = snapshot_process_resource_and_system();
|
|
process_metric_bundle_from_snapshots(resource_snapshot, process_snapshot)
|
|
}
|
|
|
|
#[inline]
|
|
pub fn collect_process_metric_bundle_with(sampler: &mut ProcessSampler) -> ProcessMetricBundle {
|
|
let (resource_snapshot, process_snapshot) = snapshot_process_resource_and_system_with(sampler);
|
|
process_metric_bundle_from_snapshots(resource_snapshot, process_snapshot)
|
|
}
|
|
|
|
fn process_metric_bundle_from_snapshots(
|
|
resource_snapshot: ProcessResourceSnapshot,
|
|
process_snapshot: ProcessSystemSnapshot,
|
|
) -> ProcessMetricBundle {
|
|
let server = current_local_node_identity();
|
|
let status = match process_snapshot.status {
|
|
ProcessStatusSnapshot::Running => ProcessStatusType::Running,
|
|
ProcessStatusSnapshot::Sleeping => ProcessStatusType::Sleeping,
|
|
ProcessStatusSnapshot::Zombie => ProcessStatusType::Zombie,
|
|
ProcessStatusSnapshot::Other => ProcessStatusType::Other,
|
|
};
|
|
|
|
let resource_stats = ResourceStats {
|
|
server: server.clone(),
|
|
cpu_percent: resource_snapshot.cpu_percent,
|
|
memory_bytes: resource_snapshot.memory_bytes,
|
|
uptime_seconds: resource_snapshot.uptime_seconds,
|
|
};
|
|
let process_stats = ProcessStats {
|
|
server,
|
|
locks_read_total: process_snapshot.locks_read_total,
|
|
locks_write_total: process_snapshot.locks_write_total,
|
|
cpu_total_seconds: process_snapshot.cpu_total_seconds,
|
|
file_descriptor_limit_total: process_snapshot.file_descriptor_limit_total,
|
|
file_descriptor_open_total: process_snapshot.file_descriptor_open_total,
|
|
go_routine_total: process_snapshot.go_routine_total,
|
|
io_rchar_bytes: process_snapshot.io_rchar_bytes,
|
|
io_read_bytes: process_snapshot.io_read_bytes,
|
|
io_wchar_bytes: process_snapshot.io_wchar_bytes,
|
|
io_write_bytes: process_snapshot.io_write_bytes,
|
|
resident_memory_bytes: process_snapshot.resident_memory_bytes,
|
|
start_time_seconds: process_snapshot.start_time_seconds,
|
|
status,
|
|
status_value: process_snapshot.status_value,
|
|
syscall_read_total: process_snapshot.syscall_read_total,
|
|
syscall_write_total: process_snapshot.syscall_write_total,
|
|
uptime_seconds: process_snapshot.uptime_seconds,
|
|
virtual_memory_bytes: process_snapshot.virtual_memory_bytes,
|
|
virtual_memory_max_bytes: process_snapshot.virtual_memory_max_bytes,
|
|
};
|
|
|
|
ProcessMetricBundle {
|
|
resource: resource_stats,
|
|
process: process_stats,
|
|
disk_read_bytes: process_snapshot.disk_read_bytes,
|
|
disk_write_bytes: process_snapshot.disk_write_bytes,
|
|
}
|
|
}
|
|
|
|
/// Collect resource and process statistics for the current process in one sysinfo refresh.
|
|
#[inline]
|
|
pub fn collect_process_resource_and_system_stats() -> (ResourceStats, ProcessStats) {
|
|
let bundle = collect_process_metric_bundle();
|
|
(bundle.resource, bundle.process)
|
|
}
|
|
|
|
/// Collect resource statistics for the current process.
|
|
#[inline]
|
|
pub fn collect_process_stats() -> ResourceStats {
|
|
collect_process_metric_bundle().resource
|
|
}
|
|
|
|
/// Collect process statistics for the current process.
|
|
#[inline]
|
|
pub fn collect_process_system_stats() -> ProcessStats {
|
|
collect_process_metric_bundle().process
|
|
}
|
|
|
|
/// Collect host network statistics from a refreshed network interface snapshot.
|
|
///
|
|
/// These counters come from system interfaces and are host-wide, not process-scoped.
|
|
pub fn collect_host_network_stats_with(networks: &Networks) -> HostNetworkStats {
|
|
let mut total_received = 0u64;
|
|
let mut total_transmitted = 0u64;
|
|
let mut per_interface = Vec::with_capacity(networks.len());
|
|
|
|
for (interface_name, data) in networks {
|
|
let received = data.received();
|
|
let transmitted = data.transmitted();
|
|
total_received += received;
|
|
total_transmitted += transmitted;
|
|
per_interface.push((interface_name.to_string(), received, transmitted));
|
|
}
|
|
|
|
HostNetworkStats {
|
|
server: current_local_node_identity(),
|
|
total_received,
|
|
total_transmitted,
|
|
per_interface,
|
|
}
|
|
}
|
|
|
|
/// Collect host network statistics using a persistent `sysinfo::Networks` snapshot.
|
|
///
|
|
/// `sysinfo` reports network I/O as deltas since the previous refresh, so
|
|
/// callers must reuse the same `Networks` instance across collection ticks.
|
|
pub fn collect_host_network_stats(networks: &mut Networks) -> HostNetworkStats {
|
|
networks.refresh(true);
|
|
collect_host_network_stats_with(networks)
|
|
}
|
|
|
|
/// Collect internode network metrics from the global internode metrics snapshot.
|
|
///
|
|
/// The returned values come directly from `global_internode_metrics().snapshot()`
|
|
/// and currently include only the counters and dial timing data tracked by the
|
|
/// internode metrics runtime.
|
|
pub fn collect_internode_network_stats() -> Option<NetworkStats> {
|
|
let snapshot = global_internode_metrics().snapshot();
|
|
|
|
Some(NetworkStats {
|
|
server: current_local_node_identity(),
|
|
internode_errors_total: snapshot.errors_total,
|
|
internode_dial_errors_total: snapshot.dial_errors_total,
|
|
internode_dial_avg_time_nanos: snapshot.dial_avg_time_nanos,
|
|
internode_sent_bytes_total: snapshot.sent_bytes_total,
|
|
internode_recv_bytes_total: snapshot.recv_bytes_total,
|
|
})
|
|
}
|
|
|
|
/// Collect cluster config metrics from backend parity configuration.
|
|
fn cluster_config_stats_from_backend_parities(
|
|
rr_sc_parity: Option<usize>,
|
|
standard_sc_parity: Option<usize>,
|
|
) -> Option<ClusterConfigStats> {
|
|
Some(ClusterConfigStats {
|
|
rrs_parity: u32::try_from(rr_sc_parity?).ok()?,
|
|
standard_parity: u32::try_from(standard_sc_parity?).ok()?,
|
|
})
|
|
}
|
|
|
|
pub async fn collect_cluster_config_stats() -> Option<ClusterConfigStats> {
|
|
let store = resolve_obs_object_store_handle()?;
|
|
let backend = StorageAdminApi::backend_info(store.as_ref()).await;
|
|
|
|
cluster_config_stats_from_backend_parities(backend.rr_sc_parity, backend.standard_sc_parity)
|
|
}
|
|
|
|
fn standard_erasure_layout_from_backend(backend: &ObsBackendInfo, pool_idx: usize) -> Option<(usize, usize)> {
|
|
let drives_per_set = backend.drives_per_set.get(pool_idx).copied()?;
|
|
if drives_per_set == 0
|
|
|| (!backend.standard_sc_data.is_empty() && backend.standard_sc_data.len() != backend.drives_per_set.len())
|
|
|| (!backend.standard_sc_parities.is_empty() && backend.standard_sc_parities.len() != backend.drives_per_set.len())
|
|
{
|
|
return None;
|
|
}
|
|
|
|
let has_data = !backend.standard_sc_data.is_empty();
|
|
let has_parities = !backend.standard_sc_parities.is_empty();
|
|
let (data, parity) = match (has_data, has_parities) {
|
|
(true, true) => (
|
|
backend.standard_sc_data.get(pool_idx).copied()?,
|
|
backend.standard_sc_parities.get(pool_idx).copied()?,
|
|
),
|
|
(true, false) => {
|
|
let data = backend.standard_sc_data.get(pool_idx).copied()?;
|
|
(data, drives_per_set.checked_sub(data)?)
|
|
}
|
|
(false, true) => {
|
|
let parity = backend.standard_sc_parities.get(pool_idx).copied()?;
|
|
(drives_per_set.checked_sub(parity)?, parity)
|
|
}
|
|
(false, false) => {
|
|
let parity = backend.standard_sc_parity?;
|
|
(drives_per_set.checked_sub(parity)?, parity)
|
|
}
|
|
};
|
|
|
|
if data == 0 || parity > data || data.checked_add(parity) != Some(drives_per_set) {
|
|
return None;
|
|
}
|
|
|
|
Some((data, parity))
|
|
}
|
|
|
|
fn erasure_set_stats_from_backend(storage_info: &ObsStorageInfo, backend: &ObsBackendInfo) -> Vec<ErasureSetStats> {
|
|
let mut grouped: HashMap<(usize, usize), ErasureSetStats> = HashMap::new();
|
|
|
|
for disk in &storage_info.disks {
|
|
let (Ok(pool_idx), Ok(set_idx)) = (usize::try_from(disk.pool_index), usize::try_from(disk.set_index)) else {
|
|
continue;
|
|
};
|
|
let Some((_, parity)) = standard_erasure_layout_from_backend(backend, pool_idx) else {
|
|
continue;
|
|
};
|
|
let set_drive_count = backend.drives_per_set[pool_idx];
|
|
let (Ok(pool_id), Ok(set_id), Ok(size), Ok(parity_metric)) = (
|
|
u32::try_from(pool_idx),
|
|
u32::try_from(set_idx),
|
|
u32::try_from(set_drive_count),
|
|
u32::try_from(parity),
|
|
) else {
|
|
continue;
|
|
};
|
|
let quorum_shape = derive_erasure_set_quorum_shape(set_drive_count, parity);
|
|
|
|
let entry = grouped.entry((pool_idx, set_idx)).or_insert_with(|| ErasureSetStats {
|
|
pool_id,
|
|
set_id,
|
|
size,
|
|
parity: parity_metric,
|
|
data_shards: quorum_shape.data_shards,
|
|
read_quorum: quorum_shape.read_quorum,
|
|
write_quorum: quorum_shape.write_quorum,
|
|
online_drives_count: 0,
|
|
healing_drives_count: 0,
|
|
health: 0,
|
|
read_tolerance: quorum_shape.read_tolerance,
|
|
write_tolerance: quorum_shape.write_tolerance,
|
|
read_health: 0,
|
|
write_health: 0,
|
|
});
|
|
|
|
if disk_is_online_for_metrics(disk.state.as_str(), disk.runtime_state.as_deref()) {
|
|
entry.online_drives_count += 1;
|
|
}
|
|
if disk.healing {
|
|
entry.healing_drives_count += 1;
|
|
}
|
|
}
|
|
|
|
for entry in grouped.values_mut() {
|
|
apply_erasure_set_health(entry);
|
|
}
|
|
|
|
let mut stats = grouped.into_values().collect::<Vec<_>>();
|
|
stats.sort_by_key(|stat| (stat.pool_id, stat.set_id));
|
|
stats
|
|
}
|
|
|
|
/// Collect cluster erasure set metrics from storage and backend topology info.
|
|
pub async fn collect_erasure_set_stats() -> Vec<ErasureSetStats> {
|
|
let Some(store) = resolve_obs_object_store_handle() else {
|
|
return Vec::new();
|
|
};
|
|
|
|
let storage_info = StorageAdminApi::storage_info(store.as_ref()).await;
|
|
let backend = StorageAdminApi::backend_info(store.as_ref()).await;
|
|
erasure_set_stats_from_backend(&storage_info, &backend)
|
|
}
|
|
|
|
pub async fn collect_iam_stats() -> Option<IamStats> {
|
|
let snapshot = iam_metrics_snapshot()?;
|
|
|
|
Some(IamStats {
|
|
last_sync_duration_millis: snapshot.last_sync_duration_millis,
|
|
plugin_authn_service_failed_requests_minute: snapshot.plugin_authn_service_failed_requests_minute,
|
|
plugin_authn_service_last_fail_seconds: snapshot.plugin_authn_service_last_fail_seconds,
|
|
plugin_authn_service_last_succ_seconds: snapshot.plugin_authn_service_last_succ_seconds,
|
|
plugin_authn_service_succ_avg_rtt_ms_minute: snapshot.plugin_authn_service_succ_avg_rtt_ms_minute,
|
|
plugin_authn_service_succ_max_rtt_ms_minute: snapshot.plugin_authn_service_succ_max_rtt_ms_minute,
|
|
plugin_authn_service_total_requests_minute: snapshot.plugin_authn_service_total_requests_minute,
|
|
since_last_sync_millis: snapshot.since_last_sync_millis,
|
|
sync_failures: snapshot.sync_failures,
|
|
sync_successes: snapshot.sync_successes,
|
|
})
|
|
}
|
|
|
|
/// Collect cluster and per-bucket usage metrics from backend usage snapshots.
|
|
///
|
|
/// This reads persisted usage data via `load_data_usage_from_backend()` and
|
|
/// builds cluster totals plus per-bucket distributions from the returned
|
|
/// histograms. It does not trigger an inline object-data rescan.
|
|
pub async fn collect_cluster_usage_metric_stats() -> Option<(ClusterUsageStats, Vec<BucketUsageStats>)> {
|
|
let store = resolve_obs_object_store_handle()?;
|
|
let data_usage = load_obs_data_usage_from_backend(store.clone()).await.ok()?;
|
|
let bucket_namespace = store
|
|
.list_bucket(&BucketOptions {
|
|
cached: true,
|
|
no_metadata: true,
|
|
..Default::default()
|
|
})
|
|
.await
|
|
.ok()?
|
|
.into_iter()
|
|
.filter(|bucket| !bucket.name.starts_with('.'))
|
|
.map(|bucket| bucket.name)
|
|
.collect::<HashSet<_>>();
|
|
collect_cluster_usage_metric_stats_from_data_usage(data_usage, &bucket_namespace).await
|
|
}
|
|
|
|
async fn collect_cluster_usage_metric_stats_from_data_usage(
|
|
data_usage: ObsDataUsageInfo,
|
|
bucket_namespace: &HashSet<String>,
|
|
) -> Option<(ClusterUsageStats, Vec<BucketUsageStats>)> {
|
|
if !data_usage_snapshot_covers_bucket_namespace(&data_usage, bucket_namespace) {
|
|
return None;
|
|
}
|
|
|
|
let mut buckets = Vec::with_capacity(data_usage.buckets_usage.len());
|
|
|
|
for (bucket_name, usage) in &data_usage.buckets_usage {
|
|
if bucket_name.starts_with('.') {
|
|
continue;
|
|
}
|
|
|
|
let quota_bytes = obs_bucket_quota_limit_bytes(bucket_name).await;
|
|
|
|
buckets.push(BucketUsageStats {
|
|
bucket: bucket_name.clone(),
|
|
total_bytes: usage.size,
|
|
objects_count: usage.objects_count,
|
|
versions_count: usage.versions_count,
|
|
delete_markers_count: usage.delete_markers_count,
|
|
quota_bytes,
|
|
object_size_distribution: usage
|
|
.object_size_histogram
|
|
.iter()
|
|
.map(|(range, count)| (range.clone(), *count))
|
|
.collect(),
|
|
version_count_distribution: usage
|
|
.object_versions_histogram
|
|
.iter()
|
|
.map(|(range, count)| (range.clone(), *count))
|
|
.collect(),
|
|
});
|
|
}
|
|
|
|
buckets.sort_by(|a, b| a.bucket.cmp(&b.bucket));
|
|
|
|
Some((
|
|
ClusterUsageStats {
|
|
since_last_update_seconds: crate::metrics::collectors::cluster_usage::usage_since_last_update_seconds(
|
|
data_usage.last_update,
|
|
SystemTime::now(),
|
|
),
|
|
total_bytes: data_usage.objects_total_size,
|
|
objects_count: data_usage.objects_total_count,
|
|
versions_count: data_usage.versions_total_count,
|
|
delete_markers_count: data_usage.delete_markers_total_count,
|
|
buckets_count: data_usage.buckets_count,
|
|
snapshot_converged: data_usage.usage_snapshot_converged,
|
|
object_size_distribution: data_usage
|
|
.buckets_usage
|
|
.values()
|
|
.flat_map(|usage| usage.object_size_histogram.iter())
|
|
.fold(HashMap::<String, u64>::new(), |mut acc, (range, count)| {
|
|
*acc.entry(range.clone()).or_default() += *count;
|
|
acc
|
|
})
|
|
.into_iter()
|
|
.collect(),
|
|
versions_distribution: data_usage
|
|
.buckets_usage
|
|
.values()
|
|
.flat_map(|usage| usage.object_versions_histogram.iter())
|
|
.fold(HashMap::<String, u64>::new(), |mut acc, (range, count)| {
|
|
*acc.entry(range.clone()).or_default() += *count;
|
|
acc
|
|
})
|
|
.into_iter()
|
|
.collect(),
|
|
},
|
|
buckets,
|
|
))
|
|
}
|
|
|
|
fn ilm_action_task_stats(ilm: &ObsIlmRuntimeSnapshot) -> Vec<IlmActionTaskStats> {
|
|
vec![
|
|
IlmActionTaskStats {
|
|
action: "expiry".to_string(),
|
|
state: "pending".to_string(),
|
|
value: ilm.expiry_pending_tasks,
|
|
},
|
|
IlmActionTaskStats {
|
|
action: "transition".to_string(),
|
|
state: "active".to_string(),
|
|
value: ilm.transition_active_tasks,
|
|
},
|
|
IlmActionTaskStats {
|
|
action: "transition".to_string(),
|
|
state: "pending".to_string(),
|
|
value: ilm.transition_pending_tasks,
|
|
},
|
|
IlmActionTaskStats {
|
|
action: "transition".to_string(),
|
|
state: "missed_immediate".to_string(),
|
|
value: ilm.transition_missed_immediate_tasks,
|
|
},
|
|
IlmActionTaskStats {
|
|
action: "transition".to_string(),
|
|
state: "queue_full".to_string(),
|
|
value: ilm.transition_queue_full_tasks,
|
|
},
|
|
IlmActionTaskStats {
|
|
action: "transition".to_string(),
|
|
state: "queue_send_timeout".to_string(),
|
|
value: ilm.transition_queue_send_timeout_tasks,
|
|
},
|
|
IlmActionTaskStats {
|
|
action: "transition".to_string(),
|
|
state: "compensation_scheduled".to_string(),
|
|
value: ilm.transition_compensation_scheduled_tasks,
|
|
},
|
|
IlmActionTaskStats {
|
|
action: "transition".to_string(),
|
|
state: "compensation_running".to_string(),
|
|
value: ilm.transition_compensation_running_tasks,
|
|
},
|
|
]
|
|
}
|
|
|
|
fn ilm_queue_task_stats(metrics: &ScannerMetricsReport) -> Vec<IlmQueueTaskStats> {
|
|
let expiry = &metrics.lifecycle_expiry;
|
|
let transition = &metrics.lifecycle_transition;
|
|
vec![
|
|
IlmQueueTaskStats {
|
|
action: "expiry".to_string(),
|
|
state: "pending".to_string(),
|
|
value: expiry.current_queued,
|
|
},
|
|
IlmQueueTaskStats {
|
|
action: "expiry".to_string(),
|
|
state: "active".to_string(),
|
|
value: expiry.current_active,
|
|
},
|
|
IlmQueueTaskStats {
|
|
action: "transition".to_string(),
|
|
state: "pending".to_string(),
|
|
value: transition.current_queued,
|
|
},
|
|
IlmQueueTaskStats {
|
|
action: "transition".to_string(),
|
|
state: "active".to_string(),
|
|
value: transition.current_active,
|
|
},
|
|
IlmQueueTaskStats {
|
|
action: "transition".to_string(),
|
|
state: "compensation_running".to_string(),
|
|
value: transition.compensation_running,
|
|
},
|
|
]
|
|
}
|
|
|
|
fn ilm_task_event_stats(metrics: &ScannerMetricsReport) -> Vec<IlmTaskEventStats> {
|
|
let expiry = &metrics.lifecycle_expiry;
|
|
let transition = &metrics.lifecycle_transition;
|
|
vec![
|
|
IlmTaskEventStats {
|
|
action: "expiry".to_string(),
|
|
result: "queued".to_string(),
|
|
value: expiry.scanner_queued,
|
|
},
|
|
IlmTaskEventStats {
|
|
action: "expiry".to_string(),
|
|
result: "missed".to_string(),
|
|
value: expiry.scanner_missed,
|
|
},
|
|
IlmTaskEventStats {
|
|
action: "expiry".to_string(),
|
|
result: "blocked".to_string(),
|
|
value: expiry.scanner_blocked,
|
|
},
|
|
IlmTaskEventStats {
|
|
action: "expiry".to_string(),
|
|
result: "not_enqueued".to_string(),
|
|
value: expiry.scanner_not_enqueued,
|
|
},
|
|
IlmTaskEventStats {
|
|
action: "expiry".to_string(),
|
|
result: "failed".to_string(),
|
|
value: expiry.delete_failed,
|
|
},
|
|
IlmTaskEventStats {
|
|
action: "transition".to_string(),
|
|
result: "queued".to_string(),
|
|
value: transition.scanner_queued,
|
|
},
|
|
IlmTaskEventStats {
|
|
action: "transition".to_string(),
|
|
result: "missed".to_string(),
|
|
value: transition.scanner_missed,
|
|
},
|
|
IlmTaskEventStats {
|
|
action: "transition".to_string(),
|
|
result: "completed".to_string(),
|
|
value: transition.completed,
|
|
},
|
|
IlmTaskEventStats {
|
|
action: "transition".to_string(),
|
|
result: "failed".to_string(),
|
|
value: transition.failed,
|
|
},
|
|
]
|
|
}
|
|
|
|
fn ilm_backpressure_stats(metrics: &ScannerMetricsReport) -> Vec<IlmBackpressureStats> {
|
|
vec![
|
|
IlmBackpressureStats {
|
|
action: "expiry".to_string(),
|
|
reason: "queue_missed".to_string(),
|
|
value: metrics.lifecycle_expiry.queue_missed,
|
|
},
|
|
IlmBackpressureStats {
|
|
action: "transition".to_string(),
|
|
reason: "queue_full".to_string(),
|
|
value: metrics.lifecycle_transition.queue_full,
|
|
},
|
|
IlmBackpressureStats {
|
|
action: "transition".to_string(),
|
|
reason: "send_timeout".to_string(),
|
|
value: metrics.lifecycle_transition.queue_send_timeout,
|
|
},
|
|
]
|
|
}
|
|
|
|
/// Collect ILM metrics from the current lifecycle runtime state.
|
|
pub async fn collect_ilm_metric_stats() -> Option<IlmStats> {
|
|
collect_ilm_runtime_metric_stats().await.map(|stats| stats.stats)
|
|
}
|
|
|
|
pub(crate) async fn collect_ilm_runtime_metric_stats() -> Option<IlmRuntimeStats> {
|
|
let ilm = obs_ilm_runtime_snapshot().await;
|
|
let metrics = global_metrics().report().await;
|
|
let versions_scanned = scanner_lifecycle_checked_versions(&metrics);
|
|
|
|
Some(IlmRuntimeStats {
|
|
server: current_local_node_identity(),
|
|
action_tasks: ilm_action_task_stats(&ilm),
|
|
queue_tasks: ilm_queue_task_stats(&metrics),
|
|
task_events: ilm_task_event_stats(&metrics),
|
|
backpressure: ilm_backpressure_stats(&metrics),
|
|
versions_scanned,
|
|
stats: IlmStats {
|
|
expiry_pending_tasks: ilm.expiry_pending_tasks,
|
|
transition_active_tasks: ilm.transition_active_tasks,
|
|
transition_pending_tasks: ilm.transition_pending_tasks,
|
|
transition_missed_immediate_tasks: ilm.transition_missed_immediate_tasks,
|
|
transition_queue_full_tasks: ilm.transition_queue_full_tasks,
|
|
transition_queue_send_timeout_tasks: ilm.transition_queue_send_timeout_tasks,
|
|
transition_compensation_scheduled_tasks: ilm.transition_compensation_scheduled_tasks,
|
|
transition_compensation_running_tasks: ilm.transition_compensation_running_tasks,
|
|
versions_scanned,
|
|
},
|
|
})
|
|
}
|
|
|
|
/// Collect scanner metrics from a runtime source.
|
|
///
|
|
/// Task 5 maps scanner runtime snapshots from `global_metrics()` into the
|
|
/// rustfs-obs scanner collector shape.
|
|
fn scanner_bucket_scans_started(life_time_ops: &HashMap<String, u64>, bucket_scans_finished: u64) -> u64 {
|
|
life_time_ops
|
|
.get("scan_bucket_drive_start")
|
|
.copied()
|
|
.unwrap_or(bucket_scans_finished)
|
|
}
|
|
|
|
fn scanner_source_work_stats(source_work: &[ScannerSourceWorkSnapshot]) -> Vec<ScannerSourceWorkStats> {
|
|
let mut stats = source_work
|
|
.iter()
|
|
.filter(|work| !work.source.is_empty())
|
|
.map(|work| ScannerSourceWorkStats {
|
|
source: work.source.clone(),
|
|
checked: work.checked,
|
|
queued: work.queued,
|
|
executed: work.executed,
|
|
failed: work.failed,
|
|
skipped: work.skipped,
|
|
missed: work.missed,
|
|
})
|
|
.collect::<Vec<_>>();
|
|
stats.sort_by(|left, right| left.source.cmp(&right.source));
|
|
stats
|
|
}
|
|
|
|
fn scanner_current_cycle_source_work_stats(metrics: &ScannerMetricsReport) -> Vec<ScannerSourceWorkStats> {
|
|
let current = scanner_source_work_stats(&metrics.current_cycle_source_work);
|
|
if !current.is_empty() {
|
|
return current;
|
|
}
|
|
|
|
let mut sources = scanner_source_work_stats(&metrics.last_cycle_source_work)
|
|
.into_iter()
|
|
.map(|work| work.source)
|
|
.chain(
|
|
scanner_source_work_stats(&metrics.source_work)
|
|
.into_iter()
|
|
.map(|work| work.source),
|
|
)
|
|
.collect::<Vec<_>>();
|
|
sources.sort();
|
|
sources.dedup();
|
|
sources
|
|
.into_iter()
|
|
.map(|source| ScannerSourceWorkStats {
|
|
source,
|
|
..Default::default()
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
fn scanner_bucket_drive_result_stats(results: &[ScannerBucketDriveResultSnapshot]) -> Vec<ScannerBucketDriveResultStats> {
|
|
let mut stats = results
|
|
.iter()
|
|
.filter(|result| !result.bucket.is_empty() && !result.drive.is_empty() && !result.result.is_empty() && result.count > 0)
|
|
.map(|result| ScannerBucketDriveResultStats {
|
|
bucket: result.bucket.clone(),
|
|
drive: result.drive.clone(),
|
|
result: result.result.clone(),
|
|
count: result.count,
|
|
})
|
|
.collect::<Vec<_>>();
|
|
stats.sort_by(|left, right| {
|
|
left.bucket
|
|
.cmp(&right.bucket)
|
|
.then_with(|| left.drive.cmp(&right.drive))
|
|
.then_with(|| left.result.cmp(&right.result))
|
|
});
|
|
stats
|
|
}
|
|
|
|
fn scanner_active_bucket_drive_stats(results: &[ScannerActiveBucketDriveSnapshot]) -> Vec<ScannerActiveBucketDriveStats> {
|
|
let mut stats = results
|
|
.iter()
|
|
.filter(|result| !result.source.is_empty() && !result.bucket.is_empty() && !result.drive.is_empty() && result.count > 0)
|
|
.map(|result| ScannerActiveBucketDriveStats {
|
|
source: result.source.clone(),
|
|
bucket: result.bucket.clone(),
|
|
drive: result.drive.clone(),
|
|
count: result.count,
|
|
age_seconds: result.age_seconds,
|
|
})
|
|
.collect::<Vec<_>>();
|
|
stats.sort_by(|left, right| {
|
|
left.source
|
|
.cmp(&right.source)
|
|
.then_with(|| left.bucket.cmp(&right.bucket))
|
|
.then_with(|| left.drive.cmp(&right.drive))
|
|
});
|
|
stats
|
|
}
|
|
|
|
pub async fn collect_scanner_metric_stats() -> Option<ScannerStats> {
|
|
collect_scanner_runtime_metric_stats().await.map(|stats| stats.stats)
|
|
}
|
|
|
|
pub(crate) async fn collect_scanner_runtime_metric_stats() -> Option<ScannerRuntimeStats> {
|
|
let (metrics, runtime_details) = global_metrics().report_with_runtime_details().await;
|
|
let now = Timestamp::now();
|
|
let bucket_scans_finished = metrics.life_time_ops.get("scan_bucket_drive").copied().unwrap_or_default();
|
|
let bucket_scans_started = scanner_bucket_scans_started(&metrics.life_time_ops, bucket_scans_finished);
|
|
let bucket_scans_failed = metrics
|
|
.life_time_ops
|
|
.get("scan_bucket_drive_failure")
|
|
.copied()
|
|
.unwrap_or_default();
|
|
let completed_cycles = metrics.life_time_ops.get("scan_cycle").copied().unwrap_or_default();
|
|
let directories_scanned = metrics.life_time_ops.get("scan_folder").copied().unwrap_or_default();
|
|
let objects_scanned = metrics.life_time_ops.get("scan_object").copied().unwrap_or_default();
|
|
// Real scan coverage: every version the scanner walked, independent of ILM
|
|
// rules. The ILM-checked subset (`scanner_lifecycle_checked_versions`) still
|
|
// feeds the ILM collector's versions_scanned, but the scanner collector must
|
|
// report the total scanned versions — otherwise clusters without lifecycle
|
|
// rules report zero here while objects_scanned keeps climbing.
|
|
let versions_scanned = metrics.versions_scanned;
|
|
let reference_time = metrics.cycles_completed_at.last().copied().unwrap_or(metrics.current_started);
|
|
let last_activity_seconds = timestamp_elapsed_seconds_since(now, reference_time);
|
|
let active_paths = metrics.active_scan_paths as u64;
|
|
let current_cycle_age_seconds = current_scanner_cycle_age_seconds(metrics.current_cycle_active, metrics.current_started, now);
|
|
let current_scan_mode = scanner_scan_mode_code(&metrics.current_scan_mode);
|
|
let current_cycle_age = current_cycle_age_seconds as f64;
|
|
let last_cycle_duration = metrics.last_cycle_duration_seconds;
|
|
|
|
Some(ScannerRuntimeStats {
|
|
server: current_local_node_identity(),
|
|
source_work: scanner_source_work_stats(&metrics.source_work),
|
|
current_cycle_source_work: scanner_current_cycle_source_work_stats(&metrics),
|
|
last_cycle_source_work: scanner_source_work_stats(&metrics.last_cycle_source_work),
|
|
bucket_drive_results: scanner_bucket_drive_result_stats(&runtime_details.bucket_drive_results),
|
|
current_cycle_bucket_drive_results: scanner_bucket_drive_result_stats(
|
|
&runtime_details.current_cycle_bucket_drive_results,
|
|
),
|
|
last_cycle_bucket_drive_results: scanner_bucket_drive_result_stats(&runtime_details.last_cycle_bucket_drive_results),
|
|
active_bucket_drive_scans: scanner_active_bucket_drive_stats(&runtime_details.active_bucket_drive_scans),
|
|
stats: ScannerStats {
|
|
bucket_scans_finished,
|
|
bucket_scans_started,
|
|
bucket_scans_failed,
|
|
directories_scanned,
|
|
objects_scanned,
|
|
versions_scanned,
|
|
last_activity_seconds,
|
|
active_paths,
|
|
oldest_active_path_age_seconds: metrics.oldest_active_path_age_seconds,
|
|
current_set_scan_concurrency_limit: metrics.current_set_scan_concurrency_limit,
|
|
current_set_scans_queued: metrics.current_set_scans_queued,
|
|
current_set_scans_active: metrics.current_set_scans_active,
|
|
current_disk_scan_concurrency_limit: metrics.current_disk_scan_concurrency_limit,
|
|
current_disk_bucket_scans_queued: metrics.current_disk_bucket_scans_queued,
|
|
current_disk_bucket_scans_active: metrics.current_disk_bucket_scans_active,
|
|
throttle_idle_mode_enabled: metrics.throttle_idle_mode_enabled,
|
|
throttle_sleep_factor: metrics.throttle_sleep_factor,
|
|
throttle_max_sleep_seconds: metrics.throttle_max_sleep_seconds,
|
|
yield_every_n_objects: metrics.yield_every_n_objects,
|
|
cycle_interval_seconds: metrics.cycle_interval_seconds,
|
|
cycle_max_duration_seconds: metrics.cycle_max_duration_seconds,
|
|
cycle_max_objects: metrics.cycle_max_objects,
|
|
cycle_max_directories: metrics.cycle_max_directories,
|
|
bitrot_cycle_enabled: metrics.bitrot_cycle_enabled,
|
|
bitrot_cycle_seconds: metrics.bitrot_cycle_seconds,
|
|
current_cycle: metrics.current_cycle,
|
|
completed_cycles,
|
|
current_cycle_age_seconds,
|
|
current_cycle_objects_scanned: metrics.current_cycle_objects_scanned,
|
|
current_cycle_directories_scanned: metrics.current_cycle_directories_scanned,
|
|
current_cycle_bucket_drive_scans: metrics.current_cycle_bucket_drive_scans,
|
|
current_cycle_bucket_drive_failures: metrics.current_cycle_bucket_drive_failures,
|
|
current_cycle_objects_per_second: scanner_work_rate_per_second(
|
|
metrics.current_cycle_objects_scanned,
|
|
current_cycle_age,
|
|
),
|
|
current_cycle_directories_per_second: scanner_work_rate_per_second(
|
|
metrics.current_cycle_directories_scanned,
|
|
current_cycle_age,
|
|
),
|
|
current_cycle_bucket_drive_scans_per_second: scanner_work_rate_per_second(
|
|
metrics.current_cycle_bucket_drive_scans,
|
|
current_cycle_age,
|
|
),
|
|
current_cycle_yield_events: metrics.current_cycle_yield_events,
|
|
current_cycle_yield_duration_seconds: metrics.current_cycle_yield_duration_seconds,
|
|
current_cycle_throttle_sleep_events: metrics.current_cycle_throttle_sleep_events,
|
|
current_cycle_throttle_sleep_duration_seconds: metrics.current_cycle_throttle_sleep_duration_seconds,
|
|
current_cycle_ilm_actions: metrics.current_cycle_ilm_actions,
|
|
current_cycle_heal_objects: metrics.current_cycle_heal_objects,
|
|
current_cycle_replication_checks: metrics.current_cycle_replication_checks,
|
|
current_cycle_usage_saves: metrics.current_cycle_usage_saves,
|
|
current_scan_mode,
|
|
last_cycle_result: metrics.last_cycle_result_code,
|
|
last_cycle_partial_reason: metrics.last_cycle_partial_reason_code,
|
|
last_cycle_duration_seconds: metrics.last_cycle_duration_seconds,
|
|
last_cycle_objects_scanned: metrics.last_cycle_objects_scanned,
|
|
last_cycle_directories_scanned: metrics.last_cycle_directories_scanned,
|
|
last_cycle_bucket_drive_scans: metrics.last_cycle_bucket_drive_scans,
|
|
last_cycle_bucket_drive_failures: metrics.last_cycle_bucket_drive_failures,
|
|
last_cycle_objects_per_second: scanner_work_rate_per_second(metrics.last_cycle_objects_scanned, last_cycle_duration),
|
|
last_cycle_directories_per_second: scanner_work_rate_per_second(
|
|
metrics.last_cycle_directories_scanned,
|
|
last_cycle_duration,
|
|
),
|
|
last_cycle_bucket_drive_scans_per_second: scanner_work_rate_per_second(
|
|
metrics.last_cycle_bucket_drive_scans,
|
|
last_cycle_duration,
|
|
),
|
|
last_cycle_yield_events: metrics.last_cycle_yield_events,
|
|
last_cycle_yield_duration_seconds: metrics.last_cycle_yield_duration_seconds,
|
|
last_cycle_throttle_sleep_events: metrics.last_cycle_throttle_sleep_events,
|
|
last_cycle_throttle_sleep_duration_seconds: metrics.last_cycle_throttle_sleep_duration_seconds,
|
|
last_cycle_ilm_actions: metrics.last_cycle_ilm_actions,
|
|
last_cycle_heal_objects: metrics.last_cycle_heal_objects,
|
|
last_cycle_replication_checks: metrics.last_cycle_replication_checks,
|
|
last_cycle_usage_saves: metrics.last_cycle_usage_saves,
|
|
failed_cycles: metrics.failed_cycles,
|
|
superseded_cycles: metrics.superseded_cycles,
|
|
partial_cycles: metrics.partial_cycles,
|
|
partial_cycles_unknown: metrics.partial_cycles_unknown,
|
|
partial_cycles_runtime: metrics.partial_cycles_runtime,
|
|
partial_cycles_objects: metrics.partial_cycles_objects,
|
|
partial_cycles_directories: metrics.partial_cycles_directories,
|
|
},
|
|
})
|
|
}
|
|
|
|
/// Collect cluster-level compression statistics.
|
|
pub async fn collect_compression_cluster_stats() -> Option<CompressionClusterStats> {
|
|
let compression_data_usage = obs_load_compression_total_from_memory().await?;
|
|
|
|
let original_bytes_total = compression_data_usage.original_bytes_total;
|
|
let compressed_bytes_total = compression_data_usage.compressed_bytes_total;
|
|
let bytes_saved_total = original_bytes_total.saturating_sub(compressed_bytes_total);
|
|
let compression_ratio = if original_bytes_total > 0 {
|
|
compressed_bytes_total as f64 / original_bytes_total as f64
|
|
} else {
|
|
0.0
|
|
};
|
|
let compression_operations_total = compression_data_usage.compression_operations_total;
|
|
|
|
Some(CompressionClusterStats {
|
|
original_bytes_total,
|
|
compressed_bytes_total,
|
|
bytes_saved_total,
|
|
compression_ratio,
|
|
compression_operations_total,
|
|
})
|
|
}
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use rustfs_common::metrics::ScannerSourceWorkSnapshot;
|
|
use std::io::{Read, Write};
|
|
use std::net::{Shutdown, TcpListener, TcpStream};
|
|
use std::thread;
|
|
use std::time::Duration;
|
|
|
|
fn storage_info_with_one_online_disk() -> ObsStorageInfo {
|
|
let mut info = ObsStorageInfo::default();
|
|
info.disks.push(Default::default());
|
|
let disk = info.disks.last_mut().expect("inserted disk should exist");
|
|
disk.pool_index = 0;
|
|
disk.set_index = 0;
|
|
disk.disk_index = 0;
|
|
disk.state = DRIVE_STATE_OK.to_string();
|
|
disk.runtime_state = Some(DRIVE_STATE_ONLINE.to_string());
|
|
info
|
|
}
|
|
|
|
#[test]
|
|
fn bucket_usage_metrics_distinguish_unknown_from_confirmed_zero() {
|
|
assert_eq!(bucket_usage_metric_values(None, "bucket"), (None, None));
|
|
|
|
let mut data_usage = ObsDataUsageInfo {
|
|
usage_snapshot_complete: true,
|
|
..Default::default()
|
|
};
|
|
data_usage
|
|
.buckets_usage
|
|
.insert("bucket".to_string(), ObsBucketUsageInfo::default());
|
|
|
|
assert_eq!(bucket_usage_metric_values(Some(&data_usage), "bucket"), (Some(0), Some(0)));
|
|
assert_eq!(bucket_usage_metric_values(Some(&data_usage), "missing"), (None, None));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn cluster_usage_metrics_skip_incomplete_snapshot() {
|
|
assert!(
|
|
collect_cluster_usage_metric_stats_from_data_usage(ObsDataUsageInfo::default(), &HashSet::new())
|
|
.await
|
|
.is_none()
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn cluster_usage_metrics_publish_complete_empty_snapshot() {
|
|
let (cluster, buckets) = collect_cluster_usage_metric_stats_from_data_usage(
|
|
ObsDataUsageInfo {
|
|
usage_snapshot_complete: true,
|
|
usage_snapshot_converged: true,
|
|
..Default::default()
|
|
},
|
|
&HashSet::new(),
|
|
)
|
|
.await
|
|
.expect("complete empty usage should remain publishable");
|
|
|
|
assert_eq!(cluster.buckets_count, 0);
|
|
assert_eq!(cluster.objects_count, 0);
|
|
assert_eq!(cluster.total_bytes, 0);
|
|
assert!(cluster.snapshot_converged);
|
|
assert!(buckets.is_empty());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn cluster_usage_metrics_publish_unknown_convergence_as_unconverged() {
|
|
let (cluster, _) = collect_cluster_usage_metric_stats_from_data_usage(
|
|
ObsDataUsageInfo {
|
|
usage_snapshot_complete: true,
|
|
usage_snapshot_converged: false,
|
|
..Default::default()
|
|
},
|
|
&HashSet::new(),
|
|
)
|
|
.await
|
|
.expect("complete usage with unknown convergence should remain publishable");
|
|
|
|
assert!(!cluster.snapshot_converged);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn cluster_usage_metrics_skip_snapshot_for_a_different_bucket_namespace() {
|
|
let data_usage = ObsDataUsageInfo {
|
|
usage_snapshot_complete: true,
|
|
buckets_count: 1,
|
|
buckets_usage: HashMap::from([("stale-bucket".to_string(), ObsBucketUsageInfo::default())]),
|
|
..Default::default()
|
|
};
|
|
|
|
assert!(
|
|
collect_cluster_usage_metric_stats_from_data_usage(data_usage, &HashSet::from(["live-bucket".to_string()]))
|
|
.await
|
|
.is_none()
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn cluster_config_stats_accept_homogeneous_backend_parities() {
|
|
let stats = cluster_config_stats_from_backend_parities(Some(1), Some(2))
|
|
.expect("homogeneous scalar parities should produce cluster config metrics");
|
|
|
|
assert_eq!(stats.rrs_parity, 1);
|
|
assert_eq!(stats.standard_parity, 2);
|
|
}
|
|
|
|
#[test]
|
|
fn cluster_config_stats_skip_heterogeneous_backend_parities() {
|
|
assert!(cluster_config_stats_from_backend_parities(Some(1), None).is_none());
|
|
assert!(cluster_config_stats_from_backend_parities(None, Some(2)).is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn cluster_config_stats_reject_parity_larger_than_u32() {
|
|
let Ok(overflow) = usize::try_from(u64::from(u32::MAX) + 1) else {
|
|
return;
|
|
};
|
|
|
|
assert!(cluster_config_stats_from_backend_parities(Some(overflow), Some(2)).is_none());
|
|
assert!(cluster_config_stats_from_backend_parities(Some(1), Some(overflow)).is_none());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn node_local_resource_stats_use_stable_local_node_identity() {
|
|
let _guard = crate::node_identity::local_node_identity_test_guard().await;
|
|
let previous = rustfs_common::get_global_local_node_name().await;
|
|
rustfs_common::set_global_local_node_name("node1:9000").await;
|
|
|
|
let mut system = System::new_all();
|
|
let (cpu, memory) = collect_system_cpu_and_memory_stats_with(&mut system);
|
|
let host_network = collect_host_network_stats_with(&Networks::new());
|
|
let process_bundle =
|
|
process_metric_bundle_from_snapshots(ProcessResourceSnapshot::default(), ProcessSystemSnapshot::default());
|
|
|
|
assert_eq!(cpu.server, "node1:9000");
|
|
assert_eq!(memory.server, "node1:9000");
|
|
assert_eq!(host_network.server, "node1:9000");
|
|
assert_eq!(process_bundle.resource.server, "node1:9000");
|
|
assert_eq!(process_bundle.process.server, "node1:9000");
|
|
|
|
rustfs_common::set_global_local_node_name(&previous).await;
|
|
}
|
|
|
|
#[test]
|
|
fn erasure_set_stats_skip_unknown_backend_layout() {
|
|
let storage_info = storage_info_with_one_online_disk();
|
|
let backend = ObsBackendInfo {
|
|
drives_per_set: vec![8],
|
|
..Default::default()
|
|
};
|
|
|
|
assert!(erasure_set_stats_from_backend(&storage_info, &backend).is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn erasure_set_stats_skip_inconsistent_exact_layout_without_scalar_fallback() {
|
|
let storage_info = storage_info_with_one_online_disk();
|
|
let backend = ObsBackendInfo {
|
|
standard_sc_data: vec![6],
|
|
standard_sc_parities: vec![4],
|
|
standard_sc_parity: Some(2),
|
|
drives_per_set: vec![8],
|
|
..Default::default()
|
|
};
|
|
|
|
assert!(erasure_set_stats_from_backend(&storage_info, &backend).is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn erasure_set_stats_accept_truthful_legacy_scalar_layout() {
|
|
let storage_info = storage_info_with_one_online_disk();
|
|
let backend = ObsBackendInfo {
|
|
standard_sc_parity: Some(2),
|
|
drives_per_set: vec![8],
|
|
..Default::default()
|
|
};
|
|
|
|
let stats = erasure_set_stats_from_backend(&storage_info, &backend);
|
|
assert_eq!(stats.len(), 1);
|
|
assert_eq!(stats[0].size, 8);
|
|
assert_eq!(stats[0].parity, 2);
|
|
assert_eq!(stats[0].data_shards, 6);
|
|
}
|
|
|
|
#[test]
|
|
fn erasure_set_stats_exact_data_ignores_stale_scalar() {
|
|
let storage_info = storage_info_with_one_online_disk();
|
|
let backend = ObsBackendInfo {
|
|
standard_sc_data: vec![6],
|
|
standard_sc_parity: Some(4),
|
|
drives_per_set: vec![8],
|
|
..Default::default()
|
|
};
|
|
|
|
let stats = erasure_set_stats_from_backend(&storage_info, &backend);
|
|
assert_eq!(stats.len(), 1);
|
|
assert_eq!(stats[0].parity, 2);
|
|
assert_eq!(stats[0].data_shards, 6);
|
|
}
|
|
|
|
fn generate_loopback_traffic() -> std::io::Result<()> {
|
|
let listener = TcpListener::bind(("127.0.0.1", 0))?;
|
|
let addr = listener.local_addr()?;
|
|
let payload = vec![0x5Au8; 64 * 1024];
|
|
let expected_len = payload.len();
|
|
|
|
let server = thread::spawn(move || -> std::io::Result<()> {
|
|
let (mut stream, _) = listener.accept()?;
|
|
let mut received = 0usize;
|
|
let mut buf = [0u8; 8192];
|
|
|
|
while received < expected_len {
|
|
let read = stream.read(&mut buf)?;
|
|
if read == 0 {
|
|
break;
|
|
}
|
|
received += read;
|
|
}
|
|
|
|
Ok(())
|
|
});
|
|
|
|
let mut client = TcpStream::connect(addr)?;
|
|
client.write_all(&payload)?;
|
|
client.flush()?;
|
|
client.shutdown(Shutdown::Write)?;
|
|
|
|
server
|
|
.join()
|
|
.expect("loopback traffic server thread should complete successfully")?;
|
|
Ok(())
|
|
}
|
|
|
|
#[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")));
|
|
}
|
|
|
|
#[test]
|
|
fn disk_topology_label_rejects_unknown_negative_index() {
|
|
assert_eq!(disk_topology_label(-1), None);
|
|
assert_eq!(disk_topology_label(3), Some("3".to_string()));
|
|
}
|
|
|
|
#[test]
|
|
fn non_empty_disk_id_rejects_blank_uuid() {
|
|
assert_eq!(non_empty_disk_id(" "), None);
|
|
assert_eq!(non_empty_disk_id("disk-1"), Some("disk-1".to_string()));
|
|
}
|
|
|
|
#[test]
|
|
fn drive_server_label_uses_node_identity_for_urls_and_local_paths() {
|
|
assert_eq!(drive_server_label("http://node1:9000/data", "local:9000"), "node1:9000");
|
|
assert_eq!(drive_server_label("https://node2:9443/export/d1", "local:9000"), "node2:9443");
|
|
assert_eq!(drive_server_label("/mnt/data1", "local:9000"), "local:9000");
|
|
}
|
|
|
|
#[test]
|
|
fn drive_inode_stats_skip_unknown_zero_inode_totals() {
|
|
assert_eq!(drive_inode_stats(0, 0), (None, None, None));
|
|
assert_eq!(drive_inode_stats(2, 3), (Some(2), Some(3), Some(5)));
|
|
}
|
|
|
|
#[test]
|
|
fn drive_api_metrics_are_sorted_and_average_latency() {
|
|
let last_minute = HashMap::from([
|
|
("write".to_string(), (2, 6_000)),
|
|
("read".to_string(), (1, 3_000)),
|
|
("zero".to_string(), (0, 9_000)),
|
|
]);
|
|
let api_calls = HashMap::from([("write".to_string(), 9), ("read".to_string(), 4)]);
|
|
|
|
assert_eq!(drive_api_latency_micros(last_minute.values().copied()), Some(3));
|
|
assert_eq!(
|
|
drive_api_latency_by_api_micros(last_minute.iter().map(|(api, (count, acc_time))| (api, *count, *acc_time))),
|
|
vec![("read".to_string(), 3), ("write".to_string(), 3), ("zero".to_string(), 0)]
|
|
);
|
|
assert_eq!(drive_api_calls(api_calls.iter()), vec![("read".to_string(), 4), ("write".to_string(), 9)]);
|
|
}
|
|
|
|
#[test]
|
|
fn drive_api_latency_skips_zero_denominators() {
|
|
let last_minute = HashMap::from([("zero".to_string(), (0, 9_000))]);
|
|
|
|
assert_eq!(drive_api_latency_micros(last_minute.values().copied()), Some(0));
|
|
assert_eq!(
|
|
drive_api_latency_by_api_micros(last_minute.iter().map(|(api, (count, acc_time))| (api, *count, *acc_time))),
|
|
vec![("zero".to_string(), 0)]
|
|
);
|
|
assert_eq!(drive_api_latency_micros([].into_iter()), None);
|
|
}
|
|
|
|
#[test]
|
|
fn derive_erasure_set_quorum_shape_handles_standard_layout() {
|
|
let shape = derive_erasure_set_quorum_shape(16, 4);
|
|
|
|
assert_eq!(
|
|
shape,
|
|
ErasureSetQuorumShape {
|
|
data_shards: 12,
|
|
read_quorum: 12,
|
|
write_quorum: 12,
|
|
read_tolerance: 4,
|
|
write_tolerance: 4,
|
|
}
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn derive_erasure_set_quorum_shape_handles_equal_data_and_parity() {
|
|
let shape = derive_erasure_set_quorum_shape(4, 2);
|
|
|
|
assert_eq!(
|
|
shape,
|
|
ErasureSetQuorumShape {
|
|
data_shards: 2,
|
|
read_quorum: 2,
|
|
write_quorum: 3,
|
|
read_tolerance: 2,
|
|
write_tolerance: 1,
|
|
}
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn apply_erasure_set_health_marks_read_and_write_health_from_online_count() {
|
|
let mut stats = ErasureSetStats {
|
|
read_quorum: 3,
|
|
write_quorum: 4,
|
|
online_drives_count: 3,
|
|
..Default::default()
|
|
};
|
|
|
|
apply_erasure_set_health(&mut stats);
|
|
assert_eq!(stats.read_health, 1);
|
|
assert_eq!(stats.write_health, 0);
|
|
assert_eq!(stats.health, 0);
|
|
|
|
stats.online_drives_count = 4;
|
|
apply_erasure_set_health(&mut stats);
|
|
assert_eq!(stats.read_health, 1);
|
|
assert_eq!(stats.write_health, 1);
|
|
assert_eq!(stats.health, 1);
|
|
}
|
|
|
|
#[test]
|
|
fn current_scanner_cycle_age_seconds_returns_zero_when_idle() {
|
|
let now = Timestamp::constant(1_700_000_000, 0);
|
|
|
|
assert_eq!(
|
|
current_scanner_cycle_age_seconds(false, now - jiff::SignedDuration::from_secs(30), now),
|
|
0
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn current_scanner_cycle_age_seconds_clamps_future_start() {
|
|
let now = Timestamp::constant(1_700_000_000, 0);
|
|
|
|
assert_eq!(current_scanner_cycle_age_seconds(true, now + jiff::SignedDuration::from_secs(30), now), 0);
|
|
}
|
|
|
|
#[test]
|
|
fn current_scanner_cycle_age_seconds_reports_active_first_cycle_elapsed_time() {
|
|
let now = Timestamp::constant(1_700_000_000, 0);
|
|
|
|
assert_eq!(
|
|
current_scanner_cycle_age_seconds(true, now - jiff::SignedDuration::from_secs(45), now),
|
|
45
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn scanner_scan_mode_code_maps_known_modes() {
|
|
assert_eq!(scanner_scan_mode_code(HealScanMode::Normal.as_str()), HealScanMode::Normal as u8 as u64);
|
|
assert_eq!(scanner_scan_mode_code(HealScanMode::Deep.as_str()), HealScanMode::Deep as u8 as u64);
|
|
}
|
|
|
|
#[test]
|
|
fn scanner_scan_mode_code_maps_unknown_mode() {
|
|
assert_eq!(scanner_scan_mode_code(""), HealScanMode::Unknown as u8 as u64);
|
|
}
|
|
|
|
#[test]
|
|
fn scanner_bucket_scans_started_uses_explicit_started_count() {
|
|
let mut life_time_ops = HashMap::new();
|
|
life_time_ops.insert("scan_bucket_drive_start".to_string(), 7);
|
|
|
|
assert_eq!(scanner_bucket_scans_started(&life_time_ops, 5), 7);
|
|
}
|
|
|
|
#[test]
|
|
fn scanner_bucket_scans_started_falls_back_to_finished_count() {
|
|
let life_time_ops = HashMap::new();
|
|
|
|
assert_eq!(scanner_bucket_scans_started(&life_time_ops, 5), 5);
|
|
}
|
|
|
|
#[test]
|
|
fn ilm_action_task_stats_maps_runtime_states() {
|
|
let stats = ilm_action_task_stats(&ObsIlmRuntimeSnapshot {
|
|
expiry_pending_tasks: 1,
|
|
transition_active_tasks: 2,
|
|
transition_pending_tasks: 3,
|
|
transition_missed_immediate_tasks: 4,
|
|
transition_queue_full_tasks: 5,
|
|
transition_queue_send_timeout_tasks: 6,
|
|
transition_compensation_scheduled_tasks: 7,
|
|
transition_compensation_running_tasks: 8,
|
|
});
|
|
|
|
assert_eq!(stats.len(), 8);
|
|
assert_eq!(stats[0].action, "expiry");
|
|
assert_eq!(stats[0].state, "pending");
|
|
assert_eq!(stats[0].value, 1);
|
|
assert!(
|
|
stats
|
|
.iter()
|
|
.any(|task| { task.action == "transition" && task.state == "queue_send_timeout" && task.value == 6 })
|
|
);
|
|
assert!(
|
|
stats
|
|
.iter()
|
|
.any(|task| { task.action == "transition" && task.state == "compensation_running" && task.value == 8 })
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn scanner_lifecycle_checked_versions_uses_lifecycle_checked_source_work() {
|
|
let report = ScannerMetricsReport {
|
|
source_work: vec![
|
|
ScannerSourceWorkSnapshot {
|
|
source: "usage".to_string(),
|
|
checked: 11,
|
|
..Default::default()
|
|
},
|
|
ScannerSourceWorkSnapshot {
|
|
source: "lifecycle".to_string(),
|
|
checked: 37,
|
|
..Default::default()
|
|
},
|
|
],
|
|
life_time_ilm: HashMap::from([("TransitionAction".to_string(), 5), ("DeleteAction".to_string(), 7)]),
|
|
..Default::default()
|
|
};
|
|
|
|
assert_eq!(scanner_lifecycle_checked_versions(&report), 37);
|
|
}
|
|
|
|
#[test]
|
|
fn ilm_detail_stats_keep_expiry_and_transition_results_separate() {
|
|
let report = ScannerMetricsReport {
|
|
lifecycle_expiry: rustfs_common::metrics::ScannerLifecycleExpirySnapshot {
|
|
current_queued: 2,
|
|
current_active: 1,
|
|
scanner_queued: 10,
|
|
scanner_missed: 3,
|
|
delete_failed: 4,
|
|
..Default::default()
|
|
},
|
|
lifecycle_transition: rustfs_common::metrics::ScannerLifecycleTransitionSnapshot {
|
|
current_queued: 5,
|
|
current_active: 6,
|
|
queue_full: 7,
|
|
queue_send_timeout: 8,
|
|
scanner_queued: 11,
|
|
completed: 12,
|
|
failed: 13,
|
|
..Default::default()
|
|
},
|
|
..Default::default()
|
|
};
|
|
|
|
let queues = ilm_queue_task_stats(&report);
|
|
assert!(
|
|
queues
|
|
.iter()
|
|
.any(|task| task.action == "expiry" && task.state == "pending" && task.value == 2)
|
|
);
|
|
assert!(
|
|
queues
|
|
.iter()
|
|
.any(|task| task.action == "transition" && task.state == "active" && task.value == 6)
|
|
);
|
|
|
|
let events = ilm_task_event_stats(&report);
|
|
assert!(
|
|
events
|
|
.iter()
|
|
.any(|event| event.action == "expiry" && event.result == "failed" && event.value == 4)
|
|
);
|
|
assert!(
|
|
events
|
|
.iter()
|
|
.any(|event| event.action == "transition" && event.result == "completed" && event.value == 12)
|
|
);
|
|
assert!(
|
|
events
|
|
.iter()
|
|
.any(|event| event.action == "transition" && event.result == "failed" && event.value == 13)
|
|
);
|
|
|
|
let backpressure = ilm_backpressure_stats(&report);
|
|
assert!(
|
|
backpressure
|
|
.iter()
|
|
.any(|event| event.action == "transition" && event.reason == "queue_full" && event.value == 7)
|
|
);
|
|
assert!(
|
|
backpressure
|
|
.iter()
|
|
.any(|event| event.action == "transition" && event.reason == "send_timeout" && event.value == 8)
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn scanner_source_work_stats_sorts_and_skips_empty_source() {
|
|
let stats = scanner_source_work_stats(&[
|
|
ScannerSourceWorkSnapshot {
|
|
source: "usage".to_string(),
|
|
checked: 11,
|
|
queued: 2,
|
|
executed: 3,
|
|
failed: 4,
|
|
skipped: 5,
|
|
missed: 6,
|
|
},
|
|
ScannerSourceWorkSnapshot {
|
|
checked: 99,
|
|
queued: 99,
|
|
executed: 99,
|
|
failed: 99,
|
|
skipped: 99,
|
|
missed: 99,
|
|
..Default::default()
|
|
},
|
|
ScannerSourceWorkSnapshot {
|
|
source: "lifecycle".to_string(),
|
|
checked: 21,
|
|
queued: 7,
|
|
executed: 8,
|
|
failed: 9,
|
|
skipped: 10,
|
|
missed: 12,
|
|
},
|
|
]);
|
|
|
|
assert_eq!(stats.len(), 2);
|
|
assert_eq!(stats[0].source, "lifecycle");
|
|
assert_eq!(stats[0].checked, 21);
|
|
assert_eq!(stats[0].missed, 12);
|
|
assert_eq!(stats[1].source, "usage");
|
|
assert_eq!(stats[1].failed, 4);
|
|
}
|
|
|
|
#[test]
|
|
fn scanner_current_cycle_source_work_stats_zeroes_idle_sources() {
|
|
let report = ScannerMetricsReport {
|
|
source_work: vec![ScannerSourceWorkSnapshot {
|
|
source: "usage".to_string(),
|
|
checked: 11,
|
|
queued: 2,
|
|
..Default::default()
|
|
}],
|
|
last_cycle_source_work: vec![ScannerSourceWorkSnapshot {
|
|
source: "lifecycle".to_string(),
|
|
checked: 21,
|
|
queued: 7,
|
|
..Default::default()
|
|
}],
|
|
..Default::default()
|
|
};
|
|
|
|
let stats = scanner_current_cycle_source_work_stats(&report);
|
|
|
|
assert_eq!(stats.len(), 2);
|
|
assert_eq!(stats[0].source, "lifecycle");
|
|
assert_eq!(stats[0].checked, 0);
|
|
assert_eq!(stats[1].source, "usage");
|
|
assert_eq!(stats[1].queued, 0);
|
|
}
|
|
|
|
#[test]
|
|
fn scanner_current_cycle_source_work_stats_keeps_active_values() {
|
|
let report = ScannerMetricsReport {
|
|
source_work: vec![ScannerSourceWorkSnapshot {
|
|
source: "usage".to_string(),
|
|
checked: 11,
|
|
..Default::default()
|
|
}],
|
|
current_cycle_source_work: vec![ScannerSourceWorkSnapshot {
|
|
source: "usage".to_string(),
|
|
checked: 3,
|
|
..Default::default()
|
|
}],
|
|
..Default::default()
|
|
};
|
|
|
|
let stats = scanner_current_cycle_source_work_stats(&report);
|
|
|
|
assert_eq!(stats.len(), 1);
|
|
assert_eq!(stats[0].source, "usage");
|
|
assert_eq!(stats[0].checked, 3);
|
|
}
|
|
|
|
#[test]
|
|
fn scanner_lifecycle_checked_versions_defaults_to_zero_when_lifecycle_missing() {
|
|
let report = ScannerMetricsReport {
|
|
source_work: vec![ScannerSourceWorkSnapshot {
|
|
source: "usage".to_string(),
|
|
checked: 11,
|
|
..Default::default()
|
|
}],
|
|
..Default::default()
|
|
};
|
|
|
|
assert_eq!(scanner_lifecycle_checked_versions(&report), 0);
|
|
}
|
|
|
|
#[test]
|
|
fn host_network_stats_require_a_persistent_networks_snapshot() -> std::io::Result<()> {
|
|
if !sysinfo::IS_SUPPORTED_SYSTEM {
|
|
return Ok(());
|
|
}
|
|
|
|
let mut persistent_networks = Networks::new();
|
|
persistent_networks.refresh(true);
|
|
let initial_stats = collect_host_network_stats_with(&persistent_networks);
|
|
assert_eq!(
|
|
initial_stats.total_received + initial_stats.total_transmitted,
|
|
0,
|
|
"the first refresh only seeds sysinfo's baseline snapshot"
|
|
);
|
|
|
|
match generate_loopback_traffic() {
|
|
Ok(()) => {}
|
|
Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => {
|
|
return Ok(());
|
|
}
|
|
Err(err) => return Err(err),
|
|
}
|
|
thread::sleep(Duration::from_millis(100));
|
|
|
|
let refreshed_stats = collect_host_network_stats(&mut persistent_networks);
|
|
assert!(
|
|
refreshed_stats.total_received > 0 || refreshed_stats.total_transmitted > 0,
|
|
"a persistent Networks instance should report non-zero loopback deltas after traffic"
|
|
);
|
|
|
|
let recreated_stats = collect_host_network_stats_with(&Networks::new_with_refreshed_list());
|
|
assert_eq!(
|
|
recreated_stats.total_received + recreated_stats.total_transmitted,
|
|
0,
|
|
"recreating Networks loses the prior refresh baseline and yields zero deltas"
|
|
);
|
|
Ok(())
|
|
}
|
|
|
|
#[test]
|
|
fn scanner_work_rate_per_second_reports_rate() {
|
|
assert_eq!(scanner_work_rate_per_second(90, 45.0), 2.0);
|
|
}
|
|
|
|
#[test]
|
|
fn scanner_work_rate_per_second_returns_zero_for_invalid_seconds() {
|
|
assert_eq!(scanner_work_rate_per_second(90, 0.0), 0.0);
|
|
assert_eq!(scanner_work_rate_per_second(90, f64::INFINITY), 0.0);
|
|
assert_eq!(scanner_work_rate_per_second(90, f64::NAN), 0.0);
|
|
}
|
|
|
|
#[test]
|
|
fn usable_capacity_used_bytes_matches_usable_total_minus_free() {
|
|
assert_eq!(usable_capacity_used_bytes(240, 90), 150);
|
|
}
|
|
|
|
#[test]
|
|
fn usable_capacity_used_bytes_saturates_when_free_exceeds_total() {
|
|
assert_eq!(usable_capacity_used_bytes(90, 240), 0);
|
|
}
|
|
}
|