diff --git a/CHANGELOG.md b/CHANGELOG.md index 527ebdef5..1905685c7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,7 +19,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - http-client: a caller can bound the response body (`http_request_bounded`, `PrpcClient::with_max_response_bytes`). Nothing is bounded by default — `dstack vmm logs --lines 100000` is a legitimate multi-megabyte fetch — but every client that talks to a guest agent opts in, in the gateway and in the VMM, because a CVM is untrusted and one of them polls on a timer against the whole fleet ### Fixed -- gateway: a node removed via `RemoveNode` silently rejoined the cluster the next time it started, because every node re-registers its own sync address on boot. Once tombstone GC is collecting, that comeback is worse than an annoyance: a stale data directory diverges from every digest, and the divergence repair's full re-exchange resurrects records whose deletes the cluster already collected. Removal now writes a durable marker — a live record, so the GC can never eat it — that every gateway's sync endpoints enforce; a removed node's envelopes are refused until an operator re-admits it with `SetNodeUrl` +- gateway: a node removed via `RemoveNode` silently rejoined the cluster the next time it started, because every node re-registers its own sync address on boot. Once tombstone GC is collecting, that comeback is worse than an annoyance: a stale data directory diverges from every digest, and the divergence repair's full re-exchange resurrects records whose deletes the cluster already collected. Removal now writes a durable marker — a live record, so the GC can never eat it — that every gateway's sync endpoints enforce; a removed node's envelopes are refused until an operator re-admits it with `SetNodeUrl`. The refused node counts HTTP 403 sync rejections (`dstack_gateway_sync_rejected_total`), including removal lockouts and app-identity mismatches. Every gateway also exposes `dstack_gateway_node_last_seen_timestamp_seconds` per known node, so a long-offline gateway is a one-line alert instead of an ack-watermark puzzle - gateway: deleted KV records left a tombstone that nothing ever collected, so every deregistered CVM stayed on disk for the life of the deployment. Tombstones every peer has acknowledged are now dropped once every `tombstone_gc_writes` replicated writes (default 10000, zero disables); the trigger counts replicated writes rather than reading a clock, so nodes in a cluster collect in the same window without depending on time synchronization. A `SetTombstoneGcConfig` admin RPC stores an operator override in the KV itself, replicating one pace to every node - gateway: `Admin.RemoveCvm` now reports the outcome of the `inst/` tombstone alone. A failure to delete associated override or telemetry records is logged instead of failing the call, so a removal that did take effect is no longer reported as failed — which also aborted the local routing cleanup that follows it. Re-issuing a removal still sweeps up override and telemetry records orphaned by an earlier partial failure - sdk: `get_compose_hash` in the Python SDK mutated the dictionary it was given, stripping `docker_config` and `requirements` from it, so hashing the same manifest twice returned two different digests -- the second one for a manifest missing those blocks diff --git a/dstack/gateway/src/kv/https_client.rs b/dstack/gateway/src/kv/https_client.rs index 7e185425a..871c3e861 100644 --- a/dstack/gateway/src/kv/https_client.rs +++ b/dstack/gateway/src/kv/https_client.rs @@ -156,6 +156,21 @@ pub struct HttpsClient { client: HyperClient, } +/// A non-success HTTP status, kept as a typed error in the chain so a caller +/// can react to a specific code. The sync path needs to tell an HTTP 403 +/// rejection apart from a peer that is down or broken without string-matching +/// a formatted error message. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct HttpStatusError(pub u16); + +impl std::fmt::Display for HttpStatusError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "request failed with HTTP status {}", self.0) + } +} + +impl std::error::Error for HttpStatusError {} + impl HttpsClient { async fn post_gzipped( &self, @@ -255,7 +270,7 @@ impl HttpsClient { .with_context(|| format!("failed to send request to {url}"))?; if !response.status().is_success() { - anyhow::bail!("request failed: {}", response.status()); + return Err(HttpStatusError(response.status().as_u16()).into()); } // Bounded like every other response: this is the bootnode GetPeers path, and @@ -271,7 +286,7 @@ impl HttpsClient { let status = response.status(); if !status.is_success() { - anyhow::bail!("request failed: {status}"); + return Err(HttpStatusError(status.as_u16()).into()); } let body = read_body_bounded(response.into_body()).await?; @@ -282,7 +297,7 @@ impl HttpsClient { pub async fn post_bytes_no_response(&self, url: &str, body: Vec) -> Result<()> { let response = self.post_gzipped(url, body).await?; if !response.status().is_success() { - anyhow::bail!("request failed: {}", response.status()); + return Err(HttpStatusError(response.status().as_u16()).into()); } Ok(()) } @@ -514,6 +529,22 @@ mod transport_tests { .await } + /// The status code survives into the error chain as a typed value: the + /// sync path tells an HTTP 403 rejection apart from a peer that is down or + /// broken, and string-matching a Display line is not a contract. + #[tokio::test] + async fn a_refusal_status_is_readable_from_the_error_chain() { + let err = request(StatusCode::FORBIDDEN, Vec::new()) + .await + .expect_err("403 must not read as success"); + let status = err + .chain() + .filter_map(|cause| cause.downcast_ref::()) + .next() + .expect("the status must be present as a typed error"); + assert_eq!(*status, HttpStatusError(403)); + } + /// A non-success status must never be decoded as a successful sync response. #[tokio::test] async fn a_server_error_is_rejected() { diff --git a/dstack/gateway/src/kv/sync_service.rs b/dstack/gateway/src/kv/sync_service.rs index 6b621a6af..a999a4fb0 100644 --- a/dstack/gateway/src/kv/sync_service.rs +++ b/dstack/gateway/src/kv/sync_service.rs @@ -11,7 +11,7 @@ use std::sync::Arc; use anyhow::{Context, Result}; use dstack_gateway_rpc::GetPeersResponse; -use tracing::{info, warn}; +use tracing::{error, info, warn}; use wavekv::{ sync::{ ExchangeInterface, PeerLinkStatus, SyncConfig as KvSyncConfig, SyncEnvelope, SyncManager, @@ -23,7 +23,7 @@ use wavekv::{ use crate::config::SyncConfig as GwSyncConfig; -use super::https_client::{HttpsClient, HttpsClientConfig}; +use super::https_client::{HttpStatusError, HttpsClient, HttpsClientConfig}; use super::KvStore; /// HTTP-based network transport for WaveKV sync. @@ -40,6 +40,26 @@ pub struct HttpSyncNetwork { } impl HttpSyncNetwork { + /// Whether a peer rejected a send with HTTP 403. + fn was_rejected(err: &anyhow::Error) -> bool { + err.chain() + .filter_map(|cause| cause.downcast_ref::()) + .any(|status| status.0 == 403) + } + + /// Count a 403 for this node's own monitoring. Sync endpoints use 403 for + /// both removal lockouts and app-identity mismatches, so the metric and log + /// deliberately report a rejection without claiming which condition caused it. + fn note_if_rejected(err: &anyhow::Error, peer: NodeId) { + if !Self::was_rejected(err) { + return; + } + crate::metrics::record_sync_rejected(); + error!( + "peer {peer} rejected this node's sync envelope (HTTP 403); \ + this can indicate a removal lockout or an app-identity mismatch" + ); + } /// `my_uuid` is passed in rather than read back out of the store. /// /// Our own uuid is local configuration, not replicated state, and sourcing @@ -105,7 +125,8 @@ impl ExchangeInterface for HttpSyncNetwork { .client .post_bytes_response(&sync_url, env.encode()?) .await - .with_context(|| format!("failed to sync to peer {peer} at {sync_url}"))?; + .with_context(|| format!("failed to sync to peer {peer} at {sync_url}")) + .inspect_err(|err| Self::note_if_rejected(err, peer))?; self.kv_store.update_peer_last_seen(peer); Ok(Some(SyncEnvelope::decode(&body)?)) @@ -118,7 +139,8 @@ impl ExchangeInterface for HttpSyncNetwork { self.client .post_bytes_no_response(&push_url, env.encode()?) .await - .with_context(|| format!("failed to push to peer {peer} at {push_url}"))?; + .with_context(|| format!("failed to push to peer {peer} at {push_url}")) + .inspect_err(|err| Self::note_if_rejected(err, peer))?; Ok(()) } } @@ -309,3 +331,45 @@ pub async fn fetch_peers_from_bootnode( Ok(()) } + +#[cfg(test)] +mod sync_rejection_tests { + use super::*; + + fn wrapped(status: u16) -> anyhow::Error { + anyhow::Error::new(HttpStatusError(status)).context( + "failed to sync to peer 2 at https://gw2.example.com:9202/wavekv/sync/persistent", + ) + } + + /// Only HTTP 403 is a rejection; transport errors and other HTTP failures + /// must not increment the rejection counter. + #[test] + fn only_a_403_reads_as_rejected() { + assert!(HttpSyncNetwork::was_rejected(&wrapped(403))); + for status in [400u16, 401, 404, 500, 503] { + assert!( + !HttpSyncNetwork::was_rejected(&wrapped(status)), + "status {status} must not read as a rejection" + ); + } + assert!(!HttpSyncNetwork::was_rejected(&anyhow::anyhow!( + "connection refused" + ))); + } + + /// The rejection must reach the node's own monitoring. + #[test] + fn a_rejection_is_counted_for_the_senders_own_monitoring() { + // Process-wide static: assert on the delta, never the absolute value. + let before = crate::metrics::sync_rejected_count(); + HttpSyncNetwork::note_if_rejected(&wrapped(500), 2); + assert_eq!( + crate::metrics::sync_rejected_count(), + before, + "an ordinary failure must not count as a rejection" + ); + HttpSyncNetwork::note_if_rejected(&wrapped(403), 2); + assert!(crate::metrics::sync_rejected_count() > before); + } +} diff --git a/dstack/gateway/src/metrics.rs b/dstack/gateway/src/metrics.rs index 89bb0f571..bf0421902 100644 --- a/dstack/gateway/src/metrics.rs +++ b/dstack/gateway/src/metrics.rs @@ -56,6 +56,7 @@ static WG_RECONFIGURE_TOTAL: AtomicU64 = AtomicU64::new(0); static WG_RECONFIGURE_FAILURES: AtomicU64 = AtomicU64::new(0); static KV_PERSIST_FAILURES: AtomicU64 = AtomicU64::new(0); static KV_WAL_SYNC_FAILURES: AtomicU64 = AtomicU64::new(0); +static SYNC_REJECTED: AtomicU64 = AtomicU64::new(0); thread_local! { /// Set while a scrape is sampling live state. @@ -128,6 +129,17 @@ pub(crate) fn record_kv_wal_sync_failure() { KV_WAL_SYNC_FAILURES.fetch_add(1, Ordering::Relaxed); } +/// Record a sync envelope of ours that a peer rejected with HTTP 403. +pub(crate) fn record_sync_rejected() { + SYNC_REJECTED.fetch_add(1, Ordering::Relaxed); +} + +/// Test-only read: the counter is process-wide, so tests assert on deltas. +#[cfg(test)] +pub(crate) fn sync_rejected_count() -> u64 { + SYNC_REJECTED.load(Ordering::Relaxed) +} + /// Index of the longest prefix in `prefixes` that `key` starts with. /// /// Longest rather than first so that adding a narrower prefix later (say @@ -163,6 +175,9 @@ pub(crate) struct Snapshot { pub stores: Vec, /// domain -> certificate `notAfter`, in seconds since the epoch. pub cert_not_after: Vec<(String, u64)>, + /// node id -> the latest time any gateway observed the node, in seconds + /// since the epoch (max across observers, future-dated reports dropped). + pub node_last_seen: Vec<(u32, u64)>, } /// One WaveKV store (`persistent` or `ephemeral`). @@ -362,6 +377,28 @@ pub(crate) fn render(snapshot: &Snapshot) -> String { "", KV_WAL_SYNC_FAILURES.load(Ordering::Relaxed), ); + counter( + &mut out, + "dstack_gateway_sync_rejected_total", + "Sync envelopes this node sent that a peer rejected with HTTP 403. This can indicate a removal lockout or an app-identity mismatch.", + "", + SYNC_REJECTED.load(Ordering::Relaxed), + ); + + header( + &mut out, + "dstack_gateway_node_last_seen_timestamp_seconds", + "Latest time any gateway observed the node, unix seconds. Replicated (max across observers): every node reports roughly the same value, so aggregate with max(), and alert on time() minus this. A node an operator removed leaves the series entirely -- absence here plus dstack_gateway_sync_rejected_total on the machine itself can indicate the removed-by-mistake picture.", + "gauge", + ); + for (node_id, last_seen) in &snapshot.node_last_seen { + line( + &mut out, + "dstack_gateway_node_last_seen_timestamp_seconds", + &format!("{{node=\"{node_id}\"}}"), + *last_seen, + ); + } gauge( &mut out, @@ -490,6 +527,7 @@ mod tests { }], }], cert_not_after: vec![("app.example.com".to_string(), 1_800_000_000)], + node_last_seen: vec![(2, 1_777_000_000)], } } @@ -529,6 +567,7 @@ mod tests { "dstack_gateway_cluster_kv_keys{store=\"persistent\"} 42", "dstack_gateway_kv_dirty{store=\"persistent\"} 1", "dstack_gateway_cluster_cert_not_after_seconds{domain=\"app.example.com\"} 1800000000", + "dstack_gateway_node_last_seen_timestamp_seconds{node=\"2\"} 1777000000", ] { assert!(rendered.contains(expected), "missing sample: {expected}"); } diff --git a/dstack/gateway/src/web_routes/metrics.rs b/dstack/gateway/src/web_routes/metrics.rs index 6a6aa132e..bdad9cbcb 100644 --- a/dstack/gateway/src/web_routes/metrics.rs +++ b/dstack/gateway/src/web_routes/metrics.rs @@ -49,6 +49,20 @@ fn sample(state: &State) -> Snapshot { .map(|(domain, data)| (domain, data.not_after)) .collect(); + // One pass over the node table plus one ephemeral read per node -- the + // cost the counting fast-path above avoids, paid here deliberately, + // because a per-node timestamp is the point: a node whose value stops + // advancing is offline, however the operator's alerting words it. + let node_last_seen = kv_store + .load_all_nodes() + .into_keys() + .filter_map(|id| { + kv_store + .get_node_latest_last_seen(id) + .map(|last_seen| (id, last_seen)) + }) + .collect(); + Snapshot { version: crate::app_version(), node_id: kv_store.my_node_id(), @@ -59,6 +73,7 @@ fn sample(state: &State) -> Snapshot { accel, stores, cert_not_after, + node_last_seen, } }