grafos_observe/

metrics.rs

//! Core metrics types for fabric observability.
//!
//! All types use atomics and work in `no_std` environments. The
//! [`FabricMetrics`] singleton collects system-wide counters and gauges.

use core::sync::atomic::{AtomicI64, AtomicU64, Ordering};

/// Monotonically increasing counter (atomic u64).
///
/// Counters only go up — use them for totals like "operations completed"
/// or "bytes transferred". Thread-safe via relaxed atomic operations.
///
/// # Examples
///
/// ```
/// use grafos_observe::MetricCounter;
///
/// let c = MetricCounter::new();
/// c.inc();
/// c.add(10);
/// assert_eq!(c.get(), 11);
///
/// let prev = c.reset();
/// assert_eq!(prev, 11);
/// assert_eq!(c.get(), 0);
/// ```
pub struct MetricCounter {
    value: AtomicU64,
}

impl Default for MetricCounter {
    fn default() -> Self {
        Self::new()
    }
}

impl MetricCounter {
    /// Create a new counter starting at zero.
    pub const fn new() -> Self {
        Self {
            value: AtomicU64::new(0),
        }
    }

    /// Increment the counter by one.
    pub fn inc(&self) {
        self.value.fetch_add(1, Ordering::Relaxed);
    }

    /// Increment the counter by `n`.
    pub fn add(&self, n: u64) {
        self.value.fetch_add(n, Ordering::Relaxed);
    }

    /// Read the current counter value.
    pub fn get(&self) -> u64 {
        self.value.load(Ordering::Relaxed)
    }

    /// Reset the counter to zero. Returns the previous value.
    pub fn reset(&self) -> u64 {
        self.value.swap(0, Ordering::Relaxed)
    }
}

/// Current-value gauge (atomic i64).
///
/// Gauges go up and down — use them for values like "active leases"
/// or "connections open".
///
/// # Examples
///
/// ```
/// use grafos_observe::MetricGauge;
///
/// let g = MetricGauge::new();
/// g.inc();
/// g.inc();
/// g.dec();
/// assert_eq!(g.get(), 1);
///
/// g.set(-5);
/// assert_eq!(g.get(), -5);
/// ```
pub struct MetricGauge {
    value: AtomicI64,
}

impl Default for MetricGauge {
    fn default() -> Self {
        Self::new()
    }
}

impl MetricGauge {
    /// Create a new gauge starting at zero.
    pub const fn new() -> Self {
        Self {
            value: AtomicI64::new(0),
        }
    }

    /// Increment the gauge by one.
    pub fn inc(&self) {
        self.value.fetch_add(1, Ordering::Relaxed);
    }

    /// Decrement the gauge by one.
    pub fn dec(&self) {
        self.value.fetch_sub(1, Ordering::Relaxed);
    }

    /// Set the gauge to an absolute value.
    pub fn set(&self, val: i64) {
        self.value.store(val, Ordering::Relaxed);
    }

    /// Read the current gauge value.
    pub fn get(&self) -> i64 {
        self.value.load(Ordering::Relaxed)
    }
}

/// Fixed-bucket latency histogram.
///
/// Tracks the distribution of values (typically durations in microseconds)
/// across predefined buckets. No allocator needed — the bucket array is
/// inline. Uses 10 buckets: 100us, 500us, 1ms, 5ms, 10ms, 50ms, 100ms,
/// 500ms, 1s, +Inf.
///
/// Buckets are cumulative: each bucket count includes all observations that
/// also fall into lower buckets. This matches the Prometheus histogram
/// convention.
///
/// # Examples
///
/// ```
/// use grafos_observe::MetricHistogram;
///
/// let h = MetricHistogram::new();
/// h.observe(50);     // 50us — lands in the <=100us bucket
/// h.observe(2_000);  // 2ms  — lands in the <=5ms bucket
///
/// assert_eq!(h.count(), 2);
/// assert_eq!(h.sum(), 2050);
/// assert_eq!(h.bucket_count(0), 1); // <=100us: only the 50us observation
/// assert_eq!(h.bucket_count(3), 2); // <=5ms: both observations
/// ```
pub struct MetricHistogram {
    /// Bucket upper bounds in microseconds. The last bucket is +Inf (u64::MAX).
    bounds: [u64; Self::NUM_BUCKETS],
    /// Counts per bucket (cumulative — each bucket includes all lower buckets).
    counts: [AtomicU64; Self::NUM_BUCKETS],
    /// Sum of all observed values.
    sum: AtomicU64,
    /// Total number of observations.
    total: AtomicU64,
}

impl Default for MetricHistogram {
    fn default() -> Self {
        Self::new()
    }
}

impl MetricHistogram {
    /// Number of histogram buckets.
    pub const NUM_BUCKETS: usize = 10;

    /// Bucket boundaries in microseconds.
    pub const BUCKET_BOUNDS: [u64; Self::NUM_BUCKETS] = [
        100,       // 100us
        500,       // 500us
        1_000,     // 1ms
        5_000,     // 5ms
        10_000,    // 10ms
        50_000,    // 50ms
        100_000,   // 100ms
        500_000,   // 500ms
        1_000_000, // 1s
        u64::MAX,  // +Inf
    ];

    /// Create a new histogram with the default bucket boundaries.
    #[allow(clippy::declare_interior_mutable_const)]
    pub const fn new() -> Self {
        Self {
            bounds: Self::BUCKET_BOUNDS,
            counts: [
                AtomicU64::new(0),
                AtomicU64::new(0),
                AtomicU64::new(0),
                AtomicU64::new(0),
                AtomicU64::new(0),
                AtomicU64::new(0),
                AtomicU64::new(0),
                AtomicU64::new(0),
                AtomicU64::new(0),
                AtomicU64::new(0),
            ],
            sum: AtomicU64::new(0),
            total: AtomicU64::new(0),
        }
    }

    /// Record an observation (value in microseconds).
    ///
    /// Increments the count for every bucket whose bound is >= the value
    /// (cumulative histogram, matching Prometheus convention).
    pub fn observe(&self, value_us: u64) {
        self.sum.fetch_add(value_us, Ordering::Relaxed);
        self.total.fetch_add(1, Ordering::Relaxed);
        for (i, &bound) in self.bounds.iter().enumerate() {
            if value_us <= bound {
                self.counts[i].fetch_add(1, Ordering::Relaxed);
            }
        }
    }

    /// Read the cumulative count for a specific bucket index.
    pub fn bucket_count(&self, index: usize) -> u64 {
        if index < Self::NUM_BUCKETS {
            self.counts[index].load(Ordering::Relaxed)
        } else {
            0
        }
    }

    /// Read the bucket upper bound for a specific bucket index (in microseconds).
    pub fn bucket_bound(&self, index: usize) -> u64 {
        if index < Self::NUM_BUCKETS {
            self.bounds[index]
        } else {
            u64::MAX
        }
    }

    /// Total number of observations.
    pub fn count(&self) -> u64 {
        self.total.load(Ordering::Relaxed)
    }

    /// Sum of all observed values (in microseconds).
    pub fn sum(&self) -> u64 {
        self.sum.load(Ordering::Relaxed)
    }
}

/// Global fabric metrics registry.
///
/// Tracks system-wide counters and gauges for lease lifecycles,
/// data-plane operations, and graph rewrites. Access the process-wide
/// singleton via [`FabricMetrics::global()`].
///
/// # Examples
///
/// ```
/// use grafos_observe::FabricMetrics;
///
/// let m = FabricMetrics::global();
/// m.leases_total.inc();
/// m.leases_active.inc();
/// m.ops_total.add(5);
/// m.bytes_read.add(1024);
/// m.op_latency.observe(300);
/// ```
pub struct FabricMetrics {
    /// Currently active leases (gauge — goes up on acquire, down on drop/expire).
    pub leases_active: MetricGauge,
    /// Total leases ever created (counter).
    pub leases_total: MetricCounter,
    /// Total leases that expired (counter).
    pub leases_expired: MetricCounter,
    /// Total leases that entered fenced state (counter).
    pub leases_fenced: MetricCounter,
    /// Total data-plane operations completed (counter).
    pub ops_total: MetricCounter,
    /// Total data-plane operations that failed (counter).
    pub ops_errors: MetricCounter,
    /// Total bytes read across all data-plane operations (counter).
    pub bytes_read: MetricCounter,
    /// Total bytes written across all data-plane operations (counter).
    pub bytes_written: MetricCounter,
    /// Total graph rewrites initiated (counter).
    pub rewrites_total: MetricCounter,
    /// Histogram of operation latencies in microseconds.
    pub op_latency: MetricHistogram,
    /// Total leases explicitly revoked (counter) — distinct from expired.
    pub leases_revoked: MetricCounter,
    /// Histogram of lease bind latency in microseconds.
    pub bind_latency: MetricHistogram,
    /// Histogram of lease renewal latency in microseconds.
    pub renew_latency: MetricHistogram,
    /// Histogram of lease revocation latency in microseconds.
    pub revoke_latency: MetricHistogram,
    /// Histogram of teardown execution latency in microseconds.
    pub teardown_latency: MetricHistogram,
    /// Total authentication failures (counter).
    pub auth_failures: MetricCounter,
    /// Total anti-replay cache rejections (counter).
    pub replay_rejections: MetricCounter,
    /// Total capability token validations (counter).
    pub token_validations: MetricCounter,
    /// Total capability token validation failures (counter).
    pub token_failures: MetricCounter,
    /// Total stale access attempts after revoke/expiry (counter).
    pub stale_access_rejections: MetricCounter,
    /// Histogram of control-plane operation latencies in microseconds.
    pub control_latency: MetricHistogram,
    /// Histogram of dataplane operation latencies in microseconds.
    pub dataplane_latency: MetricHistogram,
    /// Histogram of tasklet submit (full dispatch) latencies in microseconds.
    pub tasklet_submit_latency: MetricHistogram,
    /// Histogram of tasklet execution-only latencies in microseconds.
    pub tasklet_exec_latency: MetricHistogram,
    /// Total tasklet submissions received (counter).
    pub tasklet_submits: MetricCounter,
    /// Total tasklet executions that completed successfully (counter).
    pub tasklet_completions: MetricCounter,
    /// Total tasklet executions that failed (counter).
    pub tasklet_failures: MetricCounter,
    /// Histogram of tasklet wall-clock duration in microseconds.
    pub tasklet_duration: MetricHistogram,
    /// Total module cache hits (counter).
    pub module_cache_hits: MetricCounter,
    /// Total module cache misses (counter).
    pub module_cache_misses: MetricCounter,
    /// Total module cache stores (counter).
    pub module_cache_stores: MetricCounter,
    /// Total module cache hash mismatches (counter).
    pub module_cache_hash_mismatches: MetricCounter,
}

impl Default for FabricMetrics {
    fn default() -> Self {
        Self::new()
    }
}

impl FabricMetrics {
    /// Create a new metrics instance with all values at zero.
    pub const fn new() -> Self {
        Self {
            leases_active: MetricGauge::new(),
            leases_total: MetricCounter::new(),
            leases_expired: MetricCounter::new(),
            leases_fenced: MetricCounter::new(),
            ops_total: MetricCounter::new(),
            ops_errors: MetricCounter::new(),
            bytes_read: MetricCounter::new(),
            bytes_written: MetricCounter::new(),
            rewrites_total: MetricCounter::new(),
            op_latency: MetricHistogram::new(),
            leases_revoked: MetricCounter::new(),
            bind_latency: MetricHistogram::new(),
            renew_latency: MetricHistogram::new(),
            revoke_latency: MetricHistogram::new(),
            teardown_latency: MetricHistogram::new(),
            auth_failures: MetricCounter::new(),
            replay_rejections: MetricCounter::new(),
            token_validations: MetricCounter::new(),
            token_failures: MetricCounter::new(),
            stale_access_rejections: MetricCounter::new(),
            control_latency: MetricHistogram::new(),
            dataplane_latency: MetricHistogram::new(),
            tasklet_submit_latency: MetricHistogram::new(),
            tasklet_exec_latency: MetricHistogram::new(),
            tasklet_submits: MetricCounter::new(),
            tasklet_completions: MetricCounter::new(),
            tasklet_failures: MetricCounter::new(),
            tasklet_duration: MetricHistogram::new(),
            module_cache_hits: MetricCounter::new(),
            module_cache_misses: MetricCounter::new(),
            module_cache_stores: MetricCounter::new(),
            module_cache_hash_mismatches: MetricCounter::new(),
        }
    }

    /// Access the global metrics instance.
    pub fn global() -> &'static FabricMetrics {
        static INSTANCE: FabricMetrics = FabricMetrics::new();
        &INSTANCE
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn counter_inc_and_add() {
        let c = MetricCounter::new();
        assert_eq!(c.get(), 0);
        c.inc();
        assert_eq!(c.get(), 1);
        c.add(10);
        assert_eq!(c.get(), 11);
    }

    #[test]
    fn counter_reset() {
        let c = MetricCounter::new();
        c.add(42);
        let prev = c.reset();
        assert_eq!(prev, 42);
        assert_eq!(c.get(), 0);
    }

    #[test]
    fn gauge_inc_dec_set() {
        let g = MetricGauge::new();
        assert_eq!(g.get(), 0);
        g.inc();
        g.inc();
        assert_eq!(g.get(), 2);
        g.dec();
        assert_eq!(g.get(), 1);
        g.set(100);
        assert_eq!(g.get(), 100);
        g.set(-5);
        assert_eq!(g.get(), -5);
    }

    #[test]
    fn histogram_observe() {
        let h = MetricHistogram::new();
        // Observation of 50us should land in the 100us bucket (index 0) and all above.
        h.observe(50);
        assert_eq!(h.count(), 1);
        assert_eq!(h.sum(), 50);
        // Bucket 0 (<=100us) should have 1 count.
        assert_eq!(h.bucket_count(0), 1);
        // Bucket 9 (+Inf) should also have 1 count.
        assert_eq!(h.bucket_count(9), 1);
    }

    #[test]
    fn histogram_multiple_observations() {
        let h = MetricHistogram::new();
        h.observe(50); // <=100us bucket
        h.observe(200); // <=500us bucket
        h.observe(2_000); // <=5ms bucket
        h.observe(999_999); // <=1s bucket

        assert_eq!(h.count(), 4);
        assert_eq!(h.sum(), 50 + 200 + 2_000 + 999_999);

        // Bucket 0 (<=100us): only the 50us observation.
        assert_eq!(h.bucket_count(0), 1);
        // Bucket 1 (<=500us): 50 + 200.
        assert_eq!(h.bucket_count(1), 2);
        // Bucket 3 (<=5ms): 50 + 200 + 2000.
        assert_eq!(h.bucket_count(3), 3);
        // Bucket 8 (<=1s): all four.
        assert_eq!(h.bucket_count(8), 4);
        // +Inf: all four.
        assert_eq!(h.bucket_count(9), 4);
    }

    #[test]
    fn histogram_out_of_bounds_index() {
        let h = MetricHistogram::new();
        assert_eq!(h.bucket_count(99), 0);
        assert_eq!(h.bucket_bound(99), u64::MAX);
    }

    #[test]
    fn fabric_metrics_global_is_singleton() {
        let m1 = FabricMetrics::global();
        let m2 = FabricMetrics::global();
        // Same address.
        assert!(core::ptr::eq(m1, m2));
    }

    #[test]
    fn fabric_metrics_fields() {
        let m = FabricMetrics::new();
        m.leases_active.inc();
        m.leases_total.inc();
        m.ops_total.add(5);
        m.bytes_read.add(1024);
        m.bytes_written.add(512);
        m.rewrites_total.inc();
        m.leases_expired.inc();
        m.leases_fenced.inc();
        m.ops_errors.inc();

        assert_eq!(m.leases_active.get(), 1);
        assert_eq!(m.leases_total.get(), 1);
        assert_eq!(m.ops_total.get(), 5);
        assert_eq!(m.bytes_read.get(), 1024);
        assert_eq!(m.bytes_written.get(), 512);
        assert_eq!(m.rewrites_total.get(), 1);
        assert_eq!(m.leases_expired.get(), 1);
        assert_eq!(m.leases_fenced.get(), 1);
        assert_eq!(m.ops_errors.get(), 1);

        m.leases_revoked.inc();
        m.auth_failures.inc();
        m.replay_rejections.inc();
        m.token_validations.add(10);
        m.token_failures.inc();
        m.stale_access_rejections.inc();
        m.bind_latency.observe(100);
        m.renew_latency.observe(200);
        m.revoke_latency.observe(300);
        m.teardown_latency.observe(400);

        assert_eq!(m.leases_revoked.get(), 1);
        assert_eq!(m.auth_failures.get(), 1);
        assert_eq!(m.replay_rejections.get(), 1);
        assert_eq!(m.token_validations.get(), 10);
        assert_eq!(m.token_failures.get(), 1);
        assert_eq!(m.stale_access_rejections.get(), 1);
        assert_eq!(m.bind_latency.count(), 1);
        assert_eq!(m.renew_latency.count(), 1);
        assert_eq!(m.revoke_latency.count(), 1);
        assert_eq!(m.teardown_latency.count(), 1);
    }
}