diff --git a/Cargo.lock b/Cargo.lock index 4f83cb0e..078598d6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4839,6 +4839,8 @@ dependencies = [ name = "sush-client" version = "0.1.0" dependencies = [ + "anstream", + "anstyle", "async-recursion", "atomicwrites", "base64 0.22.1", diff --git a/Cargo.toml b/Cargo.toml index 36381c34..37766b1b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,6 +6,8 @@ resolver = "2" version = "0.1.0" [workspace.dependencies] +anstream = "1" +anstyle = "1" async-recursion = "1" attest-mock = { git = "https://github.com/oxidecomputer/dice-util", rev = "10952e8d9599b735b85d480af3560a11700e5b64" } atomicwrites = "0.4" diff --git a/client/Cargo.toml b/client/Cargo.toml index 44bf21bc..3c1a479d 100644 --- a/client/Cargo.toml +++ b/client/Cargo.toml @@ -17,6 +17,8 @@ path = "src/main.rs" permslip = ["dep:permission-slip-client", "dep:permission-slip-common"] [dependencies] +anstream.workspace = true +anstyle.workspace = true async-recursion.workspace = true atomicwrites.workspace = true base64.workspace = true diff --git a/client/src/cli.rs b/client/src/cli.rs index e4c51dab..158b3708 100644 --- a/client/src/cli.rs +++ b/client/src/cli.rs @@ -13,6 +13,8 @@ use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; use std::time::{Duration, SystemTime}; +use anstream::print; +use anstyle::{AnsiColor, Style}; use atomicwrites::{AtomicFile, OverwriteBehavior}; use bytesize::ByteSize; use chrono::TimeDelta; @@ -32,7 +34,7 @@ use sush_common::jobs::{ SessionSignerNonce, SignedJob, job_status_to_json_map, }; use sush_common::keys::{KeyId, Signature, SshPublicKey}; -use sush_common::targets::{MAX_CUBBY, SledId, SledVersion}; +use sush_common::targets::{MAX_CUBBY, SledHealth, SledId, SledVersion}; use sush_common::version::VersionInfo; use crate::AuthzSigner; @@ -1065,7 +1067,8 @@ mod test { /// Cell text width for one sled in the rack drawing. const CELL: usize = 28; -/// One sled cell: serial on the left, build on the right. +/// One sled cell: serial on the left, build on the right, colored by +/// health. fn rack_cell(sled: Option<&SledVersion>) -> String { match sled { Some(sled) => { @@ -1080,12 +1083,29 @@ fn rack_cell(sled: Option<&SledVersion>) -> String { } None => String::new(), }; - format!(" {:<12.12}{:>14.14} ", sled.baseboard.serial_number, build) + let style = health_style(sled); + format!( + "{style} {:<12.12}{:>14.14} {style:#}", + sled.baseboard.serial_number, build + ) } None => " ".repeat(CELL), } } +/// Green gossips with the answering sled, yellow was once known but is +/// out of contact, and red is in the cubby map with no other sign of +/// life. Sleds without health (an old server, a newer state than this +/// build knows) stay unstyled, which renders as nothing. +fn health_style(sled: &SledVersion) -> Style { + match sled.health { + Some(SledHealth::Linked) => AnsiColor::Green.on_default(), + Some(SledHealth::Unlinked) if sled.version.is_some() => AnsiColor::Yellow.on_default(), + Some(SledHealth::Unlinked) => AnsiColor::Red.on_default(), + Some(SledHealth::Unknown) | None => Style::new(), + } +} + /// Draw the rack as wicket does: 16 rows of two cubbies, numbered /// bottom-to-top and left-to-right per RFD 200, split where the /// switches and power shelves sit. Sleds known only by build (no @@ -1140,11 +1160,10 @@ mod rack { version: "0.1.0".to_string(), commit: "f078e863b17359031de072222bb631270f2d5157".to_string(), }), + health: None, } } - /// Compare the rack drawing against the snapshot in - /// `tests/output/`, or rewrite it under `EXPECTORATE=overwrite`. #[test] fn rack_drawing() { let mut sleds = vec![ @@ -1162,10 +1181,30 @@ mod rack { if let Some(version) = &mut sleds[1].version { version.commit.push_str("-dirty"); } - let drawing = draw_rack(&sleds); - let path = "tests/output/rack.txt"; + check(&draw_rack(&sleds), "tests/output/rack.txt"); + } + + /// A healthy sled, a silent one, and one that is only a cubby + /// number, pinning the color codes. + #[test] + fn rack_drawing_health() { + let mut sleds = vec![ + sled(14, "BRM42220030"), + sled(15, "BRM42220036"), + sled(16, "2CN2M459"), + ]; + sleds[0].health = Some(SledHealth::Linked); + sleds[1].health = Some(SledHealth::Unlinked); + sleds[2].health = Some(SledHealth::Unlinked); + sleds[2].version = None; + check(&draw_rack(&sleds), "tests/output/rack-health.txt"); + } + + /// Compare against the snapshot at `path`, or rewrite it under + /// `EXPECTORATE=overwrite`. + fn check(drawing: &str, path: &str) { if env::var("EXPECTORATE").as_deref() == Ok("overwrite") { - write(path, &drawing).unwrap(); + write(path, drawing).unwrap(); } else { let expected = read_to_string(path).expect("missing snapshot"); assert_eq!(drawing, expected, "rack drawing changed:\n{drawing}"); diff --git a/client/src/commands.rs b/client/src/commands.rs index 7c3f1036..4af83637 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -1151,22 +1151,25 @@ async fn session( }, Some(client), ) => { - let (session_id, nonce) = if let (Some(session_id), Some(nonce)) = (session_id, nonce) { - (session_id, nonce) - } else { - let (baseboard_id, nonce) = with_login(ctx, client, async || { - Ok(( - client.target().send().await?.into_inner(), - client.session_start_nonce().send().await?.into_inner(), - )) - }) - .await?; - let (session_id, nonce) = - session_create(permslip, permslip_url, &baseboard_id, nonce.nonce).await?; - ctx.session_created(Session::new(session_id), nonce); - (session_id, nonce) - }; - session_start(ctx, client, session_id, nonce, wait).await + // Creation already announces the session; don't echo it + // when the start succeeds. + let (session_id, nonce, show) = + if let (Some(session_id), Some(nonce)) = (session_id, nonce) { + (session_id, nonce, true) + } else { + let (baseboard_id, nonce) = with_login(ctx, client, async || { + Ok(( + client.target().send().await?.into_inner(), + client.session_start_nonce().send().await?.into_inner(), + )) + }) + .await?; + let (session_id, nonce) = + session_create(permslip, permslip_url, &baseboard_id, nonce.nonce).await?; + ctx.session_created(Session::new(session_id), nonce); + (session_id, nonce, false) + }; + session_start(ctx, client, session_id, nonce, wait, show).await } #[cfg(feature = "permslip")] @@ -1199,7 +1202,7 @@ async fn session( } else { return Err(CommandError::SigningUnavailable); }; - session_start(ctx, client, session_id, nonce, wait).await + session_start(ctx, client, session_id, nonce, wait, true).await } (SessionCommand::Allow { key_id, write }, Some(client)) => { @@ -1758,6 +1761,7 @@ async fn session_start( session_id: SessionId, signer_nonce: SessionSignerNonce, wait: bool, + show: bool, ) -> Result<(), CommandError> { let session = Session::new(session_id); with_login(ctx, client, async || { @@ -1771,7 +1775,9 @@ async fn session_start( }) .await? .into_inner(); - ctx.session_started(session, true); + if show { + ctx.session_started(session, true); + } Ok(()) } diff --git a/client/tests/output/rack-health.txt b/client/tests/output/rack-health.txt new file mode 100644 index 00000000..156814a9 --- /dev/null +++ b/client/tests/output/rack-health.txt @@ -0,0 +1,19 @@ + ┌────────────────────────────┬────────────────────────────┐ + 30 │ │ │ 31 + 28 │ │ │ 29 + 26 │ │ │ 27 + 24 │ │ │ 25 + 22 │ │ │ 23 + 20 │ │ │ 21 + 18 │ │ │ 19 + 16 │ 2CN2M459 │ │ 17 + ├────────────────────────────┼────────────────────────────┤ + 14 │ BRM42220030 0.1.0 f078e86 │ BRM42220036 0.1.0 f078e86 │ 15 + 12 │ │ │ 13 + 10 │ │ │ 11 + 8 │ │ │ 9 + 6 │ │ │ 7 + 4 │ │ │ 5 + 2 │ │ │ 3 + 0 │ │ │ 1 + └────────────────────────────┴────────────────────────────┘ diff --git a/common/src/targets.rs b/common/src/targets.rs index 8d865713..d1eb8e02 100644 --- a/common/src/targets.rs +++ b/common/src/targets.rs @@ -30,12 +30,28 @@ pub type Cubbies = BTreeMap; /// The highest cubby number in a rack. pub const MAX_CUBBY: u8 = 31; -/// One sled's location and build. +/// One sled's location, build, and health. #[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)] pub struct SledVersion { pub cubby: Option, pub baseboard: BaseboardId, pub version: Option, + #[serde(default)] + pub health: Option, +} + +/// One sled's gossip link health, as the answering sled sees it. A +/// silent death can lag `Linked` at TCP's pace. +#[derive(Clone, Copy, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)] +#[serde(rename_all = "lowercase")] +pub enum SledHealth { + /// Attested, holding a live gossip link. + Linked, + /// Known by version or cubby, but no live link. + Unlinked, + /// A state from a build newer than this one. + #[serde(other)] + Unknown, } /// The sleds a request names. diff --git a/server/src/gossip.rs b/server/src/gossip.rs index 6fca713e..bc64d4ad 100644 --- a/server/src/gossip.rs +++ b/server/src/gossip.rs @@ -32,6 +32,7 @@ use futures::StreamExt as _; use rumors::{Error, Joined, Network, Peer, Rumors, Ticks, Version}; use serde::Serialize; use serde::de::DeserializeOwned; +use sled_hardware_types::BaseboardId; use slog::{Logger, debug, info, o, warn}; use sprockets_tls::keys::SprocketsConfig; use tokio::sync::watch; @@ -43,9 +44,12 @@ use tokio_util::sync::CancellationToken; use rumors::link::routed::Endpoint; use crate::bookmark::{BookmarkSource, SushBookmark}; -use crate::link::{CorpusSource, SprocketsDial, SprocketsLink, Transport}; +use crate::link::{AttestedBaseboards, CorpusSource, SprocketsDial, SprocketsLink, Transport}; -/// Manager timing. The defaults suit a rack; tests shrink them. +/// The attested baseboards of our live gossip peers. +pub type LinkedBaseboards = watch::Receiver>; + +/// Manager timing. The defaults suit a rack, tests shrink them. #[derive(Clone, Debug)] pub struct GossipConfig { /// How often absent links are re-established. @@ -94,6 +98,12 @@ pub fn isolated(seed: Rumors) -> watch::Receiver rx } +/// A linked set that is forever empty, to accompany [`isolated`]. +pub fn lonely() -> LinkedBaseboards { + let (_tx, rx) = watch::channel(BTreeSet::new()); + rx +} + /// Whether the peer's universe dominates ours, by rumors' documented rule. fn remote_dominates( local_events: &Ticks, @@ -110,9 +120,10 @@ fn remote_dominates( } /// Bind a sprockets transport on `listen_addr` and run a gossip manager -/// over it until `shutdown`. Returns the address the listener bound and -/// the channel following the current universe. A caller that cannot bind -/// may fall back to [`isolated`]. +/// over it until `shutdown`. Returns the address the listener bound, the +/// channel following the current universe, and the channel following the +/// baseboards we hold live links to. A caller that cannot bind may fall +/// back to [`isolated`]. #[allow(clippy::too_many_arguments)] pub async fn spawn_gossip( log: &Logger, @@ -124,7 +135,7 @@ pub async fn spawn_gossip( seed: Rumors, bookmarks: BookmarkSource, shutdown: CancellationToken, -) -> io::Result<(SocketAddrV6, watch::Receiver>)> +) -> io::Result<(SocketAddrV6, watch::Receiver>, LinkedBaseboards)> where T: DeserializeOwned + Serialize + Send + Sync + 'static, { @@ -138,14 +149,15 @@ where ) .await?; let bound = transport.bound(); - let universe = spawn_gossip_manager(log, config, transport, peers, seed, bookmarks, shutdown); - Ok((bound, universe)) + let (universe, linked) = + spawn_gossip_manager(log, config, transport, peers, seed, bookmarks, shutdown); + Ok((bound, universe, linked)) } /// Run a gossip manager until `shutdown`. Establishes and serves links for /// the addresses on `peers`, drives gossip on every link, and resolves -/// universe collisions. The returned channel follows the current universe, -/// starting at `seed`. +/// universe collisions. The returned channels follow the current universe, +/// starting at `seed`, and the baseboards we hold live links to. #[allow(clippy::too_many_arguments)] pub fn spawn_gossip_manager( log: &Logger, @@ -155,20 +167,23 @@ pub fn spawn_gossip_manager( seed: Rumors, bookmarks: BookmarkSource, shutdown: CancellationToken, -) -> watch::Receiver> +) -> (watch::Receiver>, LinkedBaseboards) where T: DeserializeOwned + Serialize + Send + Sync + 'static, { let (publish, subscribe) = watch::channel(Universe::genesis(seed.clone())); + let (linked, subscribe_linked) = watch::channel(BTreeSet::new()); let manager = Manager { log: log.new(o!("component" => "gossip manager")), config, endpoint: transport.endpoint(), + attested: transport.baseboards().watch(), transport, peers, rumors: seed, bookmarks, publish, + linked, drivers: JoinSet::new(), live: HashMap::new(), dials: JoinSet::new(), @@ -177,7 +192,7 @@ where shutdown, }; spawn(manager.run()); - subscribe + (subscribe, subscribe_linked) } /// A link establishment that finished, and the peer it was aimed at. @@ -200,6 +215,8 @@ struct Manager { rumors: Rumors, bookmarks: BookmarkSource, publish: watch::Sender>, + attested: watch::Receiver, + linked: watch::Sender>, drivers: JoinSet<(SocketAddr, Stopped)>, live: HashMap, dials: JoinSet, @@ -224,29 +241,80 @@ where self.prune(); self.link_absent(); } + // A late attestation names a link that is already live. + Ok(()) = self.attested.changed() => {} Some((peer, link)) = self.transport.accept() => { self.on_link(peer, link).await; } - Some(done) = self.dials.join_next() => { - if let Ok((peer, result)) = done { - self.dialing.remove(&peer); - match result { - Ok(link) => self.on_link(peer, link).await, - Err(err) => { - debug!(self.log, "link failed"; "peer" => %peer, "error" => err); + Some(done) = self.dials.join_next_with_id() => { + match done { + // A dial may finish after its peer was pruned. Only + // the dial still on the books may act on its result. + Ok((id, (peer, result))) => { + if self.dialing.get(&peer).is_some_and(|dial| dial.id() == id) { + self.dialing.remove(&peer); + match result { + Ok(link) => self.on_link(peer, link).await, + Err(err) => { + debug!( + self.log, "link failed"; + "peer" => %peer, "error" => err, + ); + } + } } } + Err(err) => self.dialing.retain(|_, dial| dial.id() != err.id()), } } - Some(done) = self.drivers.join_next() => { - if let Ok((peer, stopped)) = done { - self.live.remove(&peer); - if matches!(stopped, Stopped::Dominated) { - self.joins.insert(peer); + Some(done) = self.drivers.join_next_with_id() => { + match done { + Ok((id, (peer, stopped))) => { + debug!( + self.log, "link driver stopped"; + "peer" => %peer, + "dominated" => matches!(stopped, Stopped::Dominated), + ); + // A driver may die after a fresh link to its peer + // has replaced it. Only the living driver's death + // may take the peer out of the live set. + if self.live.get(&peer).is_some_and(|live| live.id() == id) { + self.live.remove(&peer); + if matches!(stopped, Stopped::Dominated) { + self.joins.insert(peer); + } + } + } + // A panicked driver reports nothing, and its peer + // would never be re-linked. + Err(err) => { + if err.is_panic() { + warn!(self.log, "link driver panicked"; "error" => %err); + } + self.live.retain(|_, live| live.id() != err.id()); } } } } + self.publish_linked(); + } + } + + /// Publish the attested baseboards of the peers with live links. + fn publish_linked(&self) { + let attested = self.attested.borrow(); + let linked: BTreeSet = self + .live + .keys() + .filter_map(|addr| attested.resolve(addr)) + .collect(); + if *self.linked.borrow() != linked { + debug!( + self.log, "linked baseboards changed"; + "linked" => ?linked, + "live" => ?self.live.keys().collect::>(), + ); + self.linked.send_replace(linked); } } @@ -301,7 +369,7 @@ where self.live.retain(cull); self.dialing.retain(cull); self.joins.retain(|peer| want.contains(peer)); - self.transport.retain_pools(&self.peers.borrow()); + self.transport.retain_peers(&self.peers.borrow()); } /// Gossip on a fresh link, or use it to join a universe that beat ours. @@ -316,6 +384,7 @@ where /// Spawn a session driver owning `link`: push our changes, and serve /// whatever the peer initiates, until the link fails. fn drive(&mut self, peer: SocketAddr, link: SprocketsLink) { + debug!(self.log, "driving link"; "peer" => %peer); let rumors = self.rumors.clone(); let log = self.log.clone(); let handle = self diff --git a/server/src/link.rs b/server/src/link.rs index 7d94a2af..5549f81d 100644 --- a/server/src/link.rs +++ b/server/src/link.rs @@ -19,9 +19,9 @@ //! are reused: a connection dropped mid-stream is discarded at the //! pool's door. -use std::collections::{BTreeSet, HashMap}; +use std::collections::{BTreeMap, BTreeSet, HashMap}; use std::io; -use std::net::{SocketAddr, SocketAddrV6}; +use std::net::{Ipv6Addr, SocketAddr, SocketAddrV6}; use std::pin::Pin; use std::sync::{Arc, Mutex}; use std::task::{Context, Poll}; @@ -36,13 +36,14 @@ use qorb::pool::Pool; use qorb::resolvers::fixed::FixedResolver; use rumors::link::STREAM_COUNT; use rumors::link::routed::{Config, Dial, Endpoint, Incoming, Listen, RoutedLink}; -use slog::{Logger, o, warn}; +use sled_hardware_types::BaseboardId; +use slog::{Logger, debug, o, warn}; use sprockets_tls::keys::SprocketsConfig; use sprockets_tls::{Client, Server}; use tokio::io::{AsyncRead, AsyncReadExt as _, AsyncWrite, AsyncWriteExt as _, ReadBuf}; use tokio::net::TcpStream; use tokio::runtime::Handle; -use tokio::sync::mpsc; +use tokio::sync::{mpsc, watch}; use tokio::time::{sleep, timeout}; use tokio::{select, spawn}; use tokio_util::sync::CancellationToken; @@ -188,12 +189,116 @@ impl Drop for SprocketsConn { } } +/// The baseboards peers have attested at handshake, kept on a watch +/// channel so consumers follow the table as it changes. +#[derive(Clone)] +pub struct Baseboards { + known: watch::Sender, +} + +/// The attestation tables: a dialed connection names its peer's listen +/// address exactly; an accepted one names only its source IP. The +/// bootstrap network keeps IPs one-to-one with sleds, but localhost +/// tests stack peers on one IP, so an ambiguous IP resolves to no one. +/// [`Baseboards::retain`] bounds both tables to the peer set. +#[derive(Default)] +pub struct AttestedBaseboards { + dialed: BTreeMap, + accepted: BTreeMap>, +} + +impl AttestedBaseboards { + /// The baseboard attested at `addr`: seen on a dialed connection, + /// or the only baseboard to have connected from the same IP. + pub fn resolve(&self, addr: &SocketAddr) -> Option { + let SocketAddr::V6(addr) = addr else { + return None; + }; + if let Some(id) = self.dialed.get(addr) { + return Some(id.clone()); + } + match self.accepted.get(addr.ip()) { + Some(ids) if ids.len() == 1 => ids.first().cloned(), + _ => None, + } + } +} + +impl Baseboards { + fn new() -> Self { + Baseboards { + known: watch::Sender::new(AttestedBaseboards::default()), + } + } + + /// Record the peer attested on a connection dialed to `addr`. + fn dialed(&self, log: &Logger, addr: SocketAddrV6, platform_id: &str) { + let Some(id) = baseboard(log, platform_id) else { + return; + }; + self.known.send_if_modified(|known| { + let grew = known.dialed.get(&addr) != Some(&id); + if grew { + debug!(log, "attested by dial"; "addr" => %addr, "baseboard" => %id); + known.dialed.insert(addr, id); + } + grew + }); + } + + /// Record the peer attested on a connection accepted from `ip`. + fn accepted(&self, log: &Logger, ip: Ipv6Addr, platform_id: &str) { + let Some(id) = baseboard(log, platform_id) else { + return; + }; + self.known.send_if_modified(|known| { + if known.accepted.entry(ip).or_default().insert(id.clone()) { + debug!(log, "attested by accept"; "ip" => %ip, "baseboard" => %id); + true + } else { + false + } + }); + } + + /// A receiver following the attestation tables. + pub fn watch(&self) -> watch::Receiver { + self.known.subscribe() + } + + /// Forget the baseboards of peers outside `peers`. + fn retain(&self, peers: &BTreeSet) { + let ips: BTreeSet<&Ipv6Addr> = peers.iter().map(|addr| addr.ip()).collect(); + self.known.send_if_modified(|known| { + let before = (known.dialed.len(), known.accepted.len()); + known.dialed.retain(|addr, _| peers.contains(addr)); + known.accepted.retain(|ip, _| ips.contains(ip)); + (known.dialed.len(), known.accepted.len()) != before + }); + } +} + +/// The baseboard a platform id names (`prefix:part:revision:serial`). +fn baseboard(log: &Logger, platform_id: &str) -> Option { + let mut fields = platform_id.split(':'); + match (fields.nth(1), fields.nth(1)) { + (Some(part_number), Some(serial_number)) => Some(BaseboardId { + part_number: part_number.to_string(), + serial_number: serial_number.to_string(), + }), + _ => { + warn!(log, "unparseable platform id"; "platform_id" => platform_id); + None + } + } +} + /// Establishes a peer pool's connections, one attested handshake each. struct PoolConnector { log: Logger, config: SprocketsConfig, corpus: CorpusSource, - /// Ceiling on one handshake. + baseboards: Baseboards, timeout: Duration, } @@ -230,6 +335,8 @@ impl backend::Connector for PoolConnector { } }); let stream = dial.await.map_err(io::Error::other)??; + self.baseboards + .dialed(&self.log, addr, stream.peer_platform_id().as_str()); Ok(PooledConn { stream: Some(stream), dirty: false, @@ -259,8 +366,8 @@ pub struct SprocketsDial { log: Logger, config: SprocketsConfig, corpus: CorpusSource, + baseboards: Baseboards, timeout: Duration, - /// One pool per peer, created at first dial and dropped at prune. pools: Arc>>>>, } @@ -272,12 +379,14 @@ impl SprocketsDial { log: &Logger, config: SprocketsConfig, corpus: CorpusSource, + baseboards: Baseboards, timeout: Duration, ) -> Self { SprocketsDial { log: log.new(o!("component" => "sprockets dial")), config, corpus, + baseboards, timeout, pools: Arc::default(), } @@ -298,6 +407,7 @@ impl SprocketsDial { log: self.log.clone(), config: self.config.clone(), corpus: self.corpus.clone(), + baseboards: self.baseboards.clone(), timeout: self.timeout, }); let resolver = Box::new(FixedResolver::new([SocketAddr::V6(addr)])); @@ -308,9 +418,6 @@ impl SprocketsDial { set_config: SetConfig { max_count: MAX_SLOTS, min_connection_backoff: MIN_CONNECTION_BACKOFF, - // No liveness probing: a pooled connection that died - // idle is discovered by the stream that draws it, and - // the failed session re-links. health_interval: HEALTH_INTERVAL, ..SetConfig::default() }, @@ -318,7 +425,6 @@ impl SprocketsDial { }; match Pool::new(format!("gossip {addr}"), resolver, connector, policy) { Ok(pool) => pool, - // Registration is telemetry-only; the pool works either way. Err(err) => err.into_inner(), } } @@ -369,6 +475,7 @@ pub struct Transport { endpoint: Endpoint, incoming: Incoming, dial: SprocketsDial, + baseboards: Baseboards, bound: SocketAddrV6, } @@ -383,15 +490,17 @@ impl Transport { dial_timeout: Duration, shutdown: CancellationToken, ) -> io::Result { + let baseboards = Baseboards::new(); let (listen, bound) = SprocketsListen::bind( log, config.clone(), corpus.clone(), + baseboards.clone(), listen_addr, shutdown.clone(), ) .await?; - let dial = SprocketsDial::new(log, config, corpus, dial_timeout); + let dial = SprocketsDial::new(log, config, corpus, baseboards.clone(), dial_timeout); let router_config = Config { pending_headers: PENDING_HEADERS, ..Config::default() @@ -412,6 +521,7 @@ impl Transport { endpoint, incoming, dial, + baseboards, bound, }) } @@ -426,9 +536,16 @@ impl Transport { self.endpoint.clone() } - /// Drop the connection pools of peers outside `peers`. - pub fn retain_pools(&self, peers: &BTreeSet) { - self.dial.retain(peers) + /// The baseboards peers have attested to this transport. + pub fn baseboards(&self) -> &Baseboards { + &self.baseboards + } + + /// Drop the connection pools and recorded baseboards of peers + /// outside `peers`. + pub fn retain_peers(&self, peers: &BTreeSet) { + self.dial.retain(peers); + self.baseboards.retain(peers); } /// Receive the next link a peer established toward us, with the name it @@ -457,6 +574,7 @@ impl SprocketsListen { log: &Logger, config: SprocketsConfig, corpus: CorpusSource, + baseboards: Baseboards, listen_addr: SocketAddrV6, shutdown: CancellationToken, ) -> io::Result<(Self, SocketAddrV6)> { @@ -471,7 +589,7 @@ impl SprocketsListen { } }; let (tx, connections) = mpsc::channel(HANDSHAKE_QUEUE_DEPTH); - spawn(pump(server, corpus, tx, log, shutdown)); + spawn(pump(server, corpus, baseboards, tx, log, shutdown)); Ok((SprocketsListen { connections }, bound)) } } @@ -491,6 +609,7 @@ impl Listen for SprocketsListen { async fn pump( server: Server, corpus: CorpusSource, + baseboards: Baseboards, connections: mpsc::Sender, log: Logger, shutdown: CancellationToken, @@ -501,11 +620,16 @@ async fn pump( accepted = server.accept((corpus)()) => match accepted { Ok(acceptor) => { let connections = connections.clone(); + let baseboards = baseboards.clone(); let log = log.clone(); spawn(async move { match acceptor.handshake().await { // A closed queue means the endpoint is gone. - Ok((stream, _)) => { + Ok((stream, peer)) => { + if let SocketAddr::V6(peer) = peer { + let id = stream.peer_platform_id().as_str(); + baseboards.accepted(&log, *peer.ip(), id); + } let conn = SprocketsConn(Conn::Direct(Some(stream))); let _ = connections.send(conn).await; } diff --git a/server/src/main.rs b/server/src/main.rs index be837ad2..b76c43d2 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -22,7 +22,7 @@ use sush_api::sush_api_mod::api_description; use sush_common::targets::Cubbies; use sush_server::bookmark::BookmarkSource; use sush_server::executor::PathIsolation; -use sush_server::gossip::isolated; +use sush_server::gossip::{isolated, lonely}; use sush_server::manager::JobManager; use sush_server::output::JobOutputDir; use sush_server::server::ApiServer; @@ -111,6 +111,7 @@ async fn main() -> Result<(), String> { baseboard, cubbies, gossip, + lonely(), &roots, shutdown.clone(), ) diff --git a/server/src/manager.rs b/server/src/manager.rs index 0c8e2963..ccde5e8b 100644 --- a/server/src/manager.rs +++ b/server/src/manager.rs @@ -35,11 +35,12 @@ use sush_common::jobs::{ }; use sush_common::jobs::{JobOutputStream, SessionSushNonce}; use sush_common::keys::{KeyError, KeyId, SshPublicKey}; -use sush_common::targets::{Cubbies, SledVersion}; +use sush_common::targets::{Cubbies, SledHealth, SledVersion}; use sush_common::version::LONG_VERSION; use crate::error::JobError; use crate::executor::PathIsolation; +use crate::gossip::LinkedBaseboards; use crate::job::SocketSender; use crate::messages::v0::{CertRequest, IdentityRequest, JobRequest, Request, SessionRequest}; use crate::output::{JobOutputDir, JobOutputFileStream}; @@ -88,6 +89,7 @@ pub struct JobManager { session_sush_nonce: Arc>, output_dir: JobOutputDir, own_baseboard: BaseboardId, + linked: LinkedBaseboards, state: watch::Receiver, // from the state manager tx_req: mpsc::Sender, // to the state manager join_state: Option>, @@ -104,6 +106,7 @@ impl JobManager { own_baseboard: BaseboardId, cubbies: watch::Receiver, universe: watch::Receiver, + linked: LinkedBaseboards, roots: &[impl AsRef], shutdown: CancellationToken, ) -> Result { @@ -115,6 +118,7 @@ impl JobManager { own_baseboard, cubbies, universe, + linked, &roots, shutdown, ) @@ -129,6 +133,7 @@ impl JobManager { own_baseboard: BaseboardId, cubbies: watch::Receiver, universe: watch::Receiver, + linked: LinkedBaseboards, roots: &[Certificate], shutdown: CancellationToken, ) -> Result { @@ -154,6 +159,7 @@ impl JobManager { identities: Arc::new(Mutex::new(LruCache::new(MAX_CACHED_IDENTITIES))), session_sush_nonce, own_baseboard, + linked, output_dir, state: rx_state, tx_req, @@ -168,6 +174,7 @@ impl JobManager { /// Every sled known by cubby or by build, sorted by cubby first. pub fn versions(&self) -> Vec { let state = self.state.borrow(); + let linked = self.linked.borrow(); let mut sleds: BTreeSet<&BaseboardId> = state.versions().keys().collect(); sleds.extend(state.cubbies().values()); let mut rows: Vec = sleds @@ -179,6 +186,13 @@ impl JobManager { .find_map(|(cubby, b)| (b == baseboard).then_some(*cubby)), baseboard: baseboard.clone(), version: state.versions().get(baseboard).cloned(), + health: Some( + if *baseboard == self.own_baseboard || linked.contains(baseboard) { + SledHealth::Linked + } else { + SledHealth::Unlinked + }, + ), }) .collect(); rows.sort_by_key(|row| (row.cubby.is_none(), row.cubby)); diff --git a/server/tests/common/mod.rs b/server/tests/common/mod.rs index 7285d5cc..81a9c8db 100644 --- a/server/tests/common/mod.rs +++ b/server/tests/common/mod.rs @@ -22,12 +22,12 @@ use sprockets_tls::keys::{ AttestConfig, MeasurementConnectionPolicy, ResolveSetting, SprocketsConfig, }; use sprockets_tls_test_utils::{ - OutputFileExistsBehavior, alias_prefix, cert_path, certlist_path, generate_config, + OutputFileExistsBehavior, alias_prefix, cert_path, certlist_path, generate_config, platform_id, private_key_path, root_prefix, sprockets_auth_prefix, }; use sush_common::authn::{Challenge, ChallengeResponse, Identity, Nonce, RequestKey}; use sush_common::codephrases::Codephrase; -use sush_common::jobs::{JobId, JobMode, JobStartRequest, SessionId, SignedJob}; +use sush_common::jobs::{BaseboardId, JobId, JobMode, JobStartRequest, SessionId, SignedJob}; use sush_common::keys::{EphemeralKey, KeyType, Signer as _}; use sush_common::targets::Target; use sush_server::gossip::GossipConfig; @@ -99,6 +99,16 @@ pub fn corpus(dir: &Utf8PathBuf) -> CorpusSource { Arc::new(move || corpus.clone()) } +/// The baseboard the test PKI attests for `node`. +pub fn baseboard(node: usize) -> BaseboardId { + let id = platform_id(node); + let mut fields = id.split(':'); + BaseboardId { + part_number: fields.nth(1).unwrap().to_string(), + serial_number: fields.nth(1).unwrap().to_string(), + } +} + /// Dial ceiling for localhost handshakes. pub fn dial_timeout() -> Duration { Duration::from_secs(10) diff --git a/server/tests/distributed.rs b/server/tests/distributed.rs index 48d92220..9b54af25 100644 --- a/server/tests/distributed.rs +++ b/server/tests/distributed.rs @@ -22,7 +22,7 @@ use sush_common::jobs::{ JobId, JobOutputState, JobStatus, ProcessError, Session, SessionId, SessionSignerNonce, }; use sush_common::keys::pem_cert_chain; -use sush_common::targets::Cubbies; +use sush_common::targets::{Cubbies, SledHealth}; use sush_common::version::VersionInfo; use sush_server::bookmark::BookmarkSource; use sush_server::executor::PathIsolation; @@ -33,8 +33,8 @@ use sush_server::state::GossipUniverse; use sush_server::{JobManager, seed_gossip}; use common::{ - corpus, eventually, fake_identity, gossip_config, localhost, pki, sign_job, sprockets_config, - test_logger, + baseboard, corpus, eventually, fake_identity, gossip_config, localhost, pki, sign_job, + sprockets_config, test_logger, }; struct Sled { @@ -74,7 +74,7 @@ impl Sled { shutdown: &CancellationToken, ) -> Sled { let (peers, peers_rx) = watch::channel(BTreeSet::new()); - let (addr, universe) = spawn_gossip( + let (addr, universe, linked) = spawn_gossip( log, gossip_config(), sprockets_config(dir, identity), @@ -88,10 +88,9 @@ impl Sled { .await .unwrap(); let output = TempDir::with_prefix("sush-out-").unwrap(); - let baseboard = BaseboardId { - part_number: "sled".to_string(), - serial_number: identity.to_string(), - }; + // The manager's baseboard must be the one sprockets attests, + // as it is on a sled, or the health join can never match. + let baseboard = baseboard(identity); let (_cubbies, cubbies) = watch::channel(Cubbies::new()); let mgr = JobManager::new( log.clone(), @@ -100,6 +99,7 @@ impl Sled { baseboard.clone(), cubbies, universe.clone(), + linked, std::slice::from_ref(root_pem), shutdown.clone(), ) @@ -140,14 +140,15 @@ async fn jobs_gossip_between_sleds() { }) .await; - // Each sled learns the other's build. - eventually("versions gossip", 60, async || { + // Each sled learns the other's build, and sees it linked. + eventually("versions gossip", 120, async || { [&a, &b].iter().all(|sled| { let versions = sled.mgr.versions(); [&a.baseboard, &b.baseboard].iter().all(|baseboard| { versions.iter().any(|row| { row.baseboard == **baseboard && row.version.as_ref() == Some(&VersionInfo::current()) + && row.health == Some(SledHealth::Linked) }) }) }) @@ -362,10 +363,7 @@ async fn interrupted_jobs_get_stopped() { let job_id: JobId = "abandon-abandon-abandon-abandon-abandon-abandon-abandon-ability" .parse() .unwrap(); - let ghost = BaseboardId { - part_number: "sled".to_string(), - serial_number: "2".to_string(), - }; + let ghost = baseboard(2); a.universe.borrow().rumors.clone().send( Message::Event( ghost.clone(), diff --git a/server/tests/gossip.rs b/server/tests/gossip.rs index a6dcab5c..77784a27 100644 --- a/server/tests/gossip.rs +++ b/server/tests/gossip.rs @@ -16,16 +16,20 @@ use slog::Logger; use tokio::sync::watch; use tokio_util::sync::CancellationToken; +use sush_common::jobs::BaseboardId; use sush_server::bookmark::{BookmarkSource, SushBookmark}; -use sush_server::gossip::{Universe, spawn_gossip}; +use sush_server::gossip::{LinkedBaseboards, Universe, spawn_gossip}; -use common::{corpus, eventually, gossip_config, localhost, pki, sprockets_config, test_logger}; +use common::{ + baseboard, corpus, eventually, gossip_config, localhost, pki, sprockets_config, test_logger, +}; struct Node { addr: SocketAddrV6, initial: Network, peers: watch::Sender>, universe: watch::Receiver>, + linked: LinkedBaseboards, shutdown: CancellationToken, } @@ -40,7 +44,7 @@ impl Node { .into_rumors(); let initial = seed.network(); let (peers, peers_rx) = watch::channel(BTreeSet::new()); - let (addr, universe) = spawn_gossip( + let (addr, universe, linked) = spawn_gossip( log, gossip_config(), sprockets_config(dir, identity), @@ -58,6 +62,7 @@ impl Node { initial, peers, universe, + linked, shutdown, } } @@ -76,6 +81,10 @@ impl Node { .iter() .any(|(_, m)| m.as_str() == message) } + + fn linked(&self) -> BTreeSet { + self.linked.borrow().clone() + } } impl Drop for Node { @@ -179,3 +188,23 @@ async fn node_replacement_reconverges() { }) .await; } + +#[tokio::test] +async fn linked_follows_live_links() { + let (_tmp, dir) = pki("sush-gossip-", 2); + let log = test_logger("linked_follows_live_links"); + let a = Node::start(&log, &dir, 1).await; + let b = Node::start(&log, &dir, 2).await; + assert!(a.linked().is_empty()); + mesh(&[&a, &b]); + + // Both sides resolving proves the dialed and the accepted paths. + eventually("mutual attested links", 120, async || { + a.linked() == BTreeSet::from([baseboard(2)]) && b.linked() == BTreeSet::from([baseboard(1)]) + }) + .await; + + b.shutdown.cancel(); + drop(b); + eventually("dead peer unlinked", 120, async || a.linked().is_empty()).await; +} diff --git a/sush.json b/sush.json index da8853c1..cb044371 100644 --- a/sush.json +++ b/sush.json @@ -1717,8 +1717,34 @@ "signature" ] }, + "SledHealth": { + "description": "One sled's gossip link health, as the answering sled sees it. A silent death can lag `Linked` at TCP's pace.", + "oneOf": [ + { + "description": "Attested, holding a live gossip link.", + "type": "string", + "enum": [ + "linked" + ] + }, + { + "description": "Known by version or cubby, but no live link.", + "type": "string", + "enum": [ + "unlinked" + ] + }, + { + "description": "A state from a build newer than this one.", + "type": "string", + "enum": [ + "unknown" + ] + } + ] + }, "SledVersion": { - "description": "One sled's location and build.", + "description": "One sled's location, build, and health.", "type": "object", "properties": { "baseboard": { @@ -1730,6 +1756,15 @@ "format": "uint8", "minimum": 0 }, + "health": { + "nullable": true, + "default": null, + "allOf": [ + { + "$ref": "#/components/schemas/SledHealth" + } + ] + }, "version": { "nullable": true, "allOf": [ diff --git a/tests/src/manager_tests.rs b/tests/src/manager_tests.rs index 34a01308..7dbfa4e6 100644 --- a/tests/src/manager_tests.rs +++ b/tests/src/manager_tests.rs @@ -35,7 +35,7 @@ use sush_common::jobs::{ use sush_common::keys::{EphemeralKey, KeyError, KeyId, KeyType, Signer as _, pem_cert_chain}; use sush_common::targets::{Cubbies, Target}; use sush_server::bookmark::BookmarkSource; -use sush_server::gossip::{Universe, isolated}; +use sush_server::gossip::{Universe, isolated, lonely}; use sush_server::io::BATCH_OUTPUT_BUFFER_SIZE; use sush_server::messages::v0::{CertRequest, IdentityRequest, Message, Request, SessionRequest}; use sush_server::output::{JobOutputDir, OutputDirs}; @@ -580,6 +580,7 @@ async fn cubby_targets() { test_baseboard_id(), cubbies_rx, isolated(seed_gossip(&BookmarkSource::null()).await), + lonely(), &[root.cert().to_owned()], CancellationToken::new(), ) @@ -685,6 +686,7 @@ async fn root_certs_from_files() { test_baseboard_id(), no_cubbies(), isolated(seed_gossip(&BookmarkSource::null()).await), + lonely(), &[path], CancellationToken::new(), ) @@ -735,6 +737,7 @@ async fn bad_root_cert_files() { test_baseboard_id(), no_cubbies(), isolated(seed_gossip(&BookmarkSource::null()).await), + lonely(), &[path], CancellationToken::new(), ) @@ -765,6 +768,7 @@ async fn job_output_dir_moves() { test_baseboard_id(), no_cubbies(), isolated(seed_gossip(&BookmarkSource::null()).await), + lonely(), &[root.cert().to_owned()], CancellationToken::new(), ) @@ -866,6 +870,7 @@ async fn universe_swap() { test_baseboard_id(), no_cubbies(), universe_rx, + lonely(), &[root.cert().to_owned()], CancellationToken::new(), ) @@ -1014,6 +1019,7 @@ async fn cert_chain() { baseboard, no_cubbies(), gossip, + lonely(), &roots, shutdown, ) @@ -1788,6 +1794,7 @@ async fn hostile_imports_cannot_displace() { baseboard, no_cubbies(), isolated(seed), + lonely(), from_ref(&root_cert), shutdown, ) @@ -1933,6 +1940,7 @@ async fn homonym_issuer_resolves_to_true_parent() { baseboard, no_cubbies(), isolated(seed), + lonely(), from_ref(&root_cert), shutdown, ) diff --git a/tests/src/test_utils.rs b/tests/src/test_utils.rs index aea150d8..54c1bb97 100644 --- a/tests/src/test_utils.rs +++ b/tests/src/test_utils.rs @@ -31,7 +31,7 @@ use sush_common::keys::{EphemeralKey, KeyType, Signer}; use sush_common::targets::{Cubbies, Target}; use sush_server::bookmark::BookmarkSource; use sush_server::executor::PathIsolation; -use sush_server::gossip::isolated; +use sush_server::gossip::{isolated, lonely}; use sush_server::output::{JobOutputDir, JobOutputFileStream}; use sush_server::state::GossipNetwork; use sush_server::{JobError, JobManager, seed_gossip}; @@ -240,6 +240,7 @@ pub async fn manager_test_root_and_peer( test_baseboard_id(), no_cubbies(), gossip, + lonely(), &[root.cert().to_owned()], shutdown.clone(), )