clawmates/crates/cm-db/src/repo/node_metrics.rs

//! Rich per-node metrics (from the workspace's Beszel hub). One latest snapshot
//! per node: scalar columns for the fleet cards + rules engine, plus a JSONB blob
//! for the per-node monitor page.

use cm_domain::{NodeId, WorkspaceId};
use serde_json::Value;
use sqlx::{PgPool, Row};
use uuid::Uuid;

use crate::DbError;

/// The latest metric snapshot for a node (nullable scalars + the full blob).
#[derive(Debug, Clone, Default)]
pub struct NodeMetrics {
    pub cpu_pct: Option<f64>,
    pub mem_pct: Option<f64>,
    pub disk_pct: Option<f64>,
    pub gpu_pct: Option<f64>,
    pub temp_max: Option<f64>,
    pub net_sent_ps: Option<i64>,
    pub net_recv_ps: Option<i64>,
    pub disk_read_ps: Option<i64>,
    pub disk_write_ps: Option<i64>,
    pub load1: Option<f64>,
    pub container_count: Option<i32>,
    pub data: Value,
}

/// Upsert a node's latest metrics snapshot.
pub async fn upsert(pool: &PgPool, node_id: NodeId, m: &NodeMetrics) -> Result<(), DbError> {
    sqlx::query(
        "INSERT INTO node_metrics
            (node_id, updated_at, cpu_pct, mem_pct, disk_pct, gpu_pct, temp_max,
             net_sent_ps, net_recv_ps, disk_read_ps, disk_write_ps, load1, container_count, data)
         VALUES ($1, now(), $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
         ON CONFLICT (node_id) DO UPDATE SET
            updated_at = now(), cpu_pct = excluded.cpu_pct, mem_pct = excluded.mem_pct,
            disk_pct = excluded.disk_pct, gpu_pct = excluded.gpu_pct, temp_max = excluded.temp_max,
            net_sent_ps = excluded.net_sent_ps, net_recv_ps = excluded.net_recv_ps,
            disk_read_ps = excluded.disk_read_ps, disk_write_ps = excluded.disk_write_ps,
            load1 = excluded.load1, container_count = excluded.container_count, data = excluded.data",
    )
    .bind(node_id.as_uuid())
    .bind(m.cpu_pct)
    .bind(m.mem_pct)
    .bind(m.disk_pct)
    .bind(m.gpu_pct)
    .bind(m.temp_max)
    .bind(m.net_sent_ps)
    .bind(m.net_recv_ps)
    .bind(m.disk_read_ps)
    .bind(m.disk_write_ps)
    .bind(m.load1)
    .bind(m.container_count)
    .bind(&m.data)
    .execute(pool)
    .await?;
    Ok(())
}

/// A node's current evaluatable scalars (Beszel-tapped, falling back to the
/// heartbeat health), for the rules engine + metrics-aware placement.
#[derive(Debug, Clone)]
pub struct EvalRow {
    pub node_id: NodeId,
    pub workspace_id: WorkspaceId,
    pub status: String,
    pub cpu_pct: Option<f64>,
    pub mem_pct: Option<f64>,
    pub disk_pct: Option<f64>,
    pub gpu_pct: Option<f64>,
    pub temp_max: Option<f64>,
    pub load1: Option<f64>,
    /// Absolute memory, for CAPACITY rather than utilisation. `mem_pct` cannot
    /// answer "does another 8 GiB VM fit" — a node at 20% of 31 GiB and one at
    /// 20% of 60 GiB report the same percentage and hold a different number of
    /// VMs. From the 5s heartbeat, which is the only source with absolutes.
    pub mem_total_bytes: Option<i64>,
    pub mem_used_bytes: Option<i64>,
    pub disk_free_bytes: Option<i64>,
    /// Memory in use with no phase VMs committed, remembered from the last time
    /// this node was observed idle. `None` until then, which makes placement
    /// fall back to inferring it — the behaviour that over-committed morpheus.
    pub mem_baseline_mib: Option<i64>,
    /// Age of each source. Placement is fail-closed on stale health (a node whose
    /// RAM we cannot read is one we are guessing at), and demotes rather than
    /// excludes on stale Beszel metrics, which only ever break ties.
    pub health_age_secs: Option<f64>,
    pub metrics_age_secs: Option<f64>,
}

impl EvalRow {
    /// Look up a metric by rule name.
    pub fn metric(&self, name: &str) -> Option<f64> {
        match name {
            "cpu_pct" => self.cpu_pct,
            "mem_pct" => self.mem_pct,
            "disk_pct" => self.disk_pct,
            "gpu_pct" => self.gpu_pct,
            "temp_max" => self.temp_max,
            "load1" => self.load1,
            _ => None,
        }
    }
    /// Free headroom heuristic (higher = more capacity) for placement ranking.
    pub fn headroom(&self) -> f64 {
        let used = self.cpu_pct.unwrap_or(0.0).max(self.mem_pct.unwrap_or(0.0));
        100.0 - used
    }

    /// [`headroom`] when at least one source is recent, otherwise the WORST
    /// possible score.
    ///
    /// Used only as a placement TIEBREAK, never as an admission gate: stale
    /// metrics may cost a node a tie, they may never win one. Admission is
    /// decided by absolute memory from the heartbeat, which has its own
    /// freshness check.
    pub fn headroom_fresh(&self, max_age_secs: f64) -> f64 {
        let fresh = |a: Option<f64>| a.is_some_and(|x| x <= max_age_secs);
        if fresh(self.health_age_secs) || fresh(self.metrics_age_secs) {
            self.headroom()
        } else {
            0.0
        }
    }
}

/// Every node's current metric scalars (merged Beszel + heartbeat health).
pub async fn eval_all(pool: &PgPool) -> Result<Vec<EvalRow>, DbError> {
    let rows = sqlx::query(
        "SELECT n.id, n.workspace_id, n.status,
            COALESCE(m.cpu_pct, h.cpu_pct) AS cpu_pct,
            COALESCE(m.mem_pct, CASE WHEN h.mem_total > 0 THEN h.mem_used::float8 / h.mem_total * 100 END) AS mem_pct,
            COALESCE(m.disk_pct, CASE WHEN h.disk_total > 0 THEN (h.disk_total - h.disk_free)::float8 / h.disk_total * 100 END) AS disk_pct,
            m.gpu_pct, m.temp_max,
            COALESCE(m.load1, h.load1) AS load1,
            h.mem_total AS mem_total_bytes,
            h.mem_used  AS mem_used_bytes,
            h.disk_free AS disk_free_bytes,
            n.mem_baseline_mib,
            EXTRACT(EPOCH FROM now() - h.captured_at)::float8 AS health_age_secs,
            EXTRACT(EPOCH FROM now() - m.updated_at)::float8  AS metrics_age_secs
         FROM nodes n
         LEFT JOIN node_health h ON h.node_id = n.id
         LEFT JOIN node_metrics m ON m.node_id = n.id",
    )
    .fetch_all(pool)
    .await?;
    Ok(rows
        .into_iter()
        .map(|r| EvalRow {
            node_id: NodeId::from(r.get::<Uuid, _>("id")),
            workspace_id: WorkspaceId::from(r.get::<Uuid, _>("workspace_id")),
            status: r.get("status"),
            cpu_pct: r.get("cpu_pct"),
            mem_pct: r.get("mem_pct"),
            disk_pct: r.get("disk_pct"),
            gpu_pct: r.get("gpu_pct"),
            temp_max: r.get("temp_max"),
            load1: r.get("load1"),
            mem_total_bytes: r.get("mem_total_bytes"),
            mem_used_bytes: r.get("mem_used_bytes"),
            disk_free_bytes: r.get("disk_free_bytes"),
            mem_baseline_mib: r.get("mem_baseline_mib"),
            health_age_secs: r.get("health_age_secs"),
            metrics_age_secs: r.get("metrics_age_secs"),
        })
        .collect())
}

/// The latest metrics blob for a node (the full snapshot for the monitor page).
pub async fn latest(pool: &PgPool, node_id: NodeId) -> Result<Option<Value>, DbError> {
    let row = sqlx::query(
        "SELECT data, extract(epoch from updated_at)::bigint AS updated FROM node_metrics WHERE node_id = $1",
    )
    .bind(node_id.as_uuid())
    .fetch_optional(pool)
    .await?;
    Ok(row.map(|r| {
        let mut data: Value = r.get("data");
        if let Some(obj) = data.as_object_mut() {
            obj.insert(
                "updatedAt".into(),
                serde_json::json!(r.get::<i64, _>("updated")),
            );
        }
        data
    }))
}