Files
rustfs/crates/io-metrics/src/lib.rs
T
2026-04-08 20:58:17 +08:00

1082 lines
35 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.
//! RustFS metrics collection and reporting.
//!
//! This crate provides the **single source of truth** for all metrics
//! in RustFS. It uses the `metrics` crate for reporting to OTEL exporters.
//!
//! # Architecture
//!
//! - **Free functions**: Simple `record_*()` functions for quick metric reporting
//! - **PerformanceMetrics**: Shared atomic counter struct for advanced use cases
//! - **MetricsCollector**: I/O operation tracking with percentile calculation
//! - **AutoTuner**: Automatic performance optimization based on metrics
//!
//! # Usage
//!
//! ```rust,no_run
//! use rustfs_io_metrics::{MetricsCollector, PerformanceMetrics, record_get_object};
//! use std::sync::Arc;
//! use std::time::Duration;
//!
//! # #[tokio::main]
//! # async fn main() {
//! // Simple recording
//! record_get_object(100.0, 1024);
//!
//! // Advanced usage with collector
//! let metrics = Arc::new(PerformanceMetrics::new());
//! let collector = MetricsCollector::new(metrics, 1000);
//! collector.record_io_operation(1024, Duration::from_millis(10), true).await;
//! # }
//! ```
// Import macros from the metrics crate
#[macro_use]
extern crate metrics;
// Public modules
pub mod adaptive_ttl;
pub mod autotuner;
pub mod backpressure_metrics;
pub mod cache_config;
pub mod capacity_metrics;
pub mod collector;
pub mod config;
pub mod deadlock_metrics;
pub mod io_metrics;
pub mod lock_metrics;
pub mod performance;
pub mod timeout_metrics;
pub use autotuner::{AutoTuner, TunerConfig, TuningResult};
// Cache config exports
pub use cache_config::{AdaptiveTTL, CacheConfig, CacheConfigError, CacheHealthStatus, CacheStats};
// Adaptive TTL exports
pub use adaptive_ttl::{
AccessRecord, AccessTracker, AdaptiveTTLStats, record_access_pattern_change, record_early_eviction, record_ttl_adjustment,
record_ttl_expiration,
};
// Capacity metrics exports
pub use capacity_metrics::{
record_capacity_cache_hit, record_capacity_cache_miss, record_capacity_cache_served, record_capacity_current_bytes,
record_capacity_dirty_disk_count, record_capacity_dynamic_timeout, record_capacity_refresh_inflight,
record_capacity_refresh_joiner, record_capacity_refresh_request, record_capacity_refresh_result,
record_capacity_refresh_scope, record_capacity_scan_disk, record_capacity_scan_mode, record_capacity_scan_sampling,
record_capacity_stall_detected, record_capacity_symlink, record_capacity_timeout_fallback, record_capacity_update_completed,
record_capacity_update_failed, record_capacity_write_operation,
};
// I/O metrics exports
pub use io_metrics::{
IoSchedulerStats, record_bandwidth_observation, record_buffer_size_adjustment, record_io_priority_decision,
record_io_scheduler_decision, record_load_level_change, record_queue_operation, record_starvation_event,
};
// Backpressure metrics exports
pub use backpressure_metrics::{
record_backpressure_activation, record_backpressure_deactivation, record_backpressure_rejection,
record_backpressure_state_change, record_concurrent_operations,
};
// Deadlock metrics exports
pub use deadlock_metrics::{
record_deadlock_detected, record_lock_acquisition, record_lock_contention, record_lock_release, record_long_held_lock,
record_wait_edge_added, record_wait_edge_removed,
};
// Lock metrics exports
pub use lock_metrics::{
LockMetricsSummary, record_contention_event, record_early_release, record_lock_hold_time, record_lock_optimization_enabled,
record_spin_attempt, record_spin_count_change,
};
// Timeout metrics exports
pub use timeout_metrics::{
TimeoutMetricsSummary, record_dynamic_timeout, record_operation_completion, record_operation_duration,
record_operation_progress, record_stalled_operation, record_timeout_event,
};
// Config exports
pub use config::{
BackpressureSettings, CacheSettings, DEFAULT_BASE_BUFFER_SIZE, DEFAULT_CACHE_MAX_CAPACITY, DEFAULT_CACHE_MAX_MEMORY,
DEFAULT_CACHE_TTL_SECS, DEFAULT_MAX_BUFFER_SIZE, DEFAULT_MAX_CONCURRENT_READS, DEFAULT_MIN_BUFFER_SIZE,
DeadlockDetectionSettings, IoConfig, IoSchedulerSettings, TimeoutSettings,
};
// Re-exports for convenience
pub use collector::MetricsCollector;
pub use metric_names::data_plane;
pub use performance::PerformanceMetrics;
/// High-level request path selected for an I/O operation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum IoPath {
Fast,
Legacy,
}
impl IoPath {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::Fast => "fast",
Self::Legacy => "legacy",
}
}
}
/// Effective copy mode observed for an I/O operation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CopyMode {
TrueZeroCopy,
SharedBytes,
SingleCopy,
Reconstructed,
Transformed,
}
impl CopyMode {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::TrueZeroCopy => "true_zero_copy",
Self::SharedBytes => "shared_bytes",
Self::SingleCopy => "single_copy",
Self::Reconstructed => "reconstructed",
Self::Transformed => "transformed",
}
}
}
/// Stage where a data plane decision or fallback happened.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum IoStage {
Unknown,
ReadSetup,
HttpBridge,
LocalDiskChunk,
RangeGuard,
PutTransform,
}
impl IoStage {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::Unknown => "unknown",
Self::ReadSetup => "read_setup",
Self::HttpBridge => "http_bridge",
Self::LocalDiskChunk => "local_disk_chunk",
Self::RangeGuard => "range_guard",
Self::PutTransform => "put_transform",
}
}
}
/// Reason why the data plane fell back from a preferred path.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FallbackReason {
Unknown,
MmapDisabled,
MmapUnavailable,
SmallObject,
WindowLimitExceeded,
UnalignedWindow,
RangeNotSupported,
EncryptionEnabled,
CompressionEnabled,
TransformEncryptionLegacy,
TransformCompressionLegacy,
TransformCompressionEncryptionLegacy,
ChunkBridgeUnavailable,
NonLocalBackend,
}
impl FallbackReason {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::Unknown => "unknown",
Self::MmapDisabled => "mmap_disabled",
Self::MmapUnavailable => "mmap_unavailable",
Self::SmallObject => "small_object",
Self::WindowLimitExceeded => "window_limit_exceeded",
Self::UnalignedWindow => "unaligned_window",
Self::RangeNotSupported => "range_not_supported",
Self::EncryptionEnabled => "encryption_enabled",
Self::CompressionEnabled => "compression_enabled",
Self::TransformEncryptionLegacy => "transform_encryption_legacy",
Self::TransformCompressionLegacy => "transform_compression_legacy",
Self::TransformCompressionEncryptionLegacy => "transform_compression_encryption_legacy",
Self::ChunkBridgeUnavailable => "chunk_bridge_unavailable",
Self::NonLocalBackend => "non_local_backend",
}
}
}
#[inline(always)]
fn put_size_bucket_label(size_bytes: i64) -> &'static str {
match size_bytes {
..=0 => "unknown",
1..=16_384 => "le_16kib",
16_385..=65_536 => "le_64kib",
65_537..=262_144 => "le_256kib",
262_145..=1_048_576 => "le_1mib",
_ => "gt_1mib",
}
}
/// Record GetObject request start.
#[inline(always)]
pub fn record_get_object_request_start(concurrent_requests: usize) {
counter!("rustfs_io_get_object_requests_total").increment(1);
gauge!("rustfs_io_get_object_concurrent_requests").set(concurrent_requests as f64);
}
/// Record GetObject request start without concurrency context.
#[inline(always)]
pub fn record_get_object_request_started() {
counter!("rustfs_io_get_object_requests_total").increment(1);
}
/// Record GetObject request result.
#[inline(always)]
pub fn record_get_object_request_result(status: &str, duration_secs: f64) {
counter!("rustfs_io_get_object_request_results_total", "status" => status.to_string()).increment(1);
histogram!("rustfs_io_get_object_request_duration_seconds", "status" => status.to_string()).record(duration_secs);
}
/// Record GetObject timeout for a specific stage.
#[inline(always)]
pub fn record_get_object_timeout(stage: Option<&str>, elapsed_secs: Option<f64>) {
match stage {
Some(stage) => counter!("rustfs_io_get_object_timeout_total", "stage" => stage.to_string()).increment(1),
None => counter!("rustfs_io_get_object_timeout_total").increment(1),
}
if let Some(elapsed_secs) = elapsed_secs {
histogram!("rustfs_io_get_object_timeout_elapsed_seconds").record(elapsed_secs);
}
}
/// Record GetObject completion.
#[inline(always)]
pub fn record_get_object_completion(total_duration_secs: f64, response_size_bytes: i64, buffer_size_bytes: usize) {
counter!("rustfs_io_get_object_completed_total").increment(1);
histogram!("rustfs_io_get_object_total_duration_seconds").record(total_duration_secs);
histogram!("rustfs_io_get_object_response_size_bytes").record(response_size_bytes as f64);
histogram!("rustfs_io_get_object_buffer_size_bytes").record(buffer_size_bytes as f64);
}
/// Record I/O queue congestion observation.
#[inline(always)]
pub fn record_io_queue_congestion() {
counter!("rustfs_io_queue_congestion_total").increment(1);
}
/// Record I/O priority assignment.
#[inline(always)]
pub fn record_io_priority_assignment(priority: &str) {
counter!("rustfs_io_priority_assigned_total", "priority" => priority.to_string()).increment(1);
}
/// Record detailed GetObject I/O orchestration metrics.
#[inline(always)]
pub fn record_get_object_io_state(
permit_wait_secs: f64,
queue_utilization_percent: f64,
permits_in_use: usize,
permits_available: usize,
load_level: &str,
buffer_multiplier: f64,
) {
histogram!("rustfs_io_disk_permit_wait_duration_seconds").record(permit_wait_secs);
gauge!("rustfs_io_queue_utilization_percent").set(queue_utilization_percent);
gauge!("rustfs_io_queue_permits_in_use").set(permits_in_use as f64);
gauge!("rustfs_io_queue_permits_available").set(permits_available as f64);
gauge!("rustfs_io_buffer_multiplier").set(buffer_multiplier);
counter!("rustfs_io_strategy_selected_total", "level" => load_level.to_string()).increment(1);
}
/// Record which request path was selected for an operation.
#[inline(always)]
pub fn record_io_path_selected(operation: &'static str, io_path: IoPath) {
counter!(
metric_names::data_plane::PATH_SELECTED_TOTAL,
"path" => operation,
"mode" => io_path.as_str()
)
.increment(1);
}
/// Record the effective copy mode for an operation.
#[inline(always)]
pub fn record_io_copy_mode(operation: &'static str, copy_mode: CopyMode, size_bytes: usize) {
counter!(
metric_names::data_plane::COPY_MODE_BYTES_TOTAL,
"path" => operation,
"mode" => copy_mode.as_str()
)
.increment(size_bytes as u64);
}
/// Record a data plane fallback decision.
#[inline(always)]
pub fn record_io_fallback(stage: IoStage, reason: FallbackReason) {
counter!(
metric_names::data_plane::FALLBACK_TOTAL,
"stage" => stage.as_str(),
"reason" => reason.as_str()
)
.increment(1);
}
/// Record the currently active mmap bytes held by LocalDisk chunk streams.
#[inline(always)]
pub fn record_local_disk_active_mmap_bytes(active_bytes: usize) {
gauge!(metric_names::data_plane::LOCAL_DISK_ACTIVE_MMAP_BYTES).set(active_bytes as f64);
}
/// Record pooled chunk usage in LocalDisk compatibility paths.
#[inline(always)]
pub fn record_local_disk_pooled_chunk(source: &'static str, size_bytes: usize) {
counter!(
metric_names::data_plane::LOCAL_DISK_POOLED_CHUNKS_TOTAL,
"source" => source
)
.increment(1);
counter!(
metric_names::data_plane::LOCAL_DISK_POOLED_BYTES_TOTAL,
"source" => source
)
.increment(size_bytes as u64);
}
/// Record a compatibility chunk-stream aggregation performed by `read_file_zero_copy()`.
#[inline(always)]
pub fn record_local_disk_compat_collect(chunk_count: usize, total_bytes: usize) {
counter!(metric_names::data_plane::LOCAL_DISK_COMPAT_COLLECT_TOTAL).increment(1);
histogram!(metric_names::data_plane::LOCAL_DISK_COMPAT_COLLECT_CHUNKS).record(chunk_count as f64);
histogram!(metric_names::data_plane::LOCAL_DISK_COMPAT_COLLECT_BYTES).record(total_bytes as f64);
}
/// Record an attempted PUT fast path.
#[inline(always)]
pub fn record_put_object_attempted_fast_path(size_bytes: i64) {
counter!(metric_names::data_plane::PUT_FAST_PATH_ATTEMPTS_TOTAL).increment(1);
if size_bytes > 0 {
histogram!(metric_names::data_plane::PUT_FAST_PATH_ATTEMPT_SIZE_BYTES).record(size_bytes as f64);
}
}
/// Record which transformed PUT pipeline was selected.
#[inline(always)]
pub fn record_put_transform_selected(kind: &'static str, io_path: IoPath, size_bytes: usize) {
counter!(
metric_names::data_plane::PUT_TRANSFORM_SELECTED_TOTAL,
"kind" => kind,
"mode" => io_path.as_str()
)
.increment(1);
histogram!(
metric_names::data_plane::PUT_TRANSFORM_SIZE_BYTES,
"kind" => kind,
"mode" => io_path.as_str()
)
.record(size_bytes as f64);
}
/// Record PUT path selection with size-bucket context.
#[inline(always)]
pub fn record_put_path_selected(size_bytes: i64, io_path: IoPath) {
counter!(
"rustfs.s3.put_object.path.selected.total",
"mode" => io_path.as_str(),
"size_bucket" => put_size_bucket_label(size_bytes)
)
.increment(1);
}
/// Record PUT copy mode with size-bucket context.
#[inline(always)]
pub fn record_put_copy_mode(size_bytes: i64, copy_mode: CopyMode) {
counter!(
"rustfs.s3.put_object.copy_mode.total",
"mode" => copy_mode.as_str(),
"size_bucket" => put_size_bucket_label(size_bytes)
)
.increment(1);
}
/// Record PUT fallback with size-bucket context.
#[inline(always)]
pub fn record_put_fallback(size_bytes: i64, reason: FallbackReason) {
counter!(
"rustfs.s3.put_object.fallback.total",
"reason" => reason.as_str(),
"size_bucket" => put_size_bucket_label(size_bytes)
)
.increment(1);
}
/// Record inline-object selection for PUT with size-bucket context.
#[inline(always)]
pub fn record_put_inline_selected(size_bytes: i64, versioned: bool) {
counter!(
"rustfs.s3.put_object.inline.selected.total",
"versioned" => if versioned { "true" } else { "false" },
"size_bucket" => put_size_bucket_label(size_bytes)
)
.increment(1);
}
// ============================================================================
// BytesPool Metrics
// ============================================================================
/// Record BytesPool buffer acquisition.
///
/// # Arguments
///
/// * `tier` - Pool tier ("small", "medium", "large", "xlarge")
/// * `size` - Buffer size acquired
/// * `from_pool` - Whether buffer was reused from pool
#[inline(always)]
pub fn record_bytes_pool_acquire(tier: &str, size: usize, from_pool: bool) {
counter!("rustfs.bytes.pool.acquisitions.total", "tier" => tier.to_string()).increment(1);
gauge!("rustfs.bytes.pool.size.bytes", "tier" => tier.to_string()).set(size as f64);
if from_pool {
counter!("rustfs.bytes.pool.hits.total", "tier" => tier.to_string()).increment(1);
} else {
counter!("rustfs.bytes.pool.misses.total", "tier" => tier.to_string()).increment(1);
}
}
/// Record BytesPool buffer return.
///
/// # Arguments
///
/// * `tier` - Pool tier ("small", "medium", "large", "xlarge")
#[inline(always)]
pub fn record_bytes_pool_return(tier: &str) {
counter!("rustfs.bytes.pool.returns.total", "tier" => tier.to_string()).increment(1);
}
/// Record current BytesPool allocated bytes.
///
/// # Arguments
///
/// * `tier` - Pool tier
/// * `bytes` - Currently allocated bytes
#[inline(always)]
pub fn record_bytes_pool_allocated(tier: &str, bytes: u64) {
gauge!("rustfs.bytes.pool.allocated.bytes", "tier" => tier.to_string()).set(bytes as f64);
}
/// Get BytesPool hit rate as a gauge metric.
///
/// # Arguments
///
/// * `tier` - Pool tier
/// * `hit_rate` - Hit rate (0.0 - 1.0)
#[inline(always)]
pub fn record_bytes_pool_hit_rate(tier: &str, hit_rate: f64) {
gauge!("rustfs.bytes.pool.hit.rate", "tier" => tier.to_string()).set(hit_rate * 100.0);
}
// ============================================================================
// S3 Operation Metrics (GetObject, PutObject, etc.)
// ============================================================================
/// Record GetObject operation metrics.
///
/// # Arguments
///
/// * `duration_ms` - Operation duration in milliseconds
/// * `size_bytes` - Object size in bytes
///
/// Note: this function records aggregate S3 GET metrics only. It must not be
/// interpreted as the definitive source of truth for data-plane copy mode.
#[inline(always)]
pub fn record_get_object(duration_ms: f64, size_bytes: i64) {
counter!("rustfs.s3.get_object.total").increment(1);
histogram!("rustfs.s3.get_object.duration.ms").record(duration_ms);
if size_bytes > 0 {
histogram!("rustfs.s3.get_object.size.bytes").record(size_bytes as f64);
}
}
/// Record PutObject operation metrics.
///
/// # Arguments
///
/// * `duration_ms` - Operation duration in milliseconds
/// * `size_bytes` - Object size in bytes
/// * `zero_copy_enabled` - Legacy aggregate flag preserved for compatibility
///
/// Note: this function records aggregate S3 PUT metrics only. The definitive
/// outcome of request-level fast-path attempts must be tracked separately via
/// ADR 0001 data-plane helpers.
#[inline(always)]
pub fn record_put_object(duration_ms: f64, size_bytes: i64, zero_copy_enabled: bool) {
counter!("rustfs.s3.put_object.total").increment(1);
histogram!("rustfs.s3.put_object.duration.ms").record(duration_ms);
counter!(
"rustfs.s3.put_object.bucketed.total",
"size_bucket" => put_size_bucket_label(size_bytes)
)
.increment(1);
histogram!(
"rustfs.s3.put_object.bucketed.duration.ms",
"size_bucket" => put_size_bucket_label(size_bytes)
)
.record(duration_ms);
if size_bytes > 0 {
histogram!("rustfs.s3.put_object.size.bytes").record(size_bytes as f64);
histogram!(
"rustfs.s3.put_object.bucketed.size.bytes",
"size_bucket" => put_size_bucket_label(size_bytes)
)
.record(size_bytes as f64);
}
if zero_copy_enabled {
counter!("rustfs.s3.put_object.zero_copy.enabled.total").increment(1);
}
}
/// Record ListObjects operation metrics.
///
/// # Arguments
///
/// * `duration_ms` - Operation duration in milliseconds
/// * `objects_count` - Number of objects returned
/// * `is_truncated` - Whether the response was truncated
#[inline(always)]
pub fn record_list_objects(duration_ms: f64, objects_count: u64, is_truncated: bool) {
counter!("rustfs.s3.list_objects.total").increment(1);
histogram!("rustfs.s3.list_objects.duration.ms").record(duration_ms);
histogram!("rustfs.s3.list_objects.count").record(objects_count as f64);
if is_truncated {
counter!("rustfs.s3.list_objects.truncated.total").increment(1);
}
}
/// Record DeleteObject operation metrics.
///
/// # Arguments
///
/// * `duration_ms` - Operation duration in milliseconds
/// * `version_deleted` - Whether a specific version was deleted
#[inline(always)]
pub fn record_delete_object(duration_ms: f64, version_deleted: bool) {
counter!("rustfs.s3.delete_object.total").increment(1);
histogram!("rustfs.s3.delete_object.duration.ms").record(duration_ms);
if version_deleted {
counter!("rustfs.s3.delete_object.version.total").increment(1);
}
}
// ============================================================================
// I/O Scheduler Metrics
// ============================================================================
/// Record I/O scheduler strategy selection.
///
/// # Arguments
///
/// * `storage_media` - Detected storage media type ("nvme", "ssd", "hdd", "unknown")
/// * `access_pattern` - Detected access pattern ("sequential", "random", "mixed", "unknown")
/// * `buffer_size` - Selected buffer size in bytes
/// * `concurrent_requests` - Number of concurrent requests
#[inline(always)]
pub fn record_io_strategy(storage_media: &str, access_pattern: &str, buffer_size: usize, concurrent_requests: u64) {
counter!("rustfs.io.strategy.total",
"storage_media" => storage_media.to_string(),
"access_pattern" => access_pattern.to_string(),
)
.increment(1);
gauge!("rustfs.io.buffer.size.bytes",
"storage_media" => storage_media.to_string(),
)
.set(buffer_size as f64);
gauge!("rustfs.io.concurrent.requests").set(concurrent_requests as f64);
}
/// Record disk permit wait time (load tracking).
///
/// # Arguments
///
/// * `duration_ms` - Time spent waiting for disk permit
#[inline(always)]
pub fn record_permit_wait(duration_ms: f64) {
histogram!("rustfs.io.permit.wait.duration.ms").record(duration_ms);
}
/// Record I/O load level.
///
/// # Arguments
///
/// * `load_level` - Current load level ("low", "medium", "high", "critical")
/// * `concurrent_requests` - Number of concurrent requests
#[inline(always)]
pub fn record_io_load_level(load_level: &str, concurrent_requests: u64) {
counter!("rustfs.io.load.level",
"level" => load_level.to_string(),
)
.increment(1);
gauge!("rustfs.io.concurrent.requests").set(concurrent_requests as f64);
}
/// Record cache size and entry count.
///
/// # Arguments
///
/// * `tier` - Cache tier ("l1", "l2")
/// * `size_bytes` - Total cache size in bytes
/// * `entries` - Number of entries in the cache
#[inline(always)]
pub fn record_cache_size(tier: &str, size_bytes: usize, entries: u64) {
gauge!("rustfs.cache.size.bytes",
"tier" => tier.to_string(),
)
.set(size_bytes as f64);
gauge!("rustfs.cache.entries",
"tier" => tier.to_string(),
)
.set(entries as f64);
}
// ============================================================================
// Bandwidth Monitoring Metrics
// ============================================================================
/// Record bandwidth observation.
///
/// # Arguments
///
/// * `bytes_per_second` - Observed bandwidth in bytes per second
/// * `tier` - Bandwidth tier ("low", "medium", "high", "unknown")
#[inline(always)]
pub fn record_bandwidth(bytes_per_second: u64, tier: &str) {
gauge!("rustfs.bandwidth.current.bps").set(bytes_per_second as f64);
gauge!("rustfs.bandwidth.current.bps",
"tier" => tier.to_string(),
)
.set(bytes_per_second as f64);
histogram!("rustfs.bandwidth.observed.bps").record(bytes_per_second as f64);
}
/// Record data transfer for bandwidth calculation.
///
/// # Arguments
///
/// * `bytes` - Number of bytes transferred
/// * `duration_ms` - Duration of the transfer in milliseconds
#[inline(always)]
pub fn record_data_transfer(bytes: u64, duration_ms: f64) {
counter!("rustfs.io.transfer.bytes").increment(bytes);
histogram!("rustfs.io.transfer.duration.ms").record(duration_ms);
if duration_ms > 0.0 {
let bps = (bytes as f64 * 1000.0) / duration_ms;
histogram!("rustfs.io.transfer.bandwidth.bps").record(bps);
}
}
// ============================================================================
// System Resource Metrics
// ============================================================================
/// Record memory usage.
///
/// # Arguments
///
/// * `used_bytes` - Used memory in bytes
/// * `total_bytes` - Total memory in bytes
#[inline(always)]
pub fn record_memory_usage(used_bytes: u64, total_bytes: u64) {
gauge!("rustfs.memory.used.bytes").set(used_bytes as f64);
gauge!("rustfs.memory.total.bytes").set(total_bytes as f64);
if total_bytes > 0 {
let usage_percent = (used_bytes as f64 / total_bytes as f64) * 100.0;
gauge!("rustfs.memory.usage.percent").set(usage_percent);
}
}
/// Record CPU usage.
///
/// # Arguments
///
/// * `percent` - CPU usage percentage (0.0 - 100.0)
#[inline(always)]
pub fn record_cpu_usage(percent: f64) {
gauge!("rustfs.cpu.usage.percent").set(percent);
}
/// Record disk I/O statistics.
///
/// # Arguments
///
/// * `read_bytes` - Bytes read
/// * `write_bytes` - Bytes written
/// * `read_ops` - Number of read operations
/// * `write_ops` - Number of write operations
#[inline(always)]
pub fn record_disk_io(read_bytes: u64, write_bytes: u64, read_ops: u64, write_ops: u64) {
counter!("rustfs.disk.read.bytes").increment(read_bytes);
counter!("rustfs.disk.write.bytes").increment(write_bytes);
counter!("rustfs.disk.read.ops").increment(read_ops);
counter!("rustfs.disk.write.ops").increment(write_ops);
gauge!("rustfs.disk.read.bytes_total").set(read_bytes as f64);
gauge!("rustfs.disk.write.bytes_total").set(write_bytes as f64);
}
// ============================================================================
// Error and Timeout Metrics
// ============================================================================
/// Record operation error.
///
/// # Arguments
///
/// * `operation` - Operation type (e.g., "get_object", "put_object")
/// * `error_type` - Error type (e.g., "timeout", "disk_error", "network")
#[inline(always)]
pub fn record_error(operation: &str, error_type: &str) {
counter!("rustfs.errors.total",
"operation" => operation.to_string(),
"type" => error_type.to_string(),
)
.increment(1);
}
/// Record operation timeout.
///
/// # Arguments
///
/// * `operation` - Operation type that timed out
/// * `duration_ms` - Duration before timeout
#[inline(always)]
pub fn record_timeout(operation: &str, duration_ms: f64) {
counter!("rustfs.timeouts.total",
"operation" => operation.to_string(),
)
.increment(1);
histogram!("rustfs.timeouts.duration.ms",
"operation" => operation.to_string(),
)
.record(duration_ms);
}
/// Record retry attempt.
///
/// # Arguments
///
/// * `operation` - Operation being retried
/// * `attempt_number` - Attempt number (1-based)
#[inline(always)]
pub fn record_retry(operation: &str, attempt_number: u32) {
counter!("rustfs.retries.total",
"operation" => operation.to_string(),
)
.increment(1);
histogram!("rustfs.retries.attempt",
"operation" => operation.to_string(),
)
.record(attempt_number as f64);
}
// ============================================================================
// Helper Metrics (for MetricsCollector)
// ============================================================================
/// Record I/O latency in milliseconds.
///
/// # Arguments
///
/// * `latency_ms` - I/O latency in milliseconds
#[inline(always)]
pub fn record_io_latency(latency_ms: f64) {
histogram!("rustfs.io.latency.ms").record(latency_ms);
}
/// Record I/O latency P95 in milliseconds.
///
/// # Arguments
///
/// * `latency_ms` - P95 I/O latency in milliseconds
#[inline(always)]
pub fn record_io_latency_p95(latency_ms: f64) {
gauge!("rustfs.io.latency.p95.ms").set(latency_ms);
}
/// Record I/O latency P99 in milliseconds.
///
/// # Arguments
///
/// * `latency_ms` - P99 I/O latency in milliseconds
#[inline(always)]
pub fn record_io_latency_p99(latency_ms: f64) {
gauge!("rustfs.io.latency.p99.ms").set(latency_ms);
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_io_path_as_str_values_stable() {
assert_eq!(IoPath::Fast.as_str(), "fast");
assert_eq!(IoPath::Legacy.as_str(), "legacy");
}
#[test]
fn test_copy_mode_as_str_values_stable() {
assert_eq!(CopyMode::TrueZeroCopy.as_str(), "true_zero_copy");
assert_eq!(CopyMode::SharedBytes.as_str(), "shared_bytes");
assert_eq!(CopyMode::SingleCopy.as_str(), "single_copy");
assert_eq!(CopyMode::Reconstructed.as_str(), "reconstructed");
assert_eq!(CopyMode::Transformed.as_str(), "transformed");
}
#[test]
fn test_fallback_reason_as_str_values_stable() {
assert_eq!(FallbackReason::Unknown.as_str(), "unknown");
assert_eq!(FallbackReason::MmapDisabled.as_str(), "mmap_disabled");
assert_eq!(FallbackReason::MmapUnavailable.as_str(), "mmap_unavailable");
assert_eq!(FallbackReason::SmallObject.as_str(), "small_object");
assert_eq!(FallbackReason::WindowLimitExceeded.as_str(), "window_limit_exceeded");
assert_eq!(FallbackReason::UnalignedWindow.as_str(), "unaligned_window");
assert_eq!(FallbackReason::RangeNotSupported.as_str(), "range_not_supported");
assert_eq!(FallbackReason::EncryptionEnabled.as_str(), "encryption_enabled");
assert_eq!(FallbackReason::CompressionEnabled.as_str(), "compression_enabled");
assert_eq!(FallbackReason::TransformEncryptionLegacy.as_str(), "transform_encryption_legacy");
assert_eq!(FallbackReason::TransformCompressionLegacy.as_str(), "transform_compression_legacy");
assert_eq!(
FallbackReason::TransformCompressionEncryptionLegacy.as_str(),
"transform_compression_encryption_legacy"
);
assert_eq!(FallbackReason::ChunkBridgeUnavailable.as_str(), "chunk_bridge_unavailable");
assert_eq!(FallbackReason::NonLocalBackend.as_str(), "non_local_backend");
}
#[test]
fn test_record_io_path_selected() {
record_io_path_selected("get", IoPath::Fast);
record_io_path_selected("put", IoPath::Legacy);
}
#[test]
fn test_record_io_copy_mode() {
record_io_copy_mode("get", CopyMode::SharedBytes, 1024);
record_io_copy_mode("put", CopyMode::Transformed, 2048);
}
#[test]
fn test_record_io_fallback() {
record_io_fallback(IoStage::ReadSetup, FallbackReason::MmapUnavailable);
record_io_fallback(IoStage::HttpBridge, FallbackReason::ChunkBridgeUnavailable);
}
#[test]
fn test_record_local_disk_active_mmap_bytes() {
record_local_disk_active_mmap_bytes(4096);
record_local_disk_active_mmap_bytes(0);
}
#[test]
fn test_record_local_disk_pooled_chunk() {
record_local_disk_pooled_chunk("fallback", 4096);
record_local_disk_pooled_chunk("compat_collect", 8192);
}
#[test]
fn test_record_local_disk_compat_collect() {
record_local_disk_compat_collect(3, 16384);
}
#[test]
fn test_record_put_object_attempted_fast_path() {
record_put_object_attempted_fast_path(1024 * 1024);
record_put_object_attempted_fast_path(0);
}
#[test]
fn test_record_put_transform_selected() {
record_put_transform_selected("compression", IoPath::Fast, 2048);
record_put_transform_selected("compression_encryption", IoPath::Legacy, 4096);
}
#[test]
fn test_record_put_path_selected() {
record_put_path_selected(8 * 1024, IoPath::Fast);
record_put_path_selected(2 * 1024 * 1024, IoPath::Legacy);
}
#[test]
fn test_record_put_copy_mode() {
record_put_copy_mode(8 * 1024, CopyMode::SingleCopy);
record_put_copy_mode(512 * 1024, CopyMode::Transformed);
}
#[test]
fn test_record_put_fallback() {
record_put_fallback(32 * 1024, FallbackReason::CompressionEnabled);
record_put_fallback(2 * 1024 * 1024, FallbackReason::EncryptionEnabled);
}
#[test]
fn test_record_put_inline_selected() {
record_put_inline_selected(8 * 1024, false);
record_put_inline_selected(32 * 1024, true);
}
#[test]
fn test_record_bytes_pool_metrics() {
record_bytes_pool_acquire("small", 4096, true);
record_bytes_pool_return("small");
record_bytes_pool_allocated("small", 4096);
record_bytes_pool_hit_rate("small", 0.85);
}
// S3 Operation Metrics Tests
#[test]
fn test_record_get_object() {
record_get_object(100.0, 1024 * 1024);
record_get_object(50.0, 2048);
}
#[test]
fn test_record_put_object() {
record_put_object(200.0, 1024 * 1024, true);
record_put_object(100.0, 512, false);
}
#[test]
fn test_record_list_objects() {
record_list_objects(50.0, 100, false);
record_list_objects(75.0, 1000, true);
}
#[test]
fn test_record_delete_object() {
record_delete_object(25.0, false);
record_delete_object(30.0, true);
}
// I/O Scheduler Metrics Tests
#[test]
fn test_record_io_strategy() {
record_io_strategy("nvme", "sequential", 256 * 1024, 5);
record_io_strategy("ssd", "random", 64 * 1024, 10);
}
#[test]
fn test_record_permit_wait() {
record_permit_wait(5.0);
record_permit_wait(10.5);
}
#[test]
fn test_record_io_load_level() {
record_io_load_level("low", 2);
record_io_load_level("medium", 5);
record_io_load_level("high", 15);
}
#[test]
fn test_record_cache_size() {
record_cache_size("l1", 50 * 1024 * 1024, 1000);
record_cache_size("l2", 200 * 1024 * 1024, 5000);
}
// Bandwidth Metrics Tests
#[test]
fn test_record_bandwidth() {
record_bandwidth(100 * 1024 * 1024, "high");
record_bandwidth(50 * 1024 * 1024, "medium");
}
#[test]
fn test_record_data_transfer() {
record_data_transfer(1024 * 1024, 100.0);
record_data_transfer(2048, 50.0);
}
// System Resource Metrics Tests
#[test]
fn test_record_memory_usage() {
record_memory_usage(1024 * 1024 * 1024, 4 * 1024 * 1024 * 1024);
record_memory_usage(2 * 1024 * 1024 * 1024, 8 * 1024 * 1024 * 1024);
}
#[test]
fn test_record_cpu_usage() {
record_cpu_usage(25.5);
record_cpu_usage(50.0);
record_cpu_usage(75.5);
}
#[test]
fn test_record_disk_io() {
record_disk_io(1024 * 1024, 2048, 100, 50);
record_disk_io(2048, 4096, 200, 100);
}
// Error and Timeout Metrics Tests
#[test]
fn test_record_error() {
record_error("get_object", "timeout");
record_error("put_object", "disk_error");
}
#[test]
fn test_record_timeout() {
record_timeout("get_object", 5000.0);
record_timeout("list_objects", 10000.0);
}
#[test]
fn test_record_retry() {
record_retry("get_object", 1);
record_retry("put_object", 2);
}
}
pub mod bandwidth;
pub mod global_metrics;
pub mod metric_names;