fxcp_core/

metrics.rs

// SPDX-License-Identifier: GPL-2.0-or-later
// Copyright (C) 2025 Joel Wirāmu Pauling <aenertia@aenertia.net>
//
// fxcp-core/src/metrics.rs  --  Prometheus metrics definitions and initialization

//! Prometheus metric counters and gauges for copy operations, I/O stats,
//! and storage detection results.

#![allow(clippy::unwrap_used, clippy::expect_used)]

use lazy_static::lazy_static;
use prometheus::{
    register_counter, register_counter_vec, register_gauge, register_gauge_vec, register_histogram_vec,
    register_histogram, Counter, CounterVec, Gauge, GaugeVec, HistogramVec, Histogram, Registry,
};
use std::sync::atomic::{AtomicU64, Ordering};

lazy_static! {
    /// Shared Prometheus registry for all foxing metrics.
    pub static ref REGISTRY: Registry = Registry::new();

    /// End-to-end replication latency from source event to target write, per target.
    pub static ref REPLICATION_LATENCY: HistogramVec = register_histogram_vec!(
        "foxing_replication_latency_seconds",
        "End-to-end latency from source event to target write",
        &["target"]
    ).unwrap();
    /// Time spent resolving an inode number to a filesystem path.
    pub static ref INODE_LOOKUP_DURATION: Histogram = register_histogram!(
        "foxing_inode_lookup_duration_seconds",
        "Time spent resolving inode to path",
        vec![0.001, 0.01, 0.1, 1.0]
    ).unwrap();

    /// Total bytes successfully written to each replication target.
    pub static ref BYTES_REPLICATED: CounterVec = register_counter_vec!(
        "foxing_bytes_replicated_total",
        "Bytes successfully written to target",
        &["target"]
    ).unwrap();
    /// Total rename operations processed across all targets.
    pub static ref RENAME_EVENTS: Counter = register_counter!(
        "foxing_rename_events_total",
        "Total rename operations processed"
    ).unwrap();
    /// Write events merged into larger chunks by the coalescer.
    pub static ref COALESCED_WRITES: Counter = register_counter!(
        "foxing_coalesced_writes_total",
        "Number of write events merged into larger chunks"
    ).unwrap();

    /// Copies completed via CoW reflink (FICLONE), per target.
    pub static ref COPY_METHOD_REFLINK: CounterVec = register_counter_vec!(
        "foxing_copy_method_reflink_total",
        "Writes handled via CoW Reflink",
        &["target"]
    ).unwrap();
    /// Copies completed via hardware or network offload (copy_file_range/sendfile), per target.
    pub static ref COPY_METHOD_OFFLOAD: CounterVec = register_counter_vec!(
        "foxing_copy_method_offload_total",
        "Writes handled via hardware/network offload",
        &["target"]
    ).unwrap();
    /// Copies completed via standard read/write or io_uring, per target.
    pub static ref COPY_METHOD_STANDARD: CounterVec = register_counter_vec!(
        "foxing_copy_method_standard_total",
        "Writes handled via standard read/write",
        &["target"]
    ).unwrap();
    /// Atomic write attempts (tmp+rename) that fell back to buffered I/O.
    pub static ref ATOMIC_WRITE_FALLBACKS: Counter = register_counter!(
        "foxing_atomic_write_fallbacks_total",
        "Number of atomic writes that failed and fell back to buffered/standard I/O"
    ).unwrap();
    /// Copies that fell back from io_uring to sendfile due to EBADF or timeout.
    pub static ref COPY_METHOD_SENDFILE_FALLBACK: CounterVec = register_counter_vec!(
        "foxing_copy_method_sendfile_fallback_total",
        "Copies that fell back from io_uring to sendfile",
        &["target"]
    ).unwrap();
}

lazy_static! {
    /// Failed attempts to create reflink version snapshots before overwrite.
    pub static ref VERSIONING_FAILURES: Counter = register_counter!(
        "foxing_versioning_failures_total",
        "Failed attempts to create version snapshots"
    ).unwrap();
    /// Successfully created reflink version snapshots.
    pub static ref VERSIONING_SUCCESS: Counter = register_counter!(
        "foxing_versioning_success_total",
        "Successfully created version snapshots"
    ).unwrap();

    /// Whether the system is currently under stress (1 = stressed, 0 = normal).
    pub static ref GOVERNOR_STRESSED: Gauge = register_gauge!(
        "foxing_governor_stressed",
        "Current system stress state (1=stressed, 0=normal)"
    ).unwrap();
    /// Composite stress score from PSI, memory, and CPU pressure (0.0..1.0 = OK, >1.0 = critical).
    pub static ref GOVERNOR_STRESS_SCORE: Gauge = register_gauge!(
        "foxing_governor_stress_score",
        "Current system stress score (0.0-1.0=OK, >1.0=Critical)"
    ).unwrap();
    /// Events delayed because the governor detected system pressure.
    pub static ref GOVERNOR_THROTTLED_EVENTS: Counter = register_counter!(
        "foxing_governor_throttled_events_total",
        "Events delayed due to governor pressure"
    ).unwrap();
    /// System load averages (1-minute, 5-minute, 15-minute), labeled by period.
    pub static ref GOVERNOR_LOAD_AVERAGE: GaugeVec = register_gauge_vec!(
        "foxing_governor_load_average",
        "System load average (1m, 5m, 15m)",
        &["period"]
    ).unwrap();
    /// Cumulative time workers spent sleeping due to governor pacing (ms).
    pub static ref GOVERNOR_PACING_DURATION_MS: Counter = register_counter!(
        "foxing_governor_pacing_duration_milliseconds_total",
        "Total time spent sleeping due to governor pacing"
    ).unwrap();

    /// Number of device-mapper layers detected beneath the target filesystem.
    pub static ref DM_STACK_DEPTH: Gauge = register_gauge!(
        "foxing_dm_stack_depth",
        "Number of device-mapper layers beneath the filesystem"
    ).unwrap();
    /// Set to 1 when dm-crypt (LUKS) is detected in the storage stack.
    pub static ref DM_CRYPT_DETECTED: Gauge = register_gauge!(
        "foxing_dm_crypt_detected",
        "1 if dm-crypt (LUKS) detected in storage stack"
    ).unwrap();
    /// Physical block size of the underlying storage device (bytes).
    pub static ref STORAGE_PHYSICAL_BLOCK_SIZE: Gauge = register_gauge!(
        "foxing_storage_physical_block_size",
        "Physical block size of base storage device (bytes)"
    ).unwrap();

    /// Maximum configured memory budget for I/O buffers across all workers (bytes).
    pub static ref GLOBAL_BUFFER_LIMIT: Gauge = register_gauge!(
        "foxing_global_buffer_limit_bytes",
        "Maximum configured memory for buffers (Bytes)"
    ).unwrap();
    /// Atomic counter tracking the total number of allocated buffers globally.
    pub static ref GLOBAL_BUFFER_COUNT: AtomicU64 = AtomicU64::new(0);
    /// Current total memory allocated for I/O buffers across all workers (bytes).
    pub static ref GLOBAL_MEMORY_USAGE_BYTES: Gauge = register_gauge!(
        "foxing_global_memory_usage_bytes",
        "Current total memory allocated for IO buffers across all workers"
    ).unwrap();

    /// Number of buffers currently in the worker's buffer pool.
    pub static ref BUFFER_POOL_CAPACITY: Gauge = register_gauge!(
        "foxing_buffer_pool_capacity_buffers",
        "Number of buffers in the worker's buffer pool"
    ).unwrap();
    /// Size of each individual buffer chunk in the pool (bytes).
    pub static ref BUFFER_POOL_CHUNK_SIZE: Gauge = register_gauge!(
        "foxing_buffer_pool_chunk_size_bytes",
        "Size of each buffer chunk in bytes"
    ).unwrap();
    /// Total memory allocated to the buffer pool (bytes).
    pub static ref BUFFER_POOL_TOTAL_BYTES: Gauge = register_gauge!(
        "foxing_buffer_pool_total_bytes",
        "Total memory allocated to buffer pool"
    ).unwrap();
}

lazy_static! {
    /// `.foxing_meta` sidecar files created (fallback for filesystems without xattr support).
    pub static ref SIDECAR_FILES_CREATED: Counter = register_counter!(
        "foxing_sidecar_files_created_total",
        "Number of .foxing_meta files created (fallback metadata)"
    ).unwrap();

    /// BLAKE3 hash verifications performed (post-copy integrity checks).
    pub static ref HASH_VERIFICATIONS_TOTAL: Counter = register_counter!(
        "foxing_hash_verifications_total",
        "Number of BLAKE3 hash verifications performed"
    ).unwrap();
    /// Files skipped because their cached signature matched the source.
    pub static ref HASH_CACHE_HITS: Counter = register_counter!(
        "foxing_hash_cache_hits_total",
        "Files skipped due to matching signature cache"
    ).unwrap();
    /// NFS bypass fallbacks due to Kerberos auth (krb5/krb5i/krb5p) on the mount.
    pub static ref NFS_BYPASS_KRB5_FALLBACK: Counter = register_counter!(
        "foxing_nfs_bypass_krb5_fallback_total",
        "NFS bypass unavailable due to Kerberos auth  --  fell back to kernel VFS path"
    ).unwrap();
    /// NFS bypass TLS upgrade skipped  --  xprtsec=tls mount but tls feature disabled or STARTTLS rejected.
    pub static ref NFS_BYPASS_TLS_FALLBACK: Counter = register_counter!(
        "foxing_nfs_bypass_tls_fallback_total",
        "NFS bypass TLS upgrade skipped  --  xprtsec=tls mount but tls feature disabled or STARTTLS rejected"
    ).unwrap();
    /// Double encryption detected: sec=krb5p (per-RPC AES) + xprtsec=tls.
    pub static ref NFS_BYPASS_DOUBLE_ENCRYPTION: Counter = register_counter!(
        "foxing_nfs_bypass_double_encryption_total",
        "sec=krb5p + xprtsec=tls detected  --  redundant double encryption"
    ).unwrap();
    /// kTLS kernel offload successfully promoted after TLS handshake.
    pub static ref NFS_BYPASS_KTLS_PROMOTED: Counter = register_counter!(
        "foxing_nfs_bypass_ktls_promoted_total",
        "kTLS kernel offload promoted after userspace TLS handshake"
    ).unwrap();

    /// Time spent computing BLAKE3 hashes (seconds).
    pub static ref HASH_COMPUTATION_DURATION: Histogram = register_histogram!(
        "foxing_hash_computation_seconds",
        "Time spent computing BLAKE3 hashes"
    ).unwrap();
}

/// Set the global buffer memory limit gauge from a megabyte value.
pub fn initialize_metrics(global_limit_mb: u64) {
    GLOBAL_BUFFER_LIMIT.set((global_limit_mb * 1024 * 1024) as f64);
}

/// Batches increments to a Prometheus [`Counter`] to reduce atomic contention.
///
/// Accumulates increments locally and flushes to the underlying counter
/// when the batch threshold is reached or on drop.
pub struct BatchedCounter {
    counter: Counter,
    local_count: u64,
    batch_size: u64,
}

impl BatchedCounter {
    pub fn new(counter: Counter, batch_size: u64) -> Self {
        Self {
            counter,
            local_count: 0,
            batch_size,
        }
    }

    #[inline]
    pub fn inc(&mut self) {
        self.local_count += 1;
        if self.local_count >= self.batch_size {
            self.flush();
        }
    }

    #[inline]
    pub fn inc_by(&mut self, val: u64) {
        self.local_count += val;
        if self.local_count >= self.batch_size {
            self.flush();
        }
    }

    #[inline]
    pub fn flush(&mut self) {
        if self.local_count > 0 {
            self.counter.inc_by(self.local_count as f64);
            self.local_count = 0;
        }
    }
}

impl Drop for BatchedCounter {
    fn drop(&mut self) {
        self.flush();
    }
}

/// Batches increments and decrements to a global [`AtomicU64`] to reduce contention.
///
/// Flushes the local delta when the batch threshold is reached or on drop.
pub struct BatchedAtomicCounter {
    global: &'static AtomicU64,
    local_count: i64,
    batch_size: i64,
}

impl BatchedAtomicCounter {
    pub fn new(global: &'static AtomicU64, batch_size: i64) -> Self {
        Self {
            global,
            local_count: 0,
            batch_size,
        }
    }

    #[inline]
    pub fn inc(&mut self) {
        self.local_count += 1;
        if self.local_count >= self.batch_size {
            self.flush();
        }
    }

    #[inline]
    pub fn dec(&mut self) {
        self.local_count -= 1;
        if self.local_count <= -self.batch_size {
            self.flush();
        }
    }

    #[inline]
    pub fn flush(&mut self) {
        if self.local_count != 0 {
            if self.local_count > 0 {
                self.global.fetch_add(self.local_count as u64, Ordering::Relaxed);
            } else {
                self.global.fetch_sub((-self.local_count) as u64, Ordering::Relaxed);
            }
            self.local_count = 0;
        }
    }
}

impl Drop for BatchedAtomicCounter {
    fn drop(&mut self) {
        self.flush();
    }
}

lazy_static! {
    /// RDMA per-RPC round-trip latency in microseconds (Send WR post -> CQ completion).
    pub static ref RDMA_RPC_LATENCY_US: Histogram = register_histogram!(
        "foxing_rdma_rpc_latency_us",
        "RDMA per-RPC round-trip latency in microseconds",
        vec![1.0, 2.0, 5.0, 10.0, 20.0, 50.0, 100.0, 500.0, 1000.0]
    ).expect("failed to register foxing_rdma_rpc_latency_us");
}