Phase 5i: publish cache metrics via gossip

ClusterServices now periodically snapshots the router's CacheMetrics
and republishes four raw counters — GetRef hits/misses and blob
GET/PUT byte totals — through chitchat as clawstor.cache.* keys.
Peers derive the hit rate locally via PeerView::cache_get_ref_hit_rate,
eliminating a per-peer GetMetrics roundtrip for placement decisions.

* gossip.rs: 4 new well-known keys, PeerView carries the counters +
  a saturating-add derived hit rate that returns None on 0/0 or when
  either counter is missing (guards against half-writes reading as
  100% hits).
* services.rs: router hoisted out of the TLS branch so a
  cache_metric_task can hold Arc<RpcRouter>. Publishes once at start
  (initial zeros so peers don't wait 60s for first read) then every
  CACHE_METRIC_INTERVAL. Ticker skipped when RPC isn't up — counters
  only fire inside dispatch.

+3 tests:
- gossip: two-node convergence with cache counters + hit-rate math
- gossip: half-write / 0-0 / normal PeerView cases return correct rates
- services: fresh cluster sees Some(0) for all four keys, then after
  in-process router counters + explicit set_cache_metrics the peer sees
  the updated values with hit rate 0.8

241 tests pass (+3 from Phase 5g). Pre-existing macOS `du -sb` failure
in hot::tests unchanged.
This commit is contained in:
Omar Sobh
2026-07-12 04:20:45 -07:00
parent d36cec11a6
commit 29be728089
2 changed files with 371 additions and 18 deletions
+186
View File
@@ -17,6 +17,7 @@
//! RPC endpoints, hot-tier occupancy, and the warm-tier projects they //! RPC endpoints, hot-tier occupancy, and the warm-tier projects they
//! serve — the last of which drives Gitea runner label dynamics later. //! serve — the last of which drives Gitea runner label dynamics later.
use crate::cluster::metrics::MetricsReply;
use crate::config::ClusterConfig; use crate::config::ClusterConfig;
use anyhow::{bail, Context, Result}; use anyhow::{bail, Context, Result};
use chitchat::transport::UdpTransport; use chitchat::transport::UdpTransport;
@@ -49,6 +50,18 @@ pub mod keys {
pub const WARM_PROJECTS: &str = "clawstor.warm.projects"; pub const WARM_PROJECTS: &str = "clawstor.warm.projects";
/// Unix timestamp (seconds) when this node's daemon started. /// Unix timestamp (seconds) when this node's daemon started.
pub const UPTIME_UNIX: &str = "clawstor.uptime.unix"; pub const UPTIME_UNIX: &str = "clawstor.uptime.unix";
/// Phase 5i: cumulative `GetRef` hits observed by this node's cache
/// since counters were reset. Placement policy divides by
/// `hits + misses` to derive a hit rate without an RPC roundtrip.
pub const CACHE_GET_REF_HITS: &str = "clawstor.cache.get_ref.hits";
/// Phase 5i: cumulative `GetRef` misses.
pub const CACHE_GET_REF_MISSES: &str = "clawstor.cache.get_ref.misses";
/// Phase 5i: cumulative bytes served out of this node's blob store
/// via `BlobGet` / `GetChunk`. Byte-volume signal for load balancing.
pub const CACHE_BLOB_GET_BYTES: &str = "clawstor.cache.blob_get.bytes";
/// Phase 5i: cumulative bytes ingested into this node's blob store
/// via `BlobPut`.
pub const CACHE_BLOB_PUT_BYTES: &str = "clawstor.cache.blob_put.bytes";
} }
/// Cluster identifier — every node in the same fleet must agree on this /// Cluster identifier — every node in the same fleet must agree on this
@@ -89,6 +102,15 @@ pub struct PeerView {
pub warm_projects: Vec<String>, pub warm_projects: Vec<String>,
pub uptime_unix: Option<u64>, pub uptime_unix: Option<u64>,
pub alive: bool, pub alive: bool,
/// Phase 5i: cumulative `GetRef` hits — `None` when the peer hasn't
/// yet published its cache counters (fresh daemon, or no RPC).
pub cache_get_ref_hits: Option<u64>,
/// Phase 5i: cumulative `GetRef` misses.
pub cache_get_ref_misses: Option<u64>,
/// Phase 5i: cumulative bytes served out of this node's blob store.
pub cache_blob_get_bytes: Option<u64>,
/// Phase 5i: cumulative bytes ingested into this node's blob store.
pub cache_blob_put_bytes: Option<u64>,
} }
impl PeerView { impl PeerView {
@@ -99,6 +121,20 @@ impl PeerView {
_ => None, _ => None,
} }
} }
/// Phase 5i: `GetRef` hit rate derived from gossiped counters.
/// Returns `None` when the peer hasn't published counters yet, or
/// when zero lookups have been recorded (0/0 is undefined).
pub fn cache_get_ref_hit_rate(&self) -> Option<f64> {
let hits = self.cache_get_ref_hits?;
let misses = self.cache_get_ref_misses?;
let total = hits.saturating_add(misses);
if total == 0 {
None
} else {
Some(hits as f64 / total as f64)
}
}
} }
/// Running gossip service — owns the chitchat handle. Drop shuts down /// Running gossip service — owns the chitchat handle. Drop shuts down
@@ -226,6 +262,25 @@ impl ClusterGossip {
self.set(keys::HOT_MAX_BYTES, bytes.to_string()).await; self.set(keys::HOT_MAX_BYTES, bytes.to_string()).await;
} }
/// Phase 5i: publish this node's cache-metric snapshot into gossip.
///
/// Peers see the four raw counters — `GetRef` hits + misses and blob
/// GET/PUT byte totals — on the next gossip round and derive hit
/// rate locally via [`PeerView::cache_get_ref_hit_rate`]. Called
/// periodically by [`crate::cluster::services::ClusterServices`]
/// once RPC has come up.
pub async fn set_cache_metrics(&self, snap: &MetricsReply) {
// Batch all four writes under one lock — chitchat serialises
// per-node kv writes, so a single locked block avoids an
// interleaving where peers see hits without misses.
let mut cc = self.chitchat.lock().await;
let state = cc.self_node_state();
state.set(keys::CACHE_GET_REF_HITS, snap.get_ref_hits.to_string());
state.set(keys::CACHE_GET_REF_MISSES, snap.get_ref_misses.to_string());
state.set(keys::CACHE_BLOB_GET_BYTES, snap.blob_get_bytes.to_string());
state.set(keys::CACHE_BLOB_PUT_BYTES, snap.blob_put_bytes.to_string());
}
/// Publish the list of warm-tier `org/repo` projects this node serves. /// Publish the list of warm-tier `org/repo` projects this node serves.
/// Later phases use this to bias runner scheduling. /// Later phases use this to bias runner scheduling.
pub async fn set_warm_projects<S: AsRef<str>>(&self, projects: &[S]) { pub async fn set_warm_projects<S: AsRef<str>>(&self, projects: &[S]) {
@@ -329,6 +384,10 @@ fn peer_view_from_state(
warm_projects, warm_projects,
uptime_unix: get_u64(state, keys::UPTIME_UNIX), uptime_unix: get_u64(state, keys::UPTIME_UNIX),
alive, alive,
cache_get_ref_hits: get_u64(state, keys::CACHE_GET_REF_HITS),
cache_get_ref_misses: get_u64(state, keys::CACHE_GET_REF_MISSES),
cache_blob_get_bytes: get_u64(state, keys::CACHE_BLOB_GET_BYTES),
cache_blob_put_bytes: get_u64(state, keys::CACHE_BLOB_PUT_BYTES),
} }
} }
@@ -548,6 +607,129 @@ mod tests {
g.shutdown(); g.shutdown();
} }
#[tokio::test]
async fn set_cache_metrics_publishes_all_four_counters_to_peer() {
// Phase 5i: A publishes cache metrics; B's PeerView for A must
// carry all four raw counters + derive the hit rate correctly.
use crate::cluster::metrics::CacheMetrics;
let port_a = next_lan_port();
let port_b = next_lan_port();
let addr_a = loopback(port_a);
let addr_b = loopback(port_b);
let cfg_a = ClusterConfig {
zone: "fabric-10g".into(),
bind_lan: Some(addr_a),
bind_tailscale: None,
peers: vec![],
bind_rpc_lan: None,
bind_rpc_tailscale: None,
tls: None,
blob_store_root: None,
};
let cfg_b = ClusterConfig {
zone: "lan-1g".into(),
bind_lan: Some(addr_b),
bind_tailscale: None,
peers: vec![PeerEntry {
name: "a".into(),
zone: "fabric-10g".into(),
lan_addr: Some(addr_a),
tailscale_addr: None,
}],
bind_rpc_lan: None,
bind_rpc_tailscale: None,
tls: None,
blob_store_root: None,
};
let g_a = ClusterGossip::bootstrap(&cfg_a, "a").await.unwrap();
let g_b = ClusterGossip::bootstrap(&cfg_b, "b").await.unwrap();
// Feed A's metrics — 3 hits, 1 miss ⇒ 0.75 hit rate — then
// publish. Byte counters test the u64 path independently.
let m = CacheMetrics::new();
for _ in 0..3 {
m.record_get_ref_hit();
}
m.record_get_ref_miss();
m.record_blob_get_bytes(4096);
m.record_blob_put_bytes(8192);
g_a.set_cache_metrics(&m.snapshot()).await;
assert!(
wait_until_peer_alive(&g_b, "a", Duration::from_secs(10)).await,
"B must see A alive"
);
// Wait for the specific cache keys to converge — liveness is
// faster than kv propagation on the first sight of a new peer.
let deadline = Instant::now() + Duration::from_secs(10);
loop {
if let Some(v) = g_b.peer("a").await {
if v.cache_get_ref_hits.is_some() {
assert_eq!(v.cache_get_ref_hits, Some(3));
assert_eq!(v.cache_get_ref_misses, Some(1));
assert_eq!(v.cache_blob_get_bytes, Some(4096));
assert_eq!(v.cache_blob_put_bytes, Some(8192));
let rate = v.cache_get_ref_hit_rate().unwrap();
assert!((rate - 0.75).abs() < 1e-9, "hit rate = {rate}");
break;
}
}
if Instant::now() >= deadline {
panic!("cache metrics never propagated to B within 10s");
}
tokio::time::sleep(Duration::from_millis(100)).await;
}
g_a.shutdown();
g_b.shutdown();
}
#[test]
fn peer_view_cache_hit_rate_none_when_counters_missing() {
let base = PeerView {
name: "x".into(),
zone: "z".into(),
rpc_lan: None,
rpc_tailscale: None,
hot_used_bytes: None,
hot_max_bytes: None,
warm_projects: vec![],
uptime_unix: None,
alive: true,
cache_get_ref_hits: None,
cache_get_ref_misses: None,
cache_blob_get_bytes: None,
cache_blob_put_bytes: None,
};
assert_eq!(base.cache_get_ref_hit_rate(), None, "no counters → None");
// Hits published but not misses — treat as None too. Prevents a
// half-write from producing a bogus 1.0 hit rate.
let half = PeerView {
cache_get_ref_hits: Some(5),
..base.clone()
};
assert_eq!(half.cache_get_ref_hit_rate(), None, "hits only → None");
// Both zero — division would be undefined; must be None.
let zero = PeerView {
cache_get_ref_hits: Some(0),
cache_get_ref_misses: Some(0),
..base.clone()
};
assert_eq!(zero.cache_get_ref_hit_rate(), None, "0/0 → None");
// Real ratio.
let real = PeerView {
cache_get_ref_hits: Some(9),
cache_get_ref_misses: Some(1),
..base
};
let rate = real.cache_get_ref_hit_rate().unwrap();
assert!((rate - 0.9).abs() < 1e-9, "expected 0.9, got {rate}");
}
#[test] #[test]
fn peer_view_hot_fill_ratio_handles_missing_or_zero() { fn peer_view_hot_fill_ratio_handles_missing_or_zero() {
let base = PeerView { let base = PeerView {
@@ -560,6 +742,10 @@ mod tests {
warm_projects: vec![], warm_projects: vec![],
uptime_unix: None, uptime_unix: None,
alive: true, alive: true,
cache_get_ref_hits: None,
cache_get_ref_misses: None,
cache_blob_get_bytes: None,
cache_blob_put_bytes: None,
}; };
assert_eq!(base.hot_fill_ratio(), None, "no used → None"); assert_eq!(base.hot_fill_ratio(), None, "no used → None");
+185 -18
View File
@@ -30,6 +30,13 @@ use tokio::task::JoinHandle;
/// against filesystem-walk cost on nodes with fat hot tiers. /// against filesystem-walk cost on nodes with fat hot tiers.
const HOT_METRIC_INTERVAL: Duration = Duration::from_secs(30); const HOT_METRIC_INTERVAL: Duration = Duration::from_secs(30);
/// Phase 5i: how often the daemon publishes its cache-metric snapshot
/// into gossip. Cache counters change on every request so freshness
/// matters, but writing to chitchat too often bloats gossip payloads.
/// 60s is the sweet spot — a placement engine polling on a 5-minute
/// cadence still sees at-worst-60s-stale counters.
const CACHE_METRIC_INTERVAL: Duration = Duration::from_secs(60);
/// Live cluster services attached to a running daemon. /// Live cluster services attached to a running daemon.
/// ///
/// Drop shuts down all background tasks. Ownership is single: the /// Drop shuts down all background tasks. Ownership is single: the
@@ -55,12 +62,19 @@ pub struct ClusterServices {
/// `DeleteTag` / `ListTags` — the human-readable pin layer over /// `DeleteTag` / `ListTags` — the human-readable pin layer over
/// raw refs. Opened alongside the blob store for the same reason. /// raw refs. Opened alongside the blob store for the same reason.
pub tag_store: Option<Arc<TagStore>>, pub tag_store: Option<Arc<TagStore>>,
/// RPC router. Held so the cache-metric ticker can call
/// `router.metrics()` without duplicating an `Arc<CacheMetrics>`
/// field. `None` when RPC never came up (no `[cluster.tls]`).
pub router: Option<Arc<RpcRouter>>,
/// QUIC accept-loop task. `None` when `[cluster.tls]` was absent /// QUIC accept-loop task. `None` when `[cluster.tls]` was absent
/// and RPC therefore didn't come up. /// and RPC therefore didn't come up.
accept_task: Option<JoinHandle<()>>, accept_task: Option<JoinHandle<()>>,
/// Periodic hot-tier metric publisher. Always running when a /// Periodic hot-tier metric publisher. Always running when a
/// gossip service exists. /// gossip service exists.
metric_task: JoinHandle<()>, metric_task: JoinHandle<()>,
/// Phase 5i: periodic cache-metric publisher. `None` when RPC is
/// not running (no router → no counters to publish).
cache_metric_task: Option<JoinHandle<()>>,
} }
impl std::fmt::Debug for ClusterServices { impl std::fmt::Debug for ClusterServices {
@@ -143,8 +157,28 @@ impl ClusterServices {
}; };
// RPC server + accept loop — only when TLS material is configured. // RPC server + accept loop — only when TLS material is configured.
let accept_task = match &cluster.tls { // Router is built even when TLS is absent iff a blob store is
Some(tls) => { // present, so an in-process caller (dashboard, tests) can hold
// it. But we only spawn the accept loop when TLS is up.
let router: Option<Arc<RpcRouter>> = if cluster.tls.is_some() {
let mut r =
RpcRouter::new(gossip.clone(), local_name.clone(), cluster.zone.clone());
if let Some(store) = &blob_store {
r = r.with_blob_store(store.clone());
}
if let Some(store) = &ref_store {
r = r.with_ref_store(store.clone());
}
if let Some(store) = &tag_store {
r = r.with_tag_store(store.clone());
}
Some(Arc::new(r))
} else {
None
};
let accept_task = match (&cluster.tls, &router) {
(Some(tls), Some(router)) => {
let identity = let identity =
NodeIdentity::from_pem_files(&tls.ca_cert, &tls.node_cert, &tls.node_key) NodeIdentity::from_pem_files(&tls.ca_cert, &tls.node_cert, &tls.node_key)
.context("loading node identity from [cluster.tls]")?; .context("loading node identity from [cluster.tls]")?;
@@ -154,27 +188,13 @@ impl ClusterServices {
.context("no RPC bind address (need bind_lan or bind_tailscale)")?; .context("no RPC bind address (need bind_lan or bind_tailscale)")?;
let server = QuicServer::bind(bind, identity) let server = QuicServer::bind(bind, identity)
.context("binding QUIC RPC server")?; .context("binding QUIC RPC server")?;
let mut router = RpcRouter::new( let router = router.clone();
gossip.clone(),
local_name.clone(),
cluster.zone.clone(),
);
if let Some(store) = &blob_store {
router = router.with_blob_store(store.clone());
}
if let Some(store) = &ref_store {
router = router.with_ref_store(store.clone());
}
if let Some(store) = &tag_store {
router = router.with_tag_store(store.clone());
}
let router = Arc::new(router);
tracing::info!("cluster RPC server listening on {}", bind); tracing::info!("cluster RPC server listening on {}", bind);
Some(tokio::spawn(async move { Some(tokio::spawn(async move {
accept_forever(server, router).await; accept_forever(server, router).await;
})) }))
} }
None => { _ => {
tracing::info!("cluster: no [cluster.tls] configured; RPC disabled"); tracing::info!("cluster: no [cluster.tls] configured; RPC disabled");
None None
} }
@@ -194,13 +214,38 @@ impl ClusterServices {
} }
}); });
// Phase 5i: cache-metric ticker. Only meaningful when the
// router exists, because counters only increment inside RPC
// dispatch. Publish once immediately so peers don't wait a full
// interval for the first read even when no requests have fired.
let cache_metric_task = if let Some(router) = &router {
let router = router.clone();
let gossip = gossip.clone();
gossip.set_cache_metrics(&router.metrics().snapshot()).await;
Some(tokio::spawn(async move {
let mut ticker = tokio::time::interval(CACHE_METRIC_INTERVAL);
// First tick fires immediately (tokio interval default),
// covered by the initial publish above — skip it here.
ticker.tick().await;
loop {
ticker.tick().await;
let snap = router.metrics().snapshot();
gossip.set_cache_metrics(&snap).await;
}
}))
} else {
None
};
Ok(Self { Ok(Self {
gossip, gossip,
blob_store, blob_store,
ref_store, ref_store,
tag_store, tag_store,
router,
accept_task, accept_task,
metric_task, metric_task,
cache_metric_task,
}) })
} }
@@ -228,6 +273,9 @@ impl ClusterServices {
task.abort(); task.abort();
} }
self.metric_task.abort(); self.metric_task.abort();
if let Some(task) = self.cache_metric_task {
task.abort();
}
} }
} }
@@ -510,6 +558,125 @@ mod tests {
svc_b.shutdown(); svc_b.shutdown();
} }
#[tokio::test]
async fn services_publish_cache_metrics_to_gossip_on_start() {
// Phase 5i: when RPC is up, the initial cache-metric publish
// fires immediately at start(). A peer B seeded from A must see
// A's four cache counter keys within gossip convergence time,
// even though no RPC requests have been issued.
use crate::cluster::transport::FleetCa;
use crate::config::ClusterTlsConfig;
let tmp = tempfile::TempDir::new().unwrap();
let ca_dir = tmp.path().join("ca");
let a_tls_dir = tmp.path().join("a-tls");
let ca = FleetCa::generate("test CA").unwrap();
ca.save(&ca_dir).unwrap();
ca.sign_leaf_to_pem("a", &a_tls_dir).unwrap();
let port_a = next_port();
let port_b = next_port();
let cfg_a = ClusterConfig {
zone: "fabric-10g".into(),
bind_lan: Some(loopback(port_a)),
tls: Some(ClusterTlsConfig {
ca_cert: a_tls_dir.join("ca.crt"),
node_cert: a_tls_dir.join("node.crt"),
node_key: a_tls_dir.join("node.key"),
}),
..Default::default()
};
let cfg_b = ClusterConfig {
zone: "lan-1g".into(),
bind_lan: Some(loopback(port_b)),
peers: vec![PeerEntry {
name: "a".into(),
zone: "fabric-10g".into(),
lan_addr: Some(loopback(port_a)),
tailscale_addr: None,
}],
..Default::default()
};
let hot_dir = tmp.path().join("hot");
std::fs::create_dir_all(&hot_dir).unwrap();
let svc_a = ClusterServices::start(
&cfg_a,
"a".into(),
hot_dir.clone(),
1_000_000,
None,
)
.await
.unwrap();
// Drive counters BEFORE start() would be a chicken-and-egg —
// instead, drive them AFTER start and check the periodic
// republish. First, verify the initial-publish (all zeros) has
// reached B: the four keys should be Some(0), not None.
let svc_b = ClusterServices::start(
&cfg_b,
"b".into(),
hot_dir,
1_000_000,
None,
)
.await
.unwrap();
let deadline = std::time::Instant::now() + Duration::from_secs(10);
loop {
if let Some(v) = svc_b.gossip.peer("a").await {
if v.cache_get_ref_hits == Some(0) && v.alive {
// Initial publish converged. Sanity-check all four.
assert_eq!(v.cache_get_ref_misses, Some(0));
assert_eq!(v.cache_blob_get_bytes, Some(0));
assert_eq!(v.cache_blob_put_bytes, Some(0));
break;
}
}
if std::time::Instant::now() >= deadline {
panic!("initial cache-metric publish never reached B in 10s");
}
tokio::time::sleep(Duration::from_millis(100)).await;
}
// Now push counters directly through the router (in-process) and
// publish once more manually via gossip — proves the whole
// router↔gossip wiring is real, without waiting 60s for the
// ticker.
let router = svc_a.router.as_ref().expect("router when TLS up");
for _ in 0..4 {
router.metrics().record_get_ref_hit();
}
router.metrics().record_get_ref_miss();
router.metrics().record_blob_get_bytes(2048);
svc_a
.gossip
.set_cache_metrics(&router.metrics().snapshot())
.await;
let deadline = std::time::Instant::now() + Duration::from_secs(10);
loop {
if let Some(v) = svc_b.gossip.peer("a").await {
if v.cache_get_ref_hits == Some(4) {
assert_eq!(v.cache_get_ref_misses, Some(1));
assert_eq!(v.cache_blob_get_bytes, Some(2048));
let rate = v.cache_get_ref_hit_rate().unwrap();
assert!((rate - 0.8).abs() < 1e-9, "hit rate = {rate}");
break;
}
}
if std::time::Instant::now() >= deadline {
panic!("updated cache metrics never reached B in 10s");
}
tokio::time::sleep(Duration::from_millis(100)).await;
}
svc_a.shutdown();
svc_b.shutdown();
}
#[tokio::test] #[tokio::test]
async fn services_with_blob_store_serves_blob_rpc_end_to_end() { async fn services_with_blob_store_serves_blob_rpc_end_to_end() {
// Prove that config → services → RPC path plumbs BlobStore // Prove that config → services → RPC path plumbs BlobStore