diff --git a/.config/nextest.toml b/.config/nextest.toml index fef1880a35..9b11588bcb 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -36,7 +36,7 @@ test-group = "kubernetes-ha" # they kill members or gateways, or scale the pool. Never run them at once. # Test names, not binary(), for the reason above. [[profile.e2e-kubernetes.overrides]] -filter = "test(/^(losing_a_supervisor_pod_moves_its_sandboxes_without_restarting_them|scaling_a_supervisor_pool_up_and_down_keeps_its_sandboxes_running|pool_sessions_follow_their_supervisors_gateway)$/)" +filter = "test(/^(losing_a_supervisor_pod_moves_its_sandboxes_without_restarting_them|scaling_a_supervisor_pool_up_and_down_keeps_its_sandboxes_running|pool_sessions_follow_their_supervisors_gateway|supervisors_spread_evenly_across_gateways)$/)" test-group = "kubernetes-shared-supervisor-pool" # Scaling the pool runs two bounded Helm upgrades (up to 6 minutes each), waits @@ -51,6 +51,12 @@ slow-timeout = { period = "60s", terminate-after = 20 } filter = "test(/^pool_sessions_follow_their_supervisors_gateway$/)" slow-timeout = { period = "60s", terminate-after = 20 } +# One bounded rollout per supervisor and one for the gateway, each followed by +# a settle round. +[[profile.e2e-kubernetes.overrides]] +filter = "test(/^supervisors_spread_evenly_across_gateways$/)" +slow-timeout = { period = "60s", terminate-after = 20 } + # Relative to the profile store dir (`e2e/rust/target/nextest/e2e-kubernetes/`). [profile.e2e-kubernetes.junit] path = "../../../../../results/e2e-kubernetes.xml" diff --git a/crates/openshell-server/src/gateway_metrics.rs b/crates/openshell-server/src/gateway_metrics.rs index 0f55f42bbd..6e907696c7 100644 --- a/crates/openshell-server/src/gateway_metrics.rs +++ b/crates/openshell-server/src/gateway_metrics.rs @@ -23,6 +23,8 @@ use tonic::{Code, Status}; // Gauges pub const SUPERVISOR_SESSIONS: &str = "openshell_server_supervisor_sessions"; +pub const SUPERVISOR_HOSTS: &str = "openshell_server_supervisor_hosts"; +pub const SUPERVISOR_HOST_INSTANCES_MAX: &str = "openshell_server_supervisor_host_instances_max"; pub const RELAY_PENDING: &str = "openshell_server_relay_pending"; pub const RELAY_PENDING_CAPACITY: &str = "openshell_server_relay_pending_capacity"; pub const MUTATION_LOCK_CONNECTIONS_IN_USE: &str = @@ -262,6 +264,16 @@ pub fn describe_and_initialize(relay: RelayCapacity, mutation_lock_connections: Unit::Count, "Supervisor control sessions registered on this gateway replica." ); + describe_gauge!( + SUPERVISOR_HOSTS, + Unit::Count, + "Shared supervisors whose assignment streams this gateway replica serves." + ); + describe_gauge!( + SUPERVISOR_HOST_INSTANCES_MAX, + Unit::Count, + "Most sandbox instances placed on one shared supervisor this gateway replica serves." + ); describe_gauge!( RELAY_PENDING, Unit::Count, @@ -335,6 +347,8 @@ pub fn describe_and_initialize(relay: RelayCapacity, mutation_lock_connections: // `increment(0)` registers a series without overwriting a value recorded earlier. gauge!(SUPERVISOR_SESSIONS).increment(0.0); + gauge!(SUPERVISOR_HOSTS).increment(0.0); + gauge!(SUPERVISOR_HOST_INSTANCES_MAX).increment(0.0); gauge!(RELAY_PENDING).increment(0.0); gauge!(RELAY_PENDING_CAPACITY).set(count_as_f64(relay.per_replica)); for kind in RelayKind::ALL { @@ -433,6 +447,11 @@ impl Drop for GaugeSlot { } } +pub fn record_supervisor_hosts(hosts: usize, most_instances: usize) { + gauge!(SUPERVISOR_HOSTS).set(count_as_f64(hosts)); + gauge!(SUPERVISOR_HOST_INSTANCES_MAX).set(count_as_f64(most_instances)); +} + pub fn record_relay_rejected(reason: RelayRejection) { counter!(RELAY_REJECTED_TOTAL, LABEL_REASON => reason.label()).increment(1); } @@ -642,6 +661,8 @@ mod tests { ("openshell_server_mutation_lock_connections_capacity", 4), ("openshell_server_mutation_lock_connections_in_use", 0), ("openshell_server_supervisor_sessions", 0), + ("openshell_server_supervisor_hosts", 0), + ("openshell_server_supervisor_host_instances_max", 0), ("openshell_server_relay_pending", 0), ( "openshell_server_relay_rejected_total{reason=\"replica_capacity\"}", diff --git a/crates/openshell-server/src/gateway_ring.rs b/crates/openshell-server/src/gateway_ring.rs index a238c98951..b151a0e93c 100644 --- a/crates/openshell-server/src/gateway_ring.rs +++ b/crates/openshell-server/src/gateway_ring.rs @@ -84,6 +84,23 @@ impl GatewayRing { Some(self.points[index].1.as_str()) } + /// The first replica clockwise from `key` whose `load` is below `cap`, + /// or the owner of `key` when every replica is full. + pub fn first_below(&self, key: &str, cap: usize, load: impl Fn(&str) -> usize) -> Option<&str> { + let owner = self.owner_for(key)?; + let start = self + .points + .partition_point(|(candidate, _)| *candidate < hash_point(key)); + let mut seen = std::collections::HashSet::new(); + self.points[start..] + .iter() + .chain(&self.points[..start]) + .map(|(_, id)| id.as_str()) + .filter(|id| seen.insert(*id)) + .find(|id| load(id) < cap) + .or(Some(owner)) + } + /// The same ring with `replica_id` taken out. pub fn without(&self, replica_id: &str) -> Self { Self { @@ -250,4 +267,44 @@ mod tests { "adding one replica moved {moved} of {total} sandboxes" ); } + + #[test] + fn first_below_skips_full_replicas_and_falls_back_to_the_owner() { + let ring = GatewayRing::new(["gw-0", "gw-1"]); + let owner = ring.owner_for("sup-0").unwrap(); + let other = if owner == "gw-0" { "gw-1" } else { "gw-0" }; + + assert_eq!(ring.first_below("sup-0", 1, |_| 0), Some(owner)); + assert_eq!( + ring.first_below("sup-0", 1, |id| usize::from(id == owner)), + Some(other) + ); + assert_eq!(ring.first_below("sup-0", 1, |_| 1), Some(owner)); + assert_eq!(GatewayRing::default().first_below("sup-0", 1, |_| 0), None); + } + + /// Supervisors arriving one by one under a cap of their even share fill + /// every replica to within that cap. + #[test] + fn first_below_spreads_arrivals_evenly() { + for (replicas, supervisors) in [(2, 2), (3, 7), (4, 40)] { + let ids: Vec = (0..replicas).map(|i| format!("gw-{i}")).collect(); + let ring = GatewayRing::new(&ids); + let cap = usize::div_ceil(supervisors, replicas); + let mut load: HashMap = HashMap::new(); + for i in 0..supervisors { + let home = ring + .first_below(&format!("sup-{i}"), cap, |id| { + load.get(id).copied().unwrap_or(0) + }) + .unwrap() + .to_string(); + *load.entry(home).or_default() += 1; + } + assert!( + load.values().all(|count| *count <= cap) && load.len() == replicas, + "{supervisors} supervisors on {replicas} replicas: {load:?}" + ); + } + } } diff --git a/crates/openshell-server/src/grpc/supervisor_host.rs b/crates/openshell-server/src/grpc/supervisor_host.rs index 9760c4c97b..c6c6eed751 100644 --- a/crates/openshell-server/src/grpc/supervisor_host.rs +++ b/crates/openshell-server/src/grpc/supervisor_host.rs @@ -8,15 +8,21 @@ //! the handler sees a [`Principal::Supervisor`] and hands the stream to the //! controller the in-process compute driver installed. +use std::collections::HashMap; use std::pin::Pin; use std::sync::Arc; +use std::sync::atomic::Ordering; +use futures::StreamExt as _; use openshell_core::proto::supervisor_host::v1::supervisor_host_server::SupervisorHost; use openshell_core::proto::supervisor_host::v1::{HostGatewayMessage, HostMessage}; +use openshell_supervisor_control::MemberView; +use openshell_supervisor_control::controller::redirect; use tonic::{Request, Response, Status, Streaming}; use crate::ServerState; use crate::auth::principal::Principal; +use crate::gateway_ring::GatewayRing; /// gRPC path prefix of the supervisor host service. pub const SUPERVISOR_HOST_PATH_PREFIX: &str = "/openshell.supervisor.host.v1.SupervisorHost/"; @@ -65,10 +71,116 @@ impl SupervisorHost for SupervisorHostService { if *shutdown.borrow() { return Err(Status::unavailable("gateway is shutting down")); } - let outbound = controller.attach(identity, request.into_inner()).await?; - let outbound = futures::StreamExt::take_until(outbound, async move { - let _ = shutdown.wait_for(|shutdown| *shutdown).await; - }); + let state = self.state.clone(); + let (member_id, outbound) = controller + .attach(identity, request.into_inner(), |member_id| { + host_home( + &state, + &controller.pool_view().unwrap_or_default().members, + member_id, + ) + }) + .await?; + let handoff = futures::stream::once(async move { + let stopping = *state.supervisor_sessions.subscribe_shutdown().borrow(); + let members = controller.pool_view().unwrap_or_default().members; + stopping + .then(|| host_home(&state, &members, &member_id)) + .flatten() + .map(|peer_endpoint| Ok(redirect(peer_endpoint))) + }) + .filter_map(std::future::ready); + let outbound = outbound + .take_until(async move { + let _ = shutdown.wait_for(|shutdown| *shutdown).await; + }) + .chain(handoff); Ok(Response::new(Box::pin(outbound))) } } + +/// Peer endpoint of the replica that should serve the shared supervisor +/// `member_id` when that is not this replica. A stopping replica leaves +/// itself out. +fn host_home(state: &ServerState, members: &[MemberView], member_id: &str) -> Option { + if state.store.is_single_replica() { + return None; + } + let ring = state.gateway_ring.read().ok()?; + let ring = if state.gateway_shutting_down.load(Ordering::Acquire) { + ring.without(&state.replica_id) + } else { + ring.clone() + }; + let home = spread_home(&ring, members, member_id)?; + if home == state.replica_id { + return None; + } + state + .gateway_peers + .read() + .ok()? + .get(home) + .filter(|endpoint| !endpoint.is_empty()) + .cloned() +} + +/// The first replica clockwise from `member_id` on the gateway ring that +/// serves fewer than an even share of the pool's other supervisors. +fn spread_home<'a>( + ring: &'a GatewayRing, + members: &[MemberView], + member_id: &str, +) -> Option<&'a str> { + let mut load = HashMap::<&str, usize>::new(); + let mut supervisors = 1_usize; + for member in members + .iter() + .filter(|member| member.member_id != member_id && !member.leaving) + { + supervisors += 1; + *load.entry(member.gateway_replica.as_str()).or_default() += 1; + } + let cap = supervisors.div_ceil(ring.replica_count().max(1)); + ring.first_below(member_id, cap, |replica| { + load.get(replica).copied().unwrap_or(0) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn member(member_id: &str, gateway_replica: &str) -> MemberView { + MemberView { + member_id: member_id.to_string(), + process_id: String::new(), + runtime_address: String::new(), + leaving: false, + gateway_replica: gateway_replica.to_string(), + } + } + + #[test] + fn a_supervisor_goes_to_a_replica_below_its_even_share() { + let ring = GatewayRing::new(["gw-a", "gw-b"]); + let owner = ring.owner_for("sup-2").unwrap(); + let other = if owner == "gw-a" { "gw-b" } else { "gw-a" }; + + assert_eq!( + spread_home(&ring, &[member("sup-1", owner)], "sup-2"), + Some(other) + ); + assert_eq!( + spread_home(&ring, &[member("sup-1", other)], "sup-2"), + Some(owner) + ); + // Its own record and leaving members do not count against a replica. + let mut leaving = member("sup-3", owner); + leaving.leaving = true; + assert_eq!( + spread_home(&ring, &[member("sup-2", owner), leaving], "sup-2"), + Some(owner) + ); + } +} diff --git a/crates/openshell-server/src/supervisor_pool.rs b/crates/openshell-server/src/supervisor_pool.rs index 2ecc6b6f94..dbd50f0797 100644 --- a/crates/openshell-server/src/supervisor_pool.rs +++ b/crates/openshell-server/src/supervisor_pool.rs @@ -182,6 +182,7 @@ pub struct LiveMember { pub process_id: String, pub runtime_address: String, pub leaving: bool, + pub gateway_replica: String, } #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] @@ -1168,6 +1169,7 @@ impl SupervisorPool { process_id: payload.process_id, runtime_address: payload.runtime_address, leaving: payload.leaving, + gateway_replica: payload.gateway_replica, }); } members.sort_by(|left, right| left.member_id.cmp(&right.member_id)); @@ -2671,6 +2673,7 @@ impl SupervisorPool { process_id: payload.process_id, runtime_address: payload.runtime_address, leaving: payload.leaving, + gateway_replica: payload.gateway_replica, }); Ok(takeover_cause(owner, member.as_ref())) } @@ -2825,6 +2828,7 @@ impl SupervisorPool { process_id: member.process_id.clone(), runtime_address: member.runtime_address.clone(), leaving: member.leaving, + gateway_replica: member.gateway_replica.clone(), }) .collect(), owners: BTreeMap::new(), @@ -2868,6 +2872,19 @@ impl SupervisorPool { ); } lock(&self.minted).retain(|term, _| current_terms.contains(term)); + let mut placed: BTreeMap<&str, usize> = served + .iter() + .map(|(member, _)| (member.as_str(), 0)) + .collect(); + for owner in view.owners.values() { + if let Some(count) = placed.get_mut(owner.member_id.as_str()) { + *count += 1; + } + } + crate::gateway_metrics::record_supervisor_hosts( + placed.len(), + placed.values().copied().max().unwrap_or(0), + ); Some(view) } @@ -5519,6 +5536,7 @@ mod tests { process_id: process_id.to_string(), runtime_address: "10.0.0.1:5500".to_string(), leaving, + gateway_replica: String::new(), }; assert_eq!(takeover_cause(&owner, None), Some("expired")); assert_eq!( @@ -5543,6 +5561,7 @@ mod tests { process_id: format!("p-{id}"), runtime_address: "10.0.0.1:5500".to_string(), leaving, + gateway_replica: String::new(), }; let members = vec![member("m1", false), member("m2", false), member("m3", true)]; let ring = ring_of(&members); diff --git a/crates/openshell-supervisor-control/README.md b/crates/openshell-supervisor-control/README.md index 1cb7f0d115..9bc94c19c8 100644 --- a/crates/openshell-supervisor-control/README.md +++ b/crates/openshell-supervisor-control/README.md @@ -48,7 +48,10 @@ re-implementing the control contracts. `pool_timing::GATEWAY_SILENCE_TIMEOUT` of silence, and drops a member that stops reading as soon. `HostAccepted` carries this replica's peer endpoint (`set_peer_endpoint`); the host dials its instances' sandbox sessions - there, so each session runs on the replica serving its supervisor. + there, so each session runs on the replica serving its supervisor. A host + whose hello does not set `stay` is sent a `HostRedirect` instead when the + gateway places its member on another replica; the gateway picks the first + replica on its ring below an even share of supervisors. - `pool_timing`: the timing contract between pool hosts and the gateway: the 15 s membership TTL, the 2 s record refresh, 1 s heartbeats, host silence detection and bounded reconnect attempts, and `reconnect_budget`, the worst diff --git a/crates/openshell-supervisor-control/src/controller.rs b/crates/openshell-supervisor-control/src/controller.rs index 7b2abbd24e..f381f42cf2 100644 --- a/crates/openshell-supervisor-control/src/controller.rs +++ b/crates/openshell-supervisor-control/src/controller.rs @@ -40,7 +40,7 @@ use tonic::{Status, Streaming}; use crate::proto::{ AssignmentSnapshot, HostAccepted, HostGatewayHeartbeat, HostGatewayMessage, HostHello, - HostMessage, HostStatus, PoolMember, RouteOwner, SandboxInstanceAssignment, + HostMessage, HostRedirect, HostStatus, PoolMember, RouteOwner, SandboxInstanceAssignment, SandboxInstanceStatus, SandboxInstanceTakeover, host_gateway_message, host_message, }; @@ -167,6 +167,8 @@ pub struct MemberView { pub runtime_address: String, /// The member announced it is leaving the pool. pub leaving: bool, + /// Gateway replica serving the member's stream. + pub gateway_replica: String, } /// The gateway's durable view of a supervisor pool, replaced as a whole by @@ -819,12 +821,15 @@ impl SupervisorHostController { } /// Serve one authenticated host stream. Reads the hello, supersedes any - /// session it replaces, and returns the gateway-to-host stream. + /// session it replaces, and returns the host's member ID and the + /// gateway-to-host stream. A host that does not ask to stay is redirected, + /// without a session, to the peer endpoint `home` names for its member. pub async fn attach( &self, identity: HostIdentity, mut inbound: Streaming, - ) -> Result>, Status> { + home: impl FnOnce(&str) -> Option, + ) -> Result<(String, ReceiverStream>), Status> { let hello = match tokio::time::timeout(HELLO_TIMEOUT, inbound.next()).await { Ok(Some(Ok(HostMessage { payload: Some(host_message::Payload::Hello(hello)), @@ -855,8 +860,16 @@ impl SupervisorHostController { return Err(Status::failed_precondition(MISSING_MEMBER)); }; validate_member(member).map_err(Status::invalid_argument)?; - let heartbeat = POOL_HEARTBEAT_INTERVAL; + let member_id = member.member_id.clone(); let (tx, rx) = mpsc::channel(OUTBOUND_QUEUE); + if !hello.stay + && let Some(peer_endpoint) = home(&member_id) + { + tracing::info!(member_id, peer_endpoint, "redirecting supervisor host"); + let _ = tx.try_send(Ok(redirect(peer_endpoint))); + return Ok((member_id, ReceiverStream::new(rx))); + } + let heartbeat = POOL_HEARTBEAT_INTERVAL; let expires_at = identity.expires_at; let (session_id, superseded) = self.begin_session(identity, &hello)?; let peer_endpoint = lock(&self.inner.peer_endpoint).clone(); @@ -879,7 +892,7 @@ impl SupervisorHostController { .await; controller.end_session(session_id); }); - Ok(ReceiverStream::new(rx)) + Ok((member_id, ReceiverStream::new(rx))) } /// Start a session for `hello`, superseding the same member's session. A @@ -1276,6 +1289,16 @@ const MISSING_MEMBER: &str = "the shared supervisor did not name its pool member OPENSHELL_SUPERVISOR_POOL_MEMBER_ID to a stable member ID and \ OPENSHELL_SUPERVISOR_POOL_ADVERTISE_IP to an IP reachable from sandbox runtimes"; +/// Sends a host to another gateway replica. +#[must_use] +pub fn redirect(peer_endpoint: String) -> HostGatewayMessage { + HostGatewayMessage { + payload: Some(host_gateway_message::Payload::Redirect(HostRedirect { + peer_endpoint, + })), + } +} + /// Check a pool member's identity, naming the setting that is wrong. fn validate_member(member: &PoolMember) -> Result<(), &'static str> { let valid = |value: &str| { @@ -1445,6 +1468,7 @@ mod tests { member_id: member.to_string(), runtime_address: format!("10.0.0.{}:5500", member.len()), }), + stay: false, } } @@ -1469,6 +1493,7 @@ mod tests { process_id: process.to_string(), runtime_address: format!("10.0.0.{}:5500", member.len()), leaving: false, + gateway_replica: String::new(), } } @@ -1963,6 +1988,7 @@ mod tests { instance_id: "p1".to_string(), connection_epoch: 1, member: None, + stay: false, }, ) .unwrap_err(); diff --git a/crates/openshell-supervisor/src/shared_host.rs b/crates/openshell-supervisor/src/shared_host.rs index 9e6ae7d241..406e79cc79 100644 --- a/crates/openshell-supervisor/src/shared_host.rs +++ b/crates/openshell-supervisor/src/shared_host.rs @@ -1009,6 +1009,8 @@ enum StreamEnd { accepted: bool, redial: Option, }, + /// The gateway sent the host to another replica; dial it at once. + Redirected { peer_endpoint: String }, } /// What a gateway sends a host on its assignment stream. @@ -1095,6 +1097,11 @@ async fn attach_with( // unanswered ones is still quick. let mut unanswered = 0_u32; let mut redial = None; + // `redirected`: this attempt follows a redirect. `stay`: the attempt after + // a failed redirect stays wherever the Service lands it, so a stale ring + // cannot send it back to the replica it could not reach. + let mut redirected = false; + let mut stay = false; tokio::pin!(shutdown); loop { connection_epoch += 1; @@ -1105,16 +1112,25 @@ async fn attach_with( member_id: host.pool.member_id.clone(), runtime_address: host.pool.runtime_address.to_string(), }), + stay: stay || redial.is_some(), }; let attempt_timeout = timings.attempt_timeout(unanswered); let end = tokio::select! { end = run_stream(&host, connector.as_ref(), &reconciler, hello, &timings, attempt_timeout, redial.as_deref()) => end, () = &mut shutdown => return reconciler.shut_down().await, }; - let StreamEnd::Lost { - accepted, - redial: next, - } = end; + let (accepted, next) = match end { + StreamEnd::Redirected { peer_endpoint } if !redirected => { + tracing::info!(%peer_endpoint, "gateway redirected the supervisor host"); + redial = Some(peer_endpoint); + redirected = true; + continue; + } + StreamEnd::Redirected { .. } => (false, None), + StreamEnd::Lost { accepted, redial } => (accepted, redial), + }; + stay = redirected && !accepted; + redirected = false; redial = next; if accepted { failed = 0; @@ -1481,6 +1497,10 @@ async fn run_stream( tracing::error!(reason = %rejected.reason, "gateway rejected the supervisor host"); break lost(accepted); } + Some(host_gateway_message::Payload::Redirect(redirect)) if !redirect.peer_endpoint.is_empty() => { + break StreamEnd::Redirected { peer_endpoint: redirect.peer_endpoint }; + } + Some(host_gateway_message::Payload::Redirect(_)) => break lost(accepted), Some(host_gateway_message::Payload::Assignments(snapshot)) => { if !accepted { tracing::warn!("gateway sent assignments before accepting the host"); @@ -2317,6 +2337,47 @@ mod tests { assert_eq!(next.endpoint, None); } + fn hello_stays(message: Option) -> bool { + match message { + Some(HostMessage { + payload: Some(host_message::Payload::Hello(hello)), + }) => hello.stay, + other => panic!("expected a hello, got {other:?}"), + } + } + + /// A redirected host dials the named replica at once and asks to stay; + /// when that replica does not accept it, the host asks the replica the + /// Service picks to keep it. + #[tokio::test(start_paused = true)] + async fn a_redirected_host_dials_the_replica_at_once_and_stays() { + let (_host, _gateway, mut streams, _root) = + attach_pool_host(vec![Attempt::Now, Attempt::Now, Attempt::Now, Attempt::Now]); + let mut first = streams.recv().await.unwrap(); + assert!(!hello_stays(first.outbound.recv().await)); + first + .inbound + .send(Ok(openshell_supervisor_control::controller::redirect( + "https://gw-b:8443".to_string(), + ))) + .await + .unwrap(); + + let mut redirected = streams.recv().await.unwrap(); + assert_eq!(redirected.endpoint.as_deref(), Some("https://gw-b:8443")); + assert_eq!(redirected.at, first.at); + assert!(hello_stays(redirected.outbound.recv().await)); + drop(redirected); + + let mut fallback = streams.recv().await.unwrap(); + assert_eq!(fallback.endpoint, None); + assert!(hello_stays(fallback.outbound.recv().await)); + drop(fallback); + + let mut next = streams.recv().await.unwrap(); + assert!(!hello_stays(next.outbound.recv().await)); + } + /// A replica that keeps writing but stops reading fills the host's /// outbound queue; the blocked send gives up within the silence budget /// rather than the ordinary send bound. diff --git a/docs/kubernetes/high-availability.mdx b/docs/kubernetes/high-availability.mdx index c21150f762..de3cd4a886 100644 --- a/docs/kubernetes/high-availability.mdx +++ b/docs/kubernetes/high-availability.mdx @@ -195,6 +195,14 @@ A sandbox on a shared supervisor keeps its session on the gateway replica that serves its supervisor, and reconnects there when the supervisor moves to another replica. +Shared supervisors spread across gateway replicas. A supervisor that starts, +restarts, or loses its replica goes to the first replica on the gateway ring +that serves fewer than its even share of supervisors, and a stopping replica +hands its supervisors over the same way. A supervisor that reconnects after a +token refresh or a short interruption stays on its replica, and supervisors +never move when a replica joins, so a replica added without a rollout fills +only as supervisors restart. + Established streams do not move between gateway replicas. An interrupted exec, forwarding, or service stream must reconnect. File sync retries with a new SSH session instead of resuming the interrupted byte stream. Rolling updates can @@ -212,6 +220,7 @@ configuration, and access control. | Signal | Metrics | Use | |---|---|---| | Sessions per replica | `openshell_server_supervisor_sessions` | Placement and skew. Do not scale on it. | +| Shared supervisors per replica | `openshell_server_supervisor_hosts`, `openshell_server_supervisor_host_instances_max` | Supervisor spread across replicas, and the busiest supervisor's sandbox count against the 256 per-supervisor limit. | | Relay utilization | `openshell_server_relay_pending`, `openshell_server_relay_pending_capacity` | Graph it. It is transient and usually near 0, so alert on rejections instead. | | Relay rejections | `openshell_server_relay_rejected_total` | Alert on any increase. | | Relay claims | `openshell_server_relay_claim_duration_seconds`, `openshell_server_relay_expired_total` | Supervisor connect-back time for exec, SSH, forwarding, and service traffic, including relays requested through peers. A high 99th percentile while rejections stay flat points at the supervisor or its node, not at relay capacity. Relays not claimed within 10 seconds are missing from the histogram and count as expired, so alert on any increase. | diff --git a/docs/observability/gateway-metrics.mdx b/docs/observability/gateway-metrics.mdx index 24ec1b0bd9..43c659bdbf 100644 --- a/docs/observability/gateway-metrics.mdx +++ b/docs/observability/gateway-metrics.mdx @@ -174,6 +174,8 @@ Supervisor sessions: | Metric | Type | Labels | Description | |---|---|---|---| | `openshell_server_supervisor_sessions` | Gauge | None | Supervisor control sessions held by this replica. After a rollout settles, the sum across pods equals the number of connected sandboxes. | +| `openshell_server_supervisor_hosts` | Gauge | None | Shared supervisors whose assignment streams this replica serves. | +| `openshell_server_supervisor_host_instances_max` | Gauge | None | Most sandbox instances placed on one shared supervisor this replica serves. | Relays: diff --git a/e2e/rust/tests/kubernetes_shared_supervisor_ha.rs b/e2e/rust/tests/kubernetes_shared_supervisor_ha.rs index 36b16cf698..a690f608a6 100644 --- a/e2e/rust/tests/kubernetes_shared_supervisor_ha.rs +++ b/e2e/rust/tests/kubernetes_shared_supervisor_ha.rs @@ -17,7 +17,8 @@ //! The pool sits behind two gateway replicas sharing a database. Every pool //! sandbox's session runs on the replica serving its supervisor at start and //! after a gateway crash, a gateway rolling restart, and a supervisor crash; -//! exec reaches every sandbox after each step. +//! exec reaches every sandbox after each step. Supervisors that restart, and +//! those of a gateway that stops, spread evenly across the gateways. mod shared_supervisor; @@ -421,7 +422,7 @@ struct Branch { session_gateway: String, } -async fn tree(routes: &[String]) -> Vec { +async fn psql(query: &str) -> String { kubectl(&[ "-n", &release_namespace(), @@ -437,23 +438,28 @@ async fn tree(routes: &[String]) -> Vec { "-F", "|", "-c", - TREE, + query, ]) .await - .lines() - .filter_map(|row| { - let mut fields = row.split('|'); - let route = fields.next()?; - if !routes.iter().any(|ours| ours == route) { - return None; - } - Some(Branch { - member: fields.next()?.to_string(), - supervisor_gateway: fields.next()?.to_string(), - session_gateway: fields.next()?.to_string(), +} + +async fn tree(routes: &[String]) -> Vec { + psql(TREE) + .await + .lines() + .filter_map(|row| { + let mut fields = row.split('|'); + let route = fields.next()?; + if !routes.iter().any(|ours| ours == route) { + return None; + } + Some(Branch { + member: fields.next()?.to_string(), + supervisor_gateway: fields.next()?.to_string(), + session_gateway: fields.next()?.to_string(), + }) }) - }) - .collect() + .collect() } /// Wait until the session of every sandbox on `routes` runs on its @@ -646,6 +652,97 @@ async fn force_delete(pod: &str) { .await; } +const HOMES: &str = "\ +SELECT p->>'member_id', p->>'gateway_replica' +FROM (SELECT convert_from(payload, 'UTF8')::json AS p FROM objects + WHERE object_type = 'supervisor_pool_member') pm"; + +/// Wait until every live supervisor streams to a live gateway and no gateway +/// serves more than its even share. +async fn assert_spread(step: &str) { + let mut last = std::collections::BTreeMap::::new(); + let settled = tokio::time::timeout(SETTLE_DEADLINE, async { + loop { + let members = supervisor_pods().await; + let gateways: Vec = kubectl_json(&[ + "-n", + &release_namespace(), + "get", + "pods", + "-l", + GATEWAY_SELECTOR, + "-o", + "json", + ]) + .await["items"] + .as_array() + .expect("Pod items") + .iter() + .filter(|pod| pod["metadata"]["deletionTimestamp"].is_null()) + .filter_map(|pod| pod["metadata"]["name"].as_str().map(str::to_string)) + .collect(); + last = psql(HOMES) + .await + .lines() + .filter_map(|row| row.split_once('|')) + .filter(|(member, _)| members.iter().any(|live| live == member)) + .map(|(member, gateway)| (member.to_string(), gateway.to_string())) + .collect(); + let cap = members.len().div_ceil(gateways.len().max(1)); + if last.len() == members.len() + && gateways + .iter() + .all(|gateway| last.values().filter(|home| *home == gateway).count() <= cap) + && last.values().all(|home| gateways.contains(home)) + { + return; + } + tokio::time::sleep(Duration::from_secs(2)).await; + } + }) + .await; + assert!( + settled.is_ok(), + "{step}: supervisors did not spread across gateways within {SETTLE_DEADLINE:?}: {last:#?}" + ); +} + +#[tokio::test] +async fn supervisors_spread_evenly_across_gateways() { + let gateway = gateway_workload().await; + let supervisor = format!("deployment.apps/{}-supervisor", release()); + assert_eq!( + gateway_pods().await.len(), + 2, + "the lane runs two gateway replicas" + ); + // A supervisor that restarts picks the first gateway along the ring + // below its even share, so restarting each in turn spreads them. + let members = supervisor_pods().await; + for pod in &members { + kubectl(&["-n", &release_namespace(), "delete", "pod", pod]).await; + let replaced = tokio::time::timeout(SETTLE_DEADLINE, async { + loop { + let live = supervisor_pods().await; + if live.len() == members.len() && !live.contains(pod) { + return live; + } + tokio::time::sleep(Duration::from_secs(2)).await; + } + }) + .await + .unwrap_or_else(|_| panic!("{pod} was not replaced within {SETTLE_DEADLINE:?}")); + wait_until_attached(&replaced).await; + } + rollout("status", &supervisor).await; + assert_spread("supervisor restarts").await; + + // A stopping gateway hands its supervisors to the others the same way. + kubectl(&["-n", &release_namespace(), "rollout", "restart", &gateway]).await; + rollout("status", &gateway).await; + assert_spread("gateway rolling restart").await; +} + #[tokio::test] async fn pool_sessions_follow_their_supervisors_gateway() { let gateway = gateway_workload().await; diff --git a/proto/supervisor_host.proto b/proto/supervisor_host.proto index 862a5e3efa..a56c1f5784 100644 --- a/proto/supervisor_host.proto +++ b/proto/supervisor_host.proto @@ -50,6 +50,7 @@ message HostGatewayMessage { HostRejected rejected = 2; AssignmentSnapshot assignments = 3; HostGatewayHeartbeat heartbeat = 4; + HostRedirect redirect = 5; } } @@ -63,6 +64,9 @@ message HostHello { uint64 connection_epoch = 2; // Present when the host is a member of a supervisor pool. PoolMember member = 3; + // The host asks to be served by the replica it dialed: it is following a + // redirect or redialing the replica that last accepted it. + bool stay = 4; } // One member of a supervisor pool. @@ -90,6 +94,12 @@ message HostAccepted { string peer_endpoint = 3; } +// Another gateway replica should serve the host, which dials it at once with +// HostHello.stay set. The stream ends after this message. +message HostRedirect { + string peer_endpoint = 1; +} + // The gateway refuses the host; the stream ends after this message. message HostRejected { string reason = 1;