kevy 6.3.0

kevy — a pure-Rust, zero-dependency, Redis-compatible KV server.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169

//! Prometheus `/metrics` HTTP exposition endpoint.
//!
//! Std-only tiny HTTP/1.1 server (0-dep, no `hyper`). One background
//! thread per `serve()` call; accepts conns serially (scrapers are
//! polling, low-rate). Emits `text/plain; version=0.0.4` Prometheus
//! exposition format on `GET /metrics`. Anything else returns 404.
//!
//! Metric source: the shared [`RuntimeState`] — live config values
//! plus the aggregated per-shard stats. The thread holds only a
//! `Weak<RuntimeState>` and exits when the state is gone, so a
//! finished `serve` never leaves an immortal scraper thread pinning
//! the whole state alive.

use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::{Arc, Weak};
use std::time::Instant;

use crate::state::RuntimeState;

/// Spawn the metrics HTTP listener if `metrics.listen_port > 0`.
/// Returns immediately; the listener runs on a daemon thread.
pub fn spawn_if_enabled(state: &Arc<RuntimeState>) {
    let port = state.config().metrics.listen_port;
    if port == 0 {
        return;
    }
    let weak = Arc::downgrade(state);
    std::thread::Builder::new()
        .name("kevy-metrics".into())
        .spawn(move || run_listener(port, weak))
        .expect("spawn kevy-metrics thread");
}

fn run_listener(port: u16, state: Weak<RuntimeState>) {
    let addr = format!("127.0.0.1:{port}");
    let listener = match TcpListener::bind(&addr) {
        Ok(l) => l,
        Err(e) => {
            eprintln!("kevy: metrics endpoint failed to bind {addr}: {e}");
            return;
        }
    };
    eprintln!("kevy: metrics endpoint listening on http://{addr}/metrics");
    let start = Instant::now();
    loop {
        let (conn, _peer) = match listener.accept() {
            Ok(t) => t,
            Err(_) => continue,
        };
        // The runtime state owns this thread's reason to exist; once
        // it is dropped the scrape data is gone, so stop serving.
        let Some(state) = state.upgrade() else { return };
        handle_scrape_conn(conn, &state, start);
    }
}

/// Serve one scrape connection: read the request, answer `GET
/// /metrics` with the exposition body, anything else with a 404.
fn handle_scrape_conn(mut conn: TcpStream, state: &RuntimeState, start: Instant) {
    let _ = conn.set_read_timeout(Some(std::time::Duration::from_secs(2)));
    let _ = conn.set_write_timeout(Some(std::time::Duration::from_secs(2)));
    // Minimal HTTP/1.1 request parse: read up to the request-line
    // + drain to end-of-headers (\r\n\r\n).
    let mut buf = [0u8; 1024];
    let n = match conn.read(&mut buf) {
        Ok(n) => n,
        Err(_) => return,
    };
    let raw = &buf[..n];
    // We only care about the verb + path; any GET /metrics passes.
    let is_metrics = raw.starts_with(b"GET /metrics");
    let body =
        if is_metrics { render_metrics(state, start.elapsed().as_secs()) } else { String::new() };
    let resp = if is_metrics {
        format!(
            "HTTP/1.1 200 OK\r\n\
             Content-Type: text/plain; version=0.0.4\r\n\
             Content-Length: {}\r\n\
             Connection: close\r\n\
             \r\n{}",
            body.len(),
            body,
        )
    } else {
        "HTTP/1.1 404 Not Found\r\n\
         Content-Length: 0\r\n\
         Connection: close\r\n\
         \r\n"
            .to_string()
    };
    let _ = conn.write_all(resp.as_bytes());
}

/// Produce the Prometheus exposition body: aggregated shard stats plus
/// live config values like max_clients.
// LOC-WAIVER: flat metric-emission table — one help/type/value triplet per gauge.
fn render_metrics(state: &RuntimeState, uptime_seconds: u64) -> String {
    let mut out = String::with_capacity(4 * 1024);
    let cfg = state.config();
    let totals = state.obs.aggregate();

    // Uptime
    push_help(&mut out, "kevy_uptime_seconds", "Seconds since kevy started");
    push_type(&mut out, "kevy_uptime_seconds", "counter");
    push_value(&mut out, "kevy_uptime_seconds", uptime_seconds);

    // Clients
    push_help(&mut out, "kevy_maxclients", "Configured max client connections");
    push_type(&mut out, "kevy_maxclients", "gauge");
    push_value(&mut out, "kevy_maxclients", cfg.server.max_clients as u64);

    // Memory
    push_help(&mut out, "kevy_used_memory_bytes", "Resident keyspace memory");
    push_type(&mut out, "kevy_used_memory_bytes", "gauge");
    push_value(&mut out, "kevy_used_memory_bytes", totals.used_memory);

    push_help(&mut out, "kevy_used_memory_peak_bytes", "Peak resident keyspace memory");
    push_type(&mut out, "kevy_used_memory_peak_bytes", "gauge");
    push_value(&mut out, "kevy_used_memory_peak_bytes", totals.used_memory_peak);

    push_help(&mut out, "kevy_maxmemory_bytes", "Configured maxmemory ceiling (0 = unlimited)");
    push_type(&mut out, "kevy_maxmemory_bytes", "gauge");
    push_value(&mut out, "kevy_maxmemory_bytes", cfg.memory.maxmemory);

    push_help(&mut out, "kevy_evicted_keys_total", "Keys evicted due to maxmemory");
    push_type(&mut out, "kevy_evicted_keys_total", "counter");
    push_value(&mut out, "kevy_evicted_keys_total", totals.evicted_keys);

    push_help(&mut out, "kevy_expired_keys_total", "Keys expired due to TTL");
    push_type(&mut out, "kevy_expired_keys_total", "counter");
    push_value(&mut out, "kevy_expired_keys_total", totals.expired_keys);

    push_help(&mut out, "kevy_keys_total", "Number of keys across all shards");
    push_type(&mut out, "kevy_keys_total", "gauge");
    push_value(&mut out, "kevy_keys_total", totals.keys);

    push_help(&mut out, "kevy_expires_total", "Number of keys with TTL");
    push_type(&mut out, "kevy_expires_total", "gauge");
    push_value(&mut out, "kevy_expires_total", totals.expires);

    // Build info — useful for ops to see which kevy is running.
    push_help(&mut out, "kevy_build_info", "kevy version (always 1)");
    push_type(&mut out, "kevy_build_info", "gauge");
    out.push_str(&format!("kevy_build_info{{version=\"{}\"}} 1\n", env!("CARGO_PKG_VERSION")));

    out
}

fn push_help(out: &mut String, name: &str, help: &str) {
    out.push_str("# HELP ");
    out.push_str(name);
    out.push(' ');
    out.push_str(help);
    out.push('\n');
}
fn push_type(out: &mut String, name: &str, ty: &str) {
    out.push_str("# TYPE ");
    out.push_str(name);
    out.push(' ');
    out.push_str(ty);
    out.push('\n');
}
fn push_value(out: &mut String, name: &str, value: u64) {
    out.push_str(name);
    out.push(' ');
    out.push_str(&value.to_string());
    out.push('\n');
}