server/lib/src/metrics.rs
//! Prometheus metrics — a process-global counter/gauge registry backing the
//! read-only `GET /metrics` door (parity §K, the `full`-tier operability gap).
//!
//! Design (deliberately dependency-free): a handful of `AtomicU64` counters
//! bumped at the natural call sites the server already flows through — the
//! HTTP/OCI gateway (`exposed::http`) increments a per-verb request counter and
//! the bytes-served total on every backend request/search/sbom door hit, and the
//! retention executor (`Holger::execute_repo_retention`) bumps a run counter.
//! There is no scrape-time cost beyond a snapshot of the atomics; the only I/O is
//! the artifact-count gauge, which lists each backend at scrape time (bounded,
//! off the hot path, run under `spawn_blocking` by the door).
//!
//! Why a process-global (not a threaded handle): the emitting sites live on three
//! unrelated planes (the HTTP door, the gRPC services, the retention executor),
//! and threading one `Arc<Metrics>` through all of them would be a large,
//! cross-cutting change for a purely additive observability feature. A single
//! `OnceLock<Metrics>` (`metrics::global()`) lets any site bump the same registry
//! without wiring, and the render path is a **pure** function over a `&Metrics`
//! snapshot + a supplied repo→count list, so tests exercise an isolated instance
//! (no global interference) and assert exact counts (RED-when-broken).
//!
//! Output is Prometheus text exposition format 0.0.4: one `# HELP`/`# TYPE`
//! header per family, then samples. Counter families end in `_total`.
use crate::lock::LockOrRecover as _;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::OnceLock;
use crate::audit::AuditAction;
/// The logical verb a request is counted under — the same counter family is fed
/// by BOTH doors: the HTTP/OCI gateway (`exposed::http`) and the gRPC services
/// (via [`HolgerGrpc::record_audit`](crate::grpc)), so `/metrics` reflects the
/// whole registry's traffic, not just the HTTP half. GET/HEAD (and listings) →
/// Get, PUT/POST/PATCH → Put, DELETE → Delete, a DEV→PROD promotion → Promote,
/// plus the two read-only `/-/` doors as their own verbs so operators can see
/// search and SBOM traffic distinctly from artifact pulls.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Verb {
Get,
Put,
Delete,
Promote,
Search,
Sbom,
}
/// Every verb, in render order. One source of truth so `record`, `request_count`
/// and `render_prometheus` can never drift apart (a new verb is added here once).
pub const ALL_VERBS: [Verb; 6] = [
Verb::Get,
Verb::Put,
Verb::Delete,
Verb::Promote,
Verb::Search,
Verb::Sbom,
];
/// The outcome of one proxy/group-repo request — the labels of the
/// `holger_proxy_requests_total{result="…"}` family. A [`ProxyBackend`] chains a
/// primary with an ordered upstream fallback list; every read exits through
/// exactly ONE of these outcomes, so summing the family gives total proxy traffic
/// and the ratio of `primary_hit`/`upstream_hit` to `miss` is the cache hit-rate
/// an operator watches. Emitted at the clean outcome points in
/// [`ProxyBackend::fetch`](crate::proxy) / `handle_http2_request`.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ProxyOutcome {
/// Served straight from the primary backend (local cache hit).
PrimaryHit,
/// Primary missed; served from an upstream in the fallback chain (pull-through).
UpstreamHit,
/// Short-circuited by a live negative-cache entry (a known-absent coordinate
/// returned without re-walking the upstreams).
NegcacheHit,
/// Missed the primary AND every upstream (a true miss, remembered for the TTL).
Miss,
/// An upstream in the chain ERRORED (connection refused, DNS failure, TLS
/// error, a non-404 status) and was skipped so the walk could continue. Not a
/// request outcome but a per-upstream event: one read over a 3-upstream chain
/// with two dead members bumps this twice. Counted so a dead upstream is
/// visible in `/metrics` rather than merely tolerated.
UpstreamError,
}
/// Every proxy outcome, in render order. One source of truth so `record_proxy`,
/// `proxy_count` and `render_prometheus` can never drift apart.
const ALL_PROXY_OUTCOMES: [ProxyOutcome; 5] = [
ProxyOutcome::PrimaryHit,
ProxyOutcome::UpstreamHit,
ProxyOutcome::NegcacheHit,
ProxyOutcome::Miss,
ProxyOutcome::UpstreamError,
];
impl ProxyOutcome {
/// The Prometheus `result=` label value for this outcome.
pub fn label(self) -> &'static str {
match self {
ProxyOutcome::PrimaryHit => "primary_hit",
ProxyOutcome::UpstreamHit => "upstream_hit",
ProxyOutcome::NegcacheHit => "negcache_hit",
ProxyOutcome::Miss => "miss",
ProxyOutcome::UpstreamError => "upstream_error",
}
}
}
impl Verb {
/// The Prometheus label value for this verb.
pub fn label(self) -> &'static str {
match self {
Verb::Get => "get",
Verb::Put => "put",
Verb::Delete => "delete",
Verb::Promote => "promote",
Verb::Search => "search",
Verb::Sbom => "sbom",
}
}
/// The `/metrics` verb an [`AuditAction`] is counted under — the ONE
/// audit-action→verb classifier, shared by the gRPC `record_audit` choke point
/// and the HTTP door (Law #5, no twin). `None` ⇒ the action is not request
/// traffic and is not counted (e.g. [`AuditAction::Other`]). Reads (download +
/// listing) fold into `Get`; a store is `Put`; delete and promote are their
/// own verbs. (`Search`/`Sbom` are bumped explicitly by their dedicated
/// read-only HTTP doors, which don't route through an audit action here.)
pub fn from_audit_action(action: AuditAction) -> Option<Verb> {
match action {
AuditAction::Download | AuditAction::List => Some(Verb::Get),
AuditAction::Upload => Some(Verb::Put),
AuditAction::Delete => Some(Verb::Delete),
AuditAction::Promote => Some(Verb::Promote),
// A plan write is not request traffic — counting it as a verb would put
// an operator's billing action in the same series as a tenant's fetches.
AuditAction::Entitlement => None,
// One admin walk, not request traffic; its deletes are in the audit row.
AuditAction::Purge => None,
AuditAction::Scan => None,
AuditAction::Token => None,
AuditAction::Other => None,
}
}
}
/// Upper bounds (seconds) of the request-duration histogram buckets, ascending.
/// A Prometheus histogram reports the CUMULATIVE count `≤ le` for each bound plus
/// an implicit `+Inf` bucket for everything above the last bound; these are the
/// conventional web-latency boundaries (1 ms … 10 s).
const DURATION_BUCKETS: [f64; 12] = [
0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0,
];
/// A lock-free Prometheus histogram: one atomic per bucket (non-cumulative — the
/// render cumulates), plus the total observation count and the nanosecond sum.
/// The trailing bucket slot (`buckets[DURATION_BUCKETS.len()]`) is the `+Inf`
/// overflow (observations larger than the last bound).
#[derive(Debug)]
struct DurationHistogram {
buckets: [AtomicU64; DURATION_BUCKETS.len() + 1],
count: AtomicU64,
sum_nanos: AtomicU64,
}
impl Default for DurationHistogram {
fn default() -> Self {
Self {
buckets: std::array::from_fn(|_| AtomicU64::new(0)),
count: AtomicU64::new(0),
sum_nanos: AtomicU64::new(0),
}
}
}
impl DurationHistogram {
/// Record one observation of `seconds`. Non-finite or negative values are
/// ignored (a clock quirk must never corrupt the sum). One `Relaxed` add to
/// the matching bucket, the count, and the nanosecond sum.
fn observe(&self, seconds: f64) {
if !seconds.is_finite() || seconds < 0.0 {
return;
}
// First bucket whose upper bound is ≥ the observation, else the overflow slot.
let idx = DURATION_BUCKETS
.iter()
.position(|&b| seconds <= b)
.unwrap_or(DURATION_BUCKETS.len());
self.buckets[idx].fetch_add(1, Ordering::Relaxed);
self.count.fetch_add(1, Ordering::Relaxed);
self.sum_nanos
.fetch_add((seconds * 1e9) as u64, Ordering::Relaxed);
}
}
/// One mirror source's health, as numbers a scrape can read.
///
/// # Why this is here and not in a log line
///
/// A mirror that has stopped mirroring reports `is-active: active` and
/// `NRestarts: 0`, because nothing has crashed: the process is up, both doors
/// answer, and the only evidence of trouble is the ABSENCE of a line in a
/// journal — and successful cycles log at `info`, which the default `warn` floor
/// hides. Silence therefore means both "mirroring fine" and "not mirroring at
/// all", and no amount of log discipline fixes an ambiguity that lives in what
/// is *not* printed.
///
/// A timestamp fixes it. `holger_mirror_last_success_timestamp_seconds` either
/// moves or it does not, and `time() - that` is an alert an operator can write
/// without reading anything. The counters beside it say whether the mirror is
/// taking bytes or merely completing empty cycles.
///
/// All fields are `AtomicU64` and every bump is `Relaxed`: this sits on the
/// mirror's timer, not in a request path, and it must never be a reason a cycle
/// is slower.
#[derive(Debug, Default)]
pub struct MirrorSourceHealth {
/// The declared `ron_name`, verbatim — the `source=` label.
name: String,
/// When this source's health was first registered (process start, near
/// enough). The floor for "how long has it been stalled" before any cycle
/// has ever succeeded: a mirror that has NEVER completed a cycle is stalled
/// since boot, not stalled for zero seconds.
registered_unix: AtomicU64,
cycles_ok: AtomicU64,
cycles_failed: AtomicU64,
artifacts_fetched: AtomicU64,
artifacts_failed: AtomicU64,
bytes_fetched: AtomicU64,
/// Unix seconds of the last completed cycle. `0` ⇒ never.
last_success_unix: AtomicU64,
/// Failures since the last completed cycle.
consecutive_failures: AtomicU64,
/// Set to the failure count when a cycle completes after failures, and
/// cleared by [`MirrorSourceHealth::take_recovery`] once it has been said.
recovery_pending: AtomicU64,
/// The paired upstream's key state ([`crate::peers::LinkState`]), set by
/// the pin check before each cycle. `0` = unchecked or unpaired.
link_state: std::sync::atomic::AtomicU8,
}
/// Unix seconds now, `0` if the clock is before the epoch (it is not).
fn now_unix() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
}
impl MirrorSourceHealth {
/// A fresh registration for `name`, stamped now.
pub fn new(name: &str) -> Self {
let h = Self {
name: name.to_string(),
..Self::default()
};
h.registered_unix.store(now_unix(), Ordering::Relaxed);
h
}
/// The source's declared name.
pub fn name(&self) -> &str {
&self.name
}
/// Record what the pin check found.
pub fn set_link_state(&self, s: crate::peers::LinkState) {
self.link_state.store(s.to_u8(), Ordering::Relaxed);
}
/// What the last pin check found.
pub fn link_state(&self) -> crate::peers::LinkState {
crate::peers::LinkState::from_u8(self.link_state.load(Ordering::Relaxed))
}
/// One cycle completed: `fetched` artifacts, `bytes` bytes, `failed`
/// artifacts it could not take (which do NOT make the cycle a failure —
/// they are carried in the change map and asked for again).
pub fn record_success(&self, fetched: u64, bytes: u64, failed: u64) {
self.cycles_ok.fetch_add(1, Ordering::Relaxed);
self.artifacts_fetched.fetch_add(fetched, Ordering::Relaxed);
self.artifacts_failed.fetch_add(failed, Ordering::Relaxed);
self.bytes_fetched.fetch_add(bytes, Ordering::Relaxed);
self.last_success_unix.store(now_unix(), Ordering::Relaxed);
let previous = self.consecutive_failures.swap(0, Ordering::Relaxed);
if previous > 0 {
self.recovery_pending.store(previous, Ordering::Relaxed);
}
}
/// One cycle failed.
pub fn record_failure(&self) {
self.cycles_failed.fetch_add(1, Ordering::Relaxed);
self.consecutive_failures.fetch_add(1, Ordering::Relaxed);
}
/// Failures since the last completed cycle.
pub fn consecutive_failures(&self) -> u64 {
self.consecutive_failures.load(Ordering::Relaxed)
}
/// Unix seconds of the last completed cycle, `None` when there has never
/// been one.
pub fn last_success_unix(&self) -> Option<u64> {
match self.last_success_unix.load(Ordering::Relaxed) {
0 => None,
t => Some(t),
}
}
/// Completed cycles / failed cycles / artifacts fetched / artifacts failed /
/// bytes fetched.
pub fn counts(&self) -> (u64, u64, u64, u64, u64) {
(
self.cycles_ok.load(Ordering::Relaxed),
self.cycles_failed.load(Ordering::Relaxed),
self.artifacts_fetched.load(Ordering::Relaxed),
self.artifacts_failed.load(Ordering::Relaxed),
self.bytes_fetched.load(Ordering::Relaxed),
)
}
/// How many seconds since this source last completed a cycle — measured
/// from registration when it never has. `None` only if the clock went
/// backwards.
pub fn stalled_for_secs(&self) -> Option<u64> {
let since = self
.last_success_unix()
.unwrap_or_else(|| self.registered_unix.load(Ordering::Relaxed));
now_unix().checked_sub(since)
}
/// Take the pending "recovered after N failures" fact, if there is one.
/// Returns it once and then forgets it, so the recovery is announced exactly
/// once per outage rather than on every cycle afterwards.
pub fn take_recovery(&self) -> Option<u64> {
match self.recovery_pending.swap(0, Ordering::Relaxed) {
0 => None,
n => Some(n),
}
}
}
/// Process-wide registry of registry-activity counters. All fields are plain
/// atomics; construction is free, so tests spin up a throwaway instance while the
/// live server bumps [`global`].
#[derive(Debug, Default)]
pub struct Metrics {
get: AtomicU64,
put: AtomicU64,
delete: AtomicU64,
promote: AtomicU64,
search: AtomicU64,
sbom: AtomicU64,
bytes_served: AtomicU64,
errors: AtomicU64,
retention_runs: AtomicU64,
audit_missed: AtomicU64,
proxy_primary_hit: AtomicU64,
proxy_upstream_hit: AtomicU64,
proxy_negcache_hit: AtomicU64,
proxy_miss: AtomicU64,
proxy_upstream_error: AtomicU64,
request_duration: DurationHistogram,
/// One entry per declared mirror source, created at boot by the timer and
/// never removed. A `Mutex` rather than an atomic because the SET of sources
/// is written once (at boot) and only read afterwards; the numbers inside
/// each entry are the lock-free part, and no cycle ever takes this lock.
mirror_sources: std::sync::Mutex<Vec<std::sync::Arc<MirrorSourceHealth>>>,
/// Per door: answers with a 5xx status, since boot (the flow's door box; lane HOL-SVG-RON).
door_errors: std::sync::Mutex<std::collections::BTreeMap<String, u64>>,
/// Per scanner kind: scans and findings by severity, since boot (the flow's intake steps).
scans: std::sync::Mutex<std::collections::BTreeMap<String, ScanTally>>,
}
/// **One scanner kind's tally since boot** (lane HOL-SVG-RON): how many artifacts it
/// scanned, its verdicts, and its findings by severity.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct ScanTally {
pub scanned: u64,
/// Verdict word (`pass`, `warn`, `block`, `error`) → count.
pub verdicts: std::collections::BTreeMap<String, u64>,
/// Severity word (`info`, `low`, `medium`, `high`, `critical`) → count.
pub findings: std::collections::BTreeMap<String, u64>,
}
impl Metrics {
/// A fresh, all-zero registry.
pub fn new() -> Self {
Self::default()
}
/// Count one handled request under `verb`, add its response `bytes` to the
/// bytes-served total, and, when `status >= 500`, bump the error counter.
/// This is THE single bump point — every door funnels through it so a verb can
/// never be counted without its bytes/errors also being accounted.
pub fn record_request(&self, verb: Verb, status: u16, bytes: u64) {
let slot = self.slot(verb);
slot.fetch_add(1, Ordering::Relaxed);
if bytes > 0 {
self.bytes_served.fetch_add(bytes, Ordering::Relaxed);
}
if status >= 500 {
self.errors.fetch_add(1, Ordering::Relaxed);
}
}
/// Count one retention-executor invocation (a GC "run", dry-run or not).
pub fn record_retention_run(&self) {
self.retention_runs.fetch_add(1, Ordering::Relaxed);
}
/// Count one audit record that could not be written (A28 M17): the act it records
/// happened, and the append-only trail is missing it. Published as
/// `holger_audit_records_missed_total`, so a security gateway's lost record is a number
/// an alert can watch, never a dropped `Result`.
pub fn record_audit_missed(&self) {
self.audit_missed.fetch_add(1, Ordering::Relaxed);
}
/// Count one proxy/group-repo request under its [`ProxyOutcome`]. Bumped at the
/// single outcome point in the proxy read path (`fetch` / `handle_http2_request`),
/// so every proxy read lands on exactly one `result=` label.
pub fn record_proxy(&self, outcome: ProxyOutcome) {
self.proxy_slot(outcome).fetch_add(1, Ordering::Relaxed);
}
/// The counter slot backing `outcome` — the one place the outcome→field
/// mapping lives, shared by [`record_proxy`](Self::record_proxy) and
/// [`proxy_count`](Self::proxy_count) so the two can't disagree.
fn proxy_slot(&self, outcome: ProxyOutcome) -> &AtomicU64 {
match outcome {
ProxyOutcome::PrimaryHit => &self.proxy_primary_hit,
ProxyOutcome::UpstreamHit => &self.proxy_upstream_hit,
ProxyOutcome::NegcacheHit => &self.proxy_negcache_hit,
ProxyOutcome::Miss => &self.proxy_miss,
ProxyOutcome::UpstreamError => &self.proxy_upstream_error,
}
}
/// Current value of one proxy outcome's counter (test/observability accessor).
pub fn proxy_count(&self, outcome: ProxyOutcome) -> u64 {
self.proxy_slot(outcome).load(Ordering::Relaxed)
}
/// Observe one request's handling duration (seconds) in the
/// `holger_request_duration_seconds` histogram. Called at the HTTP door's
/// per-request exit, alongside the request counter, so latency is measured at
/// the same point traffic is counted.
pub fn observe_request_duration(&self, seconds: f64) {
self.request_duration.observe(seconds);
}
/// Total number of duration observations (test/observability accessor).
pub fn request_duration_count(&self) -> u64 {
self.request_duration.count.load(Ordering::Relaxed)
}
/// Sum of all observed durations, in seconds (test/observability accessor).
pub fn request_duration_sum_seconds(&self) -> f64 {
self.request_duration.sum_nanos.load(Ordering::Relaxed) as f64 / 1e9
}
/// The counter slot backing `verb` — the one place the verb→field mapping
/// lives, shared by [`record_request`](Self::record_request) and
/// [`request_count`](Self::request_count) so the two can't disagree.
fn slot(&self, verb: Verb) -> &AtomicU64 {
match verb {
Verb::Get => &self.get,
Verb::Put => &self.put,
Verb::Delete => &self.delete,
Verb::Promote => &self.promote,
Verb::Search => &self.search,
Verb::Sbom => &self.sbom,
}
}
/// Current value of one verb's request counter (test/observability accessor).
pub fn request_count(&self, verb: Verb) -> u64 {
self.slot(verb).load(Ordering::Relaxed)
}
/// Total bytes served across all verbs.
pub fn bytes_served(&self) -> u64 {
self.bytes_served.load(Ordering::Relaxed)
}
/// Total 5xx responses.
pub fn errors(&self) -> u64 {
self.errors.load(Ordering::Relaxed)
}
/// Total retention-executor runs.
pub fn retention_runs(&self) -> u64 {
self.retention_runs.load(Ordering::Relaxed)
}
/// Audit records that could not be written (see [`Metrics::record_audit_missed`]).
pub fn audit_missed(&self) -> u64 {
self.audit_missed.load(Ordering::Relaxed)
}
/// The health entry for mirror source `name`, registering it on first use.
///
/// Registration is what makes a mirror that has NEVER succeeded visible: the
/// series exists from boot with `holger_mirror_last_success_timestamp_seconds
/// = 0`, so "this source has done nothing at all" is a reading and not an
/// absent series a dashboard silently draws as a gap.
pub fn mirror_source(&self, name: &str) -> std::sync::Arc<MirrorSourceHealth> {
let mut sources = self.mirror_sources.lock_or_recover();
if let Some(h) = sources.iter().find(|h| h.name() == name) {
return h.clone();
}
let h = std::sync::Arc::new(MirrorSourceHealth::new(name));
sources.push(h.clone());
h
}
/// Count one answer of door `door` with `status`: a 5xx is a door error.
pub fn record_door_status(&self, door: &str, status: u16) {
if status >= 500 {
*self
.door_errors
.lock_or_recover()
.entry(door.to_string())
.or_insert(0) += 1;
}
}
/// Per door: its 5xx answers since boot.
pub fn door_errors(&self) -> std::collections::BTreeMap<String, u64> {
self.door_errors.lock_or_recover().clone()
}
/// Count one scan by `kind` (the scanner kind's wire word): its verdict and the
/// severity word of each finding.
pub fn record_scan<'a>(
&self,
kind: &str,
verdict: &str,
severities: impl IntoIterator<Item = &'a str>,
) {
let mut all = self.scans.lock_or_recover();
let t = all.entry(kind.to_string()).or_default();
t.scanned += 1;
*t.verdicts.entry(verdict.to_string()).or_insert(0) += 1;
for s in severities {
*t.findings.entry(s.to_string()).or_insert(0) += 1;
}
}
/// Per scanner kind: the tally since boot.
pub fn scan_tally(&self) -> std::collections::BTreeMap<String, ScanTally> {
self.scans.lock_or_recover().clone()
}
/// Every registered mirror source, in registration (declaration) order.
pub fn mirror_sources(&self) -> Vec<std::sync::Arc<MirrorSourceHealth>> {
self.mirror_sources.lock_or_recover().clone()
}
/// Render the full registry as Prometheus text exposition format 0.0.4.
/// **Pure**: the per-repo artifact gauge is driven entirely by `repo_counts`
/// (a `(repo_name, artifact_count)` list the caller gathers), so this function
/// does no I/O and one code path serves both the live door and the test.
///
/// A `None` count means the backend could NOT be enumerated — the listing
/// errored, or the backend cannot enumerate at all. That is not the number
/// zero, and rendering it as `0` made an unreadable archive look exactly
/// like an empty repository (and made `sum(holger_repo_artifacts)` silently
/// under-count a broken disk). An absent reading therefore emits no
/// `holger_repo_artifacts` sample at all and sets
/// `holger_repo_listable{repo}` to 0 instead, so a scrape can tell "nothing
/// in it" from "we could not look".
pub fn render_prometheus(&self, repo_counts: &[(String, Option<usize>)]) -> String {
let mut out = String::with_capacity(1024);
out.push_str(
"# HELP holger_requests_total Total requests handled (gRPC + HTTP), by verb.\n",
);
out.push_str("# TYPE holger_requests_total counter\n");
for verb in ALL_VERBS {
out.push_str(&format!(
"holger_requests_total{{verb=\"{}\"}} {}\n",
verb.label(),
self.request_count(verb),
));
}
out.push_str("# HELP holger_bytes_served_total Total response bytes served over HTTP.\n");
out.push_str("# TYPE holger_bytes_served_total counter\n");
out.push_str(&format!(
"holger_bytes_served_total {}\n",
self.bytes_served()
));
out.push_str("# HELP holger_errors_total Total 5xx HTTP responses.\n");
out.push_str("# TYPE holger_errors_total counter\n");
out.push_str(&format!("holger_errors_total {}\n", self.errors()));
out.push_str("# HELP holger_retention_runs_total Total retention-executor runs.\n");
out.push_str("# TYPE holger_retention_runs_total counter\n");
out.push_str(&format!(
"holger_retention_runs_total {}\n",
self.retention_runs()
));
out.push_str("# HELP holger_audit_records_missed_total Audit records that could not be written (the act happened; the trail is missing it).\n");
out.push_str("# TYPE holger_audit_records_missed_total counter\n");
out.push_str(&format!(
"holger_audit_records_missed_total {}\n",
self.audit_missed()
));
out.push_str(
"# HELP holger_proxy_requests_total Total proxy/group-repo reads, by outcome.\n",
);
out.push_str("# TYPE holger_proxy_requests_total counter\n");
for outcome in ALL_PROXY_OUTCOMES {
out.push_str(&format!(
"holger_proxy_requests_total{{result=\"{}\"}} {}\n",
outcome.label(),
self.proxy_count(outcome),
));
}
out.push_str(
"# HELP holger_request_duration_seconds HTTP request handling duration in seconds.\n",
);
out.push_str("# TYPE holger_request_duration_seconds histogram\n");
let h = &self.request_duration;
let mut cumulative = 0u64;
for (i, bound) in DURATION_BUCKETS.iter().enumerate() {
cumulative += h.buckets[i].load(Ordering::Relaxed);
out.push_str(&format!(
"holger_request_duration_seconds_bucket{{le=\"{}\"}} {}\n",
fmt_bucket_bound(*bound),
cumulative,
));
}
// The +Inf bucket includes the overflow slot — its count equals the total.
cumulative += h.buckets[DURATION_BUCKETS.len()].load(Ordering::Relaxed);
out.push_str(&format!(
"holger_request_duration_seconds_bucket{{le=\"+Inf\"}} {}\n",
cumulative,
));
out.push_str(&format!(
"holger_request_duration_seconds_sum {}\n",
self.request_duration_sum_seconds(),
));
out.push_str(&format!(
"holger_request_duration_seconds_count {}\n",
self.request_duration_count(),
));
// ── the mirror plane ──────────────────────────────────────────────
//
// Emitted for every DECLARED source, whether or not it has ever managed
// a cycle, because "no series" and "never succeeded" look the same on a
// graph and only one of them is a mirror that is not mirroring.
//
// `holger_mirror_last_success_timestamp_seconds` is the one an alert
// watches: `time() - it > 2 * ron_every_secs` is "this mirror stopped",
// and it is true whether the timer died, the upstream went away or the
// cycle is wedged inside a connect — none of which systemd can see.
let mirrors = self.mirror_sources();
out.push_str("# HELP holger_mirror_cycles_total Mirror cycles per source, by outcome.\n");
out.push_str("# TYPE holger_mirror_cycles_total counter\n");
for m in &mirrors {
let (ok, failed, _, _, _) = m.counts();
let s = escape_label(m.name());
out.push_str(&format!(
"holger_mirror_cycles_total{{source=\"{s}\",result=\"ok\"}} {ok}\n"
));
out.push_str(&format!(
"holger_mirror_cycles_total{{source=\"{s}\",result=\"failed\"}} {failed}\n"
));
}
out.push_str(
"# HELP holger_mirror_artifacts_total Artifacts a mirror took (fetched) or could not take (failed), per source.\n",
);
out.push_str("# TYPE holger_mirror_artifacts_total counter\n");
for m in &mirrors {
let (_, _, fetched, failed, _) = m.counts();
let s = escape_label(m.name());
out.push_str(&format!(
"holger_mirror_artifacts_total{{source=\"{s}\",result=\"fetched\"}} {fetched}\n"
));
out.push_str(&format!(
"holger_mirror_artifacts_total{{source=\"{s}\",result=\"failed\"}} {failed}\n"
));
}
out.push_str("# HELP holger_mirror_bytes_total Bytes a mirror pulled from its upstream.\n");
out.push_str("# TYPE holger_mirror_bytes_total counter\n");
for m in &mirrors {
let (_, _, _, _, bytes) = m.counts();
out.push_str(&format!(
"holger_mirror_bytes_total{{source=\"{}\"}} {bytes}\n",
escape_label(m.name())
));
}
out.push_str(
"# HELP holger_mirror_last_success_timestamp_seconds Unix time of the last completed cycle; 0 means never.\n",
);
out.push_str("# TYPE holger_mirror_last_success_timestamp_seconds gauge\n");
for m in &mirrors {
out.push_str(&format!(
"holger_mirror_last_success_timestamp_seconds{{source=\"{}\"}} {}\n",
escape_label(m.name()),
m.last_success_unix().unwrap_or(0)
));
}
out.push_str(
"# HELP holger_mirror_consecutive_failures Failed cycles since this source last completed one.\n",
);
out.push_str("# TYPE holger_mirror_consecutive_failures gauge\n");
for m in &mirrors {
out.push_str(&format!(
"holger_mirror_consecutive_failures{{source=\"{}\"}} {}\n",
escape_label(m.name()),
m.consecutive_failures()
));
}
out.push_str("# HELP holger_repo_artifacts Current artifact count per repository.\n");
out.push_str("# TYPE holger_repo_artifacts gauge\n");
for (repo, count) in repo_counts {
// A repository we could not enumerate emits NO sample here. See the
// note on this function: `0` and "could not look" are different
// facts and the exposition format has a way to say the second one
// — leave the series absent and mark it below.
if let Some(n) = count {
out.push_str(&format!(
"holger_repo_artifacts{{repo=\"{}\"}} {}\n",
escape_label(repo),
n,
));
}
}
out.push_str(
"# HELP holger_repo_listable 1 when the repository answered a listing, 0 when it could not be enumerated.\n",
);
out.push_str("# TYPE holger_repo_listable gauge\n");
for (repo, count) in repo_counts {
out.push_str(&format!(
"holger_repo_listable{{repo=\"{}\"}} {}\n",
escape_label(repo),
u8::from(count.is_some()),
));
}
out
}
}
/// Render a histogram bucket upper bound as a Prometheus `le` value. The default
/// `f64` formatting already yields the canonical minimal form for these bounds
/// (`0.001`, `0.05`, `1`, `10`), which Prometheus parses unambiguously.
fn fmt_bucket_bound(bound: f64) -> String {
format!("{}", bound)
}
/// Escape a Prometheus label value: backslash, double-quote and newline are the
/// three characters the exposition format requires be escaped inside `"…"`.
fn escape_label(v: &str) -> String {
let mut out = String::with_capacity(v.len());
for c in v.chars() {
match c {
'\\' => out.push_str("\\\\"),
'"' => out.push_str("\\\""),
'\n' => out.push_str("\\n"),
_ => out.push(c),
}
}
out
}
/// The process-global metrics registry every live call site bumps and the
/// `GET /metrics` door renders. Lazily created on first use.
pub fn global() -> &'static Metrics {
static GLOBAL: OnceLock<Metrics> = OnceLock::new();
GLOBAL.get_or_init(Metrics::new)
}
#[cfg(test)]
mod tests {
use super::*;
/// After N recorded requests of a verb, that verb's counter reads exactly N —
/// and a DIFFERENT verb stays at 0 (the RED-when-broken guard: a bump wired to
/// the wrong slot, or dropped entirely, fails this).
#[test]
fn request_counter_counts_exactly_n_per_verb() {
let m = Metrics::new();
for _ in 0..7 {
m.record_request(Verb::Get, 200, 100);
}
for _ in 0..3 {
m.record_request(Verb::Put, 201, 0);
}
assert_eq!(m.request_count(Verb::Get), 7, "7 GETs must read 7");
assert_eq!(m.request_count(Verb::Put), 3, "3 PUTs must read 3");
assert_eq!(
m.request_count(Verb::Delete),
0,
"an untouched verb stays 0"
);
assert_eq!(m.bytes_served(), 700, "7×100 bytes served");
}
/// Only 5xx responses count as errors; 2xx/4xx do not.
#[test]
fn errors_count_only_5xx() {
let m = Metrics::new();
m.record_request(Verb::Get, 200, 10);
m.record_request(Verb::Get, 404, 0);
m.record_request(Verb::Get, 500, 0);
m.record_request(Verb::Sbom, 503, 0);
assert_eq!(m.errors(), 2, "only the 500 and 503 are errors");
}
/// The retention-run counter climbs one per recorded run.
#[test]
fn a_missed_audit_record_is_counted_and_published() {
// A28 M17: a lost audit record is a series an alert can watch.
let m = Metrics::new();
assert!(m
.render_prometheus(&[])
.contains("holger_audit_records_missed_total 0\n"));
m.record_audit_missed();
assert_eq!(m.audit_missed(), 1);
assert!(m
.render_prometheus(&[])
.contains("# TYPE holger_audit_records_missed_total counter\n"));
assert!(m
.render_prometheus(&[])
.contains("holger_audit_records_missed_total 1\n"));
}
#[test]
fn retention_runs_increment() {
let m = Metrics::new();
assert_eq!(m.retention_runs(), 0);
m.record_retention_run();
m.record_retention_run();
assert_eq!(m.retention_runs(), 2);
}
/// The rendered text is valid Prometheus exposition format AND reflects the
/// live counter state: after 2 GETs the `verb="get"` sample reads 2, and the
/// supplied repo counts appear as gauge samples. RED-when-broken: a dropped
/// bump would render `…{verb="get"} 0` and fail the exact-value assertion.
#[test]
fn render_is_parseable_prometheus_reflecting_state() {
let m = Metrics::new();
m.record_request(Verb::Get, 200, 512);
m.record_request(Verb::Get, 200, 512);
m.record_request(Verb::Search, 200, 30);
m.record_retention_run();
let text = m.render_prometheus(&[
("rust-dev".to_string(), Some(12)),
("maven-dev".to_string(), Some(3)),
]);
// Structure: every family has a HELP + TYPE line before its samples.
assert!(text.contains("# TYPE holger_requests_total counter"));
assert!(text.contains("# TYPE holger_repo_artifacts gauge"));
// A minimal exposition-format sanity parse: every non-comment, non-blank
// line is `metric_name value` with a numeric final token.
for line in text.lines() {
if line.is_empty() || line.starts_with('#') {
continue;
}
let value = line.rsplit(' ').next().expect("a value token");
assert!(
value.parse::<f64>().is_ok(),
"sample line must end in a number: {line:?}",
);
}
// State: exact live values are reflected.
assert!(
text.contains("holger_requests_total{verb=\"get\"} 2"),
"2 GETs must render as 2:\n{text}",
);
assert!(text.contains("holger_requests_total{verb=\"search\"} 1"));
assert!(
text.contains("holger_bytes_served_total 1054"),
"512+512+30"
);
assert!(text.contains("holger_retention_runs_total 1"));
assert!(text.contains("holger_repo_artifacts{repo=\"rust-dev\"} 12"));
assert!(text.contains("holger_repo_artifacts{repo=\"maven-dev\"} 3"));
// An untouched verb renders zero, not absent.
assert!(text.contains("holger_requests_total{verb=\"put\"} 0"));
}
/// RED-when-broken: the ONE audit-action→verb classifier that both doors
/// share. A read (download OR listing) counts as `get`; a store as `put`;
/// delete and promote are their own verbs; `Other` is not request traffic and
/// is NOT counted (`None`). A mis-wired arm — e.g. promote folded into put, or
/// `Other` counted — fails here.
#[test]
fn audit_action_maps_to_the_expected_verb() {
assert_eq!(
Verb::from_audit_action(AuditAction::Download),
Some(Verb::Get)
);
assert_eq!(
Verb::from_audit_action(AuditAction::List),
Some(Verb::Get),
"a listing is a read"
);
assert_eq!(
Verb::from_audit_action(AuditAction::Upload),
Some(Verb::Put)
);
assert_eq!(
Verb::from_audit_action(AuditAction::Delete),
Some(Verb::Delete)
);
assert_eq!(
Verb::from_audit_action(AuditAction::Promote),
Some(Verb::Promote),
"promote is its own verb, not folded into put",
);
assert_eq!(
Verb::from_audit_action(AuditAction::Other),
None,
"a non-request action must not be counted",
);
}
/// The gRPC-plane bump (recording via the audit action, exactly as the
/// `record_audit` choke point does) lands on the RIGHT verb and renders as a
/// distinct `promote` series — RED-when-broken: a promote miscounted as a get,
/// or a missing `promote` series, fails here. Mirrors what
/// `HolgerGrpc::record_audit` drives so the wiring is asserted end to end at
/// the metrics layer.
#[test]
fn recording_via_audit_action_counts_the_grpc_plane_by_verb() {
let m = Metrics::new();
// Simulate a gRPC RPC flow through the shared classifier.
for action in [
AuditAction::Download, // fetch → get
AuditAction::List, // search → get
AuditAction::Upload, // put → put
AuditAction::Promote, // promote→ promote
AuditAction::Other, // ignored
] {
if let Some(v) = Verb::from_audit_action(action) {
m.record_request(v, 200, 4);
}
}
assert_eq!(
m.request_count(Verb::Get),
2,
"download + listing both count as get"
);
assert_eq!(m.request_count(Verb::Put), 1);
assert_eq!(
m.request_count(Verb::Promote),
1,
"a promotion is counted under promote"
);
assert_eq!(m.request_count(Verb::Delete), 0, "no delete happened");
let text = m.render_prometheus(&[]);
assert!(
text.contains("holger_requests_total{verb=\"promote\"} 1"),
"the promote verb must render as its own series:\n{text}",
);
// The four ignored/absent-traffic slots still render (zero, not missing).
assert!(text.contains("holger_requests_total{verb=\"delete\"} 0"));
}
/// RED-when-broken: each proxy outcome lands on its OWN counter slot and
/// renders as a distinct `result=` series. A hit bumps only the matching hit
/// counter (not miss); a miss bumps only miss (not a hit) — a mis-wired
/// `proxy_slot` arm, or a dropped bump, fails one of the exact-value asserts.
#[test]
fn proxy_outcomes_count_and_render_by_result_label() {
let m = Metrics::new();
// Distinct counts per outcome so a swapped slot can't accidentally pass.
for _ in 0..4 {
m.record_proxy(ProxyOutcome::PrimaryHit);
}
for _ in 0..3 {
m.record_proxy(ProxyOutcome::UpstreamHit);
}
for _ in 0..2 {
m.record_proxy(ProxyOutcome::NegcacheHit);
}
m.record_proxy(ProxyOutcome::Miss);
for _ in 0..5 {
m.record_proxy(ProxyOutcome::UpstreamError);
}
// A hit bumps only its own slot; a miss bumps only miss.
assert_eq!(m.proxy_count(ProxyOutcome::PrimaryHit), 4, "4 primary hits");
assert_eq!(
m.proxy_count(ProxyOutcome::UpstreamHit),
3,
"3 upstream hits"
);
assert_eq!(
m.proxy_count(ProxyOutcome::NegcacheHit),
2,
"2 negcache hits"
);
assert_eq!(m.proxy_count(ProxyOutcome::Miss), 1, "1 miss");
assert_eq!(
m.proxy_count(ProxyOutcome::UpstreamError),
5,
"5 skipped dead upstreams"
);
// RED-when-broken cross-check: a primary hit must NOT have bumped miss.
assert_ne!(
m.proxy_count(ProxyOutcome::Miss),
m.proxy_count(ProxyOutcome::PrimaryHit),
"a hit must not be counted as a miss (mis-wired slot)",
);
let text = m.render_prometheus(&[]);
assert!(text.contains("# TYPE holger_proxy_requests_total counter"));
assert!(
text.contains("holger_proxy_requests_total{result=\"primary_hit\"} 4"),
"primary_hit renders as its own series:\n{text}",
);
assert!(text.contains("holger_proxy_requests_total{result=\"upstream_hit\"} 3"));
assert!(text.contains("holger_proxy_requests_total{result=\"negcache_hit\"} 2"));
assert!(text.contains("holger_proxy_requests_total{result=\"miss\"} 1"));
assert!(
text.contains("holger_proxy_requests_total{result=\"upstream_error\"} 5"),
"a skipped dead upstream must be visible in /metrics, not silently tolerated:\n{text}",
);
}
/// RED-when-broken: the request-duration histogram counts every observation,
/// sums the durations, and renders CUMULATIVE `le` buckets that are monotone
/// non-decreasing and whose `+Inf` bucket equals `_count`. A dropped `observe`,
/// a mis-placed observation (wrong bucket), or a broken cumulation fails an
/// exact assertion here.
#[test]
fn duration_histogram_buckets_sum_and_count() {
let m = Metrics::new();
// Observations chosen to land in distinct buckets: 2ms→le=0.005, 30ms→le=0.05,
// 300ms→le=0.5, and 12s→the +Inf overflow (above the 10s last bound).
m.observe_request_duration(0.002);
m.observe_request_duration(0.030);
m.observe_request_duration(0.300);
m.observe_request_duration(12.0);
assert_eq!(m.request_duration_count(), 4, "4 observations counted");
let sum = m.request_duration_sum_seconds();
assert!(
(sum - 12.332).abs() < 1e-6,
"sum of durations must be 12.332, got {sum}"
);
let text = m.render_prometheus(&[]);
assert!(text.contains("# TYPE holger_request_duration_seconds histogram"));
// le=0.001 excludes the 2ms obs; le=0.005 includes it (cumulative 1).
assert!(
text.contains("holger_request_duration_seconds_bucket{le=\"0.001\"} 0"),
"nothing ≤1ms:\n{text}",
);
assert!(
text.contains("holger_request_duration_seconds_bucket{le=\"0.005\"} 1"),
"the 2ms obs falls in ≤5ms:\n{text}",
);
// le=0.05 now also includes the 30ms obs (cumulative 2).
assert!(text.contains("holger_request_duration_seconds_bucket{le=\"0.05\"} 2"));
// le=0.5 also includes the 300ms obs (cumulative 3); the 12s obs is NOT here.
assert!(text.contains("holger_request_duration_seconds_bucket{le=\"0.5\"} 3"));
// The last finite bound (10s) still excludes the 12s obs (cumulative 3).
assert!(
text.contains("holger_request_duration_seconds_bucket{le=\"10\"} 3"),
"the 12s obs is above the 10s bound:\n{text}",
);
// +Inf includes everything and equals _count.
assert!(
text.contains("holger_request_duration_seconds_bucket{le=\"+Inf\"} 4"),
"+Inf must equal the total count:\n{text}",
);
assert!(text.contains("holger_request_duration_seconds_count 4"));
// A non-finite/negative observation is ignored (no corruption).
m.observe_request_duration(f64::NAN);
m.observe_request_duration(-1.0);
assert_eq!(
m.request_duration_count(),
4,
"NaN/negative are not counted"
);
}
/// Label values with special characters are escaped so the output stays
/// parseable even for an oddly-named repo.
#[test]
fn label_values_are_escaped() {
assert_eq!(escape_label("plain"), "plain");
assert_eq!(escape_label("a\"b\\c"), "a\\\"b\\\\c");
let m = Metrics::new();
let text = m.render_prometheus(&[("we\"ird".to_string(), Some(1))]);
assert!(text.contains("holger_repo_artifacts{repo=\"we\\\"ird\"} 1"));
}
/// A repository that could not be enumerated must not be reported as zero.
/// Before this, `list()` errors were folded to `0` at the call site, so an
/// unreadable archive and an empty repository produced the identical
/// scrape — and `sum(holger_repo_artifacts)` under-counted without saying so.
#[test]
fn an_unreadable_repository_is_absent_not_zero() {
let m = Metrics::new();
let text = m.render_prometheus(&[
("good".to_string(), Some(7)),
("empty".to_string(), Some(0)),
("broken".to_string(), None),
]);
assert!(text.contains("holger_repo_artifacts{repo=\"good\"} 7"));
assert!(text.contains("holger_repo_artifacts{repo=\"empty\"} 0"));
assert!(
!text.contains("holger_repo_artifacts{repo=\"broken\"}"),
"an absent reading emits no count sample at all"
);
assert!(text.contains("holger_repo_listable{repo=\"good\"} 1"));
assert!(text.contains("holger_repo_listable{repo=\"empty\"} 1"));
assert!(
text.contains("holger_repo_listable{repo=\"broken\"} 0"),
"…and says so on its own series instead"
);
}
}