diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 2b168f26df..f0d3eefaa1 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -1077,6 +1077,14 @@ impl ComputeRuntime { self.supervisor_host.as_ref() } + /// Whether this gateway places supervisor sessions by the placement ring. A gateway + /// coordinating a shared supervisor does not: the host moves those sessions, so a + /// redirected hello is not an off-ring admission there and the counter is not exported. + #[must_use] + pub fn places_supervisor_sessions_by_ring(&self) -> bool { + self.supervisor_host().is_none() + } + pub(crate) fn configure_ssh_identities( &self, credentials: crate::credentials::CredentialRuntime, diff --git a/crates/openshell-server/src/gateway_members.rs b/crates/openshell-server/src/gateway_members.rs index b70bf5f82f..cafeecd1a8 100644 --- a/crates/openshell-server/src/gateway_members.rs +++ b/crates/openshell-server/src/gateway_members.rs @@ -8,6 +8,7 @@ //! kept in the store rather than read from Kubernetes so it works for any //! multi-replica deployment and adds no API server load. +use crate::gateway_metrics; use crate::gateway_ring::GatewayRing; use crate::persistence::{PersistenceError, PersistenceResult, Store, WriteCondition}; use openshell_core::time::now_ms; @@ -186,6 +187,10 @@ pub fn spawn_membership_worker( let index = GatewayMemberIndex::new(state.store.clone(), MEMBER_TTL); let replica_id = state.replica_id.clone(); + // Export the ring's size now (empty until the first refresh), so only + // replicas that join the ring carry the gauge, and they carry it from the + // start. + publish_ring(&state, GatewayRing::clone); Some(tokio::spawn(async move { let mut ticker = tokio::time::interval(interval); @@ -216,9 +221,7 @@ pub fn spawn_membership_worker( .into_iter() .map(|member| (member.replica_id, member.peer_endpoint)) .collect(); - if let Ok(mut slot) = state.gateway_ring.write() { - *slot = ring; - } + publish_ring(&state, |_| ring); if let Ok(mut slot) = state.gateway_peers.write() { *slot = peers; } @@ -249,24 +252,18 @@ async fn leave_ring(state: &crate::ServerState, index: &GatewayMemberIndex, repl .into_iter() .map(|member| (member.replica_id, member.peer_endpoint)) .collect(); - if let Ok(mut slot) = state.gateway_ring.write() { - *slot = ring.without(replica_id); - } + publish_ring(state, |_| ring.without(replica_id)); if let Ok(mut slot) = state.gateway_peers.write() { *slot = peers; } } Ok(Err(error)) => { tracing::warn!(%error, "gateway membership: shutdown ring refresh failed"); - if let Ok(mut slot) = state.gateway_ring.write() { - *slot = slot.without(replica_id); - } + publish_ring(state, |ring| ring.without(replica_id)); } Err(_) => { tracing::warn!("gateway membership: shutdown ring refresh timed out"); - if let Ok(mut slot) = state.gateway_ring.write() { - *slot = slot.without(replica_id); - } + publish_ring(state, |ring| ring.without(replica_id)); } } match tokio::time::timeout(SHUTDOWN_STORE_TIMEOUT, index.deregister(replica_id)).await { @@ -280,14 +277,26 @@ fn clear_ring_if_stale(state: &crate::ServerState, last_refreshed: tokio::time:: if last_refreshed.elapsed() <= MEMBER_TTL { return; } - if let Ok(mut slot) = state.gateway_ring.write() { - *slot = GatewayRing::default(); - } + publish_ring(state, |_| GatewayRing::default()); if let Ok(mut slot) = state.gateway_peers.write() { slot.clear(); } } +/// Replace this replica's placement ring with `next(current ring)` and record +/// its size as `openshell_server_gateway_ring_members` under the same write +/// lock. +/// +/// This is the only sanctioned non-test writer of `ServerState::gateway_ring`: +/// writing the slot anywhere else lets the gauge drift from the ring. A +/// poisoned lock leaves both unchanged. +fn publish_ring(state: &crate::ServerState, next: impl FnOnce(&GatewayRing) -> GatewayRing) { + if let Ok(mut slot) = state.gateway_ring.write() { + *slot = next(&slot); + gateway_metrics::record_gateway_ring_members(slot.replica_count()); + } +} + #[cfg(test)] mod tests { use super::*; @@ -314,6 +323,7 @@ mod tests { #[tokio::test] async fn leaving_points_ring_at_new_replicas_and_not_departed_ones() { + let metrics = gateway_metrics::MetricsCapture::install(); let state = crate::grpc::test_support::test_server_state().await; let index = GatewayMemberIndex::new(state.store.clone(), MEMBER_TTL); index.register("gw-old", "https://gw-old").await.unwrap(); @@ -331,6 +341,10 @@ mod tests { let ring = state.gateway_ring.read().unwrap().clone(); assert_eq!(ring.owner_for("sb-1"), Some("gw-new")); + assert_eq!( + metrics.value(gateway_metrics::GATEWAY_RING_MEMBERS), + Some(1) + ); assert_eq!( state .gateway_peers @@ -350,6 +364,222 @@ mod tests { assert_eq!(remaining, vec!["gw-new"]); } + #[tokio::test] + async fn leaving_without_a_membership_read_keeps_the_cached_ring_without_this_replica() { + let metrics = gateway_metrics::MetricsCapture::install(); + let state = crate::grpc::test_support::test_server_state().await; + publish_ring(&state, |_| GatewayRing::new(["gw-a", "gw-stopping"])); + // Only the index's store is closed, so the shutdown membership read + // fails while the state's store stays open. + let store = Arc::new(crate::persistence::test_store().await); + store.close().await; + let index = GatewayMemberIndex::new(store, MEMBER_TTL); + + leave_ring(&state, &index, "gw-stopping").await; + + let ring = state.gateway_ring.read().unwrap().clone(); + assert_eq!(ring.owner_for("sb-1"), Some("gw-a")); + assert_eq!( + metrics.value(gateway_metrics::GATEWAY_RING_MEMBERS), + Some(1) + ); + } + + #[tokio::test] + async fn leaving_after_a_timed_out_membership_read_keeps_the_cached_ring_without_this_replica() + { + let metrics = gateway_metrics::MetricsCapture::install(); + let state = crate::grpc::test_support::test_server_state().await; + publish_ring(&state, |_| GatewayRing::new(["gw-a", "gw-stopping"])); + // The listener accepts TCP but never answers the PostgreSQL handshake, so + // the shutdown read hangs past SHUTDOWN_STORE_TIMEOUT instead of failing. + let silent = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!( + "postgres://openshell@127.0.0.1:{}/openshell", + silent.local_addr().unwrap().port() + ); + let index = GatewayMemberIndex::new( + Arc::new(Store::Postgres( + crate::persistence::PostgresStore::connect_lazy_for_tests(&url, 1), + )), + MEMBER_TTL, + ); + + // The paused clock jumps to each timeout while the socket stays silent. + tokio::time::pause(); + let started = tokio::time::Instant::now(); + leave_ring(&state, &index, "gw-stopping").await; + // Both the read and the deregister must hit their timeouts; a read that + // failed fast would take the Ok(Err) branch and finish sooner. + assert!( + started.elapsed() >= SHUTDOWN_STORE_TIMEOUT * 2, + "the membership read must time out, not fail fast" + ); + + let ring = state.gateway_ring.read().unwrap().clone(); + assert_eq!(ring.replica_count(), 1); + assert_eq!(ring.owner_for("sb-1"), Some("gw-a")); + assert_eq!( + metrics.value(gateway_metrics::GATEWAY_RING_MEMBERS), + Some(1) + ); + drop(silent); + } + + #[tokio::test] + async fn stale_ring_is_cleared_only_after_the_member_ttl() { + let metrics = gateway_metrics::MetricsCapture::install(); + let state = crate::grpc::test_support::test_server_state().await; + publish_ring(&state, |_| GatewayRing::new(["gw-a", "gw-b"])); + *state.gateway_peers.write().unwrap() = HashMap::from([ + ("gw-a".to_string(), "https://gw-a".to_string()), + ("gw-b".to_string(), "https://gw-b".to_string()), + ]); + assert_eq!( + metrics.value(gateway_metrics::GATEWAY_RING_MEMBERS), + Some(2) + ); + + clear_ring_if_stale(&state, tokio::time::Instant::now()); + assert_eq!(state.gateway_ring.read().unwrap().replica_count(), 2); + assert_eq!(state.gateway_peers.read().unwrap().len(), 2); + assert_eq!( + metrics.value(gateway_metrics::GATEWAY_RING_MEMBERS), + Some(2) + ); + + let last_refreshed = tokio::time::Instant::now() + .checked_sub(MEMBER_TTL + Duration::from_millis(1)) + .expect("the monotonic clock is past one member TTL"); + clear_ring_if_stale(&state, last_refreshed); + assert!(state.gateway_ring.read().unwrap().is_empty()); + assert!(state.gateway_peers.read().unwrap().is_empty()); + assert_eq!( + metrics.value(gateway_metrics::GATEWAY_RING_MEMBERS), + Some(0) + ); + } + + #[tokio::test] + async fn poisoned_ring_lock_publishes_no_ring_size() { + let metrics = gateway_metrics::MetricsCapture::install(); + let state = crate::grpc::test_support::test_server_state().await; + let ring = Arc::clone(&state.gateway_ring); + let poisoned = std::thread::spawn(move || { + std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + let _guard = ring.write().unwrap(); + // `resume_unwind` drops the guard and poisons the lock without + // running the panic hook, keeping the expected panic out of + // the test output. + std::panic::resume_unwind(Box::new("poison the placement ring lock")); + })) + }) + .join() + .expect("the poisoning thread should catch its own unwind"); + assert!(poisoned.is_err()); + assert!(state.gateway_ring.is_poisoned()); + + publish_ring(&state, |_| GatewayRing::new(["gw-a", "gw-b"])); + + assert_eq!(metrics.value(gateway_metrics::GATEWAY_RING_MEMBERS), None); + } + + #[tokio::test] + async fn single_replica_store_neither_joins_the_ring_nor_exports_its_size() { + let metrics = gateway_metrics::MetricsCapture::install(); + let mut state = crate::grpc::test_support::test_server_state().await; + // SQLite keeps the replica out even with a peer endpoint. + Arc::get_mut(&mut state) + .expect("test state is not shared yet") + .peer_endpoint = Some("http://gw-self:8081".to_string()); + let (_shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false); + + assert!(spawn_membership_worker(state, MEMBER_REFRESH_INTERVAL, shutdown_rx).is_none()); + assert_eq!(metrics.value(gateway_metrics::GATEWAY_RING_MEMBERS), None); + } + + #[tokio::test] + #[ignore = "requires a disposable PostgreSQL database in OPENSHELL_TEST_POSTGRES_URL; run mise run test:rust:postgres"] + async fn postgres_replica_without_peer_endpoint_neither_joins_the_ring_nor_exports_its_size() { + let metrics = gateway_metrics::MetricsCapture::install(); + let schema = + crate::persistence::test_postgres::TestSchema::create("membership_no_peer").await; + let mut state = + crate::grpc::test_support::test_server_state_for_store(schema.connect_store().await) + .await; + // peer_endpoint is env-derived; pin it so the test does not depend on the environment. + Arc::get_mut(&mut state) + .expect("test state is not shared yet") + .peer_endpoint = None; + let (_shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false); + + assert!( + spawn_membership_worker(state.clone(), MEMBER_REFRESH_INTERVAL, shutdown_rx).is_none() + ); + assert_eq!(metrics.value(gateway_metrics::GATEWAY_RING_MEMBERS), None); + assert!( + !metrics + .render() + .contains(gateway_metrics::GATEWAY_RING_MEMBERS) + ); + + state.store.close().await; + schema.drop_schema().await; + } + + #[tokio::test] + #[ignore = "requires a disposable PostgreSQL database in OPENSHELL_TEST_POSTGRES_URL; run mise run test:rust:postgres"] + async fn postgres_membership_worker_exports_ring_size() { + let metrics = gateway_metrics::MetricsCapture::install(); + let schema = + crate::persistence::test_postgres::TestSchema::create("membership_worker").await; + let mut state = + crate::grpc::test_support::test_server_state_for_store(schema.connect_store().await) + .await; + { + let state = Arc::get_mut(&mut state).expect("test state is not shared yet"); + state.replica_id = "gw-self".to_string(); + state.peer_endpoint = Some("http://gw-self:8081".to_string()); + } + GatewayMemberIndex::new(state.store.clone(), MEMBER_TTL) + .register("gw-other", "http://gw-other:8081") + .await + .unwrap(); + let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false); + + let worker = spawn_membership_worker(state.clone(), Duration::from_millis(50), shutdown_rx) + .expect("a PostgreSQL replica with a peer endpoint joins the ring"); + assert_eq!( + metrics.value(gateway_metrics::GATEWAY_RING_MEMBERS), + Some(0) + ); + + tokio::time::timeout(Duration::from_secs(5), async { + while metrics.value(gateway_metrics::GATEWAY_RING_MEMBERS) != Some(2) { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("the first refresh publishes both replicas"); + assert_eq!(state.gateway_ring.read().unwrap().replica_count(), 2); + + // Leaving re-reads membership and keeps only the other replica. + shutdown_tx.send(true).unwrap(); + tokio::time::timeout(SHUTDOWN_MEMBERSHIP_TIMEOUT, worker) + .await + .expect("the worker leaves the ring") + .expect("membership worker task"); + let ring = state.gateway_ring.read().unwrap().clone(); + assert_eq!(ring.owner_for("sb-1"), Some("gw-other")); + assert_eq!( + metrics.value(gateway_metrics::GATEWAY_RING_MEMBERS), + Some(1) + ); + + state.store.close().await; + schema.drop_schema().await; + } + #[tokio::test] async fn register_is_idempotent_and_updates_the_endpoint() { let index = test_index(MEMBER_TTL).await; diff --git a/crates/openshell-server/src/gateway_metrics.rs b/crates/openshell-server/src/gateway_metrics.rs index 0f55f42bbd..12b34c7fd5 100644 --- a/crates/openshell-server/src/gateway_metrics.rs +++ b/crates/openshell-server/src/gateway_metrics.rs @@ -29,6 +29,7 @@ pub const MUTATION_LOCK_CONNECTIONS_IN_USE: &str = "openshell_server_mutation_lock_connections_in_use"; pub const MUTATION_LOCK_CONNECTIONS_CAPACITY: &str = "openshell_server_mutation_lock_connections_capacity"; +pub const GATEWAY_RING_MEMBERS: &str = "openshell_server_gateway_ring_members"; // Counters pub const RELAY_REJECTED_TOTAL: &str = "openshell_server_relay_rejected_total"; pub const RELAY_EXPIRED_TOTAL: &str = "openshell_server_relay_expired_total"; @@ -37,6 +38,10 @@ pub const MUTATION_LOCK_TIMEOUTS_TOTAL: &str = "openshell_server_mutation_lock_t pub const MUTATION_LOCK_ERRORS_TOTAL: &str = "openshell_server_mutation_lock_errors_total"; pub const PROVISIONING_DEADLINE_SKIPS_TOTAL: &str = "openshell_server_provisioning_deadline_skips_total"; +pub const SUPERVISOR_SESSION_REDIRECTS_TOTAL: &str = + "openshell_server_supervisor_session_redirects_total"; +pub const SUPERVISOR_SESSION_OFF_RING_ADMISSIONS_TOTAL: &str = + "openshell_server_supervisor_session_off_ring_admissions_total"; // Histograms (explicit buckets, see BUCKETED_HISTOGRAMS) pub const RELAY_CLAIM_DURATION_SECONDS: &str = "openshell_server_relay_claim_duration_seconds"; pub const PEER_REQUEST_DURATION_SECONDS: &str = "openshell_server_peer_request_duration_seconds"; @@ -127,6 +132,27 @@ impl RelayRejection { } } +/// Event that sent a supervisor session redirect: a supervisor connecting, or this replica +/// shutting down. A stopping replica's connect-time redirects count as `Connect`. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum RedirectReason { + /// `ConnectSupervisor` answered the hello with a redirect instead of accepting the session. + Connect, + /// A stopping replica queued a redirect on an established session. + Shutdown, +} + +impl RedirectReason { + pub const ALL: [Self; 2] = [Self::Connect, Self::Shutdown]; + + pub const fn label(self) -> &'static str { + match self { + Self::Connect => "connect", + Self::Shutdown => "shutdown", + } + } +} + /// Routed operation. For a peer request, the owning replica records the matching gRPC method in /// `openshell_server_grpc_requests_total`, for example `PeerRelay` for `relay`. #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -243,20 +269,27 @@ pub fn configure_exporter(builder: PrometheusBuilder) -> Result, + off_ring_admissions: bool, ) -> Result { let handle = configure_exporter(PrometheusBuilder::new())?.install_recorder()?; - describe_and_initialize(relay, mutation_lock_connections); + describe_and_initialize(relay, mutation_lock_connections, off_ring_admissions); Ok(handle) } /// Emit HELP metadata, create every fixed-label series at 0, and publish the relay caps and, /// with a lock pool, its size. An idle replica then exports 0 instead of "no data", which HPA -/// and `rate()` need. -pub fn describe_and_initialize(relay: RelayCapacity, mutation_lock_connections: Option) { +/// and `rate()` need. `off_ring_admissions` gates the off-ring counter, which only a gateway +/// that places supervisor sessions by ring exports. +pub fn describe_and_initialize( + relay: RelayCapacity, + mutation_lock_connections: Option, + off_ring_admissions: bool, +) { describe_gauge!( SUPERVISOR_SESSIONS, Unit::Count, @@ -272,6 +305,13 @@ pub fn describe_and_initialize(relay: RelayCapacity, mutation_lock_connections: Unit::Count, "Maximum pending relay channels on one gateway replica." ); + // Not zero-initialized: only replicas that join the placement ring record it, so SQLite + // gateways and replicas without a peer endpoint export nothing instead of a misleading 0. + describe_gauge!( + GATEWAY_RING_MEMBERS, + Unit::Count, + "Gateway replicas in this replica's placement ring, including itself." + ); describe_counter!( ROUTED_REQUEST_ATTEMPTS_TOTAL, Unit::Count, @@ -287,6 +327,21 @@ pub fn describe_and_initialize(relay: RelayCapacity, mutation_lock_connections: Unit::Count, "Pending relay channels dropped because the supervisor did not connect back in time." ); + describe_counter!( + SUPERVISOR_SESSION_REDIRECTS_TOTAL, + Unit::Count, + "Supervisor sessions redirected to the ring-preferred replica, counted when the redirect is sent at connect or queued at shutdown." + ); + // Only gateways that place supervisor sessions by ring export this counter; a gateway + // coordinating a shared supervisor moves sessions from the host and would read a + // misleading 0, so it gets no series at all. + if off_ring_admissions { + describe_counter!( + SUPERVISOR_SESSION_OFF_RING_ADMISSIONS_TOTAL, + Unit::Count, + "Redirected supervisor sessions admitted although this replica's ring names another owner for the sandbox. Exported only on gateways that coordinate no shared supervisor." + ); + } describe_counter!( MUTATION_LOCK_TIMEOUTS_TOTAL, Unit::Count, @@ -354,6 +409,12 @@ pub fn describe_and_initialize(relay: RelayCapacity, mutation_lock_connections: counter!(RELAY_REJECTED_TOTAL, LABEL_REASON => reason.label()).increment(0); } counter!(RELAY_EXPIRED_TOTAL).increment(0); + for reason in RedirectReason::ALL { + counter!(SUPERVISOR_SESSION_REDIRECTS_TOTAL, LABEL_REASON => reason.label()).increment(0); + } + if off_ring_admissions { + counter!(SUPERVISOR_SESSION_OFF_RING_ADMISSIONS_TOTAL).increment(0); + } for scope in LockScope::ALL { counter!(MUTATION_LOCK_TIMEOUTS_TOTAL, LABEL_SCOPE => scope.label()).increment(0); counter!(MUTATION_LOCK_ERRORS_TOTAL, LABEL_SCOPE => scope.label()).increment(0); @@ -437,6 +498,25 @@ pub fn record_relay_rejected(reason: RelayRejection) { counter!(RELAY_REJECTED_TOTAL, LABEL_REASON => reason.label()).increment(1); } +/// One supervisor session redirect, labeled by the step that sent it: `Connect` once the +/// redirect is on the response stream, `Shutdown` once it is queued on the session. +pub fn record_session_redirect(reason: RedirectReason) { + counter!(SUPERVISOR_SESSION_REDIRECTS_TOTAL, LABEL_REASON => reason.label()).increment(1); +} + +/// A redirected supervisor session was admitted although this replica's ring names another +/// owner for its sandbox. Recorded only after the session is accepted, and only on a gateway +/// that places supervisor sessions by ring: one coordinating a shared supervisor does not +/// export the series and never records it. +pub fn record_off_ring_admission() { + counter!(SUPERVISOR_SESSION_OFF_RING_ADMISSIONS_TOTAL).increment(1); +} + +/// Replicas in this replica's placement ring. Recorded together with each ring write. +pub fn record_gateway_ring_members(count: usize) { + gauge!(GATEWAY_RING_MEMBERS).set(count_as_f64(count)); +} + /// `count` pending relays were dropped unclaimed (late claim or reaper). pub fn record_relay_expired(count: usize) { if count > 0 { @@ -634,88 +714,113 @@ mod tests { #[test] fn describe_and_initialize_exports_capacity_and_zero_series() { - let metrics = MetricsCapture::install(); - describe_and_initialize(RelayCapacity { per_replica: 256 }, Some(4)); - - for (series, expected) in [ - ("openshell_server_relay_pending_capacity", 256), - ("openshell_server_mutation_lock_connections_capacity", 4), - ("openshell_server_mutation_lock_connections_in_use", 0), - ("openshell_server_supervisor_sessions", 0), - ("openshell_server_relay_pending", 0), - ( - "openshell_server_relay_rejected_total{reason=\"replica_capacity\"}", - 0, - ), - ( - "openshell_server_relay_rejected_total{reason=\"sandbox_capacity\"}", - 0, - ), - ("openshell_server_relay_expired_total", 0), - ( - "openshell_server_mutation_lock_timeouts_total{scope=\"global\"}", - 0, - ), - ( - "openshell_server_mutation_lock_timeouts_total{scope=\"workspace\"}", - 0, - ), - ( - "openshell_server_mutation_lock_timeouts_total{scope=\"sandbox\"}", - 0, - ), - ( - "openshell_server_mutation_lock_errors_total{scope=\"global\"}", - 0, - ), - ( - "openshell_server_mutation_lock_errors_total{scope=\"workspace\"}", - 0, - ), - ( - "openshell_server_mutation_lock_errors_total{scope=\"sandbox\"}", - 0, - ), - ( - "openshell_server_provisioning_deadline_skips_total{overdue=\"false\"}", - 0, - ), - ( - "openshell_server_provisioning_deadline_skips_total{overdue=\"true\"}", - 0, - ), - ] { - assert_eq!(metrics.value(series), Some(expected), "{series}"); - } - for kind in ["ssh", "tcp"] { - for route in ["local", "peer"] { + for off_ring_admissions in [true, false] { + let metrics = MetricsCapture::install(); + describe_and_initialize( + RelayCapacity { per_replica: 256 }, + Some(4), + off_ring_admissions, + ); + + for (series, expected) in [ + ("openshell_server_relay_pending_capacity", 256), + ("openshell_server_mutation_lock_connections_capacity", 4), + ("openshell_server_mutation_lock_connections_in_use", 0), + ("openshell_server_supervisor_sessions", 0), + ("openshell_server_relay_pending", 0), + ( + "openshell_server_relay_rejected_total{reason=\"replica_capacity\"}", + 0, + ), + ( + "openshell_server_relay_rejected_total{reason=\"sandbox_capacity\"}", + 0, + ), + ("openshell_server_relay_expired_total", 0), + ( + "openshell_server_mutation_lock_timeouts_total{scope=\"global\"}", + 0, + ), + ( + "openshell_server_mutation_lock_timeouts_total{scope=\"workspace\"}", + 0, + ), + ( + "openshell_server_mutation_lock_timeouts_total{scope=\"sandbox\"}", + 0, + ), + ( + "openshell_server_mutation_lock_errors_total{scope=\"global\"}", + 0, + ), + ( + "openshell_server_mutation_lock_errors_total{scope=\"workspace\"}", + 0, + ), + ( + "openshell_server_mutation_lock_errors_total{scope=\"sandbox\"}", + 0, + ), + ( + "openshell_server_provisioning_deadline_skips_total{overdue=\"false\"}", + 0, + ), + ( + "openshell_server_provisioning_deadline_skips_total{overdue=\"true\"}", + 0, + ), + ( + "openshell_server_supervisor_session_redirects_total{reason=\"connect\"}", + 0, + ), + ( + "openshell_server_supervisor_session_redirects_total{reason=\"shutdown\"}", + 0, + ), + ] { + assert_eq!(metrics.value(series), Some(expected), "{series}"); + } + // The off-ring counter exists only where the gateway places sessions by ring; a + // shared-supervisor gateway exports nothing rather than a misleading 0. + let off_ring_series = metrics.value(SUPERVISOR_SESSION_OFF_RING_ADMISSIONS_TOTAL); + assert_eq!( + off_ring_series, + off_ring_admissions.then_some(0), + "off_ring_admissions={off_ring_admissions}" + ); + // Described only, so it renders nothing, not even HELP, until the ring is published. + assert_eq!(metrics.value(GATEWAY_RING_MEMBERS), None); + assert!(!metrics.render().contains(GATEWAY_RING_MEMBERS)); + for kind in ["ssh", "tcp"] { + for route in ["local", "peer"] { + let series = format!( + "openshell_server_routed_request_attempts_total{{operation=\"relay\",route=\"{route}\",relay_kind=\"{kind}\",outcome=\"success\",grpc_code=\"ok\"}}" + ); + assert_eq!(metrics.value(&series), Some(0), "{series}"); + } + } + for operation in [ + "report_provider_readiness", + "report_endpoint_status", + "get_sandbox_provider_status", + ] { let series = format!( - "openshell_server_routed_request_attempts_total{{operation=\"relay\",route=\"{route}\",relay_kind=\"{kind}\",outcome=\"success\",grpc_code=\"ok\"}}" + "openshell_server_routed_request_attempts_total{{operation=\"{operation}\",route=\"peer\",relay_kind=\"none\",outcome=\"success\",grpc_code=\"ok\"}}" ); assert_eq!(metrics.value(&series), Some(0), "{series}"); } - } - for operation in [ - "report_provider_readiness", - "report_endpoint_status", - "get_sandbox_provider_status", - ] { - let series = format!( - "openshell_server_routed_request_attempts_total{{operation=\"{operation}\",route=\"peer\",relay_kind=\"none\",outcome=\"success\",grpc_code=\"ok\"}}" + assert!( + metrics + .render() + .contains("# HELP openshell_server_supervisor_sessions ") ); - assert_eq!(metrics.value(&series), Some(0), "{series}"); } - assert!( - metrics - .render() - .contains("# HELP openshell_server_supervisor_sessions ") - ); } #[test] fn describe_and_initialize_without_a_lock_pool_omits_lock_connection_gauges() { let metrics = MetricsCapture::install(); - describe_and_initialize(RelayCapacity { per_replica: 256 }, None); + describe_and_initialize(RelayCapacity { per_replica: 256 }, None, true); assert_eq!(metrics.value(MUTATION_LOCK_CONNECTIONS_IN_USE), None); assert_eq!(metrics.value(MUTATION_LOCK_CONNECTIONS_CAPACITY), None); @@ -730,6 +835,65 @@ mod tests { ); } + /// A gateway coordinating a shared supervisor never records off-ring admissions, so it must + /// not zero-initialize the series either: HELP, TYPE, and value all stay absent. + #[test] + fn describe_and_initialize_without_off_ring_admissions_omits_the_counter() { + let metrics = MetricsCapture::install(); + describe_and_initialize(RelayCapacity { per_replica: 256 }, None, false); + + assert_eq!( + metrics.value(SUPERVISOR_SESSION_OFF_RING_ADMISSIONS_TOTAL), + None + ); + assert!( + !metrics + .render() + .contains(SUPERVISOR_SESSION_OFF_RING_ADMISSIONS_TOTAL) + ); + assert_eq!( + metrics + .value("openshell_server_supervisor_session_redirects_total{reason=\"connect\"}"), + Some(0) + ); + } + + #[test] + fn placement_metrics_export_help_and_type_once_present() { + for off_ring_admissions in [true, false] { + let metrics = MetricsCapture::install(); + describe_and_initialize( + RelayCapacity { per_replica: 256 }, + None, + off_ring_admissions, + ); + record_gateway_ring_members(3); + + let rendered = metrics.render(); + for (name, kind) in [ + (SUPERVISOR_SESSION_REDIRECTS_TOTAL, "counter"), + (GATEWAY_RING_MEMBERS, "gauge"), + ] { + assert!(rendered.contains(&format!("# HELP {name} ")), "{name} HELP"); + assert!( + rendered.contains(&format!("# TYPE {name} {kind}")), + "{name} TYPE" + ); + } + if off_ring_admissions { + let name = SUPERVISOR_SESSION_OFF_RING_ADMISSIONS_TOTAL; + assert!(rendered.contains(&format!("# HELP {name} ")), "{name} HELP"); + assert!( + rendered.contains(&format!("# TYPE {name} counter")), + "{name} TYPE" + ); + } else { + assert!(!rendered.contains(SUPERVISOR_SESSION_OFF_RING_ADMISSIONS_TOTAL)); + } + assert_eq!(metrics.value(GATEWAY_RING_MEMBERS), Some(3)); + } + } + #[test] fn routed_attempts_have_seven_bounded_success_series_and_keep_counts_on_initialize() { let metrics = MetricsCapture::install(); @@ -739,7 +903,7 @@ mod tests { } } RoutedRequestTimer::relay(RelayKind::Tcp, RelayRoute::Peer).finish(&Ok::<(), Status>(())); - describe_and_initialize(RelayCapacity { per_replica: 256 }, None); + describe_and_initialize(RelayCapacity { per_replica: 256 }, None, true); for kind in ["ssh", "tcp"] { for route in ["local", "peer"] { diff --git a/crates/openshell-server/src/grpc/mod.rs b/crates/openshell-server/src/grpc/mod.rs index d066845fa4..0243e89b72 100644 --- a/crates/openshell-server/src/grpc/mod.rs +++ b/crates/openshell-server/src/grpc/mod.rs @@ -1001,7 +1001,8 @@ pub mod test_support { Principal, SandboxIdentitySource, SandboxPrincipal, UserPrincipal, }; use crate::compute::{ - NoopTestDriver, new_test_runtime, new_test_runtime_for_driver, new_test_runtime_with_driver, + ComputeRuntime, NoopTestDriver, new_test_runtime, new_test_runtime_for_driver, + new_test_runtime_with_driver, }; use crate::persistence::Store; use crate::sandbox_index::SandboxIndex; @@ -1075,11 +1076,34 @@ pub mod test_support { sandbox_id: &str, features: StreamFeatures, connection_epoch: u64, + ) -> Result { + connect_supervisor_stream_with_hello( + state, + SupervisorHello { + sandbox_id: sandbox_id.into(), + instance_id: "instance".into(), + supports_config_push: features == StreamFeatures::Apply, + connection_epoch, + supports_provider_readiness: false, + redirected: false, + supports_session_redirect: false, + }, + ) + .await + } + + /// Serve `state` on loopback and open a supervisor stream that sends + /// `hello` as-is, authenticated as the sandbox the hello names. Returns + /// the gRPC status when the gateway rejects the stream before accepting + /// it. + pub async fn connect_supervisor_stream_with_hello( + state: &Arc, + hello: SupervisorHello, ) -> Result { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); let principal = Principal::Sandbox(SandboxPrincipal { - sandbox_id: sandbox_id.to_string(), + sandbox_id: hello.sandbox_id.clone(), source: SandboxIdentitySource::BootstrapJwt { issuer: "openshell-gateway:test".to_string(), auth_epoch: None, @@ -1106,15 +1130,7 @@ pub mod test_support { let (outbound, rx) = mpsc::channel(4); outbound .send(SupervisorMessage { - payload: Some(supervisor_message::Payload::Hello(SupervisorHello { - sandbox_id: sandbox_id.into(), - instance_id: "instance".into(), - supports_config_push: features == StreamFeatures::Apply, - connection_epoch, - supports_provider_readiness: false, - redirected: false, - supports_session_redirect: false, - })), + payload: Some(supervisor_message::Payload::Hello(hello)), }) .await .unwrap(); @@ -1188,11 +1204,62 @@ pub mod test_support { state } + const IN_MEMORY_DATABASE_URL: &str = "sqlite::memory:?cache=shared"; + + /// Config shared by the in-memory test states. `database_url` is recorded + /// only when the state claims one; a caller-supplied store leaves it unset. + fn test_gateway_config(database_url: Option<&str>) -> Config { + let config = Config::new(None).with_credential_drivers(["test-static"]); + match database_url { + Some(url) => config.with_database_url(url), + None => config, + } + } + + /// Assemble a test state from the pieces the builders below vary on: the + /// store, the compute runtime, and the config. + fn test_server_state_from_parts( + store: Arc, + compute: ComputeRuntime, + config: Config, + ) -> Arc { + with_test_provider_profile_sources(Arc::new(ServerState::new( + config, + store, + compute, + SandboxIndex::new(), + SandboxWatchBus::new(), + TracingLogBus::new(), + Arc::new(SupervisorSessionRegistry::new()), + None, + ))) + } + /// Build an in-memory `ServerState` for unit tests. pub async fn test_server_state() -> Arc { test_server_state_with_driver("test").await } + /// Build a test state whose compute runtime coordinates a shared supervisor host, the + /// regime a gateway under `supervisor_topology: gateway` runs: it moves supervisor + /// sessions from the host instead of placing them by ring. The bare controller is enough + /// for code that only reads its presence. + pub async fn test_server_state_with_supervisor_host() -> Arc { + let store = Arc::new(Store::connect(IN_MEMORY_DATABASE_URL).await.unwrap()); + crate::ensure_default_workspace(&store).await.unwrap(); + seed_example_provider_profiles(&store).await; + let compute = new_test_runtime(store.clone()) + .await + .with_supervisor_host(Some( + openshell_supervisor_control::SupervisorHostController::new(), + )); + test_server_state_from_parts( + store, + compute, + test_gateway_config(Some(IN_MEMORY_DATABASE_URL)), + ) + } + /// Build a test state for a gateway with nothing imported. /// /// Its profile catalog is empty, which is the state a freshly installed @@ -1210,37 +1277,22 @@ pub mod test_support { driver_name: &str, driver: Arc, ) -> Arc { - let store = Arc::new( - Store::connect("sqlite::memory:?cache=shared") - .await - .unwrap(), - ); + let store = Arc::new(Store::connect(IN_MEMORY_DATABASE_URL).await.unwrap()); crate::ensure_default_workspace(&store).await.unwrap(); seed_example_provider_profiles(&store).await; let compute = new_test_runtime_with_driver(store.clone(), driver_name, driver); - with_test_provider_profile_sources(Arc::new(ServerState::new( - Config::new(None) - .with_database_url("sqlite::memory:?cache=shared") - .with_credential_drivers(["test-static"]), + test_server_state_from_parts( store, compute, - SandboxIndex::new(), - SandboxWatchBus::new(), - TracingLogBus::new(), - Arc::new(SupervisorSessionRegistry::new()), - None, - ))) + test_gateway_config(Some(IN_MEMORY_DATABASE_URL)), + ) } async fn test_server_state_for_driver( driver_name: &str, seed_profiles: bool, ) -> Arc { - let store = Arc::new( - Store::connect("sqlite::memory:?cache=shared") - .await - .unwrap(), - ); + let store = Arc::new(Store::connect(IN_MEMORY_DATABASE_URL).await.unwrap()); crate::ensure_default_workspace(&store).await.unwrap(); if seed_profiles { seed_example_provider_profiles(&store).await; @@ -1250,18 +1302,22 @@ pub mod test_support { } else { new_test_runtime_for_driver(store.clone(), driver_name).await }; - with_test_provider_profile_sources(Arc::new(ServerState::new( - Config::new(None) - .with_database_url("sqlite::memory:?cache=shared") - .with_credential_drivers(["test-static"]), + test_server_state_from_parts( store, compute, - SandboxIndex::new(), - SandboxWatchBus::new(), - TracingLogBus::new(), - Arc::new(SupervisorSessionRegistry::new()), - None, - ))) + test_gateway_config(Some(IN_MEMORY_DATABASE_URL)), + ) + } + + /// Build a test state on a store the caller provides, such as a disposable + /// `PostgreSQL` schema. Like the states above, it sets up the default + /// workspace, test compute runtime, provider sources, and credential + /// drivers, but configures no database URL and seeds no provider profiles. + pub async fn test_server_state_for_store(store: Store) -> Arc { + let store = Arc::new(store); + crate::ensure_default_workspace(&store).await.unwrap(); + let compute = new_test_runtime(store.clone()).await; + test_server_state_from_parts(store, compute, test_gateway_config(None)) } /// Build a test state whose compute driver fails the requested number of @@ -1269,27 +1325,16 @@ pub mod test_support { pub async fn test_server_state_with_workspace_cleanup_failures( failures: usize, ) -> Arc { - let store = Arc::new( - Store::connect("sqlite::memory:?cache=shared") - .await - .unwrap(), - ); + let store = Arc::new(Store::connect(IN_MEMORY_DATABASE_URL).await.unwrap()); crate::ensure_default_workspace(&store).await.unwrap(); seed_example_provider_profiles(&store).await; let driver = Arc::new(NoopTestDriver::failing_workspace_deletes(failures)); let compute = new_test_runtime_with_driver(store.clone(), "test", driver); - with_test_provider_profile_sources(Arc::new(ServerState::new( - Config::new(None) - .with_database_url("sqlite::memory:?cache=shared") - .with_credential_drivers(["test-static"]), + test_server_state_from_parts( store, compute, - SandboxIndex::new(), - SandboxWatchBus::new(), - TracingLogBus::new(), - Arc::new(SupervisorSessionRegistry::new()), - None, - ))) + test_gateway_config(Some(IN_MEMORY_DATABASE_URL)), + ) } } diff --git a/crates/openshell-server/src/lib.rs b/crates/openshell-server/src/lib.rs index 4608161785..514f8eeb6b 100644 --- a/crates/openshell-server/src/lib.rs +++ b/crates/openshell-server/src/lib.rs @@ -381,7 +381,9 @@ pub struct ServerState { /// Latest placement ring, refreshed from live gateway membership. Decides /// which replica *should* own a sandbox; the owner record in the store - /// remains authoritative for which one does. + /// remains authoritative for which one does. Production code writes it + /// only through `gateway_members::publish_ring`, which keeps the + /// ring-members gauge in step. pub gateway_ring: Arc>, /// Peer endpoints of live replicas, keyed by replica ID. Refreshed @@ -1104,6 +1106,7 @@ pub(crate) async fn run_server( gateway_metrics::install_global_recorder( supervisor_session::RELAY_CAPACITY, store.mutation_lock_connections(), + state.compute.places_supervisor_sessions_by_ring(), ) .map(|prometheus_handle| (metrics_bind_address, prometheus_handle)) }) @@ -3956,6 +3959,52 @@ mod tests { "persisted sandbox start must not run before the gateway listener is bound" ); } + + /// `run_server` installs the process-wide recorder with the off-ring gate it derives from + /// the compute runtime it built, so a gateway coordinating a shared supervisor starts with + /// no `openshell_server_supervisor_session_off_ring_admissions_total` series at all: no + /// HELP, no TYPE, no zero value. The regime tests call `describe_and_initialize` with a + /// literal bool, so this is the only test that exercises `install_global_recorder`. It pins + /// the predicate the production call site passes instead of starting `run_server`. + #[tokio::test] + async fn install_global_recorder_gates_the_off_ring_series_from_the_compute_runtime() { + let pool_state = crate::grpc::test_support::test_server_state_with_supervisor_host().await; + let ring_state = crate::grpc::test_support::test_server_state().await; + assert!( + !pool_state.compute.places_supervisor_sessions_by_ring(), + "a gateway coordinating a shared supervisor does not export the off-ring counter" + ); + assert!( + ring_state.compute.places_supervisor_sessions_by_ring(), + "a gateway placing supervisor sessions by ring exports the off-ring counter" + ); + + // The recorder installs once per process, so this test covers the regime the + // operator-facing contract cares about: a pool gateway must start with no series. The + // `describe_and_initialize` tests cover the mirrored true flag. + let handle = super::gateway_metrics::install_global_recorder( + super::supervisor_session::RELAY_CAPACITY, + None, + pool_state.compute.places_supervisor_sessions_by_ring(), + ) + .expect("install the process-wide recorder from the test process"); + + let rendered = handle.render(); + let off_ring = super::gateway_metrics::SUPERVISOR_SESSION_OFF_RING_ADMISSIONS_TOTAL; + assert!( + !rendered.contains(off_ring), + "a shared-supervisor gateway must export no {off_ring} HELP, TYPE, or value" + ); + // The rest of the catalog is still described and zero-initialized. + assert!( + rendered.contains("# HELP openshell_server_supervisor_sessions "), + "the recorder still initializes the supervisor session gauge" + ); + assert!( + rendered.contains("# HELP openshell_server_supervisor_session_redirects_total "), + "the recorder still initializes the redirect counter" + ); + } } #[cfg(test)] diff --git a/crates/openshell-server/src/supervisor_session.rs b/crates/openshell-server/src/supervisor_session.rs index fd54e7ff19..e715c60f40 100644 --- a/crates/openshell-server/src/supervisor_session.rs +++ b/crates/openshell-server/src/supervisor_session.rs @@ -34,7 +34,7 @@ use crate::auth::principal::Principal; use crate::config_delivery::MAX_SUPERVISOR_CONFIG_MESSAGE_BYTES; use crate::config_delivery::session::{self as config_session, ConfigSessionState}; use crate::gateway_metrics::{ - self, GaugeSlot, PeerRpc, RelayCapacity, RelayKind, RelayRejection, RelayRoute, + self, GaugeSlot, PeerRpc, RedirectReason, RelayCapacity, RelayKind, RelayRejection, RelayRoute, RoutedRequestTimer, }; use crate::grpc::provider_readiness::ProviderReadinessEvidence; @@ -1904,6 +1904,25 @@ fn may_redirect(hello: &SupervisorHello, shutting_down: bool) -> bool { hello.supports_session_redirect && (!hello.redirected || shutting_down) } +/// Whether admitting this hello here goes against this replica's ring: the +/// supervisor was redirected, yet the ring names another replica as the +/// sandbox's owner. An empty ring has no opinion. This only applies where the +/// gateway places supervisor sessions itself, so `hello.redirected` means the +/// supervisor is following a redirect; a gateway coordinating a shared +/// supervisor moves sessions from the host and never counts. This is only +/// counted; the session is still served so a stale ring or a failed redirect +/// dial cannot make the sandbox unreachable. +fn is_off_ring_admission( + hello: &SupervisorHello, + ring: &crate::gateway_ring::GatewayRing, + self_replica_id: &str, +) -> bool { + hello.redirected + && ring + .owner_for(&hello.sandbox_id) + .is_some_and(|owner| owner != self_replica_id) +} + fn ring_redirect( self_replica_id: &str, preferred_replica_id: &str, @@ -2224,6 +2243,9 @@ struct SessionSetup { outbound_tx: mpsc::Sender, inbound: tonic::Streaming, session_lifetime: OwnedRwLockReadGuard<()>, + /// Whether this redirected hello is an off-ring admission. Counted here, once acceptance + /// is durable, so a cancelled connect future cannot lose the count. + off_ring: bool, } async fn accept_supervisor_session(setup: SessionSetup) -> Result<(), Status> { @@ -2241,6 +2263,7 @@ async fn accept_supervisor_session(setup: SessionSetup) -> Result<(), Status> { outbound_tx, mut inbound, session_lifetime, + off_ring, } = setup; let pushes_config = matches!(mode, SessionMode::Push); // The registration seeds the session's provider environment cache, and @@ -2442,6 +2465,10 @@ async fn accept_supervisor_session(setup: SessionSetup) -> Result<(), Status> { abandon_session_setup(&state, &owner_index, &owner_guard, "accept send").await; return Err(Status::internal("failed to send session accepted")); } + // Acceptance is durable, so the admission counts even if the connect handler is gone. + if off_ring { + gateway_metrics::record_off_ring_admission(); + } info!( sandbox_id = %sandbox_id, session_id = %session_id, @@ -2597,10 +2624,22 @@ pub async fn handle_connect_supervisor( // Capacity 1 and a fresh receiver, so this cannot block. Dropping the // sender ends the stream, which closes the supervisor's connection. let _ = tx.send(Ok(message)).await; + gateway_metrics::record_session_redirect(RedirectReason::Connect); drop(tx); return Ok(Response::new(Box::pin(ReceiverStream::new(rx)))); } + // Read while the hello is whole, and counted by the session setup once acceptance is + // durable. A hello that was not redirected takes no ring lock, and a gateway coordinating + // a shared supervisor never takes it: there the host moves sessions, so a redirect does + // not express ring disagreement. + let off_ring = state.compute.places_supervisor_sessions_by_ring() + && hello.redirected + && state + .gateway_ring + .read() + .is_ok_and(|ring| is_off_ring_admission(&hello, &ring, &state.replica_id)); + // Build the bootstrap only for a session this replica keeps. let config_push = captured_seq.map(|captured_seq| ConfigPushSetup { captured_seq, @@ -2641,6 +2680,7 @@ pub async fn handle_connect_supervisor( outbound_tx: tx, inbound, session_lifetime, + off_ring, }; // Keep setup alive if the RPC caller disconnects after publication. tokio::spawn(accept_supervisor_session(setup)) @@ -2873,19 +2913,24 @@ async fn wait_for_replacement_owner( } /// Queue a shutdown redirect for a supervisor that supports one. Returns -/// whether a redirect is now queued for the supervisor. +/// whether a redirect is now queued for the supervisor, and counts it only +/// then. fn queue_shutdown_redirect( tx: &mpsc::Sender, supports_session_redirect: bool, redirect: impl FnOnce() -> Option, ) -> bool { - supports_session_redirect + let queued = supports_session_redirect && redirect().is_some_and(|redirect| { tx.try_send(GatewayMessage { payload: Some(gateway_message::Payload::SessionRedirect(redirect)), }) .is_ok() - }) + }); + if queued { + gateway_metrics::record_session_redirect(RedirectReason::Shutdown); + } + queued } /// An accepted supervisor session, shared by the handlers for its messages. @@ -3144,6 +3189,16 @@ mod tests { state } + async fn state_with_sandbox_and_supervisor_host(sandbox_id: &str) -> Arc { + let state = crate::grpc::test_support::test_server_state_with_supervisor_host().await; + state + .store + .put_message(&sandbox_record(sandbox_id, sandbox_id)) + .await + .unwrap(); + state + } + async fn first_gateway_message( harness: &mut crate::grpc::test_support::SupervisorStreamHarness, ) -> GatewayMessage { @@ -3154,6 +3209,11 @@ mod tests { .expect("gateway message") } + const SHUTDOWN_REDIRECTS: &str = + "openshell_server_supervisor_session_redirects_total{reason=\"shutdown\"}"; + const CONNECT_REDIRECTS: &str = + "openshell_server_supervisor_session_redirects_total{reason=\"connect\"}"; + #[tokio::test] async fn supervisor_without_config_push_is_accepted() { let state = state_with_sandbox("sb-legacy").await; @@ -3490,12 +3550,14 @@ mod tests { #[test] fn shutdown_redirect_is_queued_only_when_supported_available_and_sent() { + let metrics = MetricsCapture::install(); let (tx, mut rx) = mpsc::channel(1); assert!(queue_shutdown_redirect(&tx, true, || Some(test_redirect()))); assert!(matches!( rx.try_recv().unwrap().payload, Some(gateway_message::Payload::SessionRedirect(_)) )); + assert_eq!(metrics.value(SHUTDOWN_REDIRECTS), Some(1)); assert!(!queue_shutdown_redirect(&tx, true, || None)); assert!(!queue_shutdown_redirect(&tx, false, || { @@ -3509,6 +3571,12 @@ mod tests { true, || Some(test_redirect()) )); + assert_eq!( + metrics.value(SHUTDOWN_REDIRECTS), + Some(1), + "only the queued redirect counts" + ); + assert_eq!(metrics.value(CONNECT_REDIRECTS), None); } #[tokio::test] @@ -5655,6 +5723,60 @@ mod tests { assert!(!may_redirect(&hello, true)); } + /// A sandbox id the ring places on `owner`. + fn sandbox_owned_by(ring: &crate::gateway_ring::GatewayRing, owner: &str) -> String { + (0..300) + .map(|i| format!("sb-{i}")) + .find(|id| ring.owner_for(id) == Some(owner)) + .unwrap_or_else(|| panic!("the ring places no sandbox on {owner}")) + } + + fn redirect_capable_hello(sandbox_id: &str, redirected: bool) -> SupervisorHello { + SupervisorHello { + sandbox_id: sandbox_id.to_string(), + instance_id: "instance".to_string(), + redirected, + supports_session_redirect: true, + ..Default::default() + } + } + + #[test] + fn off_ring_admission_is_a_redirected_hello_whose_ring_owner_is_another_replica() { + let ring = crate::gateway_ring::GatewayRing::new(["gw-0", "gw-1", "gw-2"]); + let sandbox_id = sandbox_owned_by(&ring, "gw-1"); + let hello = redirect_capable_hello(&sandbox_id, true); + + assert!(is_off_ring_admission(&hello, &ring, "gw-0")); + assert!(is_off_ring_admission(&hello, &ring, "gw-2")); + assert!(!is_off_ring_admission(&hello, &ring, "gw-1")); + // A replica missing from its own stale ring still sees another owner. + assert!(is_off_ring_admission(&hello, &ring, "gw-3")); + assert!(!is_off_ring_admission( + &redirect_capable_hello(&sandbox_id, false), + &ring, + "gw-0" + )); + assert!(!is_off_ring_admission( + &hello, + &crate::gateway_ring::GatewayRing::default(), + "gw-0" + )); + } + + /// With two replicas every non-owner is also the owner's successor, so a + /// handoff to it must still count. + #[test] + fn off_ring_admission_counts_on_a_two_replica_ring() { + let ring = crate::gateway_ring::GatewayRing::new(["gw-a", "gw-b"]); + let sandbox_id = sandbox_owned_by(&ring, "gw-a"); + assert!(is_off_ring_admission( + &redirect_capable_hello(&sandbox_id, true), + &ring, + "gw-b" + )); + } + #[test] fn ring_redirect_serves_locally_when_this_replica_is_preferred() { assert!(ring_redirect("gw-0", "gw-0", Some("https://gw-0:8443".to_string())).is_none()); @@ -6233,4 +6355,297 @@ mod tests { .unwrap(); assert_eq!(reported.status.unwrap().exit_code, Some(1)); } + + /// Name this replica and point its ring at another replica for every + /// sandbox. + fn place_sandboxes_on(state: &mut Arc, self_replica_id: &str, owner: &str) { + Arc::get_mut(state) + .expect("test state is not shared yet") + .replica_id = self_replica_id.to_string(); + *state.gateway_ring.write().unwrap() = crate::gateway_ring::GatewayRing::new([owner]); + } + + #[tokio::test] + async fn redirected_supervisor_admitted_against_the_ring_counts_off_ring() { + let metrics = MetricsCapture::install(); + let mut state = state_with_sandbox("sb-off-ring").await; + place_sandboxes_on(&mut state, "gw-self-ut", "gw-owner-ut"); + + let mut harness = crate::grpc::test_support::connect_supervisor_stream_with_hello( + &state, + redirect_capable_hello("sb-off-ring", true), + ) + .await + .expect("a redirected supervisor is served even when the ring names another owner"); + assert!(matches!( + first_gateway_message(&mut harness).await.payload, + Some(gateway_message::Payload::SessionAccepted(_)) + )); + + assert_eq!( + metrics.value(gateway_metrics::SUPERVISOR_SESSION_OFF_RING_ADMISSIONS_TOTAL), + Some(1) + ); + let rendered = metrics.render(); + assert!(!rendered.contains(gateway_metrics::SUPERVISOR_SESSION_REDIRECTS_TOTAL)); + for id in ["sb-", "gw-self-ut", "gw-owner-ut"] { + assert!(!rendered.contains(id), "{id} must not appear in metrics"); + } + } + + #[tokio::test] + async fn failed_session_setup_does_not_count_an_off_ring_admission() { + let metrics = MetricsCapture::install(); + let mut state = state_with_sandbox("sb-off-ring-owned").await; + place_sandboxes_on(&mut state, "gw-self-ut", "gw-owner-ut"); + // Another supervisor instance holds a fresh owner record, so owner + // publication fails inside the spawned session setup. + let _owner = SupervisorOwnerIndex::new(state.store.clone(), OWNER_TTL) + .publish( + "sb-off-ring-owned", + "session-other", + "instance-other", + 0, + "gw-owner-ut", + "http://gw-owner-ut:8080", + ) + .await + .unwrap(); + + let Err(status) = crate::grpc::test_support::connect_supervisor_stream_with_hello( + &state, + redirect_capable_hello("sb-off-ring-owned", true), + ) + .await + else { + panic!("a sandbox owned by another supervisor instance must not be admitted"); + }; + assert_eq!(status.code(), tonic::Code::Unavailable); + assert_eq!( + metrics.value(gateway_metrics::SUPERVISOR_SESSION_OFF_RING_ADMISSIONS_TOTAL), + None + ); + } + + /// A gateway coordinating a shared supervisor moves sessions from the host, so a redirected + /// hello admitted here is the designed steady state, not ring disagreement: it must not + /// count, and the ring lock stays untouched. + #[tokio::test] + async fn off_ring_admission_is_not_counted_under_a_shared_supervisor() { + let metrics = MetricsCapture::install(); + let mut state = state_with_sandbox_and_supervisor_host("sb-pool-off-ring").await; + place_sandboxes_on(&mut state, "gw-self-ut", "gw-owner-ut"); + + let mut harness = crate::grpc::test_support::connect_supervisor_stream_with_hello( + &state, + redirect_capable_hello("sb-pool-off-ring", true), + ) + .await + .expect("a redirected supervisor is served even when the ring names another owner"); + assert!(matches!( + first_gateway_message(&mut harness).await.payload, + Some(gateway_message::Payload::SessionAccepted(_)) + )); + + assert_eq!( + metrics.value(gateway_metrics::SUPERVISOR_SESSION_OFF_RING_ADMISSIONS_TOTAL), + None + ); + } + + /// The off-ring read must never take the sandbox down: a poisoned placement ring cannot be + /// read, so the count is skipped while the redirected session is still served. Changing + /// this read to `unwrap()` panics the session setup, and counting on lock failure would + /// report an admission this replica cannot place. + #[tokio::test] + async fn poisoned_ring_lock_skips_the_off_ring_count_and_serves_the_session() { + let metrics = MetricsCapture::install(); + let mut state = state_with_sandbox("sb-poisoned-off-ring").await; + // The ring names another owner for every sandbox, so a readable lock would count. + place_sandboxes_on(&mut state, "gw-self-ut", "gw-owner-ut"); + let ring = Arc::clone(&state.gateway_ring); + let poisoned = std::thread::spawn(move || { + std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + let _guard = ring.write().unwrap(); + // `resume_unwind` drops the guard and poisons the lock without + // running the panic hook, keeping the expected panic out of + // the test output. + std::panic::resume_unwind(Box::new("poison the placement ring lock")); + })) + }) + .join() + .expect("the poisoning thread should catch its own unwind"); + assert!(poisoned.is_err()); + assert!(state.gateway_ring.is_poisoned()); + + let mut harness = crate::grpc::test_support::connect_supervisor_stream_with_hello( + &state, + redirect_capable_hello("sb-poisoned-off-ring", true), + ) + .await + .expect("a poisoned ring must not make the sandbox unreachable"); + assert!(matches!( + first_gateway_message(&mut harness).await.payload, + Some(gateway_message::Payload::SessionAccepted(_)) + )); + + assert_eq!( + metrics.value(gateway_metrics::SUPERVISOR_SESSION_OFF_RING_ADMISSIONS_TOTAL), + None + ); + } + + /// Decode hand-built request frames as `SupervisorMessage`; `Streaming` already strips the + /// gRPC frame header. + struct SupervisorMessageDecoder; + + impl tonic::codec::Decoder for SupervisorMessageDecoder { + type Item = SupervisorMessage; + type Error = Status; + + fn decode( + &mut self, + src: &mut tonic::codec::DecodeBuf<'_>, + ) -> Result, Self::Error> { + Message::decode(src) + .map(Some) + .map_err(|error| Status::internal(format!("decode supervisor message: {error}"))) + } + } + + /// A request stream carrying exactly the hello, the way tonic hands it to + /// `handle_connect_supervisor`. + fn hello_request_stream(hello: SupervisorHello) -> tonic::codec::Streaming { + let message = SupervisorMessage { + payload: Some(supervisor_message::Payload::Hello(hello)), + }; + let payload = message.encode_to_vec(); + let mut frame = Vec::with_capacity(5 + payload.len()); + frame.push(0); + frame.extend_from_slice(&u32::try_from(payload.len()).unwrap().to_be_bytes()); + frame.extend_from_slice(&payload); + let frames = futures::stream::iter([Ok::<_, Infallible>(Frame::data(Bytes::from(frame)))]); + tonic::codec::Streaming::new_request( + SupervisorMessageDecoder, + StreamBody::new(frames), + None, + None, + ) + } + + /// The detached session setup accepts the session even when the connect future is dropped + /// while it runs, so the off-ring count must follow acceptance rather than the caller's + /// connection. Polling the handler only until registration and never again models that + /// cancellation: an increment left in `handle_connect_supervisor` is never reached. + #[tokio::test] + async fn cancelled_connect_future_still_counts_an_accepted_off_ring_admission() { + let metrics = MetricsCapture::install(); + let mut state = state_with_sandbox("sb-cancelled-off-ring").await; + place_sandboxes_on(&mut state, "gw-self-ut", "gw-owner-ut"); + + let request = Request::new(hello_request_stream(redirect_capable_hello( + "sb-cancelled-off-ring", + true, + ))); + let mut connect = Box::pin(handle_connect_supervisor(&state, request)); + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + while !state + .supervisor_sessions + .has_session("sb-cancelled-off-ring") + { + assert!( + tokio::time::Instant::now() < deadline, + "session setup did not register a session" + ); + assert!( + futures_util::poll!(connect.as_mut()).is_pending(), + "the handler waits for the detached session setup" + ); + tokio::task::yield_now().await; + } + + let counted = tokio::time::timeout(Duration::from_secs(5), async { + while metrics.value(gateway_metrics::SUPERVISOR_SESSION_OFF_RING_ADMISSIONS_TOTAL) + != Some(1) + { + tokio::task::yield_now().await; + } + }) + .await; + drop(connect); + counted.expect("the accepted session is counted after the connect future is dropped"); + } + + #[tokio::test] + #[ignore = "requires a disposable PostgreSQL database in OPENSHELL_TEST_POSTGRES_URL; run mise run test:rust:postgres"] + async fn postgres_connect_redirect_is_counted_once() { + let metrics = MetricsCapture::install(); + let schema = + crate::persistence::test_postgres::TestSchema::create("connect_redirect").await; + let mut state = + crate::grpc::test_support::test_server_state_for_store(schema.connect_store().await) + .await; + place_sandboxes_on(&mut state, "gw-self-pg", "gw-owner-pg"); + state + .gateway_peers + .write() + .unwrap() + .insert("gw-owner-pg".to_string(), "http://127.0.0.1:9".to_string()); + state + .store + .put_message(&sandbox_record("sb-redirect-pg", "sb-redirect-pg")) + .await + .unwrap(); + + let mut harness = crate::grpc::test_support::connect_supervisor_stream_with_hello( + &state, + redirect_capable_hello("sb-redirect-pg", false), + ) + .await + .expect("a redirect is a successful response"); + let Some(gateway_message::Payload::SessionRedirect(redirect)) = + first_gateway_message(&mut harness).await.payload + else { + panic!("expected SessionRedirect"); + }; + assert_eq!(redirect.owner_replica_id, "gw-owner-pg"); + assert_eq!(metrics.value(CONNECT_REDIRECTS), Some(1)); + assert_eq!(metrics.value(SHUTDOWN_REDIRECTS), None); + assert_eq!( + metrics.value(gateway_metrics::SUPERVISOR_SESSION_OFF_RING_ADMISSIONS_TOTAL), + None + ); + assert!(!state.supervisor_sessions.has_session("sb-redirect-pg")); + let rendered = metrics.render(); + for id in ["sb-redirect-pg", "gw-self-pg", "gw-owner-pg"] { + assert!(!rendered.contains(id), "{id} must not appear in metrics"); + } + drop(harness); + + // A stopping replica redirects even an already-redirected supervisor + // at connect, which counts as a connect redirect, not a shutdown one. + state.gateway_shutting_down.store(true, Ordering::Release); + let mut harness = crate::grpc::test_support::connect_supervisor_stream_with_hello( + &state, + redirect_capable_hello("sb-redirect-pg", true), + ) + .await + .expect("a redirect is a successful response"); + let Some(gateway_message::Payload::SessionRedirect(redirect)) = + first_gateway_message(&mut harness).await.payload + else { + panic!("expected SessionRedirect from a stopping replica"); + }; + assert_eq!(redirect.owner_replica_id, "gw-owner-pg"); + assert_eq!(metrics.value(CONNECT_REDIRECTS), Some(2)); + assert_eq!(metrics.value(SHUTDOWN_REDIRECTS), None); + assert_eq!( + metrics.value(gateway_metrics::SUPERVISOR_SESSION_OFF_RING_ADMISSIONS_TOTAL), + None + ); + drop(harness); + + state.store.close().await; + schema.drop_schema().await; + } } diff --git a/docs/kubernetes/high-availability.mdx b/docs/kubernetes/high-availability.mdx index c21150f762..d79a0d08ab 100644 --- a/docs/kubernetes/high-availability.mdx +++ b/docs/kubernetes/high-availability.mdx @@ -212,6 +212,7 @@ configuration, and access control. | Signal | Metrics | Use | |---|---|---| | Sessions per replica | `openshell_server_supervisor_sessions` | Placement and skew. Do not scale on it. | +| Session placement | `openshell_server_supervisor_session_redirects_total`, `openshell_server_supervisor_session_off_ring_admissions_total`, `openshell_server_gateway_ring_members` | On gateways that place supervisor sessions by ring, redirects and off-ring admissions rise when a replica stops or joins. A sustained off-ring rate while membership is stable means replicas disagree on the ring or redirect dials fail. Once a rollout settles, every pod's ring size equals the number of gateway replicas. Under `supervisor_topology: gateway`, the shared supervisor's host moves sessions, so redirects and off-ring admissions do not reflect them. | | 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. | @@ -552,7 +553,9 @@ SQLite file is not an HA deployment. By default, the Kubernetes driver runs a supervisor Pod for each sandbox. With the multi-tenant topology, a shared supervisor Deployment hosts many sandboxes, and each sandbox keeps its own policy and identity. Supervisor replicas are -independent of the gateway replica count. +independent of the gateway replica count. Because the shared supervisor's host +places its sessions, sandbox moves under this topology do not appear in the +gateway placement metrics listed in [Monitor Capacity](#monitor-capacity). Enable the topology in your release values. Run at least two supervisor replicas so sandboxes can fail over: diff --git a/docs/observability/gateway-metrics.mdx b/docs/observability/gateway-metrics.mdx index 24ec1b0bd9..be61fd649d 100644 --- a/docs/observability/gateway-metrics.mdx +++ b/docs/observability/gateway-metrics.mdx @@ -175,6 +175,14 @@ Supervisor sessions: |---|---|---|---| | `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. | +Session placement: + +| Metric | Type | Labels | Description | +|---|---|---|---| +| `openshell_server_supervisor_session_redirects_total` | Counter | `reason` | Supervisors this replica redirected to the replica its placement ring prefers, counted once the redirect is sent or queued. | +| `openshell_server_supervisor_session_off_ring_admissions_total` | Counter | None | Redirected supervisors this replica admitted although its ring names another replica as the sandbox's owner. Exported only on gateways that place supervisor sessions by ring; under `supervisor_topology: gateway` the series does not exist. It counts each handoff from a stopping replica, which keeps receiving sessions until other rings refresh, up to 10 seconds later, and supervisors that fall back to their configured gateway address after a failed redirect dial. While membership is stable it stays flat, so alert on a sustained rate. | +| `openshell_server_gateway_ring_members` | Gauge | None | Replicas in this replica's placement ring, including itself until it leaves the ring at shutdown. It drops to `0` when membership refreshes have failed for more than 30 seconds. Exported only on replicas that join the placement ring (PostgreSQL with a peer endpoint). | + Relays: | Metric | Type | Labels | Description | @@ -221,9 +229,23 @@ The labels take these values: `PeerRelay` requests do not increment this counter on the receiving replica; they appear in that replica's gRPC request counter instead. This counter does not measure active connections, unique client requests, or unfinished attempts. -- `reason` is `replica_capacity` when the gateway replica's pending-relay budget - is full, or `sandbox_capacity` when one sandbox's budget on that replica, 32 - pending relays, is full. +- `reason` on `openshell_server_relay_rejected_total` is `replica_capacity` + when the gateway replica's pending-relay budget is full, or + `sandbox_capacity` when one sandbox's budget on that replica, 32 pending + relays, is full. +- `reason` on `openshell_server_supervisor_session_redirects_total` names the + step that sent the redirect, currently `connect` when this replica answers a + connecting supervisor with a redirect, or `shutdown` when a stopping replica + hands an established session to a remaining replica. A stopping replica's + redirects at connect time also count as `connect`. A stopping replica records + its redirects in its last seconds before exit, so a scrape can miss some or + all of them, and the counter is a lower bound for rollout handoffs. Under + `supervisor_topology: gateway` the shared supervisor's host moves sessions, so + these moves never appear in this counter; a climbing `connect` count then + measures gateway redirect attempts, which the supervisor usually undoes by + returning to its home within 5 seconds, not completed moves. For pool sessions + the `shutdown` count is likewise a lower bound, because the supervisor prefers + its home within 5 seconds. - `outcome` is `success`, `local_error`, or `remote_error`. A relay succeeds when the supervisor claims it; this does not measure the data stream. `local_error` means the attempt failed on this replica, and `remote_error` @@ -244,7 +266,11 @@ The labels take these values: - `overdue` is `true` when the skipped sandbox was already past its provisioning deadline, otherwise `false`. -Gauges and counters exist from startup. They start at `0`, except +Gauges and counters exist from startup, except +`openshell_server_gateway_ring_members`, which appears once the replica joins +the placement ring, and +`openshell_server_supervisor_session_off_ring_admissions_total`, which appears +only on gateways that coordinate no shared supervisor. They start at `0`, except `openshell_server_relay_pending_capacity` and `openshell_server_mutation_lock_connections_capacity`, which start at their limits. `openshell_server_routed_request_attempts_total` starts with seven diff --git a/skills/debug-openshell-cluster/SKILL.md b/skills/debug-openshell-cluster/SKILL.md index 837603cb26..dbe88b5b4e 100644 --- a/skills/debug-openshell-cluster/SKILL.md +++ b/skills/debug-openshell-cluster/SKILL.md @@ -837,7 +837,7 @@ for pod in $(kubectl -n openshell get pod \ -o jsonpath='{range .items[?(@.spec.containers[0].name=="openshell-gateway")]}{.metadata.name}{" "}{end}'); do echo "${pod}" kubectl get --raw "/api/v1/namespaces/openshell/pods/${pod}:9090/proxy/metrics" \ - | grep -E '^openshell_server_(supervisor_sessions|relay_pending|relay_rejected_total|routed_request_attempts_total|mutation_lock_(timeouts_total|errors_total|connections))' + | grep -E '^openshell_server_(supervisor_sessions|supervisor_session_(redirects|off_ring_admissions)_total|gateway_ring_members|relay_pending|relay_rejected_total|routed_request_attempts_total|mutation_lock_(timeouts_total|errors_total|connections))' done kubectl -n openshell get hpa kubectl -n openshell describe hpa openshell @@ -859,7 +859,7 @@ kubectl -n openshell port-forward pod/ 9090:9090 >/dev/null & pf_pid=$! sleep 2 curl -s http://localhost:9090/metrics \ - | grep -E '^openshell_server_(supervisor_sessions|relay_pending|relay_rejected_total|routed_request_attempts_total|mutation_lock_(timeouts_total|errors_total|connections))' + | grep -E '^openshell_server_(supervisor_sessions|supervisor_session_(redirects|off_ring_admissions)_total|gateway_ring_members|relay_pending|relay_rejected_total|routed_request_attempts_total|mutation_lock_(timeouts_total|errors_total|connections))' kill "${pf_pid}" ```