Merge pull request #60 from oxidecomputer/availability-fixes

Availability fixes

Author: plotnick

api/src/lib.rs

@@ -189,6 +189,7 @@ pub trait SushApi {
         ctx: RequestContext<Self::Context>,
         headers: Header<Authorization>,
         params: PathParams<JobIdParam>,
+        query: QueryParams<RoutingParam>,
     ) -> Result<HttpResponseOk<JsonJobStatusMap>, 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<String>,
 }
 
 /// Simple pagination for history list.

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<KeyId, CommandError> {
         match self.get_output_format() {
             OutputFormat::Json => Ok(key_id),

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<JobStatusMap, CommandError> {
-    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<Target, CommandError> {
@@ -2120,7 +2158,7 @@ async fn resolve_serial(
     job_id: &JobId,
     serial: &str,
 ) -> Result<BaseboardId, CommandError> {
-    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));

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);

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<KeyId, CommandError> {
         self.cli.really_revoke(what, key_id)
     }

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}}

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> {
+    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<Pid>, 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");
     }
 }
 

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.