reddb-io-server 1.28.8

//! Prometheus metrics for HTTP admission and synchronous execution.
//!
//! Exposes the legacy handler-envelope series plus execution-admission
//! gauges and queue-wait histograms per ADR-0017 (Prometheus / Grafana
//! adapters), so operators can distinguish front-door pressure from engine
//! saturation:
//!
//! - `http_active_handler_threads{transport}` — legacy-named gauge of admitted
//!   requests, sourced from `HttpConnectionLimiter::current()` at scrape time.
//! - `http_handler_cap{transport}` — static gauge, sourced from the
//!   limiter's cap. Same value for `http` and `https` since slice 3
//!   collapsed both transports onto one cap.
//! - `http_handler_rejected_total{transport, reason}` — counter,
//!   incremented per admission or deadline rejection.
//! - `http_handler_duration_seconds{transport}` — histogram of total
//!   handler wall-clock time, sampled on every handler exit
//!   (happy-path and timeout). Buckets are the standard Prometheus
//!   client default set so existing dashboards render quantiles via
//!   `histogram_quantile` without configuration.
//! - `http_execution_{workers,active,queued,queue_capacity}` — live execution
//!   lane state.
//! - `http_execution_queue_wait_seconds{transport}` — time admitted requests
//!   spend waiting for a synchronous execution slot.
//!
//! Counters and histogram updates are plain `AtomicU64` operations on
//! the hot path. No registry, no mutex.

use std::fmt::Write;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

use super::http_connection_limiter::HttpConnectionLimiter;
use crate::transport::execution_admission::ExecutionAdmission;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum HttpTransport {
    Http,
    Https,
}

impl HttpTransport {
    fn label(self) -> &'static str {
        match self {
            HttpTransport::Http => "http",
            HttpTransport::Https => "https",
        }
    }

    fn index(self) -> usize {
        match self {
            HttpTransport::Http => 0,
            HttpTransport::Https => 1,
        }
    }
}

#[derive(Debug, Clone, Copy)]
pub enum HttpRejectReason {
    CapExhausted,
    HandlerTimeout,
    ExecutionQueueFull,
    ExecutionQueueTimeout,
    /// Issue #934 — a principal hit its per-principal concurrent in-flight
    /// cap (the global cap still had room; this caller did not).
    PrincipalCapExhausted,
}

/// Number of distinct [`HttpRejectReason`] variants — array width for the
/// per-transport rejection counters.
const REJECT_REASONS: usize = 5;

impl HttpRejectReason {
    fn label(self) -> &'static str {
        match self {
            HttpRejectReason::CapExhausted => "cap_exhausted",
            HttpRejectReason::HandlerTimeout => "handler_timeout",
            HttpRejectReason::ExecutionQueueFull => "execution_queue_full",
            HttpRejectReason::ExecutionQueueTimeout => "execution_queue_timeout",
            HttpRejectReason::PrincipalCapExhausted => "principal_cap_exhausted",
        }
    }

    fn index(self) -> usize {
        match self {
            HttpRejectReason::CapExhausted => 0,
            HttpRejectReason::HandlerTimeout => 1,
            HttpRejectReason::ExecutionQueueFull => 2,
            HttpRejectReason::ExecutionQueueTimeout => 3,
            HttpRejectReason::PrincipalCapExhausted => 4,
        }
    }

    /// All variants, for render-time iteration.
    fn all() -> [HttpRejectReason; REJECT_REASONS] {
        [
            HttpRejectReason::CapExhausted,
            HttpRejectReason::HandlerTimeout,
            HttpRejectReason::ExecutionQueueFull,
            HttpRejectReason::ExecutionQueueTimeout,
            HttpRejectReason::PrincipalCapExhausted,
        ]
    }
}

/// Prometheus client default histogram buckets, in seconds. Aligned
/// with `prometheus.DefBuckets` so operator dashboards render
/// `histogram_quantile` without per-deployment tuning.
const DURATION_BUCKETS_SECONDS: [f64; 11] = [
    0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0,
];

#[derive(Debug)]
struct TransportHistogram {
    /// One counter per bucket in `DURATION_BUCKETS_SECONDS`, plus a
    /// trailing `+Inf` slot that holds the total sample count.
    buckets: [AtomicU64; 11],
    inf: AtomicU64,
    /// Sum of observed durations in microseconds (we keep the sum in
    /// `u64` and convert to seconds at render time).
    sum_micros: AtomicU64,
}

impl TransportHistogram {
    fn new() -> Self {
        Self {
            buckets: [
                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),
                AtomicU64::new(0),
            ],
            inf: AtomicU64::new(0),
            sum_micros: AtomicU64::new(0),
        }
    }

    fn observe_seconds(&self, value: f64) {
        let micros = (value * 1_000_000.0).round().clamp(0.0, u64::MAX as f64) as u64;
        self.sum_micros.fetch_add(micros, Ordering::Relaxed);
        self.inf.fetch_add(1, Ordering::Relaxed);
        for (i, le) in DURATION_BUCKETS_SECONDS.iter().enumerate() {
            if value <= *le {
                self.buckets[i].fetch_add(1, Ordering::Relaxed);
            }
        }
    }
}

#[derive(Debug)]
struct Inner {
    rejected: [[AtomicU64; REJECT_REASONS]; 2],
    duration: [TransportHistogram; 2],
    queue_wait: [TransportHistogram; 2],
}

/// One fresh zeroed rejection-counter row (`REJECT_REASONS` wide).
fn zeroed_reject_row() -> [AtomicU64; REJECT_REASONS] {
    std::array::from_fn(|_| AtomicU64::new(0))
}

#[derive(Debug, Clone)]
pub struct HttpHandlerMetrics {
    inner: Arc<Inner>,
}

impl HttpHandlerMetrics {
    pub fn new() -> Self {
        Self {
            inner: Arc::new(Inner {
                rejected: [zeroed_reject_row(), zeroed_reject_row()],
                duration: [TransportHistogram::new(), TransportHistogram::new()],
                queue_wait: [TransportHistogram::new(), TransportHistogram::new()],
            }),
        }
    }

    pub fn record_reject(&self, transport: HttpTransport, reason: HttpRejectReason) {
        self.inner.rejected[transport.index()][reason.index()].fetch_add(1, Ordering::Relaxed);
    }

    pub fn record_duration(&self, transport: HttpTransport, seconds: f64) {
        if !seconds.is_finite() || seconds < 0.0 {
            return;
        }
        self.inner.duration[transport.index()].observe_seconds(seconds);
    }

    pub fn record_queue_wait(&self, transport: HttpTransport, seconds: f64) {
        if !seconds.is_finite() || seconds < 0.0 {
            return;
        }
        self.inner.queue_wait[transport.index()].observe_seconds(seconds);
    }

    pub fn rejected_count(&self, transport: HttpTransport, reason: HttpRejectReason) -> u64 {
        self.inner.rejected[transport.index()][reason.index()].load(Ordering::Relaxed)
    }

    pub fn duration_sample_count(&self, transport: HttpTransport) -> u64 {
        self.inner.duration[transport.index()]
            .inf
            .load(Ordering::Relaxed)
    }

    pub fn queue_wait_sample_count(&self, transport: HttpTransport) -> u64 {
        self.inner.queue_wait[transport.index()]
            .inf
            .load(Ordering::Relaxed)
    }

    /// Render admission counters and accumulated histograms in Prometheus text
    /// exposition format. Kept as the source-compatible public formatter for
    /// callers that do not own the server's live execution gate.
    pub fn render(&self, body: &mut String, limiter: &HttpConnectionLimiter) {
        self.render_inner(body, limiter, None);
    }

    /// Render admission plus live execution-gate series for `/metrics`.
    pub(crate) fn render_with_execution(
        &self,
        body: &mut String,
        limiter: &HttpConnectionLimiter,
        execution: &ExecutionAdmission,
    ) {
        self.render_inner(body, limiter, Some(execution));
    }

    fn render_inner(
        &self,
        body: &mut String,
        limiter: &HttpConnectionLimiter,
        execution: Option<&ExecutionAdmission>,
    ) {
        let cap = limiter.cap();
        let current = limiter.current();

        // `http_active_handler_threads{transport}` — legacy-named gauge.
        // The clear-text and TLS accept loops share a single limiter
        // (slice 3), so the live count is duplicated on both labels
        // for dashboard ergonomics: an operator can graph the per-
        // transport pane without special-casing the shared cap.
        let _ = writeln!(
            body,
            "# HELP http_active_handler_threads Live admitted HTTP/HTTPS requests holding a limiter permit."
        );
        let _ = writeln!(body, "# TYPE http_active_handler_threads gauge");
        let _ = writeln!(
            body,
            "http_active_handler_threads{{transport=\"http\"}} {}",
            current
        );
        let _ = writeln!(
            body,
            "http_active_handler_threads{{transport=\"https\"}} {}",
            current
        );

        // `http_handler_cap{transport}` — static gauge.
        let _ = writeln!(
            body,
            "# HELP http_handler_cap Configured maximum admitted in-flight HTTP/HTTPS requests."
        );
        let _ = writeln!(body, "# TYPE http_handler_cap gauge");
        let _ = writeln!(body, "http_handler_cap{{transport=\"http\"}} {}", cap);
        let _ = writeln!(body, "http_handler_cap{{transport=\"https\"}} {}", cap);

        if let Some(execution) = execution {
            let execution_active = execution.active();
            let execution_queued = execution.queued();
            let execution_workers = execution.active_max();
            let execution_queue_capacity = execution.queued_max();
            let _ = writeln!(
                body,
                "# HELP http_execution_workers Legacy alias for the runtime-wide engine-call limit."
            );
            let _ = writeln!(body, "# TYPE http_execution_workers gauge");
            let _ = writeln!(body, "http_execution_workers {execution_workers}");
            let _ = writeln!(
                body,
                "# HELP reddb_execution_active_limit Runtime-wide maximum concurrent engine calls across server transports."
            );
            let _ = writeln!(body, "# TYPE reddb_execution_active_limit gauge");
            let _ = writeln!(body, "reddb_execution_active_limit {execution_workers}");
            let _ = writeln!(
                body,
                "# HELP http_execution_active Legacy alias for runtime-wide active engine calls."
            );
            let _ = writeln!(body, "# TYPE http_execution_active gauge");
            let _ = writeln!(body, "http_execution_active {execution_active}");
            let _ = writeln!(
                body,
                "# HELP reddb_execution_active Runtime-wide active engine calls across server transports."
            );
            let _ = writeln!(body, "# TYPE reddb_execution_active gauge");
            let _ = writeln!(body, "reddb_execution_active {execution_active}");
            let _ = writeln!(
                body,
                "# HELP http_execution_queued Legacy alias for runtime-wide operations waiting for an engine slot."
            );
            let _ = writeln!(body, "# TYPE http_execution_queued gauge");
            let _ = writeln!(body, "http_execution_queued {execution_queued}");
            let _ = writeln!(
                body,
                "# HELP reddb_execution_queued Runtime-wide requests waiting for an engine execution slot."
            );
            let _ = writeln!(body, "# TYPE reddb_execution_queued gauge");
            let _ = writeln!(body, "reddb_execution_queued {execution_queued}");
            let _ = writeln!(
                body,
                "# HELP http_execution_queue_capacity Legacy alias for the runtime-wide execution wait-queue limit."
            );
            let _ = writeln!(body, "# TYPE http_execution_queue_capacity gauge");
            let _ = writeln!(
                body,
                "http_execution_queue_capacity {execution_queue_capacity}"
            );
            let _ = writeln!(
                body,
                "# HELP reddb_execution_queue_limit Runtime-wide bounded execution wait-queue limit."
            );
            let _ = writeln!(body, "# TYPE reddb_execution_queue_limit gauge");
            let _ = writeln!(
                body,
                "reddb_execution_queue_limit {execution_queue_capacity}"
            );
        }

        // `http_handler_rejected_total{transport, reason}` — counter.
        let _ = writeln!(
            body,
            "# HELP http_handler_rejected_total HTTP/HTTPS handler rejections by reason since process start."
        );
        let _ = writeln!(body, "# TYPE http_handler_rejected_total counter");
        for transport in [HttpTransport::Http, HttpTransport::Https] {
            for reason in HttpRejectReason::all() {
                let _ = writeln!(
                    body,
                    "http_handler_rejected_total{{transport=\"{}\",reason=\"{}\"}} {}",
                    transport.label(),
                    reason.label(),
                    self.rejected_count(transport, reason)
                );
            }
        }

        // `http_handler_duration_seconds{transport}` — histogram.
        let _ = writeln!(
            body,
            "# HELP http_handler_duration_seconds Wall-clock handler duration per transport."
        );
        let _ = writeln!(body, "# TYPE http_handler_duration_seconds histogram");
        for transport in [HttpTransport::Http, HttpTransport::Https] {
            let hist = &self.inner.duration[transport.index()];
            for (i, le) in DURATION_BUCKETS_SECONDS.iter().enumerate() {
                let _ = writeln!(
                    body,
                    "http_handler_duration_seconds_bucket{{transport=\"{}\",le=\"{}\"}} {}",
                    transport.label(),
                    format_bucket_le(*le),
                    hist.buckets[i].load(Ordering::Relaxed)
                );
            }
            let inf = hist.inf.load(Ordering::Relaxed);
            let _ = writeln!(
                body,
                "http_handler_duration_seconds_bucket{{transport=\"{}\",le=\"+Inf\"}} {}",
                transport.label(),
                inf
            );
            let sum_secs = (hist.sum_micros.load(Ordering::Relaxed) as f64) / 1_000_000.0;
            let _ = writeln!(
                body,
                "http_handler_duration_seconds_sum{{transport=\"{}\"}} {}",
                transport.label(),
                sum_secs
            );
            let _ = writeln!(
                body,
                "http_handler_duration_seconds_count{{transport=\"{}\"}} {}",
                transport.label(),
                inf
            );
        }

        let _ = writeln!(
            body,
            "# HELP http_execution_queue_wait_seconds Time an admitted HTTP request waited for an engine worker."
        );
        let _ = writeln!(body, "# TYPE http_execution_queue_wait_seconds histogram");
        for transport in [HttpTransport::Http, HttpTransport::Https] {
            let histogram = &self.inner.queue_wait[transport.index()];
            for (index, upper_bound) in DURATION_BUCKETS_SECONDS.iter().enumerate() {
                let _ = writeln!(
                    body,
                    "http_execution_queue_wait_seconds_bucket{{transport=\"{}\",le=\"{}\"}} {}",
                    transport.label(),
                    format_bucket_le(*upper_bound),
                    histogram.buckets[index].load(Ordering::Relaxed)
                );
            }
            let count = histogram.inf.load(Ordering::Relaxed);
            let _ = writeln!(
                body,
                "http_execution_queue_wait_seconds_bucket{{transport=\"{}\",le=\"+Inf\"}} {}",
                transport.label(),
                count
            );
            let sum_seconds = (histogram.sum_micros.load(Ordering::Relaxed) as f64) / 1_000_000.0;
            let _ = writeln!(
                body,
                "http_execution_queue_wait_seconds_sum{{transport=\"{}\"}} {}",
                transport.label(),
                sum_seconds
            );
            let _ = writeln!(
                body,
                "http_execution_queue_wait_seconds_count{{transport=\"{}\"}} {}",
                transport.label(),
                count
            );
        }
    }
}

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

fn format_bucket_le(le: f64) -> String {
    // Match Prometheus's exposition formatting: trailing zeros are
    // significant for canonical bucket labels, so we use the same
    // shape that the upstream client library emits.
    if le == le.trunc() && le.abs() < 1e16 {
        format!("{le:.1}")
    } else {
        format!("{le}")
    }
}

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

    #[test]
    fn rejected_counters_isolated_by_label() {
        let m = HttpHandlerMetrics::new();
        m.record_reject(HttpTransport::Http, HttpRejectReason::CapExhausted);
        m.record_reject(HttpTransport::Http, HttpRejectReason::CapExhausted);
        m.record_reject(HttpTransport::Https, HttpRejectReason::HandlerTimeout);
        assert_eq!(
            m.rejected_count(HttpTransport::Http, HttpRejectReason::CapExhausted),
            2
        );
        assert_eq!(
            m.rejected_count(HttpTransport::Http, HttpRejectReason::HandlerTimeout),
            0
        );
        assert_eq!(
            m.rejected_count(HttpTransport::Https, HttpRejectReason::HandlerTimeout),
            1
        );
        assert_eq!(
            m.rejected_count(HttpTransport::Https, HttpRejectReason::CapExhausted),
            0
        );
    }

    #[test]
    fn duration_histogram_buckets_are_cumulative() {
        let m = HttpHandlerMetrics::new();
        m.record_duration(HttpTransport::Http, 0.003);
        m.record_duration(HttpTransport::Http, 0.04);
        m.record_duration(HttpTransport::Http, 3.0);
        assert_eq!(m.duration_sample_count(HttpTransport::Http), 3);

        let limiter = HttpConnectionLimiter::new(4);
        let execution = ExecutionAdmission::new(2, 2);
        let mut body = String::new();
        m.render_with_execution(&mut body, &limiter, &execution);

        // `le="0.005"` includes only the 3ms sample (cumulative).
        assert!(body
            .contains("http_handler_duration_seconds_bucket{transport=\"http\",le=\"0.005\"} 1"));
        // `le="0.05"` includes the 3ms + 40ms samples.
        assert!(
            body.contains("http_handler_duration_seconds_bucket{transport=\"http\",le=\"0.05\"} 2")
        );
        // `le="+Inf"` sees all 3 samples.
        assert!(
            body.contains("http_handler_duration_seconds_bucket{transport=\"http\",le=\"+Inf\"} 3")
        );
        // HTTPS labelset present but empty.
        assert!(body
            .contains("http_handler_duration_seconds_bucket{transport=\"https\",le=\"+Inf\"} 0"));
    }

    #[test]
    fn render_includes_cap_and_current_from_limiter() {
        let limiter = HttpConnectionLimiter::new(7);
        let execution = ExecutionAdmission::new(2, 5);
        let _p = limiter.try_acquire().unwrap();
        let m = HttpHandlerMetrics::new();
        let mut body = String::new();
        m.render_with_execution(&mut body, &limiter, &execution);
        assert!(body.contains("http_handler_cap{transport=\"http\"} 7"));
        assert!(body.contains("http_handler_cap{transport=\"https\"} 7"));
        assert!(body.contains("http_active_handler_threads{transport=\"http\"} 1"));
        assert!(body.contains("http_active_handler_threads{transport=\"https\"} 1"));
        assert!(body.contains("http_execution_workers 2"));
        assert!(body.contains("http_execution_queue_capacity 5"));
        assert!(body.contains("reddb_execution_active_limit 2"));
        assert!(body.contains("reddb_execution_queue_limit 5"));
    }

    #[test]
    fn render_emits_all_rejection_labels() {
        let m = HttpHandlerMetrics::new();
        let limiter = HttpConnectionLimiter::new(1);
        let execution = ExecutionAdmission::new(1, 0);
        let mut body = String::new();
        m.render_with_execution(&mut body, &limiter, &execution);
        for transport in ["http", "https"] {
            for reason in [
                "cap_exhausted",
                "handler_timeout",
                "execution_queue_full",
                "execution_queue_timeout",
                "principal_cap_exhausted",
            ] {
                let expected = format!(
                    "http_handler_rejected_total{{transport=\"{transport}\",reason=\"{reason}\"}} 0"
                );
                assert!(body.contains(&expected), "missing line: {expected}");
            }
        }
    }

    #[test]
    fn negative_or_nan_durations_are_ignored() {
        let m = HttpHandlerMetrics::new();
        m.record_duration(HttpTransport::Http, -1.0);
        m.record_duration(HttpTransport::Http, f64::NAN);
        m.record_duration(HttpTransport::Http, f64::INFINITY);
        assert_eq!(m.duration_sample_count(HttpTransport::Http), 0);
    }

    #[test]
    fn queue_wait_histogram_is_distinct_from_handler_duration() {
        let metrics = HttpHandlerMetrics::new();
        metrics.record_queue_wait(HttpTransport::Http, 0.04);
        assert_eq!(metrics.queue_wait_sample_count(HttpTransport::Http), 1);
        assert_eq!(metrics.duration_sample_count(HttpTransport::Http), 0);

        let limiter = HttpConnectionLimiter::new(4);
        let execution = ExecutionAdmission::new(1, 3);
        let mut body = String::new();
        metrics.render_with_execution(&mut body, &limiter, &execution);
        assert!(body.contains(
            "http_execution_queue_wait_seconds_bucket{transport=\"http\",le=\"0.05\"} 1"
        ));
        assert!(body.contains("http_execution_queue_wait_seconds_count{transport=\"http\"} 1"));
    }
}