357 lines
12 KiB
Rust
357 lines
12 KiB
Rust
use std::path::{Path, PathBuf};
|
|
use std::sync::Mutex;
|
|
|
|
use anyhow::{Context, Result};
|
|
use rusqlite::{params, Connection, OptionalExtension};
|
|
|
|
use crate::metrics::Sample;
|
|
|
|
pub struct Db {
|
|
path: PathBuf,
|
|
pub(crate) conn: Mutex<Connection>,
|
|
}
|
|
|
|
#[derive(Debug, Clone)]
|
|
pub struct CleanupReport {
|
|
pub deleted_by_age: usize,
|
|
pub deleted_by_size: usize,
|
|
pub vacuumed: bool,
|
|
pub bytes_before: u64,
|
|
pub bytes_after: u64,
|
|
pub rows_after: i64,
|
|
}
|
|
|
|
impl Db {
|
|
pub fn open(path: &Path) -> Result<Self> {
|
|
if let Some(parent) = path.parent() {
|
|
if !parent.as_os_str().is_empty() {
|
|
std::fs::create_dir_all(parent)?;
|
|
}
|
|
}
|
|
let conn = Connection::open(path).with_context(|| format!("open db {}", path.display()))?;
|
|
conn.execute_batch(
|
|
"
|
|
PRAGMA journal_mode = WAL;
|
|
PRAGMA synchronous = NORMAL;
|
|
PRAGMA temp_store = MEMORY;
|
|
PRAGMA foreign_keys = ON;
|
|
",
|
|
)?;
|
|
Ok(Self {
|
|
path: path.to_path_buf(),
|
|
conn: Mutex::new(conn),
|
|
})
|
|
}
|
|
|
|
pub fn path(&self) -> &Path {
|
|
&self.path
|
|
}
|
|
|
|
pub fn migrate(&self) -> Result<()> {
|
|
let conn = self.conn.lock().unwrap();
|
|
conn.execute_batch(
|
|
"
|
|
CREATE TABLE IF NOT EXISTS samples (
|
|
ts INTEGER PRIMARY KEY,
|
|
cpu_pct REAL NOT NULL,
|
|
mem_used_bytes INTEGER NOT NULL,
|
|
mem_total_bytes INTEGER NOT NULL,
|
|
mem_pct REAL NOT NULL,
|
|
disk_used_bytes INTEGER NOT NULL DEFAULT 0,
|
|
disk_total_bytes INTEGER NOT NULL DEFAULT 0,
|
|
disk_pct REAL NOT NULL DEFAULT 0,
|
|
net_rx_bps REAL NOT NULL,
|
|
net_tx_bps REAL NOT NULL,
|
|
load1 REAL NOT NULL,
|
|
load5 REAL NOT NULL,
|
|
load15 REAL NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_samples_ts ON samples(ts);
|
|
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
username TEXT NOT NULL UNIQUE COLLATE NOCASE,
|
|
password_hash TEXT NOT NULL,
|
|
created_at INTEGER NOT NULL
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS sessions (
|
|
token_hash TEXT PRIMARY KEY,
|
|
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
|
created_at INTEGER NOT NULL,
|
|
expires_at INTEGER NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_sessions_user ON sessions(user_id);
|
|
CREATE INDEX IF NOT EXISTS idx_sessions_expires ON sessions(expires_at);
|
|
|
|
CREATE TABLE IF NOT EXISTS health_sites (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL,
|
|
url TEXT NOT NULL,
|
|
enabled INTEGER NOT NULL DEFAULT 1,
|
|
interval_secs INTEGER NOT NULL DEFAULT 60,
|
|
timeout_ms INTEGER NOT NULL DEFAULT 10000,
|
|
expected_status INTEGER NOT NULL DEFAULT 200,
|
|
expect_body_contains TEXT,
|
|
created_at INTEGER NOT NULL,
|
|
updated_at INTEGER NOT NULL,
|
|
last_check_at INTEGER,
|
|
last_ok INTEGER,
|
|
last_status INTEGER,
|
|
last_latency_ms INTEGER,
|
|
last_error TEXT,
|
|
last_via_cloudflare INTEGER,
|
|
last_cf_fallback INTEGER,
|
|
last_cf_challenge INTEGER
|
|
);
|
|
CREATE UNIQUE INDEX IF NOT EXISTS idx_health_sites_url ON health_sites(url);
|
|
|
|
CREATE TABLE IF NOT EXISTS health_results (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
site_id INTEGER NOT NULL REFERENCES health_sites(id) ON DELETE CASCADE,
|
|
checked_at INTEGER NOT NULL,
|
|
ok INTEGER NOT NULL,
|
|
status INTEGER NOT NULL,
|
|
latency_ms INTEGER NOT NULL,
|
|
via_cloudflare INTEGER NOT NULL,
|
|
cf_fallback INTEGER NOT NULL,
|
|
cf_challenge INTEGER NOT NULL,
|
|
error TEXT,
|
|
reasons_json TEXT NOT NULL DEFAULT '[]'
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_health_results_site_ts ON health_results(site_id, checked_at DESC);
|
|
|
|
CREATE TABLE IF NOT EXISTS settings (
|
|
key TEXT PRIMARY KEY,
|
|
value TEXT NOT NULL
|
|
);
|
|
",
|
|
)?;
|
|
// Additive migrations for samples disk columns (existing DBs).
|
|
let _ = conn.execute(
|
|
"ALTER TABLE samples ADD COLUMN disk_used_bytes INTEGER NOT NULL DEFAULT 0",
|
|
[],
|
|
);
|
|
let _ = conn.execute(
|
|
"ALTER TABLE samples ADD COLUMN disk_total_bytes INTEGER NOT NULL DEFAULT 0",
|
|
[],
|
|
);
|
|
let _ = conn.execute(
|
|
"ALTER TABLE samples ADD COLUMN disk_pct REAL NOT NULL DEFAULT 0",
|
|
[],
|
|
);
|
|
Ok(())
|
|
}
|
|
|
|
pub fn insert_sample(&self, s: &Sample) -> Result<()> {
|
|
let conn = self.conn.lock().unwrap();
|
|
conn.execute(
|
|
"INSERT OR REPLACE INTO samples
|
|
(ts, cpu_pct, mem_used_bytes, mem_total_bytes, mem_pct,
|
|
disk_used_bytes, disk_total_bytes, disk_pct,
|
|
net_rx_bps, net_tx_bps, load1, load5, load15)
|
|
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
|
|
params![
|
|
s.ts,
|
|
s.cpu_pct,
|
|
s.mem_used_bytes as i64,
|
|
s.mem_total_bytes as i64,
|
|
s.mem_pct,
|
|
s.disk_used_bytes as i64,
|
|
s.disk_total_bytes as i64,
|
|
s.disk_pct,
|
|
s.net_rx_bps,
|
|
s.net_tx_bps,
|
|
s.load1,
|
|
s.load5,
|
|
s.load15,
|
|
],
|
|
)?;
|
|
Ok(())
|
|
}
|
|
|
|
pub fn prune_older_than(&self, older_than_ts: i64) -> Result<usize> {
|
|
let conn = self.conn.lock().unwrap();
|
|
let n = conn.execute("DELETE FROM samples WHERE ts < ?1", params![older_than_ts])?;
|
|
Ok(n)
|
|
}
|
|
|
|
/// Delete the oldest `limit` rows. Returns how many were removed.
|
|
pub fn prune_oldest(&self, limit: i64) -> Result<usize> {
|
|
if limit <= 0 {
|
|
return Ok(0);
|
|
}
|
|
let conn = self.conn.lock().unwrap();
|
|
let n = conn.execute(
|
|
"DELETE FROM samples WHERE ts IN (
|
|
SELECT ts FROM samples ORDER BY ts ASC LIMIT ?1
|
|
)",
|
|
params![limit],
|
|
)?;
|
|
Ok(n)
|
|
}
|
|
|
|
pub fn checkpoint_truncate(&self) -> Result<()> {
|
|
let conn = self.conn.lock().unwrap();
|
|
conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE);")?;
|
|
Ok(())
|
|
}
|
|
|
|
pub fn vacuum(&self) -> Result<()> {
|
|
// VACUUM cannot run inside a transaction; unlock and run dedicated connection.
|
|
{
|
|
let conn = self.conn.lock().unwrap();
|
|
conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE);")?;
|
|
}
|
|
let conn = Connection::open(&self.path)?;
|
|
conn.execute_batch("VACUUM;")?;
|
|
// Re-apply WAL on the long-lived connection
|
|
let conn = self.conn.lock().unwrap();
|
|
conn.execute_batch("PRAGMA journal_mode = WAL;")?;
|
|
Ok(())
|
|
}
|
|
|
|
/// On-disk size of main DB + WAL + SHM.
|
|
pub fn on_disk_bytes(&self) -> Result<u64> {
|
|
Ok(file_size(&self.path)?
|
|
+ file_size(Path::new(&format!("{}-wal", self.path.display())))?
|
|
+ file_size(Path::new(&format!("{}-shm", self.path.display())))?)
|
|
}
|
|
|
|
pub fn latest(&self) -> Result<Option<Sample>> {
|
|
let conn = self.conn.lock().unwrap();
|
|
conn.query_row(
|
|
"SELECT ts, cpu_pct, mem_used_bytes, mem_total_bytes, mem_pct,
|
|
disk_used_bytes, disk_total_bytes, disk_pct,
|
|
net_rx_bps, net_tx_bps, load1, load5, load15
|
|
FROM samples ORDER BY ts DESC LIMIT 1",
|
|
[],
|
|
row_to_sample,
|
|
)
|
|
.optional()
|
|
.map_err(Into::into)
|
|
}
|
|
|
|
pub fn query_range(&self, from_ts: i64, to_ts: i64) -> Result<Vec<Sample>> {
|
|
let conn = self.conn.lock().unwrap();
|
|
let mut stmt = conn.prepare(
|
|
"SELECT ts, cpu_pct, mem_used_bytes, mem_total_bytes, mem_pct,
|
|
disk_used_bytes, disk_total_bytes, disk_pct,
|
|
net_rx_bps, net_tx_bps, load1, load5, load15
|
|
FROM samples WHERE ts >= ?1 AND ts <= ?2 ORDER BY ts ASC",
|
|
)?;
|
|
let rows = stmt.query_map(params![from_ts, to_ts], row_to_sample)?;
|
|
let mut out = Vec::new();
|
|
for r in rows {
|
|
out.push(r?);
|
|
}
|
|
Ok(out)
|
|
}
|
|
|
|
pub fn count(&self) -> Result<i64> {
|
|
let conn = self.conn.lock().unwrap();
|
|
let n: i64 = conn.query_row("SELECT COUNT(*) FROM samples", [], |r| r.get(0))?;
|
|
Ok(n)
|
|
}
|
|
|
|
pub fn oldest_ts(&self) -> Result<Option<i64>> {
|
|
let conn = self.conn.lock().unwrap();
|
|
conn.query_row("SELECT MIN(ts) FROM samples", [], |r| r.get(0))
|
|
.optional()
|
|
.map_err(Into::into)
|
|
}
|
|
|
|
/// Age retention + hard size cap. Shrinks to `target_bytes` when over `max_bytes`.
|
|
pub fn run_cleanup(
|
|
&self,
|
|
retention_days: i64,
|
|
max_bytes: u64,
|
|
target_bytes: u64,
|
|
) -> Result<CleanupReport> {
|
|
let bytes_before = self.on_disk_bytes()?;
|
|
let mut deleted_by_age = 0usize;
|
|
let mut deleted_by_size = 0usize;
|
|
let mut vacuumed = false;
|
|
|
|
let cutoff = chrono::Utc::now().timestamp() - retention_days.max(1) * 24 * 60 * 60;
|
|
deleted_by_age = self.prune_older_than(cutoff)?;
|
|
|
|
// Keep deleting oldest batches until under target when over the hard cap.
|
|
let mut guard = 0;
|
|
while self.on_disk_bytes()? > max_bytes.max(target_bytes) && guard < 64 {
|
|
guard += 1;
|
|
let rows = self.count()?.max(1);
|
|
// Remove ~5% of rows each pass (min 1k, max 100k) so we converge quickly.
|
|
let batch = ((rows as f64) * 0.05).ceil() as i64;
|
|
let batch = batch.clamp(1_000, 100_000);
|
|
let n = self.prune_oldest(batch)?;
|
|
if n == 0 {
|
|
break;
|
|
}
|
|
deleted_by_size += n;
|
|
let _ = self.checkpoint_truncate();
|
|
}
|
|
|
|
let needs_reclaim = deleted_by_age > 0
|
|
|| deleted_by_size > 0
|
|
|| self.on_disk_bytes()? > target_bytes;
|
|
if needs_reclaim && self.on_disk_bytes()? > target_bytes {
|
|
self.vacuum()?;
|
|
vacuumed = true;
|
|
} else if deleted_by_age > 0 || deleted_by_size > 0 {
|
|
let _ = self.checkpoint_truncate();
|
|
}
|
|
|
|
// If still over hard cap after vacuum, keep trimming (pathological case).
|
|
guard = 0;
|
|
while self.on_disk_bytes()? > max_bytes && guard < 32 {
|
|
guard += 1;
|
|
let rows = self.count()?.max(1);
|
|
let batch = ((rows as f64) * 0.10).ceil() as i64;
|
|
let batch = batch.clamp(5_000, 200_000);
|
|
let n = self.prune_oldest(batch)?;
|
|
if n == 0 {
|
|
break;
|
|
}
|
|
deleted_by_size += n;
|
|
self.vacuum()?;
|
|
vacuumed = true;
|
|
}
|
|
|
|
Ok(CleanupReport {
|
|
deleted_by_age,
|
|
deleted_by_size,
|
|
vacuumed,
|
|
bytes_before,
|
|
bytes_after: self.on_disk_bytes()?,
|
|
rows_after: self.count()?,
|
|
})
|
|
}
|
|
}
|
|
|
|
fn file_size(path: &Path) -> Result<u64> {
|
|
match std::fs::metadata(path) {
|
|
Ok(m) => Ok(m.len()),
|
|
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(0),
|
|
Err(e) => Err(e.into()),
|
|
}
|
|
}
|
|
|
|
fn row_to_sample(row: &rusqlite::Row<'_>) -> rusqlite::Result<Sample> {
|
|
Ok(Sample {
|
|
ts: row.get(0)?,
|
|
cpu_pct: row.get(1)?,
|
|
mem_used_bytes: row.get::<_, i64>(2)? as u64,
|
|
mem_total_bytes: row.get::<_, i64>(3)? as u64,
|
|
mem_pct: row.get(4)?,
|
|
disk_used_bytes: row.get::<_, i64>(5)? as u64,
|
|
disk_total_bytes: row.get::<_, i64>(6)? as u64,
|
|
disk_pct: row.get(7)?,
|
|
net_rx_bps: row.get(8)?,
|
|
net_tx_bps: row.get(9)?,
|
|
load1: row.get(10)?,
|
|
load5: row.get(11)?,
|
|
load15: row.get(12)?,
|
|
})
|
|
}
|