diff --git a/docs/dstack-gateway.md b/docs/dstack-gateway.md index 2de511702..58e59a1bc 100644 --- a/docs/dstack-gateway.md +++ b/docs/dstack-gateway.md @@ -88,3 +88,49 @@ insecure_no_auth = false The admin server is fail-closed: if it is enabled with no `admin_token` and no `htpasswd_file`, and `insecure_no_auth` is `false`, it refuses to start rather than exposing an unauthenticated admin API. Clients authenticate by sending `Authorization: Bearer ` or the `X-Admin-Token: ` header. + +## Metrics + +The admin server exposes Prometheus metrics at `GET /metrics`. It is part of the +admin API, so it is only reachable when `core.admin.enabled` is true and it +requires the same credentials — unless `insecure_no_auth` is set, which exposes +it along with the rest of the admin API. The series name domains, node ids and +instance counts, which is topology that should not be readable without +authentication. + +```yaml +scrape_configs: + - job_name: dstack-gateway + static_configs: + - targets: [""] + authorization: + credentials: "" +``` + +### Cluster-scoped vs node-local series + +`dstack_gateway_cluster_*` describes replicated state: every node in the cluster +reports the same value, so summing across targets multiplies it by the number of +nodes. Everything else describes what one process did and sums normally. + +```promql +# Instances in the routing table — replicated, so take one node's view +max(dstack_gateway_cluster_instances) + +# Connections across the fleet — node-local, so add them up +sum(dstack_gateway_connections) + +# Nodes disagreeing about who is up: this is the replication-lag signal +max(dstack_gateway_cluster_nodes_active) - min(dstack_gateway_cluster_nodes_active) +``` + +### Series worth alerting on + +| Metric | Why | +|---|---| +| `dstack_gateway_wg_reconfigure_failures_total` | The gateway could not push a WireGuard config: it failed to render, failed to write, or `wg syncconf` rejected the whole file over one bad peer stanza. Routing updates have stopped reaching the data plane while the gateway still looks healthy. | +| `dstack_gateway_kv_decode_failures_total` | A replicated record that fails to decode is skipped, which makes the CVM behind it silently unroutable. Labelled by key prefix. Alert on `> 0`; the magnitude counts how often a bad record was *read*, not how many are bad, so do not read it as a severity. | +| `dstack_gateway_kv_peer_buffered_logs` | Entries still buffered for a peer. Sustained growth means that peer stopped acknowledging and the two nodes are drifting apart. | +| `dstack_gateway_cluster_cert_not_after_seconds` | Certificate expiry per domain; alert on `- time()` falling under the renewal window. Capped at 256 series — compare `dstack_gateway_cluster_cert_domains` to see whether the cap was hit. | +| `dstack_gateway_kv_persist_failures_total` | Periodic snapshots are failing, so a restart replays a growing WAL. | +| `dstack_gateway_kv_persist_failures_total` | Periodic snapshots are failing, so a restart replays a growing WAL. | diff --git a/dstack/gateway/src/kv/mod.rs b/dstack/gateway/src/kv/mod.rs index c1e6ecb02..7fe763407 100644 --- a/dstack/gateway/src/kv/mod.rs +++ b/dstack/gateway/src/kv/mod.rs @@ -265,6 +265,8 @@ pub mod keys { pub const CERT_PREFIX: &str = "cert/"; pub const DNS_CRED_PREFIX: &str = "dns_cred/"; pub const DNS_CRED_DEFAULT: &str = "dns_cred_default"; + /// Shared by the `GLOBAL_*` keys below; not itself a key. + pub const GLOBAL_PREFIX: &str = "global/"; pub const GLOBAL_CERTBOT_CONFIG: &str = "global/certbot_config"; pub const GLOBAL_ACME_CREDENTIALS: &str = "global/acme_credentials"; pub const GLOBAL_ACME_ATTESTATION: &str = "global/acme_attestation"; @@ -394,6 +396,7 @@ impl GetPutCodec for NodeState { .and_then(|entry| match decode(entry.value.as_ref()?) { Ok(value) => Some(value), Err(e) => { + crate::metrics::record_decode_failure(key); warn!("failed to decode value for key {key}: {e:?}"); None } @@ -414,6 +417,7 @@ impl GetPutCodec for NodeState { let value = match decode(entry.value.as_ref()?) { Ok(value) => value, Err(e) => { + crate::metrics::record_decode_failure(key); warn!("failed to decode value for key {key}: {e:?}"); return None; } @@ -430,6 +434,7 @@ impl GetPutCodec for NodeState { let value = match decode(entry.value.as_ref()?) { Ok(value) => value, Err(e) => { + crate::metrics::record_decode_failure(key); warn!("failed to decode value for key {key}: {e:?}"); return None; } @@ -580,6 +585,32 @@ impl KvStore { .collect() } + /// Whether a node counts as active. A node with no recorded status is up. + /// + /// The routing path and the metrics sampler both filter on this, and they + /// have to agree: a gauge that counts a node the router has dropped is + /// describing a routing table that does not exist. + pub(crate) fn node_is_active(status: Option<&NodeStatus>) -> bool { + !matches!(status, Some(NodeStatus::Down)) + } + + /// Count all and active nodes, without materialising `GatewayNodeInfo`. + /// + /// A scrape wants two numbers. Reaching them through `get_all_nodes()` and + /// `get_active_nodes()` instead means loading the node table twice, cloning + /// five strings per node, and taking the ephemeral lock once per node for a + /// `last_seen` that the count never reads -- all of it under the proxy lock + /// that the data path takes on every connection. + pub fn count_nodes(&self) -> (u64, u64) { + let statuses = self.load_all_node_statuses(); + let nodes = self.load_all_nodes(); + let active = nodes + .keys() + .filter(|id| Self::node_is_active(statuses.get(id))) + .count() as u64; + (nodes.len() as u64, active) + } + // ==================== Connection Count Sync ==================== /// Sync connection count for an instance (from this node) @@ -833,6 +864,7 @@ impl KvStore { match decode(value) { Ok(config) => Some(config), Err(e) => { + crate::metrics::record_decode_failure(key); warn!("failed to decode cert config for key {key}: {e:?}"); None } @@ -905,6 +937,7 @@ impl KvStore { match decode(value) { Ok(data) => Some((domain.to_string(), data)), Err(e) => { + crate::metrics::record_decode_failure(key); warn!("failed to decode cert data for key {key}: {e:?}"); None } @@ -1100,6 +1133,7 @@ impl KvStore { match decode(value) { Ok(att) => Some(att), Err(e) => { + crate::metrics::record_decode_failure(key); warn!("failed to decode attestation for key {key}: {e:?}"); None } diff --git a/dstack/gateway/src/main.rs b/dstack/gateway/src/main.rs index e7b2eb1a2..fc39c4e52 100644 --- a/dstack/gateway/src/main.rs +++ b/dstack/gateway/src/main.rs @@ -29,6 +29,7 @@ mod debug_service; mod distributed_certbot; mod kv; mod main_service; +mod metrics; mod models; mod pp; mod proxy; diff --git a/dstack/gateway/src/main_service.rs b/dstack/gateway/src/main_service.rs index 6167eaf1c..ee8b16fea 100644 --- a/dstack/gateway/src/main_service.rs +++ b/dstack/gateway/src/main_service.rs @@ -831,7 +831,10 @@ fn start_wavekv_watch_task(proxy: Proxy) -> Result<()> { match kv_store_for_persist.persist_if_dirty() { Ok(true) => info!("WaveKV: periodic persist completed"), Ok(false) => {} // No changes to persist - Err(err) => error!("WaveKV: periodic persist failed: {err:?}"), + Err(err) => { + crate::metrics::record_kv_persist_failure(); + error!("WaveKV: periodic persist failed: {err:?}"); + } } } }); @@ -1169,6 +1172,20 @@ impl ProxyState { } pub(crate) fn reconfigure(&mut self) -> Result<()> { + // Every way out of here that is not a clean apply leaves the data plane + // on the routing table it already had, so they all feed one counter -- + // the early returns included. A config that cannot be rendered or + // written never reaches `wg` at all, and both call sites of this + // function only log the `Err`, so a full disk would otherwise look + // exactly like having nothing to apply. + let result = self.reconfigure_inner(); + if result.is_err() { + crate::metrics::record_wg_reconfigure(false); + } + result + } + + fn reconfigure_inner(&mut self) -> Result<()> { let wg_config = self.generate_wg_config()?; // the rendered config carries the interface's WireGuard private key. safe_write_with_mode(&self.config.wg.config_path, wg_config, 0o600) @@ -1178,8 +1195,18 @@ impl ProxyState { let config_path = &self.config.wg.config_path; match cmd!(wg syncconf $ifname $config_path) { - Ok(_) => info!("wg config updated"), - Err(err) => error!("failed to set wg config: {err:?}"), + Ok(_) => { + crate::metrics::record_wg_reconfigure(true); + info!("wg config updated"); + } + Err(err) => { + // `wg syncconf` rejects the whole file when one peer stanza is + // bad, and this stays `Ok` for the caller as it always has, so + // the counter is the only signal that routing updates stopped + // reaching the data plane. + crate::metrics::record_wg_reconfigure(false); + error!("failed to set wg config: {err:?}"); + } } Ok(()) } @@ -1455,16 +1482,9 @@ impl ProxyState { self.kv_store .load_all_nodes() .into_iter() - .filter(|(id, _)| { - if !exclude_down { - return true; - } - // Exclude nodes with status "down" - match node_statuses.get(id) { - Some(NodeStatus::Down) => false, - _ => true, // Include Up or nodes without explicit status - } - }) + // Shared with the metrics sampler so the gauge and the routing + // table cannot disagree about what "active" means. + .filter(|(id, _)| !exclude_down || KvStore::node_is_active(node_statuses.get(id))) .map(|(id, node)| GatewayNodeInfo { id, uuid: node.uuid, diff --git a/dstack/gateway/src/metrics.rs b/dstack/gateway/src/metrics.rs new file mode 100644 index 000000000..7ec1a6f4c --- /dev/null +++ b/dstack/gateway/src/metrics.rs @@ -0,0 +1,697 @@ +// SPDX-FileCopyrightText: © 2026 Phala Network +// +// SPDX-License-Identifier: Apache-2.0 + +//! Prometheus metrics for the gateway. +//! +//! Counters live in process-wide atomics rather than on `Proxy` because the +//! sites that need them -- the KV codec, `reconfigure()` -- run on code paths +//! that hold no handle to the proxy state. `proxy::NUM_CONNECTIONS` and +//! `proxy::stats` already work this way. +//! +//! Gauges are not stored: they are sampled from live state when a scrape +//! arrives, so a scrape never has to be kept in sync with a mutation. +//! +//! Label values that come from replicated state (domains, and therefore +//! anything a peer can name) are escaped, and the decode-failure label set is +//! a fixed list of known prefixes plus `other`, so a peer cannot inflate +//! cardinality by inventing keys. + +use std::cell::Cell; +use std::fmt::Write as _; +use std::sync::atomic::{AtomicU64, Ordering}; + +use dstack_gateway_rpc::ProxyAccelStatus; + +use crate::kv::keys; + +/// Key prefixes that get their own decode-failure series. +/// +/// Taken from `kv::keys` rather than spelled out here: a prefix that is +/// renamed there and not here would not break anything loudly, it would just +/// start counting that key space under `other`. +/// +/// Anything unmatched lands in `other`, which is what keeps a peer from +/// inventing key spaces to inflate cardinality. +const METERED_PREFIXES: [&str; 9] = [ + keys::INST_PREFIX, + keys::NODE_PREFIX, + keys::CONN_PREFIX, + keys::HANDSHAKE_PREFIX, + keys::LAST_SEEN_NODE_PREFIX, + keys::PEER_ADDR_PREFIX, + keys::CERT_PREFIX, + keys::DNS_CRED_PREFIX, + keys::GLOBAL_PREFIX, +]; + +const OTHER_PREFIX: &str = "other"; + +/// Ceiling on `cert_not_after` series, far above any real deployment. +const MAX_CERT_SERIES: usize = 256; + +static DECODE_FAILURES: [AtomicU64; METERED_PREFIXES.len() + 1] = + [const { AtomicU64::new(0) }; METERED_PREFIXES.len() + 1]; +static WG_RECONFIGURE_TOTAL: AtomicU64 = AtomicU64::new(0); +static WG_RECONFIGURE_FAILURES: AtomicU64 = AtomicU64::new(0); +static KV_PERSIST_FAILURES: AtomicU64 = AtomicU64::new(0); + +thread_local! { + /// Set while a scrape is sampling live state. + /// + /// A scrape reads the same replicated records the data path reads, through + /// the same decoding helpers, so without this it would feed the very + /// counter it is about to report: one permanently corrupt record would + /// increment `decode_failures` once per scrape forever, and `rate()` over + /// it would measure the scrape interval rather than anything about the + /// store. Observing must not be indistinguishable from failing. + static SAMPLING: Cell = const { Cell::new(false) }; +} + +/// Suppresses decode-failure counting until dropped. +/// +/// Correctness depends on the sampler staying synchronous: it must not yield +/// to the runtime while this is alive, or the flag would apply to whatever +/// else the runtime schedules onto this thread. +pub(crate) struct ScrapeGuard(bool); + +impl Drop for ScrapeGuard { + fn drop(&mut self) { + SAMPLING.with(|sampling| sampling.set(self.0)); + } +} + +/// Mark the current thread as sampling for a scrape. See [`ScrapeGuard`]. +#[must_use = "decode-failure suppression ends as soon as the guard is dropped"] +pub(crate) fn scrape_guard() -> ScrapeGuard { + ScrapeGuard(SAMPLING.with(|sampling| sampling.replace(true))) +} + +/// Record that a replicated value could not be decoded. +/// +/// A decode failure makes the record invisible to the data plane with nothing +/// but a log line to say so, which is how a single corrupt record turns into +/// "that CVM silently stopped being routable". +/// +/// Counting is suppressed while a scrape samples; see [`ScrapeGuard`]. +pub(crate) fn record_decode_failure(key: &str) { + if SAMPLING.with(Cell::get) { + return; + } + DECODE_FAILURES[prefix_index(key)].fetch_add(1, Ordering::Relaxed); +} + +/// Record the outcome of pushing a new WireGuard config. +/// +/// Covers the whole of `reconfigure()`, not just `wg syncconf`: rendering and +/// writing the config can fail too, and all three leave the data plane on its +/// previous routing table while the gateway keeps answering. `wg syncconf` +/// additionally rejects the *whole* file when one peer stanza is bad, and its +/// call site can only log that, so without a counter a gateway that stopped +/// applying routing updates looks healthy. +pub(crate) fn record_wg_reconfigure(ok: bool) { + WG_RECONFIGURE_TOTAL.fetch_add(1, Ordering::Relaxed); + if !ok { + WG_RECONFIGURE_FAILURES.fetch_add(1, Ordering::Relaxed); + } +} + +/// Record a failed periodic snapshot. Repeated failures mean the node is one +/// restart away from replaying a very long WAL, or from losing the writes it +/// never managed to snapshot. +pub(crate) fn record_kv_persist_failure() { + KV_PERSIST_FAILURES.fetch_add(1, 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 +/// `node/status/` next to `node/`) routes keys to the narrower series instead +/// of depending on array order. The current set does not overlap, so this is +/// here to keep the next addition from being a silent mis-bucketing. +fn longest_prefix_index(prefixes: &[&str], key: &str) -> Option { + prefixes + .iter() + .enumerate() + .filter(|(_, prefix)| key.starts_with(*prefix)) + .max_by_key(|(_, prefix)| prefix.len()) + .map(|(index, _)| index) +} + +fn prefix_index(key: &str) -> usize { + longest_prefix_index(&METERED_PREFIXES, key).unwrap_or(METERED_PREFIXES.len()) +} + +fn prefix_label(index: usize) -> &'static str { + METERED_PREFIXES.get(index).copied().unwrap_or(OTHER_PREFIX) +} + +/// Live state sampled for one scrape. +pub(crate) struct Snapshot { + pub version: String, + pub node_id: u32, + pub instances: u64, + pub connections: u64, + pub nodes_total: u64, + pub nodes_active: u64, + pub accel: ProxyAccelStatus, + pub stores: Vec, + /// domain -> certificate `notAfter`, in seconds since the epoch. + pub cert_not_after: Vec<(String, u64)>, +} + +/// One WaveKV store (`persistent` or `ephemeral`). +pub(crate) struct StoreSnapshot { + pub name: &'static str, + pub keys: u64, + pub next_seq: u64, + pub dirty: bool, + pub peers: Vec, +} + +pub(crate) struct PeerSnapshot { + pub id: u32, + /// How far we have consumed this peer's log. + pub local_ack: u64, + /// How far the peer says it has consumed ours. + pub peer_ack: u64, + /// Entries still buffered for the peer. A number that only grows is a peer + /// that stopped acknowledging. + pub buffered_logs: u64, +} + +/// Render the Prometheus text exposition format. +pub(crate) fn render(snapshot: &Snapshot) -> String { + let mut out = String::with_capacity(2048); + + gauge( + &mut out, + "dstack_gateway_build_info", + "Gateway build information.", + &format!( + "{{version=\"{}\",node_id=\"{}\"}}", + escape_label(&snapshot.version), + snapshot.node_id + ), + 1, + ); + gauge( + &mut out, + "dstack_gateway_cluster_instances", + "CVM instances currently in the routing table. Replicated: every node reports the same value, so aggregate with max(), not sum().", + "", + snapshot.instances, + ); + gauge( + &mut out, + "dstack_gateway_connections", + "Proxy connections currently open.", + "", + snapshot.connections, + ); + gauge( + &mut out, + "dstack_gateway_cluster_nodes", + "Gateway nodes known to the cluster. Replicated: aggregate with max(), not sum().", + "", + snapshot.nodes_total, + ); + gauge( + &mut out, + "dstack_gateway_cluster_nodes_active", + "Gateway nodes this node does not consider down. Replicated state seen locally, so disagreement between nodes is itself the replication-lag signal.", + "", + snapshot.nodes_active, + ); + + counter( + &mut out, + "dstack_gateway_ktls_offloaded_total", + "Connections handed to the kernel TLS ULP.", + "", + snapshot.accel.ktls_offloaded, + ); + counter( + &mut out, + "dstack_gateway_ktls_offload_failed_total", + "Connections whose kernel TLS handover failed.", + "", + snapshot.accel.ktls_offload_failed, + ); + counter( + &mut out, + "dstack_gateway_splice_engaged_total", + "Connections that entered a zero-copy splice relay.", + "", + snapshot.accel.splice_engaged, + ); + + header( + &mut out, + "dstack_gateway_cluster_kv_keys", + "Keys held in a WaveKV store. Replicated: aggregate with max(), not sum().", + "gauge", + ); + for store in &snapshot.stores { + line( + &mut out, + "dstack_gateway_cluster_kv_keys", + &store_label(store), + store.keys, + ); + } + header( + &mut out, + "dstack_gateway_kv_next_seq", + "Next sequence number this node will assign in a WaveKV store.", + "gauge", + ); + for store in &snapshot.stores { + line( + &mut out, + "dstack_gateway_kv_next_seq", + &store_label(store), + store.next_seq, + ); + } + header( + &mut out, + "dstack_gateway_kv_dirty", + "1 when a WaveKV store holds changes that are not in its snapshot.", + "gauge", + ); + for store in &snapshot.stores { + line( + &mut out, + "dstack_gateway_kv_dirty", + &store_label(store), + u64::from(store.dirty), + ); + } + + header( + &mut out, + "dstack_gateway_kv_peer_local_ack", + "How far this node has consumed a peer's log.", + "gauge", + ); + for (store, peer) in peers(snapshot) { + line( + &mut out, + "dstack_gateway_kv_peer_local_ack", + &peer_label(store, peer), + peer.local_ack, + ); + } + header( + &mut out, + "dstack_gateway_kv_peer_remote_ack", + "How far a peer reports having consumed this node's log.", + "gauge", + ); + for (store, peer) in peers(snapshot) { + line( + &mut out, + "dstack_gateway_kv_peer_remote_ack", + &peer_label(store, peer), + peer.peer_ack, + ); + } + header( + &mut out, + "dstack_gateway_kv_peer_buffered_logs", + "Log entries still buffered for a peer. Sustained growth means the peer stopped acknowledging.", + "gauge", + ); + for (store, peer) in peers(snapshot) { + line( + &mut out, + "dstack_gateway_kv_peer_buffered_logs", + &peer_label(store, peer), + peer.buffered_logs, + ); + } + + header( + &mut out, + "dstack_gateway_kv_decode_failures_total", + "Replicated values that could not be decoded, by key prefix.", + "counter", + ); + for (index, counter) in DECODE_FAILURES.iter().enumerate() { + line( + &mut out, + "dstack_gateway_kv_decode_failures_total", + &format!("{{prefix=\"{}\"}}", escape_label(prefix_label(index))), + counter.load(Ordering::Relaxed), + ); + } + + counter( + &mut out, + "dstack_gateway_wg_reconfigure_total", + "WireGuard config applications attempted.", + "", + WG_RECONFIGURE_TOTAL.load(Ordering::Relaxed), + ); + counter( + &mut out, + "dstack_gateway_wg_reconfigure_failures_total", + "WireGuard config applications that did not reach the data plane: render failure, write failure, or a config wg syncconf rejected.", + "", + WG_RECONFIGURE_FAILURES.load(Ordering::Relaxed), + ); + counter( + &mut out, + "dstack_gateway_kv_persist_failures_total", + "Periodic WaveKV snapshots that failed.", + "", + KV_PERSIST_FAILURES.load(Ordering::Relaxed), + ); + + gauge( + &mut out, + "dstack_gateway_cluster_cert_domains", + "Domains holding certificate data. Exceeding the number of cert_not_after series means the series were truncated.", + "", + snapshot.cert_not_after.len() as u64, + ); + header( + &mut out, + "dstack_gateway_cluster_cert_not_after_seconds", + "Certificate expiry per domain, in seconds since the epoch. Replicated: every node reports the same series.", + "gauge", + ); + // Domains are only created through the admin API, so in practice this is a + // handful of wildcard certificates. The cap is for the case where it is + // not: the records are replicated, so a peer with write access could turn + // one label into an unbounded series count. Truncation is by domain order + // rather than by expiry so the exported set does not flap between scrapes; + // `cert_domains` above is what tells you it happened. + for (domain, not_after) in snapshot.cert_not_after.iter().take(MAX_CERT_SERIES) { + line( + &mut out, + "dstack_gateway_cluster_cert_not_after_seconds", + &format!("{{domain=\"{}\"}}", escape_label(domain)), + *not_after, + ); + } + + out +} + +fn peers(snapshot: &Snapshot) -> impl Iterator { + snapshot + .stores + .iter() + .flat_map(|store| store.peers.iter().map(move |peer| (store, peer))) +} + +fn store_label(store: &StoreSnapshot) -> String { + format!("{{store=\"{}\"}}", escape_label(store.name)) +} + +fn peer_label(store: &StoreSnapshot, peer: &PeerSnapshot) -> String { + format!( + "{{store=\"{}\",peer=\"{}\"}}", + escape_label(store.name), + peer.id + ) +} + +fn header(out: &mut String, name: &str, help: &str, kind: &str) { + let _ = writeln!(out, "# HELP {name} {help}"); + let _ = writeln!(out, "# TYPE {name} {kind}"); +} + +fn line(out: &mut String, name: &str, labels: &str, value: u64) { + let _ = writeln!(out, "{name}{labels} {value}"); +} + +fn gauge(out: &mut String, name: &str, help: &str, labels: &str, value: u64) { + header(out, name, help, "gauge"); + line(out, name, labels, value); +} + +fn counter(out: &mut String, name: &str, help: &str, labels: &str, value: u64) { + header(out, name, help, "counter"); + line(out, name, labels, value); +} + +/// Escape a label value per the exposition format. +/// +/// Domains reach this from replicated state, so an unescaped quote or newline +/// would be a peer-controlled way to forge series in the scrape output. +/// +/// The format defines exactly three escapes: `\\`, `\"` and `\n`. Escaping +/// anything else is not the safer choice it looks like -- `prometheus/common`'s +/// parser, which backs `promtool check metrics` and most client tooling, +/// rejects an unknown escape sequence outright. Emitting `\t` would hand the +/// same hostile peer a cheaper attack than the one this function exists to +/// stop: one tab in a domain and the entire scrape stops parsing. Remaining +/// control characters are dropped instead, so the output is valid and carries +/// no raw control bytes either. +fn escape_label(value: &str) -> String { + let mut escaped = String::with_capacity(value.len()); + for ch in value.chars() { + match ch { + '\\' => escaped.push_str("\\\\"), + '"' => escaped.push_str("\\\""), + '\n' => escaped.push_str("\\n"), + _ if ch.is_control() => {} + _ => escaped.push(ch), + } + } + escaped +} + +#[cfg(test)] +mod tests { + use super::*; + + fn snapshot() -> Snapshot { + Snapshot { + version: "0.0.0-test".to_string(), + node_id: 7, + instances: 3, + connections: 12, + nodes_total: 3, + nodes_active: 2, + accel: ProxyAccelStatus { + ktls_mode: "off".to_string(), + splice_mode: "off".to_string(), + ktls_offloaded: 5, + ktls_offload_failed: 1, + splice_engaged: 4, + }, + stores: vec![StoreSnapshot { + name: "persistent", + keys: 42, + next_seq: 100, + dirty: true, + peers: vec![PeerSnapshot { + id: 2, + local_ack: 9, + peer_ack: 8, + buffered_logs: 1, + }], + }], + cert_not_after: vec![("app.example.com".to_string(), 1_800_000_000)], + } + } + + #[test] + fn every_series_is_declared_before_it_is_used() { + let rendered = render(&snapshot()); + let mut declared = std::collections::HashSet::new(); + for row in rendered.lines() { + if let Some(rest) = row.strip_prefix("# TYPE ") { + let name = rest.split(' ').next().unwrap_or_default(); + declared.insert(name.to_string()); + continue; + } + if row.starts_with('#') { + continue; + } + let name = row + .split(['{', ' ']) + .next() + .expect("a sample line names a series"); + assert!( + declared.contains(name), + "sample {name} appears without a # TYPE line" + ); + } + } + + #[test] + fn samples_carry_the_values_they_were_given() { + let rendered = render(&snapshot()); + for expected in [ + "dstack_gateway_build_info{version=\"0.0.0-test\",node_id=\"7\"} 1", + "dstack_gateway_cluster_instances 3", + "dstack_gateway_connections 12", + "dstack_gateway_cluster_nodes_active 2", + "dstack_gateway_ktls_offload_failed_total 1", + "dstack_gateway_cluster_kv_keys{store=\"persistent\"} 42", + "dstack_gateway_kv_dirty{store=\"persistent\"} 1", + "dstack_gateway_kv_peer_buffered_logs{store=\"persistent\",peer=\"2\"} 1", + "dstack_gateway_cluster_cert_not_after_seconds{domain=\"app.example.com\"} 1800000000", + ] { + assert!(rendered.contains(expected), "missing sample: {expected}"); + } + } + + #[test] + fn only_the_three_escapes_the_format_defines_are_emitted() { + // The exposition format defines \\, \" and \n and nothing else, and + // `prometheus/common`'s parser errors on any other escape sequence. A + // tab that reaches a label value must therefore be dropped rather than + // written as `\t`, which would cost the whole scrape -- a cheaper + // attack than the injection this escaping exists to stop. + assert_eq!(escape_label("a\tb\rc\u{7}d"), "abcd"); + assert_eq!(escape_label("a\\b\"c\nd"), "a\\\\b\\\"c\\nd"); + + let mut snapshot = snapshot(); + snapshot.cert_not_after = vec![("tab\there\rand\u{7}bell".to_string(), 1_800_000_000)]; + let rendered = render(&snapshot); + for undefined in ["\\t", "\\r"] { + assert!( + !rendered.contains(undefined), + "emitted `{undefined}`, an escape the exposition format does not define" + ); + } + assert!(rendered.contains( + "dstack_gateway_cluster_cert_not_after_seconds{domain=\"tabhereandbell\"} 1800000000" + )); + } + + #[test] + fn a_hostile_domain_cannot_forge_a_series() { + let mut snapshot = snapshot(); + // A domain arrives from replicated state, so treat it as peer-supplied. + snapshot.cert_not_after = vec![( + "evil\" 1\ndstack_gateway_cluster_instances 999\n#".to_string(), + 1_800_000_000, + )]; + let rendered = render(&snapshot); + + // The injection stays inside one label value on one line: no forged + // sample line, and the real gauge still reads what it was given. + let forged = rendered + .lines() + .filter(|row| row.starts_with("dstack_gateway_cluster_instances ")) + .count(); + assert_eq!( + forged, 1, + "the payload escaped its label and became a sample" + ); + assert!(rendered + .lines() + .any(|row| row == "dstack_gateway_cluster_instances 3")); + assert!(rendered.contains( + "dstack_gateway_cluster_cert_not_after_seconds{domain=\"evil\\\" 1\\ndstack_gateway_cluster_instances 999\\n#\"} 1800000000" + )); + } + + #[test] + fn decode_failures_are_bucketed_by_key_prefix() { + assert_eq!(prefix_label(prefix_index("inst/abc")), "inst/"); + // No narrower `node/` prefix is metered, so this folds into `node/`. + assert_eq!(prefix_label(prefix_index("node/status/3")), "node/"); + assert_eq!(prefix_label(prefix_index("cert/example.com/data")), "cert/"); + assert_eq!(prefix_label(prefix_index("__peer_addr/3")), "__peer_addr/"); + assert_eq!( + prefix_label(prefix_index("global/certbot_config")), + "global/" + ); + // A key a peer invented does not get a series of its own. + assert_eq!(prefix_label(prefix_index("whatever/1")), "other"); + assert_eq!(prefix_label(prefix_index("")), "other"); + } + + #[test] + fn a_narrower_prefix_wins_over_a_wider_one() { + // The metered set does not overlap today, so drive the rule directly: + // adding `node/status/` later must not depend on where in the array it + // lands. + let prefixes = ["node/", "node/status/"]; + assert_eq!(longest_prefix_index(&prefixes, "node/status/3"), Some(1)); + assert_eq!(longest_prefix_index(&prefixes, "node/info/3"), Some(0)); + + let reversed = ["node/status/", "node/"]; + assert_eq!(longest_prefix_index(&reversed, "node/status/3"), Some(0)); + + assert_eq!(longest_prefix_index(&prefixes, "inst/1"), None); + } + + #[test] + fn the_per_domain_expiry_series_is_capped() { + let mut snapshot = snapshot(); + let total = MAX_CERT_SERIES + 25; + snapshot.cert_not_after = (0..total) + .map(|i| (format!("d{i:04}.example.com"), 1_800_000_000 + i as u64)) + .collect(); + let rendered = render(&snapshot); + + let exported = rendered + .lines() + .filter(|row| row.starts_with("dstack_gateway_cluster_cert_not_after_seconds{")) + .count(); + assert_eq!(exported, MAX_CERT_SERIES, "the cap did not hold"); + // The real count still reaches the operator, so truncation is visible + // rather than silent. + assert!(rendered + .lines() + .any(|row| row == format!("dstack_gateway_cluster_cert_domains {total}"))); + } + + #[test] + fn a_scrape_does_not_feed_the_counter_it_reports() { + // Process-wide statics: assert on deltas, never on absolute values. + let bucket = &DECODE_FAILURES[prefix_index("conn/probe")]; + let before = bucket.load(Ordering::Relaxed); + { + let _guard = scrape_guard(); + record_decode_failure("conn/probe"); + // Nested guards must not end suppression early. + { + let _inner = scrape_guard(); + record_decode_failure("conn/probe"); + } + record_decode_failure("conn/probe"); + } + assert_eq!( + bucket.load(Ordering::Relaxed), + before, + "a scrape counted its own reads" + ); + + record_decode_failure("conn/probe"); + assert_eq!( + bucket.load(Ordering::Relaxed), + before + 1, + "suppression outlived the scrape" + ); + } + + #[test] + fn recording_a_failure_moves_its_own_bucket_only() { + // Process-wide statics: assert on deltas, never on absolute values. + let before: Vec = DECODE_FAILURES + .iter() + .map(|counter| counter.load(Ordering::Relaxed)) + .collect(); + record_decode_failure("dns_cred/abc"); + let after: Vec = DECODE_FAILURES + .iter() + .map(|counter| counter.load(Ordering::Relaxed)) + .collect(); + + let moved = prefix_index("dns_cred/abc"); + for (index, (before, after)) in before.iter().zip(after.iter()).enumerate() { + let expected = if index == moved { before + 1 } else { *before }; + assert_eq!(*after, expected, "bucket {} moved unexpectedly", index); + } + } +} diff --git a/dstack/gateway/src/web_routes.rs b/dstack/gateway/src/web_routes.rs index 5f72735db..a1b92f37a 100644 --- a/dstack/gateway/src/web_routes.rs +++ b/dstack/gateway/src/web_routes.rs @@ -4,8 +4,9 @@ use crate::main_service::Proxy; use anyhow::Result; -use rocket::{get, response::content::RawHtml, routes, Route, State}; +use rocket::{get, http::ContentType, response::content::RawHtml, routes, Route, State}; +mod metrics; mod route_index; mod wavekv_sync; @@ -14,13 +15,26 @@ async fn index(state: &State) -> Result, String> { route_index::index(state).await.map_err(|e| format!("{e}")) } +/// Prometheus scrape endpoint. +/// +/// Mounted on the admin listener only: the series below name domains, node ids +/// and instance counts, which is topology no unauthenticated caller should be +/// able to read. +#[get("/metrics")] +fn scrape(state: &State) -> (ContentType, String) { + // Naming the exposition version lets a scraper pick its parser instead of + // inferring one from a bare `text/plain`. + let content_type = ContentType::new("text", "plain").with_params([("version", "0.0.4")]); + (content_type, metrics::render(state)) +} + #[get("/health")] fn health() -> &'static str { "OK" } pub fn routes() -> Vec { - routes![index] + routes![index, scrape] } /// Health endpoint for simple liveness checks @@ -32,3 +46,22 @@ pub fn health_routes() -> Vec { pub fn wavekv_sync_routes() -> Vec { routes![wavekv_sync::sync_store] } + +#[cfg(test)] +mod tests { + use super::*; + + /// The scrape output names domains, node ids and instance counts, so `/metrics` + /// must be part of the authenticated admin route set (`routes()`) and must not + /// be included in the route sets intended for non-admin listeners. + /// + /// This test checks the route-set membership only; it does not inspect Rocket + /// listener wiring in `main.rs`. + #[test] + fn metrics_is_mounted_on_the_admin_listener_only() { + let mounted = |set: Vec| set.iter().any(|route| route.uri.path() == "/metrics"); + assert!(mounted(routes())); + assert!(!mounted(health_routes())); + assert!(!mounted(wavekv_sync_routes())); + } +} diff --git a/dstack/gateway/src/web_routes/metrics.rs b/dstack/gateway/src/web_routes/metrics.rs new file mode 100644 index 000000000..33ff73cbc --- /dev/null +++ b/dstack/gateway/src/web_routes/metrics.rs @@ -0,0 +1,87 @@ +// SPDX-FileCopyrightText: © 2026 Phala Network +// +// SPDX-License-Identifier: Apache-2.0 + +//! `/metrics` scrape handler. + +use std::sync::atomic::Ordering; + +use rocket::State; + +use crate::{ + main_service::Proxy, + metrics::{self, PeerSnapshot, Snapshot, StoreSnapshot}, + proxy::{stats::accel_status, NUM_CONNECTIONS}, +}; + +pub fn render(state: &State) -> String { + metrics::render(&sample(state)) +} + +fn sample(state: &State) -> Snapshot { + // Everything below reads replicated records through the decoding helpers + // that feed `record_decode_failure`. A scrape reporting a counter must not + // also be a writer of it. + let _sampling = metrics::scrape_guard(); + + let kv_store = state.kv_store().clone(); + let accel = accel_status(&state.config.proxy); + + // The public data path takes this lock on every connection, so the scrape + // holds it for one O(1) count and nothing else. + // + // The node counts deliberately do not go through `get_all_nodes()` / + // `get_active_nodes()`: those read no proxy state at all -- only + // `self.kv_store` -- so routing them through the lock would drag two loads + // of the node table, a `GatewayNodeInfo` per node, and one ephemeral-lock + // acquisition per node for an unused `last_seen` in here with them. + let instances = state.lock().state.instances.len() as u64; + let (nodes_total, nodes_active) = kv_store.count_nodes(); + + let stores = vec![ + store_snapshot("persistent", kv_store.persistent()), + store_snapshot("ephemeral", kv_store.ephemeral()), + ]; + + let cert_not_after = kv_store + .load_all_cert_data() + .into_iter() + .map(|(domain, data)| (domain, data.not_after)) + .collect(); + + Snapshot { + version: crate::app_version(), + node_id: kv_store.my_node_id(), + instances, + connections: NUM_CONNECTIONS.load(Ordering::Relaxed), + nodes_total, + nodes_active, + accel, + stores, + cert_not_after, + } +} + +fn store_snapshot(name: &'static str, node: &wavekv::node::Node) -> StoreSnapshot { + // `status()` is O(peers) on wavekv 1.x, so this is cheap enough to do under + // the read lock. On the 2.0 branch it also computes a state digest over the + // whole dataset -- at that point this call hashes the entire store, twice + // per scrape, while blocking writers. Revisit when the dependency moves. + let status = node.read().status(); + StoreSnapshot { + name, + keys: status.n_kvs as u64, + next_seq: status.next_seq, + dirty: status.dirty, + peers: status + .peers + .into_iter() + .map(|peer| PeerSnapshot { + id: peer.id, + local_ack: peer.ack, + peer_ack: peer.pack, + buffered_logs: peer.logs as u64, + }) + .collect(), + } +}