use std::path::Path;
use std::sync::{Arc, Mutex};
use anyhow::{Context, Result};
use rusqlite::Connection;
use serde::Serialize;
use crate::model::ContainerMetrics;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum Metric {
Cpu,
Mem,
NetRx,
NetTx,
DiskRead,
DiskWrite,
}
impl Metric {
pub fn as_str(self) -> &'static str {
match self {
Metric::Cpu => "cpu",
Metric::Mem => "mem",
Metric::NetRx => "net_rx",
Metric::NetTx => "net_tx",
Metric::DiskRead => "disk_read",
Metric::DiskWrite => "disk_write",
}
}
}
#[derive(Debug, Clone, Serialize)]
pub struct ContainerTrend {
pub bucket_start_ms: u64,
pub bucket_secs: u64,
pub host: String,
pub id: String,
pub name: String,
pub stack: Option<String>,
pub metric: &'static str,
pub min: f64,
pub max: f64,
pub median: f64,
pub samples: u32,
}
#[derive(Debug, Clone, Serialize)]
pub struct MetricPoint {
pub ts_ms: u64,
pub cpu_percent: f64,
pub mem_used: Option<u64>,
pub net_rx: Option<f64>,
pub net_tx: Option<f64>,
pub disk_read: Option<f64>,
pub disk_write: Option<f64>,
}
#[derive(Debug, Clone, Serialize)]
pub struct HistoryPoint {
pub ts_ms: u64,
pub cpu_min: Option<f64>,
pub cpu_med: Option<f64>,
pub cpu_max: Option<f64>,
pub mem_min: Option<f64>,
pub mem_med: Option<f64>,
pub mem_max: Option<f64>,
pub net_rx_min: Option<f64>,
pub net_rx_med: Option<f64>,
pub net_rx_max: Option<f64>,
pub net_tx_min: Option<f64>,
pub net_tx_med: Option<f64>,
pub net_tx_max: Option<f64>,
pub disk_read_min: Option<f64>,
pub disk_read_med: Option<f64>,
pub disk_read_max: Option<f64>,
pub disk_write_min: Option<f64>,
pub disk_write_med: Option<f64>,
pub disk_write_max: Option<f64>,
}
#[derive(Clone)]
pub struct Store {
conn: Arc<Mutex<Connection>>,
}
impl Store {
pub fn open(path: &Path) -> Result<Self> {
let conn = Connection::open(path)
.with_context(|| format!("opening database at {}", path.display()))?;
Self::from_connection(conn)
}
#[cfg(test)]
pub fn open_in_memory() -> Result<Self> {
let conn = Connection::open_in_memory().context("opening in-memory database")?;
Self::from_connection(conn)
}
fn from_connection(conn: Connection) -> Result<Self> {
conn.pragma_update(None, "journal_mode", "DELETE")
.context("setting rollback journal mode")?;
conn.execute_batch(CREATE_TABLES)
.context("creating tables")?;
for (table, col, decl) in [
("container_sample", "host", "TEXT NOT NULL DEFAULT 'local'"),
("container_trend", "host", "TEXT NOT NULL DEFAULT 'local'"),
("container_sample", "net_rx", "REAL"),
("container_sample", "net_tx", "REAL"),
("container_sample", "disk_read", "REAL"),
("container_sample", "disk_write", "REAL"),
] {
add_column_if_missing(&conn, table, col, decl)?;
}
let old_index_present: bool = conn
.query_row(
"SELECT EXISTS (SELECT 1 FROM sqlite_master
WHERE type = 'index' AND name = 'container_trend_host_id')",
[],
|r| r.get(0),
)
.context("checking for legacy trend index")?;
if old_index_present {
tracing::info!(
"migrating trend indexes (one-time, may take a while on large databases)"
);
}
conn.execute_batch(CREATE_INDEXES)
.context("creating indexes")?;
Ok(Self {
conn: Arc::new(Mutex::new(conn)),
})
}
fn lock(&self) -> std::sync::MutexGuard<'_, Connection> {
self.conn
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
pub fn session_secret(&self) -> Result<Vec<u8>> {
use rusqlite::OptionalExtension;
let conn = self.lock();
let existing: Option<Vec<u8>> = conn
.query_row(
"SELECT value FROM meta WHERE key = 'session_secret'",
[],
|r| r.get(0),
)
.optional()
.context("reading session secret")?;
if let Some(secret) = existing {
return Ok(secret);
}
let secret = crate::auth::generate_secret();
conn.execute(
"INSERT INTO meta (key, value) VALUES ('session_secret', ?1)",
rusqlite::params![secret],
)
.context("storing session secret")?;
Ok(secret)
}
pub fn insert_samples(
&self,
host: &str,
ts_ms: u64,
containers: &[ContainerMetrics],
) -> Result<()> {
let mut conn = self.lock();
let tx = conn.transaction().context("begin sample transaction")?;
{
let mut stmt = tx
.prepare_cached(
"INSERT INTO container_sample
(ts_ms, host, id, name, cpu_percent, mem_used, mem_limit,
net_rx, net_tx, disk_read, disk_write)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
)
.context("prepare container insert")?;
for c in containers {
stmt.execute(rusqlite::params![
to_db(ts_ms),
host,
c.id,
c.name,
c.cpu_percent,
c.mem_used.map(to_db),
c.mem_limit.map(to_db),
c.net_rx_bps,
c.net_tx_bps,
c.disk_read_bps,
c.disk_write_bps,
])
.context("insert container sample")?;
}
}
tx.commit().context("commit sample transaction")?;
Ok(())
}
pub fn insert_container_trends(&self, trends: &[ContainerTrend]) -> Result<()> {
if trends.is_empty() {
return Ok(());
}
let mut conn = self.lock();
let tx = conn.transaction().context("begin trend transaction")?;
{
let mut stmt = tx
.prepare_cached(
"INSERT INTO container_trend
(bucket_start_ms, bucket_secs, host, id, name, stack, metric, min, max, median, samples)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
)
.context("prepare container trend insert")?;
for t in trends {
stmt.execute(rusqlite::params![
to_db(t.bucket_start_ms),
to_db(t.bucket_secs),
t.host,
t.id,
t.name,
t.stack,
t.metric,
t.min,
t.max,
t.median,
t.samples,
])
.context("insert container trend")?;
}
}
tx.commit().context("commit trend transaction")?;
Ok(())
}
pub fn prune_raw(&self, before_ms: u64) -> Result<usize> {
let conn = self.lock();
let containers = conn
.execute(
"DELETE FROM container_sample WHERE ts_ms < ?1",
[to_db(before_ms)],
)
.context("prune container samples")?;
Ok(containers)
}
pub fn prune_trends(&self, before_ms: u64) -> Result<usize> {
let conn = self.lock();
let containers = conn
.execute(
"DELETE FROM container_trend WHERE bucket_start_ms < ?1",
[to_db(before_ms)],
)
.context("prune container trends")?;
Ok(containers)
}
pub fn recent_container_samples(
&self,
host: &str,
id: &str,
since_ms: u64,
) -> Result<Vec<MetricPoint>> {
let conn = self.lock();
let mut stmt = conn
.prepare_cached(
"SELECT ts_ms, cpu_percent, mem_used, net_rx, net_tx, disk_read, disk_write
FROM container_sample
WHERE host = ?1 AND id = ?2 AND ts_ms >= ?3 ORDER BY ts_ms ASC",
)
.context("prepare recent container query")?;
let rows = stmt
.query_map(rusqlite::params![host, id, to_db(since_ms)], row_to_point)
.context("query recent container samples")?
.collect::<rusqlite::Result<Vec<_>>>()
.context("collect recent container samples")?;
Ok(rows)
}
pub fn history_container_raw(
&self,
host: &str,
id: &str,
since_ms: u64,
until_ms: u64,
group_ms: u64,
) -> Result<Vec<HistoryPoint>> {
let conn = self.lock();
let mut stmt = conn
.prepare_cached(RAW_HISTORY_SQL)
.context("prepare raw history query")?;
let rows = stmt
.query_map(
rusqlite::params![
host,
id,
to_db(since_ms),
to_db(until_ms),
to_db(group_ms.max(1))
],
history_row_to_point,
)
.context("query raw history")?
.collect::<rusqlite::Result<Vec<_>>>()
.context("collect raw history")?;
Ok(rows)
}
pub fn recent_stack_trends(
&self,
host: &str,
stack: &str,
since_ms: u64,
) -> Result<Vec<MetricPoint>> {
let conn = self.lock();
let mut stmt = conn
.prepare_cached(RECENT_STACK_TRENDS_SQL)
.context("prepare recent stack trend query")?;
let rows = stmt
.query_map(
rusqlite::params![host, stack, to_db(since_ms)],
trend_row_to_point,
)
.context("query recent stack trends")?
.collect::<rusqlite::Result<Vec<_>>>()
.context("collect recent stack trends")?;
Ok(rows)
}
pub fn history_container(
&self,
host: &str,
name: &str,
since_ms: u64,
until_ms: u64,
group_ms: u64,
) -> Result<Vec<HistoryPoint>> {
let conn = self.lock();
let mut stmt = conn
.prepare_cached(&history_container_sql())
.context("prepare container history query")?;
let rows = stmt
.query_map(
rusqlite::params![
host,
name,
to_db(since_ms),
to_db(until_ms),
to_db(group_ms.max(1))
],
history_row_to_point,
)
.context("query container history")?
.collect::<rusqlite::Result<Vec<_>>>()
.context("collect container history")?;
Ok(rows)
}
pub fn history_stack(
&self,
host: &str,
stack: &str,
since_ms: u64,
until_ms: u64,
group_ms: u64,
) -> Result<Vec<HistoryPoint>> {
let conn = self.lock();
let mut stmt = conn
.prepare_cached(&history_stack_sql())
.context("prepare stack history query")?;
let rows = stmt
.query_map(
rusqlite::params![
host,
stack,
to_db(since_ms),
to_db(until_ms),
to_db(group_ms.max(1))
],
history_row_to_point,
)
.context("query stack history")?
.collect::<rusqlite::Result<Vec<_>>>()
.context("collect stack history")?;
Ok(rows)
}
pub fn container_name(&self, host: &str, id: &str) -> Result<Option<String>> {
use rusqlite::OptionalExtension;
let conn = self.lock();
let name = conn
.query_row(
"SELECT name FROM container_trend WHERE host = ?1 AND id = ?2 LIMIT 1",
rusqlite::params![host, id],
|r| r.get(0),
)
.optional()
.context("resolving container name from trends")?;
Ok(name)
}
#[cfg(test)]
pub fn count(&self, table: &str) -> Result<u64> {
let conn = self.lock();
let n: i64 = conn.query_row(&format!("SELECT COUNT(*) FROM {table}"), [], |r| r.get(0))?;
Ok(from_db(n))
}
}
fn add_column_if_missing(conn: &Connection, table: &str, column: &str, decl: &str) -> Result<()> {
let mut stmt = conn
.prepare(&format!("PRAGMA table_info({table})"))
.with_context(|| format!("reading columns of {table}"))?;
let exists = stmt
.query_map([], |r| r.get::<_, String>(1))?
.filter_map(std::result::Result::ok)
.any(|name| name == column);
if !exists {
conn.execute(
&format!("ALTER TABLE {table} ADD COLUMN {column} {decl}"),
[],
)
.with_context(|| format!("adding column {column} to {table}"))?;
}
Ok(())
}
fn to_db(v: u64) -> i64 {
i64::try_from(v).unwrap_or(i64::MAX)
}
fn from_db(v: i64) -> u64 {
u64::try_from(v).unwrap_or(0)
}
fn row_to_point(r: &rusqlite::Row<'_>) -> rusqlite::Result<MetricPoint> {
Ok(MetricPoint {
ts_ms: from_db(r.get::<_, i64>(0)?),
cpu_percent: r.get(1)?,
mem_used: r.get::<_, Option<i64>>(2)?.map(from_db),
net_rx: r.get(3)?,
net_tx: r.get(4)?,
disk_read: r.get(5)?,
disk_write: r.get(6)?,
})
}
#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
fn trend_row_to_point(r: &rusqlite::Row<'_>) -> rusqlite::Result<MetricPoint> {
let mem: Option<f64> = r.get(2)?;
Ok(MetricPoint {
ts_ms: from_db(r.get::<_, i64>(0)?),
cpu_percent: r.get::<_, Option<f64>>(1)?.unwrap_or(0.0),
mem_used: mem.map(|m| m.round().max(0.0) as u64),
net_rx: r.get(3)?,
net_tx: r.get(4)?,
disk_read: r.get(5)?,
disk_write: r.get(6)?,
})
}
fn history_row_to_point(r: &rusqlite::Row<'_>) -> rusqlite::Result<HistoryPoint> {
Ok(HistoryPoint {
ts_ms: from_db(r.get::<_, i64>(0)?),
cpu_min: r.get(1)?,
cpu_med: r.get(2)?,
cpu_max: r.get(3)?,
mem_min: r.get(4)?,
mem_med: r.get(5)?,
mem_max: r.get(6)?,
net_rx_min: r.get(7)?,
net_rx_med: r.get(8)?,
net_rx_max: r.get(9)?,
net_tx_min: r.get(10)?,
net_tx_med: r.get(11)?,
net_tx_max: r.get(12)?,
disk_read_min: r.get(13)?,
disk_read_med: r.get(14)?,
disk_read_max: r.get(15)?,
disk_write_min: r.get(16)?,
disk_write_med: r.get(17)?,
disk_write_max: r.get(18)?,
})
}
const HISTORY_ENVELOPE: &str = "\
MIN(CASE WHEN metric='cpu' THEN min END), AVG(CASE WHEN metric='cpu' THEN median END), MAX(CASE WHEN metric='cpu' THEN max END),
MIN(CASE WHEN metric='mem' THEN min END), AVG(CASE WHEN metric='mem' THEN median END), MAX(CASE WHEN metric='mem' THEN max END),
MIN(CASE WHEN metric='net_rx' THEN min END), AVG(CASE WHEN metric='net_rx' THEN median END), MAX(CASE WHEN metric='net_rx' THEN max END),
MIN(CASE WHEN metric='net_tx' THEN min END), AVG(CASE WHEN metric='net_tx' THEN median END), MAX(CASE WHEN metric='net_tx' THEN max END),
MIN(CASE WHEN metric='disk_read' THEN min END), AVG(CASE WHEN metric='disk_read' THEN median END), MAX(CASE WHEN metric='disk_read' THEN max END),
MIN(CASE WHEN metric='disk_write' THEN min END), AVG(CASE WHEN metric='disk_write' THEN median END), MAX(CASE WHEN metric='disk_write' THEN max END)";
const STACK_SUM_PER_BUCKET: &str = "\
SUM(CASE WHEN metric='cpu' THEN min END) AS cpu_min, SUM(CASE WHEN metric='cpu' THEN median END) AS cpu_med, SUM(CASE WHEN metric='cpu' THEN max END) AS cpu_max,
SUM(CASE WHEN metric='mem' THEN min END) AS mem_min, SUM(CASE WHEN metric='mem' THEN median END) AS mem_med, SUM(CASE WHEN metric='mem' THEN max END) AS mem_max,
SUM(CASE WHEN metric='net_rx' THEN min END) AS net_rx_min, SUM(CASE WHEN metric='net_rx' THEN median END) AS net_rx_med, SUM(CASE WHEN metric='net_rx' THEN max END) AS net_rx_max,
SUM(CASE WHEN metric='net_tx' THEN min END) AS net_tx_min, SUM(CASE WHEN metric='net_tx' THEN median END) AS net_tx_med, SUM(CASE WHEN metric='net_tx' THEN max END) AS net_tx_max,
SUM(CASE WHEN metric='disk_read' THEN min END) AS disk_read_min, SUM(CASE WHEN metric='disk_read' THEN median END) AS disk_read_med, SUM(CASE WHEN metric='disk_read' THEN max END) AS disk_read_max,
SUM(CASE WHEN metric='disk_write' THEN min END) AS disk_write_min, SUM(CASE WHEN metric='disk_write' THEN median END) AS disk_write_med, SUM(CASE WHEN metric='disk_write' THEN max END) AS disk_write_max";
const RECENT_STACK_TRENDS_SQL: &str = "\
SELECT bucket_start_ms,
SUM(CASE WHEN metric = 'cpu' THEN median ELSE 0 END) AS cpu,
SUM(CASE WHEN metric = 'mem' THEN median ELSE 0 END) AS mem,
SUM(CASE WHEN metric = 'net_rx' THEN median END) AS net_rx,
SUM(CASE WHEN metric = 'net_tx' THEN median END) AS net_tx,
SUM(CASE WHEN metric = 'disk_read' THEN median END) AS disk_read,
SUM(CASE WHEN metric = 'disk_write' THEN median END) AS disk_write
FROM container_trend
WHERE host = ?1 AND stack = ?2 AND bucket_start_ms >= ?3
GROUP BY bucket_start_ms
ORDER BY bucket_start_ms ASC";
const RAW_HISTORY_SQL: &str = "\
SELECT (ts_ms / ?5) * ?5 AS bucket,
MIN(cpu_percent), AVG(cpu_percent), MAX(cpu_percent),
MIN(mem_used), AVG(mem_used), MAX(mem_used),
MIN(net_rx), AVG(net_rx), MAX(net_rx),
MIN(net_tx), AVG(net_tx), MAX(net_tx),
MIN(disk_read), AVG(disk_read), MAX(disk_read),
MIN(disk_write), AVG(disk_write), MAX(disk_write)
FROM container_sample
WHERE host = ?1 AND id = ?2 AND ts_ms >= ?3 AND ts_ms <= ?4
GROUP BY bucket
ORDER BY bucket ASC";
fn history_container_sql() -> String {
format!(
"SELECT (bucket_start_ms / ?5) * ?5 AS bucket, {HISTORY_ENVELOPE}
FROM container_trend
WHERE host = ?1 AND name = ?2 AND bucket_start_ms >= ?3 AND bucket_start_ms <= ?4
GROUP BY bucket ORDER BY bucket ASC"
)
}
fn history_stack_sql() -> String {
format!(
"WITH per_bucket AS (
SELECT bucket_start_ms AS b, {STACK_SUM_PER_BUCKET}
FROM container_trend
WHERE host = ?1 AND stack = ?2 AND bucket_start_ms >= ?3 AND bucket_start_ms <= ?4
GROUP BY b
)
SELECT (b / ?5) * ?5 AS bucket, {STACK_OUTER_ENVELOPE}
FROM per_bucket
GROUP BY bucket ORDER BY bucket ASC"
)
}
const STACK_OUTER_ENVELOPE: &str = "\
MIN(cpu_min), AVG(cpu_med), MAX(cpu_max),
MIN(mem_min), AVG(mem_med), MAX(mem_max),
MIN(net_rx_min), AVG(net_rx_med), MAX(net_rx_max),
MIN(net_tx_min), AVG(net_tx_med), MAX(net_tx_max),
MIN(disk_read_min), AVG(disk_read_med), MAX(disk_read_max),
MIN(disk_write_min), AVG(disk_write_med), MAX(disk_write_max)";
const CREATE_TABLES: &str = "
CREATE TABLE IF NOT EXISTS container_sample (
ts_ms INTEGER NOT NULL,
host TEXT NOT NULL,
id TEXT NOT NULL,
name TEXT NOT NULL,
cpu_percent REAL,
mem_used INTEGER,
mem_limit INTEGER,
net_rx REAL,
net_tx REAL,
disk_read REAL,
disk_write REAL
);
CREATE TABLE IF NOT EXISTS container_trend (
bucket_start_ms INTEGER NOT NULL,
bucket_secs INTEGER NOT NULL,
host TEXT NOT NULL,
id TEXT NOT NULL,
name TEXT NOT NULL,
stack TEXT,
metric TEXT NOT NULL,
min REAL NOT NULL,
max REAL NOT NULL,
median REAL NOT NULL,
samples INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS meta (
key TEXT PRIMARY KEY,
value BLOB NOT NULL
);
";
const CREATE_INDEXES: &str = "
CREATE INDEX IF NOT EXISTS container_sample_host_id_ts ON container_sample(host, id, ts_ms);
CREATE INDEX IF NOT EXISTS container_trend_host_name_ts ON container_trend(host, name, bucket_start_ms);
CREATE INDEX IF NOT EXISTS container_trend_host_stack_ts ON container_trend(host, stack, bucket_start_ms) WHERE stack IS NOT NULL;
CREATE INDEX IF NOT EXISTS container_sample_retention ON container_sample(ts_ms);
CREATE INDEX IF NOT EXISTS container_trend_retention ON container_trend(bucket_start_ms);
DROP INDEX IF EXISTS container_trend_host_id;
DROP INDEX IF EXISTS container_trend_host_name;
";
#[cfg(test)]
mod tests {
use super::*;
use crate::model::{ContainerState, HealthState};
const HOST: &str = "local";
fn container_sample(id: &str, cpu: Option<f64>) -> ContainerMetrics {
ContainerMetrics {
id: id.to_string(),
name: format!("c-{id}"),
image: "img:latest".to_string(),
state: ContainerState::Running,
status: "Up".to_string(),
health: HealthState::None,
stack: None,
cpu_percent: cpu,
mem_used: Some(123),
mem_limit: Some(456),
net_rx_bps: Some(10.0),
net_tx_bps: Some(20.0),
disk_read_bps: Some(30.0),
disk_write_bps: Some(40.0),
ports: Vec::new(),
}
}
#[test]
fn insert_and_prune_roundtrip() {
let store = Store::open_in_memory().unwrap();
let containers = vec![
container_sample("a", Some(1.0)),
container_sample("b", None),
];
store.insert_samples(HOST, 1_000, &containers).unwrap();
store.insert_samples(HOST, 5_000, &containers).unwrap();
assert_eq!(store.count("container_sample").unwrap(), 4);
let removed = store.prune_raw(5_000).unwrap();
assert_eq!(removed, 2);
assert_eq!(store.count("container_sample").unwrap(), 2);
}
#[test]
fn migrates_pre_multi_host_database_in_place() {
let conn = Connection::open_in_memory().unwrap();
conn.execute_batch(
"CREATE TABLE container_sample (
ts_ms INTEGER NOT NULL, id TEXT NOT NULL, name TEXT NOT NULL,
cpu_percent REAL, mem_used INTEGER, mem_limit INTEGER,
net_rx REAL, net_tx REAL, disk_read REAL, disk_write REAL);
CREATE INDEX container_sample_id_ts ON container_sample(id, ts_ms);
CREATE TABLE container_trend (
bucket_start_ms INTEGER NOT NULL, bucket_secs INTEGER NOT NULL,
id TEXT NOT NULL, name TEXT NOT NULL, stack TEXT, metric TEXT NOT NULL,
min REAL NOT NULL, max REAL NOT NULL, median REAL NOT NULL, samples INTEGER NOT NULL);
CREATE INDEX container_trend_ts ON container_trend(id, metric, bucket_start_ms);
CREATE TABLE meta (key TEXT PRIMARY KEY, value BLOB NOT NULL);
INSERT INTO container_sample (ts_ms, id, name) VALUES (1000, 'abc', 'c1');",
)
.unwrap();
let store = Store::from_connection(conn).unwrap();
let backfilled = {
let conn = store.lock();
conn.query_row(
"SELECT host FROM container_sample WHERE id = 'abc'",
[],
|r| r.get::<_, String>(0),
)
.unwrap()
};
assert_eq!(backfilled, "local");
store
.insert_samples("nas", 2_000, &[container_sample("z", Some(2.0))])
.unwrap();
assert_eq!(store.count("container_sample").unwrap(), 2);
}
#[test]
fn retention_prunes_seek_by_index_instead_of_scanning() {
let store = Store::open_in_memory().unwrap();
let conn = store.lock();
let plan = |sql: &str| -> String {
let mut stmt = conn
.prepare(&format!("EXPLAIN QUERY PLAN {sql}"))
.expect("prepare explain");
stmt.query_map([], |r| r.get::<_, String>(3))
.expect("query plan")
.map(Result::unwrap)
.collect::<Vec<_>>()
.join("; ")
};
let sample = plan("DELETE FROM container_sample WHERE ts_ms < 1");
assert!(
sample.contains("container_sample_retention") && !sample.contains("SCAN"),
"sample prune should seek by index, got: {sample}"
);
let trend = plan("DELETE FROM container_trend WHERE bucket_start_ms < 1");
assert!(
trend.contains("container_trend_retention") && !trend.contains("SCAN"),
"trend prune should seek by index, got: {trend}"
);
}
#[test]
fn trend_history_reads_seek_time_range_by_index() {
let store = Store::open_in_memory().unwrap();
let conn = store.lock();
let plan = |sql: &str| -> String {
let mut stmt = conn
.prepare(&format!("EXPLAIN QUERY PLAN {sql}"))
.expect("prepare explain");
let dummies = vec![rusqlite::types::Value::Integer(1); stmt.parameter_count()];
stmt.query_map(rusqlite::params_from_iter(dummies), |r| {
r.get::<_, String>(3)
})
.expect("query plan")
.map(Result::unwrap)
.collect::<Vec<_>>()
.join("; ")
};
for (what, sql, index, time_col, table) in [
(
"container history",
history_container_sql(),
"container_trend_host_name_ts",
"bucket_start_ms>",
"container_trend",
),
(
"stack history",
history_stack_sql(),
"container_trend_host_stack_ts",
"bucket_start_ms>",
"container_trend",
),
(
"stack seed",
RECENT_STACK_TRENDS_SQL.to_string(),
"container_trend_host_stack_ts",
"bucket_start_ms>",
"container_trend",
),
(
"raw history",
RAW_HISTORY_SQL.to_string(),
"container_sample_host_id_ts",
"ts_ms>",
"container_sample",
),
] {
let p = plan(&sql);
assert!(
p.contains(index) && p.contains(time_col),
"{what} should range-seek {time_col} via {index}, got: {p}"
);
assert!(
!p.contains(&format!("SCAN {table}")),
"{what} should not scan {table}, got: {p}"
);
}
}
#[test]
fn history_survives_container_recreation() {
let store = Store::open_in_memory().unwrap();
let ct = |bucket: u64, id: &str| ContainerTrend {
bucket_start_ms: bucket,
bucket_secs: 60,
host: HOST.into(),
id: id.to_string(),
name: "web-1".to_string(),
stack: None,
metric: Metric::Cpu.as_str(),
min: 1.0,
max: 3.0,
median: 2.0,
samples: 20,
};
store
.insert_container_trends(&[ct(0, "old-id"), ct(60_000, "new-id")])
.unwrap();
let h = store
.history_container(HOST, "web-1", 0, u64::MAX, 60_000)
.unwrap();
assert_eq!(h.len(), 2, "history must span both incarnations");
assert_eq!(
store.container_name(HOST, "old-id").unwrap().as_deref(),
Some("web-1")
);
assert_eq!(store.container_name(HOST, "gone").unwrap(), None);
}
#[test]
fn container_samples_roundtrip_io_rates() {
let store = Store::open_in_memory().unwrap();
store
.insert_samples(HOST, 1_000, &[container_sample("a", Some(1.0))])
.unwrap();
let points = store.recent_container_samples(HOST, "a", 0).unwrap();
assert_eq!(points.len(), 1);
let p = &points[0];
assert_eq!(p.net_rx, Some(10.0));
assert_eq!(p.net_tx, Some(20.0));
assert_eq!(p.disk_read, Some(30.0));
assert_eq!(p.disk_write, Some(40.0));
}
#[test]
fn session_secret_is_generated_once_and_stable() {
let store = Store::open_in_memory().unwrap();
let first = store.session_secret().unwrap();
assert_eq!(first.len(), 32);
assert_eq!(store.session_secret().unwrap(), first);
}
#[test]
fn insert_trends_roundtrip() {
let store = Store::open_in_memory().unwrap();
store
.insert_container_trends(&[ContainerTrend {
bucket_start_ms: 0,
bucket_secs: 60,
host: HOST.into(),
id: "a".to_string(),
name: "c-a".to_string(),
stack: Some("web".to_string()),
metric: Metric::Cpu.as_str(),
min: 1.0,
max: 9.0,
median: 4.0,
samples: 20,
}])
.unwrap();
assert_eq!(store.count("container_trend").unwrap(), 1);
}
#[test]
fn recent_stack_trends_sums_members_and_pivots() {
let store = Store::open_in_memory().unwrap();
let ct = |bucket: u64, id: &str, stack: &str, metric, median: f64| ContainerTrend {
bucket_start_ms: bucket,
bucket_secs: 60,
host: HOST.into(),
id: id.to_string(),
name: format!("c-{id}"),
stack: Some(stack.to_string()),
metric,
min: median,
max: median,
median,
samples: 20,
};
store
.insert_container_trends(&[
ct(0, "a", "web", Metric::Cpu.as_str(), 4.0),
ct(0, "a", "web", Metric::Mem.as_str(), 100.0),
ct(0, "b", "web", Metric::Cpu.as_str(), 6.0),
ct(0, "b", "web", Metric::Mem.as_str(), 200.0),
ct(60_000, "a", "web", Metric::Cpu.as_str(), 2.0),
ct(60_000, "a", "web", Metric::Mem.as_str(), 150.0),
ct(60_000, "b", "web", Metric::Cpu.as_str(), 3.0),
ct(60_000, "b", "web", Metric::Mem.as_str(), 250.0),
ct(0, "a", "web", Metric::NetRx.as_str(), 5.0),
ct(0, "b", "web", Metric::NetRx.as_str(), 15.0),
ct(0, "c", "db", Metric::Cpu.as_str(), 99.0),
])
.unwrap();
let points = store.recent_stack_trends(HOST, "web", 0).unwrap();
assert_eq!(points.len(), 2);
assert_eq!(points[0].ts_ms, 0);
assert!((points[0].cpu_percent - 10.0).abs() < 1e-9);
assert_eq!(points[0].mem_used, Some(300));
assert_eq!(points[0].net_rx, Some(20.0));
assert_eq!(points[0].disk_read, None);
assert_eq!(points[1].net_rx, None);
assert_eq!(points[1].ts_ms, 60_000);
assert!((points[1].cpu_percent - 5.0).abs() < 1e-9);
assert_eq!(points[1].mem_used, Some(400));
let recent = store.recent_stack_trends(HOST, "web", 60_000).unwrap();
assert_eq!(recent.len(), 1);
assert_eq!(recent[0].ts_ms, 60_000);
}
#[test]
fn history_container_carries_io_envelope() {
let store = Store::open_in_memory().unwrap();
store
.insert_container_trends(&[
ContainerTrend {
bucket_start_ms: 0,
bucket_secs: 60,
host: HOST.into(),
id: "a".to_string(),
name: "c-a".to_string(),
stack: None,
metric: Metric::NetRx.as_str(),
min: 100.0,
max: 900.0,
median: 400.0,
samples: 20,
},
ContainerTrend {
bucket_start_ms: 0,
bucket_secs: 60,
host: HOST.into(),
id: "a".to_string(),
name: "c-a".to_string(),
stack: None,
metric: Metric::DiskWrite.as_str(),
min: 1.0,
max: 3.0,
median: 2.0,
samples: 20,
},
])
.unwrap();
let h = store
.history_container(HOST, "c-a", 0, u64::MAX, 60_000)
.unwrap();
assert_eq!(h.len(), 1);
assert_eq!(h[0].net_rx_min, Some(100.0));
assert_eq!(h[0].net_rx_med, Some(400.0));
assert_eq!(h[0].net_rx_max, Some(900.0));
assert_eq!(h[0].disk_write_med, Some(2.0));
assert_eq!(h[0].net_tx_med, None);
assert_eq!(h[0].cpu_med, None);
}
#[test]
fn history_stack_sums_members_then_downsamples() {
let store = Store::open_in_memory().unwrap();
let ct = |bucket: u64, id: &str, min: f64, median: f64, max: f64| ContainerTrend {
bucket_start_ms: bucket,
bucket_secs: 60,
host: HOST.into(),
id: id.to_string(),
name: format!("c-{id}"),
stack: Some("web".to_string()),
metric: Metric::Cpu.as_str(),
min,
max,
median,
samples: 20,
};
store
.insert_container_trends(&[
ct(0, "a", 1.0, 4.0, 9.0),
ct(0, "b", 2.0, 6.0, 10.0),
ct(60_000, "a", 1.0, 2.0, 3.0),
ct(60_000, "b", 1.0, 3.0, 5.0),
])
.unwrap();
let fine = store
.history_stack(HOST, "web", 0, u64::MAX, 60_000)
.unwrap();
assert_eq!(fine.len(), 2);
assert_eq!(fine[0].cpu_min, Some(3.0));
assert_eq!(fine[0].cpu_med, Some(10.0));
assert_eq!(fine[0].cpu_max, Some(19.0));
let coarse = store
.history_stack(HOST, "web", 0, u64::MAX, 120_000)
.unwrap();
assert_eq!(coarse.len(), 1);
assert_eq!(coarse[0].cpu_min, Some(2.0));
assert_eq!(coarse[0].cpu_med, Some(7.5)); assert_eq!(coarse[0].cpu_max, Some(19.0));
assert!(
store
.history_stack(HOST, "nope", 0, u64::MAX, 60_000)
.unwrap()
.is_empty()
);
assert!(
store
.history_container(HOST, "nope", 0, u64::MAX, 60_000)
.unwrap()
.is_empty()
);
}
#[test]
fn raw_history_downsamples_and_respects_window() {
let store = Store::open_in_memory().unwrap();
let c = |cpu| [container_sample("a", Some(cpu))];
for (ts, cpu) in [(0, 10.0), (1_000, 30.0), (2_000, 20.0), (5_000, 40.0)] {
store.insert_samples(HOST, ts, &c(cpu)).unwrap();
}
let h = store
.history_container_raw(HOST, "a", 0, u64::MAX, 3_000)
.unwrap();
assert_eq!(h.len(), 2, "4 samples in 3s groups must yield 2 points");
assert_eq!(h[0].cpu_min, Some(10.0));
assert_eq!(h[0].cpu_med, Some(20.0)); assert_eq!(h[0].cpu_max, Some(30.0));
let full = store
.history_container_raw(HOST, "a", 0, u64::MAX, 500)
.unwrap();
assert_eq!(full.len(), 4);
assert_eq!(full[3].cpu_min, full[3].cpu_max);
let bounded = store
.history_container_raw(HOST, "a", 1_000, 2_000, 500)
.unwrap();
assert_eq!(bounded.len(), 2);
}
#[test]
fn recent_container_samples_filters_and_orders() {
let store = Store::open_in_memory().unwrap();
let c = |cpu| [container_sample("a", Some(cpu))];
store.insert_samples(HOST, 5_000, &c(20.0)).unwrap();
store.insert_samples(HOST, 1_000, &c(10.0)).unwrap();
store.insert_samples(HOST, 9_000, &c(30.0)).unwrap();
let points = store.recent_container_samples(HOST, "a", 5_000).unwrap();
assert_eq!(points.len(), 2);
assert_eq!(points[0].ts_ms, 5_000);
assert_eq!(points[1].ts_ms, 9_000);
}
#[test]
fn prune_trends_drops_old_buckets() {
let store = Store::open_in_memory().unwrap();
let trend = |start: u64| ContainerTrend {
bucket_start_ms: start,
bucket_secs: 60,
host: HOST.into(),
id: "a".to_string(),
name: "c-a".to_string(),
stack: None,
metric: Metric::Cpu.as_str(),
min: 1.0,
max: 2.0,
median: 1.5,
samples: 10,
};
store
.insert_container_trends(&[trend(1_000), trend(50_000)])
.unwrap();
let removed = store.prune_trends(50_000).unwrap();
assert_eq!(removed, 1);
assert_eq!(store.count("container_trend").unwrap(), 1);
}
}