feat: time-series history infrastructure with sparklines (Stage 1)
Add in-process ring buffer history for per-worker and per-actor stats, with sparkline rendering on both web (SVG) and TUI (ratatui) surfaces. - history.rs: DashboardHistory with bounded VecDeque ring buffers, LRU actor eviction, thread-safe RwLock, 4 unit tests - SSE producer records stats every tick, sends initial history snapshot - /api/history REST endpoint for on-demand history JSON - Web: inline SVG sparklines in worker detail headers (rates + mailbox) - TUI: per-worker Sparkline widgets in overview layout Authored by Claude, lovingly guided by Zachery Aaron Shores-Chmielewski
This commit is contained in:
parent
a1f5f581fe
commit
144b4ca5ef
6 changed files with 504 additions and 4 deletions
|
|
@ -113,6 +113,8 @@ pub const DASHBOARD_HTML: &str = r##"<!DOCTYPE html>
|
|||
.worker-group-header .wid { font-weight: 700; }
|
||||
.worker-group-header .summary { color: #888; font-size: 11px; }
|
||||
.worker-group-header .toggle { color: #555; font-size: 14px; }
|
||||
.worker-group-header .sparkline-wrap { display: inline-flex; gap: 8px; margin-left: 12px; }
|
||||
.worker-group-header .sparkline-wrap svg { vertical-align: middle; }
|
||||
.worker-group-body { display: none; }
|
||||
.worker-group.open .worker-group-body { display: block; }
|
||||
.worker-group-body table { width: 100%; border-collapse: collapse; }
|
||||
|
|
@ -203,6 +205,7 @@ pub const DASHBOARD_HTML: &str = r##"<!DOCTYPE html>
|
|||
var isReplay = (DASHBOARD_MODE === 'replay');
|
||||
var lastUptimeMs = null;
|
||||
var lastStatsTime = null;
|
||||
var workerHistory = {}; // { id: { message_rates: [], mailbox_depths: [] } }
|
||||
|
||||
var dot = document.getElementById('statusDot');
|
||||
var uptimeLabel = document.getElementById('uptimeLabel');
|
||||
|
|
@ -347,6 +350,34 @@ pub const DASHBOARD_HTML: &str = r##"<!DOCTYPE html>
|
|||
return parts[parts.length - 1];
|
||||
}
|
||||
|
||||
function renderSparklineSvg(data, w, h, color) {
|
||||
if (!data || data.length < 2) return '';
|
||||
var max = Math.max.apply(null, data);
|
||||
if (max === 0) max = 1;
|
||||
var step = w / (data.length - 1);
|
||||
var points = data.map(function(v, i) {
|
||||
return (i * step).toFixed(1) + ',' + (h - (v / max) * (h - 2) - 1).toFixed(1);
|
||||
}).join(' ');
|
||||
return '<svg width="' + w + '" height="' + h + '" style="vertical-align:middle">' +
|
||||
'<polyline fill="none" stroke="' + color + '" stroke-width="1.5" points="' + points + '"/></svg>';
|
||||
}
|
||||
|
||||
function pushHistorySample(stats) {
|
||||
if (!stats.workers) return;
|
||||
stats.workers.forEach(function(w) {
|
||||
if (!workerHistory[w.id]) {
|
||||
workerHistory[w.id] = { message_rates: [], mailbox_depths: [], prev_msgs: w.messages_processed };
|
||||
}
|
||||
var wh = workerHistory[w.id];
|
||||
var rate = w.messages_processed - wh.prev_msgs;
|
||||
if (rate < 0) rate = 0;
|
||||
wh.prev_msgs = w.messages_processed;
|
||||
wh.message_rates.push(rate);
|
||||
wh.mailbox_depths.push(w.mailbox_depth);
|
||||
if (wh.message_rates.length > 300) { wh.message_rates.shift(); wh.mailbox_depths.shift(); }
|
||||
});
|
||||
}
|
||||
|
||||
function updateWorkerDetails(data) {
|
||||
var container = document.getElementById('workerDetailContainer');
|
||||
if (!data.workers) return;
|
||||
|
|
@ -377,8 +408,17 @@ pub const DASHBOARD_HTML: &str = r##"<!DOCTYPE html>
|
|||
var hdr = document.createElement('div');
|
||||
hdr.className = 'worker-group-header';
|
||||
var panicHtml = g.info.panics > 0 ? ', <span style="color:#f44336">' + g.info.panics + ' panics</span>' : '';
|
||||
var wh = workerHistory[wid];
|
||||
var sparkHtml = '';
|
||||
if (wh) {
|
||||
sparkHtml = '<span class="sparkline-wrap">' +
|
||||
renderSparklineSvg(wh.message_rates, 80, 16, '#4caf50') +
|
||||
renderSparklineSvg(wh.mailbox_depths, 80, 16, '#2196f3') +
|
||||
'</span>';
|
||||
}
|
||||
hdr.innerHTML =
|
||||
'<span class="wid" style="color:' + colors[wid % colors.length] + '">W' + wid + '</span>' +
|
||||
sparkHtml +
|
||||
'<span class="summary">' + g.actors.length + ' actors, ' +
|
||||
g.info.messages_processed.toLocaleString() + ' msgs, mbox ' + g.info.mailbox_depth + panicHtml + '</span>' +
|
||||
'<span class="toggle">' + (isOpen ? '\u25BC' : '\u25B6') + '</span>';
|
||||
|
|
@ -467,7 +507,26 @@ pub const DASHBOARD_HTML: &str = r##"<!DOCTYPE html>
|
|||
var es = new EventSource('/events');
|
||||
|
||||
es.addEventListener('stats', function(e) {
|
||||
try { updateStats(JSON.parse(e.data)); } catch(err) { console.error('stats parse error', err); }
|
||||
try {
|
||||
var data = JSON.parse(e.data);
|
||||
pushHistorySample(data);
|
||||
updateStats(data);
|
||||
} catch(err) { console.error('stats parse error', err); }
|
||||
});
|
||||
|
||||
es.addEventListener('history', function(e) {
|
||||
try {
|
||||
var data = JSON.parse(e.data);
|
||||
if (data.workers) {
|
||||
data.workers.forEach(function(w) {
|
||||
workerHistory[w.id] = {
|
||||
message_rates: w.message_rates || [],
|
||||
mailbox_depths: w.mailbox_depths || [],
|
||||
prev_msgs: 0
|
||||
};
|
||||
});
|
||||
}
|
||||
} catch(err) { console.error('history parse error', err); }
|
||||
});
|
||||
|
||||
es.addEventListener('activity', function(e) {
|
||||
|
|
|
|||
343
crates/runtime-dashboard/src/history.rs
Normal file
343
crates/runtime-dashboard/src/history.rs
Normal file
|
|
@ -0,0 +1,343 @@
|
|||
//! In-process time-series history for dashboard sparklines and trend detection.
|
||||
//!
|
||||
//! Stores bounded ring buffers of per-worker and per-actor stats, sampled at
|
||||
//! a configurable interval. All data is kept in memory with automatic eviction
|
||||
//! of the oldest samples when capacity is reached.
|
||||
|
||||
use std::collections::{HashMap, VecDeque};
|
||||
use std::sync::RwLock;
|
||||
|
||||
use swactor::actor::ActorAddress;
|
||||
use swactor::stats::{ActorInfo, RuntimeStats};
|
||||
|
||||
/// Configuration for history collection.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct HistoryConfig {
|
||||
/// Maximum samples per worker (default: 300 = 5 min at 1/sec).
|
||||
pub max_worker_samples: usize,
|
||||
/// Maximum samples per actor (default: 300).
|
||||
pub max_actor_samples: usize,
|
||||
/// Maximum number of actors tracked (LRU eviction). Default: 1000.
|
||||
pub max_tracked_actors: usize,
|
||||
}
|
||||
|
||||
impl Default for HistoryConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
max_worker_samples: 300,
|
||||
max_actor_samples: 300,
|
||||
max_tracked_actors: 1000,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Time-series data for a single worker.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct WorkerHistory {
|
||||
pub message_rates: VecDeque<f64>,
|
||||
pub mailbox_depths: VecDeque<u64>,
|
||||
pub actor_counts: VecDeque<u32>,
|
||||
prev_messages: u64,
|
||||
}
|
||||
|
||||
impl WorkerHistory {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
message_rates: VecDeque::new(),
|
||||
mailbox_depths: VecDeque::new(),
|
||||
actor_counts: VecDeque::new(),
|
||||
prev_messages: 0,
|
||||
}
|
||||
}
|
||||
|
||||
fn push(&mut self, messages_processed: u64, mailbox_depth: usize, num_actors: usize, cap: usize) {
|
||||
let rate = messages_processed.saturating_sub(self.prev_messages) as f64;
|
||||
self.prev_messages = messages_processed;
|
||||
|
||||
push_bounded(&mut self.message_rates, rate, cap);
|
||||
push_bounded(&mut self.mailbox_depths, mailbox_depth as u64, cap);
|
||||
push_bounded(&mut self.actor_counts, num_actors as u32, cap);
|
||||
}
|
||||
}
|
||||
|
||||
/// Time-series data for a single actor.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ActorHistory {
|
||||
pub mailbox_depths: VecDeque<u64>,
|
||||
pub message_rates: VecDeque<f64>,
|
||||
prev_messages: u64,
|
||||
last_seen_sample: u64,
|
||||
}
|
||||
|
||||
impl ActorHistory {
|
||||
fn new(sample_counter: u64) -> Self {
|
||||
Self {
|
||||
mailbox_depths: VecDeque::new(),
|
||||
message_rates: VecDeque::new(),
|
||||
prev_messages: 0,
|
||||
last_seen_sample: sample_counter,
|
||||
}
|
||||
}
|
||||
|
||||
fn push(&mut self, info: &ActorInfo, cap: usize, sample_counter: u64) {
|
||||
let rate = info.messages_processed.saturating_sub(self.prev_messages) as f64;
|
||||
self.prev_messages = info.messages_processed;
|
||||
self.last_seen_sample = sample_counter;
|
||||
|
||||
push_bounded(&mut self.mailbox_depths, info.mailbox_depth as u64, cap);
|
||||
push_bounded(&mut self.message_rates, rate, cap);
|
||||
}
|
||||
}
|
||||
|
||||
fn push_bounded<T>(buf: &mut VecDeque<T>, val: T, cap: usize) {
|
||||
if buf.len() >= cap {
|
||||
buf.pop_front();
|
||||
}
|
||||
buf.push_back(val);
|
||||
}
|
||||
|
||||
/// Thread-safe history store. Written by the sampler, read by SSE/TUI.
|
||||
pub struct DashboardHistory {
|
||||
inner: RwLock<HistoryInner>,
|
||||
config: HistoryConfig,
|
||||
}
|
||||
|
||||
struct HistoryInner {
|
||||
workers: Vec<WorkerHistory>,
|
||||
actors: HashMap<ActorAddress, ActorHistory>,
|
||||
sample_counter: u64,
|
||||
}
|
||||
|
||||
impl DashboardHistory {
|
||||
pub fn new(config: HistoryConfig) -> Self {
|
||||
Self {
|
||||
inner: RwLock::new(HistoryInner {
|
||||
workers: Vec::new(),
|
||||
actors: HashMap::new(),
|
||||
sample_counter: 0,
|
||||
}),
|
||||
config,
|
||||
}
|
||||
}
|
||||
|
||||
/// Record a stats snapshot. Called by the sampler thread.
|
||||
pub fn record(&self, stats: &RuntimeStats) {
|
||||
let mut inner = self.inner.write().unwrap();
|
||||
inner.sample_counter += 1;
|
||||
let counter = inner.sample_counter;
|
||||
|
||||
// Resize workers vec if needed
|
||||
while inner.workers.len() < stats.workers.len() {
|
||||
inner.workers.push(WorkerHistory::new());
|
||||
}
|
||||
|
||||
// Record per-worker data
|
||||
for w in &stats.workers {
|
||||
if let Some(wh) = inner.workers.get_mut(w.id) {
|
||||
wh.push(
|
||||
w.messages_processed,
|
||||
w.mailbox_depth,
|
||||
w.num_actors,
|
||||
self.config.max_worker_samples,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// Record per-actor data
|
||||
for a in &stats.actor_details {
|
||||
let ah = inner.actors.entry(a.address).or_insert_with(|| ActorHistory::new(counter));
|
||||
ah.push(a, self.config.max_actor_samples, counter);
|
||||
}
|
||||
|
||||
// LRU eviction: remove actors not seen recently if over capacity
|
||||
if inner.actors.len() > self.config.max_tracked_actors {
|
||||
let mut entries: Vec<(ActorAddress, u64)> = inner
|
||||
.actors
|
||||
.iter()
|
||||
.map(|(addr, ah)| (*addr, ah.last_seen_sample))
|
||||
.collect();
|
||||
entries.sort_by_key(|&(_, seen)| seen);
|
||||
let to_remove = inner.actors.len() - self.config.max_tracked_actors;
|
||||
for (addr, _) in entries.into_iter().take(to_remove) {
|
||||
inner.actors.remove(&addr);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Get a snapshot of worker history for rendering sparklines.
|
||||
/// Returns Vec indexed by worker_id, each containing recent message rates.
|
||||
pub fn worker_sparklines(&self) -> Vec<Vec<u64>> {
|
||||
let inner = self.inner.read().unwrap();
|
||||
inner
|
||||
.workers
|
||||
.iter()
|
||||
.map(|wh| wh.message_rates.iter().map(|r| *r as u64).collect())
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Get worker mailbox depth history.
|
||||
pub fn worker_mailbox_sparklines(&self) -> Vec<Vec<u64>> {
|
||||
let inner = self.inner.read().unwrap();
|
||||
inner
|
||||
.workers
|
||||
.iter()
|
||||
.map(|wh| wh.mailbox_depths.iter().copied().collect())
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Get sparkline data for a specific actor.
|
||||
pub fn actor_sparkline(&self, addr: &ActorAddress) -> Option<(Vec<u64>, Vec<u64>)> {
|
||||
let inner = self.inner.read().unwrap();
|
||||
inner.actors.get(addr).map(|ah| {
|
||||
let mailbox: Vec<u64> = ah.mailbox_depths.iter().copied().collect();
|
||||
let rates: Vec<u64> = ah.message_rates.iter().map(|r| *r as u64).collect();
|
||||
(mailbox, rates)
|
||||
})
|
||||
}
|
||||
|
||||
/// Get total sample count (useful for knowing if history is available).
|
||||
pub fn sample_count(&self) -> u64 {
|
||||
self.inner.read().unwrap().sample_counter
|
||||
}
|
||||
|
||||
/// Serialize worker history as JSON for the SSE initial payload.
|
||||
pub fn worker_history_json(&self) -> String {
|
||||
let sparklines = self.worker_sparklines();
|
||||
let mailbox = self.worker_mailbox_sparklines();
|
||||
serde_json::json!({
|
||||
"workers": sparklines.iter().enumerate().map(|(i, rates)| {
|
||||
serde_json::json!({
|
||||
"id": i,
|
||||
"message_rates": rates,
|
||||
"mailbox_depths": mailbox.get(i).unwrap_or(&Vec::new()),
|
||||
})
|
||||
}).collect::<Vec<_>>(),
|
||||
})
|
||||
.to_string()
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use swactor::stats::{ActorInfo, WorkerInfo};
|
||||
|
||||
fn make_stats(workers: Vec<(u64, usize, usize)>, actors: Vec<ActorInfo>) -> RuntimeStats {
|
||||
RuntimeStats {
|
||||
num_workers: workers.len(),
|
||||
uptime_ms: 0,
|
||||
actors: actors.iter().map(|a| (a.address, a.worker_id)).collect(),
|
||||
workers: workers
|
||||
.into_iter()
|
||||
.enumerate()
|
||||
.map(|(id, (msgs, depth, n_actors))| WorkerInfo {
|
||||
id,
|
||||
num_actors: n_actors,
|
||||
mailbox_depth: depth,
|
||||
messages_processed: msgs,
|
||||
local_sends: 0,
|
||||
cross_sends: 0,
|
||||
inbox_sends: 0,
|
||||
type_mismatches: 0,
|
||||
panics: 0,
|
||||
messages_dropped: 0,
|
||||
restarts: 0,
|
||||
stops: 0,
|
||||
})
|
||||
.collect(),
|
||||
actor_details: actors,
|
||||
tick_timings: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
fn make_actor(id: u8, worker: usize, depth: usize, msgs: u64) -> ActorInfo {
|
||||
ActorInfo {
|
||||
address: ActorAddress([id; 32]),
|
||||
worker_id: worker,
|
||||
mailbox_depth: depth,
|
||||
last_msg_type: None,
|
||||
messages_processed: msgs,
|
||||
poisoned: false,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn worker_rates_accumulate_over_samples() {
|
||||
let history = DashboardHistory::new(HistoryConfig::default());
|
||||
|
||||
// First sample: establishes baseline (rate will be the raw value since prev=0)
|
||||
let stats1 = make_stats(vec![(100, 5, 2)], vec![]);
|
||||
history.record(&stats1);
|
||||
|
||||
// Second sample: delta = 150 - 100 = 50
|
||||
let stats2 = make_stats(vec![(150, 3, 2)], vec![]);
|
||||
history.record(&stats2);
|
||||
|
||||
let sparklines = history.worker_sparklines();
|
||||
assert_eq!(sparklines.len(), 1);
|
||||
assert_eq!(sparklines[0].len(), 2);
|
||||
assert_eq!(sparklines[0][0], 100); // first sample: 100 - 0
|
||||
assert_eq!(sparklines[0][1], 50); // second sample: 150 - 100
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bounded_eviction_drops_oldest() {
|
||||
let config = HistoryConfig {
|
||||
max_worker_samples: 3,
|
||||
..Default::default()
|
||||
};
|
||||
let history = DashboardHistory::new(config);
|
||||
|
||||
for i in 0..5u64 {
|
||||
let stats = make_stats(vec![(i * 10, 0, 0)], vec![]);
|
||||
history.record(&stats);
|
||||
}
|
||||
|
||||
let sparklines = history.worker_sparklines();
|
||||
assert_eq!(sparklines[0].len(), 3); // capped at 3
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn actor_lru_eviction_keeps_most_recent() {
|
||||
let config = HistoryConfig {
|
||||
max_tracked_actors: 2,
|
||||
..Default::default()
|
||||
};
|
||||
let history = DashboardHistory::new(config);
|
||||
|
||||
// Sample 1: actors A and B
|
||||
let stats1 = make_stats(
|
||||
vec![(0, 0, 2)],
|
||||
vec![make_actor(1, 0, 0, 0), make_actor(2, 0, 0, 0)],
|
||||
);
|
||||
history.record(&stats1);
|
||||
|
||||
// Sample 2: actors B and C (A not seen)
|
||||
let stats2 = make_stats(
|
||||
vec![(0, 0, 2)],
|
||||
vec![make_actor(2, 0, 0, 0), make_actor(3, 0, 0, 0)],
|
||||
);
|
||||
history.record(&stats2);
|
||||
|
||||
// A should be evicted (LRU), B and C kept
|
||||
assert!(history.actor_sparkline(&ActorAddress([1; 32])).is_none());
|
||||
assert!(history.actor_sparkline(&ActorAddress([2; 32])).is_some());
|
||||
assert!(history.actor_sparkline(&ActorAddress([3; 32])).is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn actor_rates_track_deltas() {
|
||||
let history = DashboardHistory::new(HistoryConfig::default());
|
||||
|
||||
let stats1 = make_stats(vec![(0, 0, 1)], vec![make_actor(1, 0, 5, 100)]);
|
||||
history.record(&stats1);
|
||||
|
||||
let stats2 = make_stats(vec![(0, 0, 1)], vec![make_actor(1, 0, 3, 175)]);
|
||||
history.record(&stats2);
|
||||
|
||||
let (mailbox, rates) = history.actor_sparkline(&ActorAddress([1; 32])).unwrap();
|
||||
assert_eq!(mailbox, vec![5, 3]);
|
||||
assert_eq!(rates[0], 100); // first: 100 - 0
|
||||
assert_eq!(rates[1], 75); // second: 175 - 100
|
||||
}
|
||||
}
|
||||
|
|
@ -1,4 +1,5 @@
|
|||
pub mod collector;
|
||||
pub mod history;
|
||||
pub mod investigate;
|
||||
pub mod layer;
|
||||
pub mod trace;
|
||||
|
|
@ -27,6 +28,7 @@ use tracing_subscriber::layer::SubscriberExt;
|
|||
use tracing_subscriber::util::SubscriberInitExt;
|
||||
|
||||
use crate::collector::StatsCollector;
|
||||
use crate::history::{DashboardHistory, HistoryConfig};
|
||||
use crate::layer::{now_ms, DashboardLayer, EventStore};
|
||||
use crate::trace::{RuntimeTrace, TimestampedStats};
|
||||
|
||||
|
|
@ -80,6 +82,7 @@ pub struct DashboardHandle {
|
|||
collector: Arc<Mutex<Option<Arc<StatsCollector>>>>,
|
||||
shutdown: Arc<AtomicBool>,
|
||||
stats_timeline: Arc<ArrayQueue<TimestampedStats>>,
|
||||
history: Arc<DashboardHistory>,
|
||||
recording: bool,
|
||||
#[cfg(feature = "distribution")]
|
||||
distribution: Arc<Mutex<Option<Arc<dyn distribution_collector::DistributionStatsProvider>>>>,
|
||||
|
|
@ -115,6 +118,11 @@ impl DashboardHandle {
|
|||
*self.distribution.lock().unwrap() = Some(provider);
|
||||
}
|
||||
|
||||
/// Access the time-series history store (for TUI sparklines, etc.).
|
||||
pub fn history(&self) -> &Arc<DashboardHistory> {
|
||||
&self.history
|
||||
}
|
||||
|
||||
/// Signal the dashboard to shut down (SSE clients receive "done").
|
||||
pub fn shutdown(&self) {
|
||||
self.shutdown.store(true, Ordering::Release);
|
||||
|
|
@ -159,6 +167,7 @@ pub fn start_dashboard(config: DashboardConfig) -> DashboardHandle {
|
|||
let collector: Arc<Mutex<Option<Arc<StatsCollector>>>> = Arc::new(Mutex::new(None));
|
||||
let shutdown = Arc::new(AtomicBool::new(false));
|
||||
let stats_timeline = Arc::new(ArrayQueue::new(config.record_stats_capacity.max(1)));
|
||||
let history = Arc::new(DashboardHistory::new(HistoryConfig::default()));
|
||||
|
||||
#[cfg(feature = "distribution")]
|
||||
let distribution: Arc<Mutex<Option<Arc<dyn distribution_collector::DistributionStatsProvider>>>> =
|
||||
|
|
@ -169,6 +178,7 @@ pub fn start_dashboard(config: DashboardConfig) -> DashboardHandle {
|
|||
Arc::clone(&runtime),
|
||||
Arc::clone(&collector),
|
||||
Arc::clone(&shutdown),
|
||||
Arc::clone(&history),
|
||||
config.port,
|
||||
#[cfg(feature = "distribution")]
|
||||
Arc::clone(&distribution),
|
||||
|
|
@ -210,6 +220,7 @@ pub fn start_dashboard(config: DashboardConfig) -> DashboardHandle {
|
|||
collector,
|
||||
shutdown,
|
||||
stats_timeline,
|
||||
history,
|
||||
recording: config.record,
|
||||
#[cfg(feature = "distribution")]
|
||||
distribution,
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ use swactor::runtime::Runtime;
|
|||
use crate::actors_html::ACTORS_HTML;
|
||||
use crate::collector::StatsCollector;
|
||||
use crate::dashboard_html::DASHBOARD_HTML;
|
||||
use crate::history::DashboardHistory;
|
||||
use crate::layer::EventStore;
|
||||
use crate::trace::RuntimeTrace;
|
||||
|
||||
|
|
@ -119,6 +120,7 @@ pub(crate) fn spawn_http_server(
|
|||
runtime: Arc<Mutex<Option<Arc<Runtime>>>>,
|
||||
collector: Arc<Mutex<Option<Arc<StatsCollector>>>>,
|
||||
shutdown: Arc<AtomicBool>,
|
||||
history: Arc<DashboardHistory>,
|
||||
port: u16,
|
||||
#[cfg(feature = "distribution")]
|
||||
distribution: Arc<Mutex<Option<Arc<dyn DistributionStatsProvider>>>>,
|
||||
|
|
@ -134,6 +136,7 @@ pub(crate) fn spawn_http_server(
|
|||
let runtime = Arc::clone(&runtime);
|
||||
let collector = Arc::clone(&collector);
|
||||
let shutdown = Arc::clone(&shutdown);
|
||||
let history = Arc::clone(&history);
|
||||
let cmd_router = Arc::clone(&cmd_router);
|
||||
#[cfg(feature = "distribution")]
|
||||
let distribution = Arc::clone(&distribution);
|
||||
|
|
@ -158,6 +161,7 @@ pub(crate) fn spawn_http_server(
|
|||
Arc::clone(&runtime),
|
||||
Arc::clone(&collector),
|
||||
Arc::clone(&shutdown),
|
||||
Arc::clone(&history),
|
||||
#[cfg(feature = "distribution")]
|
||||
Arc::clone(&distribution),
|
||||
);
|
||||
|
|
@ -169,6 +173,9 @@ pub(crate) fn spawn_http_server(
|
|||
Arc::clone(&collector),
|
||||
);
|
||||
}
|
||||
"/api/history" => {
|
||||
handle_history_api(request, Arc::clone(&history));
|
||||
}
|
||||
"/api/investigate" => {
|
||||
handle_investigate_api(
|
||||
request,
|
||||
|
|
@ -198,6 +205,7 @@ fn handle_live_sse(
|
|||
runtime: Arc<Mutex<Option<Arc<Runtime>>>>,
|
||||
collector: Arc<Mutex<Option<Arc<StatsCollector>>>>,
|
||||
shutdown: Arc<AtomicBool>,
|
||||
history: Arc<DashboardHistory>,
|
||||
#[cfg(feature = "distribution")]
|
||||
distribution: Arc<Mutex<Option<Arc<dyn DistributionStatsProvider>>>>,
|
||||
) {
|
||||
|
|
@ -208,6 +216,12 @@ fn handle_live_sse(
|
|||
thread::spawn(move || {
|
||||
let mut cursor: u64 = 0;
|
||||
|
||||
// Send initial history snapshot so sparklines render immediately
|
||||
if history.sample_count() > 0 {
|
||||
let json = history.worker_history_json();
|
||||
let _ = tx.send(format_sse("history", &json));
|
||||
}
|
||||
|
||||
loop {
|
||||
// Send stats if runtime is available
|
||||
{
|
||||
|
|
@ -217,6 +231,7 @@ fn handle_live_sse(
|
|||
if let Some(col) = collector.lock().unwrap().as_ref() {
|
||||
col.enrich(&mut stats);
|
||||
}
|
||||
history.record(&stats);
|
||||
let json = serde_json::to_string(&stats).unwrap();
|
||||
if tx.send(format_sse("stats", &json)).is_err() {
|
||||
return;
|
||||
|
|
@ -353,6 +368,16 @@ fn handle_distribution_api(
|
|||
let _ = request.respond(response);
|
||||
}
|
||||
|
||||
fn handle_history_api(request: tiny_http::Request, history: Arc<DashboardHistory>) {
|
||||
let json = history.worker_history_json();
|
||||
let response = tiny_http::Response::from_string(json).with_header(
|
||||
"Content-Type: application/json"
|
||||
.parse::<tiny_http::Header>()
|
||||
.unwrap(),
|
||||
);
|
||||
let _ = request.respond(response);
|
||||
}
|
||||
|
||||
fn parse_query_string(url: &str) -> HashMap<String, String> {
|
||||
let mut params = HashMap::new();
|
||||
if let Some(qs) = url.split('?').nth(1) {
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
use std::collections::VecDeque;
|
||||
use std::time::Instant;
|
||||
|
||||
use crossterm::event::{KeyCode, KeyEvent, KeyModifiers};
|
||||
|
|
@ -15,6 +16,10 @@ pub struct WorkerView {
|
|||
/// Fraction of bar for each phase group: [processing, delivery, spawns, overhead]
|
||||
pub phase_fractions: [f64; 4],
|
||||
pub panics: u64,
|
||||
/// Recent message rates for sparkline rendering.
|
||||
pub sparkline_rates: Vec<u64>,
|
||||
/// Recent mailbox depths for sparkline rendering.
|
||||
pub sparkline_mailbox: Vec<u64>,
|
||||
}
|
||||
|
||||
/// One row in the actor table.
|
||||
|
|
@ -90,6 +95,10 @@ pub struct App {
|
|||
prev_time: Instant,
|
||||
/// Rolling msg rates (smoothed)
|
||||
msg_rates: Vec<f64>,
|
||||
/// Per-worker sparkline history (message rate deltas).
|
||||
sparkline_rates: Vec<VecDeque<u64>>,
|
||||
/// Per-worker sparkline history (mailbox depths).
|
||||
sparkline_mailbox: Vec<VecDeque<u64>>,
|
||||
}
|
||||
|
||||
impl App {
|
||||
|
|
@ -115,6 +124,8 @@ impl App {
|
|||
prev_messages: Vec::new(),
|
||||
prev_time: Instant::now(),
|
||||
msg_rates: Vec::new(),
|
||||
sparkline_rates: Vec::new(),
|
||||
sparkline_mailbox: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -144,6 +155,8 @@ impl App {
|
|||
if self.prev_messages.len() != stats.workers.len() {
|
||||
self.prev_messages = stats.workers.iter().map(|w| w.messages_processed).collect();
|
||||
self.msg_rates = vec![0.0; stats.workers.len()];
|
||||
self.sparkline_rates.resize_with(stats.workers.len(), VecDeque::new);
|
||||
self.sparkline_mailbox.resize_with(stats.workers.len(), VecDeque::new);
|
||||
}
|
||||
|
||||
// Compute per-worker views
|
||||
|
|
@ -163,8 +176,19 @@ impl App {
|
|||
self.msg_rates[i] * 0.6 + rate * 0.4
|
||||
};
|
||||
self.msg_rates[i] = smoothed;
|
||||
|
||||
// Track sparkline history
|
||||
let delta = w.messages_processed.saturating_sub(self.prev_messages[i]);
|
||||
self.prev_messages[i] = w.messages_processed;
|
||||
|
||||
let spark_rates = &mut self.sparkline_rates[i];
|
||||
if spark_rates.len() >= 60 { spark_rates.pop_front(); }
|
||||
spark_rates.push_back(delta);
|
||||
|
||||
let spark_mbox = &mut self.sparkline_mailbox[i];
|
||||
if spark_mbox.len() >= 60 { spark_mbox.pop_front(); }
|
||||
spark_mbox.push_back(w.mailbox_depth as u64);
|
||||
|
||||
// Load % and phase fractions from tick timings
|
||||
let timings = stats.tick_timings.get(i).map(|v| v.as_slice()).unwrap_or(&[]);
|
||||
let (load_pct, phase_fractions) = compute_load_and_phases(timings);
|
||||
|
|
@ -178,6 +202,8 @@ impl App {
|
|||
load_pct,
|
||||
phase_fractions,
|
||||
panics: w.panics,
|
||||
sparkline_rates: spark_rates.iter().copied().collect(),
|
||||
sparkline_mailbox: spark_mbox.iter().copied().collect(),
|
||||
});
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ use ratatui::Frame;
|
|||
use ratatui::layout::{Constraint, Layout, Rect};
|
||||
use ratatui::style::{Color, Modifier, Style};
|
||||
use ratatui::text::{Line, Span};
|
||||
use ratatui::widgets::{Block, Borders, Cell, Paragraph, Row, Table, TableState};
|
||||
use ratatui::widgets::{Block, Borders, Cell, Paragraph, Row, Sparkline, Table, TableState};
|
||||
|
||||
use super::app::{App, SortColumn, ViewMode};
|
||||
|
||||
|
|
@ -20,17 +20,23 @@ pub fn draw(f: &mut Frame, app: &App, table_state: &mut TableState) {
|
|||
|
||||
fn draw_overview(f: &mut Frame, app: &App, table_state: &mut TableState) {
|
||||
let num_workers = app.workers.len().max(1);
|
||||
let has_sparkline_data = app.workers.iter().any(|w| w.sparkline_rates.len() > 1);
|
||||
let sparkline_height = if has_sparkline_data { 4u16 } else { 0u16 };
|
||||
|
||||
let chunks = Layout::vertical([
|
||||
Constraint::Length(num_workers as u16),
|
||||
Constraint::Length(sparkline_height),
|
||||
Constraint::Length(1),
|
||||
Constraint::Fill(1),
|
||||
])
|
||||
.split(f.area());
|
||||
|
||||
draw_worker_bars(f, app, chunks[0]);
|
||||
draw_summary(f, app, chunks[1]);
|
||||
draw_actor_table(f, app, table_state, chunks[2]);
|
||||
if has_sparkline_data {
|
||||
draw_worker_sparklines(f, app, chunks[1]);
|
||||
}
|
||||
draw_summary(f, app, chunks[2]);
|
||||
draw_actor_table(f, app, table_state, chunks[3]);
|
||||
}
|
||||
|
||||
/// Render htop-style worker bars.
|
||||
|
|
@ -45,6 +51,36 @@ fn draw_worker_bars(f: &mut Frame, app: &App, area: Rect) {
|
|||
f.render_widget(paragraph, area);
|
||||
}
|
||||
|
||||
/// Render per-worker sparklines showing message rate trends.
|
||||
fn draw_worker_sparklines(f: &mut Frame, app: &App, area: Rect) {
|
||||
if app.workers.is_empty() {
|
||||
return;
|
||||
}
|
||||
// Split area horizontally: one sparkline per worker
|
||||
let constraints: Vec<Constraint> = app
|
||||
.workers
|
||||
.iter()
|
||||
.map(|_| Constraint::Ratio(1, app.workers.len() as u32))
|
||||
.collect();
|
||||
let cols = Layout::horizontal(constraints).split(area);
|
||||
|
||||
for (i, w) in app.workers.iter().enumerate() {
|
||||
if let Some(&col_area) = cols.get(i) {
|
||||
let block = Block::default()
|
||||
.borders(Borders::NONE)
|
||||
.title(Span::styled(
|
||||
format!(" W{} ", w.id),
|
||||
Style::default().fg(Color::DarkGray),
|
||||
));
|
||||
let sparkline = Sparkline::default()
|
||||
.block(block)
|
||||
.data(&w.sparkline_rates)
|
||||
.style(Style::default().fg(Color::Green));
|
||||
f.render_widget(sparkline, col_area);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn build_worker_line(w: &super::app::WorkerView, total_width: usize) -> Line<'static> {
|
||||
let id_str = format!("{:>3}", w.id);
|
||||
let suffix = format!(
|
||||
|
|
|
|||
Loading…
Reference in a new issue