Menu
Public repository / 13c3d8eb1eaaae7d08375e84fb4a80bb674c1b8f
AkurAI-Monitor / src/db.rs
Latest change 0f9f642e4702d9b174c075dc1d32cd5bb60a63c4 - style: cargo fmt (AKURAI-MONITOR-8) by Ólafur Búi Ólafsson
use crate::config;
use crate::models::{DegradedState, HealthSummary, SourceFreshness, VersionFreshness};
use crate::schema;
use mysql::prelude::Queryable;
use mysql::{Opts, OptsBuilder, Pool, PoolConstraints, PoolOpts, PooledConn};
use std::sync::LazyLock;
static POOL: LazyLock<Pool> = LazyLock::new(|| {
let cfg = config::get();
let opts = Opts::from_url(&cfg.db_url).expect("invalid MONITOR_DB_URL");
let constraints = PoolConstraints::new(1, 8).expect("valid pool constraints");
let opts =
OptsBuilder::from_opts(opts).pool_opts(PoolOpts::default().with_constraints(constraints));
let pool = Pool::new(opts).expect("failed to create MariaDB pool");
let mut conn = pool.get_conn().expect("failed to connect to MariaDB");
schema::create_tables(&mut conn);
schema::seed_alert_rules(&mut conn);
pool
});
/// Run `f` with a pooled MariaDB connection. Connections are returned to the
/// pool on drop; the pool is bounded (1..=8) so DB memory stays flat.
pub fn with_db<F, R>(f: F) -> R
where
F: FnOnce(&mut PooledConn) -> R,
{
let mut conn = POOL.get_conn().expect("failed to get MariaDB connection");
f(&mut conn)
}
/// Append exact-match tenant predicates and their positional values.
///
/// With no tenant context, no predicates are added so legacy callers keep
/// seeing all rows. Callers must use the same table alias in both arguments.
pub fn append_tenant_filters(
sql: &mut String,
params: &mut Vec<mysql::Value>,
table_alias: &str,
org_id: Option<&str>,
workspace_id: Option<&str>,
) {
if let Some(org_id) = org_id {
sql.push_str(&format!(" AND {table_alias}.org_id = ?"));
params.push(org_id.into());
}
if let Some(workspace_id) = workspace_id {
sql.push_str(&format!(" AND {table_alias}.workspace_id = ?"));
params.push(workspace_id.into());
}
}
/// Compute tenant-isolated availability, backup age, version freshness, and
/// unresolved incident counts for the given window.
///
/// Every aggregate is scoped to exactly one (org_id, workspace_id) pair.
/// Missing data produces `None` / `"unknown"` — never a misleading zero.
pub fn query_tenant_health(org_id: &str, workspace_id: &str, window_secs: i64) -> HealthSummary {
with_db(|conn| {
let cutoff = chrono::Utc::now().timestamp() - window_secs;
let now = chrono::Utc::now().timestamp();
let mut degraded = Vec::new();
// ── availability: up / total "up" metric samples ──
let (total, up): (i64, i64) = conn
.exec_first(
"SELECT COUNT(*) AS total, \
COALESCE(SUM(CASE WHEN value >= 1.0 THEN 1 ELSE 0 END), 0) AS up \
FROM metrics \
WHERE name = 'up' \
AND org_id = ? AND workspace_id = ? \
AND ts >= ?",
(org_id, workspace_id, cutoff),
)
.ok()
.flatten()
.unwrap_or((0, 0));
let availability_pct = if total > 0 {
Some((up as f64 / total as f64) * 100.0)
} else {
None
};
// Collect evidence ids for "up" metric degradation.
let up_evidence: Vec<i64> = if total > 0 && up < total {
conn.exec_map(
"SELECT id FROM metrics \
WHERE name = 'up' \
AND org_id = ? AND workspace_id = ? \
AND ts >= ? AND value < 1.0 \
ORDER BY ts DESC LIMIT 5",
(org_id, workspace_id, cutoff),
|id: i64| id,
)
.unwrap_or_default()
} else {
Vec::new()
};
if let Some(pct) = availability_pct {
if pct < 100.0 {
degraded.push(DegradedState {
component: "availability".into(),
description: format!("{:.1}% up ({} / {})", pct, up, total),
evidence_ids: up_evidence,
});
}
} else {
degraded.push(DegradedState {
component: "availability".into(),
description: "no up metric found for this tenant".into(),
evidence_ids: Vec::new(),
});
}
// ── source freshness ──
// Telemetry is collected every minute; five missed intervals makes the
// summary unknown rather than accidentally presenting old data as healthy.
const SOURCE_STALE_SECS: i64 = 5 * 60;
let latest_sample: Option<(i64, i64)> = conn
.exec_first(
"SELECT id, ts FROM metrics \
WHERE org_id = ? AND workspace_id = ? \
ORDER BY ts DESC LIMIT 1",
(org_id, workspace_id),
)
.ok()
.flatten();
let source_freshness = match latest_sample {
Some((id, ts)) => {
let age_seconds = (now - ts).max(0);
if age_seconds > SOURCE_STALE_SECS {
degraded.push(DegradedState {
component: "source_freshness".into(),
description: format!(
"tenant telemetry is {}s old (stale > {}s)",
age_seconds, SOURCE_STALE_SECS
),
evidence_ids: vec![id],
});
SourceFreshness {
status: "stale".into(),
last_observed_at: Some(ts),
age_seconds: Some(age_seconds),
evidence_ids: vec![id],
}
} else {
SourceFreshness {
status: "fresh".into(),
last_observed_at: Some(ts),
age_seconds: Some(age_seconds),
evidence_ids: vec![id],
}
}
}
None => {
degraded.push(DegradedState {
component: "source_freshness".into(),
description: "no tenant telemetry has been received".into(),
evidence_ids: Vec::new(),
});
SourceFreshness {
status: "unknown".into(),
last_observed_at: None,
age_seconds: None,
evidence_ids: Vec::new(),
}
}
};
// ── last backup age ──
let backup_sample: Option<(i64, f64, i64)> = conn
.exec_first(
"SELECT id, value, ts FROM metrics \
WHERE name = 'backup.age_seconds' \
AND org_id = ? AND workspace_id = ? \
ORDER BY ts DESC LIMIT 1",
(org_id, workspace_id),
)
.ok()
.flatten();
let backup_age = backup_sample.map(|(_, value, _)| value as i64);
let backup_fresh = match backup_sample {
None => {
degraded.push(DegradedState {
component: "backup".into(),
description: "no backup.age_seconds metric found".into(),
evidence_ids: Vec::new(),
});
false
}
Some((id, _, ts)) if now - ts > SOURCE_STALE_SECS => {
degraded.push(DegradedState {
component: "backup".into(),
description: format!(
"backup telemetry is {}s old (stale > {}s)",
now - ts,
SOURCE_STALE_SECS
),
evidence_ids: vec![id],
});
false
}
Some((id, age, _)) => {
const BACKUP_STALE_SECS: i64 = 86400;
if age as i64 > BACKUP_STALE_SECS {
degraded.push(DegradedState {
component: "backup".into(),
description: format!(
"last backup {}s ago (stale > {}s)",
age as i64, BACKUP_STALE_SECS
),
evidence_ids: vec![id],
});
}
true
}
};
// ── version freshness ──
let framework_version = crate::framework_version::FRAMEWORK_VERSION;
let apps: Vec<(i64, Option<String>)> = conn
.exec_map(
"SELECT DISTINCT a.id, a.framework_version FROM applications a \
JOIN metrics m ON m.application_id = a.id \
WHERE a.framework = 'AkurAI-Framework' AND a.enabled = 1 \
AND m.org_id = ? AND m.workspace_id = ? AND m.ts >= ?",
(org_id, workspace_id, cutoff),
|(id, fw): (i64, Option<String>)| (id, fw),
)
.unwrap_or_default();
let app_count = apps.len() as i64;
let outdated: Vec<i64> = apps
.iter()
.filter(|(_, fw)| fw.as_deref() != Some(framework_version))
.map(|(id, _)| *id)
.collect();
let outdated_count = outdated.len() as i64;
let version_freshness = if app_count == 0 {
VersionFreshness {
status: "unknown".into(),
app_count: 0,
outdated_count: 0,
evidence_ids: Vec::new(),
}
} else if outdated_count > 0 {
VersionFreshness {
status: "outdated".into(),
app_count,
outdated_count,
evidence_ids: outdated.clone(),
}
} else {
VersionFreshness {
status: "current".into(),
app_count,
outdated_count: 0,
evidence_ids: Vec::new(),
}
};
if version_freshness.status == "outdated" {
degraded.push(DegradedState {
component: "versions".into(),
description: format!(
"{}/{} apps outdated (framework {})",
outdated_count, app_count, framework_version
),
evidence_ids: outdated,
});
}
// ── unresolved critical incidents ──
let unresolved: i64 = conn
.exec_first(
"SELECT COUNT(*) FROM alert_events \
WHERE org_id = ? AND workspace_id = ? \
AND resolved_at IS NULL",
(org_id, workspace_id),
)
.ok()
.flatten()
.unwrap_or(0);
let incident_evidence: Vec<i64> = if unresolved > 0 {
conn.exec_map(
"SELECT id FROM alert_events \
WHERE org_id = ? AND workspace_id = ? \
AND resolved_at IS NULL \
ORDER BY triggered_at DESC LIMIT 20",
(org_id, workspace_id),
|id: i64| id,
)
.unwrap_or_default()
} else {
Vec::new()
};
if unresolved > 0 {
degraded.push(DegradedState {
component: "incidents".into(),
description: format!("{} unresolved critical incidents", unresolved),
evidence_ids: incident_evidence,
});
}
// ── total sample count (any metric, scoped) ──
let sample_count: i64 = conn
.exec_first(
"SELECT COUNT(*) FROM metrics \
WHERE org_id = ? AND workspace_id = ? \
AND ts >= ?",
(org_id, workspace_id, cutoff),
)
.ok()
.flatten()
.unwrap_or(0);
// Missing or stale availability, backup, or source telemetry is unknown,
// never healthy; current violations remain an actionable degradation.
let status =
if source_freshness.status != "fresh" || availability_pct.is_none() || !backup_fresh {
"unknown"
} else if degraded.is_empty() {
"healthy"
} else {
"degraded"
};
HealthSummary {
org_id: org_id.to_string(),
workspace_id: workspace_id.to_string(),
status: status.to_string(),
availability_pct,
last_backup_age_seconds: backup_age,
source_freshness,
version_freshness,
unresolved_critical_incidents: unresolved,
degraded_states: degraded,
sample_count,
window_seconds: window_secs,
computed_at: now,
}
})
}
/// Test support: scratch databases on the dev MariaDB instance.
///
/// Tests are skipped (with a note on stderr) unless `MONITOR_TEST_DB_URL`
/// points at a user allowed to CREATE/DROP `akurai_test_%` databases —
/// on Titan/EC2 that is the dev container on port 3307.
#[cfg(test)]
pub mod testing {
use mysql::prelude::Queryable;
use mysql::{Opts, OptsBuilder, Pool, PoolConstraints, PoolOpts, PooledConn};
use std::sync::atomic::{AtomicU32, Ordering};
static COUNTER: AtomicU32 = AtomicU32::new(0);
pub struct TestDb {
pool: Pool,
admin: Pool,
pub name: String,
}
impl TestDb {
/// Returns `None` (after logging) when no test database is configured.
pub fn new() -> Option<TestDb> {
let Ok(url) = std::env::var("MONITOR_TEST_DB_URL") else {
eprintln!("skipping: MONITOR_TEST_DB_URL is not set");
return None;
};
let opts = Opts::from_url(&url).expect("invalid MONITOR_TEST_DB_URL");
let constraints = PoolConstraints::new(1, 2).expect("valid pool constraints");
let small = PoolOpts::default().with_constraints(constraints);
let admin = Pool::new(OptsBuilder::from_opts(opts.clone()).pool_opts(small.clone()))
.expect("connect test MariaDB");
let name = format!(
"akurai_test_mon_{}_{}",
std::process::id(),
COUNTER.fetch_add(1, Ordering::Relaxed)
);
admin
.get_conn()
.expect("admin conn")
.query_drop(format!("CREATE DATABASE `{name}`"))
.expect("create scratch database");
let scoped = OptsBuilder::from_opts(opts)
.db_name(Some(name.clone()))
.pool_opts(small);
let pool = Pool::new(scoped).expect("connect scratch database");
Some(TestDb { pool, admin, name })
}
pub fn conn(&self) -> PooledConn {
self.pool.get_conn().expect("scratch connection")
}
}
impl Drop for TestDb {
fn drop(&mut self) {
if let Ok(mut conn) = self.admin.get_conn() {
conn.query_drop(format!("DROP DATABASE IF EXISTS `{}`", self.name))
.ok();
}
}
}
}