Menu
Public repository / dcc3eb9a6a7cc0d8d9814f0ea82d2c8b8c76f163
AkurAI-Monitor / src/schema.rs
Latest change e2201933875db930b134cab7207632f52d288fb4 - Monitor host identity from MONITOR_HOST_NAME by Ólafur Búi Ólafsson
use mysql::{params, prelude::Queryable};
/// Name of the host this Monitor runs on; its own collected metrics, logs and seeded rules attach to it.
/// Set `MONITOR_HOST_NAME` where the Monitor does not run on the EC2 VM (it runs on titan since the
/// titan move); the default keeps the original EC2 identity.
pub const DEFAULT_MONITOR_HOST_NAME: &str = "ec2-vm-dev";
pub fn monitor_host_name() -> &'static str {
static NAME: std::sync::OnceLock<String> = std::sync::OnceLock::new();
NAME.get_or_init(|| host_name_from(std::env::var("MONITOR_HOST_NAME").ok()))
}
fn host_name_from(value: Option<String>) -> String {
value
.map(|v| v.trim().to_string())
.filter(|v| !v.is_empty())
.unwrap_or_else(|| DEFAULT_MONITOR_HOST_NAME.to_string())
}
pub const MONITOR_HOST_ADDRESS: &str = "127.0.0.1";
/// Create the full schema. Every statement is idempotent (`IF NOT EXISTS`),
/// so this is safe to run on every startup. MariaDB DDL is transactional per
/// statement; there is no legacy-SQLite migration path — the MariaDB schema
/// starts at this shape (2026-07-16 migration).
pub fn create_tables(conn: &mut impl Queryable) {
const TABLES: &[&str] = &[
"CREATE TABLE IF NOT EXISTS hosts (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(191) NOT NULL UNIQUE,
address VARCHAR(255) NOT NULL,
kind VARCHAR(32) NOT NULL DEFAULT 'linux',
enabled TINYINT(1) NOT NULL DEFAULT 1,
created_at BIGINT NOT NULL,
updated_at BIGINT NOT NULL
) ENGINE=InnoDB",
"CREATE TABLE IF NOT EXISTS applications (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
host_id BIGINT NOT NULL,
name VARCHAR(191) NOT NULL,
service_name VARCHAR(255) NULL,
health_url TEXT NULL,
enabled TINYINT(1) NOT NULL DEFAULT 1,
app_version VARCHAR(64) NULL,
framework VARCHAR(64) NULL,
framework_version VARCHAR(64) NULL,
versions_reported_at BIGINT NULL,
created_at BIGINT NOT NULL,
updated_at BIGINT NOT NULL,
UNIQUE KEY uq_applications_host_name (host_id, name),
CONSTRAINT fk_applications_host FOREIGN KEY (host_id)
REFERENCES hosts(id) ON DELETE CASCADE
) ENGINE=InnoDB",
"CREATE TABLE IF NOT EXISTS metrics (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
host_id BIGINT NULL,
application_id BIGINT NULL,
org_id VARCHAR(191) NULL,
workspace_id VARCHAR(191) NULL,
installation_id VARCHAR(191) NULL,
name VARCHAR(191) NOT NULL,
value DOUBLE NOT NULL,
ts BIGINT NOT NULL,
INDEX idx_metrics_name_ts (name, ts),
INDEX idx_metrics_host_ts (host_id, ts),
INDEX idx_metrics_tenant_ts (org_id, workspace_id, ts),
CONSTRAINT fk_metrics_host FOREIGN KEY (host_id)
REFERENCES hosts(id) ON DELETE SET NULL,
CONSTRAINT fk_metrics_application FOREIGN KEY (application_id)
REFERENCES applications(id) ON DELETE SET NULL
) ENGINE=InnoDB",
"CREATE TABLE IF NOT EXISTS logs (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
host_id BIGINT NULL,
application_id BIGINT NULL,
source VARCHAR(191) NOT NULL,
line MEDIUMTEXT NOT NULL,
level VARCHAR(16) NULL,
target VARCHAR(255) NULL,
pid BIGINT NULL,
origin VARCHAR(32) NULL,
fields MEDIUMTEXT NULL,
`cursor` VARCHAR(255) NULL,
ts BIGINT NOT NULL,
UNIQUE KEY uq_logs_cursor (`cursor`),
INDEX idx_logs_source_ts (source, ts),
INDEX idx_logs_host_ts (host_id, ts),
INDEX idx_logs_level_ts (level, ts),
INDEX idx_logs_app_ts (application_id, ts),
CONSTRAINT fk_logs_host FOREIGN KEY (host_id)
REFERENCES hosts(id) ON DELETE SET NULL,
CONSTRAINT fk_logs_application FOREIGN KEY (application_id)
REFERENCES applications(id) ON DELETE SET NULL
) ENGINE=InnoDB",
"CREATE TABLE IF NOT EXISTS alert_rules (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
host_id BIGINT NULL,
application_id BIGINT NULL,
name VARCHAR(191) NOT NULL,
metric_name VARCHAR(191) NOT NULL,
operator VARCHAR(8) NOT NULL,
threshold DOUBLE NOT NULL,
duration_secs BIGINT NOT NULL,
enabled TINYINT(1) NOT NULL DEFAULT 1,
CONSTRAINT fk_alert_rules_host FOREIGN KEY (host_id)
REFERENCES hosts(id) ON DELETE CASCADE,
CONSTRAINT fk_alert_rules_application FOREIGN KEY (application_id)
REFERENCES applications(id) ON DELETE CASCADE
) ENGINE=InnoDB",
"CREATE TABLE IF NOT EXISTS alert_events (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
rule_id BIGINT NOT NULL,
org_id VARCHAR(191) NULL,
workspace_id VARCHAR(191) NULL,
installation_id VARCHAR(191) NULL,
triggered_at BIGINT NOT NULL,
resolved_at BIGINT NULL,
value DOUBLE NOT NULL,
subject VARCHAR(191) NULL,
INDEX idx_alert_events_tenant_ts (org_id, workspace_id, triggered_at),
CONSTRAINT fk_alert_events_rule FOREIGN KEY (rule_id)
REFERENCES alert_rules(id) ON DELETE CASCADE
) ENGINE=InnoDB",
"CREATE TABLE IF NOT EXISTS channels (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(191) NOT NULL UNIQUE,
kind VARCHAR(32) NOT NULL,
endpoint TEXT NOT NULL,
token_env VARCHAR(191) NULL,
enabled TINYINT(1) NOT NULL DEFAULT 1,
created_at BIGINT NOT NULL,
updated_at BIGINT NOT NULL
) ENGINE=InnoDB",
"CREATE TABLE IF NOT EXISTS recipients (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
channel_id BIGINT NOT NULL,
name VARCHAR(191) NOT NULL,
target VARCHAR(255) NOT NULL,
enabled TINYINT(1) NOT NULL DEFAULT 1,
created_at BIGINT NOT NULL,
updated_at BIGINT NOT NULL,
UNIQUE KEY uq_recipients_channel_target (channel_id, target),
CONSTRAINT fk_recipients_channel FOREIGN KEY (channel_id)
REFERENCES channels(id) ON DELETE CASCADE
) ENGINE=InnoDB",
"CREATE TABLE IF NOT EXISTS ingestion_state (
source VARCHAR(191) PRIMARY KEY,
kind VARCHAR(32) NOT NULL,
last_cursor VARCHAR(255) NULL,
last_ts BIGINT NULL,
last_error TEXT NULL,
last_error_ts BIGINT NULL,
error_count BIGINT NOT NULL DEFAULT 0
) ENGINE=InnoDB",
"CREATE TABLE IF NOT EXISTS current_metrics (
host_id BIGINT NOT NULL,
application_id BIGINT NOT NULL DEFAULT 0,
name VARCHAR(191) NOT NULL,
value DOUBLE NOT NULL,
ts BIGINT NOT NULL,
PRIMARY KEY (host_id, application_id, name),
CONSTRAINT fk_current_metrics_host FOREIGN KEY (host_id)
REFERENCES hosts(id) ON DELETE CASCADE
) ENGINE=InnoDB",
];
for ddl in TABLES {
conn.query_drop(*ddl).expect("failed to create table");
}
migrate_alert_events_subject(conn);
migrate_tenant_columns(conn);
seed_monitor_host(conn);
create_current_metrics_trigger(conn);
create_application_host_triggers(conn);
}
/// Keep `current_metrics` as an upsert cache of the newest sample per
/// (host, application, metric). `VALUES(ts)` must be compared before the
/// `ts` column is overwritten — MariaDB applies SET clauses left to right.
fn create_current_metrics_trigger(conn: &mut impl Queryable) {
conn.query_drop(
"CREATE TRIGGER IF NOT EXISTS metrics_current_insert
AFTER INSERT ON metrics FOR EACH ROW
BEGIN
IF NEW.host_id IS NOT NULL THEN
INSERT INTO current_metrics (host_id, application_id, name, value, ts)
VALUES (NEW.host_id, COALESCE(NEW.application_id, 0), NEW.name, NEW.value, NEW.ts)
ON DUPLICATE KEY UPDATE
value = IF(VALUES(ts) >= ts, VALUES(value), value),
ts = IF(VALUES(ts) >= ts, VALUES(ts), ts);
END IF;
END",
)
.expect("failed to create current metrics trigger");
}
fn create_application_host_triggers(conn: &mut impl Queryable) {
for table in ["metrics", "logs", "alert_rules"] {
for (suffix, event) in [("insert", "INSERT"), ("update", "UPDATE")] {
conn.query_drop(format!(
"CREATE TRIGGER IF NOT EXISTS {table}_application_host_{suffix}
BEFORE {event} ON {table} FOR EACH ROW
BEGIN
IF NEW.application_id IS NOT NULL AND (
NEW.host_id IS NULL
OR NOT EXISTS (
SELECT 1 FROM applications
WHERE id = NEW.application_id AND host_id = NEW.host_id
)
) THEN
SIGNAL SQLSTATE '45000'
SET MESSAGE_TEXT = 'application does not belong to host';
END IF;
END"
))
.expect("failed to create application ownership trigger");
}
}
conn.query_drop(
"CREATE TRIGGER IF NOT EXISTS applications_host_move_guard
BEFORE UPDATE ON applications FOR EACH ROW
BEGIN
IF NEW.host_id != OLD.host_id AND (
EXISTS (SELECT 1 FROM metrics WHERE application_id = OLD.id)
OR EXISTS (SELECT 1 FROM logs WHERE application_id = OLD.id)
OR EXISTS (SELECT 1 FROM alert_rules WHERE application_id = OLD.id)
) THEN
SIGNAL SQLSTATE '45000'
SET MESSAGE_TEXT = 'application host cannot change while attributed data exists';
END IF;
END",
)
.expect("failed to create application move guard");
}
/// Add `alert_events.subject` to a table created before pattern-based alert
/// rules existed. MariaDB (10.0.2+) supports `ADD COLUMN IF NOT EXISTS`, so
/// this is safe to run unconditionally on every startup, matching every
/// other statement in this module.
fn migrate_alert_events_subject(conn: &mut impl Queryable) {
conn.query_drop("ALTER TABLE alert_events ADD COLUMN IF NOT EXISTS subject VARCHAR(191) NULL")
.expect("failed to migrate alert_events.subject");
}
/// Add tenant correlation columns and filtered-query indexes to existing
/// MariaDB installations. `IF NOT EXISTS` keeps startup migration idempotent.
fn migrate_tenant_columns(conn: &mut impl Queryable) {
for (columns, index) in [
(
[
"ALTER TABLE metrics ADD COLUMN IF NOT EXISTS org_id VARCHAR(191) NULL",
"ALTER TABLE metrics ADD COLUMN IF NOT EXISTS workspace_id VARCHAR(191) NULL",
"ALTER TABLE metrics ADD COLUMN IF NOT EXISTS installation_id VARCHAR(191) NULL",
],
"ALTER TABLE metrics ADD INDEX IF NOT EXISTS idx_metrics_tenant_ts (org_id, workspace_id, ts)",
),
(
[
"ALTER TABLE alert_events ADD COLUMN IF NOT EXISTS org_id VARCHAR(191) NULL",
"ALTER TABLE alert_events ADD COLUMN IF NOT EXISTS workspace_id VARCHAR(191) NULL",
"ALTER TABLE alert_events ADD COLUMN IF NOT EXISTS installation_id VARCHAR(191) NULL",
],
"ALTER TABLE alert_events ADD INDEX IF NOT EXISTS idx_alert_events_tenant_ts (org_id, workspace_id, triggered_at)",
),
] {
for ddl in columns {
conn.query_drop(ddl)
.expect("failed to migrate tenant column");
}
conn.query_drop(index)
.expect("failed to migrate tenant index");
}
}
fn seed_monitor_host(conn: &mut impl Queryable) {
let now = chrono::Utc::now().timestamp();
let canonical_exists = conn
.exec_first::<i64, _, _>(
"SELECT id FROM hosts WHERE name = :name",
params! { "name" => monitor_host_name() },
)
.expect("failed to query Monitor host")
.is_some();
if !canonical_exists {
conn.exec_drop(
"UPDATE hosts SET name = :name, updated_at = :now
WHERE name = 'local' AND address = :address",
params! {
"name" => monitor_host_name(),
"address" => MONITOR_HOST_ADDRESS,
"now" => now,
},
)
.expect("failed to migrate Monitor host name");
}
conn.exec_drop(
"INSERT INTO hosts (name, address, kind, enabled, created_at, updated_at)
VALUES (:name, :address, 'linux', 1, :now, :now)
ON DUPLICATE KEY UPDATE
address = VALUES(address),
kind = VALUES(kind),
enabled = VALUES(enabled),
updated_at = VALUES(updated_at)",
params! {
"name" => monitor_host_name(),
"address" => MONITOR_HOST_ADDRESS,
"now" => now,
},
)
.expect("failed to seed Monitor host");
}
pub fn monitor_host_id(conn: &mut impl Queryable) -> i64 {
conn.exec_first(
"SELECT id FROM hosts WHERE name = :name",
params! { "name" => monitor_host_name() },
)
.expect("failed to query Monitor host")
.expect("Monitor host missing")
}
pub fn seed_alert_rules(conn: &mut impl Queryable) {
let host_id = monitor_host_id(conn);
// (name, metric_name, operator, threshold, duration_secs). Each row is
// seeded independently and idempotently by (name, host_id) rather than
// gated on the whole table being empty, so defaults introduced after
// initial install (e.g. the process/systemd/docker rules below) still
// land on an already-running, non-empty deployment. `metric_name`
// entries containing `%` are SQL LIKE patterns matched against every
// concrete metric name at evaluation time (see alert.rs) — required for
// dynamic per-process/unit/container entities whose names aren't known
// ahead of time.
let rules: [(&str, &str, &str, f64, i64); 7] = [
("High CPU", "cpu.usage", "gt", 90.0, 300),
("High Memory", "mem.used_pct", "gt", 90.0, 300),
("Disk Almost Full", "disk.root.used_pct", "gt", 85.0, 60),
(
"High Process Memory",
"proc.%.mem_kb",
"gt",
2_097_152.0,
300,
),
(
"High Service Memory",
"systemd.%.mem_kb",
"gt",
2_097_152.0,
300,
),
("Service Failed", "systemd.units.failed", "gt", 0.0, 60),
(
"High Container Memory",
"docker.%.mem_kb",
"gt",
2_097_152.0,
300,
),
];
let mut seeded = 0u64;
for (name, metric, op, threshold, duration) in rules {
let exists: Option<i64> = conn
.exec_first(
"SELECT 1 FROM alert_rules WHERE name = ? AND host_id = ?",
(name, host_id),
)
.expect("failed to check existing alert rule");
if exists.is_some() {
continue;
}
conn.exec_drop(
"INSERT INTO alert_rules
(host_id, name, metric_name, operator, threshold, duration_secs)
VALUES (?, ?, ?, ?, ?, ?)",
(host_id, name, metric, op, threshold, duration),
)
.expect("failed to seed alert rule");
seeded += 1;
}
if seeded > 0 {
tracing::info!(seeded, "seeded default alert rules");
}
}
#[cfg(test)]
mod host_name_tests {
use super::*;
#[test]
fn host_name_defaults_to_the_legacy_ec2_identity() {
assert_eq!(host_name_from(None), DEFAULT_MONITOR_HOST_NAME);
assert_eq!(host_name_from(Some(" ".into())), DEFAULT_MONITOR_HOST_NAME);
}
#[test]
fn host_name_comes_from_the_environment_value() {
assert_eq!(host_name_from(Some(" titan\n".into())), "titan");
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::db::testing::TestDb;
use mysql::prelude::Queryable;
#[test]
fn creates_schema_idempotently_with_seeds_and_cache_trigger() {
let Some(db) = TestDb::new() else { return };
let mut conn = db.conn();
create_tables(&mut conn);
let legacy_host_id = monitor_host_id(&mut conn);
conn.exec_drop(
"UPDATE hosts SET name = 'local' WHERE id = ?",
(legacy_host_id,),
)
.unwrap();
// Startup migrates the legacy identity in place and remains idempotent.
create_tables(&mut conn);
assert_eq!(monitor_host_id(&mut conn), legacy_host_id);
seed_alert_rules(&mut conn);
let monitor_hosts: i64 = conn
.exec_first(
"SELECT COUNT(*) FROM hosts WHERE name = :name AND address = :address",
params! { "name" => monitor_host_name(), "address" => MONITOR_HOST_ADDRESS },
)
.unwrap()
.unwrap();
assert_eq!(monitor_hosts, 1);
let rule_count: i64 = conn
.query_first("SELECT COUNT(*) FROM alert_rules")
.unwrap()
.unwrap();
assert_eq!(rule_count, 7);
// Idempotent: re-seeding (e.g. every startup) must not duplicate
// rows, including for an already-non-empty table.
seed_alert_rules(&mut conn);
let rule_count_again: i64 = conn
.query_first("SELECT COUNT(*) FROM alert_rules")
.unwrap()
.unwrap();
assert_eq!(rule_count_again, 7);
// alert_events.subject exists (fresh-install column, and idempotent
// on an already-migrated table) and stores the concrete metric name
// a pattern rule expanded to.
let rule_id: i64 = conn
.query_first("SELECT id FROM alert_rules WHERE name = 'High Process Memory'")
.unwrap()
.unwrap();
conn.exec_drop(
"INSERT INTO alert_events (rule_id, triggered_at, value, subject) VALUES (?, 1, 2.0, 'proc.nginx.mem_kb')",
(rule_id,),
)
.unwrap();
let subject: String = conn
.exec_first(
"SELECT subject FROM alert_events WHERE rule_id = ?",
(rule_id,),
)
.unwrap()
.unwrap();
assert_eq!(subject, "proc.nginx.mem_kb");
// current_metrics cache follows newest inserts, ignores stale ones.
let local = monitor_host_id(&mut conn);
conn.exec_drop(
"INSERT INTO metrics (host_id, name, value, ts) VALUES (?, 'cpu.usage', 50, 10)",
(local,),
)
.unwrap();
conn.exec_drop(
"INSERT INTO metrics (host_id, name, value, ts) VALUES (?, 'cpu.usage', 75, 20)",
(local,),
)
.unwrap();
conn.exec_drop(
"INSERT INTO metrics (host_id, name, value, ts) VALUES (?, 'cpu.usage', 60, 15)",
(local,),
)
.unwrap();
let current: f64 = conn
.query_first("SELECT value FROM current_metrics WHERE name = 'cpu.usage'")
.unwrap()
.unwrap();
assert_eq!(current, 75.0);
// Deleting a rule cascades to its events.
conn.exec_drop(
"INSERT INTO alert_events (rule_id, triggered_at, value)
SELECT id, 1, 99 FROM alert_rules LIMIT 1",
(),
)
.unwrap();
conn.query_drop("DELETE FROM alert_rules").unwrap();
let events: i64 = conn
.query_first("SELECT COUNT(*) FROM alert_events")
.unwrap()
.unwrap();
assert_eq!(events, 0);
}
#[test]
fn rejects_application_attribution_to_the_wrong_host() {
let Some(db) = TestDb::new() else { return };
let mut conn = db.conn();
create_tables(&mut conn);
let now = chrono::Utc::now().timestamp();
conn.exec_drop(
"INSERT INTO hosts (name, address, kind, enabled, created_at, updated_at)
VALUES ('remote', '100.88.0.8', 'linux', 1, ?, ?)",
(now, now),
)
.unwrap();
let remote_host = conn.last_insert_id() as i64;
conn.exec_drop(
"INSERT INTO applications (host_id, name, enabled, created_at, updated_at)
VALUES (?, 'remote-app', 1, ?, ?)",
(remote_host, now, now),
)
.unwrap();
let application_id = conn.last_insert_id() as i64;
let local_host = monitor_host_id(&mut conn);
let mismatch = conn.exec_drop(
"INSERT INTO metrics (host_id, application_id, name, value, ts)
VALUES (?, ?, 'health.ready', 1, ?)",
(local_host, application_id, now),
);
assert!(mismatch.is_err());
conn.exec_drop(
"INSERT INTO metrics (host_id, application_id, name, value, ts)
VALUES (?, ?, 'health.ready', 1, ?)",
(remote_host, application_id, now),
)
.unwrap();
let move_application = conn.exec_drop(
"UPDATE applications SET host_id = ? WHERE id = ?",
(local_host, application_id),
);
assert!(move_application.is_err());
let ownership_triggers: i64 = conn
.query_first(
"SELECT COUNT(*) FROM information_schema.triggers
WHERE trigger_schema = DATABASE()
AND trigger_name LIKE '%_application_host_%'",
)
.unwrap()
.unwrap();
assert_eq!(ownership_triggers, 6);
}
}