FleetHealth PR 1: backend human-oriented fields #100
@@ -336,6 +336,60 @@ pub struct DashboardStatusReply {
|
|||||||
/// aggregator doesn't need a second round-trip.
|
/// aggregator doesn't need a second round-trip.
|
||||||
#[serde(default)]
|
#[serde(default)]
|
||||||
pub rustc_release: Option<String>,
|
pub rustc_release: Option<String>,
|
||||||
|
/// Human-oriented fields (added for FleetHealth landing).
|
||||||
|
#[serde(default)]
|
||||||
|
pub filesystem: Option<FilesystemUsage>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub hot: Option<HotTierUsage>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub mount: Option<MountStatus>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub cache: Option<CacheSummary>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub timers: Vec<TimerStatus>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
|
||||||
|
pub struct FilesystemUsage {
|
||||||
|
pub mount_point: String,
|
||||||
|
pub total_bytes: u64,
|
||||||
|
pub available_bytes: u64,
|
||||||
|
pub used_bytes: u64,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
|
||||||
|
pub struct HotTierUsage {
|
||||||
|
pub used_bytes: u64,
|
||||||
|
pub max_bytes: u64,
|
||||||
|
/// Bytes referenced by any tag or snapshot pin. `None` when
|
||||||
|
/// we can't cheaply compute it (blob store missing).
|
||||||
|
#[serde(default)]
|
||||||
|
pub pinned_bytes: Option<u64>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
|
||||||
|
pub struct MountStatus {
|
||||||
|
/// Configured mount point, whether or not it's currently mounted.
|
||||||
|
pub path: String,
|
||||||
|
pub active: bool,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
|
||||||
|
pub struct CacheSummary {
|
||||||
|
pub hits: u64,
|
||||||
|
pub misses: u64,
|
||||||
|
pub bytes_served: u64,
|
||||||
|
pub bytes_ingested: u64,
|
||||||
|
pub hit_rate: f64,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
|
||||||
|
pub struct TimerStatus {
|
||||||
|
pub unit: String,
|
||||||
|
#[serde(default)]
|
||||||
|
pub next_fire_unix: Option<u64>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub last_result: Option<String>,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Reply payload for [`Method::PutManifest`]. When `missing` is empty
|
/// Reply payload for [`Method::PutManifest`]. When `missing` is empty
|
||||||
@@ -409,6 +463,88 @@ pub struct DashboardRef {
|
|||||||
pub blob_id_hex: String,
|
pub blob_id_hex: String,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// statvfs on the given path. Uses libc directly — cheap enough
|
||||||
|
/// that we don't need to cache. Silent on error (returns None).
|
||||||
|
fn filesystem_usage(path: &std::path::Path) -> Option<FilesystemUsage> {
|
||||||
|
let cpath = std::ffi::CString::new(path.as_os_str().to_str()?).ok()?;
|
||||||
|
// SAFETY: statvfs writes to a zero-initialised struct; we
|
||||||
|
// read only when it returns 0.
|
||||||
|
let mut stat: libc::statvfs = unsafe { std::mem::zeroed() };
|
||||||
|
let rc = unsafe { libc::statvfs(cpath.as_ptr(), &mut stat) };
|
||||||
|
if rc != 0 {
|
||||||
|
return None;
|
||||||
|
}
|
||||||
|
let bsize = stat.f_frsize as u64;
|
||||||
|
let total = stat.f_blocks as u64 * bsize;
|
||||||
|
let avail = stat.f_bavail as u64 * bsize;
|
||||||
|
let used = total.saturating_sub(avail);
|
||||||
|
Some(FilesystemUsage {
|
||||||
|
mount_point: path.display().to_string(),
|
||||||
|
total_bytes: total,
|
||||||
|
available_bytes: avail,
|
||||||
|
used_bytes: used,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Query systemd for a user-scope timer's next-fire + last result.
|
||||||
|
/// Shells out to systemctl. Silent on any failure — dashboards
|
||||||
|
/// should degrade to "unknown" rather than 500.
|
||||||
|
fn timer_status(unit: &str) -> TimerStatus {
|
||||||
|
// NextElapseUSecRealtime returns micros since epoch, or 0.
|
||||||
|
// The Service unit (same name minus .timer) holds the last result.
|
||||||
|
let next = std::process::Command::new("systemctl")
|
||||||
|
.args(["--user", "show", unit, "--no-pager", "-p", "NextElapseUSecRealtime"])
|
||||||
|
.output()
|
||||||
|
.ok()
|
||||||
|
.and_then(|o| String::from_utf8(o.stdout).ok())
|
||||||
|
.and_then(|s| {
|
||||||
|
s.trim()
|
||||||
|
.strip_prefix("NextElapseUSecRealtime=")
|
||||||
|
.and_then(|v| v.parse::<u64>().ok())
|
||||||
|
})
|
||||||
|
.filter(|&v| v > 0)
|
||||||
|
.map(|us| us / 1_000_000);
|
||||||
|
let service = unit.trim_end_matches(".timer").to_string() + ".service";
|
||||||
|
let last = std::process::Command::new("systemctl")
|
||||||
|
.args(["--user", "show", &service, "--no-pager", "-p", "Result"])
|
||||||
|
.output()
|
||||||
|
.ok()
|
||||||
|
.and_then(|o| String::from_utf8(o.stdout).ok())
|
||||||
|
.and_then(|s| {
|
||||||
|
s.trim()
|
||||||
|
.strip_prefix("Result=")
|
||||||
|
.map(|v| v.to_string())
|
||||||
|
.filter(|v| !v.is_empty())
|
||||||
|
});
|
||||||
|
TimerStatus {
|
||||||
|
unit: unit.to_string(),
|
||||||
|
next_fire_unix: next,
|
||||||
|
last_result: last,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Is `path` currently a mount point? Cheap Linux check: read
|
||||||
|
/// /proc/mounts. On macOS returns None (aggregator doesn't run
|
||||||
|
/// mounts).
|
||||||
|
fn is_mounted(path: &std::path::Path) -> bool {
|
||||||
|
let want = path.display().to_string();
|
||||||
|
let mounts = match std::fs::read_to_string("/proc/mounts") {
|
||||||
|
Ok(s) => s,
|
||||||
|
Err(_) => return false,
|
||||||
|
};
|
||||||
|
for line in mounts.lines() {
|
||||||
|
// fields: <src> <mountpoint> <fstype> ...
|
||||||
|
let mut it = line.split_whitespace();
|
||||||
|
it.next();
|
||||||
|
if let Some(mp) = it.next() {
|
||||||
|
if mp == want {
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
false
|
||||||
|
}
|
||||||
|
|
||||||
/// Recursive byte count under a directory. Used by the dashboard
|
/// Recursive byte count under a directory. Used by the dashboard
|
||||||
/// handler for reporting only; silent on read errors.
|
/// handler for reporting only; silent on read errors.
|
||||||
fn dir_size_bytes(root: &std::path::Path) -> u64 {
|
fn dir_size_bytes(root: &std::path::Path) -> u64 {
|
||||||
@@ -621,6 +757,80 @@ impl RpcRouter {
|
|||||||
.gossip
|
.gossip
|
||||||
.self_kv(crate::cluster::gossip::keys::RUSTC_RELEASE)
|
.self_kv(crate::cluster::gossip::keys::RUSTC_RELEASE)
|
||||||
.await;
|
.await;
|
||||||
|
// Filesystem stat on the blob_store_root's disk.
|
||||||
|
let filesystem = root_path.as_deref().and_then(filesystem_usage);
|
||||||
|
// Hot tier: derive from gossip so we're not
|
||||||
|
// duplicating disk walks.
|
||||||
|
let hot_used = self
|
||||||
|
.gossip
|
||||||
|
.self_kv(crate::cluster::gossip::keys::HOT_USED_BYTES)
|
||||||
|
.await
|
||||||
|
.and_then(|s| s.parse::<u64>().ok())
|
||||||
|
.unwrap_or(0);
|
||||||
|
let hot_max = self
|
||||||
|
.gossip
|
||||||
|
.self_kv(crate::cluster::gossip::keys::HOT_MAX_BYTES)
|
||||||
|
.await
|
||||||
|
.and_then(|s| s.parse::<u64>().ok())
|
||||||
|
.unwrap_or(0);
|
||||||
|
let hot = Some(HotTierUsage {
|
||||||
|
used_bytes: hot_used,
|
||||||
|
max_bytes: hot_max,
|
||||||
|
pinned_bytes: None, // Phase later — needs
|
||||||
|
// cross-reference of pin set with blob sizes.
|
||||||
|
});
|
||||||
|
// Mount status: probe the conventional path. In
|
||||||
|
// the current fleet FUSE mounts at ~/clawstor-mount
|
||||||
|
// on Linux; we don't have a config field for this
|
||||||
|
// yet so hardcode the convention.
|
||||||
|
let mount = {
|
||||||
|
let home = std::env::var_os("HOME");
|
||||||
|
let path = home
|
||||||
|
.map(std::path::PathBuf::from)
|
||||||
|
.map(|h| h.join("clawstor-mount"))
|
||||||
|
.unwrap_or_else(|| std::path::PathBuf::from("/clawstor-mount"));
|
||||||
|
Some(MountStatus {
|
||||||
|
path: path.display().to_string(),
|
||||||
|
active: is_mounted(&path),
|
||||||
|
})
|
||||||
|
};
|
||||||
|
// Cache metrics — the router already tracks these
|
||||||
|
// in-memory. Compute hit-rate here so the
|
||||||
|
// dashboard doesn't need to divide.
|
||||||
|
let cache = {
|
||||||
|
let snap = self.metrics.snapshot();
|
||||||
|
// The dashboard cares about "did the peer find
|
||||||
|
// what someone asked for". Sum the get_ref /
|
||||||
|
// get_tag / get_chunk counters — they're what
|
||||||
|
// a claw-cargo build actually queries.
|
||||||
|
let hits = snap.get_ref_hits
|
||||||
|
+ snap.get_tag_hits
|
||||||
|
+ snap.get_chunk_hits;
|
||||||
|
let misses = snap.get_ref_misses
|
||||||
|
+ snap.get_tag_misses
|
||||||
|
+ snap.get_chunk_misses;
|
||||||
|
let total = hits.saturating_add(misses);
|
||||||
|
let rate = if total > 0 {
|
||||||
|
hits as f64 / total as f64
|
||||||
|
} else {
|
||||||
|
0.0
|
||||||
|
};
|
||||||
|
Some(CacheSummary {
|
||||||
|
hits,
|
||||||
|
misses,
|
||||||
|
bytes_served: snap.blob_get_bytes,
|
||||||
|
bytes_ingested: snap.blob_put_bytes,
|
||||||
|
hit_rate: rate,
|
||||||
|
})
|
||||||
|
};
|
||||||
|
// Well-known timer set. Missing timers just get
|
||||||
|
// next_fire_unix=None / last_result=None.
|
||||||
|
let timers = vec![
|
||||||
|
timer_status("clawstor-scrub.timer"),
|
||||||
|
timer_status("clawstor-gc.timer"),
|
||||||
|
timer_status("clawstor-ref-sweep.timer"),
|
||||||
|
timer_status("clawstor-snapshot-rotate.timer"),
|
||||||
|
];
|
||||||
let reply = DashboardStatusReply {
|
let reply = DashboardStatusReply {
|
||||||
node_name: self.local_name.clone(),
|
node_name: self.local_name.clone(),
|
||||||
zone: self.local_zone.clone(),
|
zone: self.local_zone.clone(),
|
||||||
@@ -632,6 +842,11 @@ impl RpcRouter {
|
|||||||
ref_tracking_count,
|
ref_tracking_count,
|
||||||
blob_store_bytes,
|
blob_store_bytes,
|
||||||
rustc_release,
|
rustc_release,
|
||||||
|
filesystem,
|
||||||
|
hot,
|
||||||
|
mount,
|
||||||
|
cache,
|
||||||
|
timers,
|
||||||
};
|
};
|
||||||
let json = serde_json::to_vec(&reply)
|
let json = serde_json::to_vec(&reply)
|
||||||
.context("encoding DashboardStatusReply as JSON")?;
|
.context("encoding DashboardStatusReply as JSON")?;
|
||||||
|
|||||||
@@ -23,8 +23,8 @@ use std::sync::Arc;
|
|||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
|
|
||||||
use crate::cluster::rpc::{
|
use crate::cluster::rpc::{
|
||||||
call_dashboard_status, call_dashboard_storage, DashboardStatusReply,
|
call_dashboard_status, call_dashboard_storage, CacheSummary, DashboardStatusReply,
|
||||||
DashboardStorageReply,
|
DashboardStorageReply, FilesystemUsage, HotTierUsage, MountStatus, TimerStatus,
|
||||||
};
|
};
|
||||||
use crate::cluster::transport::{NodeIdentity, QuicClient};
|
use crate::cluster::transport::{NodeIdentity, QuicClient};
|
||||||
use crate::config::{Config, PeerEntry};
|
use crate::config::{Config, PeerEntry};
|
||||||
@@ -134,6 +134,16 @@ pub struct NodeStatusV2 {
|
|||||||
pub ref_tracking_count: usize,
|
pub ref_tracking_count: usize,
|
||||||
pub blob_store_bytes: u64,
|
pub blob_store_bytes: u64,
|
||||||
pub rustc_release: Option<String>,
|
pub rustc_release: Option<String>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub filesystem: Option<FilesystemUsage>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub hot: Option<HotTierUsage>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub mount: Option<MountStatus>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub cache: Option<CacheSummary>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub timers: Vec<TimerStatus>,
|
||||||
/// `true` when the aggregator successfully talked to the peer;
|
/// `true` when the aggregator successfully talked to the peer;
|
||||||
/// `false` when the RPC failed. Frontend uses this to badge the
|
/// `false` when the RPC failed. Frontend uses this to badge the
|
||||||
/// card as offline.
|
/// card as offline.
|
||||||
@@ -155,6 +165,11 @@ impl NodeStatusV2 {
|
|||||||
ref_tracking_count: r.ref_tracking_count,
|
ref_tracking_count: r.ref_tracking_count,
|
||||||
blob_store_bytes: r.blob_store_bytes,
|
blob_store_bytes: r.blob_store_bytes,
|
||||||
rustc_release: r.rustc_release,
|
rustc_release: r.rustc_release,
|
||||||
|
filesystem: r.filesystem,
|
||||||
|
hot: r.hot,
|
||||||
|
mount: r.mount,
|
||||||
|
cache: r.cache,
|
||||||
|
timers: r.timers,
|
||||||
online: true,
|
online: true,
|
||||||
error: None,
|
error: None,
|
||||||
};
|
};
|
||||||
@@ -175,6 +190,11 @@ impl NodeStatusV2 {
|
|||||||
ref_tracking_count: 0,
|
ref_tracking_count: 0,
|
||||||
blob_store_bytes: 0,
|
blob_store_bytes: 0,
|
||||||
rustc_release: None,
|
rustc_release: None,
|
||||||
|
filesystem: None,
|
||||||
|
hot: None,
|
||||||
|
mount: None,
|
||||||
|
cache: None,
|
||||||
|
timers: Vec::new(),
|
||||||
online: false,
|
online: false,
|
||||||
error: Some(e),
|
error: Some(e),
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user