use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::OnceLock;
pub const TARGET_AUDIT: &str = "gunnar::audit";
pub const TARGET_SSH: &str = "gunnar::ssh";
pub const TARGET_GIT: &str = "gunnar::git";
pub const TARGET_POLICY: &str = "gunnar::policy";
pub const TARGET_STORE: &str = "gunnar::store";
pub const TARGET_METRICS: &str = "gunnar::metrics";
pub const TARGET_LFS: &str = "gunnar::lfs";
pub const TARGET_RUNNER: &str = "gunnar::runner";
pub const TARGET_ARCHIVE: &str = "gunnar::archive";
pub const TARGET_REPLICATION: &str = "gunnar::replication";
pub const TARGET_LIMITS: &str = "gunnar::limits";
pub const TARGET_INDEX: &str = "gunnar::index";
pub const TARGET_ENTITLEMENT: &str = "gunnar::entitlement";
pub const TARGET_TWIN: &str = "gunnar::twin";
pub const TARGET_SPACE: &str = "gunnar::space";
pub const TARGET_REPAIR: &str = "gunnar::repair";
pub const TARGET_LIVE: &str = "gunnar::live";
pub const TARGET_ACTIVITY: &str = "gunnar::activity";
pub const TARGETS: &[&str] = &[
TARGET_AUDIT,
TARGET_SSH,
TARGET_GIT,
TARGET_POLICY,
TARGET_STORE,
TARGET_METRICS,
TARGET_LFS,
TARGET_RUNNER,
TARGET_ARCHIVE,
TARGET_REPLICATION,
TARGET_LIMITS,
TARGET_INDEX,
TARGET_ENTITLEMENT,
TARGET_TWIN,
TARGET_SPACE,
TARGET_REPAIR,
TARGET_LIVE,
TARGET_ACTIVITY,
];
pub fn log_filter(base: &str, gunnar: &str) -> String {
let mut spec = String::with_capacity(base.len() + TARGETS.len() * 24);
spec.push_str(base);
for target in TARGETS {
spec.push(',');
spec.push_str(target);
spec.push('=');
spec.push_str(gunnar);
}
spec
}
pub const SERVICE_LOG_BASE: &str = "info";
pub const SERVICE_LOG_GUNNAR: &str = "debug";
pub fn service_log_filter() -> String {
log_filter(SERVICE_LOG_BASE, SERVICE_LOG_GUNNAR)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct RequestId(u64);
static NEXT_REQUEST_ID: AtomicU64 = AtomicU64::new(1);
fn process_seed() -> u64 {
static SEED: OnceLock<u64> = OnceLock::new();
*SEED.get_or_init(|| {
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos() as u64)
.unwrap_or(0);
(nanos ^ ((std::process::id() as u64) << 40)) & 0xffff_ffff
})
}
impl RequestId {
pub fn next() -> RequestId {
let n = NEXT_REQUEST_ID.fetch_add(1, Ordering::Relaxed);
RequestId((process_seed() << 32) | (n & 0xffff_ffff))
}
pub fn get(&self) -> u64 {
self.0
}
}
impl std::fmt::Display for RequestId {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{:016x}", self.0)
}
}
#[derive(Debug, Default)]
pub struct ServerMetrics {
connections_opened: AtomicU64,
connections_accepted: AtomicU64,
connections_anonymous: AtomicU64,
connections_refused: AtomicU64,
auth_faults: AtomicU64,
exec_refused: AtomicU64,
transfers_started: AtomicU64,
transfers_in_flight: AtomicU64,
transfers_failed: AtomicU64,
packs_served: AtomicU64,
objects_sent: AtomicU64,
bytes_out: AtomicU64,
bytes_in: AtomicU64,
pushes_accepted: AtomicU64,
pushes_rejected: AtomicU64,
refs_updated: AtomicU64,
refs_deleted: AtomicU64,
refs_rejected: AtomicU64,
gate_refusals: AtomicU64,
usage_reconcile_missed: AtomicU64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct MetricsSnapshot {
pub connections_opened: u64,
pub connections_accepted: u64,
pub connections_anonymous: u64,
pub connections_refused: u64,
pub auth_faults: u64,
pub exec_refused: u64,
pub transfers_started: u64,
pub transfers_in_flight: u64,
pub transfers_failed: u64,
pub packs_served: u64,
pub objects_sent: u64,
pub bytes_out: u64,
pub bytes_in: u64,
pub pushes_accepted: u64,
pub pushes_rejected: u64,
pub refs_updated: u64,
pub refs_deleted: u64,
pub refs_rejected: u64,
pub gate_refusals: u64,
pub usage_reconcile_missed: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MetricKind {
Counter,
Gauge,
}
impl MetricKind {
pub fn as_str(self) -> &'static str {
match self {
MetricKind::Counter => "counter",
MetricKind::Gauge => "gauge",
}
}
}
macro_rules! metric_kind {
(counter) => {
MetricKind::Counter
};
(gauge) => {
MetricKind::Gauge
};
}
macro_rules! counters {
($($kind:ident $field:ident => $add:ident),* $(,)?) => {
impl ServerMetrics {
$(
#[doc = concat!("Add `n` to `", stringify!($field), "`.")]
pub fn $add(&self, n: u64) {
self.$field.fetch_add(n, Ordering::Relaxed);
}
)*
pub fn snapshot(&self) -> MetricsSnapshot {
MetricsSnapshot {
$($field: self.$field.load(Ordering::Relaxed),)*
}
}
}
impl MetricsSnapshot {
pub fn fields(&self) -> Vec<(&'static str, MetricKind, u64)> {
vec![$((stringify!($field), metric_kind!($kind), self.$field),)*]
}
pub fn emit_heartbeat(&self) {
tracing::info!(
target: TARGET_METRICS,
event = "metrics",
$($field = self.$field,)*
"gunnar"
);
}
}
};
}
counters! {
counter connections_opened => add_connections_opened,
counter connections_accepted => add_connections_accepted,
counter connections_anonymous => add_connections_anonymous,
counter connections_refused => add_connections_refused,
counter auth_faults => add_auth_faults,
counter exec_refused => add_exec_refused,
counter transfers_started => add_transfers_started,
gauge transfers_in_flight => add_transfers_in_flight,
counter transfers_failed => add_transfers_failed,
counter packs_served => add_packs_served,
counter objects_sent => add_objects_sent,
counter bytes_out => add_bytes_out,
counter bytes_in => add_bytes_in,
counter pushes_accepted => add_pushes_accepted,
counter pushes_rejected => add_pushes_rejected,
counter refs_updated => add_refs_updated,
counter refs_deleted => add_refs_deleted,
counter refs_rejected => add_refs_rejected,
counter gate_refusals => add_gate_refusals,
counter usage_reconcile_missed => add_usage_reconcile_missed,
}
impl ServerMetrics {
pub fn new() -> ServerMetrics {
ServerMetrics::default()
}
pub fn transfer_started(self: &std::sync::Arc<Self>) -> InFlightGuard {
self.transfers_started.fetch_add(1, Ordering::Relaxed);
self.transfers_in_flight.fetch_add(1, Ordering::Relaxed);
InFlightGuard {
metrics: std::sync::Arc::clone(self),
}
}
}
#[derive(Debug)]
pub struct InFlightGuard {
metrics: std::sync::Arc<ServerMetrics>,
}
impl Drop for InFlightGuard {
fn drop(&mut self) {
self.metrics
.transfers_in_flight
.fetch_sub(1, Ordering::Relaxed);
}
}
impl MetricsSnapshot {
pub fn is_zero(&self) -> bool {
*self == MetricsSnapshot::default()
}
}
pub const SCRAPE_PREFIX: &str = "gunnar_";
pub const SCRAPE_CONTENT_TYPE: &str = "text/plain; version=0.0.4; charset=utf-8";
impl MetricsSnapshot {
pub fn render_prometheus(&self) -> String {
let mut out = String::with_capacity(self.fields().len() * 64);
for (name, kind, value) in self.fields() {
out.push_str("# TYPE ");
out.push_str(SCRAPE_PREFIX);
out.push_str(name);
out.push(' ');
out.push_str(kind.as_str());
out.push('\n');
out.push_str(SCRAPE_PREFIX);
out.push_str(name);
out.push(' ');
{
use std::fmt::Write as _;
let _ = write!(out, "{value}");
}
out.push('\n');
}
out
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Arc;
#[test]
fn every_target_is_a_gunnar_sibling_and_the_list_is_complete() {
for target in TARGETS {
assert!(
target.starts_with("gunnar::"),
"{target} is not a sibling of {TARGET_AUDIT}"
);
assert_eq!(
target.matches("::").count(),
1,
"{target} is not one level deep"
);
}
let mut sorted = TARGETS.to_vec();
sorted.sort_unstable();
sorted.dedup();
assert_eq!(sorted.len(), TARGETS.len(), "TARGETS holds a duplicate");
assert!(TARGETS.contains(&TARGET_AUDIT));
}
#[test]
fn the_service_filter_is_derived_from_the_target_list() {
let filter = service_log_filter();
for target in TARGETS {
let directive = format!("{target}={SERVICE_LOG_GUNNAR}");
assert!(
filter.contains(&directive),
"{directive} is missing from {filter}: a target added to TARGETS did not \
reach the filter a service is installed with"
);
}
assert!(
filter.starts_with(SERVICE_LOG_BASE),
"the filter must open with the base level: {filter}"
);
assert_ne!(
SERVICE_LOG_BASE, SERVICE_LOG_GUNNAR,
"with one level for both, this filter says nothing a bare level does not, and \
the test above cannot tell a derived filter from a blanket one"
);
assert_eq!(
filter.matches(',').count(),
TARGETS.len(),
"the filter holds a directive that is not one of ours: {filter}"
);
}
#[test]
fn request_ids_do_not_repeat() {
let a = RequestId::next();
let b = RequestId::next();
assert_ne!(a, b, "two ids from the same process collided");
assert_ne!(
a.to_string(),
b.to_string(),
"two ids RENDERED the same, which is what a log correlation sees"
);
assert_eq!(a.to_string().len(), 16, "the id must be fixed-width");
assert!(
a.to_string().chars().all(|c| c.is_ascii_hexdigit()),
"the id must be greppable"
);
assert_eq!(a.get() >> 32, b.get() >> 32);
}
#[test]
fn a_counter_moves_and_the_gauge_comes_back_down() {
let metrics = Arc::new(ServerMetrics::new());
let before = metrics.snapshot();
assert!(before.is_zero(), "a fresh set of counters must read zero");
metrics.add_bytes_out(4096);
metrics.add_objects_sent(7);
let guard = metrics.transfer_started();
let during = metrics.snapshot();
assert_eq!(during.bytes_out - before.bytes_out, 4096);
assert_eq!(during.objects_sent - before.objects_sent, 7);
assert_eq!(during.transfers_started, 1);
assert_eq!(during.transfers_in_flight, 1, "the gauge must be up");
drop(guard);
let after = metrics.snapshot();
assert_eq!(
after.transfers_in_flight, 0,
"the in-flight gauge leaked; a server that never comes back to zero \
reads as permanently saturated"
);
assert_eq!(
after.transfers_started, 1,
"the total is a total, not a gauge"
);
}
#[test]
fn every_field_of_the_snapshot_is_published() {
let snapshot = MetricsSnapshot::default();
let published: Vec<&str> = snapshot.fields().iter().map(|(name, _, _)| *name).collect();
let debug = format!("{snapshot:?}");
let declared: Vec<String> = debug
.trim_start_matches("MetricsSnapshot {")
.trim_end_matches('}')
.split(',')
.filter_map(|pair| pair.split_once(':'))
.map(|(name, _)| name.trim().to_owned())
.collect();
assert_eq!(
declared.len(),
20,
"the Debug parse found {} fields, not the 20 this struct has, so this guard is \
reading nothing and would pass against an empty `fields()`: {debug}",
declared.len()
);
for field in &declared {
assert!(
published.contains(&field.as_str()),
"`{field}` is a counter of this server that NOTHING publishes: not the \
heartbeat, not the scrape endpoint. That is exactly how \
`connections_anonymous` stayed invisible."
);
}
assert_eq!(
published.len(),
declared.len(),
"fields() publishes {} names for {} struct fields",
published.len(),
declared.len()
);
}
#[test]
fn the_scrape_body_names_every_counter_with_its_kind_and_value() {
let metrics = Arc::new(ServerMetrics::new());
metrics.add_bytes_out(4096);
metrics.add_connections_anonymous(3);
let _guard = metrics.transfer_started();
let body = metrics.snapshot().render_prometheus();
assert!(
body.contains("\ngunnar_bytes_out 4096\n"),
"the applied value is not in the body:\n{body}"
);
assert!(
body.contains("\ngunnar_connections_anonymous 3\n"),
"the counter no line ever carried is still not carried:\n{body}"
);
assert!(
body.contains("# TYPE gunnar_transfers_in_flight gauge\n"),
"the in-flight gauge is not typed as a gauge:\n{body}"
);
assert!(
body.contains("# TYPE gunnar_bytes_out counter\n"),
"a total is not typed as a counter:\n{body}"
);
let mut samples = 0;
for line in body.lines() {
if let Some(spec) = line.strip_prefix("# TYPE ") {
let (name, kind) = spec.split_once(' ').expect("a TYPE line is `name kind`");
assert!(name.starts_with(SCRAPE_PREFIX), "{name} is unprefixed");
assert!(kind == "counter" || kind == "gauge", "{kind} is not a kind");
continue;
}
assert!(line.starts_with(SCRAPE_PREFIX), "stray line: {line:?}");
let (name, value) = line.split_once(' ').expect("a sample is `name value`");
value
.parse::<u64>()
.unwrap_or_else(|_| panic!("{name} rendered {value:?}, which is not a number"));
assert!(
body.contains(&format!("# TYPE {name} ")),
"{name} is a sample with no TYPE line"
);
samples += 1;
}
assert_eq!(
samples,
metrics.snapshot().fields().len(),
"the body carries {samples} samples for {} counters",
metrics.snapshot().fields().len()
);
}
#[test]
fn counters_survive_concurrent_writers() {
let metrics = Arc::new(ServerMetrics::new());
let mut threads = Vec::new();
for _ in 0..8 {
let metrics = Arc::clone(&metrics);
threads.push(std::thread::spawn(move || {
for _ in 0..1000 {
metrics.add_bytes_out(1);
let _guard = metrics.transfer_started();
}
}));
}
for t in threads {
t.join().expect("a counting thread");
}
let snapshot = metrics.snapshot();
assert_eq!(snapshot.bytes_out, 8000);
assert_eq!(snapshot.transfers_started, 8000);
assert_eq!(snapshot.transfers_in_flight, 0);
}
}