1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
//! The host, sampled for the strip in the title bar.
//!
//! One stream rather than one route polled: the page wants a reading every
//! couple of seconds, and a poll costs a request, a connection and a set of
//! headers for every one of them. The sampler lives here rather than in the
//! page because the page cannot read `/proc`, and because one sampler serving
//! every open tab costs what one costs.
use axum::{
extract::State,
response::sse::{Event, KeepAlive, Sse},
Json,
};
use futures::stream::Stream;
use wsb::prelude::*;
use std::convert::Infallible;
use crate::web::machine_web::MachineWeb;
/// How often the host is read.
///
/// Two seconds: fast enough that a build starting is visible while somebody is
/// still looking at the strip, slow enough that the reading costs nothing. A
/// processor figure is a delta over this interval, so it is also the window
/// the number describes.
const INTERVAL: std::time::Duration = std::time::Duration::from_secs(2);
/// GET /api/v1/host — one reading, for a page that just opened.
///
/// The processor figure is absent here by construction: it is a delta, and a
/// single request has nothing to compare against. The page gets it from the
/// second frame of the stream, moments later.
pub async fn host_snapshot(State(state): State<MachineWeb>) -> Json<wsb::host::Sample> {
let root = state.volume_root.clone();
let sample = tokio::task::spawn_blocking(move || {
let mut sampler = wsb::host::Sampler::new();
sampler.sample(&root)
})
.await
.unwrap_or_default();
Json(sample)
}
/// GET /api/v1/host/stream — a reading every couple of seconds.
pub async fn host_stream(
State(state): State<MachineWeb>,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
let root = state.volume_root.clone();
let stream = async_stream::stream! {
// One sampler per subscriber, because the processor figure is a delta
// against the LAST reading this subscriber saw. Shared, a second tab
// opening would reset the first tab's baseline.
let mut sampler = wsb::host::Sampler::new();
loop {
let root = root.clone();
// Off the runtime: reading /proc is cheap, but `nvidia-smi` and
// `df` are processes, and a stalled disk must not stall the
// server.
let (next, sample) = match tokio::task::spawn_blocking(move || {
let mut moved = sampler;
let sample = moved.sample(&root);
(moved, sample)
})
.await
{
Ok(pair) => pair,
Err(error) => {
tracing::warn!(%error, "the host sampler panicked");
break;
}
};
sampler = next;
match serde_json::to_string(&sample) {
Ok(payload) => yield Ok(Event::default().event("host").data(payload)),
Err(error) => {
tracing::warn!(%error, "a host sample could not be encoded");
break;
}
}
tokio::time::sleep(INTERVAL).await;
}
};
Sse::new(stream).keep_alive(KeepAlive::default())
}