use crate::controller::{
self, send_command, ControlCommand, ControlReply, ControlRequest, ControllerCommand,
ListeningControllers, RunSettings, SendError, CONTROL_TIMEOUT,
};
use crate::metrics::dashboard_snapshot::{
Heatmap, HeatmapColumn, HeatmapCursor, RequestSeries, SnapshotSettings,
};
use crate::metrics::DashboardSnapshot;
use crate::runs;
use crate::{ControlOutcome, GooseConfiguration, GooseError};
use axum::body::{Body, Bytes};
use axum::extract::{Query, State};
use axum::http::{header, HeaderMap, HeaderValue, Response, StatusCode};
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::response::IntoResponse;
use axum::routing::{delete, get, post};
use axum::{Json, Router};
use futures::stream;
use serde::Deserialize;
use serde::Serialize;
use std::collections::{BTreeMap, HashMap};
use std::convert::Infallible;
use std::path::Path;
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tokio::io::AsyncReadExt;
use tokio::sync::{oneshot, watch, Notify};
use tower_http::set_header::SetResponseHeaderLayer;
const INDEX_HTML: &str = include_str!("dashboard/static/index.html");
const APP_JS: &str = include_str!("dashboard/static/app.js");
const APP_CSS: &str = include_str!("dashboard/static/app.css");
const CSP: &str =
"default-src 'self'; script-src 'self'; connect-src 'self'; style-src 'self'; img-src 'self' data:; frame-ancestors 'none'";
const MAX_SSE_CLIENTS: usize = 32;
const SNAPSHOT_INTERVAL: Duration = Duration::from_secs(1);
const RECENT_POLL_WINDOW: Duration = Duration::from_secs(2);
const SSE_HEARTBEAT: Duration = Duration::from_secs(15);
const SNAPSHOT_BUILD_TIMEOUT: Duration = Duration::from_secs(4);
const SERVER_DRAIN_TIMEOUT: Duration = Duration::from_secs(1);
const HEADER_READ_TIMEOUT: Duration = Duration::from_secs(10);
const EXTRA_CONNECTIONS: usize = 16;
const CONTROL_BODY_LIMIT: usize = 16 * 1024;
const MAX_CONTROL_IN_FLIGHT: usize = 4;
const MAX_SERIES_IN_FLIGHT: usize = 4;
const MAX_SERIES_QUERY_LEN: usize = 2048;
const SERIES_CACHE_TTL: Duration = Duration::from_secs(1);
#[derive(Debug)]
pub(crate) struct DashboardSetup {
pub request_rx: flume::Receiver<DashboardRequest>,
#[cfg_attr(not(test), allow(dead_code))]
pub bound_port: u16,
#[cfg_attr(not(test), allow(dead_code))]
pub build_count: Arc<AtomicU64>,
pub active_sse: Arc<AtomicUsize>,
pub running_folder: RunningFolder,
pub log_folder: RunningFolder,
pub url: String,
pub close_tx: watch::Sender<bool>,
pub server: DashboardServer,
}
#[derive(Debug)]
pub(crate) enum DashboardRequest {
GetSnapshot {
heatmap_since: Option<HeatmapCursor>,
respond: tokio::sync::oneshot::Sender<DashboardSnapshot>,
},
GetRequestSeries {
method: String,
name: String,
since: Option<HeatmapCursor>,
respond: tokio::sync::oneshot::Sender<Option<RequestSeries>>,
},
}
#[derive(Debug, Clone, Serialize)]
struct ControlResult {
ok: bool,
command: String,
#[serde(skip_serializing_if = "Option::is_none")]
error: Option<String>,
message: String,
phase: String,
active_users: u64,
#[serde(skip_serializing_if = "Option::is_none")]
target_users: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
field: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
settings: Option<Box<SnapshotSettings>>,
}
impl ControlResult {
fn from_outcome(command: &str, outcome: ControlOutcome) -> Self {
Self {
ok: outcome.ok,
command: command.to_string(),
error: outcome.error.map(str::to_string),
message: outcome.message,
phase: outcome.phase,
active_users: outcome.active_users as u64,
target_users: outcome.target_users.map(|u| u as u64),
field: outcome.field.map(str::to_string),
settings: outcome.settings,
}
}
}
#[derive(Debug, Serialize)]
struct ControlHttpErrorBody {
ok: bool,
command: String,
error: String,
message: String,
phase: Option<String>,
active_users: Option<u64>,
target_users: Option<u64>,
field: Option<String>,
}
impl ControlHttpErrorBody {
fn new(command: &str, error: &str, message: &str) -> Self {
Self {
ok: false,
command: command.to_string(),
error: error.to_string(),
message: message.to_string(),
phase: None,
active_users: None,
target_users: None,
field: None,
}
}
fn with_field(mut self, field: Option<&str>) -> Self {
self.field = field.map(str::to_string);
self
}
}
struct HeatmapStore {
generation: u64,
rev: u64,
start_second: u64,
end_second: u64,
hold_from: u64,
buckets_us: Vec<[u64; 3]>,
columns: BTreeMap<u64, (u64, HeatmapColumn)>,
}
impl HeatmapStore {
fn new(heatmap: &Heatmap) -> Self {
let mut store = HeatmapStore {
generation: heatmap.generation,
rev: 0,
start_second: 0,
end_second: 0,
hold_from: 0,
buckets_us: Vec::new(),
columns: BTreeMap::new(),
};
store.apply(heatmap);
store
}
fn apply(&mut self, heatmap: &Heatmap) {
if heatmap.since == 0 || heatmap.generation != self.generation {
self.columns.clear();
}
self.generation = heatmap.generation;
self.rev = heatmap.rev;
self.start_second = heatmap.start_second;
self.end_second = heatmap.end_second;
self.hold_from = heatmap.hold_from;
if heatmap.buckets_us.len() >= self.buckets_us.len() {
self.buckets_us = heatmap.buckets_us.clone();
}
for column in &heatmap.columns {
self.columns
.insert(column.second, (heatmap.rev, column.clone()));
}
self.columns = self.columns.split_off(&heatmap.start_second);
}
fn cursor(&self) -> HeatmapCursor {
HeatmapCursor {
generation: self.generation,
rev: self.rev,
}
}
fn since(&self, cursor: Option<HeatmapCursor>) -> Heatmap {
let since = match cursor {
Some(cursor) if cursor.generation == self.generation && cursor.rev <= self.rev => {
cursor.rev
}
_ => 0,
};
Heatmap {
generation: self.generation,
rev: self.rev,
since,
start_second: self.start_second,
end_second: self.end_second,
hold_from: self.hold_from,
buckets_us: self.buckets_us.clone(),
columns: self
.columns
.values()
.filter(|(rev, _)| *rev > since)
.map(|(_, column)| column.clone())
.collect(),
}
}
}
#[derive(Default)]
struct HubView {
snapshot: Option<DashboardSnapshot>,
heatmap: Option<HeatmapStore>,
}
#[derive(Clone, Debug)]
struct SnapshotHub {
inner: Arc<HubInner>,
}
struct HubInner {
request_tx: flume::Sender<DashboardRequest>,
latest_tx: watch::Sender<Option<Bytes>>,
active_sse: Arc<AtomicUsize>,
max_sse_clients: usize,
last_poll_ms: AtomicU64,
last_build_at_ms: AtomicU64,
last_build_ms: AtomicU64,
last_snapshot_ms: AtomicU64,
snapshot_timeouts: AtomicU64,
build_count: Arc<AtomicU64>,
snapshot_in_flight: AtomicBool,
wake: Notify,
shutdown_tx: watch::Sender<bool>,
view: Mutex<HubView>,
}
impl std::fmt::Debug for HubInner {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("HubInner")
.field("active_sse", &self.active_sse.load(Ordering::Relaxed))
.field("max_sse_clients", &self.max_sse_clients)
.field("build_count", &self.build_count.load(Ordering::Relaxed))
.field("last_poll_ms", &self.last_poll_ms.load(Ordering::Relaxed))
.field(
"last_build_at_ms",
&self.last_build_at_ms.load(Ordering::Relaxed),
)
.field("last_build_ms", &self.last_build_ms.load(Ordering::Relaxed))
.field(
"last_snapshot_ms",
&self.last_snapshot_ms.load(Ordering::Relaxed),
)
.field(
"snapshot_timeouts",
&self.snapshot_timeouts.load(Ordering::Relaxed),
)
.field(
"snapshot_in_flight",
&self.snapshot_in_flight.load(Ordering::Relaxed),
)
.finish()
}
}
impl SnapshotHub {
fn new(
request_tx: flume::Sender<DashboardRequest>,
build_count: Arc<AtomicU64>,
shutdown_tx: watch::Sender<bool>,
max_sse_clients: usize,
active_sse: Arc<AtomicUsize>,
) -> Self {
let max_sse_clients = if max_sse_clients == 0 {
MAX_SSE_CLIENTS
} else {
max_sse_clients
};
let (latest_tx, _) = watch::channel(None);
let inner = Arc::new(HubInner {
request_tx,
latest_tx,
active_sse,
max_sse_clients,
last_poll_ms: AtomicU64::new(0),
last_build_at_ms: AtomicU64::new(0),
last_build_ms: AtomicU64::new(0),
last_snapshot_ms: AtomicU64::new(0),
snapshot_timeouts: AtomicU64::new(0),
build_count,
snapshot_in_flight: AtomicBool::new(false),
wake: Notify::new(),
shutdown_tx,
view: Mutex::new(HubView::default()),
});
let loop_inner = Arc::clone(&inner);
tokio::spawn(async move {
hub_loop(loop_inner).await;
});
SnapshotHub { inner }
}
fn mark_poll(&self) {
self.inner
.last_poll_ms
.store(unix_now_ms(), Ordering::Relaxed);
self.inner.wake.notify_one();
}
fn try_acquire_sse(&self) -> bool {
if *self.inner.shutdown_tx.borrow() {
return false;
}
let cap = self.inner.max_sse_clients;
let mut current = self.inner.active_sse.load(Ordering::Relaxed);
loop {
if current >= cap {
return false;
}
match self.inner.active_sse.compare_exchange_weak(
current,
current + 1,
Ordering::AcqRel,
Ordering::Relaxed,
) {
Ok(_) => {
if *self.inner.shutdown_tx.borrow() {
self.inner.active_sse.fetch_sub(1, Ordering::AcqRel);
return false;
}
self.inner.wake.notify_one();
return true;
}
Err(observed) => current = observed,
}
}
}
fn release_sse(&self) {
self.inner.active_sse.fetch_sub(1, Ordering::AcqRel);
}
fn is_closed(&self) -> bool {
*self.inner.shutdown_tx.borrow()
}
fn active_sse_clients(&self) -> usize {
self.inner.active_sse.load(Ordering::Relaxed)
}
fn build_count(&self) -> u64 {
self.inner.build_count.load(Ordering::Relaxed)
}
fn last_build_ms(&self) -> u64 {
self.inner.last_build_ms.load(Ordering::Relaxed)
}
fn last_snapshot_ms(&self) -> u64 {
self.inner.last_snapshot_ms.load(Ordering::Relaxed)
}
fn snapshot_timeouts(&self) -> u64 {
self.inner.snapshot_timeouts.load(Ordering::Relaxed)
}
fn subscribe_latest(&self) -> watch::Receiver<Option<Bytes>> {
self.inner.latest_tx.subscribe()
}
fn subscribe_shutdown(&self) -> watch::Receiver<bool> {
self.inner.shutdown_tx.subscribe()
}
async fn snapshot_bytes(&self) -> Result<Bytes, StatusCode> {
if self.is_closed() {
return Err(StatusCode::SERVICE_UNAVAILABLE);
}
self.mark_poll();
if let Some(bytes) = self.fresh_latest() {
return Ok(bytes);
}
let gen_before = self.inner.build_count.load(Ordering::Acquire);
self.inner.wake.notify_one();
let mut rx = self.inner.latest_tx.subscribe();
let mut shutdown_rx = self.inner.shutdown_tx.subscribe();
let wait = async {
loop {
if self.is_closed() {
return Err(StatusCode::SERVICE_UNAVAILABLE);
}
if self.inner.build_count.load(Ordering::Acquire) > gen_before {
if let Some(bytes) = rx.borrow().clone() {
return Ok(bytes);
}
}
if let Some(bytes) = self.fresh_latest() {
return Ok(bytes);
}
tokio::select! {
result = rx.changed() => {
if result.is_err() {
return Err(StatusCode::SERVICE_UNAVAILABLE);
}
}
result = shutdown_rx.changed() => {
if result.is_err() || *shutdown_rx.borrow() {
return Err(StatusCode::SERVICE_UNAVAILABLE);
}
}
}
}
};
match tokio::time::timeout(Duration::from_secs(5), wait).await {
Ok(result) => result,
Err(_) => Err(StatusCode::SERVICE_UNAVAILABLE),
}
}
async fn snapshot_bytes_since(
&self,
cursor: Option<HeatmapCursor>,
) -> Result<Bytes, StatusCode> {
let shared = self.snapshot_bytes().await?;
let snapshot = {
let view = self.inner.view.lock().unwrap_or_else(|e| e.into_inner());
let Some(snapshot) = view.snapshot.as_ref() else {
return Ok(shared);
};
let mut snapshot = snapshot.clone();
snapshot.series.heatmap = view.heatmap.as_ref().map(|store| store.since(cursor));
snapshot
};
match serde_json::to_vec(&snapshot) {
Ok(json) => Ok(Bytes::from(json)),
Err(e) => {
warn!("[dashboard]: failed to serialize snapshot: {e}");
Err(StatusCode::INTERNAL_SERVER_ERROR)
}
}
}
fn fresh_latest(&self) -> Option<Bytes> {
let bytes = self.inner.latest_tx.borrow().clone()?;
let built_at = self.inner.last_build_at_ms.load(Ordering::Relaxed);
if built_at == 0 {
return None;
}
let age = unix_now_ms().saturating_sub(built_at);
if age < SNAPSHOT_INTERVAL.as_millis() as u64 {
Some(bytes)
} else {
None
}
}
}
fn unix_now_ms() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0)
}
fn should_build(inner: &HubInner) -> bool {
if *inner.shutdown_tx.borrow() {
return false;
}
if inner.active_sse.load(Ordering::Relaxed) > 0 {
return true;
}
let last_poll = inner.last_poll_ms.load(Ordering::Relaxed);
if last_poll == 0 {
return false;
}
unix_now_ms().saturating_sub(last_poll) < RECENT_POLL_WINDOW.as_millis() as u64
}
enum BuildOutcome {
Built,
Transient,
Fatal,
}
async fn hub_loop(inner: Arc<HubInner>) {
let mut shutdown_rx = inner.shutdown_tx.subscribe();
if *shutdown_rx.borrow() {
return;
}
loop {
while !should_build(&inner) {
if inner.request_tx.is_disconnected() {
let _ = inner.shutdown_tx.send(true);
return;
}
tokio::select! {
biased;
result = shutdown_rx.changed() => {
if result.is_err() || *shutdown_rx.borrow() {
return;
}
}
_ = inner.wake.notified() => {}
_ = tokio::time::sleep(Duration::from_secs(1)) => {}
}
}
match build_once(&inner, SNAPSHOT_BUILD_TIMEOUT).await {
BuildOutcome::Built => {}
BuildOutcome::Transient => {
debug!("[dashboard]: snapshot build transient; will retry");
tokio::select! {
biased;
result = shutdown_rx.changed() => {
if result.is_err() || *shutdown_rx.borrow() {
return;
}
}
_ = tokio::time::sleep(SNAPSHOT_INTERVAL) => {}
}
continue;
}
BuildOutcome::Fatal => {
let _ = inner.shutdown_tx.send(true);
return;
}
}
tokio::select! {
biased;
result = shutdown_rx.changed() => {
if result.is_err() || *shutdown_rx.borrow() {
return;
}
}
_ = tokio::time::sleep(SNAPSHOT_INTERVAL) => {
if inner.request_tx.is_disconnected() {
let _ = inner.shutdown_tx.send(true);
return;
}
}
}
}
}
async fn build_once(inner: &Arc<HubInner>, timeout: Duration) -> BuildOutcome {
if inner.request_tx.is_disconnected() {
return BuildOutcome::Fatal;
}
if inner
.snapshot_in_flight
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
.is_err()
{
return BuildOutcome::Transient;
}
let started = Instant::now();
let (respond_tx, mut respond_rx) = oneshot::channel();
let heatmap_since = inner
.view
.lock()
.unwrap_or_else(|e| e.into_inner())
.heatmap
.as_ref()
.map(HeatmapStore::cursor);
if inner
.request_tx
.send(DashboardRequest::GetSnapshot {
heatmap_since,
respond: respond_tx,
})
.is_err()
{
inner.snapshot_in_flight.store(false, Ordering::Release);
return BuildOutcome::Fatal;
}
let snapshot = tokio::select! {
result = &mut respond_rx => {
match result {
Ok(s) => s,
Err(_) => {
inner.snapshot_in_flight.store(false, Ordering::Release);
return BuildOutcome::Transient;
}
}
}
_ = tokio::time::sleep(timeout) => {
inner.snapshot_timeouts.fetch_add(1, Ordering::Relaxed);
debug!(
"[dashboard]: snapshot build timed out after {timeout:?}; waiting in background"
);
let clear_flag = Arc::clone(inner);
tokio::spawn(async move {
if respond_rx.await.is_ok() {
debug!(
"[dashboard]: late snapshot reply arrived after {}ms; dropped",
started.elapsed().as_millis()
);
}
clear_flag.snapshot_in_flight.store(false, Ordering::Release);
});
return BuildOutcome::Transient;
}
};
let build_time = snapshot.build_time;
{
let mut view = inner.view.lock().unwrap_or_else(|e| e.into_inner());
match (snapshot.series.heatmap.as_ref(), view.heatmap.as_mut()) {
(Some(heatmap), Some(store)) => store.apply(heatmap),
(Some(heatmap), None) => view.heatmap = Some(HeatmapStore::new(heatmap)),
(None, _) => view.heatmap = None,
}
let mut kept = snapshot.clone();
kept.series.heatmap = None;
view.snapshot = Some(kept);
}
let outcome = match serde_json::to_vec(&snapshot) {
Ok(json) => {
let _ = inner.latest_tx.send_replace(Some(Bytes::from(json)));
let snapshot_ms = started.elapsed().as_millis() as u64;
let build_ms = build_time.as_millis() as u64;
inner
.last_build_at_ms
.store(unix_now_ms(), Ordering::Release);
inner.last_build_ms.store(build_ms, Ordering::Release);
inner.last_snapshot_ms.store(snapshot_ms, Ordering::Release);
let count = inner.build_count.fetch_add(1, Ordering::Release) + 1;
debug!(
"[dashboard]: snapshot build {} took {}ms, round trip {}ms (active_sse={})",
count,
build_ms,
snapshot_ms,
inner.active_sse.load(Ordering::Relaxed)
);
BuildOutcome::Built
}
Err(e) => {
warn!("[dashboard]: failed to serialize snapshot: {e}");
BuildOutcome::Built
}
};
inner.snapshot_in_flight.store(false, Ordering::Release);
outcome
}
#[derive(Clone)]
struct DashboardState {
hub: SnapshotHub,
request_tx: flume::Sender<DashboardRequest>,
control_tx: Option<flume::Sender<ControlRequest>>,
auth_token: String,
no_auth: bool,
loopback: bool,
control_in_flight: Arc<AtomicUsize>,
series_in_flight: Arc<AtomicUsize>,
series_cache: Arc<Mutex<SeriesCache>>,
reports_dir: std::path::PathBuf,
listing_cache: Arc<Mutex<runs::ListingCache>>,
running_folder: RunningFolder,
log_folder: RunningFolder,
controllers: HealthControllers,
}
#[derive(Debug, Deserialize)]
struct TokenQuery {
token: Option<String>,
}
#[derive(Debug, Deserialize)]
struct SnapshotQuery {
token: Option<String>,
heatmap_generation: Option<u64>,
heatmap_since: Option<u64>,
}
fn heatmap_cursor(generation: Option<u64>, since: Option<u64>) -> Option<HeatmapCursor> {
Some(HeatmapCursor {
generation: generation?,
rev: since?,
})
}
#[derive(Debug, Deserialize)]
struct SeriesQuery {
method: Option<String>,
name: Option<String>,
token: Option<String>,
heatmap_generation: Option<u64>,
heatmap_since: Option<u64>,
}
type SeriesKey = (String, String, Option<(u64, u64)>);
type SeriesCache = HashMap<SeriesKey, (Instant, Bytes)>;
#[derive(Debug, Serialize)]
struct HealthResponse {
ok: bool,
version: String,
last_build_ms: u64,
last_snapshot_ms: u64,
snapshot_timeouts: u64,
build_count: u64,
active_sse_clients: usize,
control_enabled: bool,
auth_required: bool,
loopback: bool,
controllers: HealthControllers,
}
#[derive(Debug, Clone, Serialize)]
struct HealthControllers {
telnet: Option<String>,
websocket: Option<String>,
}
fn format_bind_address(host: &str, port: u16) -> String {
if host.contains(':') && !host.starts_with('[') {
format!("[{host}]:{port}")
} else {
format!("{host}:{port}")
}
}
pub(crate) async fn setup_dashboard(
configuration: &GooseConfiguration,
control_tx: Option<flume::Sender<ControlRequest>>,
listening: ListeningControllers,
) -> Result<Option<DashboardSetup>, GooseError> {
if !configuration.dashboard {
return Ok(None);
}
let host = configuration.dashboard_host();
let port = configuration.dashboard_port();
let address = format_bind_address(host, port);
let listener = tokio::net::TcpListener::bind(&address).await.map_err(|e| {
error!("[dashboard]: failed to bind {address}: {e} (is the port already in use?)");
GooseError::Io(e)
})?;
let bound = listener.local_addr().map_err(|e| {
error!("[dashboard]: failed to read local address after bind on {address}: {e}");
GooseError::Io(e)
})?;
let (request_tx, request_rx) = flume::unbounded();
let build_count = Arc::new(AtomicU64::new(0));
let (close_tx, _) = watch::channel(false);
let active_sse = Arc::new(AtomicUsize::new(0));
let running_folder = RunningFolder::default();
let log_folder = RunningFolder::default();
let max_sse_clients = configuration.dashboard_max_clients();
let hub = SnapshotHub::new(
request_tx.clone(),
Arc::clone(&build_count),
close_tx.clone(),
max_sse_clients,
Arc::clone(&active_sse),
);
let control_enabled = control_tx.is_some();
let no_auth = configuration.dashboard_no_auth;
let loopback = crate::config::is_loopback_bind(host);
let state = DashboardState {
hub,
request_tx,
control_tx,
auth_token: if no_auth {
String::new()
} else {
configuration
.dashboard_auth_token()
.unwrap_or_default()
.to_string()
},
no_auth,
loopback,
control_in_flight: Arc::new(AtomicUsize::new(0)),
series_in_flight: Arc::new(AtomicUsize::new(0)),
series_cache: Arc::new(Mutex::new(HashMap::new())),
reports_dir: runs::parent_folder(configuration),
listing_cache: Arc::new(Mutex::new(runs::ListingCache::new())),
running_folder: running_folder.clone(),
log_folder: log_folder.clone(),
controllers: HealthControllers {
telnet: listening.telnet.then(|| {
format_bind_address(configuration.telnet_host(), configuration.telnet_port())
}),
websocket: listening.websocket.then(|| {
format_bind_address(
configuration.websocket_host(),
configuration.websocket_port(),
)
}),
},
};
let app = build_router(state, host);
let max_connections = connection_cap(max_sse_clients);
let token = !no_auth && configuration.dashboard_auth_token().is_some();
for line in startup_lines(
&bound.to_string(),
control_enabled,
no_auth,
token,
loopback,
) {
match line {
StartupLine::Info(text) => info!("{text}"),
StartupLine::Warn(text) => warn!("{text}"),
StartupLine::Block(text) => crate::util::warning_block("[dashboard]", &text),
}
}
let (stop_tx, stop_rx) = watch::channel(false);
let task = tokio::spawn(serve_dashboard(listener, app, max_connections, stop_rx));
let server = DashboardServer {
stop_tx,
close_tx: close_tx.clone(),
task,
};
Ok(Some(DashboardSetup {
request_rx,
bound_port: bound.port(),
build_count,
active_sse,
running_folder,
log_folder,
url: format!("http://{bound}"),
close_tx,
server,
}))
}
fn open_address(bound: &str) -> String {
match bound.parse::<std::net::SocketAddr>() {
Ok(addr) if addr.ip().is_unspecified() => {
let ip: std::net::IpAddr = if addr.is_ipv4() {
std::net::Ipv4Addr::LOCALHOST.into()
} else {
std::net::Ipv6Addr::LOCALHOST.into()
};
std::net::SocketAddr::new(ip, addr.port()).to_string()
}
_ => bound.to_string(),
}
}
#[derive(Debug, PartialEq)]
enum StartupLine {
Info(String),
Warn(String),
Block(String),
}
fn startup_lines(
bound: &str,
control: bool,
no_auth: bool,
token: bool,
loopback: bool,
) -> Vec<StartupLine> {
let mut lines = Vec::new();
let listening = |mode: &str| {
StartupLine::Info(format!("[dashboard]: listening on http://{bound} ({mode})"))
};
match (control, no_auth) {
(false, false) if token => {
lines.push(listening("read-only"));
lines.push(StartupLine::Info(
"[dashboard]: read-only; for control add --dashboard-control".to_string(),
));
}
(false, false) => {
lines.push(listening("read-only"));
lines.push(StartupLine::Info(
"[dashboard]: read-only; for control add --dashboard-control and either --dashboard-no-auth (no token) or --dashboard-auth-token <token>".to_string(),
));
}
(false, true) => {
lines.push(listening("read-only, no token"));
lines.push(StartupLine::Info(
"[dashboard]: read-only; for control add --dashboard-control".to_string(),
));
}
(true, false) => {
lines.push(listening("control enabled"));
let open = open_address(bound);
lines.push(StartupLine::Info(format!(
"[dashboard]: open http://{open}/?token=<token> to control the test"
)));
}
(true, true) => {
lines.push(listening("control enabled, no token"));
if loopback {
lines.push(StartupLine::Warn(
"[dashboard]: --dashboard-no-auth: anyone on this machine can start, stop and redirect the test".to_string(),
));
}
}
}
if no_auth && !loopback {
let port = bound.rsplit(':').next().unwrap_or("5118");
let who = if control {
"Anyone who can reach this port can watch the test, start and stop it,\n\
and point the load at any host."
} else {
"Anyone who can reach this port can watch the test."
};
lines.push(StartupLine::Block(format!(
"WARNING: the dashboard on http://{bound} has no token (--dashboard-no-auth).\n\
{who}\n\
Safer: bind 127.0.0.1 and use an ssh forward (ssh -L {port}:127.0.0.1:{port} host),\n\
or use --dashboard-auth-token in place of --dashboard-no-auth. Only IP addresses,\n\
localhost and the --dashboard-host name are accepted as Host; set --dashboard-host\n\
to the DNS name you browse to."
)));
}
lines
}
#[derive(Debug)]
pub(crate) struct DashboardServer {
stop_tx: watch::Sender<bool>,
close_tx: watch::Sender<bool>,
task: tokio::task::JoinHandle<()>,
}
impl DashboardServer {
pub(crate) async fn shutdown(self) {
self.close_tx.send_replace(true);
self.stop_tx.send_replace(true);
if let Err(e) = self.task.await {
error!("[dashboard]: server task failed: {e}");
}
}
}
fn connection_cap(max_sse_clients: usize) -> usize {
max_sse_clients
.saturating_mul(2)
.saturating_add(EXTRA_CONNECTIONS)
.min(tokio::sync::Semaphore::MAX_PERMITS)
}
async fn serve_dashboard(
listener: tokio::net::TcpListener,
app: Router,
max_connections: usize,
mut stop_rx: watch::Receiver<bool>,
) {
let mut builder = hyper::server::conn::http1::Builder::new();
builder
.timer(hyper_util::rt::TokioTimer::new())
.header_read_timeout(HEADER_READ_TIMEOUT);
let open_connections = Arc::new(tokio::sync::Semaphore::new(max_connections));
let mut connections = tokio::task::JoinSet::new();
loop {
tokio::select! {
_ = stop_rx.changed() => break,
accepted = listener.accept() => match accepted {
Ok((stream, peer)) => match Arc::clone(&open_connections).try_acquire_owned() {
Ok(permit) => {
connections.spawn(serve_connection(
builder.clone(),
stream,
app.clone(),
stop_rx.clone(),
permit,
));
}
Err(_) => {
debug!(
"[dashboard]: closing connection from {peer}: {max_connections} connections already open"
);
drop(stream);
}
},
Err(e) => {
if !is_connection_error(&e) {
error!("[dashboard]: accept error: {e}");
tokio::select! {
_ = stop_rx.changed() => break,
_ = tokio::time::sleep(Duration::from_secs(1)) => {}
}
}
}
},
}
while connections.try_join_next().is_some() {}
}
drop(listener);
let drained = tokio::time::timeout(SERVER_DRAIN_TIMEOUT, async {
while connections.join_next().await.is_some() {}
})
.await;
if drained.is_err() {
debug!(
"[dashboard]: closing {} connection(s) still open after the drain deadline",
connections.len()
);
connections.shutdown().await;
}
}
async fn serve_connection(
builder: hyper::server::conn::http1::Builder,
stream: tokio::net::TcpStream,
app: Router,
mut stop_rx: watch::Receiver<bool>,
_permit: tokio::sync::OwnedSemaphorePermit,
) {
let service = hyper_util::service::TowerToHyperService::new(app);
let conn = builder.serve_connection(hyper_util::rt::TokioIo::new(stream), service);
let mut conn = std::pin::pin!(conn);
let result = tokio::select! {
result = conn.as_mut() => result,
_ = stop_rx.changed() => {
conn.as_mut().graceful_shutdown();
conn.await
}
};
if let Err(e) = result {
debug!("[dashboard]: connection error: {e}");
}
}
fn is_connection_error(e: &std::io::Error) -> bool {
matches!(
e.kind(),
std::io::ErrorKind::ConnectionRefused
| std::io::ErrorKind::ConnectionAborted
| std::io::ErrorKind::ConnectionReset
)
}
fn build_router(state: DashboardState, host: &str) -> Router {
let control_enabled = state.control_tx.is_some();
let loopback_bind = state.loopback;
let no_auth = state.no_auth;
let mut router = Router::new()
.route("/", get(index_handler))
.route("/static/app.js", get(app_js_handler))
.route("/static/app.css", get(app_css_handler))
.route("/api/v1/health", get(health_handler))
.route("/api/v1/snapshot", get(snapshot_handler))
.route("/api/v1/series", get(series_handler))
.route("/api/v1/events", get(events_handler))
.route("/api/v1/reports", get(reports_handler))
.route("/api/v1/reports/{name}/{file}", get(report_file_handler));
if control_enabled {
let control_routes = Router::new()
.route("/api/v1/control/start", post(control_start_handler))
.route("/api/v1/control/stop", post(control_stop_handler))
.route("/api/v1/control/quit", post(control_quit_handler))
.route("/api/v1/reports/{name}", delete(delete_report_handler))
.route("/api/v1/control/users", post(control_users_handler))
.route("/api/v1/control/settings", post(control_settings_handler))
.route("/api/v1/control/report", post(control_report_handler))
.route(
"/api/v1/control/report/now",
post(control_report_now_handler),
)
.layer(axum::extract::DefaultBodyLimit::max(CONTROL_BODY_LIMIT));
let control_routes = if no_auth {
control_routes.route_layer(axum::middleware::from_fn(require_same_origin_json))
} else {
control_routes
};
router = router.merge(control_routes);
}
if loopback_bind {
router = router.layer(axum::middleware::from_fn(require_loopback_host));
} else if no_auth {
router = router.layer(axum::middleware::from_fn_with_state(
Arc::<str>::from(host),
require_ip_or_configured_host,
));
}
router
.layer(SetResponseHeaderLayer::if_not_present(
header::CONTENT_SECURITY_POLICY,
HeaderValue::from_static(CSP),
))
.layer(SetResponseHeaderLayer::overriding(
header::CACHE_CONTROL,
HeaderValue::from_static("no-store"),
))
.layer(SetResponseHeaderLayer::overriding(
header::X_CONTENT_TYPE_OPTIONS,
HeaderValue::from_static("nosniff"),
))
.layer(SetResponseHeaderLayer::overriding(
header::REFERRER_POLICY,
HeaderValue::from_static("no-referrer"),
))
.with_state(Arc::new(state))
}
async fn require_loopback_host(
request: axum::extract::Request,
next: axum::middleware::Next,
) -> Response<Body> {
let allowed = request
.headers()
.get(header::HOST)
.and_then(|host| host.to_str().ok())
.is_some_and(is_loopback_host);
if allowed {
next.run(request).await
} else {
(StatusCode::FORBIDDEN, "host not allowed").into_response()
}
}
fn guard_refusal(reason: &'static str) -> Response<Body> {
(
StatusCode::FORBIDDEN,
Json(serde_json::json!({ "ok": false, "reason": reason })),
)
.into_response()
}
async fn require_same_origin_json(
request: axum::extract::Request,
next: axum::middleware::Next,
) -> Response<Body> {
let headers = request.headers();
let host = headers
.get(header::HOST)
.and_then(|host| host.to_str().ok());
let same_origin = match headers.get(header::ORIGIN) {
None => true,
Some(origin) => match (origin.to_str().ok(), host) {
(Some(origin), Some(host)) => origin
.strip_prefix("http://")
.is_some_and(|rest| rest.eq_ignore_ascii_case(host)),
_ => false,
},
};
let json = headers
.get(header::CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
.is_some_and(is_json_content_type);
if same_origin && json {
next.run(request).await
} else {
guard_refusal("cross_origin")
}
}
fn is_json_content_type(value: &str) -> bool {
value
.split(';')
.next()
.is_some_and(|media| media.trim().eq_ignore_ascii_case("application/json"))
}
async fn require_ip_or_configured_host(
State(configured): State<Arc<str>>,
request: axum::extract::Request,
next: axum::middleware::Next,
) -> Response<Body> {
let allowed = request
.headers()
.get(header::HOST)
.and_then(|host| host.to_str().ok())
.is_some_and(|host| is_ip_or_configured_host(host, &configured));
if allowed {
next.run(request).await
} else {
guard_refusal("host_not_allowed")
}
}
fn is_ip_or_configured_host(value: &str, configured: &str) -> bool {
if value.contains('@') {
return false;
}
let Ok(authority) = value.parse::<axum::http::uri::Authority>() else {
return false;
};
let host = authority.host();
let unbracketed = |h: &str| -> String {
h.strip_prefix('[')
.and_then(|h| h.strip_suffix(']'))
.unwrap_or(h)
.to_string()
};
let ip = if host.starts_with('[') {
unbracketed(host).parse::<std::net::Ipv6Addr>().is_ok()
} else {
host.parse::<std::net::Ipv4Addr>().is_ok()
};
ip || crate::config::is_loopback_bind(host)
|| unbracketed(host).eq_ignore_ascii_case(&unbracketed(configured.trim()))
}
fn is_loopback_host(value: &str) -> bool {
if value.contains('@') {
return false;
}
value
.parse::<axum::http::uri::Authority>()
.is_ok_and(|authority| crate::config::is_loopback_bind(authority.host()))
}
async fn index_handler() -> Response<Body> {
static_response("text/html; charset=utf-8", INDEX_HTML)
}
async fn app_js_handler() -> Response<Body> {
static_response("application/javascript; charset=utf-8", APP_JS)
}
async fn app_css_handler() -> Response<Body> {
static_response("text/css; charset=utf-8", APP_CSS)
}
fn static_response(content_type: &'static str, body: &'static str) -> Response<Body> {
([(header::CONTENT_TYPE, content_type)], Body::from(body)).into_response()
}
async fn health_handler(State(state): State<Arc<DashboardState>>) -> impl IntoResponse {
Json(HealthResponse {
ok: true,
version: env!("CARGO_PKG_VERSION").to_string(),
last_build_ms: state.hub.last_build_ms(),
last_snapshot_ms: state.hub.last_snapshot_ms(),
snapshot_timeouts: state.hub.snapshot_timeouts(),
build_count: state.hub.build_count(),
active_sse_clients: state.hub.active_sse_clients(),
control_enabled: state.control_tx.is_some(),
auth_required: !state.no_auth,
loopback: state.loopback,
controllers: state.controllers.clone(),
})
}
async fn snapshot_handler(
State(state): State<Arc<DashboardState>>,
headers: HeaderMap,
Query(query): Query<SnapshotQuery>,
) -> Result<impl IntoResponse, StatusCode> {
if !authorize(&state.auth_token, &headers, query.token.as_deref()) {
return Err(StatusCode::UNAUTHORIZED);
}
let cursor = heatmap_cursor(query.heatmap_generation, query.heatmap_since);
let bytes = state.hub.snapshot_bytes_since(cursor).await?;
Ok(([(header::CONTENT_TYPE, "application/json")], bytes))
}
fn series_error(status: StatusCode, error: &'static str) -> Response<Body> {
(status, Json(serde_json::json!({ "error": error }))).into_response()
}
struct SeriesInFlightGuard {
counter: Arc<AtomicUsize>,
}
impl Drop for SeriesInFlightGuard {
fn drop(&mut self) {
self.counter.fetch_sub(1, Ordering::AcqRel);
}
}
fn try_acquire_series_slot(state: &DashboardState) -> Option<SeriesInFlightGuard> {
let prev = state.series_in_flight.fetch_add(1, Ordering::AcqRel);
if prev >= MAX_SERIES_IN_FLIGHT {
state.series_in_flight.fetch_sub(1, Ordering::AcqRel);
return None;
}
Some(SeriesInFlightGuard {
counter: Arc::clone(&state.series_in_flight),
})
}
fn cached_series(state: &DashboardState, key: &SeriesKey) -> Option<Bytes> {
let cache = state.series_cache.lock().unwrap_or_else(|e| e.into_inner());
cache
.get(key)
.filter(|(built, _)| built.elapsed() < SERIES_CACHE_TTL)
.map(|(_, body)| body.clone())
}
fn cache_series(state: &DashboardState, key: SeriesKey, body: Bytes) {
let mut cache = state.series_cache.lock().unwrap_or_else(|e| e.into_inner());
cache.retain(|_, (built, _)| built.elapsed() < SERIES_CACHE_TTL);
cache.insert(key, (Instant::now(), body));
}
async fn series_handler(
State(state): State<Arc<DashboardState>>,
headers: HeaderMap,
Query(query): Query<SeriesQuery>,
) -> Response<Body> {
if !authorize(&state.auth_token, &headers, query.token.as_deref()) {
return StatusCode::UNAUTHORIZED.into_response();
}
let (Some(method), Some(name)) = (query.method, query.name) else {
return series_error(StatusCode::BAD_REQUEST, "bad_request");
};
if method.len() + name.len() > MAX_SERIES_QUERY_LEN {
return series_error(StatusCode::BAD_REQUEST, "bad_request");
}
let since = heatmap_cursor(query.heatmap_generation, query.heatmap_since);
let key = (
method,
name,
since.map(|cursor| (cursor.generation, cursor.rev)),
);
if let Some(body) = cached_series(&state, &key) {
return ([(header::CONTENT_TYPE, "application/json")], body).into_response();
}
let Some(_slot) = try_acquire_series_slot(&state) else {
return series_error(StatusCode::SERVICE_UNAVAILABLE, "busy");
};
let (respond, respond_rx) = oneshot::channel();
let request = DashboardRequest::GetRequestSeries {
method: key.0.clone(),
name: key.1.clone(),
since,
respond,
};
if state.request_tx.send(request).is_err() {
return series_error(StatusCode::SERVICE_UNAVAILABLE, "unavailable");
}
match tokio::time::timeout(CONTROL_TIMEOUT, respond_rx).await {
Ok(Ok(Some(series))) => match serde_json::to_vec(&series) {
Ok(body) => {
let body = Bytes::from(body);
cache_series(&state, key, body.clone());
([(header::CONTENT_TYPE, "application/json")], body).into_response()
}
Err(e) => {
error!("[dashboard]: failed to serialize series: {e}");
StatusCode::INTERNAL_SERVER_ERROR.into_response()
}
},
Ok(Ok(None)) => series_error(StatusCode::NOT_FOUND, "not_found"),
Ok(Err(_)) => series_error(StatusCode::SERVICE_UNAVAILABLE, "unavailable"),
Err(_) => series_error(StatusCode::SERVICE_UNAVAILABLE, "timeout"),
}
}
async fn events_handler(
State(state): State<Arc<DashboardState>>,
headers: HeaderMap,
Query(query): Query<TokenQuery>,
) -> Response<Body> {
if !authorize(&state.auth_token, &headers, query.token.as_deref()) {
return StatusCode::UNAUTHORIZED.into_response();
}
if state.hub.is_closed() {
return closed_sse_response();
}
if !state.hub.try_acquire_sse() {
if state.hub.is_closed() {
return closed_sse_response();
}
return (StatusCode::SERVICE_UNAVAILABLE, "too many SSE clients").into_response();
}
let hub = state.hub.clone();
let mut snap_rx = hub.subscribe_latest();
let mut shutdown_rx = hub.subscribe_shutdown();
let (event_tx, event_rx) = tokio::sync::mpsc::channel::<Event>(8);
tokio::spawn(async move {
struct SseGuard(SnapshotHub);
impl Drop for SseGuard {
fn drop(&mut self) {
self.0.release_sse();
}
}
let _guard = SseGuard(hub);
if *shutdown_rx.borrow() {
let _ = event_tx.send(closed_event()).await;
return;
}
let initial = snap_rx.borrow_and_update().clone();
if let Some(bytes) = initial {
if event_tx.send(snapshot_event(&bytes)).await.is_err() {
return;
}
}
loop {
tokio::select! {
result = snap_rx.changed() => {
if result.is_err() {
break;
}
if *shutdown_rx.borrow() {
let _ = event_tx.send(closed_event()).await;
break;
}
let next = snap_rx.borrow_and_update().clone();
if let Some(bytes) = next {
if event_tx.send(snapshot_event(&bytes)).await.is_err() {
break;
}
}
}
result = shutdown_rx.changed() => {
let closed = result.is_err() || *shutdown_rx.borrow();
if closed {
let _ = event_tx.send(closed_event()).await;
break;
}
}
_ = event_tx.closed() => break,
}
}
});
let stream = stream::unfold(event_rx, |mut rx| async move {
rx.recv()
.await
.map(|event| (Ok::<_, Infallible>(event), rx))
});
Sse::new(stream)
.keep_alive(KeepAlive::new().interval(SSE_HEARTBEAT).text("ping"))
.into_response()
}
fn closed_sse_response() -> Response<Body> {
let event = closed_event();
let stream = stream::once(async move { Ok::<_, Infallible>(event) });
Sse::new(stream).into_response()
}
fn closed_event() -> Event {
Event::default().event("closed").data("1")
}
fn snapshot_event(bytes: &Bytes) -> Event {
let data = std::str::from_utf8(bytes).unwrap_or("{}");
Event::default().event("snapshot").data(data)
}
#[derive(Clone, Debug, Default)]
pub(crate) struct RunningFolder(Arc<Mutex<Option<String>>>);
impl RunningFolder {
pub(crate) fn set(&self, folder: Option<&std::path::Path>) {
let name = folder
.and_then(std::path::Path::file_name)
.map(|name| name.to_string_lossy().into_owned());
match self.0.lock() {
Ok(mut running) => *running = name,
Err(poisoned) => *poisoned.into_inner() = name,
}
}
fn get(&self) -> Option<String> {
match self.0.lock() {
Ok(running) => running.clone(),
Err(poisoned) => poisoned.into_inner().clone(),
}
}
fn is(&self, parent: &Path, name: &str, folder: &Path) -> bool {
self.get().is_some_and(|own| {
own == name
|| std::fs::symlink_metadata(parent.join(&own))
.is_ok_and(|metadata| same_file(folder, &metadata))
})
}
}
#[derive(Debug, Deserialize)]
struct ReportQuery {
token: Option<String>,
baseline: Option<String>,
}
const COMPARE_MD: &str = "compare.md";
#[derive(Debug, Serialize)]
struct ReportsListing {
dir: String,
reports: Vec<runs::ListedReport>,
}
async fn reports_handler(
State(state): State<Arc<DashboardState>>,
headers: HeaderMap,
Query(query): Query<TokenQuery>,
) -> Response<Body> {
if !authorize(&state.auth_token, &headers, query.token.as_deref()) {
return StatusCode::UNAUTHORIZED.into_response();
}
let listing_state = Arc::clone(&state);
let listed = tokio::task::spawn_blocking(move || {
let running = listing_state
.running_folder
.get()
.map(|name| listing_state.reports_dir.join(name));
let mut cache = match listing_state.listing_cache.lock() {
Ok(cache) => cache,
Err(poisoned) => poisoned.into_inner(),
};
runs::list_reports(&listing_state.reports_dir, running.as_deref(), &mut cache)
})
.await;
let dir = state.reports_dir.to_string_lossy().into_owned();
match listed {
Ok(Ok(reports)) => Json(ReportsListing { dir, reports }).into_response(),
Ok(Err(e)) => {
warn!("[dashboard]: can't list the reports in {dir}: {e}");
(StatusCode::INTERNAL_SERVER_ERROR, "can't list the reports").into_response()
}
Err(_) => (StatusCode::INTERNAL_SERVER_ERROR, "can't list the reports").into_response(),
}
}
async fn report_file_handler(
State(state): State<Arc<DashboardState>>,
headers: HeaderMap,
path: Result<axum::extract::Path<(String, String)>, axum::extract::rejection::PathRejection>,
Query(query): Query<ReportQuery>,
) -> Response<Body> {
if !authorize(&state.auth_token, &headers, query.token.as_deref()) {
return StatusCode::UNAUTHORIZED.into_response();
}
let Ok(axum::extract::Path((name, file))) = path else {
return StatusCode::NOT_FOUND.into_response();
};
if file == COMPARE_MD {
return compare_response(&state.reports_dir, name, query.baseline).await;
}
let (content_type, ext) = match file.as_str() {
runs::REPORT_HTML => ("text/html; charset=utf-8", "html"),
runs::REPORT_JSON => ("application/json", "json"),
runs::REPORT_MD => ("text/markdown; charset=utf-8", "md"),
_ => return StatusCode::NOT_FOUND.into_response(),
};
let Some((folder, json)) = runs::report_folder(&state.reports_dir, &name) else {
return StatusCode::NOT_FOUND.into_response();
};
let test = {
let json = json.clone();
tokio::task::spawn_blocking(move || download_test_name(&json))
.await
.unwrap_or_else(|_| "run".to_string())
};
let filename = format!("goose-{test}-{name}.{ext}");
if let Some(stored) = runs::regular_file(&folder, &file) {
let gzipped = runs::without_gz(&stored.to_string_lossy()).1;
if !gzipped || accepts_gzip(&headers) {
return stored_file_response(&stored, content_type, &filename, gzipped).await;
}
let gunzipped = tokio::task::spawn_blocking(move || {
let file = std::fs::File::open(&stored)?;
let metadata = file.metadata()?;
if !metadata.is_file() || !same_file(&stored, &metadata) {
return Err(std::io::Error::other("the report changed"));
}
let mut body = Vec::new();
std::io::Read::read_to_end(&mut flate2::read::MultiGzDecoder::new(file), &mut body)?;
Ok(body)
})
.await;
return match gunzipped {
Ok(Ok(body)) => {
let length = body.len() as u64;
let mut response =
download_response(content_type, &filename, Body::from(body), Some(length));
response
.headers_mut()
.insert(header::VARY, HeaderValue::from_static("Accept-Encoding"));
response
}
_ => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
};
}
let rendered = tokio::task::spawn_blocking(move || render_report_file(&json, ext)).await;
match rendered {
Ok(Ok(body)) => {
let length = body.len() as u64;
download_response(content_type, &filename, Body::from(body), Some(length))
}
Ok(Err(reason)) => (
StatusCode::UNPROCESSABLE_ENTITY,
[(header::CONTENT_TYPE, "text/plain; charset=utf-8")],
reason,
)
.into_response(),
Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
}
}
fn render_report_file(json: &Path, ext: &str) -> Result<Vec<u8>, String> {
let loaded = crate::metrics::LoadedReport::read(json)?;
let configuration = GooseConfiguration::default();
let empty = crate::graph::GraphData::new();
let source = loaded.source(&configuration, &empty);
let mut out = Vec::new();
let written = match ext {
"html" => source.html(None).map(|html| out = html.into_bytes()),
_ => source.write_markdown(&mut out, None),
};
written.map_err(|e| format!("Can't render the report: {e}"))?;
Ok(out)
}
fn download_test_name(json: &Path) -> String {
runs::read_record(json)
.map(|record| {
record
.test
.chars()
.filter(|c| c.is_ascii_alphanumeric() || *c == '_' || *c == '-')
.collect::<String>()
})
.filter(|name| !name.is_empty())
.unwrap_or_else(|| "run".to_string())
}
fn download_response(
content_type: &'static str,
filename: &str,
body: Body,
length: Option<u64>,
) -> Response<Body> {
let mut response = Response::new(body);
let headers = response.headers_mut();
headers.insert(header::CONTENT_TYPE, HeaderValue::from_static(content_type));
if let Some(length) = length {
headers.insert(header::CONTENT_LENGTH, HeaderValue::from(length));
}
if let Ok(value) = HeaderValue::from_str(&format!("attachment; filename=\"{filename}\"")) {
headers.insert(header::CONTENT_DISPOSITION, value);
}
headers.insert(
header::X_CONTENT_TYPE_OPTIONS,
HeaderValue::from_static("nosniff"),
);
headers.insert(header::CACHE_CONTROL, HeaderValue::from_static("no-store"));
headers.insert(
header::CONTENT_SECURITY_POLICY,
HeaderValue::from_static("sandbox"),
);
response
}
fn accepts_gzip(headers: &HeaderMap) -> bool {
let codings: Vec<(String, bool)> = headers
.get_all(header::ACCEPT_ENCODING)
.iter()
.filter_map(|value| value.to_str().ok())
.flat_map(|value| value.split(','))
.map(|coding| {
let mut parts = coding.split(';').map(str::trim);
let name = parts.next().unwrap_or_default().to_ascii_lowercase();
let refused = parts.any(|param| {
let param = param.to_ascii_lowercase();
param
.strip_prefix("q=")
.and_then(|q| q.trim().parse::<f32>().ok())
.is_some_and(|q| q == 0.0)
});
(name, !refused)
})
.collect();
let taken = |name: &str| codings.iter().find(|(n, _)| n == name).map(|(_, ok)| *ok);
taken("gzip").or_else(|| taken("*")).unwrap_or(false)
}
async fn stored_file_response(
path: &Path,
content_type: &'static str,
filename: &str,
gzipped: bool,
) -> Response<Body> {
let Ok(file) = tokio::fs::File::open(path).await else {
return StatusCode::NOT_FOUND.into_response();
};
let Ok(metadata) = file.metadata().await else {
return StatusCode::NOT_FOUND.into_response();
};
if !metadata.is_file() || !same_file(path, &metadata) {
return StatusCode::NOT_FOUND.into_response();
}
let mut response = download_response(
content_type,
filename,
Body::from_stream(file_stream(file)),
Some(metadata.len()),
);
if gzipped {
response
.headers_mut()
.insert(header::CONTENT_ENCODING, HeaderValue::from_static("gzip"));
response
.headers_mut()
.insert(header::VARY, HeaderValue::from_static("Accept-Encoding"));
}
response
}
fn same_file(path: &Path, opened: &std::fs::Metadata) -> bool {
let Ok(at_path) = std::fs::symlink_metadata(path) else {
return false;
};
#[cfg(unix)]
{
use std::os::unix::fs::MetadataExt;
at_path.dev() == opened.dev() && at_path.ino() == opened.ino()
}
#[cfg(not(unix))]
{
at_path.file_type().is_file() && at_path.len() == opened.len()
}
}
fn file_stream(
file: tokio::fs::File,
) -> impl futures::Stream<Item = Result<Bytes, std::io::Error>> + Send {
const CHUNK: usize = 64 * 1024;
stream::unfold(Some(file), |file| async move {
let mut file = file?;
let mut buf = vec![0u8; CHUNK];
match file.read(&mut buf).await {
Ok(0) => None,
Ok(n) => {
buf.truncate(n);
Some((Ok(Bytes::from(buf)), Some(file)))
}
Err(e) => Some((Err(e), None)),
}
})
}
async fn compare_response(
reports_dir: &Path,
name: String,
baseline: Option<String>,
) -> Response<Body> {
let Some(baseline) = baseline else {
return StatusCode::NOT_FOUND.into_response();
};
let (Some((_, newer_json)), Some((_, baseline_json))) = (
runs::report_folder(reports_dir, &name),
runs::report_folder(reports_dir, &baseline),
) else {
return StatusCode::NOT_FOUND.into_response();
};
let (run_name, baseline_name) = (name.clone(), baseline.clone());
let compared = tokio::task::spawn_blocking(move || {
let report = crate::metrics::compare_reports(
&newer_json,
&run_name,
&baseline_json,
&baseline_name,
)?;
let mut out =
compared_runs_line(&newer_json, &run_name, &baseline_json, &baseline_name).into_bytes();
out.extend_from_slice(b"\n\n");
out.extend(report);
Ok::<_, String>((download_test_name(&newer_json), out))
})
.await;
match compared {
Ok(Ok((test, markdown))) => {
let length = markdown.len() as u64;
download_response(
"text/markdown; charset=utf-8",
&format!("goose-{test}-{name}-vs-{baseline}.md"),
Body::from(markdown),
Some(length),
)
}
Ok(Err(reason)) => (
StatusCode::UNPROCESSABLE_ENTITY,
[(header::CONTENT_TYPE, "text/plain; charset=utf-8")],
reason,
)
.into_response(),
Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
}
}
fn compared_runs_line(
newer_json: &Path,
run_name: &str,
baseline_json: &Path,
baseline_name: &str,
) -> String {
let started = |json: &Path| {
runs::read_record(json)
.and_then(|record| record.started)
.and_then(|started| chrono::DateTime::parse_from_rfc3339(&started).ok())
.map(|time| {
time.with_timezone(&chrono::Utc)
.format("%Y-%m-%d %H:%M:%S UTC")
.to_string()
})
.unwrap_or_else(|| "at an unknown time".to_string())
};
format!(
"Run {run_name} (started {}) compared to baseline {baseline_name} (started {}).",
started(newer_json),
started(baseline_json)
)
}
fn authorize(configured_token: &str, headers: &HeaderMap, query_token: Option<&str>) -> bool {
if configured_token.is_empty() {
return true;
}
if let Some(token) = query_token {
if constant_time_eq(token.as_bytes(), configured_token.as_bytes()) {
return true;
}
}
if let Some(token) = bearer_token_from_headers(headers) {
if constant_time_eq(token.as_bytes(), configured_token.as_bytes()) {
return true;
}
}
false
}
fn authorize_control(
configured_token: &str,
headers: &HeaderMap,
query_token: Option<&str>,
) -> bool {
if query_token.is_some() {
return false;
}
if configured_token.is_empty() {
return false;
}
if let Some(token) = bearer_token_from_headers(headers) {
return constant_time_eq(token.as_bytes(), configured_token.as_bytes());
}
false
}
impl DashboardState {
fn authorize_control(&self, headers: &HeaderMap, query_token: Option<&str>) -> bool {
self.no_auth || authorize_control(&self.auth_token, headers, query_token)
}
}
fn bearer_token_from_headers(headers: &HeaderMap) -> Option<&str> {
let value = headers.get(header::AUTHORIZATION)?;
let s = value.to_str().ok()?;
let mut parts = s.split_whitespace();
let scheme = parts.next()?;
let token = parts.next()?;
if parts.next().is_some() {
return None;
}
if scheme.eq_ignore_ascii_case("Bearer") && !token.is_empty() {
Some(token)
} else {
None
}
}
fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
let len_mismatch = (a.len() != b.len()) as u8;
let max_len = a.len().max(b.len());
let mut diff = len_mismatch;
let mut i = 0;
while i < max_len {
let x = if i < a.len() { a[i] } else { 0 };
let y = if i < b.len() { b[i] } else { 0 };
diff |= x ^ y;
i += 1;
}
diff == 0
}
fn parse_empty_control_body(body: &[u8]) -> Result<(), (&'static str, &'static str)> {
let trimmed = trim_ascii_whitespace(body);
if trimmed.is_empty() {
return Ok(());
}
let value: serde_json::Value = match serde_json::from_slice(trimmed) {
Ok(v) => v,
Err(_) => return Err(("bad_request", "request body must be empty or {}")),
};
match value.as_object() {
Some(obj) if obj.is_empty() => Ok(()),
_ => Err(("bad_request", "request body must be empty or {}")),
}
}
fn parse_users_body(body: &[u8]) -> Result<ControllerCommand, (&'static str, &'static str)> {
let trimmed = trim_ascii_whitespace(body);
if trimmed.is_empty() {
return Err(("bad_request", "request body required"));
}
let value: serde_json::Value =
serde_json::from_slice(trimmed).map_err(|_| ("bad_request", "malformed JSON body"))?;
let obj = value
.as_object()
.ok_or(("bad_request", "request body must be a JSON object"))?;
let users = obj
.get("users")
.and_then(serde_json::Value::as_u64)
.unwrap_or(0);
ControllerCommand::users(users).map_err(|message| ("invalid_users", message))
}
#[derive(Debug, Default, Deserialize)]
#[serde(deny_unknown_fields)]
struct SettingsBody {
#[serde(default, deserialize_with = "present")]
host: Option<serde_json::Value>,
#[serde(default, deserialize_with = "present")]
users: Option<serde_json::Value>,
#[serde(default, deserialize_with = "present")]
increase_rate: Option<serde_json::Value>,
#[serde(default, deserialize_with = "present")]
increase_time_secs: Option<serde_json::Value>,
#[serde(default, deserialize_with = "present")]
decrease_rate: Option<serde_json::Value>,
#[serde(default, deserialize_with = "present")]
decrease_time_secs: Option<serde_json::Value>,
#[serde(default, deserialize_with = "present")]
run_time_secs: Option<serde_json::Value>,
#[serde(default, deserialize_with = "present")]
test_plan: Option<serde_json::Value>,
}
fn present<'de, D>(deserializer: D) -> Result<Option<serde_json::Value>, D::Error>
where
D: serde::Deserializer<'de>,
{
serde_json::Value::deserialize(deserializer).map(Some)
}
const NULL_USERS_MSG: &str =
"users can not be cleared: send a user count, or a test_plan to switch to";
fn invalid_secs_msg(field: &str) -> &'static str {
match field {
"increase_time" => {
"increase_time_secs must be whole seconds from 1 to 2592000, or null to clear"
}
"decrease_time" => {
"decrease_time_secs must be whole seconds from 1 to 2592000, or null to clear"
}
_ => "run_time_secs must be whole seconds from 1 to 2592000, or null to clear",
}
}
type BodyRefusal = (&'static str, &'static str, Option<&'static str>);
pub(crate) fn parse_settings_body(body: &[u8]) -> Result<RunSettings, BodyRefusal> {
let bad_request = (
"bad_request",
"request body must be a JSON object of run settings: host, users, increase_rate, increase_time_secs, decrease_rate, decrease_time_secs, run_time_secs and test_plan",
None,
);
let body: serde_json::Value =
serde_json::from_slice(trim_ascii_whitespace(body)).map_err(|_| bad_request)?;
if !body.is_object() {
return Err(bad_request);
}
let body: SettingsBody = serde_json::from_value(body).map_err(|_| bad_request)?;
let invalid = |field, message| ("invalid_value", message, Some(field));
fn read<T>(
value: Option<serde_json::Value>,
parse: impl FnOnce(&serde_json::Value) -> Option<Result<T, &'static str>>,
error: &'static str,
) -> Result<Option<Option<T>>, &'static str> {
match value {
None => Ok(None),
Some(serde_json::Value::Null) => Ok(Some(None)),
Some(value) => parse(&value).unwrap_or(Err(error)).map(|v| Some(Some(v))),
}
}
let rate = |field, value, error| {
read(
value,
|value: &serde_json::Value| {
value
.as_f64()
.map(|rate| controller::bounded_rate(rate as f32, error))
},
error,
)
.map_err(|message| invalid(field, message))
};
let secs = |field, value| {
let error = invalid_secs_msg(field);
read(
value,
|value: &serde_json::Value| {
value
.as_u64()
.filter(|&secs| secs > 0)
.map(|secs| controller::bounded_secs(secs, error))
},
error,
)
.map_err(|message| invalid(field, message))
};
let settings = RunSettings {
host: read(
body.host,
|value| value.as_str().map(controller::host),
controller::INVALID_HOST_MSG,
)
.map_err(|message| invalid("host", message))?,
users: match read(
body.users,
|value| value.as_u64().map(controller::bounded_users),
controller::INVALID_USERS_MSG,
) {
Ok(Some(None)) => return Err(invalid("users", NULL_USERS_MSG)),
Ok(users) => users.flatten(),
Err(message) => return Err(invalid("users", message)),
},
increase_rate: rate(
"increase_rate",
body.increase_rate,
controller::INVALID_INCREASE_RATE_MSG,
)?,
increase_time: secs("increase_time", body.increase_time_secs)?,
decrease_rate: rate(
"decrease_rate",
body.decrease_rate,
controller::INVALID_DECREASE_RATE_MSG,
)?,
decrease_time: secs("decrease_time", body.decrease_time_secs)?,
run_time: secs("run_time", body.run_time_secs)?,
test_plan: read(
body.test_plan,
|value| value.as_str().map(controller::test_plan),
controller::INVALID_PLAN_MSG,
)
.map_err(|message| invalid("test_plan", message))?,
};
match settings.conflict() {
Some((field, message)) => Err(("conflicting_settings", message, Some(field))),
None => Ok(settings),
}
}
fn trim_ascii_whitespace(bytes: &[u8]) -> &[u8] {
let start = bytes
.iter()
.position(|b| !b.is_ascii_whitespace())
.unwrap_or(bytes.len());
let end = bytes
.iter()
.rposition(|b| !b.is_ascii_whitespace())
.map(|i| i + 1)
.unwrap_or(0);
if start >= end {
&[]
} else {
&bytes[start..end]
}
}
fn control_http_error(
status: StatusCode,
command: &str,
error: &str,
message: &str,
) -> Response<Body> {
let body = ControlHttpErrorBody::new(command, error, message);
(status, Json(body)).into_response()
}
struct ControlInFlightGuard {
counter: Arc<AtomicUsize>,
}
impl Drop for ControlInFlightGuard {
fn drop(&mut self) {
self.counter.fetch_sub(1, Ordering::AcqRel);
}
}
fn try_acquire_control_slot(state: &DashboardState) -> Option<ControlInFlightGuard> {
let prev = state.control_in_flight.fetch_add(1, Ordering::AcqRel);
if prev >= MAX_CONTROL_IN_FLIGHT {
state.control_in_flight.fetch_sub(1, Ordering::AcqRel);
return None;
}
Some(ControlInFlightGuard {
counter: Arc::clone(&state.control_in_flight),
})
}
async fn dispatch_control(
state: &DashboardState,
command: impl Into<ControlCommand>,
name: &str,
) -> Response<Body> {
let Some(_slot) = try_acquire_control_slot(state) else {
warn!("[dashboard]: control admission full ({name})");
return control_http_error(
StatusCode::SERVICE_UNAVAILABLE,
name,
"busy",
"too many control requests in flight — wait and retry",
);
};
let reply = match state.control_tx.as_ref() {
Some(control_tx) => send_command(control_tx, command, CONTROL_TIMEOUT).await,
None => Err(SendError::Unavailable),
};
match reply {
Ok(ControlReply::Outcome(outcome)) => {
let result = ControlResult::from_outcome(name, outcome);
if result.ok {
info!(
"[dashboard]: control {name} ok phase={} target={:?}",
result.phase, result.target_users
);
} else {
info!(
"[dashboard]: control {name} rejected error={:?} phase={}",
result.error, result.phase
);
}
(StatusCode::OK, Json(result)).into_response()
}
Err(SendError::Timeout) => {
warn!("[dashboard]: control timed out ({name}); skipped if still queued");
control_http_error(
StatusCode::SERVICE_UNAVAILABLE,
name,
"timeout",
"control timed out — action may still be applying; check phase/users before retrying",
)
}
_ => {
warn!("[dashboard]: control unavailable ({name})");
control_http_error(
StatusCode::SERVICE_UNAVAILABLE,
name,
"unavailable",
"control unavailable",
)
}
}
}
async fn control_without_body(
state: &DashboardState,
headers: &HeaderMap,
query: TokenQuery,
body: &[u8],
command: ControllerCommand,
name: &str,
) -> Response<Body> {
if !state.authorize_control(headers, query.token.as_deref()) {
warn!("[dashboard]: control {name} auth failure");
return StatusCode::UNAUTHORIZED.into_response();
}
if let Err((code, msg)) = parse_empty_control_body(body) {
return control_http_error(StatusCode::BAD_REQUEST, name, code, msg);
}
dispatch_control(state, command, name).await
}
async fn control_start_handler(
State(state): State<Arc<DashboardState>>,
headers: HeaderMap,
Query(query): Query<TokenQuery>,
body: Bytes,
) -> Response<Body> {
if parse_empty_control_body(&body).is_ok() {
return control_without_body(
&state,
&headers,
query,
&body,
ControllerCommand::Start,
"start",
)
.await;
}
control_with_settings(&state, &headers, query, &body, "start").await
}
async fn control_settings_handler(
State(state): State<Arc<DashboardState>>,
headers: HeaderMap,
Query(query): Query<TokenQuery>,
body: Bytes,
) -> Response<Body> {
control_with_settings(&state, &headers, query, &body, "settings").await
}
async fn control_with_settings(
state: &DashboardState,
headers: &HeaderMap,
query: TokenQuery,
body: &[u8],
name: &str,
) -> Response<Body> {
if !state.authorize_control(headers, query.token.as_deref()) {
warn!("[dashboard]: control {name} auth failure");
return StatusCode::UNAUTHORIZED.into_response();
}
match parse_settings_body(body) {
Ok(settings) => {
let command = ControlCommand::Settings {
settings: Box::new(settings),
start: name == "start",
};
dispatch_control(state, command, name).await
}
Err((code, message, field)) => {
let body = ControlHttpErrorBody::new(name, code, message).with_field(field);
(StatusCode::BAD_REQUEST, Json(body)).into_response()
}
}
}
async fn control_stop_handler(
State(state): State<Arc<DashboardState>>,
headers: HeaderMap,
Query(query): Query<TokenQuery>,
body: Bytes,
) -> Response<Body> {
control_without_body(
&state,
&headers,
query,
&body,
ControllerCommand::Stop,
"stop",
)
.await
}
async fn control_quit_handler(
State(state): State<Arc<DashboardState>>,
headers: HeaderMap,
Query(query): Query<TokenQuery>,
body: Bytes,
) -> Response<Body> {
control_without_body(
&state,
&headers,
query,
&body,
ControllerCommand::Quit,
"quit",
)
.await
}
fn conflict(message: &'static str) -> Response<Body> {
(
StatusCode::CONFLICT,
[(header::CONTENT_TYPE, "text/plain; charset=utf-8")],
message,
)
.into_response()
}
const RESERVED_BY_ANOTHER: &str = "Another Goose process is writing to this folder.";
async fn delete_report_handler(
State(state): State<Arc<DashboardState>>,
headers: HeaderMap,
path: Result<axum::extract::Path<String>, axum::extract::rejection::PathRejection>,
Query(query): Query<TokenQuery>,
) -> Response<Body> {
if !state.authorize_control(&headers, query.token.as_deref()) {
warn!("[dashboard]: delete report auth failure");
return StatusCode::UNAUTHORIZED.into_response();
}
let Ok(axum::extract::Path(name)) = path else {
return StatusCode::NOT_FOUND.into_response();
};
let Some((folder, json)) = runs::report_folder(&state.reports_dir, &name) else {
return StatusCode::NOT_FOUND.into_response();
};
let refused = if state.running_folder.is(&state.reports_dir, &name, &folder) {
Some("The run in progress writes to this folder.")
} else if state.log_folder.is(&state.reports_dir, &name, &folder) {
Some("Goose is writing its log to this folder.")
} else if runs::is_reserved(&folder) {
Some(RESERVED_BY_ANOTHER)
} else {
None
};
if let Some(refused) = refused {
return conflict(refused);
}
let name_for_check = name.clone();
let reports_dir = state.reports_dir.clone();
let removed = tokio::task::spawn_blocking(move || {
if runs::report_folder(&reports_dir, &name_for_check).is_none() || !runs::is_listed(&json) {
return Ok(Some(StatusCode::NOT_FOUND));
}
if runs::is_reserved(&folder) {
return Ok(Some(StatusCode::CONFLICT));
}
for file in runs::deletable_files() {
let path = folder.join(file);
if std::fs::symlink_metadata(&path).is_ok_and(|m| !m.is_dir()) {
std::fs::remove_file(&path)?;
}
}
let _ = std::fs::remove_dir(&folder);
Ok::<_, std::io::Error>(None)
})
.await;
match removed {
Ok(Ok(Some(StatusCode::CONFLICT))) => conflict(RESERVED_BY_ANOTHER),
Ok(Ok(Some(status))) => status.into_response(),
Ok(Ok(None)) => {
info!("[dashboard]: deleted the report in {name}");
StatusCode::NO_CONTENT.into_response()
}
Ok(Err(e)) => {
warn!("[dashboard]: can't delete the report in {name}: {e}");
(StatusCode::INTERNAL_SERVER_ERROR, "can't delete the report").into_response()
}
Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
}
}
async fn control_users_handler(
State(state): State<Arc<DashboardState>>,
headers: HeaderMap,
Query(query): Query<TokenQuery>,
body: Bytes,
) -> Response<Body> {
if !state.authorize_control(&headers, query.token.as_deref()) {
warn!("[dashboard]: control users auth failure");
return StatusCode::UNAUTHORIZED.into_response();
}
match parse_users_body(&body) {
Ok(command) => dispatch_control(&state, command, "users").await,
Err((code, msg)) => control_http_error(StatusCode::BAD_REQUEST, "users", code, msg),
}
}
fn parse_report_body(body: &[u8]) -> Result<ControlCommand, (&'static str, &'static str)> {
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct ReportBody {
save: Option<bool>,
graph_data: Option<bool>,
}
let body: ReportBody = serde_json::from_slice(trim_ascii_whitespace(body)).map_err(|_| {
(
"bad_request",
"request body must be {\"save\": true or false, \"graph_data\": true or false}",
)
})?;
if body.save.is_none() && body.graph_data.is_none() {
return Err((
"bad_request",
"request body names neither save nor graph_data",
));
}
Ok(ControlCommand::Report {
save: body.save,
graph_data: body.graph_data,
})
}
async fn control_report_handler(
State(state): State<Arc<DashboardState>>,
headers: HeaderMap,
Query(query): Query<TokenQuery>,
body: Bytes,
) -> Response<Body> {
if !state.authorize_control(&headers, query.token.as_deref()) {
warn!("[dashboard]: control report auth failure");
return StatusCode::UNAUTHORIZED.into_response();
}
match parse_report_body(&body) {
Ok(command) => dispatch_control(&state, command, "report").await,
Err((code, msg)) => control_http_error(StatusCode::BAD_REQUEST, "report", code, msg),
}
}
async fn control_report_now_handler(
State(state): State<Arc<DashboardState>>,
headers: HeaderMap,
Query(query): Query<TokenQuery>,
body: Bytes,
) -> Response<Body> {
if !state.authorize_control(&headers, query.token.as_deref()) {
warn!("[dashboard]: control report now auth failure");
return StatusCode::UNAUTHORIZED.into_response();
}
if let Err((code, msg)) = parse_empty_control_body(&body) {
return control_http_error(StatusCode::BAD_REQUEST, "report_now", code, msg);
}
dispatch_control(&state, ControlCommand::SaveReportNow, "report_now").await
}
#[cfg(test)]
mod tests {
use super::*;
use crate::config::GooseConfiguration;
use crate::metrics::dashboard_snapshot::{
AggregateMetrics, Percentiles, SeriesWindow, SnapshotFlags,
};
use chrono::Utc;
use std::time::Instant;
fn dummy_snapshot(seq: u64) -> DashboardSnapshot {
DashboardSnapshot {
version: 1,
generated_at: Utc::now(),
goose_version: "test".into(),
phase: "maintain".into(),
stopping: false,
duration_secs: seq,
active_users: 1,
maximum_users: 1,
target_users: 1,
total_users: 1,
hosts: vec!["http://example.test".into()],
aggregate: AggregateMetrics {
total_requests: seq,
total_failures: 0,
requests_per_second: 1.0,
failures_per_second: 0.0,
failure_rate: 0.0,
response_time_avg_ms: 1.0,
response_time_min_ms: 1,
response_time_max_ms: 1,
percentile_ms: Percentiles {
p50: 1,
p95: 1,
p99: 1,
},
response_time_us: None,
co_active: false,
co_adjusted: None,
},
requests: vec![],
errors: vec![],
transaction_errors: vec![],
scenarios: vec![],
transactions: vec![],
series: SeriesWindow::empty(),
flags: SnapshotFlags {
metrics_disabled: false,
transaction_metrics_disabled: false,
scenario_metrics_disabled: false,
requests_truncated: false,
errors_truncated: false,
transaction_errors_truncated: false,
series_seconds: 300,
percentiles_over_time: false,
},
report: crate::metrics::dashboard_snapshot::SnapshotReport::default(),
settings: crate::metrics::dashboard_snapshot::SnapshotSettings::idle(),
build_time: Duration::ZERO,
}
}
fn spawn_mock_parent(
request_rx: flume::Receiver<DashboardRequest>,
parent_builds: Arc<AtomicU64>,
) -> tokio::task::JoinHandle<()> {
tokio::spawn(async move {
while let Ok(msg) = request_rx.recv_async().await {
match msg {
DashboardRequest::GetSnapshot { respond, .. } => {
let n = parent_builds.fetch_add(1, Ordering::SeqCst);
let _ = respond.send(dummy_snapshot(n));
}
DashboardRequest::GetRequestSeries { respond, .. } => {
let _ = respond.send(None);
}
}
}
})
}
fn spawn_control_parent(
control_rx: flume::Receiver<ControlRequest>,
answer: fn(ControlCommand) -> Option<ControlReply>,
) -> tokio::task::JoinHandle<()> {
tokio::spawn(async move {
while let Ok((command, respond)) = control_rx.recv_async().await {
if let Some(reply) = answer(command) {
let _ = respond.send(reply);
}
}
})
}
fn outcome(
ok: bool,
error: Option<&'static str>,
message: &str,
phase: &str,
active_users: usize,
target_users: Option<usize>,
) -> Option<ControlReply> {
Some(ControlReply::Outcome(ControlOutcome {
ok,
error,
message: message.to_string(),
phase: phase.to_string(),
active_users,
target_users,
field: None,
settings: None,
}))
}
fn answer_ok(command: ControlCommand) -> Option<ControlReply> {
let command = match command {
ControlCommand::Settings { start: false, .. } => {
return outcome(true, None, "settings configured", "idle", 0, None);
}
ControlCommand::Settings { start: true, .. } => ControllerCommand::Start,
ControlCommand::Report { .. } => {
return outcome(true, None, "save report on", "idle", 0, None);
}
ControlCommand::SaveReportNow => {
return outcome(true, None, "saving the report", "maintain", 1, None);
}
ControlCommand::Command(command) => command,
};
match command {
ControllerCommand::Start => {
outcome(true, None, "load test started", "increase", 0, Some(1))
}
ControllerCommand::Stop => {
outcome(true, None, "load test stopped", "decrease", 1, Some(0))
}
ControllerCommand::Quit => {
outcome(true, None, "goose is shutting down", "decrease", 0, None)
}
ControllerCommand::Users(users) => {
outcome(true, None, "users configured", "idle", 0, Some(users))
}
other => panic!("unexpected dashboard command {:?}", other),
}
}
fn spawn_blip_parent(
request_rx: flume::Receiver<DashboardRequest>,
parent_builds: Arc<AtomicU64>,
drop_from: u64,
drop_until: u64,
) -> tokio::task::JoinHandle<()> {
tokio::spawn(async move {
while let Ok(msg) = request_rx.recv_async().await {
match msg {
DashboardRequest::GetSnapshot { respond, .. } => {
let n = parent_builds.fetch_add(1, Ordering::SeqCst);
if n >= drop_from && n < drop_until {
drop(respond);
} else {
let _ = respond.send(dummy_snapshot(n));
}
}
DashboardRequest::GetRequestSeries { respond, .. } => {
let _ = respond.send(None);
}
}
}
})
}
async fn bind_dashboard(token: &str) -> (DashboardSetup, String) {
bind_dashboard_with(TestBind::token(token)).await
}
struct TestBind {
token: String,
no_auth: bool,
host: String,
max_clients: u32,
control: Option<flume::Sender<ControlRequest>>,
}
impl TestBind {
fn token(token: &str) -> Self {
TestBind {
token: token.to_string(),
no_auth: false,
host: "127.0.0.1".to_string(),
max_clients: 0,
control: None,
}
}
fn row(row: &crate::config::DashboardRow, control: flume::Sender<ControlRequest>) -> Self {
let configuration = crate::config::dashboard_row_configuration(row);
TestBind {
token: configuration
.dashboard_auth_token()
.unwrap_or_default()
.to_string(),
no_auth: configuration.dashboard_no_auth,
host: configuration.dashboard_host().to_string(),
max_clients: 0,
control: configuration.dashboard_control.then_some(control),
}
}
fn control(mut self, control: flume::Sender<ControlRequest>) -> Self {
self.control = Some(control);
self
}
fn max_clients(mut self, max_clients: u32) -> Self {
self.max_clients = max_clients;
self
}
}
async fn bind_dashboard_with(bind: TestBind) -> (DashboardSetup, String) {
let config = GooseConfiguration {
dashboard: true,
dashboard_control: bind.control.is_some(),
dashboard_no_auth: bind.no_auth,
dashboard_host: Some(bind.host),
dashboard_port: Some(0),
dashboard_auth_token: Some(bind.token),
dashboard_max_clients: (bind.max_clients > 0).then_some(bind.max_clients),
..Default::default()
};
let setup = setup_dashboard(
&config,
bind.control,
ListeningControllers::configured(&config),
)
.await
.expect("bind")
.expect("enabled");
let base = format!("http://127.0.0.1:{}", setup.bound_port);
let client = reqwest::Client::new();
for _ in 0..50 {
if let Ok(r) = client.get(format!("{base}/api/v1/health")).send().await {
if r.status().is_success() {
break;
}
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
(setup, base)
}
async fn setup_with_mock(
token: &str,
) -> (
Arc<AtomicU64>,
Arc<AtomicU64>,
tokio::task::JoinHandle<()>,
String,
DashboardServer,
) {
setup_with_mock_max(token, 0).await
}
async fn setup_with_mock_max(
token: &str,
max_clients: u32,
) -> (
Arc<AtomicU64>,
Arc<AtomicU64>,
tokio::task::JoinHandle<()>,
String,
DashboardServer,
) {
let (setup, base) =
bind_dashboard_with(TestBind::token(token).max_clients(max_clients)).await;
let hub_builds = Arc::clone(&setup.build_count);
let parent_builds = Arc::new(AtomicU64::new(0));
let parent = spawn_mock_parent(setup.request_rx, Arc::clone(&parent_builds));
drop(setup.close_tx);
(hub_builds, parent_builds, parent, base, setup.server)
}
async fn read_sse_until(
resp: &mut reqwest::Response,
timeout: Duration,
predicate: impl Fn(&str) -> bool,
) -> String {
let mut buf = String::new();
let deadline = Instant::now() + timeout;
while Instant::now() < deadline {
match tokio::time::timeout(Duration::from_millis(400), resp.chunk()).await {
Ok(Ok(Some(chunk))) => {
buf.push_str(&String::from_utf8_lossy(&chunk));
if predicate(&buf) {
return buf;
}
}
Ok(Ok(None)) => break,
Ok(Err(e)) => panic!("stream error: {}", e),
Err(_) => continue,
}
}
buf
}
#[test]
fn accepts_gzip_reads_accept_encoding() {
let with = |value: Option<&str>| {
let mut headers = HeaderMap::new();
if let Some(value) = value {
headers.insert(
header::ACCEPT_ENCODING,
HeaderValue::from_str(value).unwrap(),
);
}
accepts_gzip(&headers)
};
for yes in ["gzip", "gzip, deflate, br", "br;q=1.0, GZIP;q=0.5", "*"] {
assert!(with(Some(yes)), "{}", yes);
}
for no in [
"identity",
"deflate, br",
"gzip;q=0",
"gzip; q=0.0",
"*;q=0",
"gzip;Q=0",
] {
assert!(!with(Some(no)), "{}", no);
}
assert!(!with(None));
}
#[test]
fn constant_time_eq_matches() {
assert!(constant_time_eq(b"secret", b"secret"));
assert!(!constant_time_eq(b"secret", b"Secret"));
assert!(!constant_time_eq(b"short", b"longer"));
assert!(!constant_time_eq(b"", b"x"));
assert!(constant_time_eq(b"", b""));
}
#[test]
fn authorize_empty_token_allows_all() {
let headers = HeaderMap::new();
assert!(authorize("", &headers, None));
assert!(authorize("", &headers, Some("anything")));
}
#[test]
fn authorize_requires_token_when_configured() {
let headers = HeaderMap::new();
assert!(!authorize("s3cret", &headers, None));
assert!(!authorize("s3cret", &headers, Some("wrong")));
assert!(authorize("s3cret", &headers, Some("s3cret")));
let mut headers = HeaderMap::new();
headers.insert(
header::AUTHORIZATION,
HeaderValue::from_static("Bearer s3cret"),
);
assert!(authorize("s3cret", &headers, None));
let mut headers = HeaderMap::new();
headers.insert(
header::AUTHORIZATION,
HeaderValue::from_static("Bearer wrong"),
);
assert!(!authorize("s3cret", &headers, None));
}
#[test]
fn authorize_control_always_requires_token() {
let headers = HeaderMap::new();
assert!(!authorize_control("", &headers, None));
assert!(!authorize_control("", &headers, Some("anything")));
assert!(!authorize_control("s3cret", &headers, None));
assert!(!authorize_control("s3cret", &headers, Some("wrong")));
assert!(!authorize_control("s3cret", &headers, Some("s3cret")));
let mut headers = HeaderMap::new();
headers.insert(
header::AUTHORIZATION,
HeaderValue::from_static("Bearer s3cret"),
);
assert!(authorize_control("s3cret", &headers, None));
assert!(!authorize_control("s3cret", &headers, Some("wrong")));
assert!(!authorize_control("s3cret", &headers, Some("s3cret")));
let mut headers = HeaderMap::new();
headers.insert(
header::AUTHORIZATION,
HeaderValue::from_static("bearer s3cret"),
);
assert!(authorize_control("s3cret", &headers, None));
assert!(authorize("s3cret", &headers, None));
let mut headers = HeaderMap::new();
headers.insert(
header::AUTHORIZATION,
HeaderValue::from_static("Bearer\ts3cret"),
);
assert!(authorize("s3cret", &headers, None));
assert!(authorize_control("s3cret", &headers, None));
let mut headers = HeaderMap::new();
headers.insert(
header::AUTHORIZATION,
HeaderValue::from_static("Bearer s3cret"),
);
assert!(authorize("s3cret", &headers, None));
let mut headers = HeaderMap::new();
headers.insert(
header::AUTHORIZATION,
HeaderValue::from_static("Bearer s3cret extra"),
);
assert!(!authorize("s3cret", &headers, None));
}
#[test]
fn constant_time_eq_length_mismatch() {
assert!(!constant_time_eq(b"abc", b"ab"));
assert!(!constant_time_eq(b"ab", b"abc"));
assert!(constant_time_eq(b"secret", b"secret"));
assert!(!constant_time_eq(b"secret", b"secreT"));
}
#[test]
fn parse_empty_control_body_rules() {
assert!(parse_empty_control_body(b"").is_ok());
assert!(parse_empty_control_body(b" ").is_ok());
assert!(parse_empty_control_body(b"{}").is_ok());
assert!(parse_empty_control_body(b" {} ").is_ok());
assert!(parse_empty_control_body(b"[]").is_err());
assert!(parse_empty_control_body(b"null").is_err());
assert!(parse_empty_control_body(b"{\"extra\":1}").is_err());
assert!(parse_empty_control_body(b"not-json").is_err());
}
#[test]
fn parse_users_body_rules() {
assert!(matches!(
parse_users_body(b"{\"users\":50}"),
Ok(ControllerCommand::Users(50))
));
assert!(matches!(
parse_users_body(b"{\"users\":1}"),
Ok(ControllerCommand::Users(1))
));
assert!(matches!(
parse_users_body(b"{\"users\":100000}"),
Ok(ControllerCommand::Users(100000))
));
let err = parse_users_body(b"").unwrap_err();
assert_eq!(err.0, "bad_request");
let err = parse_users_body(b"[]").unwrap_err();
assert_eq!(err.0, "bad_request");
let err = parse_users_body(b"not-json").unwrap_err();
assert_eq!(err.0, "bad_request");
let err = parse_users_body(b"{}").unwrap_err();
assert_eq!(err.0, "invalid_users");
let err = parse_users_body(b"{\"users\":0}").unwrap_err();
assert_eq!(err.0, "invalid_users");
let err = parse_users_body(b"{\"users\":100001}").unwrap_err();
assert_eq!(err.0, "invalid_users");
let err = parse_users_body(b"{\"users\":1000000}").unwrap_err();
assert_eq!(err.0, "invalid_users");
let err = parse_users_body(b"{\"users\":-1}").unwrap_err();
assert_eq!(err.0, "invalid_users");
let err = parse_users_body(b"{\"users\":1.5}").unwrap_err();
assert_eq!(err.0, "invalid_users");
let err = parse_users_body(b"{\"users\":\"10\"}").unwrap_err();
assert_eq!(err.0, "invalid_users");
let users = |digits: usize| format!("{{\"users\":{}}}", "9".repeat(digits));
let err = parse_users_body(users(76).as_bytes()).unwrap_err();
assert_eq!(err.0, "invalid_users");
let err = parse_users_body(users(308).as_bytes()).unwrap_err();
assert_eq!(err.0, "invalid_users");
let err = parse_users_body(users(309).as_bytes()).unwrap_err();
assert_eq!(err, ("bad_request", "malformed JSON body"));
}
#[test]
fn parse_settings_body_rules() {
let settings = parse_settings_body(br#"{}"#).unwrap();
assert!(settings.fields().is_empty());
let settings = parse_settings_body(
br#"{"host":"https://example.com","users":5,"increase_rate":2.5,"decrease_time_secs":null,"run_time_secs":90}"#,
)
.unwrap();
assert_eq!(settings.host, Some(Some("https://example.com".to_string())));
assert_eq!(settings.users, Some(5));
assert_eq!(settings.increase_rate, Some(Some(2.5)));
assert_eq!(settings.increase_time, None);
assert_eq!(settings.decrease_rate, None);
assert_eq!(settings.decrease_time, Some(None));
assert_eq!(settings.run_time, Some(Some(90)));
assert!(settings.test_plan.is_none());
let settings =
parse_settings_body(br#"{"host":null,"test_plan":"5,1s;0,1s","increase_rate":null}"#)
.unwrap();
assert_eq!(settings.host, Some(None));
assert_eq!(settings.increase_rate, Some(None));
assert_eq!(
settings.test_plan.unwrap().unwrap().steps,
[(5, 1_000), (0, 1_000)]
);
assert!(parse_settings_body(br#"{"users":2,"test_plan":null}"#).is_ok());
assert!(parse_settings_body(br#"{"increase_rate":2,"increase_time_secs":null}"#).is_ok());
for (body, refusal) in [
(&b"{"[..], ("bad_request", None)),
(b"[1]", ("bad_request", None)),
(b"5", ("bad_request", None)),
(
br#"{"increase_rate":0,"iterations":1}"#,
("bad_request", None),
),
(br#"{"run_time":60}"#, ("bad_request", None)),
(
br#"{"host":"ftp://example.com"}"#,
("invalid_value", Some("host")),
),
(
br#"{"host":"http://user:secret@example.com"}"#,
("invalid_value", Some("host")),
),
(br#"{"host":5}"#, ("invalid_value", Some("host"))),
(br#"{"users":null}"#, ("invalid_value", Some("users"))),
(br#"{"users":0}"#, ("invalid_value", Some("users"))),
(br#"{"users":"5"}"#, ("invalid_value", Some("users"))),
(
br#"{"increase_rate":0.001,"test_plan":"1,1s","users":3}"#,
("invalid_value", Some("increase_rate")),
),
(
br#"{"decrease_rate":"2"}"#,
("invalid_value", Some("decrease_rate")),
),
(
br#"{"decrease_rate":100001}"#,
("invalid_value", Some("decrease_rate")),
),
(
br#"{"increase_time_secs":0}"#,
("invalid_value", Some("increase_time")),
),
(
br#"{"decrease_time_secs":1.5}"#,
("invalid_value", Some("decrease_time")),
),
(
br#"{"run_time_secs":2592001}"#,
("invalid_value", Some("run_time")),
),
(
br#"{"run_time_secs":-1}"#,
("invalid_value", Some("run_time")),
),
(
br#"{"test_plan":"0,1s"}"#,
("invalid_value", Some("test_plan")),
),
(br#"{"test_plan":10}"#, ("invalid_value", Some("test_plan"))),
(
br#"{"test_plan":"1,1s","users":3}"#,
("conflicting_settings", Some("test_plan")),
),
(
br#"{"test_plan":"1,1s","run_time_secs":10}"#,
("conflicting_settings", Some("run_time")),
),
(
br#"{"test_plan":null}"#,
("conflicting_settings", Some("test_plan")),
),
(
br#"{"increase_rate":2,"increase_time_secs":5}"#,
("conflicting_settings", Some("increase_rate")),
),
(
br#"{"decrease_rate":2,"decrease_time_secs":5}"#,
("conflicting_settings", Some("decrease_rate")),
),
] {
let (code, _, field) = parse_settings_body(body).unwrap_err();
assert_eq!((code, field), refusal, "{}", String::from_utf8_lossy(body));
}
let plan = format!(r#"{{"test_plan":"{}"}}"#, "1,1s;".repeat(1000));
let (code, _, field) = parse_settings_body(plan.as_bytes()).unwrap_err();
assert_eq!((code, field), ("invalid_value", Some("test_plan")));
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn settings_routes_require_token_and_check_body() {
let (control_tx, control_rx) = flume::unbounded();
let (setup, base) =
bind_dashboard_with(TestBind::token("s3cret").control(control_tx)).await;
let parent = spawn_mock_parent(setup.request_rx, Arc::new(AtomicU64::new(0)));
let control = spawn_control_parent(control_rx, answer_ok);
drop(setup.close_tx);
let client = reqwest::Client::new();
let post = |path: &str, auth: Option<&str>, body: &str| {
let mut request = client
.post(format!("{base}/api/v1/control/{path}"))
.header(header::CONTENT_TYPE, "application/json")
.body(body.to_string());
if let Some(auth) = auth {
request = request.header(header::AUTHORIZATION, auth.to_string());
}
request.send()
};
let body = r#"{"host":"https://example.com","increase_rate":2}"#;
for path in [
"settings",
"start",
"settings?token=s3cret",
"start?token=s3cret",
] {
for auth in [None, Some("Bearer wrong")] {
let resp = post(path, auth, body).await.expect("post");
assert_eq!(resp.status(), 401, "{} {:?}", path, auth);
}
}
let resp = post("settings?token=s3cret", Some("Bearer s3cret"), body)
.await
.expect("post");
assert_eq!(resp.status(), 401);
for (path, message) in [
("settings", "settings configured"),
("start", "load test started"),
] {
let resp = post(path, Some("Bearer s3cret"), body).await.expect("post");
assert_eq!(resp.status(), 200, "{}", path);
let reply: serde_json::Value = resp.json().await.expect("json");
assert_eq!(reply["ok"], true, "{}", path);
assert_eq!(reply["command"], path);
assert_eq!(reply["message"], message);
let resp = post(path, Some("Bearer s3cret"), r#"{"increase_rate":0}"#)
.await
.expect("post");
assert_eq!(resp.status(), 400, "{}", path);
let reply: serde_json::Value = resp.json().await.expect("json");
assert_eq!(reply["error"], "invalid_value");
assert_eq!(reply["field"], "increase_rate");
assert_eq!(reply["command"], path);
let resp = post(path, Some("Bearer s3cret"), r#"{"bogus":1}"#)
.await
.expect("post");
assert_eq!(resp.status(), 400, "{}", path);
let reply: serde_json::Value = resp.json().await.expect("json");
assert_eq!(reply["error"], "bad_request");
assert_eq!(reply["field"], serde_json::Value::Null);
let resp = post(
path,
Some("Bearer s3cret"),
r#"{"increase_rate":2,"increase_time_secs":5}"#,
)
.await
.expect("post");
assert_eq!(resp.status(), 400, "{}", path);
let reply: serde_json::Value = resp.json().await.expect("json");
assert_eq!(reply["error"], "conflicting_settings");
assert_eq!(reply["field"], "increase_rate");
let big = format!(r#"{{"test_plan":"{}"}}"#, " ".repeat(CONTROL_BODY_LIMIT));
let resp = post(path, Some("Bearer s3cret"), &big).await.expect("post");
assert_eq!(resp.status(), 413, "{}", path);
}
parent.abort();
control.abort();
}
#[test]
fn format_bind_address_brackets_ipv6() {
assert_eq!(format_bind_address("127.0.0.1", 5118), "127.0.0.1:5118");
assert_eq!(format_bind_address("localhost", 5118), "localhost:5118");
assert_eq!(format_bind_address("::1", 5118), "[::1]:5118");
assert_eq!(format_bind_address("[::1]", 5118), "[::1]:5118");
assert_eq!(format_bind_address("0.0.0.0", 80), "0.0.0.0:80");
}
#[tokio::test]
async fn binds_localhost_and_ipv6_loopback() {
let config = GooseConfiguration {
dashboard: true,
dashboard_host: Some("localhost".to_string()),
dashboard_port: Some(0),
dashboard_auth_token: None,
..Default::default()
};
let setup = setup_dashboard(&config, None, ListeningControllers::configured(&config))
.await
.expect("localhost bind should succeed")
.expect("dashboard enabled");
assert!(setup.bound_port > 0, "ephemeral port must be recorded");
drop(setup);
let config = GooseConfiguration {
dashboard: true,
dashboard_host: Some("::1".to_string()),
dashboard_port: Some(0),
dashboard_auth_token: None,
..Default::default()
};
match setup_dashboard(&config, None, ListeningControllers::configured(&config)).await {
Ok(Some(setup)) => {
assert!(setup.bound_port > 0);
}
Ok(None) => panic!("dashboard enabled but setup returned None"),
Err(e) => {
let msg = format!("{e}");
eprintln!("skipping ::1 bind assertion: {msg}");
}
}
}
#[tokio::test]
async fn bind_failure_is_hard_error() {
let occupied = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("occupy port");
let port = occupied.local_addr().unwrap().port();
let config = GooseConfiguration {
dashboard: true,
dashboard_host: Some("127.0.0.1".to_string()),
dashboard_port: Some(port),
dashboard_auth_token: None,
..Default::default()
};
let err = setup_dashboard(&config, None, ListeningControllers::configured(&config))
.await
.expect_err("second bind on same port must fail hard");
match err {
GooseError::Io(_) => {}
other => panic!("expected GooseError::Io, got {:?}", other),
}
drop(occupied);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn idle_hub_issues_no_snapshot_commands() {
let (hub_builds, parent_builds, parent, _base, _server) = setup_with_mock("").await;
tokio::time::sleep(Duration::from_millis(1500)).await;
assert_eq!(
parent_builds.load(Ordering::SeqCst),
0,
"no clients ⇒ parent must not see GetSnapshot"
);
assert_eq!(hub_builds.load(Ordering::SeqCst), 0);
parent.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn multi_client_sse_coalesces_to_about_one_build_per_sec() {
let (hub_builds, parent_builds, parent, base, _server) = setup_with_mock("").await;
let client = reqwest::Client::new();
let n_clients = 6usize;
let mut responses = Vec::new();
for _ in 0..n_clients {
let resp = client
.get(format!("{base}/api/v1/events"))
.send()
.await
.expect("sse connect");
assert_eq!(resp.status(), 200, "SSE must accept client");
let ct = resp
.headers()
.get(header::CONTENT_TYPE)
.and_then(|v| v.to_str().ok())
.unwrap_or("");
assert!(
ct.contains("text/event-stream"),
"content-type was {:?}",
ct
);
responses.push(resp);
}
let start = Instant::now();
tokio::time::sleep(Duration::from_millis(3200)).await;
let elapsed = start.elapsed().as_secs_f64();
drop(responses);
let builds = parent_builds.load(Ordering::SeqCst);
let hub = hub_builds.load(Ordering::SeqCst);
assert!(
builds >= 2,
"expected several coalesced builds over {:.1}s, got {}",
elapsed,
builds
);
assert!(
builds as f64 <= elapsed + 3.0,
"builds={} over {:.1}s looks uncoalesced (clients={})",
builds,
elapsed,
n_clients
);
assert!(
(builds as f64) < (n_clients as f64) * elapsed * 0.5,
"builds={} too high for coalescing with {} clients over {:.1}s",
builds,
n_clients,
elapsed
);
assert_eq!(builds, hub, "hub and parent build counters should match");
let after_drop = parent_builds.load(Ordering::SeqCst);
tokio::time::sleep(Duration::from_millis(3500)).await;
let idle_baseline = parent_builds.load(Ordering::SeqCst);
assert!(
idle_baseline <= after_drop + 2,
"builds should stop after clients disconnect (after_drop={}, now={})",
after_drop,
idle_baseline
);
tokio::time::sleep(Duration::from_millis(1500)).await;
assert_eq!(
parent_builds.load(Ordering::SeqCst),
idle_baseline,
"idle hub must not issue further GetSnapshot after clients disconnect"
);
parent.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn sse_client_cap_rejects_33rd_with_503() {
let (_hub_builds, _parent_builds, parent, base, _server) = setup_with_mock("").await;
let client = reqwest::Client::new();
let mut held = Vec::with_capacity(MAX_SSE_CLIENTS);
for i in 0..MAX_SSE_CLIENTS {
let resp = client
.get(format!("{base}/api/v1/events"))
.send()
.await
.unwrap_or_else(|e| panic!("client {} connect failed: {}", i, e));
assert_eq!(resp.status(), 200, "client {} should be accepted", i);
held.push(resp);
}
let overflow = client
.get(format!("{base}/api/v1/events"))
.send()
.await
.expect("33rd connect");
assert_eq!(overflow.status(), 503, "33rd SSE client must receive 503");
let body = overflow.text().await.unwrap_or_default();
assert!(
body.contains("too many") || body.contains("SSE"),
"503 should include a short plain-text body, got {:?}",
body
);
drop(held);
parent.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn sse_client_cap_respects_dashboard_max_clients() {
let cap = 2u32;
let (_hub_builds, _parent_builds, parent, base, _server) =
setup_with_mock_max("", cap).await;
let client = reqwest::Client::new();
let mut held = Vec::with_capacity(cap as usize);
for i in 0..cap {
let resp = client
.get(format!("{base}/api/v1/events"))
.send()
.await
.unwrap_or_else(|e| panic!("client {} connect failed: {}", i, e));
assert_eq!(resp.status(), 200, "client {} should be accepted", i);
held.push(resp);
}
let overflow = client
.get(format!("{base}/api/v1/events"))
.send()
.await
.expect("overflow connect");
assert_eq!(
overflow.status(),
503,
"client beyond --dashboard-max-clients must receive 503"
);
let health = client
.get(format!("{base}/api/v1/health"))
.send()
.await
.expect("health");
assert_eq!(health.status(), 200);
let health_json: serde_json::Value = health.json().await.expect("health json");
assert_eq!(health_json["ok"], true);
assert_eq!(health_json["active_sse_clients"], cap);
drop(held);
parent.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn health_exposes_build_timing_and_client_counts() {
let (hub_builds, _parent_builds, parent, base, _server) = setup_with_mock("").await;
let client = reqwest::Client::new();
let idle = client
.get(format!("{base}/api/v1/health"))
.send()
.await
.expect("health idle");
assert_eq!(idle.status(), 200);
let idle_json: serde_json::Value = idle.json().await.expect("json");
assert_eq!(idle_json["ok"], true);
assert!(!idle_json["version"].as_str().unwrap().is_empty());
assert_eq!(idle_json["build_count"], 0);
assert_eq!(idle_json["last_build_ms"], 0);
assert_eq!(idle_json["last_snapshot_ms"], 0);
assert_eq!(idle_json["snapshot_timeouts"], 0);
assert_eq!(idle_json["active_sse_clients"], 0);
let snap = client
.get(format!("{base}/api/v1/snapshot"))
.send()
.await
.expect("snapshot");
assert_eq!(snap.status(), 200);
for _ in 0..50 {
if hub_builds.load(Ordering::SeqCst) >= 1 {
break;
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
assert!(
hub_builds.load(Ordering::SeqCst) >= 1,
"expected at least one hub build after snapshot poll"
);
let after = client
.get(format!("{base}/api/v1/health"))
.send()
.await
.expect("health after");
let after_json: serde_json::Value = after.json().await.expect("json");
assert!(
after_json["build_count"].as_u64().unwrap() >= 1,
"build_count should advance after a snapshot, got {}",
after_json["build_count"]
);
assert!(
after_json["last_build_ms"].as_u64().is_some(),
"last_build_ms must be a number, got {}",
after_json["last_build_ms"]
);
assert_eq!(after_json["active_sse_clients"], 0);
parent.abort();
}
fn bare_hub() -> (SnapshotHub, flume::Receiver<DashboardRequest>) {
let (request_tx, request_rx) = flume::unbounded();
let (shutdown_tx, _) = watch::channel(false);
let hub = SnapshotHub::new(
request_tx,
Arc::new(AtomicU64::new(0)),
shutdown_tx,
0,
Arc::new(AtomicUsize::new(0)),
);
(hub, request_rx)
}
#[tokio::test]
async fn build_records_processor_time_and_round_trip() {
let (hub, request_rx) = bare_hub();
let parent = tokio::spawn(async move {
while let Ok(msg) = request_rx.recv_async().await {
if let DashboardRequest::GetSnapshot { respond, .. } = msg {
let started = Instant::now();
tokio::time::sleep(Duration::from_millis(5)).await;
let mut snapshot = dummy_snapshot(0);
snapshot.build_time = started.elapsed();
let _ = respond.send(snapshot);
}
}
});
assert!(matches!(
build_once(&hub.inner, SNAPSHOT_BUILD_TIMEOUT).await,
BuildOutcome::Built
));
let build_ms = hub.last_build_ms();
let snapshot_ms = hub.last_snapshot_ms();
assert!(build_ms >= 5, "last_build_ms {}", build_ms);
assert!(
build_ms <= snapshot_ms,
"last_build_ms {} > last_snapshot_ms {}",
build_ms,
snapshot_ms
);
assert_eq!(hub.snapshot_timeouts(), 0);
assert!(!hub.inner.snapshot_in_flight.load(Ordering::Acquire));
let json = hub.inner.latest_tx.borrow().clone().expect("published");
assert!(!String::from_utf8_lossy(&json).contains("build_time"));
parent.abort();
}
#[tokio::test]
async fn snapshot_timeout_is_counted_and_late_reply_dropped() {
assert_eq!(SNAPSHOT_BUILD_TIMEOUT, Duration::from_secs(4));
let timeout = Duration::from_millis(50);
let (hub, request_rx) = bare_hub();
let parent = tokio::spawn(async move {
match request_rx.recv_async().await {
Ok(DashboardRequest::GetSnapshot { respond, .. }) => (respond, request_rx),
_ => panic!("expected GetSnapshot"),
}
});
assert!(matches!(
build_once(&hub.inner, timeout).await,
BuildOutcome::Transient
));
assert_eq!(hub.snapshot_timeouts(), 1);
assert!(hub.inner.snapshot_in_flight.load(Ordering::Acquire));
assert!(matches!(
build_once(&hub.inner, timeout).await,
BuildOutcome::Transient
));
assert_eq!(hub.snapshot_timeouts(), 1);
let (respond, _request_rx) = parent.await.expect("parent");
let _ = respond.send(dummy_snapshot(0));
for _ in 0..50 {
if !hub.inner.snapshot_in_flight.load(Ordering::Acquire) {
break;
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
assert!(!hub.inner.snapshot_in_flight.load(Ordering::Acquire));
assert!(hub.inner.latest_tx.borrow().is_none());
assert_eq!(hub.build_count(), 0);
assert_eq!(hub.last_build_ms(), 0);
assert_eq!(hub.last_snapshot_ms(), 0);
}
#[tokio::test]
async fn health_reports_controller_addresses() {
let cases = [
(
false,
true,
true,
None,
serde_json::json!({"telnet": "127.0.0.1:5116", "websocket": "[::1]:6117"}),
),
(
true,
true,
true,
None,
serde_json::json!({"telnet": null, "websocket": "[::1]:6117"}),
),
(
true,
false,
true,
None,
serde_json::json!({"telnet": null, "websocket": null}),
),
(
false,
false,
true,
None,
serde_json::json!({"telnet": "127.0.0.1:5116", "websocket": null}),
),
(
false,
false,
true,
Some("0.0.0.0"),
serde_json::json!({"telnet": "0.0.0.0:5116", "websocket": null}),
),
(
false,
true,
false,
None,
serde_json::json!({"telnet": null, "websocket": "[::1]:6117"}),
),
];
let client = reqwest::Client::new();
for (no_telnet, websocket, telnet_bound, telnet_host, expected) in cases {
let config = GooseConfiguration {
dashboard: true,
dashboard_host: Some("127.0.0.1".to_string()),
dashboard_port: Some(0),
no_telnet,
telnet_host: telnet_host.map(str::to_string),
telnet_port: Some(5116),
websocket,
websocket_host: Some("::1".to_string()),
websocket_port: Some(6117),
..Default::default()
};
let mut listening = ListeningControllers::configured(&config);
listening.telnet &= telnet_bound;
let setup = setup_dashboard(&config, None, listening)
.await
.expect("bind")
.expect("enabled");
let health: serde_json::Value = client
.get(format!(
"http://127.0.0.1:{}/api/v1/health",
setup.bound_port
))
.send()
.await
.expect("health")
.json()
.await
.expect("json");
assert_eq!(
health["controllers"], expected,
"no_telnet {}, websocket {}, telnet bound {}, telnet host {:?}",
no_telnet, websocket, telnet_bound, telnet_host
);
assert_eq!(health["ok"], true);
assert_eq!(health["control_enabled"], false);
for field in [
"version",
"last_build_ms",
"last_snapshot_ms",
"snapshot_timeouts",
"build_count",
"active_sse_clients",
] {
assert!(!health[field].is_null(), "{} missing", field);
}
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn events_require_token_when_configured() {
let token = "sse-secret";
let (_hub, _parent_builds, parent, base, _server) = setup_with_mock(token).await;
let client = reqwest::Client::new();
let unauth = client
.get(format!("{base}/api/v1/events"))
.send()
.await
.expect("unauth");
assert_eq!(unauth.status(), 401);
let wrong = client
.get(format!("{base}/api/v1/events?token=wrong"))
.send()
.await
.expect("wrong");
assert_eq!(wrong.status(), 401);
let ok = client
.get(format!("{base}/api/v1/events?token={token}"))
.send()
.await
.expect("query token");
assert_eq!(ok.status(), 200);
let bearer = client
.get(format!("{base}/api/v1/events"))
.header("Authorization", format!("Bearer {token}"))
.send()
.await
.expect("bearer");
assert_eq!(bearer.status(), 200);
parent.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn poll_only_snapshots_refresh_without_sse() {
let counter = Arc::new(AtomicU64::new(0));
let counter2 = Arc::clone(&counter);
let (tx, rx) = flume::unbounded();
let parent = tokio::spawn(async move {
while let Ok(req) = rx.recv_async().await {
match req {
DashboardRequest::GetSnapshot { respond, .. } => {
let n = counter2.fetch_add(1, Ordering::SeqCst);
let mut snap = dummy_snapshot(n + 1);
snap.active_users = n + 1;
snap.phase = if n == 0 {
"maintain".into()
} else {
"increase".into()
};
let _ = respond.send(snap);
}
DashboardRequest::GetRequestSeries { respond, .. } => {
let _ = respond.send(None);
}
}
}
});
let build_count = Arc::new(AtomicU64::new(0));
let (close_tx, _) = watch::channel(false);
let hub = SnapshotHub::new(
tx,
Arc::clone(&build_count),
close_tx,
MAX_SSE_CLIENTS,
Arc::new(AtomicUsize::new(0)),
);
let b1 = hub.snapshot_bytes().await.expect("first");
let v1: serde_json::Value = serde_json::from_slice(&b1).expect("json1");
let users1 = v1["active_users"].as_u64().unwrap();
tokio::time::sleep(SNAPSHOT_INTERVAL + Duration::from_millis(50)).await;
let b2 = hub.snapshot_bytes().await.expect("second");
let v2: serde_json::Value = serde_json::from_slice(&b2).expect("json2");
let users2 = v2["active_users"].as_u64().unwrap();
assert!(
users2 > users1,
"poll-only path must publish updated snapshots; got users {users1} then {users2}, builds={}",
build_count.load(Ordering::SeqCst)
);
assert!(
build_count.load(Ordering::SeqCst) >= 2,
"expected at least 2 hub builds, got {}",
build_count.load(Ordering::SeqCst)
);
parent.abort();
drop(hub);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn snapshot_goes_through_hub() {
let (hub_builds, parent_builds, parent, base, _server) = setup_with_mock("").await;
let client = reqwest::Client::new();
let resp = client
.get(format!("{base}/api/v1/snapshot"))
.send()
.await
.expect("snapshot");
assert_eq!(resp.status(), 200);
let body: serde_json::Value = resp.json().await.expect("json");
assert_eq!(body["version"], 1);
assert_eq!(body["phase"], "maintain");
let start_builds = parent_builds.load(Ordering::SeqCst);
let mut joins = Vec::new();
for _ in 0..8 {
let c = client.clone();
let url = format!("{base}/api/v1/snapshot");
joins.push(tokio::spawn(async move {
c.get(url).send().await.expect("poll").status()
}));
}
for j in joins {
assert_eq!(j.await.unwrap(), 200);
}
let delta = parent_builds.load(Ordering::SeqCst) - start_builds;
assert!(
delta <= 2,
"8 concurrent polls should coalesce (delta builds={})",
delta
);
assert!(hub_builds.load(Ordering::SeqCst) >= 1);
parent.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn sse_stream_emits_snapshot_event() {
let (_hub, _parent_builds, parent, base, _server) = setup_with_mock("").await;
let client = reqwest::Client::new();
let mut resp = client
.get(format!("{base}/api/v1/events"))
.send()
.await
.expect("sse");
assert_eq!(resp.status(), 200);
let buf = read_sse_until(&mut resp, Duration::from_secs(3), |b| {
b.contains("event: snapshot") && b.contains("data: {")
})
.await;
assert!(
buf.contains("event: snapshot") && buf.contains("data: {"),
"did not observe snapshot SSE event in body: {:?}",
buf
);
parent.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn sse_emits_closed_when_parent_disconnects() {
let (setup, base) = bind_dashboard("").await;
let parent_builds = Arc::new(AtomicU64::new(0));
let parent = spawn_mock_parent(setup.request_rx, Arc::clone(&parent_builds));
let _close_tx = setup.close_tx;
let client = reqwest::Client::new();
let mut resp = client
.get(format!("{base}/api/v1/events"))
.send()
.await
.expect("sse");
assert_eq!(resp.status(), 200);
let buf = read_sse_until(&mut resp, Duration::from_secs(3), |b| {
b.contains("event: snapshot")
})
.await;
assert!(
buf.contains("event: snapshot"),
"expected snapshot before close, got {:?}",
buf
);
parent.abort();
tokio::time::sleep(Duration::from_millis(50)).await;
let buf = read_sse_until(&mut resp, Duration::from_secs(3), |b| {
b.contains("event: closed")
})
.await;
assert!(
buf.contains("event: closed"),
"expected event: closed after parent disconnect, got {:?}",
buf
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn sse_emits_closed_on_proactive_close_and_late_subscriber() {
let (setup, base) = bind_dashboard("").await;
let parent_builds = Arc::new(AtomicU64::new(0));
let parent = spawn_mock_parent(setup.request_rx, Arc::clone(&parent_builds));
let close_tx = setup.close_tx;
let client = reqwest::Client::new();
let mut resp = client
.get(format!("{base}/api/v1/events"))
.send()
.await
.expect("sse");
assert_eq!(resp.status(), 200);
let _ = read_sse_until(&mut resp, Duration::from_secs(3), |b| {
b.contains("event: snapshot")
})
.await;
let _ = close_tx.send(true);
let buf = read_sse_until(&mut resp, Duration::from_secs(2), |b| {
b.contains("event: closed")
})
.await;
assert!(
buf.contains("event: closed"),
"live client must see closed after proactive signal, got {:?}",
buf
);
let mut late = client
.get(format!("{base}/api/v1/events"))
.send()
.await
.expect("late sse");
assert_eq!(late.status(), 200);
let late_buf = read_sse_until(&mut late, Duration::from_secs(2), |b| {
b.contains("event: closed")
})
.await;
assert!(
late_buf.contains("event: closed"),
"late subscriber after shutdown must receive closed, got {:?}",
late_buf
);
parent.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn oneshot_cancel_does_not_kill_hub() {
let (setup, base) = bind_dashboard("").await;
let parent_builds = Arc::new(AtomicU64::new(0));
let parent = spawn_blip_parent(setup.request_rx, Arc::clone(&parent_builds), 2, 5);
drop(setup.close_tx);
let client = reqwest::Client::new();
let mut resp = client
.get(format!("{base}/api/v1/events"))
.send()
.await
.expect("sse");
assert_eq!(resp.status(), 200);
tokio::time::sleep(Duration::from_millis(7000)).await;
let builds = parent_builds.load(Ordering::SeqCst);
assert!(
builds >= 6,
"hub must keep requesting after oneshot cancels (builds={})",
builds
);
let buf = read_sse_until(&mut resp, Duration::from_secs(2), |b| {
b.contains("event: snapshot")
})
.await;
assert!(
buf.contains("event: snapshot"),
"SSE must still deliver snapshots after transient oneshot cancels, got {:?}",
buf
);
assert!(
!buf.contains("event: closed"),
"transient oneshot cancel must not emit closed"
);
parent.abort();
}
const MATRIX_REQUESTS: [(&str, &str, &str); 10] = [
("GET", "/api/v1/health", ""),
("GET", "/api/v1/snapshot", ""),
("POST", "/api/v1/control/start", "{}"),
("POST", "/api/v1/control/stop", "{}"),
("POST", "/api/v1/control/users", "{\"users\":3}"),
("POST", "/api/v1/control/settings", "{}"),
("POST", "/api/v1/control/report", "{\"save\":true}"),
("POST", "/api/v1/control/report/now", "{}"),
("POST", "/api/v1/control/quit", "{}"),
("DELETE", "/api/v1/reports/no-such-run", ""),
];
const MATRIX_STATUS: [(&str, [[u16; 3]; 10]); 7] = [
(
"R1",
[
[200; 3], [200; 3], [404; 3], [404; 3], [404; 3], [404; 3], [404; 3], [404; 3],
[404; 3], [404; 3],
],
),
(
"R2",
[
[200; 3],
[401, 200, 401],
[404; 3],
[404; 3],
[404; 3],
[404; 3],
[404; 3],
[404; 3],
[404; 3],
[404; 3],
],
),
(
"R3",
[
[200; 3],
[401, 200, 401],
[401, 200, 401],
[401, 200, 401],
[401, 200, 401],
[401, 200, 401],
[401, 200, 401],
[401, 200, 401],
[401, 200, 401],
[401, 404, 401],
],
),
(
"R4",
[
[200; 3], [200; 3], [200; 3], [200; 3], [200; 3], [200; 3], [200; 3], [200; 3],
[200; 3], [404; 3],
],
),
(
"R5",
[
[200; 3], [200; 3], [404; 3], [404; 3], [404; 3], [404; 3], [404; 3], [404; 3],
[404; 3], [404; 3],
],
),
(
"R6",
[
[200; 3], [200; 3], [200; 3], [200; 3], [200; 3], [200; 3], [200; 3], [200; 3],
[200; 3], [404; 3],
],
),
(
"R11",
[
[200; 3],
[401, 200, 401],
[401, 200, 401],
[401, 200, 401],
[401, 200, 401],
[401, 200, 401],
[401, 200, 401],
[401, 200, 401],
[401, 200, 401],
[401, 404, 401],
],
),
];
const MATRIX_HEALTH: [(&str, [bool; 3]); 7] = [
("R1", [false, true, true]),
("R2", [false, true, true]),
("R3", [true, true, true]),
("R4", [true, false, true]),
("R5", [false, false, true]),
("R6", [true, false, false]),
("R11", [true, true, false]),
];
const MATRIX_PROBES: [(&str, [u16; 4]); 8] = [
("/api/v1/control/start", [403, 403, 403, 200]),
("/api/v1/control/stop", [403, 403, 403, 200]),
("/api/v1/control/users", [403, 403, 403, 200]),
("/api/v1/control/settings", [403, 403, 403, 200]),
("/api/v1/control/report", [403, 403, 403, 200]),
("/api/v1/control/report/now", [403, 403, 403, 200]),
("/api/v1/control/quit", [403, 403, 403, 200]),
("/api/v1/reports/no-such-run", [403, 403, 403, 404]),
];
const MATRIX_HOSTS: [(&str, [u16; 4]); 7] = [
("R1", [403, 403, 200, 200]),
("R2", [403, 403, 401, 401]),
("R3", [403, 403, 401, 401]),
("R4", [403, 403, 200, 200]),
("R5", [403, 403, 200, 200]),
("R6", [403, 403, 200, 200]),
("R11", [401, 401, 401, 401]),
];
type Headers<'a> = Vec<(&'a str, &'a str)>;
async fn raw_call(
port: u16,
method: &str,
path: &str,
host: &str,
headers: &[(&str, &str)],
body: &str,
) -> (u16, String) {
let mut request = format!(
"{method} {path} HTTP/1.1\r\nHost: {host}\r\nConnection: close\r\nContent-Length: {}\r\n",
body.len()
);
for (name, value) in headers {
request.push_str(&format!("{name}: {value}\r\n"));
}
request.push_str("\r\n");
request.push_str(body);
let reply = raw_exchange(port, &request).await;
let status = reply.get(9..12).and_then(|s| s.parse().ok()).unwrap_or(0);
let body = reply
.split_once("\r\n\r\n")
.map(|(_, body)| body.to_string())
.unwrap_or_default();
(status, body)
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn dashboard_matrix_answers_every_request() {
let token = crate::config::DASHBOARD_ROW_TOKEN;
let right = format!("Bearer {token}");
let ways: [Option<&str>; 3] = [None, Some(&right), Some("Bearer wrong")];
let mut failed: Vec<String> = Vec::new();
fn check(failed: &mut Vec<String>, cell: String, got: u16, want: u16, body: &str) {
if got != want {
failed.push(format!("{cell}: {got}, want {want} {body:?}"));
}
}
for row in crate::config::DASHBOARD_ROWS
.iter()
.filter(|r| r.refusal.is_none())
{
let (control_tx, control_rx) = flume::unbounded();
let (setup, base) = bind_dashboard_with(TestBind::row(row, control_tx)).await;
let parent = spawn_mock_parent(setup.request_rx, Arc::new(AtomicU64::new(0)));
let control = spawn_control_parent(control_rx, answer_ok);
let port = port_of(&base);
let own_host = format!("127.0.0.1:{port}");
let name = row.name;
let health = MATRIX_HEALTH
.iter()
.find(|(r, _)| *r == name)
.expect("health row")
.1;
let (_, body) = raw_call(port, "GET", "/api/v1/health", &own_host, &[], "").await;
let json: serde_json::Value = serde_json::from_str(&body).unwrap_or_default();
for (field, want) in ["control_enabled", "auth_required", "loopback"]
.iter()
.zip(health)
{
if json[field] != want {
failed.push(format!(
"{name} health {field}: {}, want {want}",
json[field]
));
}
}
let statuses = MATRIX_STATUS
.iter()
.find(|(r, _)| *r == name)
.expect("status row")
.1;
for ((method, path, body), want) in MATRIX_REQUESTS.iter().zip(statuses) {
for (way, want) in ways.iter().zip(want) {
let mut headers = vec![("Content-Type", "application/json")];
if let Some(bearer) = way {
headers.push(("Authorization", bearer));
}
let (got, reply) =
raw_call(port, method, path, &own_host, &headers, body).await;
let way = match way {
None => "no token",
Some(b) if *b == right => "right token",
Some(_) => "wrong token",
};
check(
&mut failed,
format!("{name} {method} {path} {way}"),
got,
want,
&reply,
);
}
}
if row.args.contains(&"--dashboard-no-auth")
&& row.args.contains(&"--dashboard-control")
{
let own_origin = format!("http://{own_host}");
for (path, want) in MATRIX_PROBES {
let (method, _, body) = MATRIX_REQUESTS
.iter()
.find(|(_, p, _)| *p == path)
.expect("probed request");
let probes: [(&str, Headers, &str); 4] = [
(
"form",
vec![("Content-Type", "application/x-www-form-urlencoded")],
"a=b",
),
("body-less", vec![], ""),
(
"foreign Origin",
vec![
("Content-Type", "application/json"),
("Origin", "http://evil.example"),
],
body,
),
(
"own Origin",
vec![
("Content-Type", "application/json"),
("Origin", &own_origin),
],
body,
),
];
for ((probe, headers, body), want) in probes.iter().zip(want) {
let (got, reply) =
raw_call(port, method, path, &own_host, headers, body).await;
if want == 403 && !reply.contains("\"reason\":\"cross_origin\"") {
failed.push(format!(
"{name} {method} {path} {probe}: no cross_origin reason {reply:?}"
));
}
check(
&mut failed,
format!("{name} {method} {path} {probe}"),
got,
want,
&reply,
);
}
}
}
let hosts = MATRIX_HOSTS
.iter()
.find(|(r, _)| *r == name)
.expect("host row")
.1;
let rebind = format!("rebind.example:{port}");
let rebind_origin = format!("http://{rebind}");
let configured = format!(
"{}:{port}",
crate::config::dashboard_row_configuration(row).dashboard_host()
);
let ipv6 = format!("[::1]:{port}");
let json = [("Content-Type", "application/json")];
let cells: [(&str, &str, &str, Headers, &str); 4] = [
("GET", "/api/v1/snapshot", &rebind, vec![], ""),
(
"POST",
"/api/v1/control/start",
&rebind,
vec![json[0], ("Origin", &rebind_origin)],
"{}",
),
("GET", "/api/v1/snapshot", &configured, vec![], ""),
("GET", "/api/v1/snapshot", &ipv6, vec![], ""),
];
for ((method, path, host, headers, body), want) in cells.iter().zip(hosts) {
let (got, reply) = raw_call(port, method, path, host, headers, body).await;
if name == "R6" && want == 403 && !reply.contains("\"reason\":\"host_not_allowed\"")
{
failed.push(format!(
"{name} {method} {path} as {host}: no host_not_allowed reason {reply:?}"
));
}
check(
&mut failed,
format!("{name} {method} {path} as {host}"),
got,
want,
&reply,
);
}
parent.abort();
control.abort();
drop(setup.close_tx);
}
assert!(failed.is_empty(), "{:#?}", failed);
}
#[test]
fn dashboard_matrix_startup_lines() {
let read_only_hint = "[dashboard]: read-only; for control add --dashboard-control and either --dashboard-no-auth (no token) or --dashboard-auth-token <token>";
let expected: [(&str, Vec<StartupLine>); 7] = [
(
"R1",
vec![
StartupLine::Info("[dashboard]: listening on http://127.0.0.1:5118 (read-only)".into()),
StartupLine::Info(read_only_hint.into()),
],
),
(
"R2",
vec![
StartupLine::Info("[dashboard]: listening on http://127.0.0.1:5118 (read-only)".into()),
StartupLine::Info("[dashboard]: read-only; for control add --dashboard-control".into()),
],
),
(
"R3",
vec![
StartupLine::Info("[dashboard]: listening on http://127.0.0.1:5118 (control enabled)".into()),
StartupLine::Info("[dashboard]: open http://127.0.0.1:5118/?token=<token> to control the test".into()),
],
),
(
"R4",
vec![
StartupLine::Info("[dashboard]: listening on http://127.0.0.1:5118 (control enabled, no token)".into()),
StartupLine::Warn("[dashboard]: --dashboard-no-auth: anyone on this machine can start, stop and redirect the test".into()),
],
),
(
"R5",
vec![
StartupLine::Info("[dashboard]: listening on http://127.0.0.1:5118 (read-only, no token)".into()),
StartupLine::Info("[dashboard]: read-only; for control add --dashboard-control".into()),
],
),
(
"R11",
vec![
StartupLine::Info("[dashboard]: listening on http://0.0.0.0:5118 (control enabled)".into()),
StartupLine::Info("[dashboard]: open http://127.0.0.1:5118/?token=<token> to control the test".into()),
],
),
(
"R6",
vec![
StartupLine::Info("[dashboard]: listening on http://0.0.0.0:5118 (control enabled, no token)".into()),
StartupLine::Block(
"WARNING: the dashboard on http://0.0.0.0:5118 has no token (--dashboard-no-auth).\n\
Anyone who can reach this port can watch the test, start and stop it,\n\
and point the load at any host.\n\
Safer: bind 127.0.0.1 and use an ssh forward (ssh -L 5118:127.0.0.1:5118 host),\n\
or use --dashboard-auth-token in place of --dashboard-no-auth. Only IP addresses,\n\
localhost and the --dashboard-host name are accepted as Host; set --dashboard-host\n\
to the DNS name you browse to."
.into(),
),
],
),
];
for row in crate::config::DASHBOARD_ROWS
.iter()
.filter(|r| r.refusal.is_none())
{
let configuration = crate::config::dashboard_row_configuration(row);
let host = configuration.dashboard_host();
let bound = format_bind_address(host, 5118);
let lines = startup_lines(
&bound,
configuration.dashboard_control,
configuration.dashboard_no_auth,
configuration.dashboard_auth_token().is_some(),
crate::config::is_loopback_bind(host),
);
let want = &expected
.iter()
.find(|(r, _)| *r == row.name)
.expect("row")
.1;
assert_eq!(&lines, want, "{}", row.name);
for line in &lines {
let text = format!("{line:?}");
assert!(
!text.contains(crate::config::DASHBOARD_ROW_TOKEN),
"{}",
text
);
}
}
assert_eq!(
startup_lines("[::]:5118", true, false, true, false),
vec![
StartupLine::Info(
"[dashboard]: listening on http://[::]:5118 (control enabled)".into()
),
StartupLine::Info(
"[dashboard]: open http://[::1]:5118/?token=<token> to control the test".into()
),
]
);
}
#[test]
fn ip_or_configured_host_rules() {
for (host, configured, allowed) in [
("10.1.2.3:5118", "0.0.0.0", true),
("10.1.2.3", "0.0.0.0", true),
("[fe80::1]:5118", "0.0.0.0", true),
("[::1]", "::", true),
("loadgen.example:5118", "loadgen.example", true),
("LOADGEN.example", "loadgen.example", true),
("rebind.example:5118", "0.0.0.0", false),
("rebind.example:5118", "loadgen.example", false),
("localhost:5118", "0.0.0.0", true),
("localhost.example:5118", "0.0.0.0", false),
("fe80::1", "0.0.0.0", false),
("user@10.1.2.3", "0.0.0.0", false),
("", "0.0.0.0", false),
] {
assert_eq!(
is_ip_or_configured_host(host, configured),
allowed,
"{host} with --dashboard-host {configured}"
);
}
assert!(is_json_content_type("application/json"));
assert!(is_json_content_type("Application/JSON; charset=utf-8"));
assert!(!is_json_content_type("text/plain"));
assert!(!is_json_content_type("application/x-www-form-urlencoded"));
assert!(!is_json_content_type("application/jsonp"));
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn control_requires_auth_and_validates_body() {
let (control_tx, control_rx) = flume::unbounded();
let (setup, base) =
bind_dashboard_with(TestBind::token("s3cret").control(control_tx)).await;
let parent = spawn_mock_parent(setup.request_rx, Arc::new(AtomicU64::new(0)));
let control = spawn_control_parent(control_rx, answer_ok);
drop(setup.close_tx);
let client = reqwest::Client::new();
let health: serde_json::Value = client
.get(format!("{base}/api/v1/health"))
.send()
.await
.expect("health")
.json()
.await
.expect("json");
assert_eq!(health["control_enabled"], true);
let resp = client
.post(format!("{base}/api/v1/control/start"))
.header(header::CONTENT_TYPE, "application/json")
.body("{}")
.send()
.await
.expect("post");
assert_eq!(resp.status(), 401);
let resp = client
.post(format!("{base}/api/v1/control/start"))
.header(header::AUTHORIZATION, "Bearer wrong")
.header(header::CONTENT_TYPE, "application/json")
.body("{}")
.send()
.await
.expect("post");
assert_eq!(resp.status(), 401);
let resp = client
.post(format!("{base}/api/v1/control/start"))
.header(header::AUTHORIZATION, "Bearer s3cret")
.header(header::CONTENT_TYPE, "application/json")
.body("[1]")
.send()
.await
.expect("post");
assert_eq!(resp.status(), 400);
let body: serde_json::Value = resp.json().await.expect("json");
assert_eq!(body["ok"], false);
assert_eq!(body["error"], "bad_request");
assert_eq!(body["command"], "start");
let resp = client
.post(format!("{base}/api/v1/control/users"))
.header(header::AUTHORIZATION, "Bearer s3cret")
.header(header::CONTENT_TYPE, "application/json")
.body("{\"users\":0}")
.send()
.await
.expect("post");
assert_eq!(resp.status(), 400);
let body: serde_json::Value = resp.json().await.expect("json");
assert_eq!(body["error"], "invalid_users");
let resp = client
.post(format!("{base}/api/v1/control/start"))
.header(header::AUTHORIZATION, "Bearer s3cret")
.header(header::CONTENT_TYPE, "application/json")
.body("{}")
.send()
.await
.expect("post");
assert_eq!(resp.status(), 200);
let body: serde_json::Value = resp.json().await.expect("json");
assert_eq!(body["ok"], true);
assert_eq!(body["command"], "start");
assert_eq!(body["phase"], "increase");
let resp = client
.post(format!("{base}/api/v1/control/stop?token=s3cret"))
.header(header::CONTENT_TYPE, "application/json")
.body("{}")
.send()
.await
.expect("post");
assert_eq!(resp.status(), 401);
let resp = client
.post(format!("{base}/api/v1/control/stop?token=s3cret"))
.header(header::AUTHORIZATION, "Bearer s3cret")
.header(header::CONTENT_TYPE, "application/json")
.body("{}")
.send()
.await
.expect("post");
assert_eq!(resp.status(), 401);
let resp = client
.post(format!("{base}/api/v1/control/stop"))
.header(header::AUTHORIZATION, "Bearer s3cret")
.header(header::CONTENT_TYPE, "application/json")
.body("{}")
.send()
.await
.expect("post");
assert_eq!(resp.status(), 200);
let body: serde_json::Value = resp.json().await.expect("json");
assert_eq!(body["ok"], true);
assert_eq!(body["phase"], "decrease");
let resp = client
.post(format!("{base}/api/v1/control/start"))
.header(header::AUTHORIZATION, "Bearer s3cret")
.send()
.await
.expect("post");
assert_eq!(resp.status(), 200);
parent.abort();
control.abort();
}
fn heatmap_export(
generation: u64,
rev: u64,
since: u64,
start_second: u64,
columns: &[(u64, u64)],
buckets: usize,
) -> Heatmap {
Heatmap {
generation,
rev,
since,
start_second,
end_second: start_second + 299,
hold_from: start_second + 298,
buckets_us: vec![[0, 0, 0]; buckets],
columns: columns
.iter()
.map(|&(second, count)| HeatmapColumn {
second,
measured: vec![3, count],
synthetic: Vec::new(),
})
.collect(),
}
}
fn column_counts(heatmap: &Heatmap) -> Vec<(u64, u64)> {
heatmap
.columns
.iter()
.map(|column| (column.second, column.measured[1]))
.collect()
}
#[test]
fn heatmap_store_sends_each_client_what_it_lacks() {
let cursor = |generation, rev| Some(HeatmapCursor { generation, rev });
let mut store = HeatmapStore::new(&heatmap_export(7, 5, 0, 0, &[(1, 10), (2, 4)], 205));
for held in [None, cursor(6, 5), cursor(7, 6)] {
let heatmap = store.since(held);
assert_eq!((heatmap.since, heatmap.rev), (0, 5), "{:?}", held);
assert_eq!(column_counts(&heatmap), [(1, 10), (2, 4)]);
}
assert!(store.since(cursor(7, 5)).columns.is_empty());
store.apply(&heatmap_export(7, 9, 5, 0, &[(2, 6), (3, 1)], 205));
assert_eq!(
store.cursor(),
HeatmapCursor {
generation: 7,
rev: 9
}
);
let caught_up = store.since(cursor(7, 5));
assert_eq!(caught_up.since, 5);
assert_eq!(column_counts(&caught_up), [(2, 6), (3, 1)]);
assert_eq!(
column_counts(&store.since(cursor(7, 2))),
[(1, 10), (2, 6), (3, 1)]
);
assert_eq!(column_counts(&store.since(None)), [(1, 10), (2, 6), (3, 1)]);
store.apply(&heatmap_export(7, 12, 9, 2, &[(301, 2)], 206));
store.apply(&heatmap_export(7, 13, 12, 2, &[], 205));
let whole = store.since(None);
assert_eq!(column_counts(&whole), [(2, 6), (3, 1), (301, 2)]);
assert_eq!(whole.buckets_us.len(), 206);
assert_eq!((whole.start_second, whole.end_second), (2, 301));
store.apply(&heatmap_export(8, 1, 0, 0, &[(0, 1)], 205));
assert_eq!(column_counts(&store.since(cursor(7, 13))), [(0, 1)]);
assert_eq!(store.since(cursor(7, 13)).since, 0);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn series_admission_caps_in_flight() {
let (control_tx, control_rx) = flume::unbounded();
let (setup, base) =
bind_dashboard_with(TestBind::token("s3cret").control(control_tx)).await;
let request_rx = setup.request_rx;
let client = reqwest::Client::new();
let wait_for_queue = |len: usize| {
let request_rx = request_rx.clone();
async move {
let deadline = Instant::now() + Duration::from_secs(5);
while request_rx.len() < len && Instant::now() < deadline {
tokio::time::sleep(Duration::from_millis(10)).await;
}
request_rx.len()
}
};
let mut waiting = Vec::new();
for i in 0..MAX_SERIES_IN_FLIGHT {
let client = client.clone();
let url = format!("{base}/api/v1/series?method=GET&name=/{i}&token=s3cret");
waiting.push(tokio::spawn(async move { client.get(url).send().await }));
}
assert_eq!(
wait_for_queue(MAX_SERIES_IN_FLIGHT).await,
MAX_SERIES_IN_FLIGHT
);
let resp = client
.get(format!(
"{base}/api/v1/series?method=GET&name=/{MAX_SERIES_IN_FLIGHT}&token=s3cret"
))
.send()
.await
.expect("fifth series GET");
assert_eq!(resp.status(), 503);
let body: serde_json::Value = resp.json().await.expect("json");
assert_eq!(body, serde_json::json!({ "error": "busy" }));
assert_eq!(request_rx.len(), MAX_SERIES_IN_FLIGHT);
let start_client = client.clone();
let start_url = format!("{base}/api/v1/control/start");
waiting.push(tokio::spawn(async move {
start_client
.post(start_url)
.header(header::AUTHORIZATION, "Bearer s3cret")
.header(header::CONTENT_TYPE, "application/json")
.body("{}")
.send()
.await
}));
let deadline = Instant::now() + Duration::from_secs(5);
while control_rx.is_empty() && Instant::now() < deadline {
tokio::time::sleep(Duration::from_millis(10)).await;
}
assert_eq!(control_rx.len(), 1);
for task in waiting {
task.abort();
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn control_admission_caps_in_flight() {
let (control_tx, control_rx) = flume::unbounded();
let (_setup, base) =
bind_dashboard_with(TestBind::token("s3cret").control(control_tx)).await;
let client = reqwest::Client::new();
let start = |client: reqwest::Client, url: String| async move {
client
.post(url)
.header(header::AUTHORIZATION, "Bearer s3cret")
.header(header::CONTENT_TYPE, "application/json")
.body("{}")
.send()
.await
};
let url = format!("{base}/api/v1/control/start");
let fill = || -> Vec<_> {
(0..MAX_CONTROL_IN_FLIGHT)
.map(|_| tokio::spawn(start(client.clone(), url.clone())))
.collect()
};
let wait_for_queue = || async {
let deadline = Instant::now() + Duration::from_secs(5);
while control_rx.len() < MAX_CONTROL_IN_FLIGHT && Instant::now() < deadline {
tokio::time::sleep(Duration::from_millis(10)).await;
}
control_rx.len()
};
let waiting = fill();
assert_eq!(wait_for_queue().await, MAX_CONTROL_IN_FLIGHT);
let resp = start(client.clone(), url.clone())
.await
.expect("fifth start POST");
assert_eq!(resp.status(), 503);
let body: serde_json::Value = resp.json().await.expect("json");
assert_eq!(body["ok"], false);
assert_eq!(body["command"], "start");
assert_eq!(body["error"], "busy");
assert_eq!(control_rx.len(), MAX_CONTROL_IN_FLIGHT);
drop(control_rx.drain().collect::<Vec<_>>());
for task in waiting {
let resp = task.await.expect("join").expect("waiting start POST");
assert_eq!(resp.status(), 503);
}
let again = fill();
assert_eq!(
wait_for_queue().await,
MAX_CONTROL_IN_FLIGHT,
"slots must be released"
);
for task in again {
task.abort();
}
}
#[tokio::test]
async fn control_body_limit_is_exact() {
let (control_tx, _control_rx) = flume::unbounded();
let (_setup, base) =
bind_dashboard_with(TestBind::token("s3cret").control(control_tx)).await;
let client = reqwest::Client::new();
let post = |path: &'static str, len: usize| {
client
.post(format!("{base}{path}"))
.header(header::AUTHORIZATION, "Bearer s3cret")
.header(header::CONTENT_TYPE, "application/json")
.body(vec![b'x'; len])
.send()
};
for path in [
"/api/v1/control/start",
"/api/v1/control/stop",
"/api/v1/control/quit",
"/api/v1/control/users",
] {
let resp = post(path, CONTROL_BODY_LIMIT + 1)
.await
.expect("oversized POST");
assert_eq!(resp.status(), 413, "{path}");
}
let resp = post("/api/v1/control/start", CONTROL_BODY_LIMIT)
.await
.expect("limit sized POST");
assert_eq!(resp.status(), 400);
let body: serde_json::Value = resp.json().await.expect("json");
assert_eq!(body["command"], "start");
assert_eq!(body["error"], "bad_request");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn control_oneshot_error_paths() {
let client = reqwest::Client::new();
{
let (control_tx, control_rx) = flume::unbounded();
let (setup, base) =
bind_dashboard_with(TestBind::token("s3cret").control(control_tx)).await;
drop(setup.close_tx);
drop(control_rx);
let resp = client
.post(format!("{base}/api/v1/control/start"))
.header(header::AUTHORIZATION, "Bearer s3cret")
.header(header::CONTENT_TYPE, "application/json")
.body("{}")
.send()
.await
.expect("post");
assert_eq!(resp.status(), 503, "disconnected channel must 503");
let body: serde_json::Value = resp.json().await.expect("json");
assert_eq!(body["ok"], false);
assert_eq!(body["error"], "unavailable");
assert_eq!(body["command"], "start");
}
{
let (control_tx, control_rx) = flume::unbounded();
let (setup, base) =
bind_dashboard_with(TestBind::token("s3cret").control(control_tx)).await;
let parent = spawn_control_parent(control_rx, |_| None);
drop(setup.close_tx);
let resp = client
.post(format!("{base}/api/v1/control/stop"))
.header(header::AUTHORIZATION, "Bearer s3cret")
.header(header::CONTENT_TYPE, "application/json")
.body("{}")
.send()
.await
.expect("post");
assert_eq!(resp.status(), 503, "oneshot drop must 503");
let body: serde_json::Value = resp.json().await.expect("json");
assert_eq!(body["ok"], false);
assert_eq!(body["error"], "unavailable");
assert_eq!(body["command"], "stop");
parent.abort();
}
{
let (control_tx, control_rx) = flume::unbounded();
let (setup, base) =
bind_dashboard_with(TestBind::token("s3cret").control(control_tx)).await;
let parent = spawn_control_parent(control_rx, |_| {
outcome(
false,
Some("internal"),
"internal control error",
"maintain",
1,
None,
)
});
drop(setup.close_tx);
let resp = client
.post(format!("{base}/api/v1/control/users"))
.header(header::AUTHORIZATION, "Bearer s3cret")
.header(header::CONTENT_TYPE, "application/json")
.body(r#"{"users":5}"#)
.send()
.await
.expect("post");
assert_eq!(resp.status(), 200, "internal maps to HTTP 200");
let body: serde_json::Value = resp.json().await.expect("json");
assert_eq!(body["ok"], false);
assert_eq!(body["error"], "internal");
assert_eq!(body["command"], "users");
assert_eq!(body["message"], "internal control error");
parent.abort();
}
{
let (control_tx, control_rx) = flume::unbounded();
let (setup, base) =
bind_dashboard_with(TestBind::token("s3cret").control(control_tx)).await;
let parent = tokio::spawn(async move {
let mut held = Vec::new();
while let Ok(request) = control_rx.recv_async().await {
held.push(request);
}
});
drop(setup.close_tx);
let started = Instant::now();
let resp = client
.post(format!("{base}/api/v1/control/start"))
.header(header::AUTHORIZATION, "Bearer s3cret")
.header(header::CONTENT_TYPE, "application/json")
.body("{}")
.send()
.await
.expect("post");
let elapsed = started.elapsed();
assert_eq!(resp.status(), 503, "timeout must 503");
let body: serde_json::Value = resp.json().await.expect("json");
assert_eq!(body["ok"], false);
assert_eq!(body["error"], "timeout");
assert_eq!(body["command"], "start");
assert!(
body["message"]
.as_str()
.unwrap_or("")
.contains("check phase"),
"timeout message must warn against blind retry: {:?}",
body["message"]
);
assert!(
elapsed >= Duration::from_secs(14),
"expected ~15s timeout, got {:?}",
elapsed
);
assert!(
elapsed < Duration::from_secs(30),
"timeout should not hang forever, got {:?}",
elapsed
);
parent.abort();
}
}
#[test]
fn csp_includes_frame_ancestors_none() {
assert!(
CSP.contains("frame-ancestors 'none'"),
"CSP must block embedding: {}",
CSP
);
}
async fn health_on(stream: &mut tokio::net::TcpStream) -> Option<String> {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let request = "GET /api/v1/health HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n";
stream.write_all(request.as_bytes()).await.ok()?;
let mut buf = Vec::new();
let deadline = Instant::now() + Duration::from_secs(2);
loop {
let text = String::from_utf8_lossy(&buf).into_owned();
if let Some((head, body)) = text.split_once("\r\n\r\n") {
let length = head
.lines()
.find_map(|l| l.strip_prefix("content-length: "))
.and_then(|v| v.trim().parse::<usize>().ok())?;
if body.len() >= length {
return Some(text);
}
}
let mut chunk = [0u8; 1024];
let left = deadline.checked_duration_since(Instant::now())?;
match tokio::time::timeout(left, stream.read(&mut chunk)).await {
Ok(Ok(n)) if n > 0 => buf.extend_from_slice(&chunk[..n]),
_ => return None,
}
}
}
async fn closed_by_peer(stream: &mut tokio::net::TcpStream) -> bool {
closed_within(stream, Duration::from_secs(2)).await
}
async fn closed_within(stream: &mut tokio::net::TcpStream, within: Duration) -> bool {
use tokio::io::AsyncReadExt;
let mut chunk = [0u8; 64];
matches!(
tokio::time::timeout(within, stream.read(&mut chunk)).await,
Ok(Ok(0)) | Ok(Err(_))
)
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn shutdown_closes_listener_and_keep_alive_connections() {
let (setup, _base) = bind_dashboard("").await;
let port = setup.bound_port;
let mut keep_alive = tokio::net::TcpStream::connect(("127.0.0.1", port))
.await
.expect("connect");
let first = health_on(&mut keep_alive).await.expect("health answers");
assert!(first.starts_with("HTTP/1.1 200"), "{}", first);
let started = Instant::now();
tokio::time::timeout(Duration::from_secs(5), setup.server.shutdown())
.await
.expect("shutdown must finish");
assert!(
started.elapsed() < SERVER_DRAIN_TIMEOUT,
"an idle keep-alive connection closes at once, without the drain deadline"
);
assert!(
tokio::net::TcpStream::connect(("127.0.0.1", port))
.await
.is_err(),
"no new connection may be accepted after shutdown"
);
assert_eq!(
health_on(&mut keep_alive).await,
None,
"keep-alive connection must not be served after shutdown"
);
drop(
tokio::net::TcpListener::bind(("127.0.0.1", port))
.await
.expect("port must be free when shutdown returns"),
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn shutdown_drain_is_bounded() {
use tokio::io::AsyncWriteExt;
let (control_tx, control_rx) = flume::unbounded();
let (setup, _base) =
bind_dashboard_with(TestBind::token("s3cret").control(control_tx)).await;
let port = setup.bound_port;
let mut stalled = tokio::net::TcpStream::connect(("127.0.0.1", port))
.await
.expect("connect");
stalled
.write_all(
b"POST /api/v1/control/users HTTP/1.1\r\nHost: 127.0.0.1\r\n\
Authorization: Bearer s3cret\r\nContent-Type: application/json\r\n\
Content-Length: 11\r\n\r\n{\"users\":5}",
)
.await
.expect("write control request");
let _in_flight = tokio::time::timeout(Duration::from_secs(5), control_rx.recv_async())
.await
.expect("control request reaches the parent")
.expect("request channel open");
let started = Instant::now();
let shutdown = tokio::spawn(setup.server.shutdown());
loop {
if tokio::net::TcpStream::connect(("127.0.0.1", port))
.await
.is_err()
{
break;
}
assert!(
started.elapsed() < SERVER_DRAIN_TIMEOUT / 2,
"listener must close before the drain, not after it"
);
tokio::time::sleep(Duration::from_millis(10)).await;
}
tokio::time::timeout(SERVER_DRAIN_TIMEOUT * 3, shutdown)
.await
.expect("shutdown must not wait on a stalled request")
.expect("shutdown task");
assert!(
started.elapsed() >= SERVER_DRAIN_TIMEOUT,
"a busy connection gets the drain deadline before it is closed"
);
assert!(
closed_by_peer(&mut stalled).await,
"stalled connection must be closed"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn dropping_server_stops_it() {
let (setup, _base) = bind_dashboard("").await;
let port = setup.bound_port;
let mut keep_alive = tokio::net::TcpStream::connect(("127.0.0.1", port))
.await
.expect("connect");
assert!(health_on(&mut keep_alive).await.is_some());
drop(setup.server);
assert!(
closed_by_peer(&mut keep_alive).await,
"keep-alive connection must be closed"
);
let mut refused = false;
for _ in 0..40 {
if tokio::net::TcpStream::connect(("127.0.0.1", port))
.await
.is_err()
{
refused = true;
break;
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
assert!(refused, "listener must close after the handle is dropped");
}
async fn raw_exchange(port: u16, request: &str) -> String {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let mut stream = tokio::net::TcpStream::connect(("127.0.0.1", port))
.await
.expect("connect");
stream.write_all(request.as_bytes()).await.expect("write");
let mut buf = Vec::new();
let _ = tokio::time::timeout(Duration::from_secs(2), stream.read_to_end(&mut buf)).await;
String::from_utf8_lossy(&buf).into_owned()
}
fn port_of(base: &str) -> u16 {
base.rsplit(':').next().unwrap().parse().unwrap()
}
fn get_request(path: &str, host: &str) -> String {
format!("GET {path} HTTP/1.1\r\nHost: {host}\r\nConnection: close\r\n\r\n")
}
#[test]
fn is_loopback_host_rules() {
for host in [
"localhost",
"LOCALHOST:5118",
"127.0.0.1",
"127.0.0.1:5118",
"127.1.2.3:80",
"[::1]",
"[::1]:5118",
] {
assert!(is_loopback_host(host), "{} is loopback", host);
}
for host in [
"",
"evil.example",
"evil.example:5118",
"localhost.evil.example",
"127.0.0.1.evil.example",
"0.0.0.0:5118",
"10.0.0.1:5118",
"[::]:5118",
"evil.example@127.0.0.1",
"127.0.0.1:5118/path",
] {
assert!(!is_loopback_host(host), "{:?} is not loopback", host);
}
}
#[test]
fn connection_cap_is_above_sse_cap() {
assert_eq!(connection_cap(MAX_SSE_CLIENTS), 2 * MAX_SSE_CLIENTS + 16);
assert_eq!(connection_cap(1), 18);
assert_eq!(
connection_cap(usize::MAX),
tokio::sync::Semaphore::MAX_PERMITS
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn loopback_bind_refuses_non_loopback_host() {
let (_hub_builds, _parent_builds, parent, base, _server) = setup_with_mock("").await;
let port = port_of(&base);
for path in ["/", "/api/v1/health", "/api/v1/snapshot", "/api/v1/events"] {
let reply =
raw_exchange(port, &get_request(path, &format!("evil.example:{port}"))).await;
assert!(
reply.starts_with("HTTP/1.1 403"),
"{} with a rebound Host must get 403, got {:?}",
path,
reply
);
}
let reply = raw_exchange(port, "GET /api/v1/snapshot HTTP/1.0\r\n\r\n").await;
assert!(
reply.starts_with("HTTP/1.0 403"),
"a request with no Host must get 403, got {:?}",
reply
);
for host in [
format!("127.0.0.1:{port}"),
format!("localhost:{port}"),
format!("[::1]:{port}"),
] {
let reply = raw_exchange(port, &get_request("/api/v1/snapshot", &host)).await;
assert!(
reply.starts_with("HTTP/1.1 200"),
"Host {} must be served, got {:?}",
host,
reply
);
}
parent.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn non_loopback_bind_serves_any_host() {
for bind_host in ["0.0.0.0", "127.1"] {
let config = GooseConfiguration {
dashboard: true,
dashboard_host: Some(bind_host.to_string()),
dashboard_port: Some(0),
dashboard_auth_token: Some("secret".to_string()),
..Default::default()
};
let setup = setup_dashboard(&config, None, ListeningControllers::configured(&config))
.await
.expect("bind")
.expect("enabled");
let reply = raw_exchange(
setup.bound_port,
&get_request("/api/v1/health", "loadgen.example:5118"),
)
.await;
assert!(
reply.starts_with("HTTP/1.1 200"),
"a bind to {} must not check Host, got {:?}",
bind_host,
reply
);
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn responses_carry_security_headers() {
let (_hub_builds, _parent_builds, parent, base, _server) = setup_with_mock("secret").await;
let port = port_of(&base);
let client = reqwest::Client::new();
let mut responses = Vec::new();
for (path, status) in [
("/?token=secret", 200),
("/static/app.js", 200),
("/api/v1/health", 200),
("/api/v1/snapshot?token=secret", 200),
("/api/v1/snapshot", 401),
("/api/v1/events?token=secret", 200),
("/no/such/page", 404),
] {
let resp = client
.get(format!("{base}{path}"))
.send()
.await
.expect("request");
assert_eq!(resp.status(), status, "{path}");
responses.push((path.to_string(), resp.headers().clone()));
}
let refused = raw_exchange(port, &get_request("/", "evil.example")).await;
assert!(refused.starts_with("HTTP/1.1 403"), "{:?}", refused);
for (path, headers) in &responses {
assert_eq!(headers["cache-control"], "no-store", "{path}");
assert_eq!(headers["x-content-type-options"], "nosniff", "{path}");
assert_eq!(headers["referrer-policy"], "no-referrer", "{path}");
}
for line in [
"cache-control: no-store\r\n",
"x-content-type-options: nosniff\r\n",
"referrer-policy: no-referrer\r\n",
] {
assert!(
refused.contains(line),
"403 must carry {:?}: {:?}",
line,
refused
);
}
parent.abort();
}
async fn read_until_contains(stream: &mut tokio::net::TcpStream, needle: &str) -> bool {
use tokio::io::AsyncReadExt;
let mut text = String::new();
let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
let mut chunk = [0u8; 4096];
while !text.contains(needle) {
match tokio::time::timeout_at(deadline, stream.read(&mut chunk)).await {
Ok(Ok(n)) if n > 0 => text.push_str(&String::from_utf8_lossy(&chunk[..n])),
_ => return false,
}
}
true
}
async fn open_sse(port: u16) -> tokio::net::TcpStream {
use tokio::io::AsyncWriteExt;
let mut stream = tokio::net::TcpStream::connect(("127.0.0.1", port))
.await
.expect("connect");
stream
.write_all(b"GET /api/v1/events HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n")
.await
.expect("write");
assert!(
read_until_contains(&mut stream, "event: snapshot").await,
"SSE stream must get a first snapshot"
);
stream
}
async fn sse_still_live(stream: &mut tokio::net::TcpStream) -> bool {
use tokio::io::AsyncReadExt;
let mut chunk = [0u8; 4096];
while let Ok(read) =
tokio::time::timeout(Duration::from_millis(300), stream.read(&mut chunk)).await
{
if !matches!(read, Ok(n) if n > 0) {
return false;
}
}
read_until_contains(stream, "event: snapshot").await
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn slow_header_client_is_cut_at_timeout() {
use tokio::io::AsyncWriteExt;
let (_hub_builds, _parent_builds, parent, base, _server) = setup_with_mock("").await;
let port = port_of(&base);
let mut sse = open_sse(port).await;
let started = Instant::now();
let mut silent = tokio::net::TcpStream::connect(("127.0.0.1", port))
.await
.expect("connect");
let mut slow = tokio::net::TcpStream::connect(("127.0.0.1", port))
.await
.expect("connect");
slow.write_all(b"GET /api/v1/health HTTP/1.1\r\nHost: 127")
.await
.expect("write");
assert!(
!closed_within(&mut slow, HEADER_READ_TIMEOUT - Duration::from_secs(1)).await,
"closed before the header read timeout"
);
assert!(
closed_within(&mut slow, Duration::from_secs(5)).await,
"a stalled header must be cut at the header read timeout"
);
assert!(
started.elapsed() >= HEADER_READ_TIMEOUT,
"cut after {:?}",
started.elapsed()
);
assert!(
closed_within(&mut silent, Duration::from_secs(1)).await,
"a connection that sends nothing must be cut at the header read timeout"
);
assert!(
sse_still_live(&mut sse).await,
"an SSE stream must outlive the header read timeout"
);
parent.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn connection_cap_refuses_extra_connections_and_keeps_sse() {
let max_clients = 2u32;
let cap = connection_cap(max_clients as usize);
let (_hub_builds, _parent_builds, parent, base, _server) =
setup_with_mock_max("", max_clients).await;
let port = port_of(&base);
let mut sse = Vec::new();
for _ in 0..max_clients {
sse.push(open_sse(port).await);
}
let mut held = Vec::new();
let mut after_refusal = false;
loop {
assert!(
sse.len() + held.len() <= cap,
"more than {} connections served",
cap
);
if after_refusal {
tokio::time::sleep(Duration::from_millis(300)).await;
}
let mut stream = tokio::net::TcpStream::connect(("127.0.0.1", port))
.await
.expect("connect");
match health_on(&mut stream).await {
Some(_) => {
held.push(stream);
after_refusal = false;
}
None if after_refusal => break,
None => after_refusal = true,
}
}
assert_eq!(
sse.len() + held.len(),
cap,
"the cap is reached at exactly {cap} open connections"
);
for (i, stream) in sse.iter_mut().enumerate() {
assert!(
sse_still_live(stream).await,
"SSE client {} must keep receiving snapshots at the connection cap",
i
);
}
drop(held.pop());
let mut served = false;
for _ in 0..40 {
let mut stream = tokio::net::TcpStream::connect(("127.0.0.1", port))
.await
.expect("connect");
if health_on(&mut stream).await.is_some() {
served = true;
break;
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
assert!(served, "a closed connection must free its slot");
parent.abort();
}
async fn wait_active_sse(base: &str, want: u64, within: Duration) -> u64 {
let client = reqwest::Client::new();
let deadline = Instant::now() + within;
loop {
let health: serde_json::Value = client
.get(format!("{base}/api/v1/health"))
.send()
.await
.expect("health")
.json()
.await
.expect("health json");
let active = health["active_sse_clients"].as_u64().expect("count");
if active == want || Instant::now() >= deadline {
return active;
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn sse_slot_freed_on_disconnect_without_snapshots() {
let cap = 2u32;
let (_setup, base) = bind_dashboard_with(TestBind::token("").max_clients(cap)).await;
let client = reqwest::Client::new();
let mut held = Vec::new();
for i in 0..cap {
let resp = client
.get(format!("{base}/api/v1/events"))
.send()
.await
.expect("connect");
assert_eq!(resp.status(), 200, "client {i} should be accepted");
held.push(resp);
}
drop(held);
assert_eq!(
wait_active_sse(&base, 0, Duration::from_secs(2)).await,
0,
"disconnected clients must free their slots"
);
}
}