diff --git a/api/src/lib.rs b/api/src/lib.rs index 528640c7..8c11b4c3 100644 --- a/api/src/lib.rs +++ b/api/src/lib.rs @@ -189,6 +189,7 @@ pub trait SushApi { ctx: RequestContext, headers: Header, params: PathParams, + query: QueryParams, ) -> Result, HttpError>; /// Get (a subset of) the standard output or standard error of a job. @@ -387,6 +388,8 @@ impl JobWait { pub struct JobStopParams { /// Wait for the job process to end. pub wait: JobWait, + /// Where a proxy should route this request. Sleds ignore it. + pub via: Option, } /// Simple pagination for history list. diff --git a/client/src/cli.rs b/client/src/cli.rs index 13588734..59e95642 100644 --- a/client/src/cli.rs +++ b/client/src/cli.rs @@ -27,7 +27,7 @@ use sush_common::jobs::{ SignedJob, job_status_to_json_map, }; use sush_common::keys::{KeyId, Signature, SshPublicKey}; -use sush_common::targets::{MAX_CUBBY, SledVersion}; +use sush_common::targets::{MAX_CUBBY, SledId, SledVersion}; use sush_common::version::VersionInfo; use crate::AuthzSigner; @@ -463,6 +463,15 @@ impl CommandContext for Cli { } } + fn job_watch_stalled(&mut self, job_id: &JobId) { + let guard = self.watch.lock().unwrap(); + if let Some(watch) = guard.as_ref() { + let _ = watch.multi.println(format!( + "❗ No sled has reported a status for job `{job_id}`" + )); + } + } + fn job_watch_finished(&mut self, _job_id: &JobId) { if let Some(watch) = self.watch.lock().unwrap().take() { for bar in watch.bars.values() { @@ -757,6 +766,21 @@ impl CommandContext for Cli { Ok(()) } + fn really_target(&mut self, sled: &SledId) -> Result<(), CommandError> { + match self.get_output_format() { + OutputFormat::Json => Ok(()), + OutputFormat::Text => { + let prompt = + format!("❓ Sled `{sled}` is not in the rack inventory. Proceed (yes/no)? "); + if read_bool(&prompt)? { + Ok(()) + } else { + Err(CommandError::Canceled) + } + } + } + } + fn really_revoke(&mut self, what: &str, key_id: KeyId) -> Result { match self.get_output_format() { OutputFormat::Json => Ok(key_id), diff --git a/client/src/commands.rs b/client/src/commands.rs index 8f04422b..2f962aea 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -1169,6 +1169,9 @@ async fn job( return Err(CommandError::io(path, ErrorKind::AlreadyExists.into())); } } + if let Some(client) = client.as_ref() { + preflight_target(ctx, client, &target).await?; + } let streaming = if *streaming { Streaming::Output } else { @@ -1211,7 +1214,7 @@ async fn job( } (JobCommand::Stop { job_id }, Some(client)) => { - job_stop(ctx, client, &job_id).await?; + job_stop(ctx, client, &job_id, None).await?; ctx.job_stopped(&job_id); Ok(()) } @@ -1403,6 +1406,7 @@ async fn job_start( let mut last = JobStatusMap::new(); let mut settling = Settling::default(); let mut started = false; + let mut stalled = false; let mut stopped = false; let mut sigint = signal(SignalKind::interrupt())?; let status = loop { @@ -1421,7 +1425,7 @@ async fn job_start( } _ = ticker.tick() => { - last = match job_status_map(ctx, client, &job_id).await { + last = match job_status_map(ctx, client, &job_id, job_target.single_baseboard()).await { Ok(status) => status, // The job may not be visible anywhere yet. Err(CommandError::NotFound(_)) => JobStatusMap::new(), @@ -1435,6 +1439,10 @@ async fn job_start( if settling.done(&job_target, rack, &last) && started { break last; } + if !stalled && last.is_empty() && settling.polls >= WATCH_STALL_POLLS { + ctx.job_watch_stalled(&job_id); + stalled = true; + } } // While the job runs, an interrupt stops it but keeps @@ -1448,7 +1456,7 @@ async fn job_start( break last; } for _ in 0..3 { - match job_stop(ctx, client, &job_id).await { + match job_stop(ctx, client, &job_id, job_target.single_baseboard()).await { Ok(_) => { ctx.job_stopped(&job_id); stopped = true; @@ -1518,14 +1526,14 @@ async fn job_stop( ctx: &mut impl CommandContext, client: &Client, job_id: &JobId, + via: Option<&BaseboardId>, ) -> Result<(), CommandError> { - with_login(ctx, client, async || { - client - .job_stop() - .job_id(job_id) - .wait(JobWait::Stop) - .send() - .await + with_login_via(ctx, client, via, async || { + let mut request = client.job_stop().job_id(job_id).wait(JobWait::Stop); + if let Some(via) = via { + request = request.via(via.to_string()); + } + request.send().await }) .await?; Ok(()) @@ -1537,19 +1545,26 @@ async fn job_status( job_id: &JobId, style: StatusDisplayStyle, ) -> Result<(), CommandError> { - let status = job_status_map(ctx, client, job_id).await?; + let status = job_status_map(ctx, client, job_id, None).await?; ctx.job_status(job_id, &status, style); Ok(()) } -/// Fetch a job's rack-wide status map. +/// Fetch a job's rack-wide status map. Routing `via` a single-sled +/// target gets its authoritative status and keeps the login on the +/// sled that already knows it. async fn job_status_map( ctx: &mut impl CommandContext, client: &Client, job_id: &JobId, + via: Option<&BaseboardId>, ) -> Result { - let status = with_login(ctx, client, async || { - client.job_status().job_id(job_id).send().await + let status = with_login_via(ctx, client, via, async || { + let mut request = client.job_status().job_id(job_id); + if let Some(via) = via { + request = request.via(via.to_string()); + } + request.send().await }) .await? .into_inner(); @@ -1571,7 +1586,7 @@ async fn job_watch( let mut settling = Settling::default(); let mut sigint = signal(SignalKind::interrupt())?; let status = loop { - let status = match job_status_map(ctx, client, job_id).await { + let status = match job_status_map(ctx, client, job_id, target.single_baseboard()).await { Ok(status) => status, Err(error) => { ctx.job_watch_finished(job_id); @@ -1597,6 +1612,10 @@ async fn job_watch( /// is likely still missing sleds. const WATCH_MIN_POLLS: usize = 5; +/// How many polls a watch may go without any sled reporting a status +/// before warning that the job may never run. +const WATCH_STALL_POLLS: usize = 15; + /// Rolling settlement state for a watched job. #[derive(Default)] struct Settling { @@ -1707,7 +1726,7 @@ async fn job_output( } // Fetch output from every sled with a recorded status. - let status = job_status_map(ctx, client, &args.job_id).await?; + let status = job_status_map(ctx, client, &args.job_id, None).await?; if status.is_empty() { return Err(CommandError::NotFound(format!( "Job `{}` not found", @@ -1738,14 +1757,7 @@ async fn job_output_from( }: JobOutput, ) -> Result<(), CommandError> { // Fetch job status for output length and hash. - let status = job_status_try_from_json_map( - with_login_via(ctx, client, Some(target), async || { - client.job_status().job_id(job_id).send().await - }) - .await? - .into_inner(), - ) - .map_err(CommandError::BaseboardIdParseError)?; + let status = job_status_map(ctx, client, &job_id, Some(target)).await?; let JobOutputState { stdout_len, @@ -2081,6 +2093,32 @@ impl FromStr for TargetArg { } } +/// Confirm before signing a job for a sled that doesn't appear in inventory. +#[cfg(feature = "permslip")] +async fn preflight_target( + ctx: &mut impl CommandContext, + client: &Client, + target: &Target, +) -> Result<(), CommandError> { + let Target::Sleds(sleds) = target else { + return Ok(()); + }; + let Ok(inventory) = client.versions().send().await else { + return Ok(()); + }; + let inventory = inventory.into_inner(); + for sled in sleds { + let known = match sled { + SledId::Baseboard(baseboard) => inventory.iter().any(|s| &s.baseboard == baseboard), + SledId::Cubby(cubby) => inventory.iter().any(|s| s.cubby == Some(*cubby)), + }; + if !known { + ctx.really_target(sled)?; + } + } + Ok(()) +} + /// Resolve a target argument to a target, matching bare serial /// numbers against the rack's sled inventory. async fn resolve_target_arg(client: &Client, target: &TargetArg) -> Result { @@ -2120,7 +2158,7 @@ async fn resolve_serial( job_id: &JobId, serial: &str, ) -> Result { - let status = job_status_map(ctx, client, job_id).await?; + let status = job_status_map(ctx, client, job_id, None).await?; let mut matches = status .keys() .filter(|b| b.serial_number.eq_ignore_ascii_case(serial)); diff --git a/client/src/context.rs b/client/src/context.rs index 62a7010c..c6562016 100644 --- a/client/src/context.rs +++ b/client/src/context.rs @@ -18,7 +18,7 @@ use sush_common::jobs::{ Access, JobId, JobOutputStream, JobStatusMap, Session, SessionId, SignedJob, }; use sush_common::keys::{KeyId, SshPublicKey}; -use sush_common::targets::SledVersion; +use sush_common::targets::{SledId, SledVersion}; use sush_common::version::VersionInfo; use crate::AuthzSigner; @@ -130,6 +130,7 @@ pub trait CommandContext: Clone + Send + Sync { fn cert_imported(&mut self, path: &Path, key_id: KeyId) -> Result<(), CommandError>; // Job management + fn really_target(&mut self, sled: &SledId) -> Result<(), CommandError>; fn job_started(&mut self, job: &SignedJob); fn job_stopped(&mut self, id: &JobId); fn job_error(&mut self, error: CommandError) -> CommandError; @@ -146,6 +147,7 @@ pub trait CommandContext: Clone + Send + Sync { fn job_output_finished(&mut self, id: &JobId, stream: JobOutputStream, stage: Option<&str>); fn job_watch_started(&mut self, id: &JobId); fn job_watch_update(&mut self, status: &JobStatusMap); + fn job_watch_stalled(&mut self, id: &JobId); fn job_watch_finished(&mut self, id: &JobId); fn job_attached(&mut self, id: &JobId); fn job_detached(&mut self, id: &JobId); diff --git a/client/src/repl.rs b/client/src/repl.rs index 29cd7c22..c86229d0 100644 --- a/client/src/repl.rs +++ b/client/src/repl.rs @@ -23,7 +23,7 @@ use sush_common::jobs::{ Access, JobId, JobOutputStream, JobStatusMap, Session, SessionId, SignedJob, }; use sush_common::keys::{KeyId, SshPublicKey}; -use sush_common::targets::SledVersion; +use sush_common::targets::{SledId, SledVersion}; use sush_common::version::VersionInfo; use crate::cli::Cli; @@ -338,6 +338,10 @@ impl CommandContext for Repl { self.cli.job_watch_update(status) } + fn job_watch_stalled(&mut self, job_id: &JobId) { + self.cli.job_watch_stalled(job_id) + } + fn job_watch_finished(&mut self, job_id: &JobId) { self.cli.job_watch_finished(job_id) } @@ -365,6 +369,10 @@ impl CommandContext for Repl { self.cli.please_touch(identity) } + fn really_target(&mut self, sled: &SledId) -> Result<(), CommandError> { + self.cli.really_target(sled) + } + fn really_revoke(&mut self, what: &str, key_id: KeyId) -> Result { self.cli.really_revoke(what, key_id) } diff --git a/justfile b/justfile index 862a860d..616d73dc 100644 --- a/justfile +++ b/justfile @@ -4,7 +4,7 @@ check: cargo check --workspace --all-targets lint: - cargo fmt --check && cargo clippy --tests + cargo fmt --check && cargo clippy --tests -- --no-deps --deny warnings test *FILTER: cargo nextest run --workspace {{FILTER}} diff --git a/server/src/executor.rs b/server/src/executor.rs index b8b8b849..2bf2a5f7 100644 --- a/server/src/executor.rs +++ b/server/src/executor.rs @@ -18,7 +18,7 @@ use std::sync::{Arc, RwLock}; use chrono::Utc; use futures::Stream; use pwd::Passwd; -use rustix::io::close; +use rustix::io::{Errno, close}; use rustix::process::{Pid, Signal, ioctl_tiocsctty, kill_process_group, setsid}; use slog::{Logger, debug, error, o, warn}; use tokio::fs::{DirBuilder, OpenOptions}; @@ -488,20 +488,24 @@ async fn send_error( } } -/// Kill a job's whole process group, of which `child` should be the leader. -/// -/// TODO: 2-stage stop with `SIGTERM` and a grace period. -pub fn kill_job(log: &Logger, child: &Child) { - if let Some(pid) = child.id() - && let Ok(pid) = pid.try_into() - && let Some(pid) = Pid::from_raw(pid) - { - match kill_process_group(pid, Signal::KILL) { - Ok(()) => debug!(log, "killed job processes"), - Err(error) => error!(log, "unable to kill job"; "error" => %error), +/// A job's process group ID, which its `child` leads. Capture it +/// before the leader is reaped, after which `id` returns nothing. +pub fn job_pgid(child: &Child) -> Option { + Pid::from_raw(child.id()?.try_into().ok()?) +} + +/// Send a signal to a job's whole process group. +pub fn kill_job(log: &Logger, pgid: Option, signal: Signal) { + let Some(pgid) = pgid else { + debug!(log, "job has no process group"); + return; + }; + match kill_process_group(pgid, signal) { + Ok(()) => debug!(log, "signalled job processes"; "signal" => ?signal), + Err(Errno::SRCH) => debug!(log, "job processes are already dead"), + Err(error) => { + error!(log, "unable to signal job"; "signal" => ?signal, "error" => %error) } - } else { - debug!(log, "process is already dead or has an invalid PID"); } } diff --git a/server/src/io.rs b/server/src/io.rs index 610def50..e98dcc1e 100644 --- a/server/src/io.rs +++ b/server/src/io.rs @@ -169,8 +169,8 @@ impl JobIo { /// How long to keep reading output after the child dies. /// - /// For a pty there is no portable EOF signal; a short quiet period - /// is the only way to know we've drained (OpenSSH does this too). + /// For a pty there is no portable EOF signal; a short window + /// after death is the best we can do (OpenSSH does this too). /// Pipes deliver EOF once every writer exits, so this is only a /// backstop against a descendant that escaped the process group /// while holding the inherited pipe open. diff --git a/server/src/job.rs b/server/src/job.rs index 9873f785..99a5fa4c 100644 --- a/server/src/job.rs +++ b/server/src/job.rs @@ -29,13 +29,14 @@ use blake3::Hasher; use bytes::{Bytes, BytesMut}; use dropshot::WebsocketConnectionRaw; use futures::{SinkExt as _, StreamExt as _}; +use rustix::process::Signal; use slog::{Logger, debug, error, info, warn}; use tokio::fs::File; use tokio::io::{AsyncReadExt as _, AsyncSeekExt as _, AsyncWriteExt as _}; use tokio::process::Child; use tokio::sync::mpsc; use tokio::task::{JoinError, JoinHandle}; -use tokio::time::{Duration, Instant, sleep}; +use tokio::time::{Duration, Instant, sleep, timeout}; use tokio::{pin, select, spawn}; use tokio_tungstenite::WebSocketStream; use tokio_tungstenite::tungstenite::protocol::Message as WebSocketMessage; @@ -48,7 +49,7 @@ use sush_common::jobs::{ }; use tokio_util::sync::CancellationToken; -use crate::executor::kill_job; +use crate::executor::{job_pgid, kill_job}; use crate::io::JobIo; use crate::mux::WebSocketMux; @@ -60,6 +61,20 @@ pub type SocketReceiver = mpsc::Receiver<(SocketStream, Access)>; /// attach or make progress. const STREAMING_LINGER: Duration = Duration::from_secs(600); +/// How long a single send to a streaming consumer may stall +/// before the job is cancelled. +const STREAMING_SEND_TIMEOUT: Duration = Duration::from_secs(60); + +/// How long a dead streaming job may drain in total, since an +/// escaped descendant can hold the output pipe open forever. +const STREAMING_DRAIN_LIMIT: Duration = Duration::from_secs(60); + +/// How long a stopped job may clean up after SIGTERM before SIGKILL. +const KILL_GRACE: Duration = Duration::from_secs(5); + +/// How many times a failed SIGKILL is retried before giving up. +const KILL_ATTEMPTS: usize = 3; + pub struct Job { task: JoinHandle<(Result, JobOutputState)>, tx_client: SocketSender, @@ -113,7 +128,9 @@ async fn job( let mut stderr_hasher = Hasher::new(); let mut clients = WebSocketMux::new(); let mut fatal = Option::::None; + let pgid = job_pgid(&child); let mut killed = false; + let mut kills = 0; let mut dead = false; let streaming = matches!(streaming, Streaming::Output); let mut pending = Option::::None; @@ -133,6 +150,12 @@ async fn job( pin!(drain_timeout); let linger = sleep(Duration::default()); pin!(linger); + let drain_limit = sleep(Duration::default()); + pin!(drain_limit); + let attach_deadline = sleep(STREAMING_LINGER); + pin!(attach_deadline); + let kill_timeout = sleep(Duration::default()); + pin!(kill_timeout); loop { // A finished stream is done once its consumer has taken every chunk. if eof && attached && pending.is_none() { @@ -242,7 +265,7 @@ async fn job( // Attach a new client, send it the current window size, and play back the last buffer. // Streaming jobs accept one read-write consumer, even posthumously. - Some((mut client, access)) = rx_client.recv(), if !dead || streaming => { + Some((client, access)) = rx_client.recv(), if !dead || streaming => { if streaming { if access != Access::ReadWrite || attached || tx_socket.try_send(client).is_err() { debug!(log, "refused streaming client"); @@ -251,25 +274,31 @@ async fn job( } continue; } + let mut initial = Vec::new(); match io.get_window_size() { Err(error) => error!(log, "failed to get pseudoterminal window size"; "error" => %error), Ok(size) => { - match client.send(Message::Control(Control::WindowChange(size.clone())).try_into().unwrap()).await { - Err(error) => error!(log, "failed to send pty window size"; "error" => %error), - Ok(()) => debug!(log, "sent pty window size"; "size" => ?size), - } + debug!(log, "sending pty window size"; "size" => ?size); + initial.push(Message::Control(Control::WindowChange(size)).try_into().unwrap()); } } - if let Ok(Some(playback)) = playback_buffer(&mut stdout_file, INTERACTIVE_JOB_BUFFER_SIZE).await { - let playback_len = playback.len(); - match client.send(Message::Data(playback).try_into().unwrap()).await { - Err(error) => error!(log, "failed to play back job output"; "error" => %error), - Ok(()) => debug!(log, "played back output"; "bytes" => playback_len), + match playback_buffer(&mut stdout_file, INTERACTIVE_JOB_BUFFER_SIZE).await { + Ok(Some(playback)) => { + debug!(log, "playing back output"; "bytes" => playback.len()); + initial.push(Message::Data(playback).try_into().unwrap()); + } + Ok(None) => (), + Err(error) => { + error!(log, "failed to play back job output"; "error" => %error); + if let Err(error) = stdout_file.seek(SeekFrom::End(0)).await { + error!(log, "failed to restore output file position"; "error" => %error); + stop.cancel(); + } } } - clients.add(client, access, stop.child_token()); + clients.add(client, access, initial, stop.child_token()); } // Handle a message from a client. @@ -318,24 +347,51 @@ async fn job( } } - // Stop job on cancellation signal, but only once. + // Stop job on cancellation signal: terminate first, then kill. _ = stop.cancelled(), if !killed => { - kill_job(&log, &child); + kill_job(&log, pgid, Signal::TERM); + kill_timeout.as_mut().reset(Instant::now() + KILL_GRACE); killed = true; } + // Upgrade to SIGKILL a stopped group that outlives its grace + // period, even if its leader is already dead. + _ = &mut kill_timeout, if killed && kills < KILL_ATTEMPTS => { + warn!(log, "job survived SIGTERM grace period, upgrading to SIGKILL"); + kill_job(&log, pgid, Signal::KILL); + kill_timeout.as_mut().reset(Instant::now() + KILL_GRACE); + kills += 1; + } + // Notice when the job dies, but do not exit the loop; // we must continue reading output until we hit EOF or // the drain timeout expires. _ = child.wait(), if !dead => { debug!(log, "reaped job process"); drain_timeout.as_mut().reset(Instant::now() + io.drain_timeout()); + if streaming { + drain_limit.as_mut().reset(Instant::now() + STREAMING_DRAIN_LIMIT); + } dead = true; } + // Cut off a dead stream that never reaches EOF. + _ = &mut drain_limit, if streaming && dead && !eof => { + warn!(log, "streaming job output never closed"); + break; + } + + // Give up on a live stream whose consumer never attached. + _ = &mut attach_deadline, if streaming && !attached && !eof => { + warn!(log, "streaming consumer never attached"); + stop.cancel(); + break; + } + // Give up on a stream whose consumer never drained it. _ = &mut linger, if eof => { warn!(log, "streaming consumer never drained the output"); + stop.cancel(); break; } @@ -358,11 +414,16 @@ async fn job( let _ = tx_output.try_send(buf); } - // Reap the process. + // Reap the process. A stopped job that exited the loop alive + // still gets its SIGKILL upgrade after the grace period. let exit_status = select! { status = child.wait() => status, _ = stop.cancelled(), if !killed => { - kill_job(&log, &child); + kill_job(&log, pgid, Signal::KILL); + child.wait().await + } + _ = sleep(KILL_GRACE), if killed && !dead => { + kill_job(&log, pgid, Signal::KILL); child.wait().await } }; @@ -398,14 +459,24 @@ async fn stream_output( return; }; while let Some(buf) = rx_output.recv().await { - if let Err(error) = client.send(Message::Data(buf).try_into().unwrap()).await { - error!(log, "failed to stream output to client"; "error" => %error); - stop.cancel(); - return; + match timeout( + STREAMING_SEND_TIMEOUT, + client.send(Message::Data(buf).try_into().unwrap()), + ) + .await + { + Ok(Ok(())) => continue, + Ok(Err(error)) => error!(log, "failed to stream output to client"; "error" => %error), + Err(_) => error!(log, "timed out streaming output to client"), } + stop.cancel(); + return; } - let _ = client.send(Message::Close.try_into().unwrap()).await; - let _ = client.close(None).await; + let _ = timeout(STREAMING_SEND_TIMEOUT, async { + let _ = client.send(Message::Close.try_into().unwrap()).await; + let _ = client.close(None).await; + }) + .await; } fn process_exit(exit_status: ExitStatus) -> Result { @@ -424,8 +495,8 @@ fn process_exit(exit_status: ExitStatus) -> Result { } /// Fetch the last few bytes of the output file for client play back. -/// Errors here indicate problems with the output file, and so should -/// be treated as job-ending. +/// On error the file position may be anywhere; the caller must restore +/// it before the file is written again, or end the job. async fn playback_buffer( output_file: &mut File, output_bytes: usize, diff --git a/server/src/manager.rs b/server/src/manager.rs index d5f114e1..82aa8296 100644 --- a/server/src/manager.rs +++ b/server/src/manager.rs @@ -622,7 +622,7 @@ impl JobManager { &self, authn: &Identity, job_id: &JobId, - JobStopParams { wait }: JobStopParams, + JobStopParams { wait, .. }: JobStopParams, ) -> Result<(), JobError> { self.job_request(authn, JobRequest::Stop(job_id.to_owned())) .await?; diff --git a/server/src/mux.rs b/server/src/mux.rs index 8994b992..c90d6f87 100644 --- a/server/src/mux.rs +++ b/server/src/mux.rs @@ -16,6 +16,7 @@ use futures::stream::{FuturesUnordered, SplitStream}; use futures::{SinkExt as _, Stream, StreamExt}; use tokio::sync::broadcast; use tokio::task::JoinHandle; +use tokio::time::{Duration, timeout}; use tokio::{select, spawn}; use tokio_stream::{StreamMap, StreamNotifyClose}; use tokio_tungstenite::tungstenite::error::Error as WebSocketError; @@ -30,6 +31,9 @@ use crate::job::SocketStream; /// before it is disconnected. pub const MUX_CLIENT_CHANNEL_CAPACITY: usize = 100; +/// How long a writer task may spend closing its socket. +const CLOSE_TIMEOUT: Duration = Duration::from_secs(5); + /// Client identifiers are opaque and ephemeral. We use monotonically /// increasing integers because they're simple and cheap. pub type ClientId = usize; @@ -61,6 +65,7 @@ impl WebSocketMux { &mut self, stream: SocketStream, access: Access, + initial: Vec, stop: CancellationToken, ) -> ClientId { let client_id = self.next_client_id; @@ -71,20 +76,24 @@ impl WebSocketMux { self.stop.insert(client_id, stop.clone()); let mut rx = self.send.subscribe(); let handle = spawn(async move { - loop { - select! { - recvd = rx.recv() => { - let Ok(message) = recvd else { break }; - if to.send(message).await.is_err() { - break - } + let io = async { + for message in initial { + if to.send(message).await.is_err() { + return; } - _ = stop.cancelled() => { - break; + } + loop { + let Ok(message) = rx.recv().await else { return }; + if to.send(message).await.is_err() { + return; } } + }; + select! { + _ = io => {} + _ = stop.cancelled() => {} } - let _ = to.close().await; + let _ = timeout(CLOSE_TIMEOUT, to.close()).await; stop.cancel(); client_id }); @@ -116,6 +125,14 @@ impl WebSocketMux { } } +impl Drop for WebSocketMux { + fn drop(&mut self) { + for stop in self.stop.values() { + stop.cancel(); + } + } +} + impl Stream for WebSocketMux { type Item = (ClientId, Result); diff --git a/server/src/proxy.rs b/server/src/proxy.rs index ce65dc61..f5275355 100644 --- a/server/src/proxy.rs +++ b/server/src/proxy.rs @@ -7,9 +7,8 @@ //! Terminates client connections and routes each request to a sled, //! answering only `/version` itself. //! A request that names a target goes to the first sled the target -//! resolves to. Anything else goes to the sled hosting the proxy -//! when known, else a sticky default, because identities are cached -//! on the sled that authenticated them. +//! resolves to. Anything else goes to the sled hosting the proxy, +//! which any request may reach. //! Requests are forwarded untouched: bound request signatures cover //! the exact request line, so the proxy may never rewrite one. @@ -17,7 +16,7 @@ use std::collections::BTreeMap; use std::convert::Infallible; use std::io; use std::net::SocketAddr; -use std::sync::{Arc, Mutex}; +use std::sync::Arc; use http::header::CONTENT_TYPE; use http::{Method, StatusCode}; @@ -106,11 +105,7 @@ impl ProxyServer { let listener = TcpListener::bind(local_addr).await?; let local_addr = listener.local_addr()?; let acceptor = tls.map(|config| TlsAcceptor::from(Arc::new(config))); - let router = Arc::new(Router { - targets, - home, - default: Mutex::new(None), - }); + let router = Arc::new(Router { targets, home }); spawn(listen( log.new(o!("component" => "proxy")), listener, @@ -138,7 +133,6 @@ impl ProxyServer { struct Router { targets: watch::Receiver, home: Option, - default: Mutex>, } impl Router { @@ -155,29 +149,21 @@ impl Router { StatusCode::BAD_REQUEST, format!("unable to parse target `{bad}`"), ))), - None => { - if let Some(home) = self.home.as_ref() - && let Some(addr) = targets.sleds.get(home) - { - return Ok(*addr); - } - let mut default = self.default.lock().unwrap(); - if let Some(baseboard) = default.as_ref() - && let Some(addr) = targets.sleds.get(baseboard) - { - return Ok(*addr); - } - match targets.sleds.iter().next() { - Some((baseboard, addr)) => { - *default = Some(baseboard.clone()); - Ok(*addr) - } + None => match self.home.as_ref() { + Some(home) => targets.sleds.get(home).copied().ok_or_else(|| { + Box::new(error_response( + StatusCode::SERVICE_UNAVAILABLE, + format!("home sled `{home}` unknown to the proxy"), + )) + }), + None => match targets.sleds.iter().next() { + Some((_, addr)) => Ok(*addr), None => Err(Box::new(error_response( StatusCode::SERVICE_UNAVAILABLE, "no sleds known to the proxy", ))), - } - } + }, + }, } } } @@ -288,7 +274,13 @@ where Some(response) => response, None => match router.route(&request) { Ok(addr) => forward(log, addr, request).await, - Err(response) => *response, + Err(response) => { + warn!( + log, "request not routable"; + "uri" => %request.uri(), "status" => %response.status() + ); + *response + } }, }) } @@ -377,8 +369,8 @@ mod test { s.parse().unwrap() } - /// Unrouted requests go to the home sled while the targets know - /// it, and otherwise to a sticky default. + /// Unrouted requests go to the home sled, or fail while the + /// targets don't know it. Without a home, any sled will do. #[test] fn home_preference() { let sled = |serial: &str| BaseboardId { @@ -391,20 +383,27 @@ mod test { targets.sleds.insert(sled("home"), addr(2)); let (tx, rx) = watch::channel(targets); let router = Router { - targets: rx, + targets: rx.clone(), home: Some(sled("home")), - default: Mutex::new(None), }; let get = request("/versions"); assert_eq!(router.route(&get).unwrap(), addr(2)); tx.send_modify(|t| { t.sleds.remove(&sled("home")); }); - assert_eq!(router.route(&get).unwrap(), addr(1)); + assert_eq!( + router.route(&get).unwrap_err().status(), + StatusCode::SERVICE_UNAVAILABLE + ); tx.send_modify(|t| { t.sleds.insert(sled("home"), addr(2)); }); assert_eq!(router.route(&get).unwrap(), addr(2)); + let homeless = Router { + targets: rx, + home: None, + }; + assert_eq!(homeless.route(&get).unwrap(), addr(1)); } /// The proxy answers `GET /version` itself unless `via` routes it. diff --git a/server/src/server.rs b/server/src/server.rs index 3568ac36..71979a61 100644 --- a/server/src/server.rs +++ b/server/src/server.rs @@ -299,6 +299,7 @@ impl SushApi for ApiServer { ctx: RequestContext, headers: Header, params: PathParams, + _query: QueryParams, ) -> Result, HttpError> { let mgr = ctx.context(); let Authorization { authorization } = headers.into_inner(); diff --git a/sush.json b/sush.json index 7e975112..e36a0790 100644 --- a/sush.json +++ b/sush.json @@ -613,6 +613,15 @@ "schema": { "$ref": "#/components/schemas/JobId" } + }, + { + "in": "query", + "name": "via", + "description": "Where a proxy should route this request.", + "schema": { + "nullable": true, + "type": "string" + } } ], "responses": { @@ -660,6 +669,15 @@ "$ref": "#/components/schemas/JobId" } }, + { + "in": "query", + "name": "via", + "description": "Where a proxy should route this request. Sleds ignore it.", + "schema": { + "nullable": true, + "type": "string" + } + }, { "in": "query", "name": "wait", diff --git a/tests/src/manager_tests.rs b/tests/src/manager_tests.rs index 55d0e99b..96b90b53 100644 --- a/tests/src/manager_tests.rs +++ b/tests/src/manager_tests.rs @@ -14,7 +14,7 @@ use std::time::Duration; use chrono::Utc; use function_name::named; use http_range_header::{EndPosition, StartPosition, SyntacticallyCorrectRange as Range}; -use libc::{SIGKILL, SIGXCPU}; +use libc::{SIGTERM, SIGXCPU}; use pwd::Passwd; use sled_hardware_types::BaseboardId; use slog::{Discard, Logger, o}; @@ -271,6 +271,7 @@ async fn job_stop() { &job_id, JobStopParams { wait: JobWait::Stop, + ..Default::default() }, ) .await @@ -313,6 +314,7 @@ async fn job_stop() { &job_id, JobStopParams { wait: JobWait::Stop, + ..Default::default() }, ) .await @@ -329,7 +331,7 @@ async fn job_stop() { check_status_stopped( status, &job_id, - Err(ProcessError::Killed(SIGKILL)), + Err(ProcessError::Killed(SIGTERM)), Some(0), Some(0), ); @@ -390,6 +392,7 @@ async fn cancel_queued_job() { &job_id_b, JobStopParams { wait: JobWait::Stop, + ..Default::default() }, ) .await @@ -405,6 +408,7 @@ async fn cancel_queued_job() { &job_id_a, JobStopParams { wait: JobWait::Stop, + ..Default::default() }, ) .await @@ -820,7 +824,7 @@ async fn shutdown() { check_status_stopped( status, &job_id, - Err(ProcessError::Killed(SIGKILL)), + Err(ProcessError::Killed(SIGTERM)), Some(0), Some(0), ); @@ -980,6 +984,7 @@ async fn attribution() { &job_id_b, JobStopParams { wait: JobWait::Stop, + ..Default::default() }, ) .await @@ -994,6 +999,7 @@ async fn attribution() { &job_id_a, JobStopParams { wait: JobWait::Stop, + ..Default::default() }, ) .await @@ -1399,6 +1405,7 @@ async fn attach_grants() { &job_id, JobStopParams { wait: JobWait::Stop, + ..Default::default() }, ) .await