|
|
@@ -10,7 +10,6 @@ use std::sync::{Arc, Mutex};
|
|
|
|
|
|
use tokio::task::JoinHandle;
|
|
|
|
|
|
-use crate::config::Config;
|
|
|
use crate::ssh::shq;
|
|
|
use crate::web::events::Event;
|
|
|
use crate::web::ssh_stream::stream_lines;
|
|
|
@@ -89,14 +88,14 @@ pub(crate) fn push_line(ring: &Mutex<VecDeque<String>>, line: &str) {
|
|
|
/// Idempotent per (cluster, node, recipe, rank): a live stream for the same
|
|
|
/// key is reused; a finished one is replaced.
|
|
|
pub fn attach(
|
|
|
- config: &Config,
|
|
|
- bus: &tokio::sync::broadcast::Sender<Event>,
|
|
|
+ app: &crate::web::AppState,
|
|
|
attaches: &Mutex<Vec<LogAttach>>,
|
|
|
cluster: &str,
|
|
|
node: &str,
|
|
|
recipe: &str,
|
|
|
rank: u32,
|
|
|
) -> Result<(String, Vec<String>), String> {
|
|
|
+ let config = &app.config;
|
|
|
let cl = config
|
|
|
.clusters
|
|
|
.iter()
|
|
|
@@ -136,13 +135,14 @@ pub fn attach(
|
|
|
let done = Arc::new(AtomicBool::new(false));
|
|
|
let cmd = logs_stream_cmd(recipe, rank, TAIL_LINES);
|
|
|
let (h, s, d) = (history.clone(), stop.clone(), done.clone());
|
|
|
- let (cfg, node_obj, bus) = (config.clone(), node_cfg.clone(), bus.clone());
|
|
|
+ let (cfg, node_obj, bus) = (config.clone(), node_cfg.clone(), app.bus.clone());
|
|
|
let (idc, cluster_s, node_s, recipe_s) = (
|
|
|
id.clone(),
|
|
|
cl.name.clone(),
|
|
|
node_cfg.name.clone(),
|
|
|
recipe.to_string(),
|
|
|
);
|
|
|
+ let appc = app.clone();
|
|
|
let task = tokio::spawn(async move {
|
|
|
let _ = stream_lines(&cfg, &node_obj, cmd, s, |line| {
|
|
|
push_line(&h, line);
|
|
|
@@ -156,8 +156,11 @@ pub fn attach(
|
|
|
});
|
|
|
})
|
|
|
.await;
|
|
|
- // Stream errors (node offline, container gone) just end the window.
|
|
|
+ // Stream errors (node offline, container gone) just end the window —
|
|
|
+ // refresh the board so attaches[].done (and fleet changes) reach the
|
|
|
+ // UI immediately instead of on the next 30 s tick.
|
|
|
d.store(true, Ordering::Relaxed);
|
|
|
+ crate::web::push_state(&appc).await;
|
|
|
});
|
|
|
|
|
|
let attach = LogAttach {
|
|
|
@@ -214,7 +217,6 @@ pub fn by_key(
|
|
|
#[cfg(test)]
|
|
|
mod tests {
|
|
|
use super::*;
|
|
|
- use crate::web::events::Event;
|
|
|
|
|
|
fn ring() -> Arc<Mutex<VecDeque<String>>> {
|
|
|
Arc::new(Mutex::new(VecDeque::new()))
|
|
|
@@ -267,26 +269,25 @@ mod tests {
|
|
|
|
|
|
#[tokio::test]
|
|
|
async fn attach_is_idempotent_and_bounded() {
|
|
|
- let (tx, _rx) = tokio::sync::broadcast::channel::<Event>(8);
|
|
|
+ let app = crate::web::tests::app_state(None);
|
|
|
let attaches: Mutex<Vec<LogAttach>> = Mutex::new(Vec::new());
|
|
|
- let cfg = crate::web::tests::fake_config();
|
|
|
// 8 attaches fill the cap; the 9th is refused.
|
|
|
for i in 0..MAX_ATTACHES {
|
|
|
- let (id, _) = attach(&cfg, &tx, &attaches, "fake", "n0", &format!("r{i}"), 0).unwrap();
|
|
|
+ let (id, _) = attach(&app, &attaches, "fake", "n0", &format!("r{i}"), 0).unwrap();
|
|
|
assert_eq!(id, format!("l{}", i + 1));
|
|
|
}
|
|
|
- let err = attach(&cfg, &tx, &attaches, "fake", "n0", "overflow", 0).unwrap_err();
|
|
|
+ let err = attach(&app, &attaches, "fake", "n0", "overflow", 0).unwrap_err();
|
|
|
assert!(err.contains("max"), "{err}");
|
|
|
// Unknown cluster / node are 400-class errors.
|
|
|
- assert!(attach(&cfg, &tx, &attaches, "nope", "n0", "r", 0).is_err());
|
|
|
- assert!(attach(&cfg, &tx, &attaches, "fake", "nope", "r", 0).is_err());
|
|
|
+ assert!(attach(&app, &attaches, "nope", "n0", "r", 0).is_err());
|
|
|
+ assert!(attach(&app, &attaches, "fake", "nope", "r", 0).is_err());
|
|
|
// by_key finds what exists, misses what doesn't.
|
|
|
assert!(by_key(&attaches, "fake", "n0", "r0", 0, 10).is_some());
|
|
|
assert!(by_key(&attaches, "fake", "n0", "r0", 7, 10).is_none());
|
|
|
// Detach removes + frees a slot; re-attach after detach works.
|
|
|
assert!(detach(&attaches, "l1"));
|
|
|
assert!(!detach(&attaches, "l1"));
|
|
|
- attach(&cfg, &tx, &attaches, "fake", "n0", "new", 0).unwrap();
|
|
|
+ attach(&app, &attaches, "fake", "n0", "new", 0).unwrap();
|
|
|
assert_eq!(attaches.lock().unwrap().len(), MAX_ATTACHES);
|
|
|
}
|
|
|
|
|
|
@@ -310,12 +311,11 @@ mod tests {
|
|
|
|
|
|
#[tokio::test]
|
|
|
async fn attach_reuses_live_stream_for_same_key() {
|
|
|
- let (tx, _rx) = tokio::sync::broadcast::channel::<Event>(8);
|
|
|
+ let app = crate::web::tests::app_state(None);
|
|
|
let list: Mutex<Vec<LogAttach>> = Mutex::new(Vec::new());
|
|
|
list.lock().unwrap().push(fake_attach("l1", false));
|
|
|
- let cfg = crate::web::tests::fake_config();
|
|
|
// Same key, live stream -> the existing attach is reused, no new task.
|
|
|
- let (id, history) = attach(&cfg, &tx, &list, "fake", "n0", "r", 0).unwrap();
|
|
|
+ let (id, history) = attach(&app, &list, "fake", "n0", "r", 0).unwrap();
|
|
|
assert_eq!(id, "l1");
|
|
|
assert!(history.is_empty());
|
|
|
assert_eq!(list.lock().unwrap().len(), 1, "no second attach spawned");
|
|
|
@@ -323,12 +323,11 @@ mod tests {
|
|
|
|
|
|
#[tokio::test]
|
|
|
async fn attach_replaces_finished_stream_for_same_key() {
|
|
|
- let (tx, _rx) = tokio::sync::broadcast::channel::<Event>(8);
|
|
|
+ let app = crate::web::tests::app_state(None);
|
|
|
let list: Mutex<Vec<LogAttach>> = Mutex::new(Vec::new());
|
|
|
list.lock().unwrap().push(fake_attach("l1", true));
|
|
|
- let cfg = crate::web::tests::fake_config();
|
|
|
// Same key, finished stream -> dropped and replaced with a new task.
|
|
|
- attach(&cfg, &tx, &list, "fake", "n0", "r", 0).unwrap();
|
|
|
+ attach(&app, &list, "fake", "n0", "r", 0).unwrap();
|
|
|
let g = list.lock().unwrap();
|
|
|
assert_eq!(g.len(), 1, "old attach dropped");
|
|
|
assert_eq!(g[0].id, "l1", "slot id may be reused");
|