dashboard-v2 PR 1: design doc + backend read-only endpoints #92

Merged
osobh merged 1 commits from dashboard-v2-backend into main 2026-07-14 22:52:53 +00:00
6 changed files with 671 additions and 1 deletions
Showing only changes of commit b431475af7 - Show all commits
+1
View File
@@ -29,6 +29,7 @@ mod hot;
mod manifest;
mod restore;
mod serve;
mod serve_v2;
mod snapshot;
mod sync;
mod zfs;
+1
View File
@@ -27,6 +27,7 @@ pub mod hot;
pub mod manifest;
pub mod restore;
pub mod serve;
pub mod serve_v2;
pub mod snapshot;
pub mod sync;
pub mod zfs;
+1
View File
@@ -8,6 +8,7 @@ mod hot;
mod manifest;
mod restore;
mod serve;
mod serve_v2;
mod snapshot;
mod sync;
mod zfs;
+9 -1
View File
@@ -693,7 +693,15 @@ pub fn build_app(cfg: Config, manifest_path: PathBuf, static_dir: Option<PathBuf
api = api.fallback_service(tower_http::services::ServeDir::new(dir));
}
api.layer(cors).with_state(state)
// dashboard-v2 backend (docs/dashboard-v2.md). Additive alongside
// the legacy /api/* routes above so the cutover doesn't break the
// old dashboard while the new one is being iterated on.
let v2_state = std::sync::Arc::new(crate::serve_v2::V2State::from_config(&state.cfg));
let v2_routes = crate::serve_v2::routes().with_state(v2_state);
Router::new()
.merge(api.layer(cors.clone()).with_state(state))
.merge(v2_routes.layer(cors))
}
pub async fn run_server(
+561
View File
@@ -0,0 +1,561 @@
//! dashboard-v2 backend — additive `/api/v2/*` handlers.
//!
//! Reads the distributed-architecture stores (BlobStore, TagStore,
//! RefStore, SnapshotStore, RefTracking). Kept in a separate file
//! from `serve.rs` so the legacy ZFS-era handlers there can be
//! removed cleanly at cutover without a big rebase.
//!
//! Design + endpoint spec: `docs/dashboard-v2.md`.
//!
//! Single-node reads in this cut. Cross-node fleet aggregation
//! (fan-out via QUIC RPC) lands in a follow-on PR.
use axum::{
extract::{Path, Query, State},
routing::get,
Json, Router,
};
use serde::{Deserialize, Serialize};
use std::path::PathBuf;
use std::sync::Arc;
use crate::cluster::blob::{BlobId, BlobStore};
use crate::cluster::ref_tracking::{RefEntry, RefTracking};
use crate::cluster::refs::RefStore;
use crate::cluster::snapshot::SnapshotStore;
use crate::cluster::tags::TagStore;
use crate::config::Config;
/// Runtime shared by all v2 handlers. Cheap clones — inner types
/// are `Arc<_>` or filesystem-backed with no in-memory state.
#[derive(Clone)]
pub struct V2State {
pub node_name: String,
pub blob_store_root: Option<PathBuf>,
}
impl V2State {
pub fn from_config(cfg: &Config) -> Self {
let blob_store_root = cfg
.cluster
.as_ref()
.and_then(|c| c.blob_store_root.clone());
Self {
node_name: cfg.node.name.clone(),
blob_store_root,
}
}
fn open_blob(&self) -> Option<BlobStore> {
let root = self.blob_store_root.as_ref()?;
BlobStore::open(root.clone()).ok()
}
fn open_tags(&self) -> Option<TagStore> {
let root = self.blob_store_root.as_ref()?;
TagStore::open(root.join("tags-db")).ok()
}
fn open_refs(&self) -> Option<RefStore> {
let root = self.blob_store_root.as_ref()?;
RefStore::open(root.join("refs-db")).ok()
}
fn open_snapshots(&self) -> Option<SnapshotStore> {
let root = self.blob_store_root.as_ref()?;
SnapshotStore::open(root.clone()).ok()
}
fn open_ref_tracking(&self) -> Option<RefTracking> {
let root = self.blob_store_root.as_ref()?;
RefTracking::open(root.clone()).ok()
}
}
// ── request / response shapes ────────────────────────────────────
#[derive(Serialize, Deserialize)]
pub struct NodeStatusV2 {
pub node_name: String,
pub blob_store_root: Option<PathBuf>,
pub blob_count: usize,
pub tag_count: usize,
pub ref_count: usize,
pub snapshot_count: usize,
pub ref_tracking_count: usize,
pub blob_store_bytes: u64,
}
#[derive(Deserialize)]
pub struct Pagination {
#[serde(default = "default_limit")]
pub limit: usize,
#[serde(default)]
pub offset: usize,
}
fn default_limit() -> usize {
200
}
#[derive(Serialize, Deserialize)]
pub struct BlobSummary {
pub blob_id_hex: String,
pub size_bytes: u64,
pub chunk_count: usize,
}
#[derive(Serialize, Deserialize)]
pub struct TagSummaryV2 {
pub key: String,
pub value_hex: String,
}
#[derive(Deserialize)]
pub struct TagFilter {
#[serde(default)]
pub prefix: String,
}
#[derive(Serialize, Deserialize)]
pub struct RefSummary {
pub fingerprint_hex: String,
pub blob_id_hex: String,
}
#[derive(Serialize, Deserialize)]
pub struct SnapshotSummaryV2 {
pub name: String,
pub created_at_unix: u64,
pub blob_count: usize,
pub file_bytes: u64,
}
#[derive(Serialize, Deserialize)]
pub struct RefTrackingItem {
pub fingerprint_hex: String,
pub repo: String,
pub refs: Vec<String>,
pub first_seen_unix: u64,
pub last_seen_unix: u64,
}
impl From<RefEntry> for RefTrackingItem {
fn from(e: RefEntry) -> Self {
let mut hex = String::with_capacity(64);
for b in &e.fingerprint {
hex.push_str(&format!("{b:02x}"));
}
Self {
fingerprint_hex: hex,
repo: e.repo,
refs: e.refs,
first_seen_unix: e.first_seen_unix,
last_seen_unix: e.last_seen_unix,
}
}
}
#[derive(Deserialize)]
pub struct RefTrackingFilter {
#[serde(default)]
pub repo: String,
}
#[derive(Serialize)]
pub struct EmptyReason {
pub reason: String,
}
// ── handlers ─────────────────────────────────────────────────────
async fn handle_status(State(s): State<Arc<V2State>>) -> Json<NodeStatusV2> {
let blob = s.open_blob();
let tags = s.open_tags();
let refs = s.open_refs();
let snaps = s.open_snapshots();
let rt = s.open_ref_tracking();
let root = s.blob_store_root.clone();
let blob_count = match &blob {
Some(b) => b.list_blob_ids().await.map(|v| v.len()).unwrap_or(0),
None => 0,
};
let tag_count = match &tags {
Some(t) => t.list().await.map(|v| v.len()).unwrap_or(0),
None => 0,
};
let ref_count = match &refs {
Some(r) => r.list().await.map(|v| v.len()).unwrap_or(0),
None => 0,
};
let snapshot_count = match &snaps {
Some(s) => s.list().await.map(|v| v.len()).unwrap_or(0),
None => 0,
};
let ref_tracking_count = match &rt {
Some(r) => r.list_all().await.map(|v| v.len()).unwrap_or(0),
None => 0,
};
let blob_store_bytes = match &root {
Some(p) => dir_size_bytes(p),
None => 0,
};
Json(NodeStatusV2 {
node_name: s.node_name.clone(),
blob_store_root: root,
blob_count,
tag_count,
ref_count,
snapshot_count,
ref_tracking_count,
blob_store_bytes,
})
}
async fn handle_node_status(
State(s): State<Arc<V2State>>,
Path(name): Path<String>,
) -> Result<Json<NodeStatusV2>, (axum::http::StatusCode, Json<EmptyReason>)> {
// Single-node cut: only "local" or "<own name>" resolves.
// Cross-node lookup lands with the fleet fan-out follow-on.
if name != "local" && name != s.node_name {
return Err((
axum::http::StatusCode::NOT_IMPLEMENTED,
Json(EmptyReason {
reason: format!(
"cross-node lookup not yet implemented; requested {name}, this is {}",
s.node_name
),
}),
));
}
Ok(handle_status(State(s)).await)
}
async fn handle_blobs(
State(s): State<Arc<V2State>>,
Query(p): Query<Pagination>,
) -> Json<Vec<BlobSummary>> {
let store = match s.open_blob() {
Some(b) => b,
None => return Json(Vec::new()),
};
let mut ids = store.list_blob_ids().await.unwrap_or_default();
ids.sort();
let store = std::sync::Arc::new(store);
let out = ids.into_iter().skip(p.offset).take(p.limit).map(|id| {
let store = store.clone();
async move {
let manifest = store.load_manifest(&id).await.ok().flatten();
BlobSummary {
blob_id_hex: id.to_hex(),
size_bytes: manifest.as_ref().map(|m| m.total_size).unwrap_or(0),
chunk_count: manifest.map(|m| m.chunks.len()).unwrap_or(0),
}
}
});
let collected: Vec<BlobSummary> = futures_join(out).await;
Json(collected)
}
async fn handle_tags(
State(s): State<Arc<V2State>>,
Query(f): Query<TagFilter>,
) -> Json<Vec<TagSummaryV2>> {
let store = match s.open_tags() {
Some(t) => t,
None => return Json(Vec::new()),
};
let entries = store.list().await.unwrap_or_default();
let out: Vec<_> = entries
.into_iter()
.filter(|e| f.prefix.is_empty() || e.key.starts_with(&f.prefix))
.map(|e| TagSummaryV2 {
key: e.key,
value_hex: e.value_hex,
})
.collect();
Json(out)
}
async fn handle_refs(
State(s): State<Arc<V2State>>,
Query(p): Query<Pagination>,
) -> Json<Vec<RefSummary>> {
let store = match s.open_refs() {
Some(r) => r,
None => return Json(Vec::new()),
};
let all = store.list().await.unwrap_or_default();
let out: Vec<_> = all
.into_iter()
.skip(p.offset)
.take(p.limit)
.map(|(k, v)| {
let mut kh = String::with_capacity(64);
for b in &k {
kh.push_str(&format!("{b:02x}"));
}
RefSummary {
fingerprint_hex: kh,
blob_id_hex: BlobId::from_bytes(v).to_hex(),
}
})
.collect();
Json(out)
}
async fn handle_snapshots(State(s): State<Arc<V2State>>) -> Json<Vec<SnapshotSummaryV2>> {
let store = match s.open_snapshots() {
Some(s) => s,
None => return Json(Vec::new()),
};
let entries = store.list().await.unwrap_or_default();
let out: Vec<_> = entries
.into_iter()
.map(|s| SnapshotSummaryV2 {
name: s.name,
created_at_unix: s.created_at_unix,
blob_count: s.blob_count,
file_bytes: s.file_bytes,
})
.collect();
Json(out)
}
async fn handle_ref_tracking(
State(s): State<Arc<V2State>>,
Query(f): Query<RefTrackingFilter>,
) -> Json<Vec<RefTrackingItem>> {
let store = match s.open_ref_tracking() {
Some(r) => r,
None => return Json(Vec::new()),
};
let all = store.list_all().await.unwrap_or_default();
let out: Vec<_> = all
.into_iter()
.filter(|e| f.repo.is_empty() || e.repo == f.repo)
.map(RefTrackingItem::from)
.collect();
Json(out)
}
// ── helpers ──────────────────────────────────────────────────────
/// Recursive byte count. Silent on read errors — used only for
/// reporting, not correctness.
fn dir_size_bytes(root: &std::path::Path) -> u64 {
let mut total: u64 = 0;
let mut stack = vec![root.to_path_buf()];
while let Some(dir) = stack.pop() {
let entries = match std::fs::read_dir(&dir) {
Ok(e) => e,
Err(_) => continue,
};
for entry in entries.flatten() {
let ft = match entry.file_type() {
Ok(t) => t,
Err(_) => continue,
};
if ft.is_dir() {
stack.push(entry.path());
} else if ft.is_file() {
if let Ok(m) = entry.metadata() {
total = total.saturating_add(m.len());
}
}
}
}
total
}
/// Sequentially resolve N per-item async closures. Kept simple —
/// blob-metadata reads are I/O light and page-sized (200 default);
/// concurrent-fan-out isn't worth the complexity here.
async fn futures_join<F, T>(iter: impl Iterator<Item = F>) -> Vec<T>
where
F: std::future::Future<Output = T>,
{
let mut out = Vec::new();
for f in iter {
out.push(f.await);
}
out
}
// ── route registration ──────────────────────────────────────────
/// Attach the v2 route tree onto an existing axum Router.
///
/// Caller supplies the v2 state via `.with_state(state)` when
/// building the final router.
pub fn routes() -> Router<Arc<V2State>> {
Router::new()
.route("/api/v2/node/local/status", get(handle_status))
.route("/api/v2/node/:name/status", get(handle_node_status))
.route("/api/v2/storage/blobs", get(handle_blobs))
.route("/api/v2/storage/tags", get(handle_tags))
.route("/api/v2/storage/refs", get(handle_refs))
.route("/api/v2/storage/snapshots", get(handle_snapshots))
.route("/api/v2/storage/ref-tracking", get(handle_ref_tracking))
}
#[cfg(test)]
mod tests {
use super::*;
use axum::body::to_bytes;
use axum::{body::Body, http::Request};
use tempfile::TempDir;
use tower::ServiceExt;
async fn seed_state() -> (TempDir, Arc<V2State>) {
let tmp = TempDir::new().unwrap();
let root = tmp.path().to_path_buf();
// Seed one blob so /status has non-zero numbers.
let store = BlobStore::open(root.clone()).unwrap();
store.put_bytes(b"hello v2").await.unwrap();
let snap = SnapshotStore::open(root.clone()).unwrap();
snap.create("seed", &store, 42).await.unwrap();
let state = Arc::new(V2State {
node_name: "test-node".into(),
blob_store_root: Some(root),
});
(tmp, state)
}
#[tokio::test]
async fn status_reports_seeded_blob_and_snapshot() {
let (_tmp, state) = seed_state().await;
let app: Router = routes().with_state(state);
let resp = app
.oneshot(
Request::builder()
.uri("/api/v2/node/local/status")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(resp.status(), axum::http::StatusCode::OK);
let bytes = to_bytes(resp.into_body(), 65_536).await.unwrap();
let body: NodeStatusV2 = serde_json::from_slice(&bytes).unwrap();
assert_eq!(body.node_name, "test-node");
assert_eq!(body.blob_count, 1);
assert_eq!(body.snapshot_count, 1);
assert!(body.blob_store_bytes > 0);
}
#[tokio::test]
async fn cross_node_status_returns_not_implemented() {
let (_tmp, state) = seed_state().await;
let app: Router = routes().with_state(state);
let resp = app
.oneshot(
Request::builder()
.uri("/api/v2/node/architect/status")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(resp.status(), axum::http::StatusCode::NOT_IMPLEMENTED);
}
#[tokio::test]
async fn blobs_endpoint_paginates() {
let (_tmp, state) = seed_state().await;
// Seed more blobs.
if let Some(root) = &state.blob_store_root {
let store = BlobStore::open(root.clone()).unwrap();
for i in 0..5 {
store
.put_bytes(format!("payload-{i}").as_bytes())
.await
.unwrap();
}
}
let app: Router = routes().with_state(state);
let resp = app
.oneshot(
Request::builder()
.uri("/api/v2/storage/blobs?limit=2&offset=1")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(resp.status(), axum::http::StatusCode::OK);
let bytes = to_bytes(resp.into_body(), 65_536).await.unwrap();
let body: Vec<BlobSummary> = serde_json::from_slice(&bytes).unwrap();
assert_eq!(body.len(), 2);
}
#[tokio::test]
async fn tags_endpoint_filters_by_prefix() {
let (_tmp, state) = seed_state().await;
if let Some(root) = &state.blob_store_root {
let ts = TagStore::open(root.join("tags-db")).unwrap();
ts.put("clawverse:main:latest", &[0xAA; 32]).await.unwrap();
ts.put("other:tag", &[0xBB; 32]).await.unwrap();
}
let app: Router = routes().with_state(state);
let resp = app
.oneshot(
Request::builder()
.uri("/api/v2/storage/tags?prefix=clawverse:")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
let bytes = to_bytes(resp.into_body(), 65_536).await.unwrap();
let body: Vec<TagSummaryV2> = serde_json::from_slice(&bytes).unwrap();
assert_eq!(body.len(), 1);
assert_eq!(body[0].key, "clawverse:main:latest");
}
#[tokio::test]
async fn snapshots_endpoint_returns_seeded() {
let (_tmp, state) = seed_state().await;
let app: Router = routes().with_state(state);
let resp = app
.oneshot(
Request::builder()
.uri("/api/v2/storage/snapshots")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
let bytes = to_bytes(resp.into_body(), 65_536).await.unwrap();
let body: Vec<SnapshotSummaryV2> = serde_json::from_slice(&bytes).unwrap();
assert_eq!(body.len(), 1);
assert_eq!(body[0].name, "seed");
assert_eq!(body[0].created_at_unix, 42);
}
#[tokio::test]
async fn ref_tracking_filters_by_repo() {
let (_tmp, state) = seed_state().await;
if let Some(root) = &state.blob_store_root {
let rt = RefTracking::open(root.clone()).unwrap();
rt.record([0x11; 32], "r/x", "main", 100).await.unwrap();
rt.record([0x22; 32], "r/y", "main", 100).await.unwrap();
}
let app: Router = routes().with_state(state);
let resp = app
.oneshot(
Request::builder()
.uri("/api/v2/storage/ref-tracking?repo=r/x")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
let bytes = to_bytes(resp.into_body(), 65_536).await.unwrap();
let body: Vec<RefTrackingItem> = serde_json::from_slice(&bytes).unwrap();
assert_eq!(body.len(), 1);
assert_eq!(body[0].repo, "r/x");
}
}
+98
View File
@@ -0,0 +1,98 @@
# dashboard-v2 — single-pane-of-glass command center
Redesign of the legacy `claw-store serve` dashboard for the current
distributed architecture.
## Design goals
1. **One URL for the whole fleet.** Hit any node's `:7700`; that node
fans out to every peer via existing QUIC RPC and serves an
aggregated view. No "3 browser tabs" pattern.
2. **Command-center landing page.** At-a-glance health strip + key
metrics + recent-events feed. Operator answers "is anything on
fire?" in <2 s.
3. **Node-detail drill-down.** Click any node in the strip → detail
page with per-node metrics, timers, storage counts, journal tail.
4. **Cross-cutting content browsers.** Blobs / Tags / Refs / Snapshots
aggregated across the fleet, searchable, click-through to detail.
5. **Trigger actions from the UI.** Scrub, GC, snapshot-create,
pin/unpin — anything currently a CLI invocation.
## Non-goals (v2 scope)
- Real-time streaming metrics beyond SSE snapshots. Prometheus stays
the source of truth for graphs; this dashboard is for state +
actions, not observability.
- Auth beyond a shared bearer token (fleet is trust-perimeter — CA
auth for the UI is future work).
- Editing configs. Read + trigger, never write config.
## Backend shape
Additive `/api/v2/*` alongside the legacy `/api/*` handlers so the
cutover is safe:
```
/api/v2/fleet aggregated snapshot (all peers)
/api/v2/node/<name>/status this-peer or remote via gossip lookup
/api/v2/node/<name>/timers systemd timer state for the 4 timers
/api/v2/node/<name>/journal?unit=… last N lines of journalctl
/api/v2/storage/blobs?limit&offset BlobStore::list_blob_ids + summaries
/api/v2/storage/tags?prefix TagStore::list (both layers)
/api/v2/storage/refs?limit&offset RefStore::list unioned
/api/v2/storage/snapshots SnapshotStore::list
/api/v2/storage/ref-tracking?repo RefTracking::list_all
/api/v2/cache/metrics router.metrics().snapshot() + gossiped
/api/v2/peers gossip snapshot + last-probe route
/api/v2/events SSE — fleet event stream
POST /api/v2/actions/scrub
POST /api/v2/actions/gc { evict_to_gb? }
POST /api/v2/actions/snapshot { name }
POST /api/v2/actions/pin { key, blob_id_hex }
POST /api/v2/actions/unpin { key }
```
## Aggregation model
**Server-side fan-out.** When the dashboard requests `/api/v2/fleet`,
the serving node walks its gossip peer list and issues a
`PeerStatus` RPC to each. Results collated into one JSON.
Trade-offs:
- Simpler frontend (no per-peer TLS material in browser).
- Backend caches results per-endpoint (5 s TTL) so 10 dashboard
tabs don't cause 30 peer RPCs.
- Any node can serve the dashboard — no "coordinator" single point
of failure.
## Frontend shape (implementation PR follows)
Routes:
```
/ → CommandCenter
/nodes/<name> → NodeDetail
/storage/blobs → StorageBrowser (blobs tab)
/storage/tags → StorageBrowser (tags tab)
/storage/refs → StorageBrowser (refs tab)
/storage/snapshots → StorageBrowser (snapshots tab)
/refs/tracking → RefTracking
/ops → OpsPanel (timers + actions + journal)
```
React + Vite + Tailwind, single SPA served from `/usr/share/claw-store/static-v2/`.
## Cutover plan
1. Ship v2 backend endpoints — legacy `/api/*` untouched.
2. Ship v2 frontend at `dashboard-v2/`, built to `static-v2/`.
3. `claw-store serve --v2-static-dir <path>` — new flag serves v2 assets at `/v2` while `/` still serves legacy for a burn-in period.
4. After burn-in: swap defaults; legacy accessible at `/legacy`.
5. Remove legacy after 30 days.
## Auth
Config-driven: `[dashboard] api_token = "..."`. Bearer required for
all POST endpoints; GET endpoints open on trusted-fleet networks
(tailscale + LAN). If token absent → POSTs disabled entirely
(read-only dashboard). Same shape as the legacy dashboard.